Add OpenTelemetry Proto Metrics Reader Extension

extensions-contrib/opentelemetry-extensions/README.md

@@ -0,0 +1,49 @@
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one
+  ~ or more contributor license agreements.  See the NOTICE file
+  ~ distributed with this work for additional information
+  ~ regarding copyright ownership.  The ASF licenses this file
+  ~ to you under the Apache License, Version 2.0 (the
+  ~ "License"); you may not use this file except in compliance
+  ~ with the License.  You may obtain a copy of the License at
+  ~
+  ~   http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing,
+  ~ software distributed under the License is distributed on an
+  ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  ~ KIND, either express or implied.  See the License for the
+  ~ specific language governing permissions and limitations
+  ~ under the License.
+  -->
+
+# OpenTelemetry Extensions
+
+The [OpenTelemetry](https://opentelemetry.io/) extensions provides the ability to read metrics in OTLP format
+
+## Configuration
+
+### How to enable and use the extension
+
+To enable the OpenTelemetry extensions, add the extension and enable the emitter in `common.runtime.properties`.
+
+Load the plugin:
+
+```properties
+druid.extensions.loadList=[..., "druid-opentelemetry-extensions"]
+```
+
+Now Sumbit the Supervisor Config with [source input format](https://druid.apache.org/docs/latest/ingestion/data-formats/) as:
+
+```
+"ioConfig": {
+  "topic": "topic_name",
+  "inputFormat": {
+    "type": "opentelemetry-metrics-protobuf",
+    "metricDimension": "name"
+  }
+  .
+  .
+  .
+}
+```

extensions-contrib/opentelemetry-extensions/pom.xml

@@ -0,0 +1,102 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one
+  ~ or more contributor license agreements.  See the NOTICE file
+  ~ distributed with this work for additional information
+  ~ regarding copyright ownership.  The ASF licenses this file
+  ~ to you under the Apache License, Version 2.0 (the
+  ~ "License"); you may not use this file except in compliance
+  ~ with the License.  You may obtain a copy of the License at
+  ~
+  ~   http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing,
+  ~ software distributed under the License is distributed on an
+  ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  ~ KIND, either express or implied.  See the License for the
+  ~ specific language governing permissions and limitations
+  ~ under the License.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>org.apache.druid.extensions.contrib</groupId>
+    <artifactId>druid-opentelemetry-extensions</artifactId>
+    <name>druid-opentelemetry-extensions</name>
+    <description>druid-opentelemetry-extensions</description>
+
+    <parent>
+        <artifactId>druid</artifactId>
+        <groupId>org.apache.druid</groupId>
+        <version>33.0.0-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+    <dependencies>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry.proto</groupId>
+            <artifactId>opentelemetry-proto</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.google.inject</groupId>
+            <artifactId>guice</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.google.code.findbugs</groupId>
+            <artifactId>jsr305</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.druid</groupId>
+            <artifactId>druid-processing</artifactId>
+            <version>${project.parent.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.druid</groupId>
+            <artifactId>druid-indexing-service</artifactId>
+            <version>${project.parent.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <!-- test -->
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <!-- jmh -->
+        <dependency>
+            <groupId>org.openjdk.jmh</groupId>
+            <artifactId>jmh-core</artifactId>
+            <version>1.27</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.openjdk.jmh</groupId>
+            <artifactId>jmh-generator-annprocess</artifactId>
+            <version>1.27</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

extensions-contrib/opentelemetry-extensions/src/main/java/org/apache/druid/data/input/opentelemetry/protobuf/OpenTelemetryMetricsProtobufInputFormat.java

@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.data.input.opentelemetry.protobuf;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.data.input.InputEntity;
+import org.apache.druid.data.input.InputEntityReader;
+import org.apache.druid.data.input.InputFormat;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.impl.ByteEntity;
+import org.apache.druid.indexing.seekablestream.SettableByteEntity;
+import org.apache.druid.java.util.common.StringUtils;
+
+import java.io.File;
+import java.util.Objects;
+
+public class OpenTelemetryMetricsProtobufInputFormat implements InputFormat
+{
+  private static final String DEFAULT_METRIC_DIMENSION = "metric";
+  private static final String DEFAULT_VALUE_DIMENSION = "value";
+  private static final String DEFAULT_RESOURCE_PREFIX = "resource.";
+
+  private final String metricDimension;
+  private final String valueDimension;
+  private final String metricAttributePrefix;
+  private final String resourceAttributePrefix;
+
+  public OpenTelemetryMetricsProtobufInputFormat(
+      @JsonProperty("metricDimension") String metricDimension,
+      @JsonProperty("valueDimension") String valueDimension,
+      @JsonProperty("metricAttributePrefix") String metricAttributePrefix,
+      @JsonProperty("resourceAttributePrefix") String resourceAttributePrefix
+  )
+  {
+    this.metricDimension = metricDimension != null ? metricDimension : DEFAULT_METRIC_DIMENSION;
+    this.valueDimension = valueDimension != null ? valueDimension : DEFAULT_VALUE_DIMENSION;
+    this.metricAttributePrefix = StringUtils.nullToEmptyNonDruidDataString(metricAttributePrefix);
+    this.resourceAttributePrefix = resourceAttributePrefix != null ? resourceAttributePrefix : DEFAULT_RESOURCE_PREFIX;
+  }
+
+  @Override
+  public boolean isSplittable()
+  {
+    return false;
+  }
+
+  @Override
+  public InputEntityReader createReader(InputRowSchema inputRowSchema, InputEntity source, File temporaryDirectory)
+  {
+    // Sampler passes a KafkaRecordEntity directly, while the normal code path wraps the same entity in a
+    // SettableByteEntity
+    SettableByteEntity<? extends ByteEntity> settableEntity;
+    if (source instanceof SettableByteEntity) {
+      settableEntity = (SettableByteEntity<? extends ByteEntity>) source;
+    } else {
+      SettableByteEntity<ByteEntity> wrapper = new SettableByteEntity<>();
+      wrapper.setEntity((ByteEntity) source);
+      settableEntity = wrapper;
+    }
+    return new OpenTelemetryMetricsProtobufReader(
+            inputRowSchema.getDimensionsSpec(),
+            settableEntity,
+            metricDimension,
+            valueDimension,
+            metricAttributePrefix,
+            resourceAttributePrefix
+    );
+  }
+
+  @JsonProperty
+  public String getMetricDimension()
+  {
+    return metricDimension;
+  }
+
+  @JsonProperty
+  public String getValueDimension()
+  {
+    return valueDimension;
+  }
+
+  @JsonProperty
+  public String getMetricAttributePrefix()
+  {
+    return metricAttributePrefix;
+  }
+
+  @JsonProperty
+  public String getResourceAttributePrefix()
+  {
+    return resourceAttributePrefix;
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof OpenTelemetryMetricsProtobufInputFormat)) {
+      return false;
+    }
+    OpenTelemetryMetricsProtobufInputFormat that = (OpenTelemetryMetricsProtobufInputFormat) o;
+    return Objects.equals(metricDimension, that.metricDimension)
+            && Objects.equals(valueDimension, that.valueDimension)
+            && Objects.equals(metricAttributePrefix, that.metricAttributePrefix)
+            && Objects.equals(resourceAttributePrefix, that.resourceAttributePrefix);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(metricDimension, valueDimension, metricAttributePrefix, resourceAttributePrefix);
+  }
+}