PushAgent extends AbstractAgent { protected void startOpenTsdbListener(final String strPort, ReportableEntityHandlerFactory handlerFactory) { int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); ReportableEntityDecoder<String, ReportPoint> openTSDBDecoder = new ReportPointDecoderWrapper( new OpenTSDBDecoder("unknown", proxyConfig.getCustomSourceTags())); ChannelHandler channelHandler = new OpenTSDBPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, openTSDBDecoder, handlerFactory, preprocessors.get(strPort), hostnameResolver); startAsManagedThread(port, new TcpIngester(createInitializer(channelHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-plaintext-opentsdb-" + port); logger.info("listening on port: " + strPort + " for OpenTSDB metrics"); }  static void main(String[] args); @Override void stopListeners();  }
@Test public void testOpenTSDBPortHandler() throws Exception { port = findAvailablePort(4242); proxy.proxyConfig.opentsdbPorts = String.valueOf(port); proxy.startOpenTsdbListener(proxy.proxyConfig.getOpentsdbPorts(), mockHandlerFactory); waitUntilListenerIsOnline(port); reset(mockPointHandler); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric4.test").setHost("test1").setTimestamp(startTime * 1000). setAnnotations(ImmutableMap.of("env", "prod")).setValue(0.0d).build()); expectLastCall().times(2); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric4.test").setHost("test2").setTimestamp((startTime + 1) * 1000). setAnnotations(ImmutableMap.of("env", "prod")).setValue(1.0d).build()); expectLastCall().times(2); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("metric4.test").setHost("test3").setTimestamp((startTime + 2) * 1000). setAnnotations(ImmutableMap.of("env", "prod")).setValue(2.0d).build()); expectLastCall().times(2); mockPointHandler.reject((ReportPoint) eq(null), anyString()); expectLastCall().once(); replay(mockPointHandler); String payloadStr = "[\n" + " {\n" + " \"metric\": \"metric4.test\",\n" + " \"timestamp\": " + startTime + ",\n" + " \"value\": 0.0,\n" + " \"tags\": {\n" + " \"host\": \"test1\",\n" + " \"env\": \"prod\"\n" + " }\n" + " },\n" + " {\n" + " \"metric\": \"metric4.test\",\n" + " \"timestamp\": " + (startTime + 1) + ",\n" + " \"value\": 1.0,\n" + " \"tags\": {\n" + " \"host\": \"test2\",\n" + " \"env\": \"prod\"\n" + " }\n" + " }\n" + "]\n"; String payloadStr2 = "[\n" + " {\n" + " \"metric\": \"metric4.test\",\n" + " \"timestamp\": " + (startTime + 2) + ",\n" + " \"value\": 2.0,\n" + " \"tags\": {\n" + " \"host\": \"test3\",\n" + " \"env\": \"prod\"\n" + " }\n" + " },\n" + " {\n" + " \"metric\": \"metric4.test\",\n" + " \"timestamp\": " + startTime + ",\n" + " \"tags\": {\n" + " \"host\": \"test4\",\n" + " \"env\": \"prod\"\n" + " }\n" + " }]\n"; Socket socket = SocketFactory.getDefault().createSocket("localhost", port); BufferedOutputStream stream = new BufferedOutputStream(socket.getOutputStream()); String points = "version\n" + "put metric4.test " + startTime + " 0 host=test1 env=prod\n" + "put metric4.test " + (startTime + 1) + " 1 host=test2 env=prod\n" + "put metric4.test " + (startTime + 2) + " 2 host=test3 env=prod\n"; stream.write(points.getBytes()); stream.flush(); socket.close(); assertEquals(400, gzippedHttpPost("http: assertEquals(200, gzippedHttpPost("http: assertEquals(400, gzippedHttpPost("http: assertEquals(204, gzippedHttpPost("http: assertEquals(400, gzippedHttpPost("http: verify(mockPointHandler); }