[exporter/loki] Not translating labels(from resources) correctly when multiple logRecords

[rogeriospinaintelipost]

What happened?

Description

Converting log resources for loki labels works only for the first LogRecord.

Reading the code, I figured out that after first LogRecord process, resource attributes that were promoted to label are removed, so LogRecords #1 and #2 will not be able to convert labels.

for k := 0; k < logs.Len(); k++ {

				// similarly, we may remove attributes, so change only our version
				log := plog.NewLogRecord()
				logs.At(k).CopyTo(log)

				mergedLabels := convertAttributesAndMerge(log.Attributes(), resource.Attributes())
				// remove the attributes that were promoted to labels
				removeAttributes(log.Attributes(), mergedLabels)
				removeAttributes(resource.Attributes(), mergedLabels)

Also could notice that a copy of resources are made, but I think that it should be inside LogRecord loop, not outside. Ex:

This:

for i := 0; i < rls.Len(); i++ {
		ills := rls.At(i).ScopeLogs()

		for j := 0; j < ills.Len(); j++ {
			logs := ills.At(j).LogRecords()
			for k := 0; k < logs.Len(); k++ {

				// we may remove attributes, so we make a copy and change our version
				resource := pcommon.NewResource()
				rls.At(i).Resource().CopyTo(resource)

				// similarly, we may remove attributes, so change only our version
				log := plog.NewLogRecord()
				logs.At(k).CopyTo(log)

				mergedLabels := convertAttributesAndMerge(log.Attributes(), resource.Attributes())
				// remove the attributes that were promoted to labels
				removeAttributes(log.Attributes(), mergedLabels)
				removeAttributes(resource.Attributes(), mergedLabels)

Instead of this:

for i := 0; i < rls.Len(); i++ {
		ills := rls.At(i).ScopeLogs()

		// we may remove attributes, so we make a copy and change our version
		resource := pcommon.NewResource()
		rls.At(i).Resource().CopyTo(resource)

		for j := 0; j < ills.Len(); j++ {
			logs := ills.At(j).LogRecords()
			for k := 0; k < logs.Len(); k++ {

				// similarly, we may remove attributes, so change only our version
				log := plog.NewLogRecord()
				logs.At(k).CopyTo(log)

				mergedLabels := convertAttributesAndMerge(log.Attributes(), resource.Attributes())
				// remove the attributes that were promoted to labels
				removeAttributes(log.Attributes(), mergedLabels)
				removeAttributes(resource.Attributes(), mergedLabels)

Steps to Reproduce

Application writes 3 LogRecords:

log.info("Info example #0");
log.info("Info example #1");
log.info("Info example #2");

Expected Result

I expected all logRecords to have host_name, container_name and namespace labels.

Actual Result

Only the first LogRecord contains correct labels.

Collector version

0.60.0

Environment information

No response

OpenTelemetry Collector configuration

receivers:
      otlp:
        protocols:
          grpc:
          http:

    processors:
      batch:

      attributes:
        actions:
          - action: insert
            key: loki.resource.labels
            value: host_name, container_name, namespace
          - action: insert
            key: loki.attribute.labels
            value: uri

      resource:
        attributes:
          - action: insert
            key: host_name
            from_attribute: host.name

          - action: insert
            key: namespace
            from_attribute: k8s.namespace.name

          - action: insert
            key: container_name
            from_attribute: k8s.container.name


    exporters:
      logging:
        logLevel: debug

      loki:
        endpoint: "http://{loki-url}/loki/api/v1/push"
        tls:
          insecure: true

    service:
      pipelines:
        logs:
           receivers: [otlp]
           processors: [resource, attributes]
           exporters: [loki, logging]

Log output

Resource SchemaURL: https://opentelemetry.io/schemas/1.12.0
Resource attributes:
     -> host.arch: STRING(amd64)
     -> host.name: STRING(opentelemetry-lab)
     -> k8s.container.name: STRING(opentelemetry-lab)
     -> k8s.namespace.name: STRING(nmsp-test)
     -> k8s.pod.name: STRING(opentelemetry-lab)
     -> os.type: STRING(linux)
     -> process.pid: INT(1)
     -> service.name: STRING(opentelemetry-lab)
     -> telemetry.auto.version: STRING(1.18.0)
     -> telemetry.sdk.language: STRING(java)
     -> telemetry.sdk.name: STRING(opentelemetry)
     -> telemetry.sdk.version: STRING(1.18.0)
     -> host_name: STRING(opentelemetry-lab)
     -> namespace: STRING(nmsp-test)
     -> container_name: STRING(opentelemetry-lab)
ScopeLogs #0
ScopeLogs SchemaURL:
InstrumentationScope com.acme.opentelemetrylab.OpentelemetryLabApplication
LogRecord #0
ObservedTimestamp: 1970-01-01 00:00:00 +0000 UTC
Timestamp: 2022-09-19 18:12:47.911348 +0000 UTC
Severity: INFO
Body: Info log exemplo
Attributes:
     -> loki.resource.labels: STRING(host_name, container_name, namespace)
Trace ID: 15e28b21efc2d4a1a045ee15775ee549
Span ID: b66f9fe0cec628f5
Flags: 1
LogRecord #1
ObservedTimestamp: 1970-01-01 00:00:00 +0000 UTC
Timestamp: 2022-09-19 18:12:47.911497 +0000 UTC
Severity: ERROR
Body: Error log exemplo
Attributes:
     -> loki.resource.labels: STRING(host_name, container_name, namespace)
Trace ID: 15e28b21efc2d4a1a045ee15775ee549
Span ID: b66f9fe0cec628f5
Flags: 1
LogRecord #2
ObservedTimestamp: 1970-01-01 00:00:00 +0000 UTC
Timestamp: 2022-09-19 18:12:47.91156 +0000 UTC
Severity: INFO
Body: teste
Attributes:
     -> loki.resource.labels: STRING(host_name, container_name, namespace)
Trace ID: 15e28b21efc2d4a1a045ee15775ee549
Span ID: b66f9fe0cec628f5
Flags: 1

Additional context

No response

[rogeriospinaintelipost]

@jpkrohling

[github-actions]

Pinging code owners: @gramidt @gouthamve @jpkrohling @kovrus @mar4uk. See Adding Labels via Comments if you do not have permissions to add labels yourself.

[kovrus]

I can take it over.