@Test public void runShouldDoNothingWhenMetricPusherCreationFails() { PortBindings portBindings = mock(PortBindings.class); Configuration conf = mock(Configuration.class); Map<String, String> parameters = Collections.singletonMap("reportsDirPath", "/path/to/report"); MetricPusherCreator metricPusherCreator = mock(MetricPusherCreator.class); when(metricPusherCreatorProducer.create(parameters)).thenReturn(Optional.of(metricPusherCreator)); when(metricPusherProducer.create(any(MetricPusherCreator.class), anyMapOf(String.class, String.class))).thenReturn(Optional.empty()); pushMetricsProcess.run(portBindings, conf, parameters); verify(metricPusherCreatorProducer, times(1)).create(parameters); verify(metricPusherProducer, times(1)).create(metricPusherCreator, parameters); }
@Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) { metricPusherCreatorProducer.create(parameters) .ifPresent(metricPusherCreator -> { logger.info("Using MetricPusherCreatorProducer: {}", metricPusherCreator); metricPusherProducer.create(metricPusherCreator, parameters) .ifPresent(metricPusher -> { logger.info("Using MetricPusher: {}", metricPusher); fileSystemProducer.create(conf) .ifPresent(fs -> { logger.info("Using FileSystem: {}", fs); reportLocationsFinder.find(parameters) .ifPresent(paths -> { logger.info("Using report locations of size: {}", paths.size()); labeledMetricConfByPatternProducer.create(parameters) .ifPresent(labeledMetricConfByPattern -> { logger.info("Using label map of size: {}", labeledMetricConfByPattern.size()); List<Optional<List<ReportEntry>>> collect = paths.stream() .map(path -> reportEntryReader.read(fs, new Path(path))) .collect(Collectors.toList()); unwrapOptionals(collect) .map(x -> x.stream().flatMap(Collection::stream).collect(Collectors.toList())) .ifPresent(reportEntries -> { logger.info("Using report entries of size: {}", reportEntries.size()); reportEntryConverter.convert(reportEntries, parameters.get(REPORTS_DIR_PATH), labeledMetricConfByPattern) .ifPresent(gauges -> { logger.info("Using gauges of size: {}", gauges.size()); gaugesRegistrar.register(gauges) .ifPresent(registry -> { logger.info("Using registry: {}", registry); groupingKeyProducer.create(parameters) .ifPresent(groupingKey -> { logger.info("Using grouping key of size: {}", groupingKey.size()); metricPusher.pushSafe(registry, JOB, groupingKey); }); }); }); }); }); }); }); }); }); }
PushMetricsProcess implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) { metricPusherCreatorProducer.create(parameters) .ifPresent(metricPusherCreator -> { logger.info("Using MetricPusherCreatorProducer: {}", metricPusherCreator); metricPusherProducer.create(metricPusherCreator, parameters) .ifPresent(metricPusher -> { logger.info("Using MetricPusher: {}", metricPusher); fileSystemProducer.create(conf) .ifPresent(fs -> { logger.info("Using FileSystem: {}", fs); reportLocationsFinder.find(parameters) .ifPresent(paths -> { logger.info("Using report locations of size: {}", paths.size()); labeledMetricConfByPatternProducer.create(parameters) .ifPresent(labeledMetricConfByPattern -> { logger.info("Using label map of size: {}", labeledMetricConfByPattern.size()); List<Optional<List<ReportEntry>>> collect = paths.stream() .map(path -> reportEntryReader.read(fs, new Path(path))) .collect(Collectors.toList()); unwrapOptionals(collect) .map(x -> x.stream().flatMap(Collection::stream).collect(Collectors.toList())) .ifPresent(reportEntries -> { logger.info("Using report entries of size: {}", reportEntries.size()); reportEntryConverter.convert(reportEntries, parameters.get(REPORTS_DIR_PATH), labeledMetricConfByPattern) .ifPresent(gauges -> { logger.info("Using gauges of size: {}", gauges.size()); gaugesRegistrar.register(gauges) .ifPresent(registry -> { logger.info("Using registry: {}", registry); groupingKeyProducer.create(parameters) .ifPresent(groupingKey -> { logger.info("Using grouping key of size: {}", groupingKey.size()); metricPusher.pushSafe(registry, JOB, groupingKey); }); }); }); }); }); }); }); }); }); } }
PushMetricsProcess implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) { metricPusherCreatorProducer.create(parameters) .ifPresent(metricPusherCreator -> { logger.info("Using MetricPusherCreatorProducer: {}", metricPusherCreator); metricPusherProducer.create(metricPusherCreator, parameters) .ifPresent(metricPusher -> { logger.info("Using MetricPusher: {}", metricPusher); fileSystemProducer.create(conf) .ifPresent(fs -> { logger.info("Using FileSystem: {}", fs); reportLocationsFinder.find(parameters) .ifPresent(paths -> { logger.info("Using report locations of size: {}", paths.size()); labeledMetricConfByPatternProducer.create(parameters) .ifPresent(labeledMetricConfByPattern -> { logger.info("Using label map of size: {}", labeledMetricConfByPattern.size()); List<Optional<List<ReportEntry>>> collect = paths.stream() .map(path -> reportEntryReader.read(fs, new Path(path))) .collect(Collectors.toList()); unwrapOptionals(collect) .map(x -> x.stream().flatMap(Collection::stream).collect(Collectors.toList())) .ifPresent(reportEntries -> { logger.info("Using report entries of size: {}", reportEntries.size()); reportEntryConverter.convert(reportEntries, parameters.get(REPORTS_DIR_PATH), labeledMetricConfByPattern) .ifPresent(gauges -> { logger.info("Using gauges of size: {}", gauges.size()); gaugesRegistrar.register(gauges) .ifPresent(registry -> { logger.info("Using registry: {}", registry); groupingKeyProducer.create(parameters) .ifPresent(groupingKey -> { logger.info("Using grouping key of size: {}", groupingKey.size()); metricPusher.pushSafe(registry, JOB, groupingKey); }); }); }); }); }); }); }); }); }); }  }
PushMetricsProcess implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) { metricPusherCreatorProducer.create(parameters) .ifPresent(metricPusherCreator -> { logger.info("Using MetricPusherCreatorProducer: {}", metricPusherCreator); metricPusherProducer.create(metricPusherCreator, parameters) .ifPresent(metricPusher -> { logger.info("Using MetricPusher: {}", metricPusher); fileSystemProducer.create(conf) .ifPresent(fs -> { logger.info("Using FileSystem: {}", fs); reportLocationsFinder.find(parameters) .ifPresent(paths -> { logger.info("Using report locations of size: {}", paths.size()); labeledMetricConfByPatternProducer.create(parameters) .ifPresent(labeledMetricConfByPattern -> { logger.info("Using label map of size: {}", labeledMetricConfByPattern.size()); List<Optional<List<ReportEntry>>> collect = paths.stream() .map(path -> reportEntryReader.read(fs, new Path(path))) .collect(Collectors.toList()); unwrapOptionals(collect) .map(x -> x.stream().flatMap(Collection::stream).collect(Collectors.toList())) .ifPresent(reportEntries -> { logger.info("Using report entries of size: {}", reportEntries.size()); reportEntryConverter.convert(reportEntries, parameters.get(REPORTS_DIR_PATH), labeledMetricConfByPattern) .ifPresent(gauges -> { logger.info("Using gauges of size: {}", gauges.size()); gaugesRegistrar.register(gauges) .ifPresent(registry -> { logger.info("Using registry: {}", registry); groupingKeyProducer.create(parameters) .ifPresent(groupingKey -> { logger.info("Using grouping key of size: {}", groupingKey.size()); metricPusher.pushSafe(registry, JOB, groupingKey); }); }); }); }); }); }); }); }); }); }  @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters); }
PushMetricsProcess implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) { metricPusherCreatorProducer.create(parameters) .ifPresent(metricPusherCreator -> { logger.info("Using MetricPusherCreatorProducer: {}", metricPusherCreator); metricPusherProducer.create(metricPusherCreator, parameters) .ifPresent(metricPusher -> { logger.info("Using MetricPusher: {}", metricPusher); fileSystemProducer.create(conf) .ifPresent(fs -> { logger.info("Using FileSystem: {}", fs); reportLocationsFinder.find(parameters) .ifPresent(paths -> { logger.info("Using report locations of size: {}", paths.size()); labeledMetricConfByPatternProducer.create(parameters) .ifPresent(labeledMetricConfByPattern -> { logger.info("Using label map of size: {}", labeledMetricConfByPattern.size()); List<Optional<List<ReportEntry>>> collect = paths.stream() .map(path -> reportEntryReader.read(fs, new Path(path))) .collect(Collectors.toList()); unwrapOptionals(collect) .map(x -> x.stream().flatMap(Collection::stream).collect(Collectors.toList())) .ifPresent(reportEntries -> { logger.info("Using report entries of size: {}", reportEntries.size()); reportEntryConverter.convert(reportEntries, parameters.get(REPORTS_DIR_PATH), labeledMetricConfByPattern) .ifPresent(gauges -> { logger.info("Using gauges of size: {}", gauges.size()); gaugesRegistrar.register(gauges) .ifPresent(registry -> { logger.info("Using registry: {}", registry); groupingKeyProducer.create(parameters) .ifPresent(groupingKey -> { logger.info("Using grouping key of size: {}", groupingKey.size()); metricPusher.pushSafe(registry, JOB, groupingKey); }); }); }); }); }); }); }); }); }); }  @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters);  }