feat: add metrics generator for num_calls

.github/workflows/pr-build.yml

@@ -40,7 +40,7 @@ jobs:
       - name: Build with Gradle
         uses: hypertrace/github-actions/gradle@main
         with: 
-          args: build dockerBuildImages
+          args: build -x :hypertrace-metrics-generator:hypertrace-metrics-generator-api:avroCompatibilityCheck dockerBuildImages
 
   validate-helm-charts:
     runs-on: ubuntu-20.04

hypertrace-metrics-generator/build.gradle.kts

@@ -0,0 +1,3 @@
+subprojects {
+  group = "org.hypertrace.metrics.generator"
+}

hypertrace-metrics-generator/hypertrace-metrics-generator-api/build.gradle.kts

@@ -0,0 +1,13 @@
+plugins {
+  `java-library`
+  id("org.hypertrace.publish-plugin")
+  id("org.hypertrace.avro-plugin")
+}
+
+dependencies {
+  api("org.apache.avro:avro:1.10.2")
+}
+
+tasks.named<org.hypertrace.gradle.avro.CheckAvroCompatibility>("avroCompatibilityCheck") {
+  enabled = false
+}

hypertrace-metrics-generator/hypertrace-metrics-generator-api/src/main/avro/metric-identity.avdl

@@ -0,0 +1,7 @@
+@namespace("org.hypertrace.metrics.generator.api")
+protocol MetricIdentityProtocol {
+	record MetricIdentity {
+		long timestamp_millis = 0;
+		union {null, string} metric_key = null;
+	}
+}

hypertrace-metrics-generator/hypertrace-metrics-generator-api/src/main/avro/metric.avdl

@@ -0,0 +1,9 @@
+@namespace("org.hypertrace.metrics.generator.api")
+protocol MetricProtocol {
+	record Metric {
+		union {null, string} name = null;
+		map<string> attributes = {};
+		union {null, string} description = null;
+		union {null, string} unit = null;
+	}
+}

hypertrace-metrics-generator/hypertrace-metrics-generator/build.gradle.kts

@@ -0,0 +1,43 @@
+plugins {
+  java
+  application
+  jacoco
+  id("org.hypertrace.docker-java-application-plugin")
+  id("org.hypertrace.docker-publish-plugin")
+  id("org.hypertrace.jacoco-report-plugin")
+}
+
+application {
+  mainClass.set("org.hypertrace.core.serviceframework.PlatformServiceLauncher")
+}
+
+hypertraceDocker {
+  defaultImage {
+    javaApplication {
+      serviceName.set("${project.name}")
+      adminPort.set(8099)
+    }
+  }
+}
+
+tasks.test {
+  useJUnitPlatform()
+}
+
+dependencies {
+  // common and framework
+  implementation(project(":hypertrace-metrics-generator:hypertrace-metrics-generator-api"))
+  implementation(project(":hypertrace-view-generator:hypertrace-view-generator-api"))
+  implementation("org.hypertrace.core.serviceframework:platform-service-framework:0.1.30-SNAPSHOT")
+  implementation("org.hypertrace.core.serviceframework:platform-metrics:0.1.30-SNAPSHOT")
+  implementation("org.hypertrace.core.kafkastreams.framework:kafka-streams-framework:0.1.21")
+
+  // open telemetry proto
+  implementation("io.opentelemetry:opentelemetry-proto:1.6.0-alpha")
+
+  // test
+  testImplementation("org.junit.jupiter:junit-jupiter:5.7.1")
+  testImplementation("org.mockito:mockito-core:3.8.0")
+  testImplementation("org.junit-pioneer:junit-pioneer:1.3.8")
+  testImplementation("org.apache.kafka:kafka-streams-test-utils:6.0.1-ccs")
+}

hypertrace-metrics-generator/hypertrace-metrics-generator/src/main/java/org/hypertrace/metrics/generator/MetricEmitPunctuator.java

@@ -0,0 +1,132 @@
+package org.hypertrace.metrics.generator;
+
+import io.opentelemetry.proto.common.v1.AnyValue;
+import io.opentelemetry.proto.common.v1.InstrumentationLibrary;
+import io.opentelemetry.proto.common.v1.KeyValue;
+import io.opentelemetry.proto.metrics.v1.Gauge;
+import io.opentelemetry.proto.metrics.v1.InstrumentationLibraryMetrics;
+import io.opentelemetry.proto.metrics.v1.NumberDataPoint;
+import io.opentelemetry.proto.metrics.v1.ResourceMetrics;
+import io.opentelemetry.proto.resource.v1.Resource;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.To;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.hypertrace.metrics.generator.api.Metric;
+import org.hypertrace.metrics.generator.api.MetricIdentity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MetricEmitPunctuator implements Punctuator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(MetricEmitPunctuator.class);
+
+  private static final String RESOURCE_KEY_SERVICE = "service";
+  private static final String RESOURCE_KEY_SERVICE_VALUE = "metrics-generator";
+  private static final String INSTRUMENTATION_LIB_NAME = "Generated-From-View";
+
+  private MetricIdentity key;
+  private ProcessorContext context;
+  private KeyValueStore<MetricIdentity, Long> metricIdentityStore;
+  private KeyValueStore<MetricIdentity, Metric> metricStore;
+  private Cancellable cancellable;
+  private To outputTopicProducer;
+
+  public MetricEmitPunctuator(
+      MetricIdentity key,
+      ProcessorContext context,
+      KeyValueStore<MetricIdentity, Long> metricIdentityStore,
+      KeyValueStore<MetricIdentity, Metric> metricStore,
+      To outputTopicProducer) {
+    this.key = key;
+    this.context = context;
+    this.metricIdentityStore = metricIdentityStore;
+    this.metricStore = metricStore;
+    this.outputTopicProducer = outputTopicProducer;
+  }
+
+  public void setCancellable(Cancellable cancellable) {
+    this.cancellable = cancellable;
+  }
+
+  @Override
+  public void punctuate(long timestamp) {
+    // always cancel the punctuator else it will get re-scheduled automatically
+    cancellable.cancel();
+
+    // read the value from a key
+    Long value = metricIdentityStore.get(this.key);
+    if (value != null) {
+      long diff = timestamp - this.key.getTimestampMillis();
+      LOGGER.debug("Metrics with key:{} is emitted after duration {}", this.key, diff);
+
+      Metric metric = metricStore.get(this.key);
+      metricIdentityStore.delete(this.key);
+      metricStore.delete(this.key);
+      // convert to Resource Metrics
+      ResourceMetrics resourceMetrics = convertToResourceMetric(this.key, value, metric);
+      context.forward(null, resourceMetrics, outputTopicProducer);
+    } else {
+      LOGGER.debug("The value for metrics with key:{} is null", this.key);
+    }
+  }
+
+  private ResourceMetrics convertToResourceMetric(
+      MetricIdentity metricIdentity, Long value, Metric metric) {
+    ResourceMetrics.Builder resourceMetricsBuilder = ResourceMetrics.newBuilder();
+    resourceMetricsBuilder.setResource(
+        Resource.newBuilder()
+            .addAttributes(
+                io.opentelemetry.proto.common.v1.KeyValue.newBuilder()
+                    .setKey(RESOURCE_KEY_SERVICE)
+                    .setValue(
+                        AnyValue.newBuilder().setStringValue(RESOURCE_KEY_SERVICE_VALUE).build())
+                    .build()));
+
+    io.opentelemetry.proto.metrics.v1.Metric.Builder metricBuilder =
+        io.opentelemetry.proto.metrics.v1.Metric.newBuilder();
+    metricBuilder.setName(metric.getName());
+    metricBuilder.setDescription(metric.getDescription());
+    metricBuilder.setUnit(metric.getUnit());
+
+    NumberDataPoint.Builder numberDataPointBuilder = NumberDataPoint.newBuilder();
+    List<KeyValue> attributes = toAttributes(metric.getAttributes());
+    numberDataPointBuilder.addAllAttributes(attributes);
+    numberDataPointBuilder.setTimeUnixNano(
+        TimeUnit.NANOSECONDS.convert(metricIdentity.getTimestampMillis(), TimeUnit.MILLISECONDS));
+    numberDataPointBuilder.setAsInt(value);
+
+    Gauge.Builder gaugeBuilder = Gauge.newBuilder();
+    gaugeBuilder.addDataPoints(numberDataPointBuilder.build());
+    metricBuilder.setGauge(gaugeBuilder.build());
+
+    resourceMetricsBuilder.addInstrumentationLibraryMetrics(
+        InstrumentationLibraryMetrics.newBuilder()
+            .addMetrics(metricBuilder.build())
+            .setInstrumentationLibrary(
+                InstrumentationLibrary.newBuilder().setName(INSTRUMENTATION_LIB_NAME).build())
+            .build());
+
+    return resourceMetricsBuilder.build();
+  }
+
+  private List<io.opentelemetry.proto.common.v1.KeyValue> toAttributes(Map<String, String> labels) {
+    List<io.opentelemetry.proto.common.v1.KeyValue> attributes =
+        labels.entrySet().stream()
+            .map(
+                k -> {
+                  io.opentelemetry.proto.common.v1.KeyValue.Builder keyValueBuilder =
+                      io.opentelemetry.proto.common.v1.KeyValue.newBuilder();
+                  keyValueBuilder.setKey(k.getKey());
+                  String value = k.getValue() != null ? k.getValue() : "";
+                  keyValueBuilder.setValue(AnyValue.newBuilder().setStringValue(value).build());
+                  return keyValueBuilder.build();
+                })
+            .collect(Collectors.toList());
+    return attributes;
+  }
+}

hypertrace-metrics-generator/hypertrace-metrics-generator/src/main/java/org/hypertrace/metrics/generator/MetricsGenerator.java

@@ -0,0 +1,124 @@
+package org.hypertrace.metrics.generator;
+
+import com.typesafe.config.Config;
+import io.opentelemetry.proto.metrics.v1.ResourceMetrics;
+import java.util.List;
+import java.util.Map;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.Serdes.LongSerde;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.Named;
+import org.apache.kafka.streams.kstream.Produced;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.Stores;
+import org.hypertrace.core.kafkastreams.framework.KafkaStreamsApp;
+import org.hypertrace.core.serviceframework.config.ConfigClient;
+import org.hypertrace.metrics.generator.api.Metric;
+import org.hypertrace.metrics.generator.api.MetricIdentity;
+import org.hypertrace.viewgenerator.api.RawServiceView;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MetricsGenerator extends KafkaStreamsApp {
+  private static final Logger LOGGER = LoggerFactory.getLogger(MetricsGenerator.class);
+  public static final String INPUT_TOPIC_CONFIG_KEY = "input.topic";
+  public static final String OUTPUT_TOPIC_CONFIG_KEY = "output.topic";
+  public static final String METRICS_GENERATOR_JOB_CONFIG = "metrics-generator-job-config";
+  public static final String METRICS_IDENTITY_STORE = "metric-identity-store";
+  public static final String METRICS_IDENTITY_VALUE_STORE = "metric-identity-value-Store";
+  public static final String OUTPUT_TOPIC_PRODUCER = "output-topic-producer";
+
+  public MetricsGenerator(ConfigClient configClient) {
+    super(configClient);
+  }
+
+  @Override
+  public StreamsBuilder buildTopology(
+      Map<String, Object> streamsProperties,
+      StreamsBuilder streamsBuilder,
+      Map<String, KStream<?, ?>> inputStreams) {
+
+    Config jobConfig = getJobConfig(streamsProperties);
+    String inputTopic = jobConfig.getString(INPUT_TOPIC_CONFIG_KEY);
+    String outputTopic = jobConfig.getString(OUTPUT_TOPIC_CONFIG_KEY);
+
+    KStream<String, RawServiceView> inputStream =
+        (KStream<String, RawServiceView>) inputStreams.get(inputTopic);
+    if (inputStream == null) {
+      inputStream = streamsBuilder.stream(inputTopic, Consumed.with(Serdes.String(), (Serde) null));
+      inputStreams.put(inputTopic, inputStream);
+    }
+
+    // Retrieve the default value serde defined in config and use it
+    Serde valueSerde = defaultValueSerde(streamsProperties);
+    Serde keySerde = defaultKeySerde(streamsProperties);
+
+    StoreBuilder<KeyValueStore<MetricIdentity, Long>> metricIdentityStoreBuilder =
+        Stores.keyValueStoreBuilder(
+                Stores.persistentKeyValueStore(METRICS_IDENTITY_STORE), keySerde, new LongSerde())
+            .withCachingEnabled();
+
+    StoreBuilder<KeyValueStore<MetricIdentity, Metric>> metricIdentityToValueStoreBuilder =
+        Stores.keyValueStoreBuilder(
+                Stores.persistentKeyValueStore(METRICS_IDENTITY_VALUE_STORE), keySerde, valueSerde)
+            .withCachingEnabled();
+
+    streamsBuilder.addStateStore(metricIdentityStoreBuilder);
+    streamsBuilder.addStateStore(metricIdentityToValueStoreBuilder);
+
+    Produced<byte[], ResourceMetrics> outputTopicProducer =
+        Produced.with(Serdes.ByteArray(), new OtlpMetricsSerde());
+    outputTopicProducer = outputTopicProducer.withName(OUTPUT_TOPIC_PRODUCER);
+
+    inputStream
+        .transform(
+            MetricsProcessor::new,
+            Named.as(MetricsProcessor.class.getSimpleName()),
+            METRICS_IDENTITY_STORE,
+            METRICS_IDENTITY_VALUE_STORE)
+        .to(outputTopic, outputTopicProducer);
+
+    return streamsBuilder;
+  }
+
+  @Override
+  public String getJobConfigKey() {
+    return METRICS_GENERATOR_JOB_CONFIG;
+  }
+
+  @Override
+  public Logger getLogger() {
+    return LOGGER;
+  }
+
+  @Override
+  public List<String> getInputTopics(Map<String, Object> properties) {
+    Config jobConfig = getJobConfig(properties);
+    return List.of(jobConfig.getString(INPUT_TOPIC_CONFIG_KEY));
+  }
+
+  @Override
+  public List<String> getOutputTopics(Map<String, Object> properties) {
+    Config jobConfig = getJobConfig(properties);
+    return List.of(jobConfig.getString(OUTPUT_TOPIC_CONFIG_KEY));
+  }
+
+  private Config getJobConfig(Map<String, Object> properties) {
+    return (Config) properties.get(getJobConfigKey());
+  }
+
+  private Serde defaultValueSerde(Map<String, Object> properties) {
+    StreamsConfig config = new StreamsConfig(properties);
+    return config.defaultValueSerde();
+  }
+
+  private Serde defaultKeySerde(Map<String, Object> properties) {
+    StreamsConfig config = new StreamsConfig(properties);
+    return config.defaultKeySerde();
+  }
+}