OpenTelemetry as a Security Data Pipeline

Most people think of OpenTelemetry as just another observability tool. I’ve been exploring it for more than that in projects, as a flexible data pipeline for basic security monitoring.

Standard Setup with Helm

Getting started is straightforward with the official Helm chart:

helm repo add open-telemetry https://open-telemetry.github.io/opentelemetry-helm-charts
helm install otel-collector open-telemetry/opentelemetry-collector

For Kubernetes, use the official Helm chart with values overrides for security monitoring:

# Install the collector as a DaemonSet for node access
helm install otel-collector open-telemetry/opentelemetry-collector \
  --namespace observability \
  --create-namespace \
  --set mode=daemonset \
  --set image.repository=otel/opentelemetry-collector-contrib \
  --set image.tag=0.140.1

For security monitoring, you’ll need custom values. Create a values.yaml with just the overrides needed:

# Deploy as DaemonSet to access node logs
mode: daemonset

# Use contrib image for additional receivers
image:
  repository: otel/opentelemetry-collector-contrib
  tag: "0.140.1"

# Override collector config for security monitoring
config:
  receivers:
    # Container logs for application security events
    filelog/containers:
      include:
      - /var/log/pods/*/*/*.log
      exclude:
      - /var/log/pods/kube-system_*/*/*.log
      start_at: end
      operators:
      - type: router
        routes:
        - output: parse_docker
          expr: 'body matches "^{"'
        - output: parse_containerd
          expr: 'body matches "^[^ ]+ stdout"'
      - type: json_parser
        output: parse_docker
        timestamp:
          parse_from: attributes.time
          layout: '%Y-%m-%dT%H:%M:%S.%fZ'
      - type: regex_parser
        output: parse_containerd
        regex: '^(?P<time>[^ ]+) (?P<stream>stdout|stderr) (?P<logtag>[^ ]*) (?P<message>.*)'
        timestamp:
          parse_from: attributes.time
          layout: '%Y-%m-%dT%H:%M:%S.%fZ'

    # Node-level authentication logs
    filelog/auth:
      include:
      - /var/log/auth.log
      - /var/log/secure
      start_at: end
      operators:
      - type: regex_parser
        regex: '^(?P<timestamp>\w+\s+\d+\s+[\d:]+)\s+(?P<hostname>\S+)\s+(?P<process>\S+):\s+(?P<message>.*)'
        timestamp:
          parse_from: attributes.timestamp
          layout: '%b %d %H:%M:%S'

    # Kubernetes audit logs
    filelog/k8s_audit:
      include:
      - /var/log/kubernetes/audit.log
      - /var/log/audit/audit.log
      start_at: end
      operators:
      - type: json_parser

    # Kubernetes events
    k8sobjects/events:
      auth_type: serviceAccount
      objects:
      - name: events
        mode: watch
        group: ""
        exclude:
        - name: events
          filter:
            fields:
              - key: type
                value: Normal

    # Cluster metrics
    k8s_cluster:
      auth_type: serviceAccount
      collect_cluster_metrics: true

  processors:
    filter/security:
      logs:
        log_record:
        - 'IsMatch(attributes["k8s.namespace.name"], "kube-system") and severity_number < SEVERITY_NUMBER_WARN'
        - 'IsMatch(body, "(?i)healthcheck|readiness|liveness")'

    attributes/tagging:
      actions:
      - key: pipeline
        value: security
        action: insert

    transform/risk_scoring:
      log_statements:
      - context: log
        statements:
        - set(attributes["risk.level"], "high") where IsMatch(body, "(?i)(failed password|invalid user|connection closed)")

  exporters:
    otlphttp/siem:
      endpoint: https://siem.example.com:4318
      headers:
        Authorization: Bearer ${env:SIEM_TOKEN}

    prometheus:
      endpoint: "0.0.0.0:8889"
      namespace: security_metrics

  service:
    pipelines:
      logs/security:
        receivers: [filelog/containers, filelog/auth, filelog/k8s_audit, k8sobjects/events]
        processors: [filter/security, attributes/tagging, transform/risk_scoring, batch]
        exporters: [otlphttp/siem]
      metrics/cluster:
        receivers: [k8s_cluster]
        processors: [batch]
        exporters: [prometheus]

# Mount host volumes for log access
hostPathMounts:
  - name: varlog
    mountPath: /var/log
    hostPath: /var/log
    readOnly: true
  - name: varlibdockercontainers
    mountPath: /var/lib/docker/containers
    hostPath: /var/lib/docker/containers
    readOnly: true
  - name: auditlog
    mountPath: /var/log/kubernetes
    hostPath: /var/log/kubernetes
    readOnly: true

# Additional RBAC rules for security monitoring (default has basic rules)
rbac:
  rules:
  - apiGroups: [""]
    resources: ["events", "pods", "nodes"]
    verbs: ["get", "list", "watch"]
  - apiGroups: ["apps"]
    resources: ["replicasets"]
    verbs: ["get", "list", "watch"]

# Resource limits for security workload
resources:
  limits:
    cpu: 200m
    memory: 200Mi
  requests:
    cpu: 100m
    memory: 100Mi

Deploy with custom values:

helm install otel-collector open-telemetry/opentelemetry-collector \
  --namespace observability \
  --create-namespace \
  --values values.yaml

Note: The Helm chart handles ServiceAccount, ClusterRole, and proper RBAC automatically. You may need to adjust audit log paths based on your Kubernetes distribution.

Beyond Traditional Observability

The OpenTelemetry Collector offers a vendor-agnostic implementation for receiving, processing, and exporting telemetry data. This architecture makes it suitable for security event collection.

Security Log Collection

The filelog receiver tails log files and parses them into OpenTelemetry log format. For Kubernetes security monitoring, there are several key log sources:

Container Logs:

receivers:
  filelog/containers:
    include:
    - /var/log/pods/*/*/*.log
    exclude:
    - /var/log/pods/kube-system_*/*/*.log
    start_at: end
    operators:
    - type: router
      routes:
      - output: parse_docker
        expr: 'body matches "^{"'
      - output: parse_containerd
        expr: 'body matches "^[^ ]+ stdout"'
    - type: json_parser
      output: parse_docker
      timestamp:
        parse_from: attributes.time
        layout: '%Y-%m-%dT%H:%M:%S.%fZ'
    - type: regex_parser
      output: parse_containerd
      regex: '^(?P<time>[^ ]+) (?P<stream>stdout|stderr) (?P<logtag>[^ ]*) (?P<message>.*)'
      timestamp:
        parse_from: attributes.time
        layout: '%Y-%m-%dT%H:%M:%S.%fZ'

Node-level Authentication Logs:

receivers:
  filelog/auth:
    include:
    - /var/log/auth.log
    - /var/log/secure
    start_at: end
    operators:
    - type: regex_parser
      regex: '^(?P<timestamp>\w+\s+\d+\s+[\d:]+)\s+(?P<hostname>\S+)\s+(?P<process>\S+):\s+(?P<message>.*)'
      timestamp:
        parse_from: attributes.timestamp
        layout: '%b %d %H:%M:%S'
    - type: add
      field: attributes.log_type
      value: security

Kubernetes Audit Log Collection

For Kubernetes security monitoring, the audit log is the authoritative source. Note that audit logging must be enabled on the API server:

receivers:
  filelog/k8s_audit:
    include:
    - /var/log/kubernetes/audit.log
    - /var/log/audit/audit.log
    start_at: end
    operators:
    - type: json_parser
      timestamp:
        parse_from: attributes.stageTimestamp
        layout: '%Y-%m-%dT%H:%M:%S.%LZ'

This captures all API server activity including: - RBAC decisions - Resource creation/deletion - Authentication attempts - Admission controller decisions

Note: The audit log path depends on your Kubernetes distribution. You may need to mount these files into the collector DaemonSet with proper permissions.

Kubernetes Events Monitoring

The k8sobjects receiver watches Kubernetes objects as logs. For security monitoring, watch Events which capture cluster activity:

receivers:
  k8sobjects/events:
    auth_type: serviceAccount
    objects:
    - name: events
      mode: watch
      group: ""

processors:
  filter/warning_events_only:
    logs:
      log_record:
      - 'attributes["k8s.event.type"] == "Normal"'

Kubernetes Cluster Receiver

For cluster-level metrics and metadata:

receivers:
  k8s_cluster:
    auth_type: serviceAccount
    collect_cluster_metrics: true
    node_conditions_to_report: [Ready, MemoryPressure, PIDPressure, DiskPressure, NetworkUnavailable]
    allocation_to_collect:
    - cpu
    - memory

Network Traffic Analysis

The hostmetrics receiver provides network connection data:

receivers:
  hostmetrics:
    collection_interval: 30s
    scrapers:
      network:
        include:
          devices: ['eth0', 'ens5']
        metrics:
          system.network.connections:
            enabled: true
          system.network.io:
            enabled: true
          system.network.packets:
            enabled: true

Processing Security Data

Filtering with OTTL

The filter processor uses OpenTelemetry Transformation Language (OTTL) to drop or keep telemetry based on conditions.

To keep only security-relevant logs:

processors:
  filter/security_logs:
    logs:
      log_record:
      - 'not IsMatch(body, "(?i)(authentication|authorization|denied|failed|error|warning)")'

To drop debug-level noise:

processors:
  filter/drop_debug:
    logs:
      log_record:
      - 'severity_number < SEVERITY_NUMBER_INFO'

Adding Security Context with Attributes Processor

The attributes processor modifies attributes on telemetry data:

processors:
  attributes/security_enrichment:
    actions:
    - key: security.pipeline
      value: kubernetes_audit
      action: insert
    - key: environment
      value: production
      action: insert

Resource Attribution

Group telemetry by Kubernetes metadata using the resource processor:

processors:
  resource/k8s:
    attributes:
    - key: k8s.cluster.name
      value: production-sg
      action: upsert
    - key: deployment.environment
      value: production
      action: upsert

Transform Processor for Security Tagging

The transform processor uses OTTL statements to modify telemetry:

processors:
  transform/security_tags:
    log_statements:
    - context: log
      statements:
      - set(attributes["security.severity"], "high") where IsMatch(body, "(?i)authentication failure")
      - set(attributes["security.severity"], "critical") where IsMatch(body, "(?i)privilege escalation")
      - set(attributes["security.event_type"], "brute_force") where IsMatch(body, "(?i)repeated authentication failures")

Grouping by Attributes

The groupbyattrs processor groups metrics, logs, and spans with matching attributes under the corresponding Resource:

processors:
  groupbyattrs:
    keys:
    - k8s.namespace.name
    - k8s.pod.name
    - k8s.container.name

Complete Security Pipeline Configuration

receivers:
  # Container logs for application security events
  filelog/containers:
    include:
    - /var/log/pods/*/*/*.log
    exclude:
    - /var/log/pods/kube-system_*/*/*.log
    start_at: end
    operators:
    - type: router
      routes:
      - output: parse_docker
        expr: 'body matches "^{"'
      - output: parse_containerd
        expr: 'body matches "^[^ ]+ stdout"'
    - type: json_parser
      output: parse_docker
      timestamp:
        parse_from: attributes.time
        layout: '%Y-%m-%dT%H:%M:%S.%fZ'
    - type: regex_parser
      output: parse_containerd
      regex: '^(?P<time>[^ ]+) (?P<stream>stdout|stderr) (?P<logtag>[^ ]*) (?P<message>.*)'
      timestamp:
        parse_from: attributes.time
        layout: '%Y-%m-%dT%H:%M:%S.%fZ'

  # Node-level authentication logs
  filelog/auth:
    include:
    - /var/log/auth.log
    - /var/log/secure
    start_at: end
    operators:
    - type: regex_parser
      regex: '^(?P<timestamp>\w+\s+\d+\s+[\d:]+)\s+(?P<hostname>\S+)\s+sshd\[(?P<pid>\d+)\]:\s+(?P<message>.*)'
      timestamp:
        parse_from: attributes.timestamp
        layout: '%b %d %H:%M:%S'

  # Kubernetes audit logs
  filelog/k8s_audit:
    include:
    - /var/log/kubernetes/audit.log
    - /var/log/audit/audit.log
    start_at: end
    operators:
    - type: json_parser
      timestamp:
        parse_from: attributes.stageTimestamp
        layout: '%Y-%m-%dT%H:%M:%S.%LZ'

  # Kubernetes events
  k8sobjects/events:
    auth_type: serviceAccount
    objects:
    - name: events
      mode: watch
      group: ""
      exclude:
      - name: events
        filter:
          fields:
            - key: type
              value: Normal

  # Cluster metrics
  k8s_cluster:
    auth_type: serviceAccount
    collect_cluster_metrics: true

  # Host metrics for network monitoring
  hostmetrics:
    collection_interval: 60s
    scrapers:
      network:
        metrics:
          system.network.connections:
            enabled: true

processors:
  # Filter out non-security events
  filter/security:
    logs:
      log_record:
      - 'IsMatch(attributes["k8s.namespace.name"], "kube-system") and severity_number < SEVERITY_NUMBER_WARN'
      - 'IsMatch(body, "(?i)healthcheck|readiness|liveness")'

  # Add security metadata
  attributes/tagging:
    actions:
    - key: pipeline
      value: security
      action: insert
    - key: environment
      value: production
      action: insert

  # Risk scoring for security events
  transform/risk_scoring:
    log_statements:
    - context: log
      statements:
      - set(attributes["risk.level"], "high") where IsMatch(body, "(?i)(failed password|invalid user|connection closed)")
      - set(attributes["risk.level"], "critical") where IsMatch(body, "(?i)(root login|sudo.*FAILED|authentication failure)")

  # Batch processing
  batch:
    timeout: 10s
    send_batch_size: 1000

exporters:
  # SIEM integration
  otlphttp/siem:
    endpoint: https://siem.example.com:4318
    headers:
      Authorization: Bearer ${env:SIEM_TOKEN}

  # Metrics for Prometheus
  prometheus:
    endpoint: "0.0.0.0:8889"
    namespace: security_metrics

  # Debug output
  debug:
    verbosity: basic

service:
  pipelines:
    logs/security:
      receivers: [filelog/containers, filelog/auth, filelog/k8s_audit, k8sobjects/events]
      processors: [filter/security, attributes/tagging, transform/risk_scoring, batch]
      exporters: [otlphttp/siem, debug]
    metrics/cluster:
      receivers: [k8s_cluster, hostmetrics]
      processors: [batch]
      exporters: [prometheus]

Exporting to Security Tools

SIEM Integration

Send processed security logs to your SIEM:

exporters:
  otlphttp:
    endpoint: https://siem.example.com:4318
    tls:
      cert_file: /certs/client.crt
      key_file: /certs/client.key
      ca_file: /certs/ca.crt

Prometheus for Security Metrics

Expose metrics for Prometheus scraping:

exporters:
  prometheus:
    endpoint: "0.0.0.0:8889"
    namespace: security
    const_labels:
      cluster: production

Alerting is handled downstream in Prometheus/Alertmanager:

# alertmanager rules (separate from OTel config)
groups:
- name: security
  rules:
  - alert: HighFailedSSHAttempts
    expr: rate(security_failed_ssh_total[5m]) > 10
    for: 2m
    labels:
      severity: warning

Multiple Destinations

The same data can be exported to multiple backends simultaneously:

exporters:
  otlphttp/siem:
    endpoint: https://siem.example.com:4318
  otlphttp/backup:
    endpoint: https://backup-siem.example.com:4318
  file/compliance:
    path: /var/log/otel/security.json
    rotation:
      max_megabytes: 100
      max_days: 365

service:
  pipelines:
    logs/security:
      receivers: [filelog/auth]
      processors: [batch]
      exporters: [otlphttp/siem, otlphttp/backup, file/compliance]

Data Pipeline Benefits

Benefit Description
Unified Pipeline Single infrastructure for observability and security
Vendor Neutral Works with any OTLP-compatible backend
Protocol Agnostic Ingests syslog, JSON, OTLP, Prometheus formats
Cloud Native Built for Kubernetes environments
Real-time Stream processing with configurable batching

Challenges and Mitigations

Data Volume

Security events generate significant volume. Use filtering aggressively:

processors:
  filter/noise_reduction:
    logs:
      log_record:
      - 'IsMatch(body, "(?i)healthcheck")'
      - 'IsMatch(body, "(?i)readiness probe")'
      - 'attributes["k8s.namespace.name"] == "kube-system" and severity_number < SEVERITY_NUMBER_WARN'

Storage Costs

Implement tiered retention through your SIEM/storage backend, not in the collector. The collector is a transit layer.

False Positives

Handle alert tuning in your downstream alerting system (Prometheus Alertmanager, SIEM correlation rules). The collector’s role is collection and enrichment, not alert decision-making.

Security of the Collector Itself

The collector handles sensitive data. Follow security best practices:

# Bind to pod IP only, not 0.0.0.0
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: ${env:POD_IP}:4317
      http:
        endpoint: ${env:POD_IP}:4318

Additional Configurations for Security Use Cases

1. Redact Sensitive Data

Security logs may contain sensitive data (IPs, usernames, tokens). This type of processor can scrub this in the logs before sending it out from your environment:

processors:
  attributes/redact:
    actions:
    - key: user.password
      action: delete
    - key: http.request.header.authorization
      action: delete
    - key: user.email
      action: hash
    - key: source.ip
      pattern: '^(?P<first>\d+)\.(?P<second>\d+)\.(?P<third>\d+)\.(?P<fourth>\d+)'
      action: extract
    - key: source.ip
      value: "${first}.${second}.xxx.xxx"
      action: update

Or use the transform processor for more complex redaction:

processors:
  transform/redact:
    log_statements:
    - context: log
      statements:
      - replace_pattern(body, "password=\\S+", "password=REDACTED")
      - replace_pattern(body, "Bearer [A-Za-z0-9\\-._~+/]+=*", "Bearer REDACTED")
      - replace_pattern(body, "\\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Z|a-z]{2,}\\b", "EMAIL_REDACTED")

2. Kubernetes Attributes Processor

Automatically enrich logs with pod/namespace metadata for security correlation:

processors:
  k8sattributes:
    auth_type: serviceAccount
    extract:
      metadata:
      - k8s.namespace.name
      - k8s.pod.name
      - k8s.pod.uid
      - k8s.deployment.name
      - k8s.node.name
      labels:
      - tag_name: app
        key: app.kubernetes.io/name
      annotations:
      - tag_name: owner
        key: owner
    pod_association:
    - sources:
      - from: resource_attribute
        name: k8s.pod.ip

This automatically tags every log with its Kubernetes origin, critical for tracing security events to specific workloads.

3. Memory Limiter Processor

Security pipelines can spike during incidents. Protect the collector from OOM:

processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 400
    spike_limit_mib: 100

Place this first in your processor chain:

service:
  pipelines:
    logs/security:
      receivers: [filelog/auth]
      processors: [memory_limiter, filter/security, batch]  # memory_limiter first
      exporters: [otlphttp/siem]

4. Syslog Receiver for Network Devices

Collect logs from firewalls, load balancers, and network infrastructure. This would allow you to have all the capabilities of the pipeline use syslog based data sources:

receivers:
  syslog:
    tcp:
      listen_address: "0.0.0.0:54526"
    protocol: rfc5424
    operators:
    - type: add
      field: attributes.log_source
      value: network_infrastructure

5. Count Connector for Security Metrics

Convert log events into metrics for alerting/dashboards:

connectors:
  count:
    logs:
      security.failed_logins:
        description: Count of failed login attempts
        conditions:
        - 'IsMatch(body, "(?i)failed password")'
        attributes:
        - key: k8s.namespace.name
        - key: source.ip
      security.privilege_escalation:
        description: Count of privilege escalation attempts
        conditions:
        - 'IsMatch(body, "(?i)(sudo|su|privilege)")'

service:
  pipelines:
    logs/security:
      receivers: [filelog/auth]
      processors: [batch]
      exporters: [count]
    metrics/security_counts:
      receivers: [count]
      processors: [batch]
      exporters: [prometheus]

This enables Prometheus alerting on security_failed_logins_total without external log aggregation.

6. Routing Connector

Route events to different pipelines based on attributes / severity / or other characteristics:

receivers:
    otlp:

exporters:
  file/other:
    path: ./other.log
  file/east:
    path: ./east.log
  file/west:
    path: ./west.log

connectors:
  routing:
    default_pipelines: [logs/other]
    table:
      - context: log
        condition: attributes["region"] == "east"
        pipelines: [logs/east]
      - context: log
        condition: attributes["region"] == "west"
        pipelines: [logs/west]

service:
  pipelines:
    logs/in:
      receivers: [otlp]
      exporters: [routing]
    logs/east:
      receivers: [routing]
      exporters: [file/east]
    logs/west:
      receivers: [routing]
      exporters: [file/west]
    logs/other:
      receivers: [routing]
      exporters: [file/other]

7. Persistent Queue for Reliability

Security logs must not be lost. Enable persistent queues in case of network instability:

exporters:
  otlphttp/siem:
    endpoint: https://siem.example.com:4318
    sending_queue:
      enabled: true
      storage: file_storage
    retry_on_failure:
      enabled: true
      initial_interval: 5s
      max_interval: 30s
      max_elapsed_time: 300s

extensions:
  file_storage:
    directory: /var/lib/otelcol/queue
    timeout: 10s

service:
  extensions: [file_storage]

Limitations

OpenTelemetry Collector is not a replacement for dedicated security tools:

Conclusion

OpenTelemetry provides a solid foundation for security log collection and enrichment in Kubernetes environments. Its strength lies in unified data collection, vendor-neutral exports, and flexible transformation capabilities. The collector handles the data pipeline; security analysis and alerting belong in downstream systems purpose-built for those functions.

Published on by Stanley