Orion Broker Set UI Implementation

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

@@ -16,7 +16,9 @@
 package com.pinterest.orion.common;
 
 import java.io.Serializable;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 
 public class NodeInfo implements Serializable {
 
@@ -33,6 +35,32 @@ public class NodeInfo implements Serializable {
   private Map<String, String> serviceInfo;
   private Map<String, String> agentSettings;
   private Map<String, String> environment;
+  private Set<String> brokersets = new HashSet<>();
+  private Map<String, String> brokerStatus;
+  /**
+   * @param brokerStatus
+   */
+  public void setBrokerStatus(Map<String, String> brokerStatus) {
+    this.brokerStatus = brokerStatus;
+  }
+  /**
+   * @return the brokerStatus
+   */
+  public Map<String, String> getBrokerStatus() {
+    return brokerStatus;
+  }
+  /**
+   * @param brokersets
+   */
+  public void setBrokersets(Set<String> brokersets) {
+    this.brokersets = brokersets;
+  }
+  /**
+   * @return the brokersets
+   */
+  public Set<String> getBrokersets() {
+      return brokersets;
+  }
   /**
    * @return the timestamp
    */

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

@@ -0,0 +1,113 @@
+package com.pinterest.orion.core.automation.sensor.kafka;
+
+import com.pinterest.orion.core.Attribute;
+import com.pinterest.orion.core.Node;
+import com.pinterest.orion.core.kafka.Brokerset;
+import com.pinterest.orion.core.kafka.BrokersetState;
+import com.pinterest.orion.core.kafka.KafkaCluster;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.logging.Logger;
+
+public class BrokersetStateSensor extends KafkaSensor {
+    private static final Logger logger = Logger.getLogger(BrokersetStateSensor.class.getName());
+    public static final String ATTR_BROKERSET_STATE_KEY = "brokersetState";
+
+    /**
+     * KafkaClusterInfoSensor loads the brokerset information from the config files.
+     * This sensor generates brokerset state information based on the brokerset information.
+     * @param cluster Kafka cluster to sense.
+     * @throws Exception If there is an error sensing the cluster.
+     */
+    @Override
+    public void sense(KafkaCluster cluster) throws Exception {
+        // Brokerset information and broker information are loaded from other sensors.
+        // If they are not ready, skip this sensor.
+        if (!cluster.containsAttribute(KafkaClusterInfoSensor.ATTR_BROKERSET_KEY)) {
+            // No brokerset information available. Skip.
+            return;
+        }
+        Attribute brokersetMapAttr = cluster.getAttribute(KafkaClusterInfoSensor.ATTR_BROKERSET_KEY);
+        if (brokersetMapAttr == null) {
+            // No brokerset information available. Skip.
+            return;
+        }
+        if (cluster.getNodeMap() == null || cluster.getNodeMap().isEmpty()) {
+            // No node information available. Skip.
+            return;
+        }
+        // Brokerset info from config files
+        Map<String, Brokerset> brokersetMap = brokersetMapAttr.getValue();
+        // Brokerset state map.
+        Map<String, BrokersetState> brokersetStateMap = new HashMap<>();
+        // BrokerId to brokerset alias set map
+        Map<String, Set<String>> brokerToBrokersetsMap = new HashMap<>();
+        for (String nodeId : cluster.getNodeMap().keySet()) {
+            brokerToBrokersetsMap.put(nodeId, new HashSet<>());
+        }
+        // Get broker ids for each brokerset.
+        for (Brokerset brokerset : brokersetMap.values()) {
+            String brokersetAlias = brokerset.getBrokersetAlias();
+            BrokersetState brokersetState = new BrokersetState(brokersetAlias);
+            Set<String> brokerIds = new HashSet<>();
+            List<Brokerset.BrokersetRange> brokersetRanges = brokerset.getEntries();
+            if (brokersetRanges == null || brokersetRanges.isEmpty()) {
+                logger.warning(String.format("Brokerset %s in cluster %s has no brokerset range.",
+                    brokersetAlias, cluster.getName()));
+                continue;
+            }
+            boolean invalidBrokerset = false;
+            for (Brokerset.BrokersetRange brokersetRange : brokersetRanges) {
+                int start = brokersetRange.getStartBrokerIdx();
+                int end = brokersetRange.getEndBrokerIdx();
+                for (int i = start; i <= end; i++) {
+                    String nodeId = Integer.toString(i);
+                    if (cluster.getNodeMap().containsKey(nodeId)) {
+                        Node node = cluster.getNodeMap().get(nodeId);
+                        brokerIds.add(node.getCurrentNodeInfo().getNodeId());
+                        brokerToBrokersetsMap.get(nodeId).add(brokersetAlias);
+                    } else {
+                        invalidBrokerset = true;
+                    }
+                }
+                brokersetState.addBrokerRange(Arrays.asList(start, end));
+            }
+            brokersetState.setBrokerIds(new ArrayList<>(brokerIds));
+            brokersetStateMap.put(brokersetAlias, brokersetState);
+            if (invalidBrokerset) {
+                handleInvalidBrokerset(brokersetAlias, cluster.getName());
+            }
+        }
+        // TODO: Add other brokerset status information (ex. usage data).
+        cluster.setAttribute(ATTR_BROKERSET_STATE_KEY, brokersetStateMap);
+        // Update node info with brokerset information.
+        for (Node node : cluster.getNodeMap().values()) {
+            if (node != null && node.getCurrentNodeInfo() != null) {
+                String nodeId = node.getCurrentNodeInfo().getNodeId();
+                node.getCurrentNodeInfo().setBrokersets(brokerToBrokersetsMap.get(nodeId));
+            }
+        }
+    }
+
+    @Override
+    public String getName() {
+        return "BrokersetStateSensor";
+    }
+
+    /**
+     * Handle invalid brokerset.
+     * Invalid brokerset means that some brokers recorded in brokerset config files are not found in the cluster.
+     * @param brokersetAlias Brokerset alias.
+     * @param clusterId Cluster id.
+     */
+    protected void handleInvalidBrokerset(String brokersetAlias, String clusterId) {
+        logger.warning(String.format("Brokerset %s in cluster %s has invalid brokers.",
+            brokersetAlias, clusterId));
+    }
+}

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

@@ -0,0 +1,111 @@
+package com.pinterest.orion.core.kafka;
+
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class BrokersetState {
+    /**
+     * The brokersetAlias is the alias of the brokerset.
+     */
+    private String brokersetAlias;
+    /**
+     * The brokersetRanges are the ranges of brokerset that are in the brokerset.
+     * The brokersetRanges are obtained from the brokerset configuration file.
+     */
+    private List<List<Integer>> brokersetRanges = new ArrayList<>();
+    /**
+     * The brokerIds are the ids of running broker that are in the brokerset.
+     * The brokerIds are obtained from the cluster state.
+     */
+    private List<String> brokerIds = new ArrayList<>();
+    /**
+     * The constructor of BrokersetState.
+     * @param brokersetAlias
+     */
+    public BrokersetState(String brokersetAlias) {
+        this.brokersetAlias = brokersetAlias;
+    }
+    /**
+     * The constructor of BrokersetState.
+     * @param brokersetAlias
+     * @param brokersetRanges
+     * @param brokerIds
+     */
+    public BrokersetState(String brokersetAlias, List<List<Integer>> brokersetRanges, List<String> brokerIds) {
+        this.brokersetAlias = brokersetAlias;
+        this.brokersetRanges = brokersetRanges;
+        this.brokerIds = brokerIds;
+    }
+    /**
+     * The method is used to get the size of brokerset.
+     * The size of brokerset is the number of brokers in the brokerset.
+     * It may not be equal to the sum of brokersetRanges if some brokers are not running.
+     * @return the size of brokerset.
+     */
+    public int getSize() {
+        return brokerIds.size();
+    }
+    /**
+     * The method is used to get the alias of brokerset.
+     * @return the alias of brokerset.
+     */
+    public String getBrokersetAlias() {
+        return brokersetAlias;
+    }
+    /**
+     * The method is used to set the alias of brokerset.
+     * @param brokersetAlias
+     */
+    public void setBrokersetAlias(String brokersetAlias) {
+        this.brokersetAlias = brokersetAlias;
+    }
+    /**
+     * The method is used to get the ranges of brokerset.
+     * @return the ranges of brokerset.
+     */
+    public List<List<Integer>> getBrokersetRanges() {
+        return brokersetRanges;
+    }
+    /**
+     * The method is used to add the range of brokerset.
+     * Each range is a list of two integers.
+     * @param brokerRange
+     * @throws IllegalArgumentException
+     */
+    public void addBrokerRange(List<Integer> brokerRange) throws IllegalArgumentException {
+        if (brokerRange.size() != 2) {
+            throw new IllegalArgumentException(
+                "The size of brokerRange should be 2 for brokerset " + brokersetAlias + ".");
+        }
+        brokersetRanges.add(brokerRange);
+    }
+    /**
+     * The method is used to set the ranges of brokerset.
+     * @param brokersetRanges
+     * @throws IllegalArgumentException
+     */
+    public void setBrokersetRanges(List<List<Integer>> brokersetRanges) throws IllegalArgumentException {
+        for (List<Integer> brokerRange : brokersetRanges) {
+            if (brokerRange.size() != 2) {
+                throw new IllegalArgumentException(
+                    "The size of brokerRange should be 2 for brokerset " + brokersetAlias + ".");
+            }
+        }
+        this.brokersetRanges = brokersetRanges;
+    }
+    /**
+     * The method is used to get the ids of brokers in the brokerset.
+     * @return brokerIds
+     */
+    public List<String> getBrokerIds() {
+        return brokerIds;
+    }
+    /**
+     * The method is used to set the ids of brokers in the brokerset.
+     * @param brokerIds
+     */
+    public void setBrokerIds(List<String> brokerIds) {
+        this.brokerIds = brokerIds;
+    }
+}