opensearch-project / opensearch-project/data-prepper

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

Open
#3,868 3 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

bug
Dominant language
Java
Stars
374
Forks
354
Avg merge
3d 18h
Merged PRs (30d)
8

Description

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

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start in data-prepper-plugins/otel-proto-common/src/main/java/org/opensearch/dataprepper/plugins/otel/codec/OTelProtoCodec.java, especially unpackKeyValueListLog and processLogsList, then reproduce the failure with the linked docker-compose steps and curl request. Confirm that repeated log attributes such as ConnectionId no longer cause ingestion to fail and that the sample log is accepted.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
observability-sre
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.