DataflowMetrics extends MetricResults { private JobMetrics getJobMetrics() throws IOException { if (cachedMetricResults != null) { return cachedMetricResults; } JobMetrics result = dataflowClient.getJobMetrics(dataflowPipelineJob.jobId); if (dataflowPipelineJob.getState().isTerminal()) { cachedMetricResults = result; } return result; } DataflowMetrics(DataflowPipelineJob dataflowPipelineJob, DataflowClient dataflowClient); @Override MetricQueryResults queryMetrics(@Nullable MetricsFilter filter);  }
@Test public void testEmptyMetricUpdates() throws IOException { Job modelJob = new Job(); modelJob.setCurrentState(State.RUNNING.toString()); DataflowPipelineJob job = mock(DataflowPipelineJob.class); DataflowPipelineOptions options = mock(DataflowPipelineOptions.class); when(options.isStreaming()).thenReturn(false); when(job.getDataflowOptions()).thenReturn(options); when(job.getState()).thenReturn(State.RUNNING); job.jobId = JOB_ID; JobMetrics jobMetrics = new JobMetrics(); jobMetrics.setMetrics(null ); DataflowClient dataflowClient = mock(DataflowClient.class); when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics); DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient); MetricQueryResults result = dataflowMetrics.allMetrics(); assertThat(ImmutableList.copyOf(result.getCounters()), is(empty())); assertThat(ImmutableList.copyOf(result.getDistributions()), is(empty())); }
@Test public void testCachingMetricUpdates() throws IOException { Job modelJob = new Job(); modelJob.setCurrentState(State.RUNNING.toString()); DataflowPipelineJob job = mock(DataflowPipelineJob.class); DataflowPipelineOptions options = mock(DataflowPipelineOptions.class); when(options.isStreaming()).thenReturn(false); when(job.getDataflowOptions()).thenReturn(options); when(job.getState()).thenReturn(State.DONE); job.jobId = JOB_ID; JobMetrics jobMetrics = new JobMetrics(); jobMetrics.setMetrics(ImmutableList.of()); DataflowClient dataflowClient = mock(DataflowClient.class); when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics); DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient); verify(dataflowClient, times(0)).getJobMetrics(JOB_ID); dataflowMetrics.allMetrics(); verify(dataflowClient, times(1)).getJobMetrics(JOB_ID); dataflowMetrics.allMetrics(); verify(dataflowClient, times(1)).getJobMetrics(JOB_ID); }
@Test public void testSingleCounterUpdates() throws IOException { JobMetrics jobMetrics = new JobMetrics(); DataflowPipelineJob job = mock(DataflowPipelineJob.class); DataflowPipelineOptions options = mock(DataflowPipelineOptions.class); when(options.isStreaming()).thenReturn(false); when(job.getDataflowOptions()).thenReturn(options); when(job.getState()).thenReturn(State.RUNNING); job.jobId = JOB_ID; AppliedPTransform<?, ?, ?> myStep = mock(AppliedPTransform.class); when(myStep.getFullName()).thenReturn("myStepName"); job.transformStepNames = HashBiMap.create(); job.transformStepNames.put(myStep, "s2"); MetricUpdate update = new MetricUpdate(); long stepValue = 1234L; update.setScalar(new BigDecimal(stepValue)); MetricUpdate mu1 = makeCounterMetricUpdate("counterName", "counterNamespace", "s2", 1234L, false); MetricUpdate mu1Tentative = makeCounterMetricUpdate("counterName", "counterNamespace", "s2", 1233L, true); jobMetrics.setMetrics(ImmutableList.of(mu1, mu1Tentative)); DataflowClient dataflowClient = mock(DataflowClient.class); when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics); DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient); MetricQueryResults result = dataflowMetrics.allMetrics(); assertThat( result.getCounters(), containsInAnyOrder( attemptedMetricsResult("counterNamespace", "counterName", "myStepName", 1234L))); assertThat( result.getCounters(), containsInAnyOrder( committedMetricsResult("counterNamespace", "counterName", "myStepName", 1234L))); }
@Test public void testIgnoreDistributionButGetCounterUpdates() throws IOException { JobMetrics jobMetrics = new JobMetrics(); DataflowClient dataflowClient = mock(DataflowClient.class); when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics); DataflowPipelineJob job = mock(DataflowPipelineJob.class); DataflowPipelineOptions options = mock(DataflowPipelineOptions.class); when(options.isStreaming()).thenReturn(false); when(job.getDataflowOptions()).thenReturn(options); when(job.getState()).thenReturn(State.RUNNING); job.jobId = JOB_ID; AppliedPTransform<?, ?, ?> myStep = mock(AppliedPTransform.class); when(myStep.getFullName()).thenReturn("myStepName"); job.transformStepNames = HashBiMap.create(); job.transformStepNames.put(myStep, "s2"); jobMetrics.setMetrics( ImmutableList.of( makeCounterMetricUpdate("counterName", "counterNamespace", "s2", 1233L, false), makeCounterMetricUpdate("counterName", "counterNamespace", "s2", 1233L, true), makeCounterMetricUpdate("otherCounter[MIN]", "otherNamespace", "s2", 0L, false), makeCounterMetricUpdate("otherCounter[MIN]", "otherNamespace", "s2", 0L, true))); DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient); MetricQueryResults result = dataflowMetrics.allMetrics(); assertThat( result.getCounters(), containsInAnyOrder( attemptedMetricsResult("counterNamespace", "counterName", "myStepName", 1233L))); assertThat( result.getCounters(), containsInAnyOrder( committedMetricsResult("counterNamespace", "counterName", "myStepName", 1233L))); }
@Test public void testDistributionUpdates() throws IOException { JobMetrics jobMetrics = new JobMetrics(); DataflowClient dataflowClient = mock(DataflowClient.class); when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics); DataflowPipelineJob job = mock(DataflowPipelineJob.class); DataflowPipelineOptions options = mock(DataflowPipelineOptions.class); when(options.isStreaming()).thenReturn(false); when(job.getDataflowOptions()).thenReturn(options); when(job.getState()).thenReturn(State.RUNNING); job.jobId = JOB_ID; AppliedPTransform<?, ?, ?> myStep2 = mock(AppliedPTransform.class); when(myStep2.getFullName()).thenReturn("myStepName"); job.transformStepNames = HashBiMap.create(); job.transformStepNames.put(myStep2, "s2"); jobMetrics.setMetrics( ImmutableList.of( makeDistributionMetricUpdate( "distributionName", "distributionNamespace", "s2", 18L, 2L, 2L, 16L, false), makeDistributionMetricUpdate( "distributionName", "distributionNamespace", "s2", 18L, 2L, 2L, 16L, true))); DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient); MetricQueryResults result = dataflowMetrics.allMetrics(); assertThat( result.getDistributions(), contains( attemptedMetricsResult( "distributionNamespace", "distributionName", "myStepName", DistributionResult.create(18, 2, 2, 16)))); assertThat( result.getDistributions(), contains( committedMetricsResult( "distributionNamespace", "distributionName", "myStepName", DistributionResult.create(18, 2, 2, 16)))); }
@Test public void testDistributionUpdatesStreaming() throws IOException { JobMetrics jobMetrics = new JobMetrics(); DataflowClient dataflowClient = mock(DataflowClient.class); when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics); DataflowPipelineJob job = mock(DataflowPipelineJob.class); DataflowPipelineOptions options = mock(DataflowPipelineOptions.class); when(options.isStreaming()).thenReturn(true); when(job.getDataflowOptions()).thenReturn(options); when(job.getState()).thenReturn(State.RUNNING); job.jobId = JOB_ID; AppliedPTransform<?, ?, ?> myStep2 = mock(AppliedPTransform.class); when(myStep2.getFullName()).thenReturn("myStepName"); job.transformStepNames = HashBiMap.create(); job.transformStepNames.put(myStep2, "s2"); jobMetrics.setMetrics( ImmutableList.of( makeDistributionMetricUpdate( "distributionName", "distributionNamespace", "s2", 18L, 2L, 2L, 16L, false), makeDistributionMetricUpdate( "distributionName", "distributionNamespace", "s2", 18L, 2L, 2L, 16L, true))); DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient); MetricQueryResults result = dataflowMetrics.allMetrics(); try { result.getDistributions().iterator().next().getCommitted(); fail("Expected UnsupportedOperationException"); } catch (UnsupportedOperationException expected) { assertThat( expected.getMessage(), containsString( "This runner does not currently support committed" + " metrics results. Please use 'attempted' instead.")); } assertThat( result.getDistributions(), contains( attemptedMetricsResult( "distributionNamespace", "distributionName", "myStepName", DistributionResult.create(18, 2, 2, 16)))); }
@Test public void testMultipleCounterUpdates() throws IOException { JobMetrics jobMetrics = new JobMetrics(); DataflowClient dataflowClient = mock(DataflowClient.class); when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics); DataflowPipelineJob job = mock(DataflowPipelineJob.class); DataflowPipelineOptions options = mock(DataflowPipelineOptions.class); when(options.isStreaming()).thenReturn(false); when(job.getDataflowOptions()).thenReturn(options); when(job.getState()).thenReturn(State.RUNNING); job.jobId = JOB_ID; AppliedPTransform<?, ?, ?> myStep2 = mock(AppliedPTransform.class); when(myStep2.getFullName()).thenReturn("myStepName"); job.transformStepNames = HashBiMap.create(); job.transformStepNames.put(myStep2, "s2"); AppliedPTransform<?, ?, ?> myStep3 = mock(AppliedPTransform.class); when(myStep3.getFullName()).thenReturn("myStepName3"); job.transformStepNames.put(myStep3, "s3"); AppliedPTransform<?, ?, ?> myStep4 = mock(AppliedPTransform.class); when(myStep4.getFullName()).thenReturn("myStepName4"); job.transformStepNames.put(myStep4, "s4"); jobMetrics.setMetrics( ImmutableList.of( makeCounterMetricUpdate("counterName", "counterNamespace", "s2", 1233L, false), makeCounterMetricUpdate("counterName", "counterNamespace", "s2", 1234L, true), makeCounterMetricUpdate("otherCounter", "otherNamespace", "s3", 12L, false), makeCounterMetricUpdate("otherCounter", "otherNamespace", "s3", 12L, true), makeCounterMetricUpdate("counterName", "otherNamespace", "s4", 1200L, false), makeCounterMetricUpdate("counterName", "otherNamespace", "s4", 1233L, true), makeCounterMetricUpdate("lostName", "otherNamespace", "s5", 1200L, false), makeCounterMetricUpdate("lostName", "otherNamespace", "s5", 1200L, true))); DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient); MetricQueryResults result = dataflowMetrics.allMetrics(); assertThat( result.getCounters(), containsInAnyOrder( attemptedMetricsResult("counterNamespace", "counterName", "myStepName", 1233L), attemptedMetricsResult("otherNamespace", "otherCounter", "myStepName3", 12L), attemptedMetricsResult("otherNamespace", "counterName", "myStepName4", 1200L))); assertThat( result.getCounters(), containsInAnyOrder( committedMetricsResult("counterNamespace", "counterName", "myStepName", 1233L), committedMetricsResult("otherNamespace", "otherCounter", "myStepName3", 12L), committedMetricsResult("otherNamespace", "counterName", "myStepName4", 1200L))); }
@Test public void testMultipleCounterUpdatesStreaming() throws IOException { JobMetrics jobMetrics = new JobMetrics(); DataflowClient dataflowClient = mock(DataflowClient.class); when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics); DataflowPipelineJob job = mock(DataflowPipelineJob.class); DataflowPipelineOptions options = mock(DataflowPipelineOptions.class); when(options.isStreaming()).thenReturn(true); when(job.getDataflowOptions()).thenReturn(options); when(job.getState()).thenReturn(State.RUNNING); job.jobId = JOB_ID; AppliedPTransform<?, ?, ?> myStep2 = mock(AppliedPTransform.class); when(myStep2.getFullName()).thenReturn("myStepName"); job.transformStepNames = HashBiMap.create(); job.transformStepNames.put(myStep2, "s2"); AppliedPTransform<?, ?, ?> myStep3 = mock(AppliedPTransform.class); when(myStep3.getFullName()).thenReturn("myStepName3"); job.transformStepNames.put(myStep3, "s3"); AppliedPTransform<?, ?, ?> myStep4 = mock(AppliedPTransform.class); when(myStep4.getFullName()).thenReturn("myStepName4"); job.transformStepNames.put(myStep4, "s4"); jobMetrics.setMetrics( ImmutableList.of( makeCounterMetricUpdate("counterName", "counterNamespace", "s2", 1233L, false), makeCounterMetricUpdate("counterName", "counterNamespace", "s2", 1234L, true), makeCounterMetricUpdate("otherCounter", "otherNamespace", "s3", 12L, false), makeCounterMetricUpdate("otherCounter", "otherNamespace", "s3", 12L, true), makeCounterMetricUpdate("counterName", "otherNamespace", "s4", 1200L, false), makeCounterMetricUpdate("counterName", "otherNamespace", "s4", 1233L, true))); DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient); MetricQueryResults result = dataflowMetrics.allMetrics(); try { result.getCounters().iterator().next().getCommitted(); fail("Expected UnsupportedOperationException"); } catch (UnsupportedOperationException expected) { assertThat( expected.getMessage(), containsString( "This runner does not currently support committed" + " metrics results. Please use 'attempted' instead.")); } assertThat( result.getCounters(), containsInAnyOrder( attemptedMetricsResult("counterNamespace", "counterName", "myStepName", 1233L), attemptedMetricsResult("otherNamespace", "otherCounter", "myStepName3", 12L), attemptedMetricsResult("otherNamespace", "counterName", "myStepName4", 1200L))); }