v3.2.x: Kafka rev2 with multi-topic and header support

raddb/mods-available/kafka

@@ -1,6 +1,5 @@
 kafka {
 	bootstrap-servers = "localhost:9092"
-	topic = "radius"
 
 	#
 	#  This initialises the default, synchronous instance of the kafka
@@ -28,12 +27,30 @@ kafka {
 		#ssl.ca.location = "${cadir}"
 		#sasl.username = "kafkauser"
 		#sasl.password = "kafkapass"
-		# ...
+		#  ...
 		#
 
 	}
 
-	topic-config {
+	#
+	#  Example of a user-defined topic called by reference from another
+	#  section, or by using its name (which may be changed) in an xlat of
+	#  either form:
+	#
+	#    - "%{kafka:custom-topic (<header-list>) &Key-Attr-Ref Remainder is the message}"
+	#    - "%{kafka:custom-topic (<header-list>)  Remainder is the message}", if no key
+	#    - "%{kafka:custom-topic &Key-Attr-Ref Remainder is the message}"
+	#    - "%{kafka:custom-topic  Remainder is the message}", if no key
+	#
+	#  The attribute referred to by the optional &Key-Attr-Ref holds the
+	#  Kafka message's key. Notice that it is replaced with a mandatory
+	#  space if there is no key.
+	#
+	#  <header-list> has the same format as the json_encode xlat. For
+	#  example `&request[*] !&User-Password` creates Kafka headers from all
+	#  attributes in the RADIUS request, except for `User-Password`.
+	#
+	topic-config custom-topic {
 
 		#
 		#  The message durabily strategy: Number of acks required by all
@@ -43,7 +60,7 @@ kafka {
 
 		#
 		#  How long a message can remain undelivered in the local
-		#  in-memory queue, with the  before it is timed out.
+		#  in-memory queue before it is timed out.
 		#
 		#  For synchronous delivery, the default local queue is set to
 		#  25 sec. It should be some value less than max_request_time
@@ -64,6 +81,41 @@ kafka {
 
 	}
 
+	#
+	#  Topic used by calls from an accounting section
+	#
+	topic-config accounting {
+
+		#
+		#  You may either provide topic configuration parameters
+		#  or reference an existing topic (declared above this) to
+		#  reuse it.
+		#
+		#reference = custom-topic
+
+	}
+
+	#
+	#  Topic used by calls from an authorize section
+	#
+	topic-config authorize {
+
+		#
+		#  Used to set the topic name to something other than the
+		#  section from which it was called.
+		#
+		#name = "radius.auth_request"
+
+	}
+
+	#
+	#  Topic used by calls from a post-auth section
+	#
+	topic-config post-auth {
+		#name = "radius.auth_result"
+		#request.required.acks = 2
+	}
+
 	#
 	#  Uncomment to generate producer statistics
 	#

raddb/mods-available/kafka_async

@@ -1,6 +1,5 @@
 kafka kafka_async {
 	bootstrap-servers = "localhost:9092"
-	topic = "radius"
 
 	#
 	#  This initialises an asynchronous instance of the kafka module.
@@ -58,7 +57,25 @@ kafka kafka_async {
 
 	}
 
-	topic-config {
+	#
+	#  Example of a user-defined topic called by reference from another
+	#  section, or by using its name (which may be changed) in an xlat of
+	#  either form:
+	#
+	#    - "%{kafka_async:custom-topic (<header-list>) &Key-Attr-Ref Remainder is the message}"
+	#    - "%{kafka_async:custom-topic (<header-list>)  Remainder is the message}", if no key
+	#    - "%{kafka_async:custom-topic &Key-Attr-Ref Remainder is the message}"
+	#    - "%{kafka_async:custom-topic  Remainder is the message}", if no key
+	#
+	#  The attribute referred to by the optional &Key-Attr-Ref holds the
+	#  Kafka message's key. Notice that it is replaced with a mandatory
+	#  space if there is no key.
+	#
+	#  <header-list> has the same format as the json_encode xlat. For
+	#  example `&request[*] !&User-Password` creates Kafka headers from all
+	#  attributes in the RADIUS request, except for `User-Password`.
+	#
+	topic-config custom-topic {
 
 		#
 		#  The message durabily strategy: Number of acks required by all
@@ -90,8 +107,41 @@ kafka kafka_async {
 	}
 
 	#
-	#  Uncomment to generate producer statistics
+	#  Topic used by calls from an accounting section
+	#
+	topic-config accounting {
+
+		#
+		#  You may either provide topic configuration parameters
+		#  or reference an existing topic (declared above this) to
+		#  reuse it.
+		#
+		#reference = custom-topic
+
+	}
+
+	#
+	#  Topic used by calls from an authorize section
+	#
+	topic-config authorize {
+
+		#
+		#  Used to set the topic name to something other than the
+		#  section from which it was called.
+		#
+		#name = "radius.auth_request"
+
+	}
+
+	#
+	#  Topic used by calls from a post-auth section
 	#
+	topic-config post-auth {
+		#name = "radius.auth_result"
+		#request.required.acks = 2
+	}
+
+
 #	statistics {
 #		file = /tmp/kafka_async_stats.json
 #	}

raddb/mods-config/kafka/messages-json.conf

@@ -1,19 +1,31 @@
 #
-#  To use the json_encode xlats in the schemas below, you must first enable the
-#  json module, and may probably want to set json.encode.output_mode = object_simple
+#  key:       An optional key for the message.
+#
+#  headers:   An optional specification for which attributes should
+#             be set as Kafka headers, with their values encoded as strings.
+#
+#  message:   The message itself.
+#
+#  reference: Expanded to a configuration item reference that contains the
+#             message.
+#
+#  To use the json_encode xlats in the schemas examples below, you must first
+#  enable the json module, and may probably want to set
+#  json.encode.output_mode = object_simple
 #
 
 authorize {
 
 	key = "%{User-Name}"
+	headers = "&Calling-Station-Id"
 	message = "%{json_encode:&request:[*]}"
 
 }
 
 post-auth {
 
 	key = "%{User-Name}"
-
+	headers = "&reply:Framed-IP-Address"
 	reference = "messages.%{%{reply:Packet-Type}:-default}"
 
 	messages {
@@ -29,6 +41,7 @@ post-auth {
 accounting {
 
 	key = "%{Acct-Unique-Session-Id}"
+	headers = "&Acct-Status-Type &NAS-Identifier"
 
 	reference = "messages.%{%{Acct-Status-Type}:-default}"