@Test public void testDeltaCounterHandlerDataStream() throws Exception { deltaPort = findAvailablePort(5888); proxy.proxyConfig.deltaCountersAggregationListenerPorts = String.valueOf(deltaPort); proxy.proxyConfig.deltaCountersAggregationIntervalSeconds = 10; proxy.startDeltaCounterListener(proxy.proxyConfig.getDeltaCountersAggregationListenerPorts(), null, mockSenderTaskFactory, new SpanSampler(new RateSampler(1.0D), proxy.proxyConfig.isTraceAlwaysSampleErrors())); waitUntilListenerIsOnline(deltaPort); reset(mockSenderTask); Capture<String> capturedArgument = Capture.newInstance(CaptureType.ALL); mockSenderTask.add(EasyMock.capture(capturedArgument)); expectLastCall().atLeastOnce(); replay(mockSenderTask); String payloadStr = "∆test.mixed 1.0 " + startTime + " source=test1\n"; assertEquals(202, httpPost("http: ReportableEntityHandler<?, ?> handler = proxy.deltaCounterHandlerFactory. getHandler(HandlerKey.of(ReportableEntityType.POINT, String.valueOf(deltaPort))); if (!(handler instanceof DeltaCounterAccumulationHandlerImpl)) fail(); ((DeltaCounterAccumulationHandlerImpl) handler).flushDeltaCounters(); assertEquals(202, httpPost("http: assertEquals(202, httpPost("http: ((DeltaCounterAccumulationHandlerImpl) handler).flushDeltaCounters(); verify(mockSenderTask); assertEquals(2, capturedArgument.getValues().size()); assertTrue(capturedArgument.getValues().get(0).startsWith("\"∆test.mixed\" 2.0" )); assertTrue(capturedArgument.getValues().get(1).startsWith("\"∆test.mixed\" 3.0" )); }
@VisibleForTesting protected void startDeltaCounterListener(String strPort, SharedGraphiteHostAnnotator hostAnnotator, SenderTaskFactory senderTaskFactory, SpanSampler sampler) { final int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); if (this.deltaCounterHandlerFactory == null) { this.deltaCounterHandlerFactory = new ReportableEntityHandlerFactory() { private final Map<String, ReportableEntityHandler<?, ?>> handlers = new ConcurrentHashMap<>(); @Override public <T, U> ReportableEntityHandler<T, U> getHandler(HandlerKey handlerKey) { return (ReportableEntityHandler<T, U>) handlers.computeIfAbsent(handlerKey.getHandle(), k -> new DeltaCounterAccumulationHandlerImpl(handlerKey, proxyConfig.getPushBlockedSamples(), senderTaskFactory.createSenderTasks(handlerKey), validationConfiguration, proxyConfig.getDeltaCountersAggregationIntervalSeconds(), rate -> entityProps.get(ReportableEntityType.POINT). reportReceivedRate(handlerKey.getHandle(), rate), blockedPointsLogger, VALID_POINTS_LOGGER)); } @Override public void shutdown(@Nonnull String handle) { if (handlers.containsKey(handle)) { handlers.values().forEach(ReportableEntityHandler::shutdown); } } }; } shutdownTasks.add(() -> deltaCounterHandlerFactory.shutdown(strPort)); WavefrontPortUnificationHandler wavefrontPortUnificationHandler = new WavefrontPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, decoderSupplier.get(), deltaCounterHandlerFactory, hostAnnotator, preprocessors.get(strPort), () -> false, () -> false, () -> false, sampler); startAsManagedThread(port, new TcpIngester(createInitializer(wavefrontPortUnificationHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-deltaCounter-" + port); }
PushAgent extends AbstractAgent { @VisibleForTesting protected void startDeltaCounterListener(String strPort, SharedGraphiteHostAnnotator hostAnnotator, SenderTaskFactory senderTaskFactory, SpanSampler sampler) { final int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); if (this.deltaCounterHandlerFactory == null) { this.deltaCounterHandlerFactory = new ReportableEntityHandlerFactory() { private final Map<String, ReportableEntityHandler<?, ?>> handlers = new ConcurrentHashMap<>(); @Override public <T, U> ReportableEntityHandler<T, U> getHandler(HandlerKey handlerKey) { return (ReportableEntityHandler<T, U>) handlers.computeIfAbsent(handlerKey.getHandle(), k -> new DeltaCounterAccumulationHandlerImpl(handlerKey, proxyConfig.getPushBlockedSamples(), senderTaskFactory.createSenderTasks(handlerKey), validationConfiguration, proxyConfig.getDeltaCountersAggregationIntervalSeconds(), rate -> entityProps.get(ReportableEntityType.POINT). reportReceivedRate(handlerKey.getHandle(), rate), blockedPointsLogger, VALID_POINTS_LOGGER)); } @Override public void shutdown(@Nonnull String handle) { if (handlers.containsKey(handle)) { handlers.values().forEach(ReportableEntityHandler::shutdown); } } }; } shutdownTasks.add(() -> deltaCounterHandlerFactory.shutdown(strPort)); WavefrontPortUnificationHandler wavefrontPortUnificationHandler = new WavefrontPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, decoderSupplier.get(), deltaCounterHandlerFactory, hostAnnotator, preprocessors.get(strPort), () -> false, () -> false, () -> false, sampler); startAsManagedThread(port, new TcpIngester(createInitializer(wavefrontPortUnificationHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-deltaCounter-" + port); } }
PushAgent extends AbstractAgent { @VisibleForTesting protected void startDeltaCounterListener(String strPort, SharedGraphiteHostAnnotator hostAnnotator, SenderTaskFactory senderTaskFactory, SpanSampler sampler) { final int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); if (this.deltaCounterHandlerFactory == null) { this.deltaCounterHandlerFactory = new ReportableEntityHandlerFactory() { private final Map<String, ReportableEntityHandler<?, ?>> handlers = new ConcurrentHashMap<>(); @Override public <T, U> ReportableEntityHandler<T, U> getHandler(HandlerKey handlerKey) { return (ReportableEntityHandler<T, U>) handlers.computeIfAbsent(handlerKey.getHandle(), k -> new DeltaCounterAccumulationHandlerImpl(handlerKey, proxyConfig.getPushBlockedSamples(), senderTaskFactory.createSenderTasks(handlerKey), validationConfiguration, proxyConfig.getDeltaCountersAggregationIntervalSeconds(), rate -> entityProps.get(ReportableEntityType.POINT). reportReceivedRate(handlerKey.getHandle(), rate), blockedPointsLogger, VALID_POINTS_LOGGER)); } @Override public void shutdown(@Nonnull String handle) { if (handlers.containsKey(handle)) { handlers.values().forEach(ReportableEntityHandler::shutdown); } } }; } shutdownTasks.add(() -> deltaCounterHandlerFactory.shutdown(strPort)); WavefrontPortUnificationHandler wavefrontPortUnificationHandler = new WavefrontPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, decoderSupplier.get(), deltaCounterHandlerFactory, hostAnnotator, preprocessors.get(strPort), () -> false, () -> false, () -> false, sampler); startAsManagedThread(port, new TcpIngester(createInitializer(wavefrontPortUnificationHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-deltaCounter-" + port); }  }
PushAgent extends AbstractAgent { @VisibleForTesting protected void startDeltaCounterListener(String strPort, SharedGraphiteHostAnnotator hostAnnotator, SenderTaskFactory senderTaskFactory, SpanSampler sampler) { final int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); if (this.deltaCounterHandlerFactory == null) { this.deltaCounterHandlerFactory = new ReportableEntityHandlerFactory() { private final Map<String, ReportableEntityHandler<?, ?>> handlers = new ConcurrentHashMap<>(); @Override public <T, U> ReportableEntityHandler<T, U> getHandler(HandlerKey handlerKey) { return (ReportableEntityHandler<T, U>) handlers.computeIfAbsent(handlerKey.getHandle(), k -> new DeltaCounterAccumulationHandlerImpl(handlerKey, proxyConfig.getPushBlockedSamples(), senderTaskFactory.createSenderTasks(handlerKey), validationConfiguration, proxyConfig.getDeltaCountersAggregationIntervalSeconds(), rate -> entityProps.get(ReportableEntityType.POINT). reportReceivedRate(handlerKey.getHandle(), rate), blockedPointsLogger, VALID_POINTS_LOGGER)); } @Override public void shutdown(@Nonnull String handle) { if (handlers.containsKey(handle)) { handlers.values().forEach(ReportableEntityHandler::shutdown); } } }; } shutdownTasks.add(() -> deltaCounterHandlerFactory.shutdown(strPort)); WavefrontPortUnificationHandler wavefrontPortUnificationHandler = new WavefrontPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, decoderSupplier.get(), deltaCounterHandlerFactory, hostAnnotator, preprocessors.get(strPort), () -> false, () -> false, () -> false, sampler); startAsManagedThread(port, new TcpIngester(createInitializer(wavefrontPortUnificationHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-deltaCounter-" + port); }  static void main(String[] args); @Override void stopListeners(); }
PushAgent extends AbstractAgent { @VisibleForTesting protected void startDeltaCounterListener(String strPort, SharedGraphiteHostAnnotator hostAnnotator, SenderTaskFactory senderTaskFactory, SpanSampler sampler) { final int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); if (this.deltaCounterHandlerFactory == null) { this.deltaCounterHandlerFactory = new ReportableEntityHandlerFactory() { private final Map<String, ReportableEntityHandler<?, ?>> handlers = new ConcurrentHashMap<>(); @Override public <T, U> ReportableEntityHandler<T, U> getHandler(HandlerKey handlerKey) { return (ReportableEntityHandler<T, U>) handlers.computeIfAbsent(handlerKey.getHandle(), k -> new DeltaCounterAccumulationHandlerImpl(handlerKey, proxyConfig.getPushBlockedSamples(), senderTaskFactory.createSenderTasks(handlerKey), validationConfiguration, proxyConfig.getDeltaCountersAggregationIntervalSeconds(), rate -> entityProps.get(ReportableEntityType.POINT). reportReceivedRate(handlerKey.getHandle(), rate), blockedPointsLogger, VALID_POINTS_LOGGER)); } @Override public void shutdown(@Nonnull String handle) { if (handlers.containsKey(handle)) { handlers.values().forEach(ReportableEntityHandler::shutdown); } } }; } shutdownTasks.add(() -> deltaCounterHandlerFactory.shutdown(strPort)); WavefrontPortUnificationHandler wavefrontPortUnificationHandler = new WavefrontPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, decoderSupplier.get(), deltaCounterHandlerFactory, hostAnnotator, preprocessors.get(strPort), () -> false, () -> false, () -> false, sampler); startAsManagedThread(port, new TcpIngester(createInitializer(wavefrontPortUnificationHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-deltaCounter-" + port); }  static void main(String[] args); @Override void stopListeners();  }