@Test public void testWavefrontUnifiedPortHandlerPlaintextUncompressedMixedDataPayload() throws Exception { port = findAvailablePort(2888); proxy.proxyConfig.pushListenerPorts = String.valueOf(port); proxy.startGraphiteListener(proxy.proxyConfig.getPushListenerPorts(), mockHandlerFactory, null, new SpanSampler(new RateSampler(1.0D), proxy.proxyConfig.isTraceAlwaysSampleErrors())); waitUntilListenerIsOnline(port); reset(mockHistogramHandler); reset(mockPointHandler); reset(mockSourceTagHandler); reset(mockEventHandler); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric.test.mixed").setHost("test2").setTimestamp((startTime + 1) * 1000).setValue(10d).build()); mockHistogramHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric.test.mixed").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()); mockEventHandler.report(ReportEvent.newBuilder(). setStartTime(startTime * 1000). setEndTime(startTime * 1000 + 1). setName("Event name for testing"). setHosts(ImmutableList.of("host1", "host2")). setDimensions(ImmutableMap.of("multi", ImmutableList.of("bar", "baz"))). setAnnotations(ImmutableMap.of("severity", "INFO")). setTags(ImmutableList.of("tag1")). build()); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric.test.mixed").setHost("test2").setTimestamp((startTime + 1) * 1000). setValue(9d).build()); mockSourceTagHandler.report(ReportSourceTag.newBuilder(). setOperation(SourceOperationType.SOURCE_TAG).setAction(SourceTagAction.SAVE). setSource("testSource").setAnnotations(ImmutableList.of("newtag1", "newtag2")).build()); expectLastCall(); replay(mockPointHandler); replay(mockHistogramHandler); replay(mockEventHandler); Socket socket = SocketFactory.getDefault().createSocket("localhost", port); BufferedOutputStream stream = new BufferedOutputStream(socket.getOutputStream()); String payloadStr = "metric.test.mixed 10.0 " + (startTime + 1) + " source=test2\n" + "!M " + startTime + " #5 10.0 #10 100.0 metric.test.mixed source=test1\n" + "@SourceTag action=save source=testSource newtag1 newtag2\n" + "metric.test.mixed 9.0 " + (startTime + 1) + " source=test2\n" + "@Event " + startTime + " \"Event name for testing\" host=host1 host=host2 tag=tag1 " + "severity=INFO multi=bar multi=baz\n"; stream.write(payloadStr.getBytes()); stream.flush(); socket.close(); verifyWithTimeout(500, mockPointHandler, mockHistogramHandler, 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();  }