private void updateClusterMetrics ( String topologyId , TopologyMetric tpMetric ) { if ( tpMetric . get_topologyMetric ( ) . get_metrics_size ( ) == 0 ) { return ; } MetricInfo topologyMetrics = tpMetric . get_topologyMetric ( ) ; // make a new MetricInfo to save the topologyId's metric MetricInfo clusterMetrics = MetricUtils . mkMetricInfo ( ) ; Set < String > metricNames = new HashSet <> ( ) ; for ( Map . Entry < String , Map < Integer , MetricSnapshot > > entry : topologyMetrics . get_metrics ( ) . entrySet ( ) ) { String metricName = MetricUtils . topo2clusterName ( entry . getKey ( ) ) ; MetricType metricType = MetricUtils . metricType ( metricName ) ; Map < Integer , MetricSnapshot > winData = new HashMap <> ( ) ; for ( Map . Entry < Integer , MetricSnapshot > entryData : entry . getValue ( ) . entrySet ( ) ) { MetricSnapshot snapshot = entryData . getValue ( ) . deepCopy ( ) ; winData . put ( entryData . getKey ( ) , snapshot ) ; if ( metricType == MetricType . HISTOGRAM ) { // reset topology metric points entryData . getValue ( ) . set_points ( new byte [ 0 ] ) ; entryData . getValue ( ) . set_pointSize ( 0 ) ; } } clusterMetrics . put_to_metrics ( metricName , winData ) ; metricNames . add ( metricName ) ; } // save to local cache, waiting for merging TopologyMetricContext clusterTpMetricContext = context . getClusterTopologyMetricContext ( ) ; clusterTpMetricContext . addToMemCache ( topologyId , clusterMetrics ) ; context . registerMetrics ( JStormMetrics . CLUSTER_METRIC_KEY , metricNames ) ; }
update cluster metrics local cache