diff --git a/apm-collector/apm-collector-agent/collector-agent-grpc-provider/src/main/java/org/skywalking/apm/collector/agent/grpc/handler/ApplicationRegisterServiceHandler.java b/apm-collector/apm-collector-agent/collector-agent-grpc-provider/src/main/java/org/skywalking/apm/collector/agent/grpc/handler/ApplicationRegisterServiceHandler.java index d6f6649a8..1008f22c8 100644 --- a/apm-collector/apm-collector-agent/collector-agent-grpc-provider/src/main/java/org/skywalking/apm/collector/agent/grpc/handler/ApplicationRegisterServiceHandler.java +++ b/apm-collector/apm-collector-agent/collector-agent-grpc-provider/src/main/java/org/skywalking/apm/collector/agent/grpc/handler/ApplicationRegisterServiceHandler.java @@ -22,8 +22,6 @@ import com.google.protobuf.ProtocolStringList; import io.grpc.stub.StreamObserver; import org.skywalking.apm.collector.agent.stream.worker.register.ApplicationIDService; import org.skywalking.apm.collector.core.module.ModuleManager; -import org.skywalking.apm.collector.core.module.ModuleNotFoundException; -import org.skywalking.apm.collector.core.module.ServiceNotProvidedException; import org.skywalking.apm.collector.server.grpc.GRPCHandler; import org.skywalking.apm.network.proto.Application; import org.skywalking.apm.network.proto.ApplicationMapping; @@ -52,12 +50,7 @@ public class ApplicationRegisterServiceHandler extends ApplicationRegisterServic ApplicationMapping.Builder builder = ApplicationMapping.newBuilder(); for (int i = 0; i < applicationCodes.size(); i++) { String applicationCode = applicationCodes.get(i); - int applicationId = 0; - try { - applicationId = applicationIDService.getOrCreate(applicationCode); - } catch (ModuleNotFoundException | ServiceNotProvidedException e) { - logger.error(e.getMessage(), e); - } + int applicationId = applicationIDService.getOrCreate(applicationCode); if (applicationId != 0) { KeyWithIntegerValue value = KeyWithIntegerValue.newBuilder().setKey(applicationCode).setValue(applicationId).build(); diff --git a/apm-collector/apm-collector-agent/collector-agent-grpc-provider/src/main/java/org/skywalking/apm/collector/agent/grpc/handler/InstanceDiscoveryServiceHandler.java b/apm-collector/apm-collector-agent/collector-agent-grpc-provider/src/main/java/org/skywalking/apm/collector/agent/grpc/handler/InstanceDiscoveryServiceHandler.java index 449afddd3..456c1cd26 100644 --- a/apm-collector/apm-collector-agent/collector-agent-grpc-provider/src/main/java/org/skywalking/apm/collector/agent/grpc/handler/InstanceDiscoveryServiceHandler.java +++ b/apm-collector/apm-collector-agent/collector-agent-grpc-provider/src/main/java/org/skywalking/apm/collector/agent/grpc/handler/InstanceDiscoveryServiceHandler.java @@ -23,8 +23,6 @@ import com.google.gson.JsonObject; import io.grpc.stub.StreamObserver; import org.skywalking.apm.collector.agent.stream.worker.register.InstanceIDService; import org.skywalking.apm.collector.core.module.ModuleManager; -import org.skywalking.apm.collector.core.module.ModuleNotFoundException; -import org.skywalking.apm.collector.core.module.ServiceNotProvidedException; import org.skywalking.apm.collector.core.util.TimeBucketUtils; import org.skywalking.apm.collector.server.grpc.GRPCHandler; import org.skywalking.apm.network.proto.ApplicationInstance; @@ -52,12 +50,7 @@ public class InstanceDiscoveryServiceHandler extends InstanceDiscoveryServiceGrp @Override public void register(ApplicationInstance request, StreamObserver responseObserver) { long timeBucket = TimeBucketUtils.INSTANCE.getSecondTimeBucket(request.getRegisterTime()); - int instanceId = 0; - try { - instanceId = instanceIDService.getOrCreate(request.getApplicationId(), request.getAgentUUID(), timeBucket, buildOsInfo(request.getOsinfo())); - } catch (ModuleNotFoundException | ServiceNotProvidedException e) { - logger.error(e.getMessage(), e); - } + int instanceId = instanceIDService.getOrCreate(request.getApplicationId(), request.getAgentUUID(), timeBucket, buildOsInfo(request.getOsinfo())); ApplicationInstanceMapping.Builder builder = ApplicationInstanceMapping.newBuilder(); builder.setApplicationId(request.getApplicationId()); builder.setApplicationInstanceId(instanceId); @@ -68,11 +61,7 @@ public class InstanceDiscoveryServiceHandler extends InstanceDiscoveryServiceGrp @Override public void registerRecover(ApplicationInstanceRecover request, StreamObserver responseObserver) { long timeBucket = TimeBucketUtils.INSTANCE.getSecondTimeBucket(request.getRegisterTime()); - try { - instanceIDService.recover(request.getApplicationInstanceId(), request.getApplicationId(), timeBucket, buildOsInfo(request.getOsinfo())); - } catch (ModuleNotFoundException | ServiceNotProvidedException e) { - logger.error(e.getMessage(), e); - } + instanceIDService.recover(request.getApplicationInstanceId(), request.getApplicationId(), timeBucket, buildOsInfo(request.getOsinfo())); responseObserver.onNext(Downstream.newBuilder().build()); responseObserver.onCompleted(); } diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/AgentModuleJettyProvider.java b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/AgentModuleJettyProvider.java index 68ac58805..75d911b44 100644 --- a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/AgentModuleJettyProvider.java +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/AgentModuleJettyProvider.java @@ -20,6 +20,9 @@ package org.skywalking.apm.collector.agent.jetty; import java.util.Properties; import org.skywalking.apm.collector.agent.AgentModule; +import org.skywalking.apm.collector.agent.jetty.handler.ApplicationRegisterServletHandler; +import org.skywalking.apm.collector.agent.jetty.handler.InstanceDiscoveryServletHandler; +import org.skywalking.apm.collector.agent.jetty.handler.ServiceNameDiscoveryServiceHandler; import org.skywalking.apm.collector.agent.jetty.handler.TraceSegmentServletHandler; import org.skywalking.apm.collector.agent.jetty.handler.naming.AgentJettyNamingHandler; import org.skywalking.apm.collector.agent.jetty.handler.naming.AgentJettyNamingListener; @@ -89,6 +92,9 @@ public class AgentModuleJettyProvider extends ModuleProvider { } private void addHandlers(Server jettyServer) { - jettyServer.addHandler(new TraceSegmentServletHandler()); + jettyServer.addHandler(new TraceSegmentServletHandler(getManager())); + jettyServer.addHandler(new ApplicationRegisterServletHandler(getManager())); + jettyServer.addHandler(new InstanceDiscoveryServletHandler(getManager())); + jettyServer.addHandler(new ServiceNameDiscoveryServiceHandler(getManager())); } } diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/handler/ApplicationRegisterServletHandler.java b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/handler/ApplicationRegisterServletHandler.java new file mode 100644 index 000000000..5fb5db4b6 --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/handler/ApplicationRegisterServletHandler.java @@ -0,0 +1,75 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.agent.jetty.handler; + +import com.google.gson.Gson; +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import java.io.IOException; +import javax.servlet.http.HttpServletRequest; +import org.skywalking.apm.collector.agent.stream.worker.register.ApplicationIDService; +import org.skywalking.apm.collector.core.module.ModuleManager; +import org.skywalking.apm.collector.server.jetty.ArgumentsParseException; +import org.skywalking.apm.collector.server.jetty.JettyHandler; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng + */ +public class ApplicationRegisterServletHandler extends JettyHandler { + + private final Logger logger = LoggerFactory.getLogger(ApplicationRegisterServletHandler.class); + + private final ApplicationIDService applicationIDService; + private Gson gson = new Gson(); + private static final String APPLICATION_CODE = "c"; + private static final String APPLICATION_ID = "i"; + + public ApplicationRegisterServletHandler(ModuleManager moduleManager) { + this.applicationIDService = new ApplicationIDService(moduleManager); + } + + @Override public String pathSpec() { + return "/application/register"; + } + + @Override protected JsonElement doGet(HttpServletRequest req) throws ArgumentsParseException { + throw new UnsupportedOperationException(); + } + + @Override protected JsonElement doPost(HttpServletRequest req) throws ArgumentsParseException { + JsonArray responseArray = new JsonArray(); + try { + JsonArray applicationCodes = gson.fromJson(req.getReader(), JsonArray.class); + for (int i = 0; i < applicationCodes.size(); i++) { + String applicationCode = applicationCodes.get(i).getAsString(); + int applicationId = applicationIDService.getOrCreate(applicationCode); + JsonObject mapping = new JsonObject(); + mapping.addProperty(APPLICATION_CODE, applicationCode); + mapping.addProperty(APPLICATION_ID, applicationId); + responseArray.add(mapping); + } + } catch (IOException e) { + logger.error(e.getMessage(), e); + } + return responseArray; + } +} diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/handler/InstanceDiscoveryServletHandler.java b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/handler/InstanceDiscoveryServletHandler.java new file mode 100644 index 000000000..658d7d419 --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/handler/InstanceDiscoveryServletHandler.java @@ -0,0 +1,78 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.agent.jetty.handler; + +import com.google.gson.Gson; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import java.io.IOException; +import javax.servlet.http.HttpServletRequest; +import org.skywalking.apm.collector.agent.stream.worker.register.InstanceIDService; +import org.skywalking.apm.collector.core.module.ModuleManager; +import org.skywalking.apm.collector.server.jetty.ArgumentsParseException; +import org.skywalking.apm.collector.server.jetty.JettyHandler; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng + */ +public class InstanceDiscoveryServletHandler extends JettyHandler { + + private final Logger logger = LoggerFactory.getLogger(InstanceDiscoveryServletHandler.class); + + private final InstanceIDService instanceIDService; + private Gson gson = new Gson(); + + private static final String APPLICATION_ID = "ai"; + private static final String AGENT_UUID = "au"; + private static final String REGISTER_TIME = "rt"; + private static final String INSTANCE_ID = "ii"; + private static final String OS_INFO = "oi"; + + public InstanceDiscoveryServletHandler(ModuleManager moduleManager) { + this.instanceIDService = new InstanceIDService(moduleManager); + } + + @Override public String pathSpec() { + return "/instance/register"; + } + + @Override protected JsonElement doGet(HttpServletRequest req) throws ArgumentsParseException { + throw new UnsupportedOperationException(); + } + + @Override protected JsonElement doPost(HttpServletRequest req) throws ArgumentsParseException { + JsonObject responseJson = new JsonObject(); + try { + JsonObject instance = gson.fromJson(req.getReader(), JsonObject.class); + int applicationId = instance.get(APPLICATION_ID).getAsInt(); + String agentUUID = instance.get(AGENT_UUID).getAsString(); + long registerTime = instance.get(REGISTER_TIME).getAsLong(); + JsonObject osInfo = instance.get(OS_INFO).getAsJsonObject(); + + int instanceId = instanceIDService.getOrCreate(applicationId, agentUUID, registerTime, osInfo.toString()); + responseJson.addProperty(APPLICATION_ID, applicationId); + responseJson.addProperty(INSTANCE_ID, instanceId); + } catch (IOException e) { + logger.error(e.getMessage(), e); + } + return responseJson; + } +} diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/handler/ServiceNameDiscoveryServiceHandler.java b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/handler/ServiceNameDiscoveryServiceHandler.java new file mode 100644 index 000000000..b842e17c9 --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/handler/ServiceNameDiscoveryServiceHandler.java @@ -0,0 +1,82 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.agent.jetty.handler; + +import com.google.gson.Gson; +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import java.io.IOException; +import javax.servlet.http.HttpServletRequest; +import org.skywalking.apm.collector.agent.stream.worker.register.ServiceNameService; +import org.skywalking.apm.collector.core.module.ModuleManager; +import org.skywalking.apm.collector.server.jetty.ArgumentsParseException; +import org.skywalking.apm.collector.server.jetty.JettyHandler; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng + */ +public class ServiceNameDiscoveryServiceHandler extends JettyHandler { + + private final Logger logger = LoggerFactory.getLogger(ServiceNameDiscoveryServiceHandler.class); + + private final ServiceNameService serviceNameService; + private Gson gson = new Gson(); + + private static final String APPLICATION_ID = "ai"; + private static final String SERVICE_NAME = "sn"; + private static final String SERVICE_ID = "si"; + private static final String ELEMENT = "el"; + + public ServiceNameDiscoveryServiceHandler(ModuleManager moduleManager) { + this.serviceNameService = new ServiceNameService(moduleManager); + } + + @Override public String pathSpec() { + return "/servicename/discovery"; + } + + @Override protected JsonElement doGet(HttpServletRequest req) throws ArgumentsParseException { + throw new UnsupportedOperationException(); + } + + @Override protected JsonElement doPost(HttpServletRequest req) throws ArgumentsParseException { + JsonArray responseArray = new JsonArray(); + try { + JsonArray services = gson.fromJson(req.getReader(), JsonArray.class); + for (JsonElement service : services) { + int applicationId = service.getAsJsonObject().get(APPLICATION_ID).getAsInt(); + String serviceName = service.getAsJsonObject().get(SERVICE_NAME).getAsString(); + + int serviceId = serviceNameService.getOrCreate(applicationId, serviceName); + if (serviceId != 0) { + JsonObject responseJson = new JsonObject(); + responseJson.addProperty(SERVICE_ID, serviceId); + responseJson.add(ELEMENT, service); + responseArray.add(responseJson); + } + } + } catch (IOException e) { + logger.error(e.getMessage(), e); + } + return responseArray; + } +} diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/handler/TraceSegmentServletHandler.java b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/handler/TraceSegmentServletHandler.java index dd3ac8ed2..6c873bd77 100644 --- a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/handler/TraceSegmentServletHandler.java +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/handler/TraceSegmentServletHandler.java @@ -26,6 +26,7 @@ import javax.servlet.http.HttpServletRequest; import org.skywalking.apm.collector.agent.jetty.handler.reader.TraceSegment; import org.skywalking.apm.collector.agent.jetty.handler.reader.TraceSegmentJsonReader; import org.skywalking.apm.collector.agent.stream.parser.SegmentParse; +import org.skywalking.apm.collector.core.module.ModuleManager; import org.skywalking.apm.collector.server.jetty.ArgumentsParseException; import org.skywalking.apm.collector.server.jetty.JettyHandler; import org.slf4j.Logger; @@ -38,6 +39,12 @@ public class TraceSegmentServletHandler extends JettyHandler { private final Logger logger = LoggerFactory.getLogger(TraceSegmentServletHandler.class); + private final ModuleManager moduleManager; + + public TraceSegmentServletHandler(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + @Override public String pathSpec() { return "/segments"; } @@ -64,7 +71,7 @@ public class TraceSegmentServletHandler extends JettyHandler { reader.beginArray(); while (reader.hasNext()) { - SegmentParse segmentParse = new SegmentParse(null); + SegmentParse segmentParse = new SegmentParse(moduleManager); TraceSegment traceSegment = jsonReader.read(reader); segmentParse.parse(traceSegment.getUpstreamSegment(), SegmentParse.Source.Agent); } diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/ApplicationRegisterPost.java b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/ApplicationRegisterPost.java new file mode 100644 index 000000000..cd5343285 --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/ApplicationRegisterPost.java @@ -0,0 +1,33 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.agent.jetty.handler; + +import com.google.gson.JsonElement; +import java.io.IOException; + +/** + * @author peng-yongsheng + */ +public class ApplicationRegisterPost { + + public void send(String jsonFile) throws IOException { + JsonElement application = JsonFileReader.INSTANCE.read(jsonFile); + HttpClientTools.INSTANCE.post("http://localhost:12800/application/register", application.toString()); + } +} diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/HttpClientTools.java b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/HttpClientTools.java new file mode 100644 index 000000000..e97b3857a --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/HttpClientTools.java @@ -0,0 +1,95 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.agent.jetty.handler; + +import java.io.IOException; +import java.net.URI; +import java.util.List; +import org.apache.http.Consts; +import org.apache.http.HttpEntity; +import org.apache.http.NameValuePair; +import org.apache.http.client.entity.UrlEncodedFormEntity; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.entity.StringEntity; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.util.EntityUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng + */ +public enum HttpClientTools { + INSTANCE; + + private final Logger logger = LoggerFactory.getLogger(HttpClientTools.class); + + public String get(String url, List params) throws IOException { + CloseableHttpClient httpClient = HttpClients.createDefault(); + try { + HttpGet httpget = new HttpGet(url); + String paramStr = EntityUtils.toString(new UrlEncodedFormEntity(params)); + httpget.setURI(new URI(httpget.getURI().toString() + "?" + paramStr)); + logger.debug("executing get request {}", httpget.getURI()); + + try (CloseableHttpResponse response = httpClient.execute(httpget)) { + HttpEntity entity = response.getEntity(); + if (entity != null) { + return EntityUtils.toString(entity); + } + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } finally { + try { + httpClient.close(); + } catch (IOException e) { + logger.error(e.getMessage(), e); + } + } + return null; + } + + public String post(String url, String data) throws IOException { + CloseableHttpClient httpClient = HttpClients.createDefault(); + try { + HttpPost httppost = new HttpPost(url); + httppost.setEntity(new StringEntity(data, Consts.UTF_8)); + logger.debug("executing post request {}", httppost.getURI()); + try (CloseableHttpResponse response = httpClient.execute(httppost)) { + HttpEntity entity = response.getEntity(); + if (entity != null) { + return EntityUtils.toString(entity); + } + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } finally { + try { + httpClient.close(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + return null; + } +} diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/InstanceRegisterPost.java b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/InstanceRegisterPost.java new file mode 100644 index 000000000..12ef75dda --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/InstanceRegisterPost.java @@ -0,0 +1,33 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.agent.jetty.handler; + +import com.google.gson.JsonElement; +import java.io.IOException; + +/** + * @author peng-yongsheng + */ +public class InstanceRegisterPost { + + public void send(String jsonFile) throws IOException { + JsonElement instance = JsonFileReader.INSTANCE.read(jsonFile); + HttpClientTools.INSTANCE.post("http://localhost:12800/instance/register", instance.toString()); + } +} diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/JsonFileReader.java b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/JsonFileReader.java new file mode 100644 index 000000000..0353c8214 --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/JsonFileReader.java @@ -0,0 +1,42 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.agent.jetty.handler; + +import com.google.gson.JsonElement; +import com.google.gson.JsonParser; +import java.io.FileNotFoundException; +import java.io.FileReader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng + */ +public enum JsonFileReader { + INSTANCE; + + private final Logger logger = LoggerFactory.getLogger(JsonFileReader.class); + + public JsonElement read(String fileName) throws FileNotFoundException { + String path = this.getClass().getClassLoader().getResource(fileName).getFile(); + logger.debug("path: {}", path); + JsonParser jsonParser = new JsonParser(); + return jsonParser.parse(new FileReader(path)); + } +} diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/SegmentPost.java b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/SegmentPost.java new file mode 100644 index 000000000..29b3efc31 --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/SegmentPost.java @@ -0,0 +1,47 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.agent.jetty.handler; + +import com.google.gson.JsonElement; +import java.io.IOException; + +/** + * @author peng-yongsheng + */ +public class SegmentPost { + + public static void main(String[] args) throws IOException { + ApplicationRegisterPost applicationRegisterPost = new ApplicationRegisterPost(); + applicationRegisterPost.send("json/application-register-consumer.json"); + applicationRegisterPost.send("json/application-register-provider.json"); + + InstanceRegisterPost instanceRegisterPost = new InstanceRegisterPost(); + instanceRegisterPost.send("json/instance-register-consumer.json"); + instanceRegisterPost.send("json/instance-register-provider.json"); + + ServiceNameRegisterPost serviceNameRegisterPost = new ServiceNameRegisterPost(); + serviceNameRegisterPost.send("json/servicename-register-consumer.json"); + serviceNameRegisterPost.send("json/servicename-register-provider.json"); + + JsonElement provider = JsonFileReader.INSTANCE.read("json/dubbox-provider.json"); + HttpClientTools.INSTANCE.post("http://localhost:12800/segments", provider.toString()); + JsonElement consumer = JsonFileReader.INSTANCE.read("json/dubbox-consumer.json"); + HttpClientTools.INSTANCE.post("http://localhost:12800/segments", consumer.toString()); + } +} diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/ServiceNameRegisterPost.java b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/ServiceNameRegisterPost.java new file mode 100644 index 000000000..6e6cde68d --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/ServiceNameRegisterPost.java @@ -0,0 +1,33 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.agent.jetty.handler; + +import com.google.gson.JsonElement; +import java.io.IOException; + +/** + * @author peng-yongsheng + */ +public class ServiceNameRegisterPost { + + public void send(String jsonFile) throws IOException { + JsonElement instance = JsonFileReader.INSTANCE.read(jsonFile); + HttpClientTools.INSTANCE.post("http://localhost:12800/servicename/discovery", instance.toString()); + } +} diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/application-register-consumer.json b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/application-register-consumer.json new file mode 100644 index 000000000..b7f243635 --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/application-register-consumer.json @@ -0,0 +1,3 @@ +[ + "dubbox-consumer" +] \ No newline at end of file diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/application-register-provider.json b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/application-register-provider.json new file mode 100644 index 000000000..08a2c156b --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/application-register-provider.json @@ -0,0 +1,3 @@ +[ + "dubbox-provider" +] \ No newline at end of file diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/dubbox-consumer.json b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/dubbox-consumer.json new file mode 100644 index 000000000..453dce3eb --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/dubbox-consumer.json @@ -0,0 +1,71 @@ +[ + { + "gt": [ + [ + 230150, + 185809, + 24040000 + ] + ], + "sg": { + "ts": [ + 230150, + 185809, + 24040000 + ], + "ai": -1, + "ii": 1, + "rs": [], + "ss": [ + { + "si": 1, + "tv": 1, + "lv": 1, + "ps": 0, + "st": 1501858094526, + "et": 1501858097004, + "ci": 3, + "cn": "", + "oi": 0, + "on": "org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()", + "pi": 0, + "pn": "172.25.0.4:20880", + "ie": false, + "to": [ + { + "k": "url", + "v": "rest://172.25.0.4:20880/org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()" + } + ], + "lo": [] + }, + { + "si": 0, + "tv": 0, + "lv": 2, + "ps": -1, + "st": 1501858092409, + "et": 1501858097033, + "ci": 1, + "cn": "", + "oi": 0, + "on": "/dubbox-case/case/dubbox-rest", + "pi": 0, + "pn": "", + "ie": false, + "to": [ + { + "k": "url", + "v": "http://localhost:18080/dubbox-case/case/dubbox-rest" + }, + { + "k": "http.method", + "v": "GET" + } + ], + "lo": [] + } + ] + } + } +] \ No newline at end of file diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/dubbox-provider.json b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/dubbox-provider.json new file mode 100644 index 000000000..709f57760 --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/dubbox-provider.json @@ -0,0 +1,67 @@ +[ + { + "gt": [ + [ + 230150, + 185809, + 24040000 + ] + ], + "sg": { + "ts": [ + 137150, + 185809, + 48780000 + ], + "ai": 2, + "ii": 2, + "rs": [ + { + "ts": [ + 230150, + 185809, + 24040000 + ], + "ai": -1, + "si": 1, + "vi": 0, + "vn": "/dubbox-case/case/dubbox-rest", + "ni": 0, + "nn": "172.25.0.4:20880", + "ea": 2, + "ei": 0, + "en": "/dubbox-case/case/dubbox-rest", + "rn": 0 + } + ], + "ss": [ + { + "si": 0, + "tv": 0, + "lv": 2, + "ps": -1, + "st": 1501858094726, + "et": 1501858096804, + "ci": 3, + "cn": "", + "oi": 0, + "on": "org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()", + "pi": 0, + "pn": "", + "ie": false, + "to": [ + { + "k": "url", + "v": "rest://172.25.0.4:20880/org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()" + }, + { + "k": "http.method", + "v": "GET" + } + ], + "lo": [] + } + ] + } + } +] \ No newline at end of file diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/instance-register-consumer.json b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/instance-register-consumer.json new file mode 100644 index 000000000..139dc5743 --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/instance-register-consumer.json @@ -0,0 +1,9 @@ +{ + "ai": -1, + "au": "dubbox-consumer", + "rt": 1501858094526, + "oi": { + "any_name": "any_value", + "any_name1": "any_value1" + } +} \ No newline at end of file diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/instance-register-provider.json b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/instance-register-provider.json new file mode 100644 index 000000000..976928f9c --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/instance-register-provider.json @@ -0,0 +1,9 @@ +{ + "ai": 2, + "au": "dubbox-provider", + "rt": 1501858094526, + "oi": { + "any_name": "any_value", + "any_name1": "any_value1" + } +} \ No newline at end of file diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/servicename-register-consumer.json b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/servicename-register-consumer.json new file mode 100644 index 000000000..9cc919204 --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/servicename-register-consumer.json @@ -0,0 +1,6 @@ +[ + { + "ai": -1, + "sn": "org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()" + } +] \ No newline at end of file diff --git a/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/servicename-register-provider.json b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/servicename-register-provider.json new file mode 100644 index 000000000..0e9a96153 --- /dev/null +++ b/apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/test/resources/json/servicename-register-provider.json @@ -0,0 +1,6 @@ +[ + { + "ai": 2, + "sn": "org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()" + } +] \ No newline at end of file diff --git a/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferManager.java b/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferManager.java index 938d0e980..2e2a3d86c 100644 --- a/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferManager.java +++ b/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferManager.java @@ -21,6 +21,7 @@ package org.skywalking.apm.collector.agent.stream.buffer; import java.io.File; import java.io.FileOutputStream; import java.io.IOException; +import org.skywalking.apm.collector.core.module.ModuleManager; import org.skywalking.apm.collector.core.util.Const; import org.skywalking.apm.collector.core.util.StringUtils; import org.skywalking.apm.collector.core.util.TimeBucketUtils; @@ -39,7 +40,7 @@ public enum SegmentBufferManager { public static final String DATA_FILE_PREFIX = "data"; private FileOutputStream outputStream; - public synchronized void initialize() { + public synchronized void initialize(ModuleManager moduleManager) { logger.info("segment buffer initialize"); try { OffsetManager.INSTANCE.initialize(); @@ -58,7 +59,7 @@ public enum SegmentBufferManager { newDataFile(); } } - SegmentBufferReader.INSTANCE.initialize(); + SegmentBufferReader.INSTANCE.initialize(moduleManager); } catch (IOException e) { logger.error(e.getMessage(), e); } diff --git a/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferReader.java b/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferReader.java index a073b212b..148b33d9b 100644 --- a/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferReader.java +++ b/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferReader.java @@ -27,6 +27,7 @@ import java.io.InputStream; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import org.skywalking.apm.collector.agent.stream.parser.SegmentParse; +import org.skywalking.apm.collector.core.module.ModuleManager; import org.skywalking.apm.collector.core.util.CollectionUtils; import org.skywalking.apm.collector.core.util.Const; import org.skywalking.apm.collector.core.util.StringUtils; @@ -42,8 +43,10 @@ public enum SegmentBufferReader { private final Logger logger = LoggerFactory.getLogger(SegmentBufferReader.class); private InputStream inputStream; + private ModuleManager moduleManager; - public void initialize() { + public void initialize(ModuleManager moduleManager) { + this.moduleManager = moduleManager; Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(this::preRead, 3, 3, TimeUnit.SECONDS); } @@ -117,7 +120,7 @@ public enum SegmentBufferReader { while (readFile.length() > readFileOffset && readFileOffset < endPoint) { UpstreamSegment upstreamSegment = UpstreamSegment.parser().parseDelimitedFrom(inputStream); - SegmentParse parse = new SegmentParse(null); + SegmentParse parse = new SegmentParse(moduleManager); if (!parse.parse(upstreamSegment, SegmentParse.Source.Buffer)) { return false; } diff --git a/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/parser/standardization/ReferenceIdExchanger.java b/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/parser/standardization/ReferenceIdExchanger.java index 9b6e5aa55..e0ea1401a 100644 --- a/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/parser/standardization/ReferenceIdExchanger.java +++ b/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/parser/standardization/ReferenceIdExchanger.java @@ -18,9 +18,9 @@ package org.skywalking.apm.collector.agent.stream.parser.standardization; +import org.skywalking.apm.collector.agent.stream.worker.register.ApplicationIDService; import org.skywalking.apm.collector.agent.stream.worker.register.ServiceNameService; import org.skywalking.apm.collector.cache.CacheModule; -import org.skywalking.apm.collector.cache.service.ApplicationCacheService; import org.skywalking.apm.collector.cache.service.InstanceCacheService; import org.skywalking.apm.collector.core.module.ModuleManager; import org.skywalking.apm.collector.core.util.Const; @@ -36,9 +36,9 @@ public class ReferenceIdExchanger implements IdExchanger { private final Logger logger = LoggerFactory.getLogger(ReferenceIdExchanger.class); private static ReferenceIdExchanger EXCHANGER; - private ServiceNameService serviceNameService; + private final ServiceNameService serviceNameService; + private final ApplicationIDService applicationIDService; private final InstanceCacheService instanceCacheService; - private final ApplicationCacheService applicationCacheService; public static ReferenceIdExchanger getInstance(ModuleManager moduleManager) { if (EXCHANGER == null) { @@ -48,9 +48,9 @@ public class ReferenceIdExchanger implements IdExchanger { } private ReferenceIdExchanger(ModuleManager moduleManager) { + applicationIDService = new ApplicationIDService(moduleManager); serviceNameService = new ServiceNameService(moduleManager); instanceCacheService = moduleManager.find(CacheModule.NAME).getService(InstanceCacheService.class); - applicationCacheService = moduleManager.find(CacheModule.NAME).getService(ApplicationCacheService.class); } @Override public boolean exchange(ReferenceDecorator standardBuilder, int applicationId) { @@ -58,6 +58,10 @@ public class ReferenceIdExchanger implements IdExchanger { int entryServiceId = serviceNameService.getOrCreate(instanceCacheService.get(standardBuilder.getEntryApplicationInstanceId()), standardBuilder.getEntryServiceName()); if (entryServiceId == 0) { + if (logger.isDebugEnabled()) { + int entryApplicationId = instanceCacheService.get(standardBuilder.getEntryApplicationInstanceId()); + logger.debug("entry service name: {} from application id: {} exchange failed", standardBuilder.getEntryServiceName(), entryApplicationId); + } return false; } else { standardBuilder.toBuilder(); @@ -70,6 +74,10 @@ public class ReferenceIdExchanger implements IdExchanger { int parentServiceId = serviceNameService.getOrCreate(instanceCacheService.get(standardBuilder.getParentApplicationInstanceId()), standardBuilder.getParentServiceName()); if (parentServiceId == 0) { + if (logger.isDebugEnabled()) { + int parentApplicationId = instanceCacheService.get(standardBuilder.getParentApplicationInstanceId()); + logger.debug("parent service name: {} from application id: {} exchange failed", standardBuilder.getParentServiceName(), parentApplicationId); + } return false; } else { standardBuilder.toBuilder(); @@ -79,8 +87,11 @@ public class ReferenceIdExchanger implements IdExchanger { } if (standardBuilder.getNetworkAddressId() == 0 && StringUtils.isNotEmpty(standardBuilder.getNetworkAddress())) { - int networkAddressId = applicationCacheService.get(standardBuilder.getNetworkAddress()); + int networkAddressId = applicationIDService.getOrCreate(standardBuilder.getNetworkAddress()); if (networkAddressId == 0) { + if (logger.isDebugEnabled()) { + logger.debug("network address: {} from application id: {} exchange failed", standardBuilder.getNetworkAddress(), applicationId); + } return false; } else { standardBuilder.toBuilder(); diff --git a/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/parser/standardization/SegmentStandardizationWorker.java b/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/parser/standardization/SegmentStandardizationWorker.java index 8fe297ad2..0383d9327 100644 --- a/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/parser/standardization/SegmentStandardizationWorker.java +++ b/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/parser/standardization/SegmentStandardizationWorker.java @@ -38,7 +38,7 @@ public class SegmentStandardizationWorker extends AbstractLocalAsyncWorker { private final Logger logger = LoggerFactory.getLogger(SpanIdExchanger.class); private static SpanIdExchanger EXCHANGER; + private final ApplicationIDService applicationIDService; private final ServiceNameService serviceNameService; - private final ApplicationCacheService applicationCacheService; public static SpanIdExchanger getInstance(ModuleManager moduleManager) { if (EXCHANGER == null) { @@ -45,15 +44,16 @@ public class SpanIdExchanger implements IdExchanger { return EXCHANGER; } - public SpanIdExchanger(ModuleManager moduleManager) { - this.applicationCacheService = moduleManager.find(CacheModule.NAME).getService(ApplicationCacheService.class); + private SpanIdExchanger(ModuleManager moduleManager) { + this.applicationIDService = new ApplicationIDService(moduleManager); this.serviceNameService = new ServiceNameService(moduleManager); } @Override public boolean exchange(SpanDecorator standardBuilder, int applicationId) { if (standardBuilder.getPeerId() == 0 && StringUtils.isNotEmpty(standardBuilder.getPeer())) { - int peerId = applicationCacheService.get(standardBuilder.getPeer()); + int peerId = applicationIDService.getOrCreate(standardBuilder.getPeer()); if (peerId == 0) { + logger.debug("peer: {} in application: {} exchange failed", standardBuilder.getPeer(), applicationId); return false; } else { standardBuilder.toBuilder(); @@ -66,6 +66,7 @@ public class SpanIdExchanger implements IdExchanger { int operationNameId = serviceNameService.getOrCreate(applicationId, standardBuilder.getOperationName()); if (operationNameId == 0) { + logger.debug("service name: {} from application id: {} exchange failed", standardBuilder.getOperationName(), applicationId); return false; } else { standardBuilder.toBuilder(); diff --git a/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/ApplicationIDService.java b/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/ApplicationIDService.java index bdd80b658..1595a8cf0 100644 --- a/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/ApplicationIDService.java +++ b/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/ApplicationIDService.java @@ -24,8 +24,6 @@ import org.skywalking.apm.collector.cache.service.ApplicationCacheService; import org.skywalking.apm.collector.core.graph.Graph; import org.skywalking.apm.collector.core.graph.GraphManager; import org.skywalking.apm.collector.core.module.ModuleManager; -import org.skywalking.apm.collector.core.module.ModuleNotFoundException; -import org.skywalking.apm.collector.core.module.ServiceNotProvidedException; import org.skywalking.apm.collector.storage.table.register.Application; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -45,7 +43,7 @@ public class ApplicationIDService { this.applicationRegisterGraph = GraphManager.INSTANCE.createIfAbsent(RegisterStreamGraph.APPLICATION_REGISTER_GRAPH_ID, Application.class); } - public int getOrCreate(String applicationCode) throws ModuleNotFoundException, ServiceNotProvidedException { + public int getOrCreate(String applicationCode) { ApplicationCacheService service = moduleManager.find(CacheModule.NAME).getService(ApplicationCacheService.class); int applicationId = service.get(applicationCode); diff --git a/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/InstanceIDService.java b/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/InstanceIDService.java index efa5e5956..831b9dcbf 100644 --- a/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/InstanceIDService.java +++ b/apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/InstanceIDService.java @@ -24,8 +24,6 @@ import org.skywalking.apm.collector.cache.service.InstanceCacheService; import org.skywalking.apm.collector.core.graph.Graph; import org.skywalking.apm.collector.core.graph.GraphManager; import org.skywalking.apm.collector.core.module.ModuleManager; -import org.skywalking.apm.collector.core.module.ModuleNotFoundException; -import org.skywalking.apm.collector.core.module.ServiceNotProvidedException; import org.skywalking.apm.collector.storage.StorageModule; import org.skywalking.apm.collector.storage.dao.IInstanceRegisterDAO; import org.skywalking.apm.collector.storage.table.register.Instance; @@ -49,8 +47,7 @@ public class InstanceIDService { this.instanceRegisterDAO = moduleManager.find(StorageModule.NAME).getService(IInstanceRegisterDAO.class); } - public int getOrCreate(int applicationId, String agentUUID, long registerTime, - String osInfo) throws ModuleNotFoundException, ServiceNotProvidedException { + public int getOrCreate(int applicationId, String agentUUID, long registerTime, String osInfo) { logger.debug("get or create instance id, application id: {}, agentUUID: {}, registerTime: {}, osInfo: {}", applicationId, agentUUID, registerTime, osInfo); InstanceCacheService service = moduleManager.find(CacheModule.NAME).getService(InstanceCacheService.class); int instanceId = service.getInstanceId(applicationId, agentUUID); @@ -69,8 +66,7 @@ public class InstanceIDService { return instanceId; } - public void recover(int instanceId, int applicationId, long registerTime, - String osInfo) throws ModuleNotFoundException, ServiceNotProvidedException { + public void recover(int instanceId, int applicationId, long registerTime, String osInfo) { logger.debug("instance recover, instance id: {}, application id: {}, register time: {}", instanceId, applicationId, registerTime); Instance instance = new Instance(String.valueOf(instanceId)); diff --git a/apm-collector/apm-collector-boot/src/main/resources/application.yml b/apm-collector/apm-collector-boot/src/main/resources/application.yml index 275e975bb..254f84d23 100644 --- a/apm-collector/apm-collector-boot/src/main/resources/application.yml +++ b/apm-collector/apm-collector-boot/src/main/resources/application.yml @@ -23,10 +23,10 @@ ui: host: localhost port: 12800 context_path: / -#storage: -# elasticsearch: -# cluster_name: CollectorDBCluster -# cluster_transport_sniffer: true -# cluster_nodes: localhost:9300 -# index_shards_number: 2 -# index_replicas_number: 0 \ No newline at end of file +storage: + elasticsearch: + cluster_name: CollectorDBCluster + cluster_transport_sniffer: true + cluster_nodes: localhost:9300 + index_shards_number: 2 + index_replicas_number: 0 \ No newline at end of file diff --git a/apm-collector/apm-collector-cluster/collector-cluster-standalone-provider/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterModuleStandaloneProvider.java b/apm-collector/apm-collector-cluster/collector-cluster-standalone-provider/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterModuleStandaloneProvider.java index 8788acc6a..3778504f3 100644 --- a/apm-collector/apm-collector-cluster/collector-cluster-standalone-provider/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterModuleStandaloneProvider.java +++ b/apm-collector/apm-collector-cluster/collector-cluster-standalone-provider/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterModuleStandaloneProvider.java @@ -26,6 +26,8 @@ import org.skywalking.apm.collector.cluster.service.ModuleListenerService; import org.skywalking.apm.collector.cluster.service.ModuleRegisterService; import org.skywalking.apm.collector.cluster.standalone.service.StandaloneModuleListenerService; import org.skywalking.apm.collector.cluster.standalone.service.StandaloneModuleRegisterService; +import org.skywalking.apm.collector.core.CollectorException; +import org.skywalking.apm.collector.core.UnexpectedException; import org.skywalking.apm.collector.core.module.Module; import org.skywalking.apm.collector.core.module.ModuleProvider; import org.skywalking.apm.collector.core.module.ServiceNotProvidedException; @@ -75,7 +77,11 @@ public class ClusterModuleStandaloneProvider extends ModuleProvider { } @Override public void notifyAfterCompleted() throws ServiceNotProvidedException { - + try { + dataMonitor.start(); + } catch (CollectorException e) { + throw new UnexpectedException(e.getMessage()); + } } @Override public String[] requiredModules() {