Make retention.ms be long; Attach broker id into metrics datapoints

Fixes: - Convert topicSize and retentionMs from interger to long. So the large topic or long rention topic wont cause sensor parsing error. - Attach broker id into metrics datapoints to help the engineer find out which brokers are in bad state.
pinterest · Oct 8, 2024 · 3a37e31 · 3a37e31
2 parents 4f11d45 + a622de5
commit 3a37e31
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 14 deletions.
diff --git a/...c/main/java/com/pinterest/orion/core/automation/operator/kafka/BrokerHealingOperator.java b/...c/main/java/com/pinterest/orion/core/automation/operator/kafka/BrokerHealingOperator.java
@@ -159,12 +159,15 @@ public void operate(KafkaCluster cluster) throws Exception {
           "Orion agents on " + cluster.getClusterId() + " are unhealthy, no URPs on the cluster: " + alertableUnhealthyAgentBrokersWithoutURPs,
           "orion"
       ));
-      OrionServer.metricsCounterInc(
-              "broker.agent.unhealthy",
-              new HashMap<String, String>() {{
-                put("clusterId", cluster.getClusterId());
-              }}
-      );
+      for (String brokerId : alertableUnhealthyAgentBrokersWithoutURPs) {
+        OrionServer.metricsCounterInc(
+                "broker.agent.unhealthy",
+                new HashMap<String, String>() {{
+                  put("clusterId", cluster.getClusterId());
+                  put("brokerId", brokerId);
+                }}
+        );
+      }
     }
 
     // alert on brokers where broker service is unhealthy but there are no URPs if they show up for 3 consecutive times
@@ -186,12 +189,15 @@ public void operate(KafkaCluster cluster) throws Exception {
           "Kafka service on " + cluster.getClusterId() + " are unhealthy, no URPs on the cluster: " + alertableUnhealthyBrokersWithoutURPs,
           "orion"
       ));
-      OrionServer.metricsCounterInc(
-              "broker.service.unhealthy",
-              new HashMap<String, String>() {{
-                put("clusterId", cluster.getClusterId());
-              }}
-      );
+      for (String brokerId : alertableUnhealthyBrokersWithoutURPs) {
+        OrionServer.metricsCounterInc(
+                "broker.service.unhealthy",
+                new HashMap<String, String>() {{
+                  put("clusterId", cluster.getClusterId());
+                  put("brokerId", brokerId);
+                }}
+        );
+      }
     }
 
     setMessage("offline brokers: " + unhealthyKafkaBrokers + "\nunhealthy agent orion nodes: " + unhealthyAgentNodes +

diff --git a/...rver/src/main/java/com/pinterest/orion/core/automation/sensor/kafka/KafkaTopicSensor.java b/...rver/src/main/java/com/pinterest/orion/core/automation/sensor/kafka/KafkaTopicSensor.java
@@ -156,7 +156,7 @@ public void populateTopicMetrics(KafkaCluster cluster, Map<String, KafkaTopicDes
         put("clusterId", cluster.getClusterId());
       }};
       // Topic size
-      int topicSize = (int) getTopicSizeByteFromTopicDescription(topicDescription);
+      long topicSize = (long) getTopicSizeByteFromTopicDescription(topicDescription);
       OrionServer.metricsHistogram(
               "kafka.topic.size.bytes",
               topicSize,
@@ -170,7 +170,7 @@ public void populateTopicMetrics(KafkaCluster cluster, Map<String, KafkaTopicDes
               metricsTags
       );
       // Retention
-      int retentionMs= Integer.parseInt(
+      long retentionMs= Long.parseLong(
               topicDescription.getTopicConfigs().getOrDefault("retention.ms", "0.0"));
       OrionServer.metricsHistogram(
               "kafka.topic.retention.ms",