@Test public void testLargeHistogramDataOnWavefrontUnifiedPortHandler() throws Exception { port = findAvailablePort(2988); 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); List<Double> bins = new ArrayList<>(); List<Integer> counts = new ArrayList<>(); for (int i = 0; i < 50; i++) bins.add(10.0d); for (int i = 0; i < 150; i++) bins.add(99.0d); for (int i = 0; i < 200; i++) counts.add(1); mockHistogramHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric.test.histo").setHost("test1").setTimestamp(startTime * 1000).setValue( Histogram.newBuilder() .setType(HistogramType.TDIGEST) .setDuration(60000) .setBins(bins) .setCounts(counts) .build()) .build()); expectLastCall(); replay(mockHistogramHandler); Socket socket = SocketFactory.getDefault().createSocket("localhost", port); BufferedOutputStream stream = new BufferedOutputStream(socket.getOutputStream()); StringBuilder payloadStr = new StringBuilder("!M "); payloadStr.append(startTime); for (int i = 0; i < 50; i++) payloadStr.append(" #1 10.0"); for (int i = 0; i < 150; i++) payloadStr.append(" #1 99.0"); payloadStr.append(" metric.test.histo source=test1\n"); stream.write(payloadStr.toString().getBytes()); stream.flush(); socket.close(); verifyWithTimeout(500, mockHistogramHandler); }
@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();  }