Display Broker and Brokerset Information in Orion UI

This update enables Orion to store and display metrics for brokers and brokersets within the Orion interface.  
  
## Details  
  
### Broker Class  
  
- Introduces two new maps:  
  - **`rawBrokerStatus`**: Stores raw data from the metrics API call as a map of `String -> Double`. Example: `cpu_usage -> 0.22`.  
  - **`brokerStatus`**: Contains the keys and values that are displayed in the Orion UI. Both values are strings. Example: `"CPU Usage"` and `"22%"`.  
  
### Brokerset Class  
  
- Also includes two new maps:  
  - **`rawBrokersetStatus`**: Stores raw data after aggregating values across all brokers as a map of `String -> Double`. Example: `max_cpu_usage -> 0.22`.  
  - **`brokersetStatus`**: Contains the keys and values to be shown in the Orion UI. Both values are strings. Example: `"Max CPU Usage Across All Brokers"` and `"22%"`.  
  
### BrokerMetricsSensor Class  
  
- This class will be overridden by a sensor from the internal package.  
- The sensor is intended to periodically call metrics APIs, such as the CloudWatch API, and update the `rawBrokerStatus`.  
  
### JavaScript Changes  
  
- Multiple JavaScript updates were made to display the information.  
- The Brokersets page in the Service section will show:  
  - Maximum CPU usage over 7 days.  
  - Maximum disk usage over 7 days for all brokers.  
  
## Testing  
  
- The changes have been successfully tested within Pinterest.

Author: yisheng-zhou

orion-commons/src/main/java/com/pinterest/orion/common/NodeInfo.java

@@ -36,7 +36,26 @@ public class NodeInfo implements Serializable {
   private Map<String, String> agentSettings;
   private Map<String, String> environment;
   private Set<String> brokersets = new HashSet<>();
+  /**
+   * Save the broker status in printable format (string)
+   */
   private Map<String, String> brokerStatus;
+  /**
+   * Save the broker status in raw format (number)
+   */
+  private Map<String, Double> rawBrokerStatus;
+  /**
+   * @param rawBrokerStatus
+   */
+  public void setRawBrokerStatus(Map<String, Double> rawBrokerStatus) {
+    this.rawBrokerStatus = rawBrokerStatus;
+  }
+  /**
+   * @return the rawBrokerStatus
+   */
+  public Map<String, Double> getRawBrokerStatus() {
+      return rawBrokerStatus;
+  }
   /**
    * @param brokerStatus
    */
@@ -202,15 +221,26 @@ public void setNodeType(String nodeType) {
     this.nodeType = nodeType;
   }
 
-  /* (non-Javadoc)
-   * @see java.lang.Object#toString()
-   */
   @Override
   public String toString() {
-    return "NodeInfo [nodeId=" + nodeId + ", hostname=" + hostname + ", ip=" + ip + ", clusterId="
-        + clusterId + ", servicePort=" + servicePort + ", localtime="
-        + localtime + ", rack=" + rack + ", serviceInfo=" + serviceInfo + ", agentSettings="
-        + agentSettings + ", environment=" + environment + "]";
+    return String.format(
+        "NodeInfo [timestamp=%d, nodeId=%s, hostname=%s, ip=%s, clusterId=%s, servicePort=%d, " +
+            "localtime=%d, rack=%s, nodeType=%s, serviceInfo=%s, agentSettings=%s, environment=%s, brokersets=%s, " +
+            "brokerStatus=%s, rawBrokerStatus=%s]",
+        timestamp,
+        nodeId,
+        hostname,
+        ip,
+        clusterId,
+        servicePort,
+        localtime,
+        rack,
+        nodeType,
+        serviceInfo,
+        agentSettings,
+        environment,
+        brokersets,
+        brokerStatus,
+        rawBrokerStatus);
   }
-
 }

orion-server/src/main/java/com/pinterest/orion/core/automation/sensor/kafka/BrokerMetricsSensor.java

@@ -0,0 +1,15 @@
+package com.pinterest.orion.core.automation.sensor.kafka;
+
+import com.pinterest.orion.core.kafka.KafkaCluster;
+
+public class BrokerMetricsSensor extends KafkaSensor {
+
+    @Override
+    public String getName() {
+        return "BrokerMetricsSensor";
+    }
+
+    @Override
+    public void sense(KafkaCluster cluster) throws Exception {
+    }
+}

orion-server/src/main/java/com/pinterest/orion/core/automation/sensor/kafka/BrokersetStateSensor.java

@@ -79,6 +79,16 @@ public void sense(KafkaCluster cluster) throws Exception {
                 brokersetState.addBrokerRange(Arrays.asList(start, end));
             }
             brokersetState.setBrokerIds(new ArrayList<>(brokerIds));
+            try {
+                updateBrokersetStateWithMetrics(cluster, brokersetState, brokerIds);
+            } catch (Exception e) {
+                logger.warning(
+                    String.format(
+                        "Failed to update brokerset state with metrics for brokerset %s in cluster %s. Error: %s",
+                        brokersetAlias,
+                        cluster.getName(),
+                        e.getMessage()));
+            }
             brokersetStateMap.put(brokersetAlias, brokersetState);
             if (invalidBrokerset) {
                 handleInvalidBrokerset(brokersetAlias, cluster.getName());
@@ -95,6 +105,19 @@ public void sense(KafkaCluster cluster) throws Exception {
         }
     }
 
+    /**
+     * Update brokerset state with metrics.
+     * This method should be overridden by subclasses to update brokerset state with metrics.
+     * @param cluster Kafka cluster.
+     * @param brokersetState Brokerset state to update. It has state fields and raw metrics fields to update.
+     * @param brokerIds Broker ids in the brokerset.
+     */
+    protected void updateBrokersetStateWithMetrics(
+        KafkaCluster cluster,
+        BrokersetState brokersetState,
+        Set<String> brokerIds) {
+    }
+
     @Override
     public String getName() {
         return "BrokersetStateSensor";

orion-server/src/main/java/com/pinterest/orion/core/kafka/BrokersetState.java

@@ -3,12 +3,17 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 public class BrokersetState {
     /**
      * The brokersetAlias is the alias of the brokerset.
      */
     private String brokersetAlias;
+    /**
+     * The instanceType is the type of the brokerset.
+     */
+    private String instanceType;
     /**
      * The brokersetRanges are the ranges of brokerset that are in the brokerset.
      * The brokersetRanges are obtained from the brokerset configuration file.
@@ -19,6 +24,14 @@ public class BrokersetState {
      * The brokerIds are obtained from the cluster state.
      */
     private List<String> brokerIds = new ArrayList<>();
+    /**
+     * Save the broker status in printable format (string).
+     */
+    private Map<String, Double> rawBrokersetStatus;
+    /**
+     * Save the broker status in raw format (number).
+     */
+    private Map<String, String> brokersetStatus;
     /**
      * The constructor of BrokersetState.
      * @param brokersetAlias
@@ -108,4 +121,22 @@ public List<String> getBrokerIds() {
     public void setBrokerIds(List<String> brokerIds) {
         this.brokerIds = brokerIds;
     }
+    public void setRawBrokersetStatus(Map<String, Double> rawBrokersetStatus) {
+        this.rawBrokersetStatus = rawBrokersetStatus;
+    }
+    public Map<String, Double> getRawBrokersetStatus() {
+        return rawBrokersetStatus;
+    }
+    public void setBrokersetStatus(Map<String, String> brokersetStatus) {
+        this.brokersetStatus = brokersetStatus;
+    }
+    public Map<String, String> getBrokersetStatus() {
+        return brokersetStatus;
+    }
+    public void setInstanceType(String instanceType) {
+        this.instanceType = instanceType;
+    }
+    public String getInstanceType() {
+        return instanceType;
+    }
 }

orion-server/src/main/resources/webapp/src/basic-components/Kafka/BrokersetEntry.js

@@ -30,7 +30,13 @@ const routes = [
 function getStatsData(clusterId, rawData) {
     let brokersetStats = [];
     let brokersetData = rawData.brokersetData;
-    brokersetStats.push({ key: "Broker Count", value: brokersetData.size});
+    brokersetStats.push({ key: "Broker_Count", value: brokersetData.size});
+    let brokersetStatus = brokersetData.brokersetStatus;
+    if (brokersetStatus !== undefined && brokersetStatus !== null) {
+        for (let key of Object.keys(brokersetStatus)) {
+            brokersetStats.push({ key: key, value: brokersetStatus[key] });
+        }
+    }
     return brokersetStats;
 }
 
@@ -111,6 +117,10 @@ function getBrokersetInfoHeader(rawData, clusterId) {
     let brokersetData = rawData.brokersetData;
     let brokersetAlias = brokersetData.brokersetAlias;
     let brokerCount = brokersetData.size;
+    let instanceType = "Unknown";
+    if (brokersetData.instanceType !== undefined && brokersetData.instanceType !== null) {
+        instanceType = brokersetData.instanceType;
+    }
     return (
         <Box my={2}>
             <Grid container display="flex" alignItems="center" spacing={2}>
@@ -130,6 +140,14 @@ function getBrokersetInfoHeader(rawData, clusterId) {
                         label={brokerCount + " brokers"}
                     />
                 </Grid>
+                <Grid item>
+                    <Chip
+                        variant="outlined"
+                        color="primary"
+                        size="small"
+                        label={instanceType}
+                    />
+                </Grid>
             </Grid>
         </Box>
     );

orion-server/src/main/resources/webapp/src/basic-components/Kafka/Brokersets.js

@@ -42,17 +42,37 @@ export default function Brokersets(props) {
     }
     let columns = [
         { title: "Name", field: "brokersetAlias" },
-        { title: "Broker Count", field: "brokerCount" }
+        { title: "Broker Count", field: "brokerCount" },
+        { title: "Max CPU Usage 7D", field: "maxCpuUsage7Day" },
+        { title: "Max Disk Usage 7D", field: "maxDiskUsage7Day" },
+        { title: "Updated Time", field: "timestamp" }
     ]
     let clusterId = props.cluster.clusterId;
     let brokersetToRowValuesMap = {};
     for (let brokerset of brokersets) {
         let brokersetAlias = brokerset.brokersetAlias;
+        let maxCpuUsage7Day = "N/A";
+        let maxDiskUsage7Day = "N/A";
+        let timestamp = "N/A"
+        if (brokerset.brokersetStatus) {
+            if (brokerset.brokersetStatus["CPU_Usage_Max_All_Brokers_7Days"] !== undefined) {
+                maxCpuUsage7Day = brokerset.brokersetStatus["CPU_Usage_Max_All_Brokers_7Days"];
+            }
+            if (brokerset.brokersetStatus["Disk_Usage_Max_All_Brokers_7Days"] !== undefined) {
+                maxDiskUsage7Day = brokerset.brokersetStatus["Disk_Usage_Max_All_Brokers_7Days"];
+            }
+            if (brokerset.brokersetStatus["Short_Timestamp"] !== undefined) {
+                timestamp = brokerset.brokersetStatus["Short_Timestamp"];
+            }
+        }
         brokersetToRowValuesMap[brokersetAlias] = {
             "brokersetAlias": brokersetAlias,
             "clusterId": clusterId,
             "brokerCount": brokerset.size,
-            "brokersetData": brokerset
+            "brokersetData": brokerset,
+            "maxCpuUsage7Day": maxCpuUsage7Day,
+            "maxDiskUsage7Day": maxDiskUsage7Day,
+            "timestamp": timestamp
         }
     }
 

orion-server/src/main/resources/webapp/src/basic-components/Kafka/KafkaNode.js

@@ -268,6 +268,13 @@ function getBrokersetColumns() {
 
 function getBrokerStatsData(cluster, node) {
   let brokerStatsData = [];
+  let brokerStats = node.currentNodeInfo.brokerStatus;
+  if (brokerStats === undefined || brokerStats === null) {
+    return brokerStatsData;
+  }
+  for (let [key, value] of Object.entries(brokerStats)) {
+    brokerStatsData.push({key: key, value: JSON.stringify(value)});
+  }
   return brokerStatsData;
 }