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

raddb/mods-available/kafka

@@ -32,18 +32,25 @@ kafka {
 
 	}
 
-
-        #
-        #  A user-defined topic called by reference or by using its name (which may
-        #  be changed) in an xlat of either form:
-        #
-        #    - "%{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 &Key-Attr-Ref holds the Kafka message's key.
-        #  Notice two spaces leading the message in case there is no key.
-        #
-        topic-config custom-topic {
+	#
+	#  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

raddb/mods-available/kafka_async

@@ -58,14 +58,22 @@ kafka kafka_async {
 	}
 
 	#
-	#  A user-defined topic called by reference or by using its name (which may
-	#  be changed) in an xlat of either form:
+	#  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 &Key-Attr-Ref holds the Kafka message's key.
-	#  Notice two spaces leading the message in case there is 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 {
 

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}"