<traffic>\$type \$time_generated {"src_ip": "\$src", "sport":"\$sport", "dst_ip": "\$dst", "dport":"\$dport", "proto": "\$proto", "app": "\$app", "rule":"\$rule", "action": "\$action", "bytes_recv":"\$bytes_received", "bytes_sent":"\$bytes_sent", "pkts_received": "\$pkts_received", "pkts_sent":"\$pkts_sent", "start_time": "\$start", "elapsed_time": "\$elapsed", "repeat_count":"\$repeatcnt", "category":"\$category", "src country":"\$srcloc", "dst country":"\$dstloc", "session_end_reason": "\$session_end_reason", "xff_ip": "\$xff_ip"}</traffic>
Threat Format- XPath:
/config/shared/log-settings/syslog/entry[@name='syslog-kafka-threat']/format
- Element:
<threat>\$type \$time_generated {"src_ip": "\$src", "sport":"\$sport", "dst":"\$dst", "dport":"\$dport", "proto": "\$proto", "app":"\$app", "rule":"\$rule", "action": "\$action", "threat_category": "\$thr_category", "sub_type": "\$subtype", "threat_content_name":"\$threatid", "severity": "\$severity", "direction": "\$direction", "repeatcnt":"\$repeatent", "data_filter_reason": "\$reason", "filetype": "\$filetype", "contentver": "\$contentver"}</threat>
- URL Format
- XPath:
/config/shared/log-settings/syslog/entry[@name='syslog-kafka-url']/format
- Element:
<url>\$type \$time_generated {"src_ip":"\$src", "sport":"\$sport", "dst":"\$dst", "dport":"\$dport", "proto": "\$proto", "app":"\$app", "rule":"\$rule", "action": "\$action", "url_idx":"\$url_idx", "url_category_list": "\$url_category_list", "severity": "\$severity", "direction": "\$direction", "repeatent":"\$repeatent", "xff":"\$xff", "url_filename":"\$misc"}</url>
- Configure log forwarding profile.
Route each individual log type to its
corresponding newly created syslog profile. Run the following in
configuration
mode:
set shared log-settings profiles kafka-log-fwd match-list traffic-fwd filter "All Logs" log-type traffic send-syslog syslog-kafka-traffic
set shared log-settings profiles kafka-log-fwd match-list threat-fwd filter "All Logs" log-type threat send-syslog syslog-kafka-threat
set shared log-settings profiles kafka-log-fwd match-list url-fwd filter "All Logs" log-type url send-syslog syslog-kafka-url
- Attach log forwarding profile to security rules
Attach the forwarding
profile to the specific security rules governing the traffic you intend
to stream. Make sure log-end yes is verified to cleanly capture session
termination traffic
logs.
set rulebase security rules allow-interzone log-setting kafka-log-fwd
set rulebase security rules allow-interzone log-end yes
- Commit the configuration.
Verification and Health Check
To check Kafka server statuses: Execute the command:
show plugins vm_series kafka-servers (verify traffic,
threat, and url outputs show correct brokers and cert profiles).
To check log port mapping: Execute the command:
show plugins vm_series kafka-server-log-type-mapping
(verify traffic points to 5141, threat to 5142, and url to 5143).
Check Pipeline Health
Validate the running states of your connections and streaming
statistics:
Connectivity/mTLS Health: Run
debug plugins vm_series kafka config-status and confirm
that it outputs: STATUS: All checks passed, mTLS OK, Private Key: Present on
all 3 outputs.
Live Statistics: Run
debug plugins vm_series kafka statistics. Ensure
traffic_logs_input > 0 after passing some ICMP traffic,
Total Dropped: 0, and the output status reads
OK for all three log outputs (traffic_output,
threat_output, url_output).