PushAgent extends AbstractAgent { protected void startJsonListener(String strPort, ReportableEntityHandlerFactory handlerFactory) { final int port = Integer.parseInt(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); ChannelHandler channelHandler = new JsonMetricsPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, handlerFactory, proxyConfig.getPrefix(), proxyConfig.getHostname(), preprocessors.get(strPort)); startAsManagedThread(port, new TcpIngester(createInitializer(channelHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-plaintext-json-" + port); logger.info("listening on port: " + strPort + " for JSON metrics data"); }  static void main(String[] args); @Override void stopListeners();  }
@Test public void testJsonMetricsPortHandler() throws Exception { port = findAvailablePort(3878); proxy.proxyConfig.jsonListenerPorts = String.valueOf(port); proxy.startJsonListener(proxy.proxyConfig.jsonListenerPorts, mockHandlerFactory); waitUntilListenerIsOnline(port); reset(mockPointHandler); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric.test").setHost("testSource").setTimestamp(startTime * 1000). setAnnotations(ImmutableMap.of("env", "prod", "dc", "test1")).setValue(1.0d).build()); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric.test.cpu.usage.idle").setHost("testSource"). setTimestamp(startTime * 1000).setValue(99.0d).build()); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric.test.cpu.usage.user").setHost("testSource"). setTimestamp(startTime * 1000).setValue(0.5d).build()); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric.test.cpu.usage.system").setHost("testSource"). setTimestamp(startTime * 1000).setValue(0.7d).build()); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric.test.disk.free").setHost("testSource"). setTimestamp(startTime * 1000).setValue(0.0d).build()); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric.test.mem.used").setHost("testSource"). setTimestamp(startTime * 1000).setValue(50.0d).build()); replay(mockPointHandler); String payloadStr = "{\n" + " \"value\": 1.0,\n" + " \"tags\": {\n" + " \"dc\": \"test1\",\n" + " \"env\": \"prod\"\n" + " }\n" + "}\n"; String payloadStr2 = "{\n" + " \"cpu.usage\": {\n" + " \"idle\": 99.0,\n" + " \"user\": 0.5,\n" + " \"system\": 0.7\n" + " },\n" + " \"disk.free\": 0.0,\n" + " \"mem\": {\n" + " \"used\": 50.0\n" + " }\n" + "}\n"; assertEquals(200, gzippedHttpPost("http: "d=" + startTime * 1000, payloadStr)); assertEquals(200, gzippedHttpPost("http: "d=" + startTime * 1000, payloadStr2)); verify(mockPointHandler); }