Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Test otel proto exporter v2 #268

Draft
wants to merge 20 commits into
base: main
Choose a base branch
from
Draft

Test otel proto exporter v2 #268

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
6497b7d
adding otel steps
kotharironak Jul 22, 2021
f9a94c4
adds support for creating topic
kotharironak Aug 11, 2021
a794db6
Merge remote-tracking branch 'origin/main' into test-otel-proto
kotharironak Aug 11, 2021
d649fe7
update the input
kotharironak Sep 17, 2021
50d54c6
Merge remote-tracking branch 'origin/main' into test-otel-proto
kotharironak Sep 17, 2021
bde5b5c
feat : basic framework layout for metrics generator and processor
kotharironak Sep 22, 2021
7ed5ae6
clean up the span normalizer
kotharironak Sep 22, 2021
f0e329b
feat: adds metrics exporter
kotharironak Sep 23, 2021
93edd27
feat : working demo e2e for ingestion side
kotharironak Sep 24, 2021
a1220e8
feat: adding exporter for direct sink from kafka
kotharironak Sep 29, 2021
47a12e8
feat: adding promethues exporter
kotharironak Sep 30, 2021
471c084
feat: working copy of pull based exporter
kotharironak Oct 1, 2021
ebfb694
wip: modification of metrics extractor in progress
kotharironak Oct 7, 2021
d1fd23f
feat: testing the timestamp changes
kotharironak Oct 8, 2021
4a6c82b
feat: added e2e stuff
kotharironak Oct 8, 2021
c2af53e
progress on using prometheus collector
kotharironak Oct 8, 2021
548a632
final changes
kotharironak Oct 8, 2021
406b68f
testing final changes
kotharironak Oct 8, 2021
0780658
feat: e2e working
kotharironak Oct 11, 2021
f465ffb
working
kotharironak Oct 13, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion ...etrics-exporter/src/main/java/org/hypertrace/metrics/exporter/MetricsExporterServlet.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

import io.opentelemetry.exporter.prometheus.PrometheusCollector;
import io.opentelemetry.sdk.metrics.export.MetricReader;
import io.prometheus.client.Collector;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.exporter.MetricsServlet;
import java.io.IOException;
import javax.servlet.ServletException;
Expand All @@ -11,9 +13,13 @@
public class MetricsExporterServlet extends MetricsServlet {
private InMemoryMetricsProducer inMemoryMetricsProducer;
private MetricReader metricReader;
private Collector collector;
private static final CollectorRegistry collectorRegistry = new CollectorRegistry(false);

public MetricsExporterServlet(InMemoryMetricsProducer producer) {
metricReader = PrometheusCollector.create().apply(producer);
super(collectorRegistry);
collector = PrometheusCollector.getNewInstance(producer);
collectorRegistry.register(collector);
inMemoryMetricsProducer = producer;
}

Expand All @@ -22,6 +28,7 @@ protected void doGet(final HttpServletRequest req, final HttpServletResponse res
throws ServletException, IOException {
try {
super.doGet(req, resp);
// List<MetricFamilySamples> test = collector.collect();
inMemoryMetricsProducer.setCommitOffset();
} catch (ServletException e) {
throw e;
Expand Down
13 changes: 8 additions & 5 deletions ...ce-metrics-generator/src/main/java/org/hypertrace/metrics/generator/MetricsExtractor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,11 +123,14 @@ private List<io.opentelemetry.proto.common.v1.KeyValue> toAttributes(Map<String,
List<io.opentelemetry.proto.common.v1.KeyValue> attributes =
labels.entrySet().stream()
.map(
k ->
io.opentelemetry.proto.common.v1.KeyValue.newBuilder()
.setKey(k.getKey())
.setValue(AnyValue.newBuilder().setStringValue(k.getValue()).build())
.build())
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;
}
Expand Down
12 changes: 11 additions & 1 deletion ...ce-metrics-generator/src/main/java/org/hypertrace/metrics/generator/OtelMetricsSerde.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,11 @@
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class OtelMetricsSerde implements Serde<ResourceMetrics> {
private static final Logger LOGGER = LoggerFactory.getLogger(OtelMetricsSerde.class);

@Override
public void configure(Map<String, ?> configs, boolean isKey) {}
Expand All @@ -28,16 +31,23 @@ public Deserializer<ResourceMetrics> deserializer() {
public static class Ser implements Serializer<ResourceMetrics> {
@Override
public byte[] serialize(String topic, ResourceMetrics data) {
return data.toByteArray();
try {
return data.toByteArray();
} catch (Exception e) {
LOGGER.error("serilzation error:", e);
}
return null;
}
}

public static class De implements Deserializer<ResourceMetrics> {
@Override
public ResourceMetrics deserialize(String topic, byte[] data) {
try {
LOGGER.info("deserilize:{}", OtelMetricsSerde.class.getName());
return ResourceMetrics.parseFrom(data);
} catch (InvalidProtocolBufferException e) {
LOGGER.error("error:", e);
throw new RuntimeException(e);
}
}
Expand Down
4 changes: 4 additions & 0 deletions ...ace-metrics-processor/src/main/java/org/hypertrace/metrics/processor/MetricsEnricher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,16 +4,20 @@
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.Transformer;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MetricsEnricher
implements Transformer<byte[], ResourceMetrics, KeyValue<byte[], ResourceMetrics>> {
private static final Logger logger = LoggerFactory.getLogger(MetricsEnricher.class);

@Override
public void init(ProcessorContext context) {}

@Override
public KeyValue<byte[], ResourceMetrics> transform(byte[] key, ResourceMetrics value) {
// noop enricher for now
logger.info("got the value in MetricsEnricher");
return new KeyValue<>(key, value);
}

Expand Down
5 changes: 5 additions & 0 deletions ...ce-metrics-processor/src/main/java/org/hypertrace/metrics/processor/OtlpMetricsSerde.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,11 @@
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class OtlpMetricsSerde implements Serde<ResourceMetrics> {
private static final Logger logger = LoggerFactory.getLogger(OtlpMetricsSerde.class);

@Override
public void configure(Map<String, ?> configs, boolean isKey) {}
Expand Down Expand Up @@ -36,8 +39,10 @@ public static class De implements Deserializer<ResourceMetrics> {
@Override
public ResourceMetrics deserialize(String topic, byte[] data) {
try {
logger.info("Desereized from: {}", OtlpMetricsSerde.class.getName());
return ResourceMetrics.parseFrom(data);
} catch (InvalidProtocolBufferException e) {
logger.error("Failed in deserilzed", e);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
logger.error("Failed in deserilzed", e);
logger.error("Failed in deserialized", e);

throw new RuntimeException(e);
}
}
Expand Down