Metrics { public static Metric metric(String name) { return MetricsImpl.metric(name, Unit.COUNT); } private  Metrics(); static Metric metric(String name); static Metric metric(String name, Unit unit); static Metric threadSafeMetric(String name); static Metric threadSafeMetric(String name, Unit unit);  }
@Test public void unusedMetrics() { pipeline.readFrom(TestSources.items(0L, 1L, 2L, 3L, 4L)) .filter(l -> { boolean pass = l % 2 == 0; if (!pass) { Metrics.metric("dropped"); } return pass; }) .writeTo(Sinks.noop()); Job job = runPipeline(pipeline.toDag()); JobMetricsChecker checker = new JobMetricsChecker(job); checker.assertSummedMetricValue("dropped", 0); checker.assertNoMetricValues("total"); }
@Test public void typicalUsage() { pipeline.readFrom(TestSources.items(5L, 4L, 3L, 2L, 1L, 0L)) .filter(l -> { boolean pass = l % 2 == 0; if (pass) { Metrics.metric("single-flip-flop").decrement(); Metrics.metric("multi-flip-flop").decrement(10); } else { Metrics.metric("dropped").increment(); Metrics.metric("single-flip-flop").increment(); Metrics.metric("multi-flip-flop").increment(10); } Metrics.metric("total").increment(); Metrics.metric("last").set(l); return pass; }) .writeTo(Sinks.noop()); Job job = runPipeline(pipeline.toDag()); JobMetricsChecker checker = new JobMetricsChecker(job); checker.assertSummedMetricValue("dropped", 3); checker.assertSummedMetricValue("total", 6); checker.assertSummedMetricValue("single-flip-flop", 0); checker.assertSummedMetricValue("multi-flip-flop", 0); }
@Test public void customUnit() { pipeline.readFrom(TestSources.items(0L, 1L, 2L, 3L, 4L)) .mapStateful(LongAccumulator::new, (acc, i) -> { acc.add(i); Metric metric = Metrics.metric("sum", Unit.COUNT); metric.set(acc.get()); return acc.get(); }) .writeTo(Sinks.noop()); Job job = runPipeline(pipeline.toDag()); JobMetricsChecker checker = new JobMetricsChecker(job); checker.assertSummedMetricValue("sum", 10); }
@Test public void customUnit_notUsed() { pipeline.readFrom(TestSources.items(0L, 1L, 2L, 3L, 4L)) .mapStateful(LongAccumulator::new, (acc, i) -> { acc.add(i); Metrics.metric("sum", Unit.COUNT); return acc.get(); }) .writeTo(Sinks.noop()); Job job = runPipeline(pipeline.toDag()); JobMetricsChecker checker = new JobMetricsChecker(job); checker.assertSummedMetricValue("sum", 0L); }
@Test public void nonCooperativeProcessor() { DAG dag = new DAG(); Vertex source = dag.newVertex("source", TestProcessors.ListSource.supplier(asList(1L, 2L, 3L))); Vertex map = dag.newVertex("map", new NonCoopTransformPSupplier((FunctionEx<Long, Long>) l -> { Metrics.metric("mapped").increment(); return l * 10L; })); Vertex sink = dag.newVertex("sink", writeListP("results")); dag.edge(between(source, map)).edge(between(map, sink)); Job job = runPipeline(dag); JobMetricsChecker checker = new JobMetricsChecker(job); checker.assertSummedMetricValue("mapped", 3L); assertEquals( new HashSet<>(Arrays.asList(10L, 20L, 30L)), new HashSet<>(instance().getList("results")) ); }
@Test public void metricsDisabled() { Long[] input = {0L, 1L, 2L, 3L, 4L}; pipeline.readFrom(TestSources.items(input)) .map(l -> { Metrics.metric("mapped").increment(); Metrics.metric("total", Unit.COUNT).set(input.length); return l; }) .writeTo(Sinks.noop()); Job job = instance().newJob(pipeline, new JobConfig().setMetricsEnabled(false)); job.join(); JobMetrics metrics = job.getMetrics(); assertTrue(metrics.get("mapped").isEmpty()); assertTrue(metrics.get("total").isEmpty()); }
@Test public void availableDuringJobExecution() { int generatedItems = 1000; pipeline.readFrom(TestSources.itemStream(1_000)) .withIngestionTimestamps() .filter(l -> l.sequence() < generatedItems) .map(t -> { Metrics.metric("total").increment(); return t; }) .writeTo(Sinks.noop()); Job job = instance().newJob(pipeline, JOB_CONFIG_WITH_METRICS); JobMetricsChecker checker = new JobMetricsChecker(job); assertTrueEventually(() -> checker.assertSummedMetricValue("total", generatedItems)); }
@Test public void when_metricsDisabled_then_unavailableDuringJobExecution() { int generatedItems = 1000; pipeline.readFrom(TestSources.itemStream(1_000)) .withIngestionTimestamps() .filter(l -> l.sequence() < generatedItems) .map(t -> { Metrics.metric("total").increment(); return t; }) .writeTo(Sinks.list("sink")); Job job = instance().newJob(pipeline, new JobConfig().setMetricsEnabled(false)); List<Object> list = instance().getList("sink"); assertTrueEventually(() -> assertFalse(list.isEmpty())); assertTrue(job.getMetrics().get("total").isEmpty()); }
@Test public void when_jetMetricNameIsUsed_then_itIsNotOverwritten() { Long[] items = {0L, 1L, 2L, 3L, 4L}; pipeline.readFrom(TestSources.items(items)) .filter(l -> { Metrics.metric("emittedCount").increment(1000); return true; }) .writeTo(Sinks.noop()); Job job = runPipeline(pipeline.toDag()); new JobMetricsChecker(job, MeasurementPredicates.tagValueEquals("user", "true").negate()) .assertSummedMetricValue("emittedCount", 10); new JobMetricsChecker(job) .assertSummedMetricValue("emittedCount", 10 + items.length * 1000); }
@Test public void metricInFusedStages() { int inputSize = 100_000; Integer[] inputs = new Integer[inputSize]; Arrays.setAll(inputs, i -> i); pipeline.readFrom(TestSources.items(inputs)) .filter(l -> { Metrics.metric("onlyInFilter").increment(); Metrics.metric("inBoth").increment(); return true; }) .map(t -> { Metrics.metric("onlyInMap").increment(); Metrics.metric("inBoth").increment(); return t; }) .writeTo(Sinks.noop()); Job job = runPipeline(pipeline.toDag()); JobMetricsChecker checker = new JobMetricsChecker(job); checker.assertSummedMetricValue("onlyInFilter", inputSize); checker.assertSummedMetricValue("onlyInMap", inputSize); checker.assertSummedMetricValue("inBoth", 2 * inputSize); }
@Test public void metricInNotFusedStages() { int inputSize = 100_000; Integer[] inputs = new Integer[inputSize]; Arrays.setAll(inputs, i -> i); pipeline.readFrom(TestSources.items(inputs)) .filter(l -> { Metrics.metric("onlyInFilter").increment(); Metrics.metric("inBoth").increment(); return true; }) .groupingKey(t -> t) .aggregate(counting()) .map(t -> { Metrics.metric("onlyInMap").increment(); Metrics.metric("inBoth").increment(); return t; }) .writeTo(Sinks.noop()); Job job = runPipeline(pipeline.toDag()); JobMetricsChecker checker = new JobMetricsChecker(job); checker.assertSummedMetricValue("onlyInFilter", inputSize); checker.assertSummedMetricValue("onlyInMap", inputSize); checker.assertSummedMetricValue("inBoth", 2 * inputSize); }
@Test public void metricsAreRestartedIfPipelineIsRunTwice() { pipeline.readFrom(TestSources.items(0L, 1L, 2L, 3L, 4L)) .filter(l -> { Metrics.metric("total").increment(); return true; }) .writeTo(Sinks.noop()); Job job = runPipeline(pipeline.toDag()); new JobMetricsChecker(job).assertSummedMetricValue("total", 5); job = runPipeline(pipeline.toDag()); new JobMetricsChecker(job).assertSummedMetricValue("total", 5); }
@Test public void metricsAreJobIndependent() { pipeline.readFrom(TestSources.itemStream(1_000)) .withIngestionTimestamps() .filter(l -> l.sequence() < 4000) .map(t -> { if (t.sequence() % 4 == 0) { Metrics.metric("total").increment(); } return t; }) .writeTo(Sinks.noop()); Pipeline pipeline2 = Pipeline.create(); pipeline2.readFrom(TestSources.itemStream(1_000)) .withIngestionTimestamps() .filter(l -> l.sequence() < 4000) .map(t -> { if (t.sequence() % 4 != 0) { Metrics.metric("total").increment(); } return t; }) .writeTo(Sinks.noop()); Job job = instance().newJob(pipeline, JOB_CONFIG_WITH_METRICS); Job job2 = instance().newJob(pipeline2, JOB_CONFIG_WITH_METRICS); JobMetricsChecker checker1 = new JobMetricsChecker(job); assertTrueEventually(() -> checker1.assertSummedMetricValue("total", 1000)); JobMetricsChecker checker2 = new JobMetricsChecker(job2); assertTrueEventually(() -> checker2.assertSummedMetricValue("total", 3000)); }
@Test public void availableViaJmx() throws Exception { int generatedItems = 1000; pipeline.readFrom(TestSources.itemStream(1_000)) .withIngestionTimestamps() .filter(l -> l.sequence() < generatedItems) .map(t -> { Metrics.metric("total").increment(); return t; }) .writeTo(Sinks.noop()); Job job = instance().newJob(pipeline, JOB_CONFIG_WITH_METRICS); JobMetricsChecker jobMetricsChecker = new JobMetricsChecker(job); assertTrueEventually(() -> jobMetricsChecker.assertSummedMetricValue("total", generatedItems)); String instanceName = instance().getHazelcastInstance().getName(); long sum = 0; int availableProcessors = Runtime.getRuntime().availableProcessors(); for (int i = 0; i < availableProcessors; i++) { JmxMetricsChecker jmxMetricsChecker = new JmxMetricsChecker(instanceName, job, "vertex=fused(filter, map)", "procType=TransformP", "proc=" + i, "user=true"); long attributeValue = jmxMetricsChecker.getMetricValue("total"); sum += attributeValue; } assertEquals(generatedItems, sum); }