deal with muliple tags with the same key (#829)

slackhq · Mar 26, 2024 · c95ed27 · c95ed27
1 parent cc194a2
commit c95ed27
Show file tree

Hide file tree

Showing 4 changed files with 43 additions and 11 deletions.
diff --git a/kaldb/src/main/java/com/slack/kaldb/bulkIngestApi/opensearch/BulkApiRequestParser.java b/kaldb/src/main/java/com/slack/kaldb/bulkIngestApi/opensearch/BulkApiRequestParser.java
@@ -112,14 +112,22 @@ public static Trace.Span fromIngestDocument(
     sourceAndMetadata.remove(IngestDocument.Metadata.ID.getFieldName());
     sourceAndMetadata.remove(IngestDocument.Metadata.INDEX.getFieldName());
 
-    sourceAndMetadata.forEach(
-        (key, value) -> spanBuilder.addTags(SpanFormatter.convertKVtoProto(key, value, schema)));
-    spanBuilder.addTags(
-        Trace.KeyValue.newBuilder()
-            .setKey(SERVICE_NAME_KEY)
-            .setVType(Trace.ValueType.STRING)
-            .setVStr(index)
-            .build());
+    boolean tagsContainServiceName = false;
+    for (Map.Entry<String, Object> kv : sourceAndMetadata.entrySet()) {
+      if (!tagsContainServiceName && kv.getKey().equals(SERVICE_NAME_KEY)) {
+        tagsContainServiceName = true;
+      }
+      spanBuilder.addTags(SpanFormatter.convertKVtoProto(kv.getKey(), kv.getValue(), schema));
+    }
+    if (!tagsContainServiceName) {
+      spanBuilder.addTags(
+          Trace.KeyValue.newBuilder()
+              .setKey(SERVICE_NAME_KEY)
+              .setVType(Trace.ValueType.STRING)
+              .setFieldType(Schema.SchemaFieldType.KEYWORD)
+              .setVStr(index)
+              .build());
+    }
 
     return spanBuilder.build();
   }

diff --git a/kaldb/src/main/java/com/slack/kaldb/logstore/schema/SchemaAwareLogDocumentBuilderImpl.java b/kaldb/src/main/java/com/slack/kaldb/logstore/schema/SchemaAwareLogDocumentBuilderImpl.java
@@ -430,7 +430,9 @@ public Document fromMessage(Trace.Span message) throws JsonProcessingException {
     Map<String, Trace.KeyValue> tags =
         message.getTagsList().stream()
             .map(keyValue -> Map.entry(keyValue.getKey(), keyValue))
-            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey, Map.Entry::getValue, (firstKV, dupKV) -> firstKV));
 
     // This should just be top level Trace.Span fields. This is error prone - what if type is
     // not a string?

diff --git a/kaldb/src/test/java/com/slack/kaldb/logstore/schema/RaiseErrorFieldValueTest.java b/kaldb/src/test/java/com/slack/kaldb/logstore/schema/RaiseErrorFieldValueTest.java
@@ -166,7 +166,7 @@ public void testRaiseErrorOnConflictingReservedField() {
                         1, "Test message", Instant.now(), List.of(hostField, tagField))))
         .isInstanceOf(FieldDefMismatchException.class);
 
-    assertThat(docBuilder.getSchema().size()).isEqualTo(19);
+    assertThat(docBuilder.getSchema().size()).isEqualTo(18);
     assertThat(docBuilder.getSchema().keySet()).contains(hostNameField);
     assertThat(docBuilder.getSchema().get(hostNameField).fieldType).isEqualTo(FieldType.STRING);
     assertThat(MetricsUtil.getCount(DROP_FIELDS_COUNTER, meterRegistry)).isZero();
@@ -206,7 +206,7 @@ public void testRaiseErrorOnConflictingReservedFieldWithoutFullTextSearch() {
                         1, "Test message", Instant.now(), List.of(hostField, tagField))))
         .isInstanceOf(FieldDefMismatchException.class);
 
-    assertThat(docBuilder.getSchema().size()).isEqualTo(18);
+    assertThat(docBuilder.getSchema().size()).isEqualTo(17);
     assertThat(docBuilder.getSchema().keySet()).contains(hostNameField);
     assertThat(docBuilder.getSchema().get(hostNameField).fieldType).isEqualTo(FieldType.STRING);
     assertThat(MetricsUtil.getCount(DROP_FIELDS_COUNTER, meterRegistry)).isZero();

diff --git a/kaldb/src/test/java/com/slack/kaldb/schema/SpanTagFormatterWithSchemaTest.java b/kaldb/src/test/java/com/slack/kaldb/schema/SpanTagFormatterWithSchemaTest.java
@@ -337,5 +337,27 @@ public void testDuplicateFieldAsTag() {
 
     logStore.addMessage(span);
     assertThat(getCount(MESSAGES_FAILED_COUNTER, meterRegistry)).isEqualTo(0);
+
+    // duplicate tags
+    span =
+        Trace.Span.newBuilder()
+            .setName("service1")
+            .setId(ByteString.copyFrom("123".getBytes()))
+            .addTags(
+                Trace.KeyValue.newBuilder()
+                    .setKey("tag1")
+                    .setVStr("value1")
+                    .setVType(Trace.ValueType.STRING)
+                    .setFieldType(Schema.SchemaFieldType.KEYWORD))
+            .addTags(
+                Trace.KeyValue.newBuilder()
+                    .setKey("tag1")
+                    .setVStr("value1")
+                    .setVType(Trace.ValueType.STRING)
+                    .setFieldType(Schema.SchemaFieldType.KEYWORD))
+            .build();
+
+    logStore.addMessage(span);
+    assertThat(getCount(MESSAGES_FAILED_COUNTER, meterRegistry)).isEqualTo(0);
   }
 }