# LogRhythm Forwarder Building a forwarder from Apica Ascent to LogRhythm requires a slightly different approach than modern SaaS OTLP targets. While LogRhythm is transitioning toward JSON-native ingestion via its Open Collector, it does not yet provide a public, direct OTLP/HTTP endpoint for general log streaming like Grafana or Honeycomb. The most reliable "forwarder" approach involves using Apica to transform OTel data into a structured JSON format and delivering it to the LogRhythm Open Collector via Syslog or a Beat. ### 1. Prerequisites from LogRhythm To receive data, you must have the LogRhythm Open Collector installed and configured: 1. Open Collector IP/Hostname: The address of the VM running your Open Collector. 2. Syslog Beat / Generic JSON Beat: Ensure a "Beat" is configured to listen on a specific port (typically `5044` for Beats or `514/1514` for Syslog). 3. Log Source Type: In the LogRhythm Console, create a new Log Source and set the type to "System : JSON - Open Collector". ### 2. Configuration Strategy: The Forwarder Pipeline In the Apica Flow (Ascent) UI, you will configure a Syslog or Webhook forwarder that bridges OTel data into LogRhythm's JSON-native ingestion layer. #### Step A: Destination Settings | **Field** | **Value** | | ---------------- | ----------------------------------- | | Destination Name | `LogRhythm_SIEM_Forwarder` | | Endpoint | `` | | Protocol | `TCP` (Recommended for reliability) | | Port | `1514` (Default for JSON Syslog) | | Format | `JSON` | #### Step B: Metadata Enrichment (OTTL) LogRhythm's "Schema Dictionary" maps fields based on key names. To ensure your Apica logs are parsed automatically (auto-mapped), you should normalize your OTel attributes to match LogRhythm's expectations. SQL ``` # 1. Map OTel severity to LogRhythm's priority scale set(attributes["priority"], 5) where severity_number >= 13 set(attributes["priority"], 1) where severity_number <= 9 # 2. Extract the 'body' into a 'message' field (Standard for LR) set(attributes["message"], body) # 3. Add LogRhythm-specific metadata tags set(attributes["vendor"], "Apica") set(attributes["product"], "Ascent") set(attributes["log_source_id"], "Apica_OTel_Pipeline") # 4. Flatten high-cardinality OTel resource attributes set(attributes["hostname"], resource.attributes["host.name"]) ``` ### 3. Detailed Reference: OTel Collector Bridge If you are running a standalone OTel Collector managed by Apica to bridge data to LogRhythm, use the following configuration to output a LogRhythm-friendly stream: YAML ``` exporters: # Using the TCP exporter to send JSON logs to LogRhythm Open Collector tcp/logrhythm: endpoint: ":1514" # Convert OTel LogRecord into a single-line JSON object format: json service: pipelines: logs: receivers: [otlp] processors: [batch, transform/lr_mapping] exporters: [tcp/logrhythm] ``` ### 4. Key Implementation Notes * JSON Mapping: The LogRhythm Open Collector is designed to grab JSON fields. If you send nested OTel objects (e.g., `attributes.http.method`), ensure your Open Collector mapping file (the `.yaml` file for that specific Beat) is configured to flatten those fields or that you flatten them in Apica before sending. * Timestamping: LogRhythm uses the `normalised_time` field for its index. Ensure Apica is sending the `timestamp` in ISO8601 format to avoid "Time Out of Sync" errors in the LogRhythm SIEM. * Volume Control: Since SIEM licensing is often based on volume (Messages Per Second), use Apica's Flow Control to drop non-security related logs (like "Debug" or "Info") before they reach the Open Collector. ## Detailed Mapping Template and Policy Configuration To implement the forwarder approach for LogRhythm, you will primarily use the LogRhythm JSON Policy Builder (introduced in version 7.17) or a custom JQ-based Pipeline if you are on an older version. Below is the detailed mapping template and policy configuration designed specifically for the Apica OTel attributes we established (e.g., `priority`, `message`, `vendor`, `product`). ### 1. The Logic: JSON Policy Mapping In the LogRhythm Web Console (Settings > Open Collector > JSON Policy Builder), you will map the incoming JSON fields from Apica to the LogRhythm Schema Dictionary. | **Apica JSON Field (Source)** | **LogRhythm Schema Field (Target)** | **Logic / Reason** | | ----------------------------- | ----------------------------------- | ------------------------------------------- | | `attributes.priority` | `Priority` | Maps the calculated 1–5 scale from Apica. | | `attributes.message` | `Message` | The core log body text for indexing. | | `attributes.vendor` | `Vendor` | Set to `Apica` for parser identification. | | `attributes.product` | `Product` | Set to `Ascent` for product categorization. | | `attributes.hostname` | `Origin Host` | Essential for UEBA and entity linking. | | `attributes.log_source_id` | `Subject` | Identifies the specific Apica pipeline. | | `severity_text` | `Severity` | Maps "INFO", "WARN", "ERROR" etc. | | `timestamp` | `Normalised Time` | Must be ISO8601 for correct sequencing. | ### 2. The Template: Custom JQ Pipeline (`transform.jq`) If you are using the high-performance Open Collector Pipeline (CLI-based), use this JQ snippet to normalize the data before it hits the System Monitor. Code snippet ``` # Apica-to-LogRhythm Mapping Template # This JQ filter transforms OTel-standard JSON into LogRhythm Schema def transform: # 1. Initialize the LogRhythm IO format { "input": ., "output": { "original_message": . | tojson, "beatname": "apica_ascent_otel", "device_type": "apica_siem_forwarder" } } | # 2. Map core identity fields .output.vendor = .input.attributes.vendor | .output.product = .input.attributes.product | # 3. Map Security & Severity .output.priority = .input.attributes.priority | .output.severity = .input.severity_text | # 4. Map Message & Metadata .output.message = .input.attributes.message | .output.sname = .input.attributes.hostname | # Subject Name .output.subject = .input.attributes.log_source_id | # 5. Final output for Syslog delivery .output; ``` ### 3. Implementation Steps in LogRhythm Once you have the mapping defined, follow these steps to activate the forwarder: 1. Initialize the Beat: Configure a Generic Beat or Webhook Beat on your Open Collector to receive the TCP/HTTP stream from Apica. 2. Upload the Policy: If using the Web UI, upload the generated JSON Policy. If using the CLI, import the `.pipe` file: `cat apica_ascent.pipe | ./lrctl oc pipe import` 3. Log Source Virtualization (LSV): \* In the Client Console, go to Log Source Virtualization. * Create a new Virtual Log Source for "Apica" using the `device_type` defined in the JQ above (`apica_siem_forwarder`). * This ensures LogRhythm treats these logs as a unique source rather than generic Syslog. ### 4. Key Reference: Field Mapping Table Use this table as a reference for your MPE (Message Processing Engine) rules if you need to build custom sub-rules for specific Apica events. | **Apica Event Attribute** | **Suggested Common Event** | **SIEM Classification** | | -------------------------- | -------------------------- | ----------------------- | | `log.type: "auth_failure"` | `Authentication Failed` | Access Control | | `log.type: "system_alert"` | `System Warning` | Operations | | `log.type: "api_error"` | `Application Error` | Error | --- # Agent Instructions: Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter: ``` GET https://docs.apica.io/flow/list-of-forwarders/logrhythm-forwarder.md?ask= ``` The question should be specific, self-contained, and written in natural language. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.