add Sensor to MetricCollector

[qoo332001]

此PR結合MetricCollector,ClusterBean和Sensor的功能，讓CostFunction在計算分數時可以透過ClusterBean拿到Sensor統計一段時間內的值

• 新增一個介面MetricSensors，內有一個方法用來處理一種Metrics要如何做統計，以及回傳自行產生的HasBeanObject
• 在MetricCollector中新增 addMetricSensors()，用來增加現有的MetricSensors
• 在MetricCollector更新metrics到beans的同時會呼叫各個MetricSensors中的record方法，並取回由該方法產生的HasBeanObject儲存到beans
• ClusterBean中新增兩個方法分別可以用來取得TopicPartitionReplica level 與 broker level透過Sensor計算出來的統計資料
  • statisticsByReplica
  • statisticsByNode

目前如何統計使用Sensor統計這些HasBeanObject會完全交給CostFunction來決定，因此如果需要使用自行統計後的值會需要先實做一個方法CostFunction#sensors()，這個方法會需要回傳Collection，每個MetricsSensors會紀錄"一種Metrics"(ex. BytesInPerSec, Size)，每一種Metrics可能會有多個Sensors(看metrics種類，有可能是一個replica對應一個Sensor或是一個Broker對應一個Sensor)，MetricsSensors中有一個方法record需要實做，每當MetricsCollector那邊收集到新的metrics時，會將identitiy與其收集到的一包metrics送進MetricsSensors#record進行自訂的統計，統計後會回傳一個CostFUnction中自行產生的，統計後的HasBeanObject，最後存到MetricCollector中，當CostFunction要取用時，可以透過ClusterBean來獲取

一個實做MetricsSensors#record的範例：

    @Override
    public Collection<MetricSensors> sensors() {
        return List.of(
                new MetricSensors() {
                    final Map<TopicPartitionReplica, Sensor<Double>> sensors = new HashMap<>();

                    @Override
                    public Map<Integer, Collection<? extends HasBeanObject>> record(
                            int identity, Collection<? extends HasBeanObject> beans) {
                        var statisticalBeans = new HashMap<TopicPartitionReplica, HasBeanObject>();
                        beans.forEach(
                                bean -> {
                                    if (bean != null) {
                                        if (bean.beanObject().domainName().equals(LogMetrics.DOMAIN_NAME)
                                                && bean.beanObject().properties().get("type").equals(LogMetrics.LOG_TYPE)) {
                                            var tpr =
                                                    TopicPartitionReplica.of(
                                                            bean.beanObject().properties().get("topic"),
                                                            Integer.parseInt(bean.beanObject().properties().get("partition")),
                                                            identity);
                                            sensors
                                                    .computeIfAbsent(
                                                            tpr,
                                                            ignore ->
                                                                    new SensorBuilder<Double>()
                                                                            .addStat(Avg.AVG_KEY, Avg.of())
                                                                            .build())
                                                    .record(
                                                            Double.valueOf(
                                                                    bean.beanObject().attributes().get("Value").toString()));
                                            statisticalBeans.put(tpr, bean);
                                        }
                                    }
                                });
                        return Map.of(
                                identity,
                                sensors.entrySet().stream()
                                        .map(
                                                sensor -> {
                                                    var bean = statisticalBeans.get(sensor.getKey()).beanObject();
                                                    return new SizeStatisticalBean(
                                                            new BeanObject(
                                                                    bean.domainName(),
                                                                    bean.properties(),
                                                                    Map.of("Value", sensor.getValue().measure(Avg.AVG_KEY)),
                                                                    bean.createdTimestamp()));
                                                })
                                        .collect(Collectors.toList()));
                    }
                });
    }

CostFunction會回傳的自訂HasBeanObject，實做如下:

  public class SizeStatisticalBean implements HasBeanObject{
      BeanObject beanObject;
      SizeStatisticalBean(BeanObject beanObject){
          this.beanObject = beanObject;
      }
      @Override
      public BeanObject beanObject() {
          return beanObject;
      }

      public String topic() {
          return beanObject().properties().get("topic");
      }

      public int partition() {
          return Integer.parseInt(beanObject().properties().get("partition"));
      }

      public double value(){
          return Double.parseDouble(beanObject().attributes().get("Value").toString());
      }
  }

[chia7712]

@qoo332001 終於看到這個功能，很棒。整個專案目前最重要的項目就是如何提升metrics的品質，如同11/25 很多人在詢問我們是透過哪些線索來“處理負載平衡”

[chia7712]

LGTM

麻煩接著試試看用你新增的方法來套用到 cost function，自我檢驗一下

[qoo332001]

麻煩接著試試看用你新增的方法來套用到 cost function，自我檢驗一下

另外開一PR做對嗎

[chia7712]

另外開一PR做對嗎

yep