[BUG] Duplicate key exception when integrate with open-telemetry exporter

[hoangthanh28]

Describe the bug
Data prepper throws java.lang.IllegalStateException: Duplicate key xxx when ingest through log pipeline.
See the error below, due to open telemetry exporter exporting a duplicate ConnectionId key. Not sure this is expected behavior or not. Please help considering to fix this issue.

2023-12-29T07:28:02,516 [pool-13-thread-1] ERROR org.opensearch.dataprepper.plugins.source.otellogs.OTelLogsGrpcService - Failed to parse the request resource_logs {
  resource {
    ...truncate...
  scope_logs {
    scope {
      name: "Microsoft.Extensions.Hosting.Internal.Host"
    }
    log_records {
      time_unix_nano: 1703834880595497100
      severity_number: SEVERITY_NUMBER_DEBUG
      severity_text: "Debug"
      body {
        string_value: "Connection id \"{ConnectionId}\" sending FIN because: \"{Reason}\""
      }
      attributes {
        key: "ConnectionId"
        value {
          string_value: "0HN087UNTCNA9"
        }
      }
      attributes {
        key: "Reason"
        value {
          string_value: "The Socket transport\'s send loop completed gracefully."
        }
      }
      attributes {
        key: "ConnectionId"
        value {
          string_value: "0HN087UNTCNA9"
        }
      }
      observed_time_unix_nano: 1703834880595497100
    }
  }
}
java.lang.IllegalStateException: Duplicate key log.attributes.ConnectionId (attempted merging values 0HN087UNTCNA9 and 0HN087UNTCNA9)
	at java.base/java.util.stream.Collectors.duplicateKeyException(Collectors.java:135) ~[?:?]
	at java.base/java.util.stream.Collectors.lambda$uniqKeysMapAccumulator$1(Collectors.java:182) ~[?:?]
	at java.base/java.util.stream.ReduceOps$3ReducingSink.accept(ReduceOps.java:169) ~[?:?]
	at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625) ~[?:?]
	at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
	at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
	at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921) ~[?:?]
	at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:?]
	at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682) ~[?:?]
	at org.opensearch.dataprepper.plugins.otel.codec.OTelProtoCodec.unpackKeyValueListLog(OTelProtoCodec.java:1094) ~[otel-proto-common-2.6.1.jar:?]
	at org.opensearch.dataprepper.plugins.otel.codec.OTelProtoCodec$OTelProtoDecoder.lambda$processLogsList$7(OTelProtoCodec.java:376) ~[otel-proto-common-2.6.1.jar:?]

To Reproduce
Steps to reproduce the behavior:

1. I have pushed the sample repository, please visit here
2. Run docker-compose -f docker-compose.yml up -d
3. curl --location 'http://localhost:5000'
4. Wait a couple of seconds, then check data-prepper log. docker logs -f data-prepper

Expected behavior
Remove the duplicate key and ingest the log without issue.
Not sure whether this is a right place to fix or not ( forgive me if I'm wrong, I'm a .net developer - FYI)
Code

 protected List<OpenTelemetryLog> processLogsList(final List<LogRecord> logsList,
                                                                                         final String serviceName,
                                                                                         final Map<String, Object> ils,
                                                                                         final Map<String, Object> resourceAttributes,
                                                                                         final String schemaUrl) {
            return logsList.stream()
                    .map(log -> JacksonOtelLog.builder()
                            .withTime(OTelProtoCodec.convertUnixNanosToISO8601(log.getTimeUnixNano()))
                            .withObservedTime(OTelProtoCodec.convertUnixNanosToISO8601(log.getObservedTimeUnixNano()))
                            .withServiceName(serviceName)
                            .withAttributes(OTelProtoCodec.mergeAllAttributes(
                                    Arrays.asList(
                                            // would be possible to de-duplicate log.getAttributesList() first ???, so that will solve the problem.
                                            OTelProtoCodec.unpackKeyValueListLog(log.getAttributesList()),
                                            resourceAttributes,
                                            ils
                                    )
                            ))
                            .withSchemaUrl(schemaUrl)
                            .withFlags(log.getFlags())
                            .withTraceId(OTelProtoCodec.convertByteStringToString(log.getTraceId()))
                            .withSpanId(OTelProtoCodec.convertByteStringToString(log.getSpanId()))
                            .withSeverityNumber(log.getSeverityNumberValue())
                            .withSeverityText(log.getSeverityText())
                            .withDroppedAttributesCount(log.getDroppedAttributesCount())
                            .withBody(OTelProtoCodec.convertAnyValue(log.getBody()))
                            .build())
                    .collect(Collectors.toList());
        }

Screenshots
If applicable, add screenshots to help explain your problem.

Environment (please complete the following information):

• OS: Debian
• Version: 12

Additional context
N/A

[dlvenable]

@hoangthanh28 , We see an issue with the DotNet exporter that may be related - open-telemetry/opentelemetry-dotnet#4324. Are you using the DotNet exporter?

Also, how would you like merges to be resolved? The first one wins, second one wins, or we make an array of them?

[hoangthanh28]

Hi @dlvenable
Yes, I use dotnet exporter, this is problematic with open-telemetry dotnet, but data-prepper should be agnostics from the exporter I think.
The merge mechanism should not be critical, duplicates have the same key/value anyway. I suggest we get the last value.
In C#, we can implement like this:

dic[item.key] = item.value