ElasticMetricsReporter implements Consumer<MetricsDump> { private Single<BulkRequest> requestFor (Observable<Map<String, Object>> metricsAsMaps) { return metricsAsMaps .map(map -> new IndexRequest() .index(indexNameSupplier.get()) .source(map) ) .reduce(Requests.bulkRequest(), (bulkRequestBuilder, indexRequest) -> { bulkRequestBuilder.add(indexRequest); return bulkRequestBuilder; }) ; } ElasticMetricsReporter(String indexName, String ... elasticServers);  ElasticMetricsReporter(String indexName, boolean createDailyIndices, String ... elasticServers);  ElasticMetricsReporter(String indexName, HttpHost ... elasticServers);  ElasticMetricsReporter(String indexName, boolean createDailyIndices, HttpHost ... elasticServers); @Override void accept(MetricsDump metricsDump); void accept(MetricsDump metricsDump, Consumer<Throwable> exceptionHandler); void accept(SerializableMetricsDump metricsDump); void accept(SerializableMetricsDump metricsDump, Consumer<Throwable> exceptionHandler); void startup(); void shutdown();  }
@Test void requestFromMetricsDumpIsCreatedAsExpected() throws Exception { Metrics metrics = new Metrics(); metrics.getCounter("test","counter1"); metrics.getMeter("test","meter1"); metrics.register(new MyMetric(), "test","myMetric1"); metrics.addAdditionalInfoForDumps("host", "myMachine"); metrics.addAdditionalInfoForDumps("hostAddress", "myAddress"); metrics.addAdditionalInfoForDumps("additionalKey", "additionalValue"); MetricsDump dump = metrics.dump(); TestObserver<BulkRequest> bulkRequestObserver = sut.requestFor(dump).test(); bulkRequestObserver.assertComplete(); bulkRequestObserver.assertValueCount(1); BulkRequest bulkRequest = bulkRequestObserver.values().get(0); List<DocWriteRequest<?>> requests = bulkRequest.requests(); assertThat(requests.size(), is(3)); for (DocWriteRequest request: requests) { assertTrue(request instanceof IndexRequest); IndexRequest ir = (IndexRequest)request; Map<String,Object> sourceAsMap = getMapFrom(ir.source()); assertThat(sourceAsMap.get("@timestamp"), is(notNullValue())); dump.additionalInfo.forEach(entry -> { assertThat(sourceAsMap.get(entry._1), is(entry._2)); }); assertThat(sourceAsMap.get("name"),Matchers.oneOf("test.counter1","test.meter1","test.myMetric1")); } }
@Test void requestFromSerializableMetricsDumpIsCreatedAsExpected() throws Exception { Metrics metrics = new Metrics(); metrics.getCounter("test","counter1"); metrics.getMeter("test","meter1"); metrics.register(new MyMetric(), "test","myMetric1"); metrics.addAdditionalInfoForDumps("host", "myMachine2"); metrics.addAdditionalInfoForDumps("hostAddress", "myAddress2"); metrics.addAdditionalInfoForDumps("additionalKey", "additionalValue2"); MetricsDump dump = metrics.dump(); SerializableMetricsDump serializableDump = SerializableMetricsDump.createFor(dump); TestObserver<BulkRequest> bulkRequestObserver = sut.requestFor(serializableDump).test(); bulkRequestObserver.assertComplete(); bulkRequestObserver.assertValueCount(1); BulkRequest bulkRequest = bulkRequestObserver.values().get(0); List<DocWriteRequest<?>> requests = bulkRequest.requests(); assertThat(requests.size(), is(3)); for (DocWriteRequest request: requests) { assertTrue(request instanceof IndexRequest); IndexRequest ir = (IndexRequest)request; Map<String,Object> sourceAsMap = getMapFrom(ir.source()); assertThat(sourceAsMap.get("@timestamp"), is(notNullValue())); dump.additionalInfo.forEach(entry -> { assertThat(sourceAsMap.get(entry._1), is(entry._2)); }); assertThat(sourceAsMap.get("name"), Matchers.oneOf("test.counter1", "test.meter1", "test.myMetric1")); } }
@Test void requestFromMapDumpIsCreatedAsExpected() throws Exception { Map<String, Object> dumpAsMap = new HashMap<>(); Arrays.asList(new Pair<>("counter", "counter1"), new Pair<>("meter", "meter1"), new Pair<>("MyMetric", "myMetric1")) .forEach(pair -> { Map<String, Object> metricMap = new HashMap<>(); metricMap.put("type", pair._1); metricMap.put("someAttribute", "someVal"); dumpAsMap.put("test." + pair._2, metricMap); }); dumpAsMap.put("hostName", "someVal"); TestObserver<BulkRequest> bulkRequestObserver = sut.requestFor(dumpAsMap).test(); bulkRequestObserver.assertComplete(); bulkRequestObserver.assertValueCount(1); BulkRequest bulkRequest = bulkRequestObserver.values().get(0); List<DocWriteRequest<?>> requests = bulkRequest.requests(); assertThat(requests.size(), is(3)); for (DocWriteRequest request: requests) { assertTrue(request instanceof IndexRequest); IndexRequest ir = (IndexRequest) request; Map<String, Object> sourceAsMap = getMapFrom(ir.source()); assertNotNull(sourceAsMap.get("@timestamp")); assertNotNull(sourceAsMap.get("someAttribute")); assertThat(sourceAsMap.get("name"), Matchers.oneOf("test.counter1", "test.meter1", "test.myMetric1")); assertThat(sourceAsMap.get("type"), Matchers.oneOf("counter", "meter", "MyMetric")); } }