Add replicas/leaders metrics to API response, display node replicas

api/src/main/java/com/github/eyefloaters/console/api/service/KafkaClusterService.java

@@ -174,13 +174,14 @@ CompletionStage<KafkaCluster> addMetrics(KafkaCluster cluster, List<String> fiel
 
         String namespace = cluster.getNamespace();
         String name = cluster.getName();
+        int controllerId = cluster.getController().id();
 
         try (var rangesStream = getClass().getResourceAsStream("/metrics/queries/kafkaCluster_ranges.promql");
              var valuesStream = getClass().getResourceAsStream("/metrics/queries/kafkaCluster_values.promql")) {
             String rangeQuery = new String(rangesStream.readAllBytes(), StandardCharsets.UTF_8)
                     .formatted(namespace, name);
             String valueQuery = new String(valuesStream.readAllBytes(), StandardCharsets.UTF_8)
-                    .formatted(namespace, name);
+                    .formatted(namespace, name, controllerId);
 
             var rangeResults = metricsService.queryRanges(rangeQuery).toCompletableFuture();
             var valueResults = metricsService.queryValues(valueQuery).toCompletableFuture();

api/src/main/resources/metrics/queries/kafkaCluster_ranges.promql

@@ -1,11 +1,11 @@
   sum by (nodeId, __console_metric_name__) (
     label_replace(
       label_replace(
-        rate(container_cpu_usage_seconds_total{namespace="%1$s",pod=~"%2$s-kafka-\\d+"}[1m]),
+        rate(container_cpu_usage_seconds_total{namespace="%1$s",pod=~"%2$s-.+-\\d+"}[1m]),
         "nodeId",
         "$1",
         "pod",
-        ".+-kafka-(\\d+)"
+        ".+-(\\d+)"
       ),
       "__console_metric_name__",
       "cpu_usage_seconds",
@@ -17,11 +17,11 @@ or
   sum by (nodeId, __console_metric_name__) (
     label_replace(
       label_replace(
-        container_memory_usage_bytes{namespace="%1$s",pod=~"%2$s-kafka-\\d+"},
+        container_memory_usage_bytes{namespace="%1$s",pod=~"%2$s-.+-\\d+"},
         "nodeId",
         "$1",
         "pod",
-        ".+-kafka-(\\d+)"
+        ".+-(\\d+)"
       ),
       "__console_metric_name__",
       "memory_usage_bytes",
@@ -32,7 +32,7 @@ or
 or
   sum by (__console_metric_name__) (
     label_replace(
-      irate(kafka_server_brokertopicmetrics_bytesin_total{topic!="",namespace="%1$s",pod=~"%2$s-kafka-\\d+"}[5m]),
+      irate(kafka_server_brokertopicmetrics_bytesin_total{topic!="",namespace="%1$s",pod=~"%2$s-.+-\\d+",strimzi_io_kind="Kafka"}[5m]),
       "__console_metric_name__",
       "incoming_byte_rate",
       "",
@@ -42,7 +42,7 @@ or
 or
   sum by (__console_metric_name__) (
     label_replace(
-      irate(kafka_server_brokertopicmetrics_bytesout_total{topic!="",namespace="%1$s",pod=~"%2$s-kafka-\\d+"}[5m]),
+      irate(kafka_server_brokertopicmetrics_bytesout_total{topic!="",namespace="%1$s",pod=~"%2$s-.+-\\d+",strimzi_io_kind="Kafka"}[5m]),
       "__console_metric_name__",
       "outgoing_byte_rate",
       "",
@@ -53,11 +53,11 @@ or
   sum by (nodeId, __console_metric_name__) (
     label_replace(
       label_replace(
-        kubelet_volume_stats_capacity_bytes{namespace="%1$s",persistentvolumeclaim=~"data(?:-\\d+)?-%2$s-kafka-\\d+"},
+        kubelet_volume_stats_capacity_bytes{namespace="%1$s",persistentvolumeclaim=~"data(?:-\\d+)?-%2$s-.+-\\d+"},
         "nodeId",
         "$1",
         "persistentvolumeclaim",
-        ".+-kafka-(\\d+)"
+        ".+-(\\d+)"
       ),
       "__console_metric_name__",
       "volume_stats_capacity_bytes",
@@ -69,11 +69,11 @@ or
   sum by (nodeId, __console_metric_name__) (
     label_replace(
       label_replace(
-        kubelet_volume_stats_used_bytes{namespace="%1$s",persistentvolumeclaim=~"data(?:-\\d+)?-%2$s-kafka-\\d+"},
+        kubelet_volume_stats_used_bytes{namespace="%1$s",persistentvolumeclaim=~"data(?:-\\d+)?-%2$s-.+-\\d+"},
         "nodeId",
         "$1",
         "persistentvolumeclaim",
-        ".+-kafka-(\\d+)"
+        ".+-(\\d+)"
       ),
       "__console_metric_name__",
       "volume_stats_used_bytes",

api/src/main/resources/metrics/queries/kafkaCluster_values.promql

@@ -1,11 +1,11 @@
 sum by (__console_metric_name__, nodeId) (
   label_replace(
     label_replace(
-      kafka_server_kafkaserver_brokerstate{namespace="%1$s",pod=~"%2$s-kafka-\\d+"} > 0,
+      kafka_server_kafkaserver_brokerstate{namespace="%1$s",pod=~"%2$s-.+-\\d+",strimzi_io_kind="Kafka"} > 0,
       "nodeId",
       "$1",
       "pod",
-      ".+-kafka-(\\d+)"
+      ".+-(\\d+)"
     ),
     "__console_metric_name__",
     "broker_state",
@@ -18,7 +18,7 @@ or
 
 sum by (__console_metric_name__) (
   label_replace(
-    kafka_controller_kafkacontroller_globaltopiccount{namespace="%1$s",pod=~"%2$s-kafka-\\d+"} > 0,
+    kafka_controller_kafkacontroller_globaltopiccount{namespace="%1$s",pod=~"%2$s-.+-%3$d",strimzi_io_kind="Kafka"} > 0,
     "__console_metric_name__",
     "total_topics",
     "",
@@ -30,7 +30,7 @@ or
 
 sum by (__console_metric_name__) (
   label_replace(
-    kafka_controller_kafkacontroller_globalpartitioncount{namespace="%1$s",pod=~"%2$s-kafka-\\d+"} > 0,
+    kafka_controller_kafkacontroller_globalpartitioncount{namespace="%1$s",pod=~"%2$s-.+-%3$d",strimzi_io_kind="Kafka"} > 0,
     "__console_metric_name__",
     "total_partitions",
     "",
@@ -41,9 +41,52 @@ sum by (__console_metric_name__) (
 or
 
 label_replace(
-  count(sum by (topic) (kafka_cluster_partition_underreplicated{namespace="%1$s",pod=~"%2$s-kafka-\\d+"})),
+  (
+    count(
+      sum by (topic) (
+        kafka_cluster_partition_underreplicated{namespace="%1$s",pod=~"%2$s-.+-\\d+",strimzi_io_kind="Kafka"} > 0
+      )
+    )
+    OR on() vector(0)
+  ),
   "__console_metric_name__",
   "underreplicated_topics",
   "",
   ""
-)
+)
+
+or
+
+sum by (__console_metric_name__, nodeId) (
+  label_replace(
+    label_replace(
+      kafka_cluster_partition_replicascount{namespace="%1$s",pod=~"%2$s-.+-\\d+",strimzi_io_kind="Kafka"} > 0,
+      "nodeId",
+      "$1",
+      "pod",
+      ".+-(\\d+)"
+    ),
+    "__console_metric_name__",
+    "replica_count",
+    "",
+    ""
+  )
+)
+
+or
+
+sum by (__console_metric_name__, nodeId) (
+  label_replace(
+    label_replace(
+      kafka_server_replicamanager_leadercount{namespace="%1$s",pod=~"%2$s-.+-\\d+",strimzi_io_kind="Kafka"} > 0,
+      "nodeId",
+      "$1",
+      "pod",
+      ".+-(\\d+)"
+    ),
+    "__console_metric_name__",
+    "leader_count",
+    "",
+    ""
+  )
+)

ui/api/kafka/actions.ts

@@ -27,7 +27,7 @@ export async function getKafkaClusters(): Promise<ClusterList[]> {
 export async function getKafkaCluster(
   clusterId: string,
 ): Promise<ClusterDetail | null> {
-  const url = `${process.env.BACKEND_URL}/api/kafkas/${clusterId}/?fields%5Bkafkas%5D=name,namespace,creationTimestamp,nodes,controller,authorizedOperations,bootstrapServers,authType`;
+  const url = `${process.env.BACKEND_URL}/api/kafkas/${clusterId}/?fields%5Bkafkas%5D=name,namespace,creationTimestamp,nodes,controller,authorizedOperations,bootstrapServers,authType,metrics`;
   try {
     const res = await fetch(url, {
       headers: await getHeaders(),

ui/api/kafka/schema.ts

@@ -39,6 +39,24 @@ const ClusterDetailSchema = z.object({
         authType: z.string().nullable(),
       })
     ).optional() /* remove .optional() when `listeners` is added to the fetched fields */,
+    metrics: z.object({
+      values: z.record(
+        z.array(
+          z.object({
+            value: z.string(),
+            nodeId: z.string().optional(),
+          })
+        )
+      ),
+      ranges: z.record(
+        z.array(
+          z.object({
+            range: z.array(z.array(z.string())),
+            nodeId: z.string().optional(),
+          })
+        )
+      ),
+    }).optional(),
   }),
 });
 export const ClusterResponse = z.object({

ui/app/[locale]/kafka/[kafkaId]/nodes/NodesTable.tsx

@@ -6,14 +6,16 @@ import { ClipboardCopy, Label } from "@patternfly/react-core";
 import { ServerIcon } from "@patternfly/react-icons";
 import Link from "next/link";
 
-const columns = ["id", "host", "rack"] as const;
+const columns = ["id", "replicas", "rack"] as const;
 
 export function NodesTable({
   nodes,
   controller,
+  metrics,
 }: {
   nodes: KafkaNode[];
   controller: KafkaNode;
+  metrics: Record<string, any>;
 }) {
   return (
     <ResponsiveTable
@@ -29,8 +31,8 @@ export function NodesTable({
                 &nbsp; Node
               </Th>
             );
-          case "host":
-            return <Th>Host</Th>;
+          case "replicas":
+            return <Th>Total Replicas</Th>;
           case "rack":
             return <Th>Rack</Th>;
         }
@@ -50,17 +52,11 @@ export function NodesTable({
                 )}
               </Td>
             );
-          case "host":
+          case "replicas":
             return (
               <Td key={key} dataLabel={"Host"}>
-                <ClipboardCopy
-                  hoverTip="Copy"
-                  clickTip="Copied"
-                  variant="inline-compact"
-                  isBlock={true}
-                >
-                  {row.host}:{row.port}
-                </ClipboardCopy>
+                { metrics.values.replica_count
+                    .find((e: any) => parseInt(e.nodeId) == row.id)?.value ?? "-" }
               </Td>
             );
           case "rack":

ui/app/[locale]/kafka/[kafkaId]/nodes/page.tsx

@@ -15,6 +15,7 @@ export default async function NodesPage({ params }: { params: KafkaParams }) {
       <NodesTable
         nodes={cluster.attributes.nodes}
         controller={cluster.attributes.controller}
+        metrics={cluster.attributes.metrics ?? {}}
       />
     </PageSection>
   );