# Storm Metrics
Storm 開放了一個 metrics 接口,用來匯報整個 topology 的匯總信息 。 Storm 內部通過該接口可以跟蹤各類統計數據:executor 和 acker 的數量;每個 bolt 的平均處理時延、worker 節點的堆棧使用情況,這些信息都可以在 Nimbus 的 UI 界面中看到。
### Metric Types
Metrics 必須實現 [`IMetric`](http://github.com/apache/storm/blob/master%0A/storm-core/src/jvm/org/apache/storm/metric/api/IMetric.java)接口,IMetric接口只包含一個方法 `getValueAndReset` -- 得到匯總值,并且重置為初始狀態。例如,在 MeanReducer 中實現 running total/running count 的均值,然后兩個值都重新設置為0.
Storm 提供了以下幾種 metric 類型:
* [AssignableMetric](http://github.com/apache/storm/blob/master%0A/storm-core/src/jvm/org/apache/storm/metric/api/AssignableMetric.java) -- 將 metric 設置為指定值。此類型在兩種情況下有用:1\. metric 本身為外部設置的值;2\. 你已經另外計算出了匯總的統計值。
* [CombinedMetric](http://github.com/apache/storm/blob/master%0A/storm-core/src/jvm/org/apache/storm/metric/api/CombinedMetric.java) -- 可以對 metric 進行關聯更新的通用接口。
* [CountMetric](http://github.com/apache/storm/blob/master%0A/storm-core/src/jvm/org/apache/storm/metric/api/CountMetric.java) -- 返回 metric 的匯總結果。可以調用 `incr()` 方法來將加過自增;調用 `incrBy(n)` 方法來將結果加上或者減去給定值。
* [MultiCountMetric](http://github.com/apache/storm/blob/master%0A/storm-core/src/jvm/org/apache/storm/metric/api/MultiCountMetric.java) -- 返回包含一組 CountMetric 的 HashMap
* [ReducedMetric](http://github.com/apache/storm/blob/master%0A/storm-core/src/jvm/org/apache/storm/metric/api/ReducedMetric.java)
* [MeanReducer](http://github.com/apache/storm/blob/master%0A/storm-core/src/jvm/org/apache/storm/metric/api/MeanReducer.java) -- 跟蹤 `reduce()` 方法提供的運行狀態均值結果(可以接受 `Double`、`Integer`、`Long` 等類型,內置的均值結果是 `Double` 類型)。MeanReducer 確實是一個相當棒的家伙。
* [MultiReducedMetric](http://github.com/apache/storm/blob/master%0A/storm-core/src/jvm/org/apache/storm/metric/api/MultiReducedMetric.java) -- 返回包含一組 ReducedMetric 的 HashMap
### Metrics Consumer
你可以監聽和處理 topology(拓撲)的metrics ,通過 注冊 Metrics Consumer到你的topology.
注冊 metrics consumer 到你的 topology,添加到你的 topology(拓撲)配置,像下面這樣:
```
conf.registerMetricsConsumer(org.apache.storm.metric.LoggingMetricsConsumer.class, 1);
```
你可以參考 [Config#registerMetricsConsumer](javadocs/org/apache/storm/Config.html#registerMetricsConsumer-java.lang.Class-) ,然后根據javadoc 覆蓋 java 方法.
否則編輯 storm.yaml 配置文件:
```
topology.metrics.consumer.register:
- class: "org.apache.storm.metric.LoggingMetricsConsumer"
parallelism.hint: 1
- class: "org.apache.storm.metric.HttpForwardingMetricsConsumer"
parallelism.hint: 1
argument: "http://example.com:8080/metrics/my-topology/"
```
注冊 metrics consumer后,Storm 會添加 MetricsConsumerBolt 到你的 topology,每個 MetricsConsumerBolt 會從所有的 tasks 接受 metrics。 相應的 MetricsConsumerBolt 的 parallelism 設置為 `parallelism.hint` ,`component id` 設置為 `__metrics_<metrics consumer class name>`. 如果你注冊多次相同的類, component id 后會添加 `#<sequence number>`。
Storm 提供了一些內置的 metrics consumers,我們來看一下 topology(拓撲)中提供了哪些 metrics.
* [`LoggingMetricsConsumer`](http://github.com/apache/storm/blob/master%0A/storm-core/src/jvm/org/apache/storm/metric/LoggingMetricsConsumer.java) -- 監聽所有的 metrics ,然后將數據dump 到日志文件(Tab Separated Values).
* [`HttpForwardingMetricsConsumer`](http://github.com/apache/storm/blob/master%0A/storm-core/src/jvm/org/apache/storm/metric/HttpForwardingMetricsConsumer.java) -- 監聽所有的 metrics,并且將數據序列化,然后通過http post 到配置的url。Storm 提供 [`HttpForwardingMetricsServer`](http://github.com/apache/storm/blob/master%0A/storm-core/src/jvm/org/apache/storm/metric/HttpForwardingMetricsServer.java) 抽象類,你可以繼承這個類,并且啟動一個 HTTP sever,通過 HttpForwardingMetricsConsumer 處理發送的 metrics.
當然,Storm 開放了實現 Metrics Consumer 的 [`IMetricsConsumer`](http://github.com/apache/storm/blob/master%0A/storm-core/src/jvm/org/apache/storm/metric/api/IMetricsConsumer.java) 接口,你可以創建自定義 metrics consumers ,綁定到相應的 topology(拓撲)。或者使用 Storm 社區內其他比較好的 Metrics Consumers 實現. 你可以參考 [versign/storm-graphite](https://github.com/verisign/storm-graphite) 和 [storm-metrics-statsd](https://github.com/endgameinc/storm-metrics-statsd).
當你實現自己的 metrics consumers,調用 [IMetricsConsumer#prepare](javadocs/org/apache/storm/metric/api/IMetricsConsumer.html#prepare-java.util.Map-java.lang.Object-org.apache.storm.task.TopologyContext-org.apache.storm.task.IErrorReporter-)的時候,`argument` 需要傳給 consumer 對象。所以你要參考 yaml 文件中配置的Java 類型,做好類型的分配。
請記住 MetricsConsumerBolt 只是 Bolt 類型的一種,所以 如果 metrics consumers 如果不能持續處理 metrics,topology 的吞吐量將會下降,所以你要和其他 Bolt 一樣關注好 MetricsConsumerBolt。一個比較好的方式就是將 Metrics Consumers 設計為 `non-blocking`(非阻塞)的。
### 構建你自己的 metric(task level)
你可以通過注冊 `IMetric` 到 Metric Registry(登記處),然后度量你的 metric. 假定你想要度量 Bolt#execute 的執行次數。我們一起來定義這個 metric 實例。CountMetric 符合我們的應用場景。
```
private transient CountMetric countMetric;
```
你會發現,我們將CountMetric 定義為 transient。因為IMetric 并不是 Serializable 的,所以定義為 transient 可以避免很多問題。
下一步,我們初始化和注冊 metric 實例。
```
@Override
public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
// other intialization here.
countMetric = new CountMetric();
context.registerMetric("execute_count", countMetric, 60);
}
```
第一個和第二個參數很簡單,metric 名稱和 IMetric 實例。第三個參數[TopologyContext#registerMetric](javadocs/org/apache/storm/task/TopologyContext.html#registerMetric-java.lang.String-T-int-) 是發布和重置 metric 的間隔時間。
最后,當 Bolt.execute 執行的時候,自增countMetric的值。
```
public void execute(Tuple input) {
countMetric.incr();
// handle tuple here.
}
```
請注意, topology 的sample rate 不適用于自定義 metrics,所以我們自己調用 incr() 方法。
`countMetric.getValueAndReset()` 每隔60秒都會被調用, ("execute_count", value)也會被發送到 MetricsConsumer。
### Build your own metrics (worker level)
你可以注冊 worker 級別的 metrics,將他們添加到集群所有的workers 的 `Config.WORKER_METRICS` 配置,或者所有workers上的指定 topology,通過 `Config.TOPOLOGY_WORKER_METRICS` 配置。
例如,我們可以添加 `worker.metrics` 配置到集群的 storm.yaml。
```
worker.metrics:
metricA: "aaa.bbb.ccc.ddd.MetricA"
metricB: "aaa.bbb.ccc.ddd.MetricB"
...
```
或者按照 `Map<String,String>`(metric name, metric class name)格式,key是`Config.TOPOLOGY_WORKER_METRICS`,來設置 config map。
worker metrics 實例有下面一些限制:
A) worker 級別的 metrics 應該是一種 gauge 類,因為它是從 SystemBolt 初始化和注冊的,不會暴露給 user tasks。
B) Metrics 通過默認的構造器初始化,并且不會對執行配置注入或者對象注入。
C) metrics Bucket size(secounds) 已經修正為 `Config.TOPOLOGY_BUILTIN_METRICS_BUCKET_SIZE_SECS`.
### Builtin Metrics
Storm 的 [builtin metrics](http://github.com/apache/storm/blob/master%0A/storm-core/src/clj/org/apache/storm/daemon/builtin_metrics.clj) 工具。
[builtin_metrics.clj](http://github.com/apache/storm/blob/master%0A/storm-core/src/clj/org/apache/storm/daemon/builtin_metrics.clj) 為內置的metrics 設置了數據結構,其他框架組件可以使用 facade 方法來更新數據。metrics在被調用的時候走計算邏輯-- 可以看例子 [`ack-spout-msg`](http://github.com/apache/storm/blob/master%0A/storm-core/src/clj/org/apache/storm/daemon/executor.clj#358) 的 `clj/b/s/daemon/daemon/executor.clj` 部分。
- Storm 基礎
- 概念
- Scheduler(調度器)
- Configuration
- Guaranteeing Message Processing
- 守護進程容錯
- 命令行客戶端
- Storm UI REST API
- 理解 Storm Topology 的 Parallelism(并行度)
- FAQ
- Layers on Top of Storm
- Storm Trident
- Trident 教程
- Trident API 綜述
- Trident State
- Trident Spouts
- Trident RAS API
- Storm SQL
- Storm SQL 集成
- Storm SQL 示例
- Storm SQL 語言參考
- Storm SQL 內部實現
- Flux
- Storm 安裝和部署
- 設置Storm集群
- 本地模式
- 疑難解答
- 在生產集群上運行 Topology
- Maven
- 安全地運行 Apache Storm
- CGroup Enforcement
- Pacemaker
- 資源感知調度器 (Resource Aware Scheduler)
- 用于分析 Storm 的各種內部行為的 Metrics
- Windows 用戶指南
- Storm 中級
- 序列化
- 常見 Topology 模式
- Clojure DSL
- 使用沒有jvm的語言編輯storm
- Distributed RPC
- Transactional Topologies
- Hooks
- Storm Metrics
- Storm 狀態管理
- Windowing Support in Core Storm
- Joining Streams in Storm Core
- Storm Distributed Cache API
- Storm 調試
- 動態日志級別設置
- Storm Logs
- 動態員工分析
- 拓撲事件檢查器
- Storm 與外部系統, 以及其它庫的集成
- Storm Kafka Integration
- Storm Kafka 集成(0.10.x+)
- Storm HBase Integration
- Storm HDFS Integration
- Storm Hive 集成
- Storm Solr 集成
- Storm Cassandra 集成
- Storm JDBC 集成
- Storm JMS 集成
- Storm Redis 集成
- Azue Event Hubs 集成
- Storm Elasticsearch 集成
- Storm MQTT(Message Queuing Telemetry Transport, 消息隊列遙測傳輸) 集成
- Storm MongoDB 集成
- Storm OpenTSDB 集成
- Storm Kinesis 集成
- Storm Druid 集成
- Storm and Kestrel
- Container, Resource Management System Integration
- Storm 高級
- 針對 Storm 定義一個不是 JVM 的 DSL
- 多語言協議
- Storm 內部實現
- 翻譯進度