Provide OOTB mapping (with OTLP) for incoming Opentelemetry logs to outgoing Opentelemetry logs

[someStrangerFromTheAbyss]

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

Currently, if you setup a Vector.dev to receive logs with the Opentelemetry Source then send them with Opentelemetry Sink whitout any mapping or changes to these logs, your logs does not match the Opentelemetry format. The way i found out about it was when trying to use the Loki Otlp endpoint. None of the attributes/Ressource_attributes were not detected since the logs are not in a RessourceLogs array.

You have to do a remap of these logs for it to corresponds to the Opentelemetry format.

Currently, the best way i got around this problem is to have a vrl remap with the best of my opentelemetry/vector knowledge. This is what i came up with right now:

attributes_array = []
resource_attributes_array = []
# for each: https://vector.dev/docs/reference/vrl/examples/#for_each
for_each(object!(.attributes)) -> |key,value|{
    insert_key = {}
    insert_key.key = key
    insert_key.value = {}
    # The attribute value is either:
    # A primitive type: string, boolean, double precision floating point (IEEE 754-1985) or signed 64 bit integer.
    # An array of primitive type values. The array MUST be homogeneous, i.e., it MUST NOT contain values of different types.
    # Example of type: https://github.com/open-telemetry/opentelemetry-proto/blob/d7770822d70c7bd47a6891fc9faacc66fc4af3d3/examples/logs.json#L9
    if is_string(value) {
        insert_key.value.stringValue = value
    } else if is_boolean(value) {
        insert_key.value.boolValue = value
    } else if is_integer(value){
        insert_key.value.intValue = value
    } else if is_float(value){
        insert_key.value.doubleValue = value
    } else {
        # in such a case, we try to transform to string then, insert value has string. If cannot be string, simply put empty.
        insert_key.value.stringValue = string(value) ?? ""
    }
    attributes_array = push(attributes_array,insert_key)
}
for_each(object!(.resources)) -> |key,value|{
    insert_key = {}
    insert_key.key = key
    insert_key.value = {}
    # The attribute value is either:
    # A primitive type: string, boolean, double precision floating point (IEEE 754-1985) or signed 64 bit integer.
    # An array of primitive type values. The array MUST be homogeneous, i.e., it MUST NOT contain values of different types.
    # Example of type: https://github.com/open-telemetry/opentelemetry-proto/blob/d7770822d70c7bd47a6891fc9faacc66fc4af3d3/examples/logs.json#L9
    if is_string(value) {
        insert_key.value.stringValue = value
    } else if is_boolean(value) {
        insert_key.value.boolValue = value
    } else if is_integer(value){
        insert_key.value.intValue = value
    } else if is_float(value){
        insert_key.value.doubleValue = value
    } else {
        # in such a case, we try to transform to string then, insert value has string. If cannot be string, simply put empty.
        insert_key.value.stringValue = string(value) ?? ""
    }
    resource_attributes_array = push(resource_attributes_array,insert_key)
}
#map_values(object!(.attributes)) -> |value| { upcase!(value) }
.resourceLogs = [{
  "resource": {
    "attributes": resource_attributes_array
  },
  "scopeLogs": [{
    "scope": object(.scope) ?? {},
    "logRecords": [{
      "timeUnixNano": to_unix_timestamp!(.timestamp, unit: "nanoseconds"),
      "body": { "stringValue": .message },
      "severityText": .severity_text,
      "severityNumber": .severity_number,
      "attributes": attributes_array,
      "droppedAttributesCount": .dropped_attributes_count,
      "traceId": string(.traceId) ?? "",
      "spanId": string(.spanId) ?? ""
    }]
  }]
}]
del(.severity_number)
del(.severity_text)
del(.resources)
del(.attributes)
del(.message)
del(.timestamp)
del(.dropped_attributes_count)
del(.observed_timestamp)
del(.source_type)
del(.scope)

Any suggestion is appreciated.

Configuration

sources:
  open_telemetry:
    grpc:
      address: 0.0.0.0:4320
    http:
      address: 0.0.0.0:4321
    type: opentelemetry
transforms:
  remap_opentelemetry:
    type: remap
    inputs: [open_telemetry.logs]
    file: "/home/charles/vector/otel_remap.vrl"
sinks:
  debug_console:
    type: console
    encoding:
      codec: json
    inputs: 
    - remap_opentelemetry

Version

0.45.0

Debug Output

Example Data

Logs sent by the opentelemetry collector after being received by telemetrygen:

{
    "resourceLogs": [
        {
            "resource": {
                "attributes": [
                    {
                        "key": "service.name",
                        "value": {
                            "stringValue": "test"
                        }
                    },
                    {
                        "key": "k8s.cluster.name",
                        "value": {
                            "stringValue": "logs-dev-ne1-z02"
                        }
                    }
                ]
            },
            "scopeLogs": [
                {
                    "scope": {},
                    "logRecords": [
                        {
                            "timeUnixNano": "1742408939363760620",
                            "severityNumber": 9,
                            "severityText": "Info",
                            "body": {
                                "stringValue": "the message"
                            },
                            "attributes": [
                                {
                                    "key": "app",
                                    "value": {
                                        "stringValue": "server"
                                    }
                                },
                                {
                                    "key": "component",
                                    "value": {
                                        "stringValue": "logslocal"
                                    }
                                }
                            ],
                            "droppedAttributesCount": 1,
                            "traceId": "",
                            "spanId": ""
                        }
                    ]
                }
            ]
        }
    ]
}

How its parsed by vector:

{
    "attributes": {
        "app": "server",
        "component": "logslocal"
    },
    "dropped_attributes_count": 1,
    "message": "the message",
    "observed_timestamp": "2025-03-19T14:29:39.120971960Z",
    "resources": {
        "k8s.cluster.name": "logs-dev-ne1-z02",
        "service.name": "test"
    },
    "severity_number": 9,
    "severity_text": "Info",
    "source_type": "opentelemetry",
    "timestamp": "2025-03-19T14:29:38.903143790Z"
}

The timestamp are different because these are different samples, but the result would be the same.

Additional Context

Here is my otelcollector config:

processors:
  attributes:
    actions:
      - key: component
        value: "logslocal"
        action: upsert
  resource:
    attributes:
    - key: service.name
      value: "test"
      action: upsert
    - key: k8s.cluster.name
      value: "logs-dev-ne1-z02"
      action: upsert
receivers:
  otlp:
    protocols:
      grpc:
       endpoint: 0.0.0.0:4317
exporters:
  otlphttp:
    endpoint: http://localhost:4321
  debug:
    verbosity: detailed
    sampling_initial: 5
    sampling_thereafter: 200
  file:
    path: ./foo
service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [attributes,resource]
      exporters: [otlphttp,debug,file]

I then use telemetrygen to generate logs in otel format.

References

I also saw that there is a bug for the Sink opentelemetry, #22188

Ill try to find a way to use the reduce to combine into one request

[pront]

Hi @someStrangerFromTheAbyss, thank you adding all the details to this issue. I agree our goal should be OOTB compatibility between OTEL sources and sinks. What you described above is a known limitation in our implementation because we don't have a mapping from OTEL to Vector events and then a mapping from Vector events back to OTEL format. This PR is an example of this for metrics: #22550.

[someStrangerFromTheAbyss]

@pront No prob !

Problem is, for our production, we need it has soon has possible.

So ill be continuing to improve that VRL (arrays are not implemented and are posssible). But that vrl should cover most cases until the functionnality is implemented.

And thanks for changing the tags to the correct one.

[pront]

arrays are not implemented and are possible

Can you expand on this? Maybe I can help.

[someStrangerFromTheAbyss]

@pront
From my understanding, attributes can have the following types for its values:

• string
• boolean,
• double precision floating point,
• signed 64 bit integer
• an array of primitive type values

Looking at my unniversal vrl script for any otlp incoming logs, the only type that i did not include is for attributes of type array. I could do a "is_array", then do a loop through that array and repeat the code just above once more, so something like this:

for_each(object!(.attributes)) -> |key,value|{
    insert_key = {}
    insert_key.key = key
    insert_key.value = {}
    # The attribute value is either:
    # A primitive type: string, boolean, double precision floating point (IEEE 754-1985) or signed 64 bit integer.
    # An array of primitive type values. The array MUST be homogeneous, i.e., it MUST NOT contain values of different types.
    # Example of type: https://github.com/open-telemetry/opentelemetry-proto/blob/d7770822d70c7bd47a6891fc9faacc66fc4af3d3/examples/logs.json#L9
    if is_string(value) {
        insert_key.value.stringValue = value
    } else if is_boolean(value) {
        insert_key.value.boolValue = value
    } else if is_integer(value){
        insert_key.value.intValue = value
    } else if is_float(value){
        insert_key.value.doubleValue = value
    } else if is_array(value){
        insert_key.value.arrayValue = []
        for_each(array!(value)) -> |_index,array_value|{
            array_key = {}
            if is_string(array_value) {
                array_key.stringValue = array_value
                push(insert_key.value.arrayValue,array_key)
            } else if is_boolean(array_value) {
                array_key.boolValue = array_value
                push(insert_key.value.arrayValue,array_key)
            } else if is_integer(array_value){
                array_key.intValue = array_value
                push(insert_key.value.arrayValue,array_key)
            } else if is_float(array_value){
                array_key.doubleValue = array_value
                push(insert_key.value.arrayValue,array_key)
            } else {
                array_key.stringValue = string(value) ?? ""
                push(insert_key.value.arrayValue,array_key)
            }
        }
    } else {
        # in such a case, we try to transform to string then, insert value has string. If cannot be string, simply put empty.
        insert_key.value.stringValue = string(value) ?? ""
    }
    attributes_array = push(attributes_array,insert_key)
}

Not sure of the syntax, but i think it will work ? Not very pretty and not sure if its the most optimal way to do it.

Ill try it on monday see it works with array value for attributes. If you have suggestion to make it nicer, ill take them.

Thank for answering.

[someStrangerFromTheAbyss]

@pront So i got some progress to have a working mapping. This is where im at:

attributes_array = []
resource_attributes_array = []
# for each: https://vector.dev/docs/reference/vrl/examples/#for_each
for_each(object!(.attributes)) -> |key,value|{
    insert_key = {}
    insert_key.key = key
    insert_key.value = {}
    # The attribute value is either:
    # A primitive type: string, boolean, double precision floating point (IEEE 754-1985) or signed 64 bit integer.
    # An array of primitive type values. The array MUST be homogeneous, i.e., it MUST NOT contain values of different types.
    # Example of type: https://github.com/open-telemetry/opentelemetry-proto/blob/d7770822d70c7bd47a6891fc9faacc66fc4af3d3/examples/logs.json#L9
    if is_string(value) {
        insert_key.value.stringValue = value
    } else if is_boolean(value) {
        insert_key.value.boolValue = value
    } else if is_integer(value){
        insert_key.value.intValue = value
    } else if is_float(value){
        insert_key.value.doubleValue = value
    } else if is_object(value){
        insert_key.value.kvlistValue = {}
        insert_key.value.kvlistValue.values = []
        for_each(object!(value)) -> |inner_key,inner_value|{
            array_object_key = {}
            array_object_key.key = inner_key
            array_object_key.value = {}
            if is_string(inner_value) {
                array_object_key.value.stringValue = inner_value
                insert_key.value.kvlistValue.values = push(insert_key.value.kvlistValue.values,array_object_key)
            } else if is_boolean(inner_value) {
                array_object_key.value.boolValue = inner_value
                insert_key.value.kvlistValue.values = push(insert_key.value.kvlistValue.values,array_object_key)
            } else if is_integer(inner_value){
                array_object_key.value.intValue = inner_value
                insert_key.value.kvlistValue.values = push(insert_key.value.kvlistValue.values,array_object_key)
            } else if is_float(inner_value){
                array_object_key.value.doubleValue = inner_value
                insert_key.value.kvlistValue.values = push(insert_key.value.kvlistValue.values,array_object_key)
            } else {
                array_object_key.value.stringValue = string(inner_value) ?? ""
                insert_key.value.kvlistValue.values = push(insert_key.value.kvlistValue.values,array_object_key)
            }
        }
    } else if is_array(value){
        insert_key.value.arrayValue = {}
        insert_key.value.arrayValue.values = []
        for_each(array!(value)) -> |_index,array_value|{
            array_key = {}
            if is_string(array_value) {
                array_key.stringValue = array_value
                insert_key.value.arrayValue.values = push(insert_key.value.arrayValue.values,array_key)
            } else if is_boolean(array_value) {
                array_key.boolValue = array_value
                insert_key.value.arrayValue.values = push(insert_key.value.arrayValue.values,array_key)
            } else if is_integer(array_value){
                array_key.intValue = array_value
                insert_key.value.arrayValue.values = push(insert_key.value.arrayValue.values,array_key)
            } else if is_float(array_value){
                array_key.doubleValue = array_value
                insert_key.value.arrayValue.values = push(insert_key.value.arrayValue.values,array_key)
            } else {
                array_key.stringValue = string(array_value) ?? ""
                insert_key.value.arrayValue.values = push(insert_key.value.arrayValue.values,array_key)
            }
        }
    } else {
        # in such a case, we try to transform to string then, insert value has string. If cannot be string, simply put empty.
        insert_key.value.stringValue = string(value) ?? ""
    }
    attributes_array = push(attributes_array,insert_key)
}
for_each(object!(.resources)) -> |key,value|{
    insert_key = {}
    insert_key.key = key
    insert_key.value = {}
    # The attribute value is either:
    # A primitive type: string, boolean, double precision floating point (IEEE 754-1985) or signed 64 bit integer.
    # An array of primitive type values. The array MUST be homogeneous, i.e., it MUST NOT contain values of different types.
    # Example of type: https://github.com/open-telemetry/opentelemetry-proto/blob/d7770822d70c7bd47a6891fc9faacc66fc4af3d3/examples/logs.json#L9
    if is_string(value) {
        insert_key.value.stringValue = value
    } else if is_boolean(value) {
        insert_key.value.boolValue = value
    } else if is_integer(value){
        insert_key.value.intValue = value
    } else if is_float(value){
        insert_key.value.doubleValue = value
        } else if is_object(value){
        insert_key.value.kvlistValue = {}
        insert_key.value.kvlistValue.values = []
        for_each(object!(value)) -> |inner_key,inner_value|{
            array_object_key = {}
            array_object_key.key = inner_key
            array_object_key.value = {}
            if is_string(inner_value) {
                array_object_key.value.stringValue = inner_value
                insert_key.value.kvlistValue.values = push(insert_key.value.kvlistValue.values,array_object_key)
            } else if is_boolean(inner_value) {
                array_object_key.value.boolValue = inner_value
                insert_key.value.kvlistValue.values = push(insert_key.value.kvlistValue.values,array_object_key)
            } else if is_integer(inner_value){
                array_object_key.value.intValue = inner_value
                insert_key.value.kvlistValue.values = push(insert_key.value.kvlistValue.values,array_object_key)
            } else if is_float(inner_value){
                array_object_key.value.doubleValue = inner_value
                insert_key.value.kvlistValue.values = push(insert_key.value.kvlistValue.values,array_object_key)
            } else {
                array_object_key.value.stringValue = string(inner_value) ?? ""
                insert_key.value.kvlistValue.values = push(insert_key.value.kvlistValue.values,array_object_key)
            }
        }
    } else if is_array(value){
        insert_key.value.arrayValue = {}
        insert_key.value.arrayValue.values = []
        for_each(array!(value)) -> |_index,array_value|{
            array_key = {}
            if is_string(array_value) {
                array_key.stringValue = array_value
                insert_key.value.arrayValue.values = push(insert_key.value.arrayValue.values,array_key)
            } else if is_boolean(array_value) {
                array_key.boolValue = array_value
                insert_key.value.arrayValue.values = push(insert_key.value.arrayValue.values,array_key)
            } else if is_integer(array_value){
                array_key.intValue = array_value
                insert_key.value.arrayValue.values = push(insert_key.value.arrayValue.values,array_key)
            } else if is_float(array_value){
                array_key.doubleValue = array_value
                insert_key.value.arrayValue.values = push(insert_key.value.arrayValue.values,array_key)
            } else {
                array_key.stringValue = string(array_value) ?? ""
                insert_key.value.arrayValue.values = push(insert_key.value.arrayValue.values,array_key)
            }
        }
    } else {
        # in such a case, we try to transform to string then, insert value has string. If cannot be string, simply put empty.
        insert_key.value.stringValue = string(value) ?? ""
    }
    resource_attributes_array = push(resource_attributes_array,insert_key)
}
#map_values(object!(.attributes)) -> |value| { upcase!(value) }
.resourceLogs = [{
  "resource": {
    "attributes": resource_attributes_array
  },
  "scopeLogs": [{
    "scope": object(.scope) ?? {},
    "logRecords": [{
      "timeUnixNano": to_unix_timestamp!(.timestamp, unit: "nanoseconds"),
      "body": { "stringValue": .message },
      "severityText": string(.severity_text) ?? "",
      "severityNumber": int(.severity_number) ?? 0,
      "attributes": attributes_array,
      "droppedAttributesCount": .dropped_attributes_count,
      "traceId": string(.traceId) ?? "",
      "spanId": string(.spanId) ?? ""
    }]
  }]
}]
del(.severity_number)
del(.severity_text)
del(.resources)
del(.attributes)
del(.message)
del(.timestamp)
del(.dropped_attributes_count)
del(.observed_timestamp)
del(.source_type)
del(.scope)

There is one problem.... Sometime attributes can be objects and, from my testing, they can have inner object. From the vector documentation, i saw that for_each does not support recursive iteration

I saw that map_values and map_keys can be recursive. I assume i can use those when hitting attributes that objects and go through any inner objects ?

Thanks for the help.

[pront]

I saw that map_values and map_keys can be recursive. I assume i can use those when hitting attributes that objects and go through any inner objects ?

Yes they can, here is an example: playground

And again, thank you for sharing your solution with the community. In the future, a similar mapping will be done natively with a Rust implemenation.

[someStrangerFromTheAbyss]

@pront Thanks for the sharing.

Problem is, i would need recursive of both the keys and the values at the same time....the for_each does not allow it so the object cannot be implemented correctly right now with only VRL. Sadly, i won't have time to do it in Rust since i have date limit and i do not know how Rust works (mostly a golang/Java/C# DEV).

Me and the client came to a solution: Everything that is object found with the VRL script will be flatten. Not perfect, but will allow us to go around the problem.

I'll share the final version of the VRL script when its done.

Thanks again for the help.