PushAgent extends AbstractAgent { @VisibleForTesting protected void startRelayListener(String strPort, ReportableEntityHandlerFactory handlerFactory, SharedGraphiteHostAnnotator hostAnnotator) { final int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); ReportableEntityHandlerFactory handlerFactoryDelegate = proxyConfig. isPushRelayHistogramAggregator() ? new DelegatingReportableEntityHandlerFactoryImpl(handlerFactory) { @Override public <T, U> ReportableEntityHandler<T, U> getHandler(HandlerKey handlerKey) { if (handlerKey.getEntityType() == ReportableEntityType.HISTOGRAM) { ChronicleMap<HistogramKey, AgentDigest> accumulator = ChronicleMap. of(HistogramKey.class, AgentDigest.class). keyMarshaller(HistogramKeyMarshaller.get()). valueMarshaller(AgentDigestMarshaller.get()). entries(proxyConfig.getPushRelayHistogramAggregatorAccumulatorSize()). averageKeySize(proxyConfig.getHistogramDistAvgKeyBytes()). averageValueSize(proxyConfig.getHistogramDistAvgDigestBytes()). maxBloatFactor(1000). create(); AgentDigestFactory agentDigestFactory = new AgentDigestFactory(() -> (short) Math.min(proxyConfig.getPushRelayHistogramAggregatorCompression(), entityProps.getGlobalProperties().getHistogramStorageAccuracy()), TimeUnit.SECONDS.toMillis(proxyConfig.getPushRelayHistogramAggregatorFlushSecs()), proxyConfig.getTimeProvider()); AccumulationCache cachedAccumulator = new AccumulationCache(accumulator, agentDigestFactory, 0, "histogram.accumulator.distributionRelay", null); return (ReportableEntityHandler<T, U>) new HistogramAccumulationHandlerImpl( handlerKey, cachedAccumulator, proxyConfig.getPushBlockedSamples(), null, validationConfiguration, true, rate -> entityProps.get(ReportableEntityType.HISTOGRAM). reportReceivedRate(handlerKey.getHandle(), rate), blockedHistogramsLogger, VALID_HISTOGRAMS_LOGGER); } return delegate.getHandler(handlerKey); } } : handlerFactory; Map<ReportableEntityType, ReportableEntityDecoder<?, ?>> filteredDecoders = decoderSupplier.get().entrySet().stream(). filter(x -> !x.getKey().equals(ReportableEntityType.SOURCE_TAG)). collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); ChannelHandler channelHandler = new RelayPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, filteredDecoders, handlerFactoryDelegate, preprocessors.get(strPort), hostAnnotator, () -> entityProps.get(ReportableEntityType.HISTOGRAM).isFeatureDisabled(), () -> entityProps.get(ReportableEntityType.TRACE).isFeatureDisabled(), () -> entityProps.get(ReportableEntityType.TRACE_SPAN_LOGS).isFeatureDisabled()); startAsManagedThread(port, new TcpIngester(createInitializer(channelHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-relay-" + port); }  static void main(String[] args); @Override void stopListeners();  }
@Test public void testRelayPortHandlerGzipped() throws Exception { port = findAvailablePort(2888); proxy.proxyConfig.pushRelayListenerPorts = String.valueOf(port); proxy.proxyConfig.pushRelayHistogramAggregator = true; proxy.proxyConfig.pushRelayHistogramAggregatorAccumulatorSize = 10L; proxy.proxyConfig.pushRelayHistogramAggregatorFlushSecs = 1; proxy.startRelayListener(proxy.proxyConfig.getPushRelayListenerPorts(), mockHandlerFactory, null); waitUntilListenerIsOnline(port); reset(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler); String traceId = UUID.randomUUID().toString(); long timestamp1 = startTime * 1000000 + 12345; long timestamp2 = startTime * 1000000 + 23456; String spanData = "testSpanName parent=parent1 source=testsource spanId=testspanid " + "traceId=\"" + traceId + "\" parent=parent2 " + startTime + " " + (startTime + 1); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric4.test").setHost("test1").setTimestamp(startTime * 1000).setValue(0.0d).build()); expectLastCall(); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric4.test").setHost("test2").setTimestamp((startTime + 1) * 1000).setValue(1.0d).build()); expectLastCall(); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric4.test").setHost("test3").setTimestamp((startTime + 2) * 1000).setValue(2.0d).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", "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() )). setSpan(spanData). build()); expectLastCall(); mockTraceHandler.report(Span.newBuilder().setCustomer("dummy").setStartMillis(startTime * 1000) .setDuration(1000) .setName("testSpanName") .setSource("testsource") .setSpanId("testspanid") .setTraceId(traceId) .setAnnotations(ImmutableList.of(new Annotation("parent", "parent1"), new Annotation("parent", "parent2"))) .build()); expectLastCall(); mockPointHandler.reject(anyString(), anyString()); expectLastCall().times(2); replay(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler); 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 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 badData = "@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"; assertEquals(200, gzippedHttpPost("http: "{}")); assertEquals(200, gzippedHttpPost("http: "/api/v2/wfproxy/report?format=wavefront", payloadStr)); assertEquals(200, gzippedHttpPost("http: "/api/v2/wfproxy/report?format=histogram", histoData)); assertEquals(200, gzippedHttpPost("http: "/api/v2/wfproxy/report?format=trace", spanData)); assertEquals(200, gzippedHttpPost("http: "/api/v2/wfproxy/report?format=spanLogs", spanLogData)); assertEquals(200, gzippedHttpPost("http: "/api/v2/wfproxy/report?format=spanLogs", spanLogDataWithSpanField)); proxy.entityProps.get(ReportableEntityType.HISTOGRAM).setFeatureDisabled(true); assertEquals(403, gzippedHttpPost("http: "/api/v2/wfproxy/report?format=histogram", histoData)); proxy.entityProps.get(ReportableEntityType.TRACE).setFeatureDisabled(true); assertEquals(403, gzippedHttpPost("http: "/api/v2/wfproxy/report?format=trace", spanData)); proxy.entityProps.get(ReportableEntityType.TRACE_SPAN_LOGS).setFeatureDisabled(true); assertEquals(403, gzippedHttpPost("http: "/api/v2/wfproxy/report?format=spanLogs", spanLogData)); assertEquals(403, gzippedHttpPost("http: "/api/v2/wfproxy/report?format=spanLogs", spanLogDataWithSpanField)); assertEquals(400, gzippedHttpPost("http: "/api/v2/wfproxy/report?format=wavefront", badData)); verify(mockPointHandler, mockHistogramHandler, mockTraceHandler, mockTraceSpanLogsHandler); }