feat(canary): add mandatory config metronome.server.dataplane.id (#1205)

Co-authored-by: Saul Piña <[email protected]>

canary/canary.properties

@@ -15,3 +15,4 @@ lh.canary.metronome.run.threads=1
 lh.canary.metronome.run.requests=300
 lh.canary.metronome.run.sample.rate=1
 lh.canary.metronome.server.id=lh
+lh.canary.metronome.server.dataplane.id=localhost

canary/src/main/java/io/littlehorse/canary/Main.java

@@ -67,6 +67,7 @@ private static void initialize(final String[] args) throws IOException {
                     lhConfig.getApiBootstrapHost(),
                     lhConfig.getApiBootstrapPort(),
                     lhClient.getServerVersion(),
+                    canaryConfig.getMetronomeServerDataplaneId(),
                     canaryConfig.getMetronomeServerId(),
                     canaryConfig.getTopicName(),
                     canaryConfig.toKafkaConfig().toMap(),

canary/src/main/java/io/littlehorse/canary/aggregator/internal/MetricStoreExporter.java

@@ -43,6 +43,7 @@ private static List<Tag> toTags(final MetricKey key) {
         tags.add(Tag.of("server", "%s:%s".formatted(key.getServerHost(), key.getServerPort())));
         tags.add(Tag.of("server_version", key.getServerVersion()));
         tags.add(Tag.of("server_id", key.getServerId()));
+        tags.add(Tag.of("dataplane_id", key.getDataplaneId()));
         tags.addAll(key.getTagsList().stream()
                 .map(tag -> Tag.of(tag.getKey(), tag.getValue()))
                 .toList());

canary/src/main/java/io/littlehorse/canary/aggregator/topology/MetricsTopology.java

@@ -173,6 +173,7 @@ private static MetricKey buildMetricKey(final BeatKey key, final String id) {
                 .setServerPort(key.getServerPort())
                 .setServerHost(key.getServerHost())
                 .setServerId(key.getServerId())
+                .setDataplaneId(key.getDataplaneId())
                 .setId("canary_%s".formatted(id));
 
         if (key.hasStatus() && !Strings.isNullOrEmpty(key.getStatus())) {
@@ -206,6 +207,7 @@ private static BeatKey removeWfId(final BeatKey key, final BeatValue value) {
                 .setServerPort(key.getServerPort())
                 .setStatus(key.getStatus())
                 .setServerId(key.getServerId())
+                .setDataplaneId(key.getDataplaneId())
                 .addAllTags(key.getTagsList())
                 .build();
     }

canary/src/main/java/io/littlehorse/canary/config/CanaryConfig.java

@@ -33,6 +33,7 @@ public class CanaryConfig implements Config {
     public static final String METRONOME_WORKER_ENABLE = "metronome.worker.enable";
     public static final String METRONOME_DATA_PATH = "metronome.data.path";
     public static final String METRONOME_SERVER_ID = "metronome.server.id";
+    public static final String METRONOME_SERVER_DATAPLANE_ID = "metronome.server.dataplane.id";
     public static final String METRONOME_BEAT_EXTRA_TAGS = "metronome.beat.extra.tags";
     public static final String METRONOME_BEAT_EXTRA_TAGS_PREFIX = "%s.".formatted(METRONOME_BEAT_EXTRA_TAGS);
 
@@ -190,4 +191,8 @@ public String getMetronomeDataPath() {
     public String getMetronomeServerId() {
         return getConfig(METRONOME_SERVER_ID);
     }
+
+    public String getMetronomeServerDataplaneId() {
+        return getConfig(METRONOME_SERVER_DATAPLANE_ID);
+    }
 }

canary/src/main/java/io/littlehorse/canary/metronome/internal/BeatProducer.java

@@ -26,19 +26,22 @@ public class BeatProducer {
     private final String lhServerVersion;
     private final String topicName;
     private final String lhServerId;
+    private final String dataplaneId;
 
     public BeatProducer(
             final String lhServerHost,
             final int lhServerPort,
             final String lhServerVersion,
             final String lhServerId,
+            final String dataplaneId,
             final String topicName,
             final Map<String, Object> producerConfig,
             final Map<String, String> extraTags) {
         this.lhServerHost = lhServerHost;
         this.lhServerPort = lhServerPort;
         this.lhServerVersion = lhServerVersion;
         this.lhServerId = lhServerId;
+        this.dataplaneId = dataplaneId;
         this.topicName = topicName;
         this.extraTags = extraTags;
 
@@ -93,6 +96,7 @@ private BeatKey buildKey(final String id, final BeatType type, final String stat
                 .setServerPort(lhServerPort)
                 .setServerVersion(lhServerVersion)
                 .setServerId(lhServerId)
+                .setDataplaneId(dataplaneId)
                 .setId(id)
                 .setType(type);
 

canary/src/main/proto/beats.proto

@@ -28,6 +28,7 @@ message BeatKey {
   optional string id = 6;
   repeated Tag tags = 7;
   string server_id = 8;
+  string dataplane_id = 9;
 }
 
 message BeatValue {

canary/src/main/proto/metrics.proto

@@ -13,6 +13,7 @@ message MetricKey {
   string id = 4;
   repeated Tag tags = 5;
   string server_id = 6;
+  string dataplane_id = 7;
 }
 
 message MetricValue {

canary/src/test/java/io/littlehorse/canary/aggregator/internal/MetricStoreExporterTest.java

@@ -79,7 +79,7 @@ public void shouldScrapeSimpleMetric() throws InterruptedException {
 
         assertThat(prometheusRegistry.scrape())
                 .contains(
-                        "my_metric{custom_tag=\"custom_value\",server=\"localhost:2023\",server_id=\"my_server\",server_version=\"test\"} 1.0");
+                        "my_metric{custom_tag=\"custom_value\",dataplane_id=\"my-dataplane\",server=\"localhost:2023\",server_id=\"my_server\",server_version=\"test\"} 1.0");
     }
 
     private static MetricKey createMetricsKey(List<Tag> tags) {
@@ -93,6 +93,7 @@ private static MetricKey createMetricsKey(String host, List<Tag> tags) {
                 .setServerVersion("test")
                 .setId("my_metric")
                 .setServerId("my_server")
+                .setDataplaneId("my-dataplane")
                 .addAllTags(tags)
                 .build();
     }
@@ -125,7 +126,7 @@ void printMetricsWithTwoDifferentServers() throws InterruptedException {
         assertThat(prometheusRegistry.scrape())
                 .isEqualTo(
                         "# HELP my_metric  \n" + "# TYPE my_metric gauge\n"
-                                + "my_metric{custom_tag=\"custom_value\",server=\"localhost2:2023\",server_id=\"my_server\",server_version=\"test\"} 1.0\n"
-                                + "my_metric{custom_tag=\"custom_value\",server=\"localhost:2023\",server_id=\"my_server\",server_version=\"test\"} 1.0\n");
+                                + "my_metric{custom_tag=\"custom_value\",dataplane_id=\"my-dataplane\",server=\"localhost2:2023\",server_id=\"my_server\",server_version=\"test\"} 1.0\n"
+                                + "my_metric{custom_tag=\"custom_value\",dataplane_id=\"my-dataplane\",server=\"localhost:2023\",server_id=\"my_server\",server_version=\"test\"} 1.0\n");
     }
 }

canary/src/test/java/io/littlehorse/canary/aggregator/topology/MetricsTopologyTest.java

@@ -31,6 +31,7 @@ class MetricsTopologyTest {
     public static final String HOST_2 = "localhost2";
     public static final int PORT_2 = 2024;
     public static final String SERVER_ID = "LH";
+    public static final String DATAPLANE_ID = "DP";
 
     private TopologyTestDriver testDriver;
     private TestInputTopic<BeatKey, BeatValue> inputTopic;
@@ -65,7 +66,8 @@ private static MetricKey newMetricKey(String host, int port, String id, String s
                 .setServerHost(host)
                 .setServerPort(port)
                 .setId(id)
-                .setServerId(SERVER_ID);
+                .setServerId(SERVER_ID)
+                .setDataplaneId(DATAPLANE_ID);
 
         if (status != null) {
             builder.addTags(Tag.newBuilder().setKey("status").setValue(status).build());
@@ -110,6 +112,7 @@ private static TestRecord<BeatKey, BeatValue> newBeat(
                 .setServerPort(port)
                 .setType(type)
                 .setServerId(SERVER_ID)
+                .setDataplaneId(DATAPLANE_ID)
                 .setId(id);
         BeatValue.Builder valueBuilder = BeatValue.newBuilder().setTime(Timestamps.now());
 

docs/CANARY_CONFIGURATIONS.md

@@ -16,6 +16,7 @@
       * [`lh.canary.metronome.data.path`](#lhcanarymetronomedatapath)
       * [`lh.canary.metronome.beat.extra.tags.<additional tag>`](#lhcanarymetronomebeatextratagsadditional-tag)
       * [`lh.canary.metronome.server.id`](#lhcanarymetronomeserverid)
+      * [`lh.canary.metronome.server.dataplane.id`](#lhcanarymetronomeserverdataplaneid)
     * [Kafka Configurations](#kafka-configurations)
     * [LH Client Configurations](#lh-client-configurations)
   * [Task Worker](#task-worker)
@@ -176,6 +177,15 @@ For example: `lh.canary.metronome.server.id=lh`.
 - **Default:** null
 - **Importance:** high
 
+#### `lh.canary.metronome.server.dataplane.id`
+
+Add the tag dataplane id the prometheus metrics (**mandatory**).
+For example: `lh.canary.metronome.server.dataplane.id=my-cluster-aws`.
+
+- **Type:** string
+- **Default:** null
+- **Importance:** high
+
 ### Kafka Configurations
 
 LH Canary supports all kafka configurations. Use the prefix `lh.canary.kafka` and append the kafka config.