Creating Kafka connections

Entering Field mapping information

Unlike other connection types, you can specify what kind of information you want to run through your Kafka connection. For example, if you want to collect data from multiple custom sources, you can set them up as separate Kafka connections.

1. In the Field mapping section, set the Data source to specify the type of data incoming from the Kafka connection. You can select Events or Logs.

2. Specify the Mapping type, which creates the field mapping for your specified log type.

   • Mapping types for Events include None, PagerDuty, and NOI.
   • Mapping types for Logs are None, ELK, Falcon LogScale, LogDNA, Custom, and Splunk.
   • If you have log data that is already normalized, choose None.
   • If your log data is in one of the supported log formats, such as ELK or Falcon LogScale, choose the corresponding log format.
   • For field mapping for Custom types, see Creating custom connections.

3. The Topic name for a Kafka connection is predefined. Note this topic and send your data to it.

4. For log connections, enter the Maximum number of logs per second rounded to the nearest thousand. The maximum number of logs per second is collected in increments of 1,000 up to 18,000 and is rounded up to the nearest thousand sent from Kafka to IBM Cloud Pak for Watson AIOps.

5. Set the Field mapping mode to Live data for initial AI training for initial training or to Live data for continuous AI training and anomaly detection.

   • You can improve search performance by mapping the fields from your implementation fields to standard IBM Cloud Pak for Watson AIOps fields. For more information about how field mappings are defined, see Mapping data from incoming sources.
   • For more information about using mappings to clean your data for use in IBM Cloud Pak for Watson AIOps, see Cleaning mapped data using regular expressions.

6. In the Mapping section, if you use a Mapping type other than None, verify whether the JSON document that is automatically assigned to this field is valid. The validity depends on the structure of your log records. For example, for a PoC deployment, an example default Field mapping for ELK can resemble the following mapping:

     {
       "codec": "elk",
       "message_field": "@message",
       "log_entity_types": "@hostname, @bundleName, @context.Environment",
       "instance_id_field": "@properties.processtype",
       "rolling_time": 10,
       "timestamp_field": "@timestamp"
     }
   

   If you chose None for the Mapping type, each log record in your log files conforms to the following format. Set both of the application_id and application_group_id values to "1000", and make sure that a 13-digit timestamp value is assigned to the timestamp field:

   {
       "timestamp": 1581549409000,
       "utc_timestamp": "2020-02-12 23:16:49.000",
       "instance_id": "calico-node-jn2d2",
       "application_group_id": "1000",
       "application_id": "1000",
       "features": [],
       "meta_features": [],
       "level": 1,
       "message": "[64] ipsets.go 254: Resyncing ipsets with dataplane. family=\"inet\"",
       "entities": {
           "pod": "calico-node-jn2d2",
           "cluster": null,
           "container": "calico-node",
           "node": "kube-bmgcm5td0stjujtqv8m0-ai4itsimula-wp16cpu-00002c34"
       },
       "type": "StandardLog"
   }
   

   Important: The Events data that is collected must follow the Kafka Connection Normalized Event schema. For more information, see Normalized mapping rules.

7. Click Next.

Entering AI training and log data information

1. If you want to enable the connection or if you want to enable data collection for AI training and log data, switch the Data collection toggle to On.
2. Click Save to save your connection.

You created a Kafka connection in your instance, and you can now use the Kafka connection as the basis for your offline AI model training. After you create your connection, enable the data collection to connect your connection with the AI of IBM Cloud Pak for Watson AIOps.

• For more information about enabling your connection, see Enabling Kafka connections.
• To create more connections, such as a ChatOps connection, see Configuring data and tool connections.
• For more information about working with the insights provided by your connections, see ChatOps insight management.
• For more information about AI model training, see Configuring AI models.

Using a Kafka utility to publish log data

1. Install a Kafka utility, such as the kcat utility.

2. Log in to the Red Hat® OpenShift® cluster where IBM Cloud Pak for Watson AIOps is installed.

3. Run the following commands from the namespace where IBM Cloud Pak for Watson AIOps is installed:

   oc extract secret/iaf-system-cluster-ca-cert --keys=ca.crt --to=- > ca.crt
   
   export sasl_password=$(oc get secret cp4waiops-cartridge-kafka-auth-0 --template={{.data.password}} | base64 --decode);
   
   export BROKER=$(oc get routes iaf-system-kafka-bootstrap -o=jsonpath='{.status.ingress[0].host}{"\n"}'):443
   

4. Publish the log data to the Kafka topic. This example uses kcat. If you also use kcat, update the variables and run the following commands:

   export KAFKA_TOPIC=<kafka topic>
   
   export LOG_FILE=<log file>
   
   kcat -X security.protocol=SASL_SSL -X ssl.ca.location=ca.crt -X sasl.mechanisms=SCRAM-SHA-512 -X sasl.username=cp4waiops-cartridge-kafka-auth-0 -X sasl.password=$sasl_password -X enable.ssl.certificate.verification=false -b $BROKER -P -t $KAFKA_TOPIC -l $LOG_FILE
   

Copying log data into the Kafka cluster

Use this option only if you can't use a Kafka utility such as kcat for your environment. Complete the following steps to publish the log data to the corresponding Kafka topic by copying the log data into the Kafka cluster:

1. Run the following commands from the namespace where IBM Cloud Pak for Watson AIOps is installed to obtain the Kafka password and Kafka broker. Record both values, which you need in a following step.

   export kafka_password=$(oc get secret cp4waiops-cartridge-kafka-auth-0 --template={{.data.password}} | base64 --decode)
   
   export kafka_broker=$(oc get routes iaf-system-kafka-bootstrap -o=jsonpath='{.status.ingress[0].host}{"\n"}'):443
   
   echo $kafka_password
   
   echo $kafka_broker
   

2. Copy the log file to the Kafka cluster:

   oc cp <log file> iaf-system-kafka-0:/tmp
   

3. Check that the file is copied to the Kafka cluster from the namespace where IBM Cloud Pak for Watson AIOps is installed:

   oc exec -it iaf-system-kafka-0 -- bash
   
   ls /tmp/<log file>
   
   exit
   

4. Get the ca.cert certificate file:

   oc extract secret/iaf-system-cluster-ca-cert --keys=ca.crt --to=- > ca.crt
   

5. Convert ca.crt to Java keystore format (JKS):

   keytool -import -trustcacerts -alias root -file ca.crt -keystore truststore.jks -storepass password -noprompt
   

6. Copy the truststore file to the Kafka cluster:

   oc cp truststore.jks iaf-system-kafka-0:/tmp
   

7. Give bash access to the Kafka cluster again:

   oc exec -it iaf-system-kafka-0 -- bash
   

8. Create a file within the Kafka pod that is called producer.properties. The contents of the producer.properties file can resemble the following properties. Update the variable placeholders before you use this content. The ssl.truststore.location value points to the JSK file that you created earlier.

   security.protocol=SASL_SSL
   ssl.truststore.location=/tmp/truststore.jks
   ssl.truststore.password=password
   sasl.mechanism=SCRAM-SHA-512
   sasl.jaas.config=org.apache.Kafka.common.security.scram.ScramLoginModule required \
   username="cp4waiops-cartridge-kafka-auth-0" \
   password="<kafka password>"
   

9. From the pod, run the Kafka producer script to publish the log data to the corresponding Kafka topic. Update the variable placeholders:

   /opt/Kafka/bin/Kafka-console-producer.sh --broker-list <kafka broker> --producer.config producer.properties --topic <kafka topic> < <log file>