PushAgent extends AbstractAgent { @VisibleForTesting protected void startCustomTracingListener(final String strPort, ReportableEntityHandlerFactory handlerFactory, @Nullable WavefrontSender wfSender, SpanSampler sampler) { final int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); WavefrontInternalReporter wfInternalReporter = null; if (wfSender != null) { wfInternalReporter = new WavefrontInternalReporter.Builder(). prefixedWith("tracing.derived").withSource("custom_tracing").reportMinuteDistribution(). build(wfSender); wfInternalReporter.start(1, TimeUnit.MINUTES); } ChannelHandler channelHandler = new CustomTracingPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, new SpanDecoder("unknown"), new SpanLogsDecoder(), preprocessors.get(strPort), handlerFactory, sampler, () -> entityProps.get(ReportableEntityType.TRACE).isFeatureDisabled(), () -> entityProps.get(ReportableEntityType.TRACE_SPAN_LOGS).isFeatureDisabled(), wfSender, wfInternalReporter, proxyConfig.getTraceDerivedCustomTagKeys(), proxyConfig.getCustomTracingApplicationName(), proxyConfig.getCustomTracingServiceName()); startAsManagedThread(port, new TcpIngester(createInitializer(channelHandler, port, proxyConfig.getTraceListenerMaxReceivedLength(), proxyConfig.getTraceListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-custom-trace-" + port); logger.info("listening on port: " + strPort + " for custom trace data"); }  static void main(String[] args); @Override void stopListeners();  }
@Test public void testCustomTraceUnifiedPortHandlerDerivedMetrics() throws Exception { customTracePort = findAvailablePort(51233); proxy.proxyConfig.customTracingListenerPorts = String.valueOf(customTracePort); setUserPreprocessorForTraceDerivedREDMetrics(customTracePort); proxy.startCustomTracingListener(proxy.proxyConfig.getCustomTracingListenerPorts(), mockHandlerFactory, mockWavefrontSender, new SpanSampler(new RateSampler(1.0D), proxy.proxyConfig.isTraceAlwaysSampleErrors())); waitUntilListenerIsOnline(customTracePort); reset(mockTraceHandler); reset(mockWavefrontSender); String traceId = UUID.randomUUID().toString(); String spanData = "testSpanName source=testsource spanId=testspanid " + "traceId=\"" + traceId + "\" " + startTime + " " + (startTime + 1) + "\n"; mockTraceHandler.report(Span.newBuilder().setCustomer("dummy").setStartMillis(startTime * 1000). setDuration(1000). setName("testSpanName"). setSource(PREPROCESSED_SOURCE_VALUE). setSpanId("testspanid"). setTraceId(traceId). setAnnotations(ImmutableList.of( new Annotation("application", PREPROCESSED_APPLICATION_TAG_VALUE), new Annotation("service", PREPROCESSED_SERVICE_TAG_VALUE), new Annotation("cluster", PREPROCESSED_CLUSTER_TAG_VALUE), new Annotation("shard", PREPROCESSED_SHARD_TAG_VALUE))).build()); expectLastCall(); Capture<HashMap<String, String>> tagsCapture = EasyMock.newCapture(); mockWavefrontSender.sendMetric(eq(HEART_BEAT_METRIC), eq(1.0), anyLong(), eq(PREPROCESSED_SOURCE_VALUE), EasyMock.capture(tagsCapture)); expectLastCall().anyTimes(); replay(mockTraceHandler, mockWavefrontSender); Socket socket = SocketFactory.getDefault().createSocket("localhost", customTracePort); BufferedOutputStream stream = new BufferedOutputStream(socket.getOutputStream()); stream.write(spanData.getBytes()); stream.flush(); socket.close(); Thread.sleep(100); verifyWithTimeout(500, mockTraceHandler, mockWavefrontSender); HashMap<String, String> tagsReturned = tagsCapture.getValue(); assertEquals(PREPROCESSED_APPLICATION_TAG_VALUE, tagsReturned.get(APPLICATION_TAG_KEY)); assertEquals(PREPROCESSED_SERVICE_TAG_VALUE, tagsReturned.get(SERVICE_TAG_KEY)); assertEquals(PREPROCESSED_CLUSTER_TAG_VALUE, tagsReturned.get(CLUSTER_TAG_KEY)); assertEquals(PREPROCESSED_SHARD_TAG_VALUE, tagsReturned.get(SHARD_TAG_KEY)); }
@Test public void testCustomTraceUnifiedPortHandlerPlaintext() throws Exception { customTracePort = findAvailablePort(50000); proxy.proxyConfig.customTracingListenerPorts = String.valueOf(customTracePort); proxy.startCustomTracingListener(proxy.proxyConfig.getCustomTracingListenerPorts(), mockHandlerFactory, mockWavefrontSender, new SpanSampler(new RateSampler(1.0D), proxy.proxyConfig.isTraceAlwaysSampleErrors())); waitUntilListenerIsOnline(customTracePort); reset(mockTraceHandler); reset(mockTraceSpanLogsHandler); reset(mockWavefrontSender); String traceId = UUID.randomUUID().toString(); long timestamp1 = startTime * 1000000 + 12345; long timestamp2 = startTime * 1000000 + 23456; String spanData = "testSpanName source=testsource spanId=testspanid " + "traceId=\"" + traceId + "\" application=application1 service=service1 " + startTime + " " + (startTime + 1) + "\n"; 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", "key4", "value4")). build() )). 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("application", "application1"), new Annotation("service", "service1"))).build()); expectLastCall(); Capture<HashMap<String, String>> tagsCapture = EasyMock.newCapture(); mockWavefrontSender.sendMetric(eq(HEART_BEAT_METRIC), eq(1.0), anyLong(), eq("testsource"), EasyMock.capture(tagsCapture)); EasyMock.expectLastCall().anyTimes(); replay(mockTraceHandler, mockTraceSpanLogsHandler, mockWavefrontSender); Socket socket = SocketFactory.getDefault().createSocket("localhost", customTracePort); BufferedOutputStream stream = new BufferedOutputStream(socket.getOutputStream()); String payloadStr = spanData + "{\"spanId\":\"testspanid\",\"traceId\":\"" + traceId + "\",\"logs\":[{\"timestamp\":" + timestamp1 + ",\"fields\":{\"key\":\"value\",\"key2\":\"value2\"}},{\"timestamp\":" + timestamp2 + ",\"fields\":{\"key3\":\"value3\",\"key4\":\"value4\"}}]}\n" + "{\"spanId\":\"testspanid\",\"traceId\":\"" + traceId + "\",\"logs\":[{\"timestamp\":" + timestamp1 + ",\"fields\":{\"key\":\"value\",\"key2\":\"value2\"}},{\"timestamp\":" + timestamp2 + ",\"fields\":{\"key3\":\"value3\",\"key4\":\"value4\"}}]," + "\"span\":\"" + escapeSpanData(spanData) + "\"}\n"; stream.write(payloadStr.getBytes()); stream.flush(); socket.close(); verifyWithTimeout(500, mockTraceHandler, mockTraceSpanLogsHandler, mockWavefrontSender); HashMap<String, String> tagsReturned = tagsCapture.getValue(); assertEquals("application1", tagsReturned.get(APPLICATION_TAG_KEY)); assertEquals("service1", tagsReturned.get(SERVICE_TAG_KEY)); assertEquals("none", tagsReturned.get(CLUSTER_TAG_KEY)); assertEquals("none", tagsReturned.get(SHARD_TAG_KEY)); }