WIP: Kafka/AVRO output bot.

intelmq/bots/BOTS

@@ -623,6 +623,15 @@
  "replacement_char": "_"
  }
  },
+ "Kafka": {
+ "description": "Kafka is the bot responsible to send events to a kafka topic.",
+ "module": "intelmq.bots.outputs.kafka.output",
+ "parameters": {
+ "kafka_broker_list": "127.0.0.1",
+ "kafka_topic": "intelmq",
+ "flatten_fields": "extra"
+ }
+ },
  "File": {
  "description": "File is the bot responsible to send events to a file.",
  "module": "intelmq.bots.outputs.file.output",

intelmq/bots/outputs/kafka/README.md

@@ -0,0 +1,29 @@
+# Kafka Output Bot
+
+### Output Bot that sends events to kafka
+
+
+Bot parameters:
+
+Kafka Producer
+* kafka_broker_list : comma seperated list of kafka brokers. defaults to 127.0.0.1
+
+* kafka_topic : Index for the ElasticSearch output, defaults to intelmq
+
+* flatten_fields : In ES, some query and aggregations work better if the fields are flat and not JSON. 
+ Here you can provide a list of fields to convert. Can be a list of strings (fieldnames) 
+ or a string with field names separated by a comma (,). eg `extra,field2` or `['extra', 'field2']`
+ Default: ['extra']
+
+AVRO Producer
+* avro_topic_schema : a file path, pointing to a file containing a dict object containing expected values, and their destination topics.
+ the keys in this dict should be all expected values in 'avro_topic_field' 
+
+ IMPORTANT:schema must contain "other" keyword, with either None, or a topic name as the value.
+ If None is declared, other values are dropped.
+
+* avro_topic_field : The field to be used to map intelligence identy to destination topic
+
+* avro_value_schema_file: The schema file 
+
+* avro_schema_registry: URL where the schema registry is defined.

intelmq/bots/outputs/kafka/REQUIREMENTS.txt

@@ -0,0 +1 @@
+confluent-kafka==0.11.4

intelmq/bots/outputs/kafka/output.py

@@ -0,0 +1,126 @@
+# -*- coding: utf-8 -*-
+
+from json import loads
+from json import dumps
+from json import load
+from collections.abc import Mapping
+from intelmq.lib.utils import load_configuration
+from intelmq.lib.bot import Bot
+try:
+ from confluent_kafka import Producer
+except ImportError:
+ Producer = None
+try:
+ from confluent_kafka import avro
+except ImportError:
+ avro = None
+
+if avro is not None:
+ from confluent_kafka.avro import AvroProducer
+
+
+def replace_keys(obj, key_char='.', replacement='_'):
+ if isinstance(obj, Mapping):
+ replacement_obj = {}
+ for key, val in obj.items():
+ replacement_key = key.replace(key_char, replacement)
+ replacement_obj[replacement_key] = replace_keys(val, key_char, replacement)
+ return replacement_obj
+ return obj
+
+
+class KafkaOutputBot(Bot):
+
+ def init(self):
+ if Producer is None:
+ raise ValueError('Missing confluent-kafka-python module.')
+ if avro is None:
+ raise ValueError('Missing python3-avro module')
+
+ self.broker_list = getattr(self.parameters,
+ 'kafka_broker_list', '127.0.0.1:9092')
+ self.kafka_topic = getattr(self.parameters,
+ 'kafka_topic', 'intelmq')
+ self.flatten_fields = getattr(self.parameters,
+ 'flatten_fields', ['extra'])
+ self.enable_avro = getattr(self.parameters,
+ 'enable_avro', False)
+
+ # Fields below are to define an output schema via AVRO
+ if self.enable_avro is True:
+
+ self.avro_value_schema = avro.loads(
+ dumps(load_configuration(self.parameters.avro_value_schema_file)))
+ self.avro_key_schema = avro.loads(
+ dumps(load_configuration(self.parameters.avro_key_schema_file)))
+
+ self.avro_topic_schema = load_configuration(self.parameters.avro_topic_schema)
+
+ self.avro_topic_field = getattr(
+ self.parameters, 'avro_topic_field', None)
+ self.avro_schema_registry = getattr(
+ self.parameters, 'avro_schema_registry', None)
+
+ # Build a list of producers for each destination topic
+ self.producer = AvroProducer({
+ 'bootstrap.servers': self.broker_list,
+ 'schema.registry.url': self.avro_schema_registry
+ },
+ default_key_schema=self.avro_key_schema,
+ default_value_schema=self.avro_value_schema
+ )
+ else:
+ self.producer = Producer({
+ 'bootstrap.servers': self.broker_list
+ })
+
+ if isinstance(self.flatten_fields, str):
+ self.flatten_fields = self.flatten_fields.split(',')
+
+ def process(self):
+ event = self.receive_message()
+ event_dict = event.to_dict(hierarchical=False)
+
+ for field in self.flatten_fields:
+ if field in event_dict:
+ val = event_dict[field]
+ # if it's a string try to parse it as JSON
+ if isinstance(val, str):
+ try:
+ val = loads(val)
+ except ValueError:
+ pass
+ if isinstance(val, Mapping):
+ for key, value in val.items():
+ event_dict[field + '.' + key] = value
+ event_dict.pop(field)
+
+ event_dict = replace_keys(event_dict)
+
+ if self.enable_avro is False:
+ self.producer.produce(self.kafka_topic, dumps(event_dict).encode('utf-8'), callback=self.delivery_report)
+ self.acknowledge_message()
+ self.kafka.flush()
+ else:
+ try:
+ key_field = str(event_dict[self.avro_topic_field]).replace('.', '_')
+ except KeyError:
+ self.logger.debug('Event %s has no field %s, dropping.', format(event_dict), format(self.avro_topic_field))
+ self.acknowledge_message()
+ return
+
+ submit_key = {'indicator': event_dict[key_field]}
+ event_topic = self.avro_topic_schema[key_field]
+ self.logger.debug('Shipped %s to topic: %s', format(submit_key), format(event_topic))
+ self.producer.produce(topic=event_topic, value=event_dict, key=submit_key)
+ self.acknowledge_message()
+ self.producer.poll(0)
+
+ def delivery_report(self, err, msg):
+ """ Called once for each message produced to flag for failure.
+ Triggered by poll() or flush()."""
+ if err is not None:
+ self.logger.exception('Message delivery failed: {}'.format(err))
+
+
+BOT = KafkaOutputBot