OTel Collector Agent Processing

Table of Contents

Overview
Log Processing
External References

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 Sensitive Data

Use the transform processor to replace sensitive values in the log body before the data leaves your environment. Each replace_pattern statement applies a regular expression to body and substitutes every match. Because this runs in the collector, the original value never reaches Kloudfuse or crosses the network.

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

processors:
  transform/mask_sensitive:
    log_statements:
      - context: log
        statements:
          # card security codes and track data — $$1$$2 echoes the key and separator so only the value is replaced.
          # key-anchored rules run before the broad card rule so an embedded PAN cannot fragment a keyed value.
          - replace_pattern(body, "(?i)\\b(cvv2?|cvc2?|cid|pin_?block|track[12]|service_code)([\"']?\\s*[:=]\\s*[\"']?)[;%]?[^\\s,;\"'&}]+", "$$1$$2[CSC REDACTED]")
            where IsString(body)   (1)
          # credentials and tokens
          - replace_pattern(body, "(?i)\\b(password|passwd|secret(?:_?key)?|api[_-]?key|x-api-key|auth(?:oriz(?:ation|ed))?_?token|access_?token|bearer)([\"']?\\s*[:=]\\s*[\"']?)[^\\s,;\"'&}]+", "$$1$$2[SECRET REDACTED]")
            where IsString(body)
          # card numbers — major brand prefixes, 13–19 digits, contiguous or grouped in any layout; runs last
          - replace_pattern(body, "\\b(?:4\\d{3}|5[1-5]\\d{2}|2(?:22[1-9]|2[3-9]\\d|[3-6]\\d\\d|7[01]\\d|720)|3[47]\\d{2}|30[0-5]\\d|3(?:09|[689]\\d)\\d|6(?:011|5\\d\\d|4[4-9]\\d|22\\d)|35(?:2[89]|[3-8]\\d)|62\\d\\d)(?:[ -]?\\d){9,15}\\b", "[CARD REDACTED]")
            where IsString(body)
          - replace_pattern(body, "\\b[A-Za-z0-9._%+\\-]+@[A-Za-z0-9.\\-]+\\.[A-Za-z]{2,}\\b", "***@***.***")
            where IsString(body)

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

service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [transform/mask_sensitive]
      exporters: [otlp]

yaml

1	`replace_pattern` on `body` is a silent no-op when the body is a map rather than a string. The `where IsString(body)` guard makes that explicit; for structured bodies use `replace_all_patterns(body, "value", …)` instead, which walks every value in the map.

The same statements work on span attributes in a trace_statements block — for example replace_all_patterns(attributes, "value", …) with context: span — so a card number in db.statement or an error message is masked too.

For a processor that walks string, map, and slice bodies without per-statement guards, the redaction processor accepts the same patterns under blocked_values. Set allow_all_keys: true with it, or it drops every attribute not on its allowlist.

Test every pattern against a sample of real log lines before deploying it. The card-number rule also matches other long numeric identifiers that begin with a valid card prefix, such as trace IDs and order numbers — safe for cardholder data, but it can break log-to-trace correlation. Scope the statement with a where clause on resource.attributes["service.name"] if only some services handle payments.

Masking matched card and credential values in the collector reduces the amount of sensitive data sent to the observability platform. Compliance scope still depends on the complete data flow and the requirements of the applicable framework. Metric attributes can be masked the same way with metric_statements in context: datapoint (see Drop High-Cardinality Labels for the pattern), and where you control the application the OpenTelemetry SDK can redact in-process before anything is emitted. For how collector-side masking fits with ingest authentication, access policies, and audit logging on the Kloudfuse side, see Keeping Sensitive Data Out of Kloudfuse.

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

Attribute key	Value
`level`	`error`
`msg`	`connection refused`
`service`	`payments`
`trace_id`	`abc123`

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

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

SEVERITY_NUMBER_TRACE

TRACE

SEVERITY_NUMBER_DEBUG

DEBUG

SEVERITY_NUMBER_INFO

INFO

SEVERITY_NUMBER_WARN

WARN

SEVERITY_NUMBER_ERROR

ERROR

SEVERITY_NUMBER_FATAL

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.

OTel Collector Agent Processing

Overview

Log Processing

Mask Sensitive Data

Add Attributes to Logs

Parse JSON Logs and Extract Fields as Attributes

Filter Out Unwanted Logs

Extract a Field and Set It as the Log Message Body

Create Metrics from Log Records

External References