Modern cloud architectures demand seamless visibility, yet integrating disparate logging systems remains a common challenge. In enterprise environments, piping security event data from cloud infrastructure to on-premises or cloud-native Security Information and Event Management (SIEM) systems requires a highly reliable and performant pipeline.
This guide provides a comprehensive configuration for ingesting security logs from an Oracle Cloud Infrastructure (OCI) Streaming (Kafka) source, transforming them into Common Event Format (CEF) via Fluent Bit, and delivering them securely to a Trend Micro Service Gateway via TCP Syslog.
This setup directly addresses the challenge of handling nested JSON objects (typical in OCI Web Application Firewall and Load Balancer logs) and complies with the strict connection constraints of the Trend Micro receiver.
Architecture Overview
The pipeline architecture consists of three core components:
- Input: OCI Streaming (Kafka API) authenticated via SASL/PLAIN.
- Processor: A robust Lua filter that performs recursive table flattening on complex structures.
- Output: A TCP Syslog stream utilizing the RFC3164 format and targeting a specific custom message key.
1. Lua Transformation Script
Converting complex OCI JSON into a flat string that SIEMs can successfully parse requires specialized processing. The following Lua script resolves the common table: 0x... memory address leak issue by recursively drilling into nested JSON objects. It maps security fields specifically for Trend Micro compatibility and formats the resulting CEF payload safely.
Path: /etc/fluent-bit/scripts/cef_wrapper.lua
-- Helper to escape special CEF characters (|, =, \)
local function escape_cef(val)
if val == nil then return "" end
local s = tostring(val)
-- CEF requires escaping backslashes, pipes, and equal signs in values
s = s:gsub("\\", "\\\\"):gsub("|", "\\|"):gsub("=", "\\=")
return s
end
-- Improved Recursive function to flatten OCI tables
-- Uses colons (:) to avoid "Parser Drift" in SIEMs caused by nested "="
function flatten_table(obj)
if type(obj) ~= "table" then
return tostring(obj)
end
local res = ""
for k, v in pairs(obj) do
if type(v) == "table" then
-- Use brackets and colons for nested structures
res = res .. k .. ":[" .. flatten_table(v) .. "] "
else
-- Use colons to ensure SIEM doesn't think this is a new CEF key
res = res .. k .. ":" .. tostring(v) .. " "
end
end
return res
end
function wrap_cef(tag, timestamp, record)
-- 1. Determine the root of the log data
local log_body = record["payload"] or record
local data = log_body["data"] or {}
local request = data["request"] or {}
-- 2. Extract Specific Security Fields (Correctly mapped for Trend Micro)
local src_ip = data["clientAddr"] or ""
local req_path = request["path"] or ""
local method = request["method"] or ""
local action = data["action"] or ""
local host = data["host"] or ""
local event_id = log_body["type"] or "com.oraclecloud.loadbalancer.waf"
-- 3. Trend Micro Mandatory CEF Headers
local vendor = "Oracle"
local product = "OCI"
local version = "1.0"
local severity = "3"
local name = "OCI_Log_Entry"
-- 4. Build the Extension String (Standard CEF Keys)
-- Placing these FIRST ensures they are indexed even if 'msg' is truncated
local extensions = string.format("src=%s request=%s method=%s act=%s cs1Label=Host cs1=%s ",
src_ip,
escape_cef(req_path),
escape_cef(method),
escape_cef(action),
escape_cef(host))
-- 5. Flatten the rest into the 'msg' field using the safer colon format
local msg_content = flatten_table(log_body)
-- 6. Construct the Final CEF String with Syslog Priority <13>
local cef_msg = string.format("<13>CEF:0|%s|%s|%s|%s|%s|%s|%smsg=%s",
vendor, product, version, event_id, name, severity, extensions, msg_content)
-- 7. Hard Truncation at 8000 bytes for network safety
if #cef_msg > 8000 then
cef_msg = cef_msg:sub(1, 7980) .. " [TRUNCATED]"
end
-- 8. Final Output Preparation
local new_record = {}
-- The newline \n is critical for TCP stream delimited parsing
new_record["formatted_cef"] = cef_msg .. "\n"
return 1, timestamp, new_record
end
2. Fluent Bit Configuration
To deliver our formatted CEF string, we configure the Fluent Bit syslog output plugin. By leveraging the syslog_message_key directive, we guarantee that only our newly constructed CEF string is sent downstream rather than the entire wrapped JSON structure.
Path: /etc/fluent-bit/fluent-bit.conf
[SERVICE]
Flush 1
Daemon Off
Log_Level info
HTTP_Server On
HTTP_Port 2020
[INPUT]
Name kafka
Tag oci_logs
Brokers <OCI_STREAMING_ENDPOINT>:9092
Topics oci_logs
Group_Id siem-shipper-v1
Format json
rdkafka.security.protocol SASL_SSL
rdkafka.sasl.mechanism PLAIN
rdkafka.sasl.username <TENANCY>/<USER>/<STREAMPOOL_OCID>
rdkafka.sasl.password <AUTH_TOKEN>
rdkafka.auto.offset.reset latest
[FILTER]
Name lua
Match oci_logs
Script /etc/fluent-bit/scripts/cef_wrapper.lua
Call wrap_cef
[OUTPUT]
Name syslog
Match oci_logs
Host 172.18.7.10
Port 6515
Mode tcp
syslog_message_key formatted_cef
Syslog_Format rfc3164
# Disable keepalive to prevent 'Connection Reset' by Trend Micro
net.keepalive off
TLS Off
3. Troubleshooting and Pipeline Architecture Validation
When running a high-throughput syslog pipeline, verifying packet ingest and processing health is critical. Use the following built-in Fluent Bit metrics endpoints to ensure your input records are balanced with successful output records.
To verify that proc_records matches kafka.0 records and that retries_failed remains at 0, query the API using:
curl -s http://127.0.0.1:2020/api/v1/metrics | jq '.output'
curl -s http://127.0.0.1:2020/api/v1/metrics | jq '.input'
4. Testing the Infrastructure Locally
Before launching into production, it is highly recommended to redirect outputs to stdout to confirm the Lua engine is flattening the data payloads correctly.
Open and edit your configuration parser file:
nano /etc/fluent-bit/fluent-bit.conf
Temporarily swap the active Output block to print directly to your terminal:
[OUTPUT]
Name stdout
Match oci_logs
Execute the Fluent Bit daemon manually to audit the console output:
sudo /opt/fluent-bit/bin/fluent-bit -c /etc/fluent-bit/fluent-bit.conf
With this setup complete, you now possess a reliable, robust, and highly structured pipeline. It smoothly processes high-velocity, nested security events from Oracle Cloud Infrastructure, transforming and deliverying them cleanly to your Trend Micro gateway for real-time security monitoring.
_