-
Notifications
You must be signed in to change notification settings - Fork 11
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(canary): add run wf latency and duplicated task metrics (#661)
- Calculates any latency and duplicates tasks by lh cluster - Using functional semantic - Remove metrics if they are not present any more in the store
- Loading branch information
Showing
16 changed files
with
501 additions
and
234 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
75 changes: 0 additions & 75 deletions
75
canary/src/main/java/io/littlehorse/canary/aggregator/internal/LatencyMetricExporter.java
This file was deleted.
Oops, something went wrong.
65 changes: 65 additions & 0 deletions
65
...ry/src/main/java/io/littlehorse/canary/aggregator/topology/DuplicatedTaskRunTopology.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| package io.littlehorse.canary.aggregator.topology; | ||
|
|
||
| import io.littlehorse.canary.aggregator.serdes.ProtobufSerdes; | ||
| import io.littlehorse.canary.proto.Beat; | ||
| import io.littlehorse.canary.proto.BeatKey; | ||
| import io.littlehorse.canary.proto.MetricKey; | ||
| import java.time.Duration; | ||
| import lombok.Getter; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.kafka.common.serialization.Serdes; | ||
| import org.apache.kafka.common.utils.Bytes; | ||
| import org.apache.kafka.streams.kstream.Grouped; | ||
| import org.apache.kafka.streams.kstream.KStream; | ||
| import org.apache.kafka.streams.kstream.Materialized; | ||
| import org.apache.kafka.streams.state.KeyValueStore; | ||
|
|
||
| @Getter | ||
| @Slf4j | ||
| public class DuplicatedTaskRunTopology { | ||
|
|
||
| private final KStream<MetricKey, Double> stream; | ||
|
|
||
| public DuplicatedTaskRunTopology(final KStream<BeatKey, Beat> mainStream, final Duration storeRetetion) { | ||
| stream = mainStream | ||
| .filter((key, value) -> value.hasTaskRunBeat()) | ||
| .groupByKey() | ||
| // count all the records with the same idempotency key and attempt number | ||
| .count(Materialized.<BeatKey, Long, KeyValueStore<Bytes, byte[]>>with( | ||
| ProtobufSerdes.BeatKey(), Serdes.Long()) | ||
| .withRetention(storeRetetion)) | ||
| // filter by duplicated | ||
| .filter((key, value) -> value > 1L) | ||
| .toStream() | ||
| .mapValues((readOnlyKey, value) -> Double.valueOf(value)) | ||
| // debug peek aggregate | ||
| .peek(DuplicatedTaskRunTopology::peekAggregate) | ||
| // re-key from task run to lh cluster | ||
| .groupBy((key, value) -> toMetricKey(key), Grouped.with(ProtobufSerdes.MetricKey(), Serdes.Double())) | ||
| // count how many task were duplicated | ||
| .count(Materialized.<MetricKey, Long, KeyValueStore<Bytes, byte[]>>with( | ||
| ProtobufSerdes.MetricKey(), Serdes.Long()) | ||
| .withRetention(storeRetetion)) | ||
| .mapValues((readOnlyKey, value) -> Double.valueOf(value)) | ||
| .toStream(); | ||
| } | ||
|
|
||
| private static MetricKey toMetricKey(final BeatKey key) { | ||
| return MetricKey.newBuilder() | ||
| .setId("duplicated_task_run_max_count") | ||
| .setServerHost(key.getServerHost()) | ||
| .setServerPort(key.getServerPort()) | ||
| .setServerVersion(key.getServerVersion()) | ||
| .build(); | ||
| } | ||
|
|
||
| private static void peekAggregate(final BeatKey key, final Double count) { | ||
| log.debug( | ||
| "server={}:{}, idempotency_key={}, attempt_number={}, count={}", | ||
| key.getServerHost(), | ||
| key.getServerPort(), | ||
| key.getTaskRunBeatKey().getIdempotencyKey(), | ||
| key.getTaskRunBeatKey().getAttemptNumber(), | ||
| count); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
52 changes: 52 additions & 0 deletions
52
canary/src/main/java/io/littlehorse/canary/aggregator/topology/MetricsTopology.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,52 @@ | ||
| package io.littlehorse.canary.aggregator.topology; | ||
|
|
||
| import io.littlehorse.canary.aggregator.internal.BeatTimeExtractor; | ||
| import io.littlehorse.canary.aggregator.serdes.ProtobufSerdes; | ||
| import io.littlehorse.canary.proto.Beat; | ||
| import io.littlehorse.canary.proto.BeatKey; | ||
| import io.littlehorse.canary.proto.MetricKey; | ||
| import java.time.Duration; | ||
| import org.apache.kafka.common.serialization.Serdes; | ||
| import org.apache.kafka.common.utils.Bytes; | ||
| import org.apache.kafka.streams.StreamsBuilder; | ||
| import org.apache.kafka.streams.Topology; | ||
| import org.apache.kafka.streams.kstream.Consumed; | ||
| import org.apache.kafka.streams.kstream.KStream; | ||
| import org.apache.kafka.streams.kstream.Materialized; | ||
| import org.apache.kafka.streams.kstream.TimeWindows; | ||
| import org.apache.kafka.streams.state.KeyValueStore; | ||
|
|
||
| public class MetricsTopology { | ||
|
|
||
| private static final Consumed<BeatKey, Beat> BEATS_SERDES = Consumed.with( | ||
| ProtobufSerdes.BeatKey(), ProtobufSerdes.Beat()) | ||
| .withTimestampExtractor(new BeatTimeExtractor()); | ||
|
|
||
| public static final String METRICS_STORE = "metrics"; | ||
| private final StreamsBuilder streamsBuilder; | ||
|
|
||
| public MetricsTopology(final String inputTopic, final long storeRetention) { | ||
| streamsBuilder = new StreamsBuilder(); | ||
|
|
||
| final KStream<BeatKey, Beat> beatsStream = streamsBuilder.stream(inputTopic, BEATS_SERDES); | ||
|
|
||
| final LatencyTopology latencyTopology = new LatencyTopology( | ||
| beatsStream, | ||
| TimeWindows.ofSizeAndGrace(Duration.ofMinutes(1), Duration.ofSeconds(5)), | ||
| Duration.ofMillis(storeRetention)); | ||
| final DuplicatedTaskRunTopology taskRunTopology = | ||
| new DuplicatedTaskRunTopology(beatsStream, Duration.ofMillis(storeRetention)); | ||
|
|
||
| latencyTopology | ||
| .getStream() | ||
| .merge(taskRunTopology.getStream()) | ||
| .toTable(Materialized.<MetricKey, Double, KeyValueStore<Bytes, byte[]>>as(METRICS_STORE) | ||
| .withKeySerde(ProtobufSerdes.MetricKey()) | ||
| .withValueSerde(Serdes.Double()) | ||
| .withRetention(Duration.ofMillis(storeRetention))); | ||
| } | ||
|
|
||
| public Topology toTopology() { | ||
| return streamsBuilder.build(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.