OTel Collector Agent Processing

Overview

The OpenTelemetry Collector provides a set of processors and connectors that let you reshape telemetry data before it reaches Kloudfuse. This page covers common log processing patterns using the transform processor.

Replace <kloudfuse-hostname> with your Kloudfuse cluster hostname and <api-key> with your Kloudfuse API key.

Log Processing

Mask PII Data

Use the transform processor to replace sensitive values in the log body with a placeholder before the data leaves your environment:

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  transform/mask_pii:
    log_statements:
      - context: log
        statements:
          - replace_pattern(body, "password=[^\\s&]+", "password=***")
          - replace_pattern(body, "\\b[A-Za-z0-9._%+\\-]+@[A-Za-z0-9.\\-]+\\.[A-Za-z]{2,}\\b", "***@***.***")
          - replace_pattern(body, "\\b\\d{4}[\\s\\-]?\\d{4}[\\s\\-]?\\d{4}[\\s\\-]?\\d{4}\\b", "****-****-****-****")

exporters:
  otlp:
    endpoint: https://<kloudfuse-hostname>/ingester/otlp
    headers:
      Kf-Api-Key: <api-key>

service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [transform/mask_pii]
      exporters: [otlp]
yaml
Masking is applied as a regex substitution on the raw string value of body. Test patterns against representative log samples to avoid unintended replacements.

Add Attributes to Logs

Use the transform processor to attach static or dynamic attributes to every log record:

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  transform/add_labels:
    log_statements:
      - context: log
        statements:
          - set(attributes["environment"], "production")
          - set(attributes["team"], "platform")
          - set(attributes["region"], resource.attributes["cloud.region"])   (1)

exporters:
  otlp:
    endpoint: https://<kloudfuse-hostname>/ingester/otlp
    headers:
      Kf-Api-Key: <api-key>

service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [transform/add_labels]
      exporters: [otlp]
yaml
1 Copies a resource attribute onto each log record as a log-level attribute, making it available as a label in Kloudfuse log queries.

Parse JSON Logs and Extract Fields as Attributes

When a log body is a JSON string, the entire JSON blob is stored as a single opaque value. To make individual fields queryable in Kloudfuse, you need to parse the JSON and write each field as a separate log attribute.

ParseJSON(body) parses the JSON string in body into a map of key-value pairs. merge_maps(attributes, …​, "upsert") writes each of those key-value pairs into the log record’s attributes map, adding new keys and overwriting any that already exist.

For example, if a log record arrives with this body:

{"level":"error","msg":"connection refused","service":"payments","trace_id":"abc123"}
json

After the processor runs, the log record has these attributes:

Attribute key Value

level

error

msg

connection refused

service

payments

trace_id

abc123

Each attribute is then available as a label in Kloudfuse log queries and can be used for filtering, grouping, and alerting.

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  transform/parse_json:
    log_statements:
      - context: log
        statements:
          - merge_maps(attributes, ParseJSON(body), "upsert")
            where IsMatch(body, "^\\{")   (1)

exporters:
  otlp:
    endpoint: https://<kloudfuse-hostname>/ingester/otlp
    headers:
      Kf-Api-Key: <api-key>

service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [transform/parse_json]
      exporters: [otlp]
yaml
1 The where guard limits parsing to records whose body begins with {, leaving non-JSON logs unchanged and avoiding parse errors.

Filter Out Unwanted Logs

Use the filter processor to drop log records that add noise without value. Records matching any condition are dropped.

severity_number is a numeric field on every log record defined by the OpenTelemetry log data model. The OTTL constants such as SEVERITY_NUMBER_WARN are built-in named values for each level — they are not variables you define:

Constant Numeric value Equivalent level

SEVERITY_NUMBER_TRACE

1

TRACE

SEVERITY_NUMBER_DEBUG

5

DEBUG

SEVERITY_NUMBER_INFO

9

INFO

SEVERITY_NUMBER_WARN

13

WARN

SEVERITY_NUMBER_ERROR

17

ERROR

SEVERITY_NUMBER_FATAL

21

FATAL

The following example drops health check requests and drops below-WARN logs from a specific service:

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  filter/drop_noise:
    logs:
      log_record:
        - 'IsMatch(body, ".*GET /healthz.*")'
        - 'IsMatch(body, ".*GET /readyz.*")'
        - 'severity_number < SEVERITY_NUMBER_WARN and IsMatch(attributes["service.name"], ".*load-balancer.*")'   (1)

exporters:
  otlp:
    endpoint: https://<kloudfuse-hostname>/ingester/otlp
    headers:
      Kf-Api-Key: <api-key>

service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [filter/drop_noise]
      exporters: [otlp]
yaml
1 Drops any log record from load-balancer services with a severity number below 13 (WARN). Records matching any condition in the list are dropped.

To keep only WARN and above from all services, drop everything with a severity number below 13:

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  filter/warn_and_above:
    logs:
      log_record:
        - 'severity_number < SEVERITY_NUMBER_WARN'   (1)

exporters:
  otlp:
    endpoint: https://<kloudfuse-hostname>/ingester/otlp
    headers:
      Kf-Api-Key: <api-key>

service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [filter/warn_and_above]
      exporters: [otlp]
yaml
1 Drops TRACE (1–4), DEBUG (5–8), and INFO (9–12) records. WARN (13+), ERROR (17+), and FATAL (21+) records are forwarded.

Extract a Field and Set It as the Log Message Body

Use ExtractPatterns to pull a value out of a structured log line and promote it to the log body:

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  transform/extract_message:
    log_statements:
      - context: log
        statements:
          - set(body, ExtractPatterns(body, "msg=(?P<msg>[^\\s]+)")["msg"])
            where IsMatch(body, "msg=")   (1)

exporters:
  otlp:
    endpoint: https://<kloudfuse-hostname>/ingester/otlp
    headers:
      Kf-Api-Key: <api-key>

service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [transform/extract_message]
      exporters: [otlp]
yaml
1 Replaces the full log line with just the extracted msg value. Adjust the capture group name and regex to match your log format.

Create Metrics from Log Records

Use the count connector to generate a metric that counts log records matching a condition. The connector bridges the logs pipeline to the metrics pipeline, appearing as an exporter in logs and a receiver in metrics:

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  transform/parse_json:
    log_statements:
      - context: log
        statements:
          - merge_maps(attributes, ParseJSON(body), "upsert")
            where IsMatch(body, "^\\{")

connectors:
  count:
    logs:
      log.error.count:
        description: "Count of error and above log records per service"
        conditions:
          - 'severity_number >= SEVERITY_NUMBER_ERROR'
        attributes:
          - key: service.name
          - key: environment

exporters:
  otlp:
    endpoint: https://<kloudfuse-hostname>/ingester/otlp
    headers:
      Kf-Api-Key: <api-key>

service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [transform/parse_json]
      exporters: [count]    (1)
    metrics:
      receivers: [count]    (2)
      processors: []
      exporters: [otlp]
yaml
1 The count connector appears as an exporter in the logs pipeline.
2 The same count connector appears as a receiver in the metrics pipeline.

External References