feat: Adding a new feature which allows us to drop spans matching some criterion. (#122)

* feat: Adding a new feature which allows us to drop spans matching some
criterion.
This comes in handy when there is a lot of unnecessary data coming into hypertrace platform
that's not really useful for API observability.

Author: buchi-busireddy

span-normalizer/helm/templates/span-normalizer-config.yaml

@@ -49,6 +49,10 @@ data:
       {{- if hasKey .Values.spanNormalizerConfig.processor "defaultTenantId" }}
       defaultTenantId = "{{ .Values.spanNormalizerConfig.processor.defaultTenantId }}"
       {{- end }}
+
+      {{- if hasKey .Values.spanNormalizerConfig.processor "spanDropCriterion" }}
+      spanDropCriterion = {{ .Values.spanNormalizerConfig.processor.spanDropCriterion | toJson }}
+      {{- end }}
     }
     {{- end }}
 

span-normalizer/span-normalizer/src/main/java/org/hypertrace/core/spannormalizer/jaeger/JaegerSpanNormalizer.java

@@ -13,11 +13,13 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.Callable;
@@ -33,6 +35,7 @@
 import org.apache.avro.specific.SpecificDatumWriter;
 import org.apache.avro.specific.SpecificRecordBase;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
 import org.hypertrace.core.datamodel.AttributeValue;
 import org.hypertrace.core.datamodel.Attributes;
 import org.hypertrace.core.datamodel.Event;
@@ -71,13 +74,27 @@ public class JaegerSpanNormalizer implements SpanNormalizer<Span, RawSpan> {
   private static final String DEFAULT_TENANT_ID_CONFIG = "processor.defaultTenantId";
   // list of tenant ids to exclude
   private static final String TENANT_IDS_TO_EXCLUDE_CONFIG = "processor.excludeTenantIds";
-  private final List<String> tenantIdsToExclude;
+
+  /**
+   * Config key using which a list of criterion can be specified to drop the matching spans.
+   * Any span matching any one of the criterion is dropped. Each criteria is a comma separated
+   * list of key:value pairs and multiple pairs in one criteria are AND'ed.
+   *
+   * For example: ["messaging.destination_kind:queue,messaging.operation:receive,messaging.system:jms"]
+   * drops all spans which have all 3 attribute:value pairs.
+   */
+  private static final String SPAN_DROP_CRITERION_CONFIG = "processor.spanDropCriterion";
+
+  private static final String COMMA = ",";
+  private static final String COLON = ":";
 
   private static final String SPAN_NORMALIZATION_TIME_METRIC = "span.normalization.time";
 
   private static JaegerSpanNormalizer INSTANCE;
   private final FieldsGenerator fieldsGenerator;
   private final TenantIdProvider tenantIdProvider;
+  private final List<String> tenantIdsToExclude;
+  private final List<List<Pair<String, String>>> spanDropCriterion;
   private final ConcurrentMap<String, Timer> tenantToSpanNormalizationTimer = new ConcurrentHashMap<>();
 
   public static JaegerSpanNormalizer get(Config config) {
@@ -139,6 +156,32 @@ public JaegerSpanNormalizer(Config config) {
     if (!this.tenantIdsToExclude.isEmpty()) {
       LOG.info("list of tenant ids to exclude : {}", this.tenantIdsToExclude);
     }
+
+    List<String> criterion = config.hasPath(SPAN_DROP_CRITERION_CONFIG)
+        ? config.getStringList(SPAN_DROP_CRITERION_CONFIG)
+        : Collections.emptyList();
+
+    // Parse the config to see if there is any criteria to drop spans.
+    this.spanDropCriterion = criterion.stream()
+        // Split each criteria based on comma
+        .map(s -> s.split(COMMA))
+        .map(a -> Arrays.stream(a).map(this::convertToPair).filter(Objects::nonNull).collect(Collectors.toList()))
+        .collect(Collectors.toList());
+
+    if (!this.spanDropCriterion.isEmpty()) {
+      LOG.info("Span drop criterion: {}", this.spanDropCriterion);
+    }
+  }
+
+  @Nullable
+  private Pair<String, String> convertToPair(String s) {
+    if (s != null && s.contains(COLON)) {
+      String[] parts = s.split(COLON);
+      if (parts.length == 2) {
+        return Pair.of(parts[0], parts[1]);
+      }
+    }
+    return null;
   }
 
   public Timer getSpanNormalizationTimer(String tenantId) {
@@ -166,6 +209,10 @@ public RawSpan convert(Span jaegerSpan) throws Exception {
       return null;
     }
 
+    if (!this.spanDropCriterion.isEmpty() && shouldDropSpan(tags)) {
+      return null;
+    }
+
     // Record the time taken for converting the span, along with the tenant id tag.
     return tenantToSpanNormalizationTimer
         .computeIfAbsent(
@@ -176,6 +223,17 @@ public RawSpan convert(Span jaegerSpan) throws Exception {
         .recordCallable(getRawSpanNormalizerCallable(jaegerSpan, tags, tenantId));
   }
 
+  /**
+   * Method to check if the given span attributes match any of the drop criterion. Returns true if the span should be
+   * dropped, false otherwise.
+   */
+  private boolean shouldDropSpan(Map<String, KeyValue> tags) {
+    return this.spanDropCriterion.stream()
+        .anyMatch(l -> l.stream()
+            .allMatch(p -> tags.containsKey(p.getLeft())
+                && StringUtils.equals(tags.get(p.getLeft()).getVStr(), p.getRight())));
+  }
+
   @Nonnull
   private Callable<RawSpan> getRawSpanNormalizerCallable(Span jaegerSpan,
       Map<String, KeyValue> spanTags, String tenantId) {
@@ -220,7 +278,7 @@ private Event buildEvent(
 
     // SPAN REFS
     List<JaegerSpanInternalModel.SpanRef> referencesList = jaegerSpan.getReferencesList();
-    if (referencesList != null && referencesList.size() > 0) {
+    if (referencesList.size() > 0) {
       eventBuilder.setEventRefList(new ArrayList<>());
       // Convert the reflist to a set to remove duplicate references. This has been observed in the
       // field.
@@ -272,13 +330,13 @@ private Event buildEvent(
 
     // WARNINGS
     ProtocolStringList warningsList = jaegerSpan.getWarningsList();
-    if (warningsList != null && warningsList.size() > 0) {
+    if (warningsList.size() > 0) {
       jaegerFieldsBuilder.setWarnings(warningsList);
     }
     // LOGS - NOTE: This is modeled as a google.protobuf.Any
     List<JaegerSpanInternalModel.Log> logsList = jaegerSpan.getLogsList();
     List<String> eventLogsList = new ArrayList<>();
-    if (logsList != null && logsList.size() > 0) {
+    if (logsList.size() > 0) {
       for (Log log : logsList) {
         try {
           String json = JsonFormat.printer().omittingInsignificantWhitespace().print(log);
@@ -292,11 +350,11 @@ private Event buildEvent(
 
     // Jaeger service name can come from either first class field in Span or the tag `jaeger.servicename`
     String serviceName =
-        (jaegerSpan.getProcess() != null && !StringUtils.isEmpty(jaegerSpan.getProcess().getServiceName()))
+        !StringUtils.isEmpty(jaegerSpan.getProcess().getServiceName())
         ? jaegerSpan.getProcess().getServiceName()
-        : (attributeFieldMap.containsKey(OLD_JAEGER_SERVICENAME_KEY)
+        : attributeFieldMap.containsKey(OLD_JAEGER_SERVICENAME_KEY)
             ? attributeFieldMap.get(OLD_JAEGER_SERVICENAME_KEY).getValue()
-            : StringUtils.EMPTY);
+            : StringUtils.EMPTY;
 
     if (!StringUtils.isEmpty(serviceName)) {
       eventBuilder.setServiceName(serviceName);
@@ -337,7 +395,7 @@ public static <T extends SpecificRecordBase> String convertToJsonString(T object
       writer.write(object, encoder);
       encoder.flush();
       output.flush();
-      return new String(output.toByteArray());
+      return output.toString();
     }
   }
 }

span-normalizer/span-normalizer/src/main/java/org/hypertrace/core/spannormalizer/jaeger/JaegerSpanToAvroRawSpanTransformer.java

@@ -30,7 +30,6 @@ public class JaegerSpanToAvroRawSpanTransformer implements
   private static final String VALID_SPAN_RECEIVED_COUNT = "hypertrace.reported.spans.processed";
   private static final ConcurrentMap<String, Counter> tenantToSpanReceivedCount = new ConcurrentHashMap<>();
 
-
   private JaegerSpanNormalizer converter;
 
   @Override

span-normalizer/span-normalizer/src/test/java/org/hypertrace/core/spannormalizer/JaegerSpanNormalizerTest.java

@@ -2,12 +2,10 @@
 
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
-import io.jaegertracing.api_v2.JaegerSpanInternalModel;
 import io.jaegertracing.api_v2.JaegerSpanInternalModel.KeyValue;
 import io.jaegertracing.api_v2.JaegerSpanInternalModel.Process;
 import io.jaegertracing.api_v2.JaegerSpanInternalModel.Span;
 import io.micrometer.core.instrument.Timer;
-import java.io.File;
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.util.HashMap;
@@ -160,6 +158,83 @@ public void testServiceNameAddedToEvent_excludeTenantIds() throws Exception {
     Assertions.assertNotNull(rawSpan2);
   }
 
+  @Test
+  public void testSpanDropCriterion() throws Exception {
+    String tenantId = "tenant-" + random.nextLong();
+    Map<String, Object> configs = new HashMap<>(getCommonConfig());
+    configs.putAll(
+        Map.of(
+            "processor",
+            Map.of("tenantIdTagKey", "tenant-key",
+                "spanDropCriterion", List.of("foo:bar,k1:v1"))));
+    JaegerSpanNormalizer normalizer = JaegerSpanNormalizer.get(ConfigFactory.parseMap(configs));
+    Process process = Process.newBuilder().setServiceName("testService").build();
+    Span span1 =
+        Span.newBuilder()
+            .setProcess(process)
+            .addTags(KeyValue.newBuilder().setKey("tenant-key").setVStr(tenantId).build())
+            .addTags(KeyValue.newBuilder().setKey("foo").setVStr("bar").build())
+            .build();
+    RawSpan rawSpan1 = normalizer.convert(span1);
+    Assertions.assertNotNull(rawSpan1);
+
+    Span span2 =
+        Span.newBuilder()
+            .setProcess(process)
+            .addTags(KeyValue.newBuilder().setKey("tenant-key").setVStr(tenantId).build())
+            .addTags(KeyValue.newBuilder().setKey("foo").setVStr("bar").build())
+            .addTags(KeyValue.newBuilder().setKey("k1").setVStr("v1").build())
+            .build();
+    RawSpan rawSpan2 = normalizer.convert(span2);
+    Assertions.assertNull(rawSpan2);
+  }
+
+  @Test
+  public void testDropSpanWithMultipleCriterion() throws Exception {
+    String tenantId = "tenant-" + random.nextLong();
+    Map<String, Object> configs = new HashMap<>(getCommonConfig());
+    configs.putAll(
+        Map.of(
+            "processor",
+            Map.of("tenantIdTagKey", "tenant-key",
+                "spanDropCriterion", List.of("foo:bar,k1:v1", "k2:v2"))));
+
+    JaegerSpanNormalizer normalizer = JaegerSpanNormalizer.get(ConfigFactory.parseMap(configs));
+    Process process = Process.newBuilder().setServiceName("testService").build();
+    Span span3 =
+        Span.newBuilder()
+            .setProcess(process)
+            .addTags(KeyValue.newBuilder().setKey("tenant-key").setVStr(tenantId).build())
+            .addTags(KeyValue.newBuilder().setKey("foo").setVStr("bar").build())
+            .addTags(KeyValue.newBuilder().setKey("k2").setVStr("v2").build())
+            .build();
+    RawSpan rawSpan3 = normalizer.convert(span3);
+    Assertions.assertNull(rawSpan3);
+  }
+
+  @Test
+  public void testDropSpanWithEmptyCriterion() throws Exception {
+    String tenantId = "tenant-" + random.nextLong();
+    Map<String, Object> configs = new HashMap<>(getCommonConfig());
+    configs.putAll(
+        Map.of(
+            "processor",
+            Map.of("tenantIdTagKey", "tenant-key",
+                "spanDropCriterion", List.of())));
+
+    JaegerSpanNormalizer normalizer = JaegerSpanNormalizer.get(ConfigFactory.parseMap(configs));
+    Process process = Process.newBuilder().setServiceName("testService").build();
+    Span span3 =
+        Span.newBuilder()
+            .setProcess(process)
+            .addTags(KeyValue.newBuilder().setKey("tenant-key").setVStr(tenantId).build())
+            .addTags(KeyValue.newBuilder().setKey("foo").setVStr("bar").build())
+            .addTags(KeyValue.newBuilder().setKey("k2").setVStr("v2").build())
+            .build();
+    RawSpan rawSpan3 = normalizer.convert(span3);
+    Assertions.assertNotNull(rawSpan3);
+  }
+
   @Test
   public void testServiceNameAddedToEvent() throws Exception {
     String tenantId = "tenant-" + random.nextLong();