JonZeolla commented on a change in pull request #47:
URL:
https://github.com/apache/metron-bro-plugin-kafka/pull/47#discussion_r428631726
##
File path: README.md
##
@@ -179,6 +179,48 @@ event zeek_init() =-10
}
```
+ Dynamically send each zeek log to a topic with its same name.
+
+ * ej. `CONN::LOG` logs are sent to the `conn` topic or `Known::CERTS_LOG` to
the `known-certs` topic.
+
+```
+@load packages/metron-bro-plugin-kafka/Apache/Kafka
+redef Kafka::logs_to_send = set(DHCP::LOG, RADIUS::LOG, DNS::LOG);
+redef Kafka::topic_name = "";
+redef Kafka::tag_json = T;
+
+event zeek_init() =-10
+{
+for (stream_id in Log::active_streams) {
+# Convert stream type enum to string
+const stream_string: string = fmt("%s", stream_id);
+
+# replace `::` by `_` from the log string name
+ # ej. CONN::LOG to CONN_LOG or Known::CERTS_LOG to Known_CERTS_LOG
+const stream_name: string = sub(stream_string, /::/, "_");
+
+# lowercase the whole string for nomalization
+const topic_name_lower: string = to_lower(stream_name);
+
+# remove the _log at the of each topic name
+const topic_name_under: string = sub(topic_name_lower, /_log$/, "");
+
+# replace `_` by `-` for compatibility with acceptable Kafka topic naes
Review comment:
I assume you are referring to
[this](https://github.com/apache/kafka/blob/d63e0181bb7b9b4f5ed088abc00d7b32aeb0/core/src/main/scala/kafka/admin/TopicCommand.scala#L198-L199)?
Because otherwise, _ appears to be an allowed character in a topic name.
##
File path: README.md
##
@@ -179,6 +179,48 @@ event zeek_init() =-10
}
```
+ Dynamically send each zeek log to a topic with its same name.
+
+ * ej. `CONN::LOG` logs are sent to the `conn` topic or `Known::CERTS_LOG` to
the `known-certs` topic.
+
+```
+@load packages/metron-bro-plugin-kafka/Apache/Kafka
+redef Kafka::logs_to_send = set(DHCP::LOG, RADIUS::LOG, DNS::LOG);
+redef Kafka::topic_name = "";
+redef Kafka::tag_json = T;
+
+event zeek_init() =-10
+{
+for (stream_id in Log::active_streams) {
+# Convert stream type enum to string
+const stream_string: string = fmt("%s", stream_id);
+
+# replace `::` by `_` from the log string name
+ # ej. CONN::LOG to CONN_LOG or Known::CERTS_LOG to Known_CERTS_LOG
Review comment:
Nit: Please keep the indentation consistent.
##
File path: README.md
##
@@ -179,6 +179,48 @@ event zeek_init() =-10
}
```
+ Dynamically send each zeek log to a topic with its same name.
+
+ * ej. `CONN::LOG` logs are sent to the `conn` topic or `Known::CERTS_LOG` to
the `known-certs` topic.
+
+```
+@load packages/metron-bro-plugin-kafka/Apache/Kafka
+redef Kafka::logs_to_send = set(DHCP::LOG, RADIUS::LOG, DNS::LOG);
+redef Kafka::topic_name = "";
+redef Kafka::tag_json = T;
+
+event zeek_init() =-10
+{
+for (stream_id in Log::active_streams) {
+# Convert stream type enum to string
+const stream_string: string = fmt("%s", stream_id);
+
+# replace `::` by `_` from the log string name
+ # ej. CONN::LOG to CONN_LOG or Known::CERTS_LOG to Known_CERTS_LOG
+const stream_name: string = sub(stream_string, /::/, "_");
+
+# lowercase the whole string for nomalization
+const topic_name_lower: string = to_lower(stream_name);
+
+# remove the _log at the of each topic name
+const topic_name_under: string = sub(topic_name_lower, /_log$/, "");
+
+# replace `_` by `-` for compatibility with acceptable Kafka topic naes
Review comment:
naes typo
##
File path: README.md
##
@@ -179,6 +179,48 @@ event zeek_init() =-10
}
```
+ Dynamically send each zeek log to a topic with its same name.
+
+ * ej. `CONN::LOG` logs are sent to the `conn` topic or `Known::CERTS_LOG` to
the `known-certs` topic.
+
+```
+@load packages/metron-bro-plugin-kafka/Apache/Kafka
+redef Kafka::logs_to_send = set(DHCP::LOG, RADIUS::LOG, DNS::LOG);
+redef Kafka::topic_name = "";
+redef Kafka::tag_json = T;
+
+event zeek_init() =-10
+{
+for (stream_id in Log::active_streams) {
+# Convert stream type enum to string
+const stream_string: string = fmt("%s", stream_id);
+
+# replace `::` by `_` from the log string name
+ # ej. CONN::LOG to CONN_LOG or Known::CERTS_LOG to Known_CERTS_LOG
+const stream_name: string = sub(stream_string, /::/, "_");
+
+# lowercase the whole string for nomalization
+const topic_name_lower: string = to_lower(stream_name);
+
+# remove the _log at the of each topic name
+const topic_name_under: string = sub(topic_name_lower, /_log$/, "");
+
+# replace `_` by `-` for compatibility with acceptable Kafka topic naes
+const topic_name: string = sub(topic_name_under, /_/, "-");
+
+if (|Kafka::logs_to_send| == 0 || stream_id in
