@Test public void testWavefrontHandlerAsDDIEndpoint() throws Exception { port = findAvailablePort(2978); proxy.proxyConfig.pushListenerPorts = String.valueOf(port); proxy.proxyConfig.dataBackfillCutoffHours = 8640; proxy.startGraphiteListener(proxy.proxyConfig.getPushListenerPorts(), mockHandlerFactory, null, new SpanSampler(new DurationSampler(5000), proxy.proxyConfig.isTraceAlwaysSampleErrors())); waitUntilListenerIsOnline(port); String traceId = UUID.randomUUID().toString(); long timestamp1 = startTime * 1000000 + 12345; long timestamp2 = startTime * 1000000 + 23456; String payloadStr = "metric4.test 0 " + startTime + " source=test1\n" + "metric4.test 1 " + (startTime + 1) + " source=test2\n" + "metric4.test 2 " + (startTime + 2) + " source=test3"; String histoData = "!M " + startTime + " #5 10.0 #10 100.0 metric.test.histo source=test1\n" + "!M " + (startTime + 60) + " #5 20.0 #6 30.0 #7 40.0 metric.test.histo source=test2"; String spanData = "testSpanName parent=parent1 source=testsource spanId=testspanid " + "traceId=\"" + traceId + "\" parent=parent2 " + startTime + " " + (startTime + 10); String spanDataToDiscard = "testSpanName parent=parent1 source=testsource spanId=testspanid " + "traceId=\"" + traceId + "\" parent=parent2 " + startTime + " " + (startTime + 1); String spanLogData = "{\"spanId\":\"testspanid\",\"traceId\":\"" + traceId + "\",\"logs\":[{\"timestamp\":" + timestamp1 + ",\"fields\":{\"key\":\"value\",\"key2\":\"value2\"}},{\"timestamp\":" + timestamp2 + ",\"fields\":{\"key3\":\"value3\",\"key4\":\"value4\"}}]}\n"; String spanLogDataWithSpanField = "{\"spanId\":\"testspanid\",\"traceId\":\"" + traceId + "\",\"logs\":[{\"timestamp\":" + timestamp1 + ",\"fields\":{\"key\":\"value\",\"key2\":\"value2\"}},{\"timestamp\":" + timestamp2 + ",\"fields\":{\"key3\":\"value3\"}}]," + "\"span\":\"" + escapeSpanData(spanData) + "\"}\n"; String spanLogDataWithSpanFieldToDiscard = "{\"spanId\":\"testspanid\",\"traceId\":\"" + traceId + "\",\"logs\":[{\"timestamp\":" + timestamp1 + ",\"fields\":{\"key\":\"value\",\"key2\":\"value2\"}}]," + "\"span\":\"" + escapeSpanData(spanDataToDiscard) + "\"}\n"; String mixedData = "@SourceTag action=save source=testSource newtag1 newtag2\n" + "@Event " + startTime + " \"Event name for testing\" host=host1 host=host2 tag=tag1 " + "severity=INFO multi=bar multi=baz\n" + "!M " + (startTime + 60) + " #5 20.0 #6 30.0 #7 40.0 metric.test.histo source=test2\n" + "metric4.test 0 " + startTime + " source=test1\n" + spanLogData + spanLogDataWithSpanField; String invalidData = "{\"spanId\"}\n@SourceTag\n@Event\n!M #5\nmetric.name\n" + "metric5.test 0 1234567890 source=test1\n"; reset(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric4.test").setHost("test1").setTimestamp(startTime * 1000). setValue(0.0d).build()); expectLastCall().times(2); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric4.test").setHost("test2").setTimestamp((startTime + 1) * 1000). setValue(1.0d).build()); expectLastCall().times(2); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric4.test").setHost("test3").setTimestamp((startTime + 2) * 1000). setValue(2.0d).build()); expectLastCall().times(2); replay(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); assertEquals(202, gzippedHttpPost("http: assertEquals(202, gzippedHttpPost("http: "/report?format=wavefront", payloadStr)); verify(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); reset(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); mockHistogramHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric.test.histo").setHost("test1").setTimestamp(startTime * 1000).setValue( Histogram.newBuilder() .setType(HistogramType.TDIGEST) .setDuration(60000) .setBins(ImmutableList.of(10.0d, 100.0d)) .setCounts(ImmutableList.of(5, 10)) .build()) .build()); expectLastCall(); mockHistogramHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric.test.histo").setHost("test2").setTimestamp((startTime + 60) * 1000). setValue(Histogram.newBuilder() .setType(HistogramType.TDIGEST) .setDuration(60000) .setBins(ImmutableList.of(20.0d, 30.0d, 40.0d)) .setCounts(ImmutableList.of(5, 6, 7)) .build()) .build()); expectLastCall(); replay(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); assertEquals(202, gzippedHttpPost("http: "/report?format=histogram", histoData)); verify(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); reset(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); mockTraceSpanLogsHandler.report(SpanLogs.newBuilder(). setCustomer("dummy"). setTraceId(traceId). setSpanId("testspanid"). setLogs(ImmutableList.of( SpanLog.newBuilder(). setTimestamp(timestamp1). setFields(ImmutableMap.of("key", "value", "key2", "value2")). build(), SpanLog.newBuilder(). setTimestamp(timestamp2). setFields(ImmutableMap.of("key3", "value3", "key4", "value4")). build() )). build()); expectLastCall(); mockTraceSpanLogsHandler.report(SpanLogs.newBuilder(). setCustomer("dummy"). setTraceId(traceId). setSpanId("testspanid"). setLogs(ImmutableList.of( SpanLog.newBuilder(). setTimestamp(timestamp1). setFields(ImmutableMap.of("key", "value", "key2", "value2")). build(), SpanLog.newBuilder(). setTimestamp(timestamp2). setFields(ImmutableMap.of("key3", "value3")). build() )). build()); mockTraceHandler.report(Span.newBuilder().setCustomer("dummy").setStartMillis(startTime * 1000) .setDuration(10000) .setName("testSpanName") .setSource("testsource") .setSpanId("testspanid") .setTraceId(traceId) .setAnnotations(ImmutableList.of(new Annotation("parent", "parent1"), new Annotation("parent", "parent2"))) .build()); expectLastCall(); replay(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); assertEquals(202, gzippedHttpPost("http: "/report?format=trace", spanData)); assertEquals(202, gzippedHttpPost("http: "/report?format=spanLogs", spanLogData)); assertEquals(202, gzippedHttpPost("http: "/report?format=spanLogs", spanLogDataWithSpanField)); assertEquals(202, gzippedHttpPost("http: "/report?format=trace", spanDataToDiscard)); assertEquals(202, gzippedHttpPost("http: "/report?format=spanLogs", spanLogDataWithSpanFieldToDiscard)); verify(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); reset(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); mockSourceTagHandler.report(ReportSourceTag.newBuilder(). setOperation(SourceOperationType.SOURCE_TAG).setAction(SourceTagAction.SAVE). setSource("testSource").setAnnotations(ImmutableList.of("newtag1", "newtag2")).build()); expectLastCall(); mockEventHandler.report(ReportEvent.newBuilder().setStartTime(startTime * 1000). setEndTime(startTime * 1000 + 1).setName("Event name for testing"). setHosts(ImmutableList.of("host1", "host2")).setTags(ImmutableList.of("tag1")). setAnnotations(ImmutableMap.of("severity", "INFO")). setDimensions(ImmutableMap.of("multi", ImmutableList.of("bar", "baz"))).build()); expectLastCall(); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric4.test").setHost("test1").setTimestamp(startTime * 1000). setValue(0.0d).build()); expectLastCall(); replay(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); proxy.entityProps.get(ReportableEntityType.HISTOGRAM).setFeatureDisabled(true); assertEquals(403, gzippedHttpPost("http: "/report?format=histogram", histoData)); proxy.entityProps.get(ReportableEntityType.TRACE).setFeatureDisabled(true); assertEquals(403, gzippedHttpPost("http: "/report?format=trace", spanData)); proxy.entityProps.get(ReportableEntityType.TRACE_SPAN_LOGS).setFeatureDisabled(true); assertEquals(403, gzippedHttpPost("http: "/report?format=spanLogs", spanLogData)); assertEquals(403, gzippedHttpPost("http: "/report?format=spanLogs", spanLogDataWithSpanField)); assertEquals(202, gzippedHttpPost("http: verify(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); reset(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); mockSourceTagHandler.report(ReportSourceTag.newBuilder(). setOperation(SourceOperationType.SOURCE_TAG).setAction(SourceTagAction.SAVE). setSource("testSource").setAnnotations(ImmutableList.of("newtag1", "newtag2")).build()); expectLastCall(); mockEventHandler.report(ReportEvent.newBuilder().setStartTime(startTime * 1000). setEndTime(startTime * 1000 + 1).setName("Event name for testing"). setHosts(ImmutableList.of("host1", "host2")).setTags(ImmutableList.of("tag1")). setAnnotations(ImmutableMap.of("severity", "INFO")). setDimensions(ImmutableMap.of("multi", ImmutableList.of("bar", "baz"))).build()); expectLastCall(); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy").setMetric("metric4.test"). setHost("test1").setTimestamp(startTime * 1000).setValue(0.0d).build()); expectLastCall(); mockSourceTagHandler.reject(eq("@SourceTag"), anyString()); expectLastCall(); mockEventHandler.reject(eq("@Event"), anyString()); expectLastCall(); mockPointHandler.reject(eq("metric.name"), anyString()); expectLastCall(); mockPointHandler.reject(eq(ReportPoint.newBuilder().setTable("dummy").setMetric("metric5.test"). setHost("test1").setTimestamp(1234567890000L).setValue(0.0d).build()), startsWith("WF-402: Point outside of reasonable timeframe")); expectLastCall(); replay(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); assertEquals(202, gzippedHttpPost("http: mixedData + "\n" + invalidData)); verify(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler, mockSourceTagHandler, mockEventHandler); }
@VisibleForTesting protected void startGraphiteListener(String strPort, ReportableEntityHandlerFactory handlerFactory, SharedGraphiteHostAnnotator hostAnnotator, SpanSampler sampler) { final int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); WavefrontPortUnificationHandler wavefrontPortUnificationHandler = new WavefrontPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, decoderSupplier.get(), handlerFactory, hostAnnotator, preprocessors.get(strPort), () -> entityProps.get(ReportableEntityType.HISTOGRAM).isFeatureDisabled(), () -> entityProps.get(ReportableEntityType.TRACE).isFeatureDisabled(), () -> entityProps.get(ReportableEntityType.TRACE_SPAN_LOGS).isFeatureDisabled(), sampler); startAsManagedThread(port, new TcpIngester(createInitializer(wavefrontPortUnificationHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-graphite-" + port); }
PushAgent extends AbstractAgent { @VisibleForTesting protected void startGraphiteListener(String strPort, ReportableEntityHandlerFactory handlerFactory, SharedGraphiteHostAnnotator hostAnnotator, SpanSampler sampler) { final int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); WavefrontPortUnificationHandler wavefrontPortUnificationHandler = new WavefrontPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, decoderSupplier.get(), handlerFactory, hostAnnotator, preprocessors.get(strPort), () -> entityProps.get(ReportableEntityType.HISTOGRAM).isFeatureDisabled(), () -> entityProps.get(ReportableEntityType.TRACE).isFeatureDisabled(), () -> entityProps.get(ReportableEntityType.TRACE_SPAN_LOGS).isFeatureDisabled(), sampler); startAsManagedThread(port, new TcpIngester(createInitializer(wavefrontPortUnificationHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-graphite-" + port); } }
PushAgent extends AbstractAgent { @VisibleForTesting protected void startGraphiteListener(String strPort, ReportableEntityHandlerFactory handlerFactory, SharedGraphiteHostAnnotator hostAnnotator, SpanSampler sampler) { final int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); WavefrontPortUnificationHandler wavefrontPortUnificationHandler = new WavefrontPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, decoderSupplier.get(), handlerFactory, hostAnnotator, preprocessors.get(strPort), () -> entityProps.get(ReportableEntityType.HISTOGRAM).isFeatureDisabled(), () -> entityProps.get(ReportableEntityType.TRACE).isFeatureDisabled(), () -> entityProps.get(ReportableEntityType.TRACE_SPAN_LOGS).isFeatureDisabled(), sampler); startAsManagedThread(port, new TcpIngester(createInitializer(wavefrontPortUnificationHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-graphite-" + port); }  }
PushAgent extends AbstractAgent { @VisibleForTesting protected void startGraphiteListener(String strPort, ReportableEntityHandlerFactory handlerFactory, SharedGraphiteHostAnnotator hostAnnotator, SpanSampler sampler) { final int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); WavefrontPortUnificationHandler wavefrontPortUnificationHandler = new WavefrontPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, decoderSupplier.get(), handlerFactory, hostAnnotator, preprocessors.get(strPort), () -> entityProps.get(ReportableEntityType.HISTOGRAM).isFeatureDisabled(), () -> entityProps.get(ReportableEntityType.TRACE).isFeatureDisabled(), () -> entityProps.get(ReportableEntityType.TRACE_SPAN_LOGS).isFeatureDisabled(), sampler); startAsManagedThread(port, new TcpIngester(createInitializer(wavefrontPortUnificationHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-graphite-" + port); }  static void main(String[] args); @Override void stopListeners(); }
PushAgent extends AbstractAgent { @VisibleForTesting protected void startGraphiteListener(String strPort, ReportableEntityHandlerFactory handlerFactory, SharedGraphiteHostAnnotator hostAnnotator, SpanSampler sampler) { final int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); WavefrontPortUnificationHandler wavefrontPortUnificationHandler = new WavefrontPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, decoderSupplier.get(), handlerFactory, hostAnnotator, preprocessors.get(strPort), () -> entityProps.get(ReportableEntityType.HISTOGRAM).isFeatureDisabled(), () -> entityProps.get(ReportableEntityType.TRACE).isFeatureDisabled(), () -> entityProps.get(ReportableEntityType.TRACE_SPAN_LOGS).isFeatureDisabled(), sampler); startAsManagedThread(port, new TcpIngester(createInitializer(wavefrontPortUnificationHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-graphite-" + port); }  static void main(String[] args); @Override void stopListeners();  }