OTel Collector Agent Processing

Table of Contents

Overview
Log Processing
Metric 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 and metric processing patterns using the transform processor, filter processor, groupbyattrs processor, and count connector.

Each example shows a complete collector configuration including receivers, processors, exporters, and the service pipeline. 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

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.

Metric Processing

Drop Labels and Aggregate Metrics

Use the transform processor to remove attributes from metric datapoints, then use groupbyattrs to re-aggregate datapoints that now share the same attribute set:

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

processors:
  transform/drop_labels:
    metric_statements:
      - context: datapoint
        statements:
          - delete_key(attributes, "pod_id")
          - delete_key(attributes, "container_id")

  groupbyattrs/aggregate:
    keys:
      - service.name
      - environment

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

service:
  pipelines:
    metrics:
      receivers: [otlp]
      processors: [transform/drop_labels, groupbyattrs/aggregate]   (1)
      exporters: [otlp]

yaml

1	Order matters: `transform/drop_labels` runs first to remove the attributes, then `groupbyattrs/aggregate` sums the datapoints that now share the same key set.

Drop High-Cardinality Labels

Remove attributes that create an unbounded number of time series — such as request IDs, user IDs, or raw URLs — before they reach Kloudfuse:

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

processors:
  transform/drop_high_cardinality:
    metric_statements:
      - context: datapoint
        statements:
          - delete_key(attributes, "request_id")
          - delete_key(attributes, "user_id")
          - replace_pattern(attributes["http.target"], "\\?.*$", "")   (1)

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

service:
  pipelines:
    metrics:
      receivers: [otlp]
      processors: [transform/drop_high_cardinality]
      exporters: [otlp]

yaml

1	Strips query strings from the HTTP target attribute to normalize the label value and reduce cardinality.

Rename Metrics

Use the transform processor to rename metrics to a new name:

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

processors:
  transform/rename_metrics:
    metric_statements:
      - context: metric
        statements:
          - set(name, "http.server.request.duration") where name == "http_server_duration"
          - set(name, "process.cpu.utilization") where name == "process_cpu_usage"

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

service:
  pipelines:
    metrics:
      receivers: [otlp]
      processors: [transform/rename_metrics]
      exporters: [otlp]

yaml

Add Metrics Together to Create a New Metric

Rename both source metrics to the target name, then use groupbyattrs to sum the datapoints that now share the same attribute set into a single time series:

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

processors:
  transform/combine_metrics:
    metric_statements:
      - context: metric
        statements:
          - set(name, "http.requests.total")
            where name == "http.read.requests" or name == "http.write.requests"   (1)

  groupbyattrs/sum_combined:
    keys:
      - service.name
      - environment

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

service:
  pipelines:
    metrics:
      receivers: [otlp]
      processors: [transform/combine_metrics, groupbyattrs/sum_combined]
      exporters: [otlp]

yaml

1	Both source metrics are renamed to `http.requests.total`. `groupbyattrs` then sums the datapoints that share the same attribute set into a single combined time series.

This approach merges the original metrics into the new name. If you need to preserve the originals, use the routing connector to duplicate them through separate pipelines before renaming.

Drop Metrics by Name

Use the filter processor to drop metrics before they reach Kloudfuse. Each entry under metrics.metric: is a drop condition — any metric that matches is dropped and not forwarded. This is different from the transform processor, which modifies metrics rather than removing them.

To drop by exact name:

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

processors:
  filter/drop_by_name:
    metrics:
      metric:
        - 'name == "go.gc.duration"'
        - 'name == "process.runtime.go.mem.heap_alloc"'

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

service:
  pipelines:
    metrics:
      receivers: [otlp]
      processors: [filter/drop_by_name]
      exporters: [otlp]

yaml

To drop all metrics whose names match a regex pattern:

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

processors:
  filter/drop_by_regex:
    metrics:
      metric:
        - 'IsMatch(name, "^go\\.runtime\\..*")'    (1)
        - 'IsMatch(name, "^process\\.runtime\\..*")'

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

service:
  pipelines:
    metrics:
      receivers: [otlp]
      processors: [filter/drop_by_regex]
      exporters: [otlp]

yaml

1	Dots in metric names must be escaped as `\\.` in the regex because `.` matches any character.

To keep only metrics whose names match a pattern and drop everything else, use not to invert the drop condition. Because every entry under metrics.metric: is a drop condition, not IsMatch(…) means "drop any metric that does not match" — which has the effect of keeping only the metrics that do match:

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

processors:
  filter/keep_by_regex:
    metrics:
      metric:
        - 'not IsMatch(name, "^http\\..*")'    (1)

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

service:
  pipelines:
    metrics:
      receivers: [otlp]
      processors: [filter/keep_by_regex]
      exporters: [otlp]

yaml

1	Drops any metric whose name does not start with `http.`. Only `http.*` metrics are forwarded to Kloudfuse.

OTel Collector Agent Processing

Overview

Log Processing

Mask PII 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

Metric Processing

Drop Labels and Aggregate Metrics

Drop High-Cardinality Labels

Rename Metrics

Add Metrics Together to Create a New Metric

Drop Metrics by Name

External References