feat: adds metrics processor component and topology test #270
Conversation
kotharironak
requested review from
a team,
laxmanchekka,
findingrish and
surajpuvvada
Codecov Report
@@ Coverage Diff @@
## main #270 +/- ##
============================================
- Coverage 80.29% 80.18% -0.11%
- Complexity 1166 1180 +14
============================================
Files 102 106 +4
Lines 4542 4588 +46
Branches 423 424 +1
============================================
+ Hits 3647 3679 +32
- Misses 696 709 +13
- Partials 199 200 +1
Flags with carried forward coverage won't be shown. Click here to find out more.
Continue to review full report at Codecov.
|
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
| @Override | ||
| public KeyValue<byte[], ResourceMetrics> transform(byte[] key, ResourceMetrics value) { | ||
| // noop enricher for now | ||
| return new KeyValue<>(key, value); |
There was a problem hiding this comment.
In the first iteration, we will be extracting metrics from traces, and they will already have enriched attributes as labels. We will enhance this later, for now, they are just pass through. If requires, we can check if the metrics contain API and service id, and if not, we can drop them.
| implementation("org.hypertrace.core.kafkastreams.framework:kafka-streams-framework:0.1.21") | ||
|
|
||
| // serde (todo) | ||
| // implementation("io.confluent:kafka-streams-protobuf-serde:6.0.1") |
There was a problem hiding this comment.
don't you need this ?
There was a problem hiding this comment.
Currently, I am using internally defined OltpMetricsSerde. I was facing some issue with this so working as follow up item on this. For now, removed this.
| try { | ||
| return ResourceMetrics.parseFrom(data); | ||
| } catch (InvalidProtocolBufferException e) { | ||
| throw new RuntimeException(e); |
There was a problem hiding this comment.
assuming an exception handler is part of config that can deal with any exceptions ?
There was a problem hiding this comment.
Yes, this will be handle by default for deserialization exception handler - LogAndContinueExceptionHandler. So it will just be logged and such messages will be dropped.
|
|
||
| schema.registry.url = "http://localhost:8081" | ||
| schema.registry.url = ${?SCHEMA_REGISTRY_URL} | ||
| value.subject.name.strategy = "io.confluent.kafka.serializers.subject.TopicRecordNameStrategy" |
There was a problem hiding this comment.
why is it TopicRecordNameStrategy ? are there going to be different message types sent on the topic ?
There was a problem hiding this comment.
are there going to be different message types sent on the topic ?
No. Removed it.
This comment has been minimized.
This comment has been minimized.
Description
As part of building a metrics pipeline, this PR adds the new metrics process component.
Ref : hypertrace/hypertrace#297
Testing
Have added the topology test which ingests the gauge metric and verify.
Checklist: