Statistic on gauge

[chinghongfang]

#623 (comment)

有些數據不適合 "對每一次紀錄都做統計"。這裡的作法是間隔段時間才紀錄數據，可以用來紀錄 "觀測值" 的速率。

比如說，我們一秒鐘會觀測數次 "topic size" ，所以計算速率的方法是 "間隔一秒才紀錄新數據，算差值"。

這裡解釋 RangedDataCalculator 的想法：這個方法是為了 "不對每一次紀錄都做統計"，原因如上，觀測的頻率不固定，如果對每一次的觀測值統計 (例：對每一次觀測統計平均)，那麼觀測頻率較高的時段將會有較高的比重。因此，想讓 Stat 在紀錄前，先檢查是否要將這次的數據納入計算，或者是在某一時間區間的數據中取一個代表值 (例：平均、中位數等，弭平觀測誤差)。

以上是我想到的問題與解法，應該還有我沒有考慮到的，還請不吝賜教。

[chia7712]

以上是我想到的問題與解法，應該還有我沒有考慮到的，還請不吝賜教。

我們可能要先去看現在最可能的幾個用法：

1. Fix lock issue of Receiver  #1010

這隻PR改進我們收集 beans 的方法後，如何幫使用者將HasGague轉換成具時間統計的數字 (例如RateByTime)，或是轉換成既有的介面HasMeter，這樣的好處是使用者 (尤其是 cost function)就不需要自己再去統計或轉換

2. perf 中 consumer rebalance 的統計，我們需要統計每次 rebalance 後 partitions 的變化率，這部分需要 @harryteng9527 給一些建議

因此，麻煩你先就上述兩個例子給予一些想法（或是就這隻PR給一些範例），然後我們再來看看程式碼本身

[harryteng9527]

perf 中 consumer rebalance 的統計，我們需要統計每次 rebalance 後 partitions 的變化率，這部分需要 @harryteng9527 給一些建議

這邊的變化率是要計算哪種？ 例如：

1. 單純 assigned partition 跟 revoked partition 的數量差異
2. sticky partition - 即 revoked partition 跟 assigned partition 的交集
3. re-balance 後沒有 sticky住 的 partitions 數量 - assigned partitions 減去 revoked partitions

如何取得 revoked、assigned partitions

可以利用 ConsumerRebalanceListener 的 callback function 取得

如何取得 consumer instance 被分配的 partition 數量

1. 利用 Listener 的 assigned partitions 來取得
2. Client 端要設定 JMX port ，可透過 MBean client 取得 consumer 的 assigned partitions 數量

[chia7712]

單純 assigned partition 跟 revoked partition 的數量差異
sticky partition - 即 revoked partition 跟 assigned partition 的交集
re-balance 後沒有 sticky住 的 partitions 數量 - assigned partitions 減去 revoked partitions

這幾個都是不錯的例子，不管是哪一個，我的想像是透過這支PR 提供的統計功能，我們可以輕鬆獲得例如從開始到現在的變化率、或是最近一段時間的變化率等等

[chinghongfang]

1. Fix lock issue of Receiver Fix lock issue of Receiver  #1010
   這隻PR改進我們收集 beans 的方法後，如何幫使用者將HasGague轉換成具時間統計的數字 (例如RateByTime)，或是轉換成既有的介面HasMeter，這樣的好處是使用者 (尤其是 cost function)就不需要自己再去統計或轉換

這邊聽起來像是要把統計放在 BeanCollector 內完成，讓使用者呼叫 current() 時，直接取得 HasMeter 之類的。
不過，這樣在 register reciever 前，就要先決定好需要哪些統計值，要交給 1) 使用者指定 、 2) BeanCollector 內計算一組常用的統計值。這將是下一步 (把統計值放在BeanCollector 內完成) 要考慮的事情。

所以現在要考慮的是，"要如何讓 BeanCollector 也方便使用這些統計方法"。不過因為不只有 BeanCollector 要使用，個人比較傾向讓 Sensor 保持簡單 (只有 3 個方法 record, measure, metrics，raw data 的提取由 caller 負責)，也許可以在 Sensor 新增 static method 來幫助把 beanObject 紀錄進多個 Sensor 中，不過可能也要看 之後要怎麼儲存 fetch 到的資料。

1. 單純 assigned partition 跟 revoked partition 的數量差異
2. sticky partition - 即 revoked partition 跟 assigned partition 的交集
3. re-balance 後沒有 sticky住 的 partitions 數量 - assigned partitions 減去 revoked partitions

感謝提供例子，我想這些都是可以新增的功能，不過看起來這些數據的紀錄時間會更分散 (可能有做 rebalance 時才會紀錄) ，所以可能要定義一下該統計值的意義，比如說如果我有 "一段時間" 沒有紀錄(or rebalance) 那回傳的 "數量差異" 該是 1) 上一次的差，還是 2) "0 沒有差異" 。

就這隻PR給一些範例

使用者可以如此使用

var topicSizeSensor = new SensorBuilder<Double>().add("rate", new RateByTime(Duration.ofSeconds(1))).build();
while (true){
  // Fetch raw data
  var beanObjects = receiver.current();

  // Get value from benaObjects
  // Double value = beanObjects.stream().filter( ...

  topicSizeSensor.record(value);
  var statistics = topicSizeSensor.measure("rate");

  /****** Make use of statistics... ******/

}

比如說，每次我需要統計值時，我都會先去抓最新的數據，那麼我就把最新的數據丟進 Sensor 中更新統計值，然後拿取統計結果來使用。以上面的例子來說，這個統計值是以1秒為單位的增減速率。

---

以 2. 舉例，假設已經做好 Difference 這個 Stat

var assignedSensor = new SensorBuilder<Integer>().add("difference", new Difference());
new ConsumerRebalanceListener{
  @Override
  void onPartitionAssigned(Set<TopicPartition> partitions){
    assignedSensor.record(partitions.size());
  }
};

// Put listener to consumer

var statistics = assignedSensor.measure("difference");

/****** Make use of statistics ******/

[chia7712]

這樣在 register reciever 前，就要先決定好需要哪些統計值，要交給 1) 使用者指定 、 2) BeanCollector 內計算一組常用的統計值。這將是下一步 (把統計值放在BeanCollector 內完成) 要考慮的事情。

我建議一開始要“統計”哪些東西不用讓使用者決定，可以先選幾個常用的統計就好，例如單位時間內的變化就好。

使用者可以如此使用

@chinghongfang 這邊給的範例已經可以使用了嗎？還是需要等這隻PR?

[chinghongfang]

這邊給的範例已經可以使用了嗎？還是需要等這隻PR?

要等這隻 PR，RateByTime 是在這隻 PR 題的。

[chia7712]

要等這隻 PR，RateByTime 是在這隻 PR 題的。

@chinghongfang 可否隨後發一隻PR來增加我們觀察 consumer 的指標？ https://github.com/skiptests/astraea/blob/main/app/src/main/java/org/astraea/app/performance/TrackerThread.java#L156

[chia7712]

LGTM!