Collecting own logs

Collecting SOC4Kafka own logs¶

Basic use case¶

To collect logs from SOC4Kafka, you can extend the configuration file to include a telemetry section under the service block. This section allows you to specify the logging level and output paths for the logs generated by the OpenTelemetry Collector itself. The logs can be directed to both a file and standard output (stdout) for easier monitoring and debugging.

Note

Ensure that the specified log file paths exist and are writable by the user running the collector.

In this example, we are directing the logs to a directory named soc4kafka-otel, which should be created beforehand. The directory is in the same location where the collector binary is executed.

mkdir -p ./soc4kafka-otel

Here is the telemetry section to be added to your configuration file:

service:
  telemetry:
    logs:
      level: info
      output_paths:
        - ./soc4kafka-otel/otel-collector.log
        - stdout
      error_output_paths:
        - ./soc4kafka-otel/otel-collector-errors.log
        - stderr

Here is a complete example configuration that includes the telemetry section for collecting SOC4Kafka logs:

receivers:
  kafka:
    brokers: ["kafka-broker-1:9092", "kafka-broker-2:9092"]
    logs:
      topics: 
        - example-topic
      encoding: text

processors:
  resourcedetection:
    detectors: ["system"]
    system:
      hostname_sources: ["os"]

exporters:
  splunk_hec:
    token: "your-splunk-hec-token"
    endpoint: "https://splunk-hec-endpoint:8088/services/collector"
    source: my-kafka
    sourcetype: kafka-otel
    index: kafka_otel
    splunk_app_name: "soc4kafka"
    sending_queue:
      enabled: true
      num_consumers: 10
      queue_size: 10000
      block_on_overflow: true
      sizer: items
      batch:
        min_size: 1000

service:
  telemetry:
    logs:
      level: info
      output_paths:
        - ./soc4kafka-otel/otel-collector.log
        - stdout
      error_output_paths:
        - ./soc4kafka-otel/otel-collector-errors.log
        - stderr
  pipelines:
    logs:
      receivers: [kafka]
      processors: [resourcedetection]
      exporters: [splunk_hec]

Advanced use case¶

Furthermore, you might want to monitor multiple SOC4Kafka instances running on different hosts. To achieve this, and reduce the number of ssh/remote sessions to different VMs, you can simply extend the above configuration by adding ability to forward your logs into Splunk using the filelog receiver. We are using also file_storage extension to keep track of the read position (checkpoint) in the log file, so in case of a restart, SOC4Kafka will continue reading from where it left off and not resend the whole log file again. Our example configuration below demonstrates how to set this up:

receivers:
  kafka:
    brokers: ["kafka-broker-1:9092", "kafka-broker-2:9092"]
    logs:
      topics: 
        - example-topic
      encoding: text

  filelog:
    include:
      - "./soc4kafka-otel/*.log"
    start_at: beginning
    storage: file_storage

processors:
  resourcedetection:
    detectors: ["system"]
    system:
      hostname_sources: ["os"]

exporters:
  splunk_hec:
    token: "your-splunk-hec-token"
    endpoint: "https://splunk-hec-endpoint:8088/services/collector"
    source: my-kafka
    sourcetype: kafka-otel
    index: kafka_otel
    splunk_app_name: "soc4kafka"
    sending_queue:
      enabled: true
      num_consumers: 10
      queue_size: 10000
      block_on_overflow: true
      sizer: items
      batch:
        min_size: 1000

  splunk_hec/internal_logs:
    token: "your-splunk-hec-token"
    endpoint: "https://splunk-hec-endpoint:8088/services/collector"
    index: kafka-logs
    splunk_app_name: "soc4kafka"
    sending_queue:
      enabled: true
      num_consumers: 10
      queue_size: 10000
      block_on_overflow: true
      sizer: items
      batch:
        min_size: 1000

extensions:
  file_storage:
    directory: "./soc4kafka-otel/checkpoint"
    create_directory: true

service:
  extensions: [file_storage]
  telemetry:
    logs:
      level: info
      output_paths:
        - ./soc4kafka-otel/otel-collector.log
        - stdout
      error_output_paths:
        - ./soc4kafka-otel/otel-collector-errors.log
        - stderr
  pipelines:
    logs:
      receivers: [kafka]
      processors: [resourcedetection]
      exporters: [splunk_hec]
    logs/internal:
      receivers: [filelog]
      processors: [resourcedetection]
      exporters: [splunk_hec/internal_logs]

Additionally, we have added a separate pipeline to forward the internal logs collected by the filelog receiver to Splunk using a dedicated exporter (splunk_hec/internal_logs). This gives us flexibility of forwarding the internal logs to a different index in Splunk, making it easier to separate and analyze them. In production environments with multiple SOC4Kafka instances, this setup allows for centralized log management and monitoring, enhancing operational efficiency and troubleshooting capabilities.

Advanced use case - extended configuration: timestamping collector log files¶

To make it easier to identify and manage log files, especially when dealing with multiple instances or frequent restarts, you can include timestamps in the log file names. This can be achieved by using environment variables in the configuration file. Below is an example of how to set this up: 1. Prepare the configuration file (values_timestamp.yaml) with ${TIMESTAMP} variable in the log file names:

receivers:
  kafka:
    brokers: ["kafka-broker-1:9092", "kafka-broker-2:9092"]
    logs:
      topics: 
        - test-v
      encoding: text

  filelog:
    include:
      - "./soc4kafka-otel/*.log"
    start_at: beginning
    storage: file_storage
    include_file_path: true
    include_file_name: false

processors:
  resourcedetection:
    detectors: ["system"]
    system:
      hostname_sources: ["os"]

exporters:
  splunk_hec:
    token: "your-splunk-hec-token"
    endpoint: "https://splunk-hec-endpoint:8088/services/collector"
    source: my-kafka
    sourcetype: kafka-otel
    index: kafka_otel
    splunk_app_name: "soc4kafka"
    sending_queue:
      enabled: true
      num_consumers: 10
      queue_size: 10000
      block_on_overflow: true
      sizer: items
      batch:
        min_size: 1000

  splunk_hec/internal_logs:
    token: "your-splunk-hec-token"
    endpoint: "https://splunk-hec-endpoint:8088/services/collector"
    index: kafka-logs
    splunk_app_name: "soc4kafka"
    sending_queue:
      enabled: true
      num_consumers: 10
      queue_size: 10000
      block_on_overflow: true
      sizer: items
      batch:
        min_size: 1000

extensions:
  file_storage:
    directory: "./soc4kafka-otel/checkpoint"
    create_directory: true

service:
  extensions: [file_storage]
  telemetry:
    logs:
      level: info
      output_paths:
        - ./soc4kafka-otel/otel-collector-${TIMESTAMP}.log
        - stdout
      error_output_paths:
        - ./soc4kafka-otel/otel-collector-errors-${TIMESTAMP}.log
        - stderr
  pipelines:
    logs:
      receivers: [kafka]
      processors: [resourcedetection]
      exporters: [splunk_hec]
    logs/internal:
      receivers: [filelog]
      processors: [resourcedetection]
      exporters: [splunk_hec/internal_logs]

Let’s use now simple bash script to export current timestamp as an environment variable and start the collector with a config file that uses this variable.

#!/bin/bash
# Create timestamp and export it
export TIMESTAMP=$(date +"%Y%m%d_%H%M%S")

# Create logs directory if it doesn't exist
mkdir -p ./soc4kafka-otel

# Start the collector with a config file that uses the TIMESTAMP variable
echo "Starting OpenTelemetry Collector with timestamp: $TIMESTAMP"
./<otel_package> --config values_timestamp.yaml

After running the above script, you should see log files with timestamps in their names, such as otel-collector-20231005_143200.log and otel-collector-errors-20231005_143200.log, making it easier to identify when each log file was created.