FlinkMetricContainer { public void updateMetrics(String stepName, List<MonitoringInfo> monitoringInfos) { getMetricsContainer(stepName).update(monitoringInfos); updateMetrics(stepName); } FlinkMetricContainer(RuntimeContext runtimeContext); MetricsContainerImpl getMetricsContainer(String stepName); void registerMetricsForPipelineResult(); void updateMetrics(String stepName, List<MonitoringInfo> monitoringInfos); static final String ACCUMULATOR_NAME; }
@Test public void testMonitoringInfoUpdate() { SimpleCounter userCounter = new SimpleCounter(); when(metricGroup.counter("ns1.metric1")).thenReturn(userCounter); SimpleCounter pCollectionCounter = new SimpleCounter(); when(metricGroup.counter("pcoll.metric:element_count:v1")).thenReturn(pCollectionCounter); SimpleCounter pTransformCounter = new SimpleCounter(); when(metricGroup.counter("anyPTransform.myMetric")).thenReturn(pTransformCounter); MonitoringInfo userCountMonitoringInfo = new SimpleMonitoringInfoBuilder() .setUrn(MonitoringInfoConstants.Urns.USER_SUM_INT64) .setLabel(MonitoringInfoConstants.Labels.NAMESPACE, "ns1") .setLabel(MonitoringInfoConstants.Labels.NAME, "metric1") .setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, "anyPTransform") .setInt64SumValue(111) .build(); assertNotNull(userCountMonitoringInfo); MonitoringInfo pCollectionScoped = new SimpleMonitoringInfoBuilder() .setUrn(MonitoringInfoConstants.Urns.ELEMENT_COUNT) .setInt64SumValue(222) .setLabel(MonitoringInfoConstants.Labels.PCOLLECTION, "pcoll") .setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, "anyPTransform") .build(); assertNotNull(pCollectionScoped); MonitoringInfo transformScoped = new SimpleMonitoringInfoBuilder() .setUrn(MonitoringInfoConstants.Urns.START_BUNDLE_MSECS) .setInt64SumValue(333) .setLabel(MonitoringInfoConstants.Labels.NAME, "myMetric") .setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, "anyPTransform") .build(); assertNotNull(transformScoped); assertThat(userCounter.getCount(), is(0L)); assertThat(pCollectionCounter.getCount(), is(0L)); assertThat(pTransformCounter.getCount(), is(0L)); container.updateMetrics( "step", ImmutableList.of(userCountMonitoringInfo, pCollectionScoped, transformScoped)); assertThat(userCounter.getCount(), is(111L)); assertThat(pCollectionCounter.getCount(), is(222L)); assertThat(pTransformCounter.getCount(), is(333L)); }
@Test public void testMonitoringInfoUpdate() { SimpleCounter userCounter = new SimpleCounter(); when(metricGroup.counter("ns1.metric1")).thenReturn(userCounter); SimpleCounter pCollectionCounter = new SimpleCounter(); when(metricGroup.counter("pcoll.metric:element_count:v1")).thenReturn(pCollectionCounter); SimpleCounter pTransformCounter = new SimpleCounter(); when(metricGroup.counter("anyPTransform.myMetric")).thenReturn(pTransformCounter); MonitoringInfo userCountMonitoringInfo = new SimpleMonitoringInfoBuilder() .setUrn(MonitoringInfoConstants.Urns.USER_COUNTER) .setLabel(MonitoringInfoConstants.Labels.NAMESPACE, "ns1") .setLabel(MonitoringInfoConstants.Labels.NAME, "metric1") .setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, "anyPTransform") .setInt64Value(111) .build(); assertNotNull(userCountMonitoringInfo); MonitoringInfo pCollectionScoped = new SimpleMonitoringInfoBuilder() .setUrn(MonitoringInfoConstants.Urns.ELEMENT_COUNT) .setInt64Value(222) .setLabel(MonitoringInfoConstants.Labels.PCOLLECTION, "pcoll") .setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, "anyPTransform") .build(); assertNotNull(pCollectionScoped); MonitoringInfo transformScoped = new SimpleMonitoringInfoBuilder() .setUrn(MonitoringInfoConstants.Urns.START_BUNDLE_MSECS) .setInt64Value(333) .setLabel(MonitoringInfoConstants.Labels.NAME, "myMetric") .setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, "anyPTransform") .build(); assertNotNull(transformScoped); assertThat(userCounter.getCount(), is(0L)); assertThat(pCollectionCounter.getCount(), is(0L)); assertThat(pTransformCounter.getCount(), is(0L)); container.updateMetrics( "step", ImmutableList.of(userCountMonitoringInfo, pCollectionScoped, transformScoped)); assertThat(userCounter.getCount(), is(111L)); assertThat(pCollectionCounter.getCount(), is(222L)); assertThat(pTransformCounter.getCount(), is(333L)); }