diff --git a/apm-collector/apm-collector-agent-jetty/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/SegmentPost.java b/apm-collector/apm-collector-agent-jetty/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/SegmentPost.java index 29b3efc31..297eab2a2 100644 --- a/apm-collector/apm-collector-agent-jetty/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/SegmentPost.java +++ b/apm-collector/apm-collector-agent-jetty/collector-agent-jetty-provider/src/test/java/org/skywalking/apm/collector/agent/jetty/handler/SegmentPost.java @@ -40,8 +40,11 @@ public class SegmentPost { 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()); + + for (int i = 0; i < 1000; i++) { + HttpClientTools.INSTANCE.post("http://localhost:12800/segments", provider.toString()); + HttpClientTools.INSTANCE.post("http://localhost:12800/segments", consumer.toString()); + } } } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/AgentStreamModuleProvider.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/AgentStreamModuleProvider.java index 34444b476..c8c0d6e78 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/AgentStreamModuleProvider.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/AgentStreamModuleProvider.java @@ -29,6 +29,7 @@ import org.skywalking.apm.collector.agent.stream.service.register.IApplicationID import org.skywalking.apm.collector.agent.stream.service.register.IInstanceIDService; import org.skywalking.apm.collector.agent.stream.service.register.IServiceNameService; import org.skywalking.apm.collector.agent.stream.service.trace.ITraceSegmentService; +import org.skywalking.apm.collector.agent.stream.worker.AgentStreamRemoteDataRegister; import org.skywalking.apm.collector.agent.stream.worker.jvm.CpuMetricService; import org.skywalking.apm.collector.agent.stream.worker.jvm.GCMetricService; import org.skywalking.apm.collector.agent.stream.worker.jvm.InstanceHeartBeatService; @@ -42,6 +43,8 @@ import org.skywalking.apm.collector.cache.CacheModule; import org.skywalking.apm.collector.core.module.Module; import org.skywalking.apm.collector.core.module.ModuleProvider; import org.skywalking.apm.collector.core.module.ServiceNotProvidedException; +import org.skywalking.apm.collector.remote.RemoteModule; +import org.skywalking.apm.collector.remote.service.RemoteDataRegisterService; import org.skywalking.apm.collector.storage.StorageModule; /** @@ -75,6 +78,10 @@ public class AgentStreamModuleProvider extends ModuleProvider { } @Override public void start(Properties config) throws ServiceNotProvidedException { + RemoteDataRegisterService remoteDataRegisterService = getManager().find(RemoteModule.NAME).getService(RemoteDataRegisterService.class); + AgentStreamRemoteDataRegister agentStreamRemoteDataRegister = new AgentStreamRemoteDataRegister(remoteDataRegisterService); + agentStreamRemoteDataRegister.register(); + AgentStreamBootStartup bootStartup = new AgentStreamBootStartup(getManager()); bootStartup.start(); } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/BufferFileConfig.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/BufferFileConfig.java index 1b393918e..812048b72 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/BufferFileConfig.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/BufferFileConfig.java @@ -26,13 +26,19 @@ import java.util.Properties; public class BufferFileConfig { static int BUFFER_OFFSET_MAX_FILE_SIZE = 10 * 1024 * 1024; static int BUFFER_SEGMENT_MAX_FILE_SIZE = 10 * 1024 * 1024; + static String BUFFER_PATH = "../buffer/"; + private static final String BUFFER_PATH_KEY = "buffer_file_path"; private static final String BUFFER_OFFSET_MAX_FILE_SIZE_KEY = "buffer_offset_max_file_size"; private static final String BUFFER_SEGMENT_MAX_FILE_SIZE_KEY = "buffer_segment_max_file_size"; public static class Parser { public void parse(Properties config) { + if (config.containsKey(BUFFER_PATH_KEY)) { + BUFFER_PATH = config.getProperty(BUFFER_PATH_KEY); + } + if (config.containsKey(BUFFER_OFFSET_MAX_FILE_SIZE_KEY)) { String sizeStr = config.getProperty(BUFFER_OFFSET_MAX_FILE_SIZE_KEY).toUpperCase(); if (sizeStr.endsWith("K")) { diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/OffsetManager.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/OffsetManager.java index e9b4e25d6..a78aa1ac4 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/OffsetManager.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/OffsetManager.java @@ -49,7 +49,7 @@ public enum OffsetManager { public synchronized void initialize() throws IOException { if (!initialized) { this.offset = new Offset(); - File dataPath = new File(SegmentBufferConfig.BUFFER_PATH); + File dataPath = new File(BufferFileConfig.BUFFER_PATH); if (dataPath.mkdirs()) { createOffsetFile(); } else { @@ -77,7 +77,7 @@ public enum OffsetManager { private void createOffsetFile() throws IOException { String timeBucket = String.valueOf(TimeBucketUtils.INSTANCE.getSecondTimeBucket(System.currentTimeMillis())); String offsetFileName = OFFSET_FILE_PREFIX + "_" + timeBucket + "." + Const.FILE_SUFFIX; - offsetFile = new File(SegmentBufferConfig.BUFFER_PATH + offsetFileName); + offsetFile = new File(BufferFileConfig.BUFFER_PATH + offsetFileName); this.offset.getWriteOffset().setWriteFileName(Const.EMPTY_STRING); this.offset.getWriteOffset().setWriteFileOffset(0); this.offset.getReadOffset().setReadFileName(Const.EMPTY_STRING); @@ -99,7 +99,7 @@ public enum OffsetManager { private void nextFile() { String timeBucket = String.valueOf(TimeBucketUtils.INSTANCE.getSecondTimeBucket(System.currentTimeMillis())); String offsetFileName = OFFSET_FILE_PREFIX + "_" + timeBucket + "." + Const.FILE_SUFFIX; - File newOffsetFile = new File(SegmentBufferConfig.BUFFER_PATH + offsetFileName); + File newOffsetFile = new File(BufferFileConfig.BUFFER_PATH + offsetFileName); offsetFile.delete(); offsetFile = newOffsetFile; this.flush(); diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferManager.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferManager.java index 2e2a3d86c..2d3a87807 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferManager.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferManager.java @@ -44,14 +44,14 @@ public enum SegmentBufferManager { logger.info("segment buffer initialize"); try { OffsetManager.INSTANCE.initialize(); - if (new File(SegmentBufferConfig.BUFFER_PATH).mkdirs()) { + if (new File(BufferFileConfig.BUFFER_PATH).mkdirs()) { newDataFile(); } else { String writeFileName = OffsetManager.INSTANCE.getWriteFileName(); if (StringUtils.isNotEmpty(writeFileName)) { - File dataFile = new File(SegmentBufferConfig.BUFFER_PATH + writeFileName); + File dataFile = new File(BufferFileConfig.BUFFER_PATH + writeFileName); if (dataFile.exists()) { - outputStream = new FileOutputStream(new File(SegmentBufferConfig.BUFFER_PATH + writeFileName), true); + outputStream = new FileOutputStream(new File(BufferFileConfig.BUFFER_PATH + writeFileName), true); } else { newDataFile(); } @@ -83,7 +83,7 @@ public enum SegmentBufferManager { logger.debug("create new segment buffer file"); String timeBucket = String.valueOf(TimeBucketUtils.INSTANCE.getSecondTimeBucket(System.currentTimeMillis())); String writeFileName = DATA_FILE_PREFIX + "_" + timeBucket + "." + Const.FILE_SUFFIX; - File dataFile = new File(SegmentBufferConfig.BUFFER_PATH + writeFileName); + File dataFile = new File(BufferFileConfig.BUFFER_PATH + writeFileName); dataFile.createNewFile(); OffsetManager.INSTANCE.setWriteOffset(writeFileName, 0); try { @@ -98,6 +98,5 @@ public enum SegmentBufferManager { } public synchronized void flush() { - } } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferReader.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferReader.java index 148b33d9b..0126686cb 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferReader.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferReader.java @@ -53,7 +53,7 @@ public enum SegmentBufferReader { private void preRead() { String readFileName = OffsetManager.INSTANCE.getReadFileName(); if (StringUtils.isNotEmpty(readFileName)) { - File readFile = new File(SegmentBufferConfig.BUFFER_PATH + readFileName); + File readFile = new File(BufferFileConfig.BUFFER_PATH + readFileName); if (readFile.exists()) { deleteTheDataFilesBeforeReadFile(readFileName); long readFileOffset = OffsetManager.INSTANCE.getReadFileOffset(); @@ -69,7 +69,7 @@ public enum SegmentBufferReader { } private void deleteTheDataFilesBeforeReadFile(String readFileName) { - File[] dataFiles = new File(SegmentBufferConfig.BUFFER_PATH).listFiles(new PrefixFileNameFilter()); + File[] dataFiles = new File(BufferFileConfig.BUFFER_PATH).listFiles(new PrefixFileNameFilter()); long readFileCreateTime = getFileCreateTime(readFileName); for (File dataFile : dataFiles) { @@ -90,7 +90,7 @@ public enum SegmentBufferReader { private void readEarliestCreateDataFile() { String readFileName = OffsetManager.INSTANCE.getReadFileName(); - File[] dataFiles = new File(SegmentBufferConfig.BUFFER_PATH).listFiles(new PrefixFileNameFilter()); + File[] dataFiles = new File(BufferFileConfig.BUFFER_PATH).listFiles(new PrefixFileNameFilter()); if (CollectionUtils.isNotEmpty(dataFiles)) { if (dataFiles[0].getName().equals(readFileName)) { diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/parser/SegmentParse.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/parser/SegmentParse.java index 668987fc7..97d518eb0 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/parser/SegmentParse.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/parser/SegmentParse.java @@ -163,7 +163,7 @@ public class SegmentParse { } private void writeToBufferFile(String id, UpstreamSegment upstreamSegment) { - logger.debug("send to segment buffer write worker, id: {}", id); + logger.debug("push to segment buffer write worker, id: {}", id); SegmentStandardization standardization = new SegmentStandardization(id); standardization.setUpstreamSegment(upstreamSegment); Graph graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SEGMENT_STANDARDIZATION_GRAPH_ID, SegmentStandardization.class); diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/AgentStreamRemoteDataRegister.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/AgentStreamRemoteDataRegister.java new file mode 100644 index 000000000..65eee6eed --- /dev/null +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/AgentStreamRemoteDataRegister.java @@ -0,0 +1,53 @@ +/* + * 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.stream.worker; + +import org.skywalking.apm.collector.remote.service.RemoteDataRegisterService; +import org.skywalking.apm.collector.storage.table.node.NodeComponent; +import org.skywalking.apm.collector.storage.table.node.NodeMapping; +import org.skywalking.apm.collector.storage.table.noderef.NodeReference; +import org.skywalking.apm.collector.storage.table.register.Application; +import org.skywalking.apm.collector.storage.table.register.Instance; +import org.skywalking.apm.collector.storage.table.register.ServiceName; +import org.skywalking.apm.collector.storage.table.service.ServiceEntry; +import org.skywalking.apm.collector.storage.table.serviceref.ServiceReference; + +/** + * @author peng-yongsheng + */ +public class AgentStreamRemoteDataRegister { + + private final RemoteDataRegisterService remoteDataRegisterService; + + public AgentStreamRemoteDataRegister(RemoteDataRegisterService remoteDataRegisterService) { + this.remoteDataRegisterService = remoteDataRegisterService; + } + + public void register() { + remoteDataRegisterService.register(Application.class, Application::new); + remoteDataRegisterService.register(Instance.class, Instance::new); + remoteDataRegisterService.register(ServiceName.class, ServiceName::new); + + remoteDataRegisterService.register(NodeComponent.class, NodeComponent::new); + remoteDataRegisterService.register(NodeMapping.class, NodeMapping::new); + remoteDataRegisterService.register(NodeReference.class, NodeReference::new); + remoteDataRegisterService.register(ServiceEntry.class, ServiceEntry::new); + remoteDataRegisterService.register(ServiceReference.class, ServiceReference::new); + } +} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/CpuMetricService.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/CpuMetricService.java index 4fc8c80a2..1bbb03bd0 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/CpuMetricService.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/CpuMetricService.java @@ -23,6 +23,7 @@ import org.skywalking.apm.collector.agent.stream.service.jvm.ICpuMetricService; import org.skywalking.apm.collector.core.graph.Graph; import org.skywalking.apm.collector.core.graph.GraphManager; import org.skywalking.apm.collector.core.util.Const; +import org.skywalking.apm.collector.core.util.ObjectUtils; import org.skywalking.apm.collector.storage.table.jvm.CpuMetric; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -34,10 +35,13 @@ public class CpuMetricService implements ICpuMetricService { private final Logger logger = LoggerFactory.getLogger(CpuMetricService.class); - private final Graph cpuMetricGraph; + private Graph cpuMetricGraph; - public CpuMetricService() { - cpuMetricGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.CPU_METRIC_GRAPH_ID, CpuMetric.class); + private Graph getCpuMetricGraph() { + if (ObjectUtils.isEmpty(cpuMetricGraph)) { + cpuMetricGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.CPU_METRIC_GRAPH_ID, CpuMetric.class); + } + return cpuMetricGraph; } @Override public void send(int instanceId, long timeBucket, double usagePercent) { @@ -46,7 +50,7 @@ public class CpuMetricService implements ICpuMetricService { cpuMetric.setUsagePercent(usagePercent); cpuMetric.setTimeBucket(timeBucket); - logger.debug("send to cpu metric graph, id: {}", cpuMetric.getId()); - cpuMetricGraph.start(cpuMetric); + logger.debug("push to cpu metric graph, id: {}", cpuMetric.getId()); + getCpuMetricGraph().start(cpuMetric); } } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/GCMetricService.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/GCMetricService.java index 8d8f9c7a0..e2ae3a348 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/GCMetricService.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/GCMetricService.java @@ -23,6 +23,7 @@ import org.skywalking.apm.collector.agent.stream.service.jvm.IGCMetricService; import org.skywalking.apm.collector.core.graph.Graph; import org.skywalking.apm.collector.core.graph.GraphManager; import org.skywalking.apm.collector.core.util.Const; +import org.skywalking.apm.collector.core.util.ObjectUtils; import org.skywalking.apm.collector.storage.table.jvm.GCMetric; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -34,10 +35,13 @@ public class GCMetricService implements IGCMetricService { private final Logger logger = LoggerFactory.getLogger(GCMetricService.class); - private final Graph gcMetricGraph; + private Graph gcMetricGraph; - public GCMetricService() { - gcMetricGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.GC_METRIC_GRAPH_ID, GCMetric.class); + private Graph getGcMetricGraph() { + if (ObjectUtils.isEmpty(gcMetricGraph)) { + gcMetricGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.GC_METRIC_GRAPH_ID, GCMetric.class); + } + return gcMetricGraph; } @Override public void send(int instanceId, long timeBucket, int phraseValue, long count, long time) { @@ -48,7 +52,7 @@ public class GCMetricService implements IGCMetricService { gcMetric.setTime(time); gcMetric.setTimeBucket(timeBucket); - logger.debug("send to gc metric graph, id: {}", gcMetric.getId()); - gcMetricGraph.start(gcMetric); + logger.debug("push to gc metric graph, id: {}", gcMetric.getId()); + getGcMetricGraph().start(gcMetric); } } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/InstanceHeartBeatService.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/InstanceHeartBeatService.java index 78084e83b..8ef9c3506 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/InstanceHeartBeatService.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/InstanceHeartBeatService.java @@ -22,6 +22,7 @@ import org.skywalking.apm.collector.agent.stream.graph.JvmMetricStreamGraph; import org.skywalking.apm.collector.agent.stream.service.jvm.IInstanceHeartBeatService; import org.skywalking.apm.collector.core.graph.Graph; import org.skywalking.apm.collector.core.graph.GraphManager; +import org.skywalking.apm.collector.core.util.ObjectUtils; import org.skywalking.apm.collector.core.util.TimeBucketUtils; import org.skywalking.apm.collector.storage.table.register.Instance; import org.slf4j.Logger; @@ -34,10 +35,13 @@ public class InstanceHeartBeatService implements IInstanceHeartBeatService { private final Logger logger = LoggerFactory.getLogger(InstanceHeartBeatService.class); - private final Graph heartBeatGraph; + private Graph heartBeatGraph; - public InstanceHeartBeatService() { - this.heartBeatGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.INST_HEART_BEAT_GRAPH_ID, Instance.class); + private Graph getHeartBeatGraph() { + if (ObjectUtils.isEmpty(heartBeatGraph)) { + this.heartBeatGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.INST_HEART_BEAT_GRAPH_ID, Instance.class); + } + return heartBeatGraph; } @Override public void send(int instanceId, long heartBeatTime) { @@ -45,7 +49,7 @@ public class InstanceHeartBeatService implements IInstanceHeartBeatService { instance.setHeartBeatTime(TimeBucketUtils.INSTANCE.getSecondTimeBucket(heartBeatTime)); instance.setInstanceId(instanceId); - logger.debug("send to instance heart beat persistence worker, id: {}", instance.getId()); - heartBeatGraph.start(instance); + logger.debug("push to instance heart beat persistence worker, id: {}", instance.getId()); + getHeartBeatGraph().start(instance); } } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/MemoryMetricService.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/MemoryMetricService.java index 9909f7235..32d5e8f2e 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/MemoryMetricService.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/MemoryMetricService.java @@ -23,6 +23,7 @@ import org.skywalking.apm.collector.agent.stream.service.jvm.IMemoryMetricServic import org.skywalking.apm.collector.core.graph.Graph; import org.skywalking.apm.collector.core.graph.GraphManager; import org.skywalking.apm.collector.core.util.Const; +import org.skywalking.apm.collector.core.util.ObjectUtils; import org.skywalking.apm.collector.storage.table.jvm.MemoryMetric; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -34,10 +35,13 @@ public class MemoryMetricService implements IMemoryMetricService { private final Logger logger = LoggerFactory.getLogger(MemoryMetricService.class); - private final Graph memoryMetricGraph; + private Graph memoryMetricGraph; - public MemoryMetricService() { - this.memoryMetricGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.MEMORY_METRIC_GRAPH_ID, MemoryMetric.class); + private Graph getMemoryMetricGraph() { + if (ObjectUtils.isEmpty(memoryMetricGraph)) { + this.memoryMetricGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.MEMORY_METRIC_GRAPH_ID, MemoryMetric.class); + } + return memoryMetricGraph; } @Override @@ -51,7 +55,7 @@ public class MemoryMetricService implements IMemoryMetricService { memoryMetric.setCommitted(commited); memoryMetric.setTimeBucket(timeBucket); - logger.debug("send to memory metric graph, id: {}", memoryMetric.getId()); - memoryMetricGraph.start(memoryMetric); + logger.debug("push to memory metric graph, id: {}", memoryMetric.getId()); + getMemoryMetricGraph().start(memoryMetric); } } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/MemoryPoolMetricService.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/MemoryPoolMetricService.java index 3af24bc2b..8e4aed00a 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/MemoryPoolMetricService.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/MemoryPoolMetricService.java @@ -23,6 +23,7 @@ import org.skywalking.apm.collector.agent.stream.service.jvm.IMemoryPoolMetricSe import org.skywalking.apm.collector.core.graph.Graph; import org.skywalking.apm.collector.core.graph.GraphManager; import org.skywalking.apm.collector.core.util.Const; +import org.skywalking.apm.collector.core.util.ObjectUtils; import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetric; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -34,10 +35,13 @@ public class MemoryPoolMetricService implements IMemoryPoolMetricService { private final Logger logger = LoggerFactory.getLogger(MemoryPoolMetricService.class); - private final Graph memoryPoolMetricGraph; + private Graph memoryPoolMetricGraph; - public MemoryPoolMetricService() { - this.memoryPoolMetricGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.MEMORY_POOL_METRIC_GRAPH_ID, MemoryPoolMetric.class); + private Graph getMemoryPoolMetricGraph() { + if (ObjectUtils.isEmpty(memoryPoolMetricGraph)) { + this.memoryPoolMetricGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.MEMORY_POOL_METRIC_GRAPH_ID, MemoryPoolMetric.class); + } + return memoryPoolMetricGraph; } @Override @@ -51,7 +55,7 @@ public class MemoryPoolMetricService implements IMemoryPoolMetricService { memoryPoolMetric.setCommitted(commited); memoryPoolMetric.setTimeBucket(timeBucket); - logger.debug("send to memory pool metric graph, id: {}", memoryPoolMetric.getId()); - memoryPoolMetricGraph.start(memoryPoolMetric); + logger.debug("push to memory pool metric graph, id: {}", memoryPoolMetric.getId()); + getMemoryPoolMetricGraph().start(memoryPoolMetric); } } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/ApplicationIDService.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/ApplicationIDService.java index d218b537b..42aa307ba 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/ApplicationIDService.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/ApplicationIDService.java @@ -25,6 +25,7 @@ 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.util.ObjectUtils; import org.skywalking.apm.collector.storage.table.register.Application; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -37,23 +38,36 @@ public class ApplicationIDService implements IApplicationIDService { private final Logger logger = LoggerFactory.getLogger(ApplicationIDService.class); private final ModuleManager moduleManager; - private final Graph applicationRegisterGraph; + private ApplicationCacheService applicationCacheService; + private Graph applicationRegisterGraph; public ApplicationIDService(ModuleManager moduleManager) { this.moduleManager = moduleManager; - this.applicationRegisterGraph = GraphManager.INSTANCE.createIfAbsent(RegisterStreamGraph.APPLICATION_REGISTER_GRAPH_ID, Application.class); + } + + private Graph getApplicationRegisterGraph() { + if (ObjectUtils.isEmpty(applicationRegisterGraph)) { + this.applicationRegisterGraph = GraphManager.INSTANCE.createIfAbsent(RegisterStreamGraph.APPLICATION_REGISTER_GRAPH_ID, Application.class); + } + return this.applicationRegisterGraph; + } + + private ApplicationCacheService getApplicationCacheService() { + if (ObjectUtils.isEmpty(applicationCacheService)) { + this.applicationCacheService = moduleManager.find(CacheModule.NAME).getService(ApplicationCacheService.class); + } + return applicationCacheService; } public int getOrCreate(String applicationCode) { - ApplicationCacheService service = moduleManager.find(CacheModule.NAME).getService(ApplicationCacheService.class); - int applicationId = service.get(applicationCode); + int applicationId = getApplicationCacheService().get(applicationCode); if (applicationId == 0) { Application application = new Application(applicationCode); application.setApplicationCode(applicationCode); application.setApplicationId(0); - applicationRegisterGraph.start(application); + getApplicationRegisterGraph().start(application); } return applicationId; } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/InstanceIDService.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/InstanceIDService.java index c68919e66..4f1c94231 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/InstanceIDService.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/InstanceIDService.java @@ -25,6 +25,7 @@ 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.util.ObjectUtils; import org.skywalking.apm.collector.storage.StorageModule; import org.skywalking.apm.collector.storage.dao.IInstanceRegisterDAO; import org.skywalking.apm.collector.storage.table.register.Instance; @@ -39,17 +40,38 @@ public class InstanceIDService implements IInstanceIDService { private final Logger logger = LoggerFactory.getLogger(InstanceIDService.class); private final ModuleManager moduleManager; - private final Graph instanceRegisterGraph; + private InstanceCacheService instanceCacheService; + private Graph instanceRegisterGraph; + private IInstanceRegisterDAO instanceRegisterDAO; public InstanceIDService(ModuleManager moduleManager) { this.moduleManager = moduleManager; - this.instanceRegisterGraph = GraphManager.INSTANCE.createIfAbsent(RegisterStreamGraph.INSTANCE_REGISTER_GRAPH_ID, Instance.class); + } + + private InstanceCacheService getInstanceCacheService() { + if (ObjectUtils.isEmpty(instanceCacheService)) { + instanceCacheService = moduleManager.find(CacheModule.NAME).getService(InstanceCacheService.class); + } + return instanceCacheService; + } + + private Graph getInstanceRegisterGraph() { + if (ObjectUtils.isEmpty(instanceRegisterGraph)) { + this.instanceRegisterGraph = GraphManager.INSTANCE.createIfAbsent(RegisterStreamGraph.INSTANCE_REGISTER_GRAPH_ID, Instance.class); + } + return instanceRegisterGraph; + } + + private IInstanceRegisterDAO getInstanceRegisterDAO() { + if (ObjectUtils.isEmpty(instanceRegisterDAO)) { + instanceRegisterDAO = moduleManager.find(StorageModule.NAME).getService(IInstanceRegisterDAO.class); + } + return instanceRegisterDAO; } 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); + int instanceId = getInstanceCacheService().getInstanceId(applicationId, agentUUID); if (instanceId == 0) { Instance instance = new Instance("0"); @@ -60,15 +82,13 @@ public class InstanceIDService implements IInstanceIDService { instance.setInstanceId(0); instance.setOsInfo(osInfo); - instanceRegisterGraph.start(instance); + getInstanceRegisterGraph().start(instance); } return instanceId; } public void recover(int instanceId, int applicationId, long registerTime, String osInfo) { logger.debug("instance recover, instance id: {}, application id: {}, register time: {}", instanceId, applicationId, registerTime); - final IInstanceRegisterDAO instanceRegisterDAO = moduleManager.find(StorageModule.NAME).getService(IInstanceRegisterDAO.class); - Instance instance = new Instance(String.valueOf(instanceId)); instance.setApplicationId(applicationId); instance.setAgentUUID(""); @@ -77,6 +97,6 @@ public class InstanceIDService implements IInstanceIDService { instance.setInstanceId(instanceId); instance.setOsInfo(osInfo); - instanceRegisterDAO.save(instance); + getInstanceRegisterDAO().save(instance); } } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/InstanceRegisterRemoteWorker.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/InstanceRegisterRemoteWorker.java index f36fae2ba..07ba0fcc9 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/InstanceRegisterRemoteWorker.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/InstanceRegisterRemoteWorker.java @@ -39,7 +39,7 @@ public class InstanceRegisterRemoteWorker extends AbstractRemoteWorker serviceNameRegisterGraph; + private ServiceIdCacheService serviceIdCacheService; + private Graph serviceNameRegisterGraph; public ServiceNameService(ModuleManager moduleManager) { this.moduleManager = moduleManager; - this.serviceNameRegisterGraph = GraphManager.INSTANCE.createIfAbsent(RegisterStreamGraph.SERVICE_NAME_REGISTER_GRAPH_ID, ServiceName.class); + } + + private ServiceIdCacheService getServiceIdCacheService() { + if (ObjectUtils.isEmpty(serviceIdCacheService)) { + serviceIdCacheService = moduleManager.find(CacheModule.NAME).getService(ServiceIdCacheService.class); + } + return serviceIdCacheService; + } + + private Graph getServiceNameRegisterGraph() { + if (ObjectUtils.isEmpty(serviceNameRegisterGraph)) { + this.serviceNameRegisterGraph = GraphManager.INSTANCE.createIfAbsent(RegisterStreamGraph.SERVICE_NAME_REGISTER_GRAPH_ID, ServiceName.class); + } + return serviceNameRegisterGraph; } public int getOrCreate(int applicationId, String serviceName) { - ServiceIdCacheService idCacheService = moduleManager.find(CacheModule.NAME).getService(ServiceIdCacheService.class); - int serviceId = idCacheService.get(applicationId, serviceName); + int serviceId = getServiceIdCacheService().get(applicationId, serviceName); if (serviceId == 0) { ServiceName service = new ServiceName("0"); @@ -54,7 +68,7 @@ public class ServiceNameService implements IServiceNameService { service.setServiceName(serviceName); service.setServiceId(0); - serviceNameRegisterGraph.start(service); + getServiceNameRegisterGraph().start(service); } return serviceId; } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeComponentRemoteWorker.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeComponentRemoteWorker.java index 732a4e16d..d1b86441c 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeComponentRemoteWorker.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeComponentRemoteWorker.java @@ -31,7 +31,7 @@ import org.skywalking.apm.collector.stream.worker.base.WorkerException; */ public class NodeComponentRemoteWorker extends AbstractRemoteWorker { - public NodeComponentRemoteWorker(ModuleManager moduleManager) { + NodeComponentRemoteWorker(ModuleManager moduleManager) { super(moduleManager); } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeMappingRemoteWorker.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeMappingRemoteWorker.java index 6f8b8fc4e..56a4f8734 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeMappingRemoteWorker.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeMappingRemoteWorker.java @@ -31,7 +31,7 @@ import org.skywalking.apm.collector.stream.worker.base.WorkerException; */ public class NodeMappingRemoteWorker extends AbstractRemoteWorker { - public NodeMappingRemoteWorker(ModuleManager moduleManager) { + NodeMappingRemoteWorker(ModuleManager moduleManager) { super(moduleManager); } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeMappingSpanListener.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeMappingSpanListener.java index d1ac5bfaf..e42275d98 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeMappingSpanListener.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeMappingSpanListener.java @@ -67,7 +67,7 @@ public class NodeMappingSpanListener implements RefsListener, FirstSpanListener for (NodeMapping nodeMapping : nodeMappings) { nodeMapping.setId(timeBucket + Const.ID_SPLIT + nodeMapping.getId()); nodeMapping.setTimeBucket(timeBucket); - logger.debug("send to node mapping aggregation worker, id: {}", nodeMapping.getId()); + logger.debug("push to node mapping aggregation worker, id: {}", nodeMapping.getId()); graph.start(nodeMapping); } } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/noderef/NodeReferenceRemoteWorker.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/noderef/NodeReferenceRemoteWorker.java index f12a7a1e7..a06dcc133 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/noderef/NodeReferenceRemoteWorker.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/noderef/NodeReferenceRemoteWorker.java @@ -31,7 +31,7 @@ import org.skywalking.apm.collector.stream.worker.base.WorkerException; */ public class NodeReferenceRemoteWorker extends AbstractRemoteWorker { - public NodeReferenceRemoteWorker(ModuleManager moduleManager) { + NodeReferenceRemoteWorker(ModuleManager moduleManager) { super(moduleManager); } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntrySpanListener.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntrySpanListener.java index 0312bbf6d..1540fcf89 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntrySpanListener.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntrySpanListener.java @@ -84,7 +84,7 @@ public class ServiceEntrySpanListener implements RefsListener, FirstSpanListener serviceEntry.setRegisterTime(timeBucket); serviceEntry.setNewestTime(timeBucket); - logger.debug("send to service entry aggregation worker, id: {}", serviceEntry.getId()); + logger.debug("push to service entry aggregation worker, id: {}", serviceEntry.getId()); Graph graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SERVICE_ENTRY_GRAPH_ID, ServiceEntry.class); graph.start(serviceEntry); } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/serviceref/ServiceReferenceSpanListener.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/serviceref/ServiceReferenceSpanListener.java index 45aa2a1e3..9c8256d74 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/serviceref/ServiceReferenceSpanListener.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/serviceref/ServiceReferenceSpanListener.java @@ -130,7 +130,7 @@ public class ServiceReferenceSpanListener implements FirstSpanListener, EntrySpa serviceReference.setId(idBuilder.toString()); serviceReference.setTimeBucket(timeBucket); - logger.debug("send to service reference aggregation worker, id: {}", serviceReference.getId()); + logger.debug("push to service reference aggregation worker, id: {}", serviceReference.getId()); Graph graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SERVICE_REFERENCE_GRAPH_ID, ServiceReference.class); graph.start(serviceReference); 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 d8628b391..570863a22 100644 --- a/apm-collector/apm-collector-boot/src/main/resources/application.yml +++ b/apm-collector/apm-collector-boot/src/main/resources/application.yml @@ -5,8 +5,12 @@ naming: jetty: host: localhost - port: 10800 + port: 12800 context_path: / +remote: + gRPC: + host: localhost + port: 10800 agent_gRPC: gRPC: host: localhost @@ -18,6 +22,7 @@ agent_jetty: context_path: / agent_stream: default: + buffer_file_path: ../buffer/ buffer_offset_max_file_size: 10M buffer_segment_max_file_size: 500M ui: @@ -25,10 +30,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 diff --git a/apm-collector/apm-collector-boot/src/main/resources/log4j2.xml b/apm-collector/apm-collector-boot/src/main/resources/log4j2.xml index db54a6402..e6915ec9a 100644 --- a/apm-collector/apm-collector-boot/src/main/resources/log4j2.xml +++ b/apm-collector/apm-collector-boot/src/main/resources/log4j2.xml @@ -17,7 +17,7 @@ ~ Project repository: https://github.com/OpenSkywalking/skywalking --> - + @@ -29,7 +29,7 @@ - + diff --git a/apm-collector/apm-collector-cache/collector-cache-guava-provider/pom.xml b/apm-collector/apm-collector-cache/collector-cache-guava-provider/pom.xml index dda14c71e..54fe99b65 100644 --- a/apm-collector/apm-collector-cache/collector-cache-guava-provider/pom.xml +++ b/apm-collector/apm-collector-cache/collector-cache-guava-provider/pom.xml @@ -36,5 +36,10 @@ collector-cache-define ${project.version} + + com.google.guava + guava + 19.0 + \ No newline at end of file diff --git a/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ApplicationCacheGuavaService.java b/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ApplicationCacheGuavaService.java index 80770f160..743390be9 100644 --- a/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ApplicationCacheGuavaService.java +++ b/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ApplicationCacheGuavaService.java @@ -23,6 +23,7 @@ import com.google.common.cache.CacheBuilder; import org.skywalking.apm.collector.cache.service.ApplicationCacheService; import org.skywalking.apm.collector.core.module.ModuleManager; import org.skywalking.apm.collector.core.util.Const; +import org.skywalking.apm.collector.core.util.ObjectUtils; import org.skywalking.apm.collector.core.util.StringUtils; import org.skywalking.apm.collector.storage.StorageModule; import org.skywalking.apm.collector.storage.dao.IApplicationCacheDAO; @@ -38,22 +39,30 @@ public class ApplicationCacheGuavaService implements ApplicationCacheService { private final Cache codeCache = CacheBuilder.newBuilder().initialCapacity(100).maximumSize(1000).build(); - private final IApplicationCacheDAO applicationCacheDAO; + private final ModuleManager moduleManager; + private IApplicationCacheDAO applicationCacheDAO; public ApplicationCacheGuavaService(ModuleManager moduleManager) { - this.applicationCacheDAO = moduleManager.find(StorageModule.NAME).getService(IApplicationCacheDAO.class); + this.moduleManager = moduleManager; + } + + private IApplicationCacheDAO getApplicationCacheDAO() { + if (ObjectUtils.isEmpty(applicationCacheDAO)) { + this.applicationCacheDAO = moduleManager.find(StorageModule.NAME).getService(IApplicationCacheDAO.class); + } + return this.applicationCacheDAO; } public int get(String applicationCode) { int applicationId = 0; try { - applicationId = codeCache.get(applicationCode, () -> applicationCacheDAO.getApplicationId(applicationCode)); + applicationId = codeCache.get(applicationCode, () -> getApplicationCacheDAO().getApplicationId(applicationCode)); } catch (Throwable e) { logger.error(e.getMessage(), e); } if (applicationId == 0) { - applicationId = applicationCacheDAO.getApplicationId(applicationCode); + applicationId = getApplicationCacheDAO().getApplicationId(applicationCode); if (applicationId != 0) { codeCache.put(applicationCode, applicationId); } @@ -66,13 +75,13 @@ public class ApplicationCacheGuavaService implements ApplicationCacheService { public String get(int applicationId) { String applicationCode = Const.EMPTY_STRING; try { - applicationCode = idCache.get(applicationId, () -> applicationCacheDAO.getApplicationCode(applicationId)); + applicationCode = idCache.get(applicationId, () -> getApplicationCacheDAO().getApplicationCode(applicationId)); } catch (Throwable e) { logger.error(e.getMessage(), e); } if (StringUtils.isEmpty(applicationCode)) { - applicationCode = applicationCacheDAO.getApplicationCode(applicationId); + applicationCode = getApplicationCacheDAO().getApplicationCode(applicationId); if (StringUtils.isNotEmpty(applicationCode)) { codeCache.put(applicationCode, applicationId); } diff --git a/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/InstanceCacheGuavaService.java b/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/InstanceCacheGuavaService.java index 330b3b9a1..0f2e9a27e 100644 --- a/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/InstanceCacheGuavaService.java +++ b/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/InstanceCacheGuavaService.java @@ -23,6 +23,7 @@ import com.google.common.cache.CacheBuilder; import org.skywalking.apm.collector.cache.service.InstanceCacheService; import org.skywalking.apm.collector.core.module.ModuleManager; import org.skywalking.apm.collector.core.util.Const; +import org.skywalking.apm.collector.core.util.ObjectUtils; import org.skywalking.apm.collector.storage.StorageModule; import org.skywalking.apm.collector.storage.dao.IInstanceCacheDAO; import org.slf4j.Logger; @@ -39,23 +40,31 @@ public class InstanceCacheGuavaService implements InstanceCacheService { private final Cache stringCache = CacheBuilder.newBuilder().initialCapacity(100).maximumSize(5000).build(); - private final IInstanceCacheDAO instanceCacheDAO; + private final ModuleManager moduleManager; + private IInstanceCacheDAO instanceCacheDAO; public InstanceCacheGuavaService(ModuleManager moduleManager) { - this.instanceCacheDAO = moduleManager.find(StorageModule.NAME).getService(IInstanceCacheDAO.class); + this.moduleManager = moduleManager; + } + + private IInstanceCacheDAO getInstanceCacheDAO() { + if (ObjectUtils.isEmpty(instanceCacheDAO)) { + this.instanceCacheDAO = moduleManager.find(StorageModule.NAME).getService(IInstanceCacheDAO.class); + } + return this.instanceCacheDAO; } public int get(int applicationInstanceId) { int applicationId = 0; try { - applicationId = integerCache.get(applicationInstanceId, () -> instanceCacheDAO.getApplicationId(applicationInstanceId)); + applicationId = integerCache.get(applicationInstanceId, () -> getInstanceCacheDAO().getApplicationId(applicationInstanceId)); } catch (Throwable e) { logger.error(e.getMessage(), e); } if (applicationId == 0) { - applicationId = instanceCacheDAO.getApplicationId(applicationInstanceId); + applicationId = getInstanceCacheDAO().getApplicationId(applicationInstanceId); if (applicationId != 0) { integerCache.put(applicationInstanceId, applicationId); } @@ -68,13 +77,13 @@ public class InstanceCacheGuavaService implements InstanceCacheService { int instanceId = 0; try { - instanceId = stringCache.get(key, () -> instanceCacheDAO.getInstanceId(applicationId, agentUUID)); + instanceId = stringCache.get(key, () -> getInstanceCacheDAO().getInstanceId(applicationId, agentUUID)); } catch (Throwable e) { logger.error(e.getMessage(), e); } if (instanceId == 0) { - instanceId = instanceCacheDAO.getInstanceId(applicationId, agentUUID); + instanceId = getInstanceCacheDAO().getInstanceId(applicationId, agentUUID); if (applicationId != 0) { stringCache.put(key, instanceId); } diff --git a/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ServiceIdCacheGuavaService.java b/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ServiceIdCacheGuavaService.java index eb59ec750..85d6aa5aa 100644 --- a/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ServiceIdCacheGuavaService.java +++ b/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ServiceIdCacheGuavaService.java @@ -23,6 +23,7 @@ import com.google.common.cache.CacheBuilder; import org.skywalking.apm.collector.cache.service.ServiceIdCacheService; import org.skywalking.apm.collector.core.module.ModuleManager; import org.skywalking.apm.collector.core.util.Const; +import org.skywalking.apm.collector.core.util.ObjectUtils; import org.skywalking.apm.collector.storage.StorageModule; import org.skywalking.apm.collector.storage.dao.IServiceNameCacheDAO; import org.slf4j.Logger; @@ -37,22 +38,30 @@ public class ServiceIdCacheGuavaService implements ServiceIdCacheService { private final Cache serviceIdCache = CacheBuilder.newBuilder().maximumSize(1000).build(); - private final IServiceNameCacheDAO serviceNameCacheDAO; + private final ModuleManager moduleManager; + private IServiceNameCacheDAO serviceNameCacheDAO; public ServiceIdCacheGuavaService(ModuleManager moduleManager) { - this.serviceNameCacheDAO = moduleManager.find(StorageModule.NAME).getService(IServiceNameCacheDAO.class); + this.moduleManager = moduleManager; + } + + private IServiceNameCacheDAO getServiceNameCacheDAO() { + if (ObjectUtils.isEmpty(serviceNameCacheDAO)) { + this.serviceNameCacheDAO = moduleManager.find(StorageModule.NAME).getService(IServiceNameCacheDAO.class); + } + return this.serviceNameCacheDAO; } public int get(int applicationId, String serviceName) { int serviceId = 0; try { - serviceId = serviceIdCache.get(applicationId + Const.ID_SPLIT + serviceName, () -> serviceNameCacheDAO.getServiceId(applicationId, serviceName)); + serviceId = serviceIdCache.get(applicationId + Const.ID_SPLIT + serviceName, () -> getServiceNameCacheDAO().getServiceId(applicationId, serviceName)); } catch (Throwable e) { logger.error(e.getMessage(), e); } if (serviceId == 0) { - serviceId = serviceNameCacheDAO.getServiceId(applicationId, serviceName); + serviceId = getServiceNameCacheDAO().getServiceId(applicationId, serviceName); if (serviceId != 0) { serviceIdCache.put(applicationId + Const.ID_SPLIT + serviceName, serviceId); } diff --git a/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ServiceNameCacheGuavaService.java b/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ServiceNameCacheGuavaService.java index 2b235ef9d..55b9d7792 100644 --- a/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ServiceNameCacheGuavaService.java +++ b/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ServiceNameCacheGuavaService.java @@ -23,6 +23,7 @@ import com.google.common.cache.CacheBuilder; import org.skywalking.apm.collector.cache.service.ServiceNameCacheService; import org.skywalking.apm.collector.core.module.ModuleManager; import org.skywalking.apm.collector.core.util.Const; +import org.skywalking.apm.collector.core.util.ObjectUtils; import org.skywalking.apm.collector.core.util.StringUtils; import org.skywalking.apm.collector.storage.StorageModule; import org.skywalking.apm.collector.storage.dao.IServiceNameCacheDAO; @@ -38,22 +39,30 @@ public class ServiceNameCacheGuavaService implements ServiceNameCacheService { private final Cache serviceNameCache = CacheBuilder.newBuilder().maximumSize(10000).build(); - private final IServiceNameCacheDAO serviceNameCacheDAO; + private final ModuleManager moduleManager; + private IServiceNameCacheDAO serviceNameCacheDAO; public ServiceNameCacheGuavaService(ModuleManager moduleManager) { - this.serviceNameCacheDAO = moduleManager.find(StorageModule.NAME).getService(IServiceNameCacheDAO.class); + this.moduleManager = moduleManager; + } + + private IServiceNameCacheDAO getServiceNameCacheDAO() { + if (ObjectUtils.isEmpty(serviceNameCacheDAO)) { + this.serviceNameCacheDAO = moduleManager.find(StorageModule.NAME).getService(IServiceNameCacheDAO.class); + } + return this.serviceNameCacheDAO; } public String get(int serviceId) { String serviceName = Const.EMPTY_STRING; try { - serviceName = serviceNameCache.get(serviceId, () -> serviceNameCacheDAO.getServiceName(serviceId)); + serviceName = serviceNameCache.get(serviceId, () -> getServiceNameCacheDAO().getServiceName(serviceId)); } catch (Throwable e) { logger.error(e.getMessage(), e); } if (StringUtils.isEmpty(serviceName)) { - serviceName = serviceNameCacheDAO.getServiceName(serviceId); + serviceName = getServiceNameCacheDAO().getServiceName(serviceId); if (StringUtils.isNotEmpty(serviceName)) { serviceNameCache.put(serviceId, serviceName); } diff --git a/apm-collector/apm-collector-component/client-component/pom.xml b/apm-collector/apm-collector-component/client-component/pom.xml index 8d1c5e84f..da4098524 100644 --- a/apm-collector/apm-collector-component/client-component/pom.xml +++ b/apm-collector/apm-collector-component/client-component/pom.xml @@ -32,6 +32,34 @@ snakeyaml org.yaml + + netty-common + io.netty + + + netty-transport + io.netty + + + netty-codec + io.netty + + + netty-codec-http + io.netty + + + netty-buffer + io.netty + + + netty-handler + io.netty + + + netty-resolver + io.netty + @@ -50,9 +78,15 @@ - io.grpc - grpc-core - 1.4.0 + org.skywalking + apm-network + ${project.version} + + + guava + com.google.guava + + diff --git a/apm-collector/apm-collector-component/server-component/pom.xml b/apm-collector/apm-collector-component/server-component/pom.xml index 817f6385b..6a243b0df 100644 --- a/apm-collector/apm-collector-component/server-component/pom.xml +++ b/apm-collector/apm-collector-component/server-component/pom.xml @@ -21,6 +21,12 @@ org.skywalking apm-network ${project.version} + + + guava + com.google.guava + + org.eclipse.jetty diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/config/SystemConfig.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cache/Collection.java similarity index 71% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/config/SystemConfig.java rename to apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cache/Collection.java index ec787f684..fe2003221 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/config/SystemConfig.java +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cache/Collection.java @@ -16,11 +16,28 @@ * Project repository: https://github.com/OpenSkywalking/skywalking */ -package org.skywalking.apm.collector.core.config; +package org.skywalking.apm.collector.core.cache; /** * @author peng-yongsheng */ -public class SystemConfig { - public static String DATA_PATH = "../data"; +public interface Collection { + + void reading(); + + boolean isReading(); + + void writing(); + + boolean isWriting(); + + void clear(); + + int size(); + + void finishReading(); + + void finishWriting(); + + Data collection(); } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Window.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cache/Window.java similarity index 74% rename from apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Window.java rename to apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cache/Window.java index a1b53450a..8dbebcd9b 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Window.java +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cache/Window.java @@ -16,28 +16,30 @@ * Project repository: https://github.com/OpenSkywalking/skywalking */ -package org.skywalking.apm.collector.stream.worker.impl.data; +package org.skywalking.apm.collector.core.cache; import java.util.concurrent.atomic.AtomicInteger; /** * @author peng-yongsheng */ -public abstract class Window { +public abstract class Window { private AtomicInteger windowSwitch = new AtomicInteger(0); - private DataCollection pointer; + private WindowCollection pointer; - private DataCollection windowDataA; - private DataCollection windowDataB; + private WindowCollection windowDataA; + private WindowCollection windowDataB; - public Window() { - windowDataA = new DataCollection(); - windowDataB = new DataCollection(); - pointer = windowDataA; + protected Window() { + this.windowDataA = collectionInstance(); + this.windowDataB = collectionInstance(); + this.pointer = windowDataA; } + public abstract WindowCollection collectionInstance(); + public boolean trySwitchPointer() { return windowSwitch.incrementAndGet() == 1 && !getLast().isReading(); } @@ -55,7 +57,7 @@ public abstract class Window { getLast().reading(); } - protected DataCollection getCurrentAndWriting() { + protected WindowCollection getCurrentAndWriting() { if (pointer == windowDataA) { windowDataA.writing(); return windowDataA; @@ -65,11 +67,11 @@ public abstract class Window { } } - protected DataCollection getCurrent() { + protected WindowCollection getCurrent() { return pointer; } - public DataCollection getLast() { + public WindowCollection getLast() { if (pointer == windowDataA) { return windowDataB; } else { diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/data/Data.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/data/Data.java index 6f6beca17..cb0eb02df 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/data/Data.java +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/data/Data.java @@ -183,6 +183,7 @@ public abstract class Data extends EndOfBatchQueueMessage { for (Boolean dataBoolean : dataBooleans) { dataStr.append(dataBoolean).append(","); } + dataStr.append("]"); return dataStr.toString(); } } diff --git a/apm-collector/apm-collector-core/src/main/resources/application-default.yml b/apm-collector/apm-collector-core/src/main/resources/application-default.yml index 6d3aefc73..200609e65 100644 --- a/apm-collector/apm-collector-core/src/main/resources/application-default.yml +++ b/apm-collector/apm-collector-core/src/main/resources/application-default.yml @@ -15,6 +15,20 @@ naming: host: localhost port: 10800 context_path: / +agent_gRPC: + gRPC: + host: localhost + port: 11800 +agent_jetty: + jetty: + host: localhost + port: 12800 + context_path: / +agent_stream: + default: + buffer_file_path: ../buffer/ + buffer_offset_max_file_size: 10M + buffer_segment_max_file_size: 500M ui: jetty: host: localhost diff --git a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/RemoteException.java b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/RemoteException.java new file mode 100644 index 000000000..3c9339ec8 --- /dev/null +++ b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/RemoteException.java @@ -0,0 +1,35 @@ +/* + * 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.remote; + +import org.skywalking.apm.collector.core.CollectorException; + +/** + * @author peng-yongsheng + */ +public abstract class RemoteException extends CollectorException { + + public RemoteException(String message) { + super(message); + } + + public RemoteException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/RemoteModule.java b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/RemoteModule.java index b3e6fd673..e91b96574 100644 --- a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/RemoteModule.java +++ b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/RemoteModule.java @@ -19,8 +19,8 @@ package org.skywalking.apm.collector.remote; import org.skywalking.apm.collector.core.module.Module; +import org.skywalking.apm.collector.remote.service.RemoteDataRegisterService; import org.skywalking.apm.collector.remote.service.RemoteSenderService; -import org.skywalking.apm.collector.remote.service.RemoteServerService; /** * @author peng-yongsheng @@ -34,6 +34,6 @@ public class RemoteModule extends Module { } @Override public Class[] services() { - return new Class[] {RemoteServerService.class, RemoteSenderService.class}; + return new Class[] {RemoteSenderService.class, RemoteDataRegisterService.class}; } } diff --git a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/CommonRemoteDataRegisterService.java b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/CommonRemoteDataRegisterService.java new file mode 100644 index 000000000..6b13da96a --- /dev/null +++ b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/CommonRemoteDataRegisterService.java @@ -0,0 +1,71 @@ +/* + * 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.remote.service; + +import java.util.HashMap; +import java.util.Map; +import org.skywalking.apm.collector.core.data.Data; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng + */ +public class CommonRemoteDataRegisterService implements RemoteDataRegisterService, RemoteDataIDGetter, RemoteDataInstanceCreatorGetter { + + private final Logger logger = LoggerFactory.getLogger(CommonRemoteDataRegisterService.class); + + private Integer id; + private final Map, Integer> dataClassMapping; + private final Map dataInstanceCreatorMapping; + + public CommonRemoteDataRegisterService() { + this.id = 1; + this.dataClassMapping = new HashMap<>(); + this.dataInstanceCreatorMapping = new HashMap<>(); + } + + @Override public void register(Class dataClass, RemoteDataInstanceCreator instanceCreator) { + if (!dataClassMapping.containsKey(dataClass)) { + dataClassMapping.put(dataClass, this.id); + dataInstanceCreatorMapping.put(this.id, instanceCreator); + this.id++; + } else { + logger.warn("The data class {} was registered.", dataClass.getName()); + } + } + + @Override + public Integer getRemoteDataId(Class dataClass) throws RemoteDataMappingIdNotFoundException { + if (dataClassMapping.containsKey(dataClass)) { + return dataClassMapping.get(dataClass); + } else { + throw new RemoteDataMappingIdNotFoundException("Could not found the id of remote data class " + dataClass.getName()); + } + } + + @Override public RemoteDataInstanceCreator getInstanceCreator( + Integer remoteDataId) throws RemoteDataInstanceCreatorNotFoundException { + if (dataInstanceCreatorMapping.containsKey(remoteDataId)) { + return dataInstanceCreatorMapping.get(remoteDataId); + } else { + throw new RemoteDataInstanceCreatorNotFoundException("Could not found the instance creator of remote data id " + remoteDataId); + } + } +} diff --git a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteClient.java b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteClient.java index 8771cd7f6..150e1d9e9 100644 --- a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteClient.java +++ b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteClient.java @@ -26,7 +26,7 @@ import org.skywalking.apm.collector.core.data.Data; public interface RemoteClient extends Comparable { String getAddress(); - void send(int graphId, int nodeId, Data data); + void push(int graphId, int nodeId, Data data); boolean equals(String address); } diff --git a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteClientService.java b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteClientService.java index 109e9320f..c3e493da5 100644 --- a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteClientService.java +++ b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteClientService.java @@ -24,5 +24,5 @@ import org.skywalking.apm.collector.core.module.Service; * @author peng-yongsheng */ public interface RemoteClientService extends Service { - RemoteClient create(String host, int port); + RemoteClient create(String host, int port, int channelSize, int bufferSize); } diff --git a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/DataReceiver.java b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataIDGetter.java similarity index 85% rename from apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/DataReceiver.java rename to apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataIDGetter.java index 7baeb3478..04912f3eb 100644 --- a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/DataReceiver.java +++ b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataIDGetter.java @@ -23,8 +23,6 @@ import org.skywalking.apm.collector.core.data.Data; /** * @author peng-yongsheng */ -public interface DataReceiver { - Output output(int graphId, int nodeId); - - void receive(Output data); +public interface RemoteDataIDGetter { + Integer getRemoteDataId(Class dataClass) throws RemoteDataMappingIdNotFoundException; } diff --git a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteServerService.java b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataInstanceCreatorGetter.java similarity index 79% rename from apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteServerService.java rename to apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataInstanceCreatorGetter.java index 17b6febf3..f27e626ec 100644 --- a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteServerService.java +++ b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataInstanceCreatorGetter.java @@ -18,11 +18,10 @@ package org.skywalking.apm.collector.remote.service; -import org.skywalking.apm.collector.core.module.Service; - /** * @author peng-yongsheng */ -public interface RemoteServerService extends Service { - void registerReceiver(DataReceiver receiver); +public interface RemoteDataInstanceCreatorGetter { + RemoteDataRegisterService.RemoteDataInstanceCreator getInstanceCreator( + Integer remoteDataId) throws RemoteDataInstanceCreatorNotFoundException; } diff --git a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/DataReceiverRegisterListener.java b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataInstanceCreatorNotFoundException.java similarity index 75% rename from apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/DataReceiverRegisterListener.java rename to apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataInstanceCreatorNotFoundException.java index dcc9b4e3e..4a3f418ff 100644 --- a/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/DataReceiverRegisterListener.java +++ b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataInstanceCreatorNotFoundException.java @@ -18,18 +18,13 @@ package org.skywalking.apm.collector.remote.service; +import org.skywalking.apm.collector.remote.RemoteException; + /** * @author peng-yongsheng */ -public class DataReceiverRegisterListener { - - private DataReceiver dataReceiver; - - public DataReceiver getDataReceiver() { - return dataReceiver; - } - - public void setDataReceiver(DataReceiver dataReceiver) { - this.dataReceiver = dataReceiver; +public class RemoteDataInstanceCreatorNotFoundException extends RemoteException { + public RemoteDataInstanceCreatorNotFoundException(String message) { + super(message); } } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferConfig.java b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataMappingIdNotFoundException.java similarity index 72% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferConfig.java rename to apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataMappingIdNotFoundException.java index b27b7290f..ea1c82932 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/skywalking/apm/collector/agent/stream/buffer/SegmentBufferConfig.java +++ b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataMappingIdNotFoundException.java @@ -16,13 +16,15 @@ * Project repository: https://github.com/OpenSkywalking/skywalking */ -package org.skywalking.apm.collector.agent.stream.buffer; +package org.skywalking.apm.collector.remote.service; -import org.skywalking.apm.collector.core.config.SystemConfig; +import org.skywalking.apm.collector.remote.RemoteException; /** * @author peng-yongsheng */ -public class SegmentBufferConfig { - public static String BUFFER_PATH = SystemConfig.DATA_PATH + "/buffer/"; +public class RemoteDataMappingIdNotFoundException extends RemoteException { + public RemoteDataMappingIdNotFoundException(String message) { + super(message); + } } diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/config/SystemConfigParser.java b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataRegisterService.java similarity index 63% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/config/SystemConfigParser.java rename to apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataRegisterService.java index aa90354dd..e6dfe74bb 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/config/SystemConfigParser.java +++ b/apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/skywalking/apm/collector/remote/service/RemoteDataRegisterService.java @@ -16,21 +16,18 @@ * Project repository: https://github.com/OpenSkywalking/skywalking */ -package org.skywalking.apm.collector.core.config; +package org.skywalking.apm.collector.remote.service; -import org.skywalking.apm.collector.core.util.StringUtils; +import org.skywalking.apm.collector.core.data.Data; +import org.skywalking.apm.collector.core.module.Service; /** * @author peng-yongsheng */ -public enum SystemConfigParser { - INSTANCE; +public interface RemoteDataRegisterService extends Service { + void register(Class dataClass, RemoteDataInstanceCreator instanceCreator); - private static final String DATA_PATH = "data.path"; - - public void parse() { - if (!StringUtils.isEmpty(System.getProperty(DATA_PATH))) { - SystemConfig.DATA_PATH = System.getProperty(DATA_PATH); - } + interface RemoteDataInstanceCreator { + RemoteData createInstance(String id); } } diff --git a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/pom.xml b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/pom.xml index 6fb3137e1..1f3cc9054 100644 --- a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/pom.xml +++ b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/pom.xml @@ -41,6 +41,11 @@ collector-grpc-manager-define ${project.version} + + org.skywalking + apm-datacarrier + ${project.version} + diff --git a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/RemoteModuleGRPCProvider.java b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/RemoteModuleGRPCProvider.java index 3493cd7f4..01c815637 100644 --- a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/RemoteModuleGRPCProvider.java +++ b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/RemoteModuleGRPCProvider.java @@ -12,10 +12,9 @@ import org.skywalking.apm.collector.grpc.manager.service.GRPCManagerService; import org.skywalking.apm.collector.remote.RemoteModule; import org.skywalking.apm.collector.remote.grpc.handler.RemoteCommonServiceHandler; import org.skywalking.apm.collector.remote.grpc.service.GRPCRemoteSenderService; -import org.skywalking.apm.collector.remote.grpc.service.GRPCRemoteServerService; -import org.skywalking.apm.collector.remote.service.DataReceiverRegisterListener; +import org.skywalking.apm.collector.remote.service.CommonRemoteDataRegisterService; +import org.skywalking.apm.collector.remote.service.RemoteDataRegisterService; import org.skywalking.apm.collector.remote.service.RemoteSenderService; -import org.skywalking.apm.collector.remote.service.RemoteServerService; import org.skywalking.apm.collector.server.Server; /** @@ -24,10 +23,14 @@ import org.skywalking.apm.collector.server.Server; public class RemoteModuleGRPCProvider extends ModuleProvider { public static final String NAME = "gRPC"; + private static final String HOST = "host"; private static final String PORT = "port"; - private final DataReceiverRegisterListener listener = new DataReceiverRegisterListener(); + private static final String CHANNEL_SIZE = "channel_size"; + private static final String BUFFER_SIZE = "buffer_size"; + private GRPCRemoteSenderService remoteSenderService; + private CommonRemoteDataRegisterService remoteDataRegisterService; @Override public String name() { return NAME; @@ -40,10 +43,13 @@ public class RemoteModuleGRPCProvider extends ModuleProvider { @Override public void prepare(Properties config) throws ServiceNotProvidedException { String host = config.getProperty(HOST); Integer port = (Integer)config.get(PORT); + Integer channelSize = (Integer)config.getOrDefault(CHANNEL_SIZE, 5); + Integer bufferSize = (Integer)config.getOrDefault(BUFFER_SIZE, 1000); - remoteSenderService = new GRPCRemoteSenderService(host, port); - this.registerServiceImplementation(RemoteServerService.class, new GRPCRemoteServerService(listener)); + remoteDataRegisterService = new CommonRemoteDataRegisterService(); + remoteSenderService = new GRPCRemoteSenderService(host, port, channelSize, bufferSize, remoteDataRegisterService); this.registerServiceImplementation(RemoteSenderService.class, remoteSenderService); + this.registerServiceImplementation(RemoteDataRegisterService.class, remoteDataRegisterService); } @Override public void start(Properties config) throws ServiceNotProvidedException { @@ -52,7 +58,7 @@ public class RemoteModuleGRPCProvider extends ModuleProvider { GRPCManagerService managerService = getManager().find(GRPCManagerModule.NAME).getService(GRPCManagerService.class); Server gRPCServer = managerService.createIfAbsent(host, port); - gRPCServer.addHandler(new RemoteCommonServiceHandler(listener)); + gRPCServer.addHandler(new RemoteCommonServiceHandler(remoteDataRegisterService)); ModuleRegisterService moduleRegisterService = getManager().find(ClusterModule.NAME).getService(ModuleRegisterService.class); moduleRegisterService.register(RemoteModule.NAME, this.name(), new RemoteModuleGRPCRegistration(host, port)); diff --git a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/handler/RemoteCommonServiceHandler.java b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/handler/RemoteCommonServiceHandler.java index e1a5f47fc..06c8ac65d 100644 --- a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/handler/RemoteCommonServiceHandler.java +++ b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/handler/RemoteCommonServiceHandler.java @@ -20,12 +20,16 @@ package org.skywalking.apm.collector.remote.grpc.handler; import io.grpc.stub.StreamObserver; import org.skywalking.apm.collector.core.data.Data; +import org.skywalking.apm.collector.core.graph.GraphManager; +import org.skywalking.apm.collector.core.graph.Next; +import org.skywalking.apm.collector.core.util.Const; import org.skywalking.apm.collector.remote.grpc.proto.Empty; import org.skywalking.apm.collector.remote.grpc.proto.RemoteCommonServiceGrpc; import org.skywalking.apm.collector.remote.grpc.proto.RemoteData; import org.skywalking.apm.collector.remote.grpc.proto.RemoteMessage; import org.skywalking.apm.collector.remote.grpc.service.GRPCRemoteDeserializeService; -import org.skywalking.apm.collector.remote.service.DataReceiverRegisterListener; +import org.skywalking.apm.collector.remote.service.RemoteDataInstanceCreatorGetter; +import org.skywalking.apm.collector.remote.service.RemoteDataInstanceCreatorNotFoundException; import org.skywalking.apm.collector.server.grpc.GRPCHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -37,24 +41,31 @@ public class RemoteCommonServiceHandler extends RemoteCommonServiceGrpc.RemoteCo private final Logger logger = LoggerFactory.getLogger(RemoteCommonServiceHandler.class); - private final DataReceiverRegisterListener listener; + private final RemoteDataInstanceCreatorGetter instanceCreatorGetter; private final GRPCRemoteDeserializeService service; - public RemoteCommonServiceHandler(DataReceiverRegisterListener listener) { - this.listener = listener; + public RemoteCommonServiceHandler(RemoteDataInstanceCreatorGetter instanceCreatorGetter) { + this.instanceCreatorGetter = instanceCreatorGetter; this.service = new GRPCRemoteDeserializeService(); } + @SuppressWarnings("unchecked") @Override public StreamObserver call(StreamObserver responseObserver) { return new StreamObserver() { @Override public void onNext(RemoteMessage message) { int graphId = message.getGraphId(); int nodeId = message.getNodeId(); + int remoteDataId = message.getRemoteDataId(); RemoteData remoteData = message.getRemoteData(); - Data output = listener.getDataReceiver().output(graphId, nodeId); - service.deserialize(remoteData, output); - listener.getDataReceiver().receive(output); + try { + Data output = instanceCreatorGetter.getInstanceCreator(remoteDataId).createInstance(Const.EMPTY_STRING); + service.deserialize(remoteData, output); + Next next = GraphManager.INSTANCE.findGraph(graphId).toFinder().findNext(nodeId); + next.execute(output); + } catch (RemoteDataInstanceCreatorNotFoundException e) { + logger.error(e.getMessage(), e); + } } @Override public void onError(Throwable throwable) { @@ -62,6 +73,7 @@ public class RemoteCommonServiceHandler extends RemoteCommonServiceGrpc.RemoteCo } @Override public void onCompleted() { + responseObserver.onNext(Empty.newBuilder().build()); responseObserver.onCompleted(); } }; diff --git a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteClient.java b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteClient.java index 979e4c71e..f69e8763c 100644 --- a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteClient.java +++ b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteClient.java @@ -19,36 +19,146 @@ package org.skywalking.apm.collector.remote.grpc.service; import io.grpc.stub.StreamObserver; +import java.util.List; +import org.skywalking.apm.collector.client.grpc.GRPCClient; import org.skywalking.apm.collector.core.data.Data; +import org.skywalking.apm.collector.remote.grpc.proto.Empty; +import org.skywalking.apm.collector.remote.grpc.proto.RemoteCommonServiceGrpc; import org.skywalking.apm.collector.remote.grpc.proto.RemoteMessage; import org.skywalking.apm.collector.remote.service.RemoteClient; +import org.skywalking.apm.collector.remote.service.RemoteDataIDGetter; +import org.skywalking.apm.collector.remote.service.RemoteDataMappingIdNotFoundException; +import org.skywalking.apm.commons.datacarrier.DataCarrier; +import org.skywalking.apm.commons.datacarrier.buffer.BufferStrategy; +import org.skywalking.apm.commons.datacarrier.consumer.IConsumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * @author peng-yongsheng */ public class GRPCRemoteClient implements RemoteClient { - private final GRPCRemoteSerializeService service; - private final StreamObserver streamObserver; - private final String address; + private final Logger logger = LoggerFactory.getLogger(GRPCRemoteClient.class); - public GRPCRemoteClient(String host, int port, StreamObserver streamObserver) { - this.address = host + ":" + String.valueOf(port); - this.streamObserver = streamObserver; + private final GRPCRemoteSerializeService service; + private final GRPCClient client; + private final DataCarrier carrier; + private final String address; + private final RemoteDataIDGetter remoteDataIDGetter; + + GRPCRemoteClient(GRPCClient client, RemoteDataIDGetter remoteDataIDGetter, int channelSize, int bufferSize) { + this.address = client.toString(); + this.client = client; this.service = new GRPCRemoteSerializeService(); + this.remoteDataIDGetter = remoteDataIDGetter; + this.carrier = new DataCarrier<>(channelSize, bufferSize); + this.carrier.setBufferStrategy(BufferStrategy.BLOCKING); + this.carrier.consume(new RemoteMessageConsumer(), 1); } @Override public final String getAddress() { return this.address; } - @Override public void send(int graphId, int nodeId, Data data) { - RemoteMessage.Builder builder = RemoteMessage.newBuilder(); - builder.setGraphId(graphId); - builder.setNodeId(nodeId); - builder.setRemoteData(service.serialize(data)); + @Override public void push(int graphId, int nodeId, Data data) { + try { + Integer remoteDataId = remoteDataIDGetter.getRemoteDataId(data.getClass()); + RemoteMessage.Builder builder = RemoteMessage.newBuilder(); + builder.setGraphId(graphId); + builder.setNodeId(nodeId); + builder.setRemoteDataId(remoteDataId); + builder.setRemoteData(service.serialize(data)); - streamObserver.onNext(builder.build()); + this.carrier.produce(builder.build()); + logger.debug("put remote message into queue, id: {}", data.getId()); + } catch (RemoteDataMappingIdNotFoundException e) { + logger.error(e.getMessage(), e); + } + } + + class RemoteMessageConsumer implements IConsumer { + @Override public void init() { + } + + @Override public void consume(List remoteMessages) { + StreamObserver streamObserver = createStreamObserver(); + for (RemoteMessage remoteMessage : remoteMessages) { + streamObserver.onNext(remoteMessage); + } + streamObserver.onCompleted(); + } + + @Override public void onError(List remoteMessages, Throwable t) { + logger.error(t.getMessage(), t); + } + + @Override public void onExit() { + } + } + + private StreamObserver createStreamObserver() { + RemoteCommonServiceGrpc.RemoteCommonServiceStub stub = RemoteCommonServiceGrpc.newStub(client.getChannel()); + + StreamStatus status = new StreamStatus(false); + return stub.call(new StreamObserver() { + @Override public void onNext(Empty empty) { + } + + @Override public void onError(Throwable throwable) { + logger.error(throwable.getMessage(), throwable); + } + + @Override public void onCompleted() { + status.finished(); + } + }); + } + + class StreamStatus { + + private final Logger logger = LoggerFactory.getLogger(StreamStatus.class); + + private volatile boolean status; + + StreamStatus(boolean status) { + this.status = status; + } + + public boolean isFinish() { + return status; + } + + void finished() { + this.status = true; + } + + /** + * @param maxTimeout max wait time, milliseconds. + */ + public void wait4Finish(long maxTimeout) { + long time = 0; + while (!status) { + if (time > maxTimeout) { + break; + } + try2Sleep(5); + time += 5; + } + } + + /** + * Try to sleep, and ignore the {@link InterruptedException} + * + * @param millis the length of time to sleep in milliseconds + */ + private void try2Sleep(long millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException e) { + logger.error(e.getMessage(), e); + } + } } @Override public boolean equals(String address) { diff --git a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteClientService.java b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteClientService.java index e98a1f103..69eddb461 100644 --- a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteClientService.java +++ b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteClientService.java @@ -18,14 +18,11 @@ package org.skywalking.apm.collector.remote.grpc.service; -import io.grpc.stub.StreamObserver; import org.skywalking.apm.collector.client.ClientException; import org.skywalking.apm.collector.client.grpc.GRPCClient; -import org.skywalking.apm.collector.remote.grpc.proto.Empty; -import org.skywalking.apm.collector.remote.grpc.proto.RemoteCommonServiceGrpc; -import org.skywalking.apm.collector.remote.grpc.proto.RemoteMessage; import org.skywalking.apm.collector.remote.service.RemoteClient; import org.skywalking.apm.collector.remote.service.RemoteClientService; +import org.skywalking.apm.collector.remote.service.RemoteDataIDGetter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,74 +33,19 @@ public class GRPCRemoteClientService implements RemoteClientService { private final Logger logger = LoggerFactory.getLogger(GRPCRemoteClientService.class); - @Override public RemoteClient create(String host, int port) { + private final RemoteDataIDGetter remoteDataIDGetter; + + GRPCRemoteClientService(RemoteDataIDGetter remoteDataIDGetter) { + this.remoteDataIDGetter = remoteDataIDGetter; + } + + @Override public RemoteClient create(String host, int port, int channelSize, int bufferSize) { GRPCClient client = new GRPCClient(host, port); try { client.initialize(); } catch (ClientException e) { - e.printStackTrace(); - } - RemoteCommonServiceGrpc.RemoteCommonServiceStub stub = RemoteCommonServiceGrpc.newStub(client.getChannel()); - StreamObserver streamObserver = createStreamObserver(stub); - return new GRPCRemoteClient(host, port, streamObserver); - } - - private StreamObserver createStreamObserver(RemoteCommonServiceGrpc.RemoteCommonServiceStub stub) { - StreamStatus status = new StreamStatus(false); - return stub.call(new StreamObserver() { - @Override public void onNext(Empty empty) { - } - - @Override public void onError(Throwable throwable) { - logger.error(throwable.getMessage(), throwable); - } - - @Override public void onCompleted() { - status.finished(); - } - }); - } - - class StreamStatus { - private volatile boolean status; - - public StreamStatus(boolean status) { - this.status = status; - } - - public boolean isFinish() { - return status; - } - - public void finished() { - this.status = true; - } - - /** - * @param maxTimeout max wait time, milliseconds. - */ - public void wait4Finish(long maxTimeout) { - long time = 0; - while (!status) { - if (time > maxTimeout) { - break; - } - try2Sleep(5); - time += 5; - } - } - - /** - * Try to sleep, and ignore the {@link InterruptedException} - * - * @param millis the length of time to sleep in milliseconds - */ - private void try2Sleep(long millis) { - try { - Thread.sleep(millis); - } catch (InterruptedException e) { - - } + logger.error(e.getMessage(), e); } + return new GRPCRemoteClient(client, remoteDataIDGetter, channelSize, bufferSize); } } diff --git a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteDeserializeService.java b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteDeserializeService.java index 5ffbb5421..1c6d39b10 100644 --- a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteDeserializeService.java +++ b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteDeserializeService.java @@ -43,8 +43,5 @@ public class GRPCRemoteDeserializeService implements RemoteDeserializeService remoteClients; private final String selfAddress; private final HashCodeSelector hashCodeSelector; private final ForeverFirstSelector foreverFirstSelector; private final RollingSelector rollingSelector; + private final int channelSize; + private final int bufferSize; @Override public Mode send(int graphId, int nodeId, Data data, Selector selector) { RemoteClient remoteClient; @@ -66,18 +69,21 @@ public class GRPCRemoteSenderService extends ClusterModuleListener implements Re if (remoteClient.equals(selfAddress)) { return Mode.Local; } else { - remoteClient.send(graphId, nodeId, data); + remoteClient.push(graphId, nodeId, data); return Mode.Remote; } } - public GRPCRemoteSenderService(String host, int port) { - this.service = new GRPCRemoteClientService(); + public GRPCRemoteSenderService(String host, int port, int channelSize, int bufferSize, + RemoteDataIDGetter remoteDataIDGetter) { + this.service = new GRPCRemoteClientService(remoteDataIDGetter); this.remoteClients = new ArrayList<>(); this.selfAddress = host + ":" + String.valueOf(port); this.hashCodeSelector = new HashCodeSelector(); this.foreverFirstSelector = new ForeverFirstSelector(); this.rollingSelector = new RollingSelector(); + this.channelSize = channelSize; + this.bufferSize = bufferSize; } @Override public String path() { @@ -90,7 +96,7 @@ public class GRPCRemoteSenderService extends ClusterModuleListener implements Re String host = serverAddress.split(":")[0]; int port = Integer.parseInt(serverAddress.split(":")[1]); - RemoteClient remoteClient = service.create(host, port); + RemoteClient remoteClient = service.create(host, port, channelSize, bufferSize); newRemoteClients.add(remoteClient); Collections.sort(newRemoteClients); diff --git a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteSerializeService.java b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteSerializeService.java index dbe08251b..374e76696 100644 --- a/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteSerializeService.java +++ b/apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/skywalking/apm/collector/remote/grpc/service/GRPCRemoteSerializeService.java @@ -18,7 +18,6 @@ package org.skywalking.apm.collector.remote.grpc.service; -import com.google.protobuf.ByteString; import org.skywalking.apm.collector.core.data.Data; import org.skywalking.apm.collector.remote.grpc.proto.RemoteData; import org.skywalking.apm.collector.remote.service.RemoteSerializeService; @@ -31,22 +30,22 @@ public class GRPCRemoteSerializeService implements RemoteSerializeService extends Wo private final RemoteSenderService remoteSenderService; private final int graphId; - public RemoteWorkerRef(AbstractRemoteWorker remoteWorker, RemoteSenderService remoteSenderService, + RemoteWorkerRef(AbstractRemoteWorker remoteWorker, RemoteSenderService remoteSenderService, int graphId) { super(remoteWorker); this.remoteWorker = remoteWorker; diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/AggregationWorker.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/AggregationWorker.java index 2da917796..31e00ffe4 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/AggregationWorker.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/AggregationWorker.java @@ -63,7 +63,7 @@ public abstract class AggregationWorker throw new WorkerException(e.getMessage(), e); } } - dataCache.getLast().asMap().forEach((String id, Data data) -> { + dataCache.getLast().collection().forEach((String id, Data data) -> { logger.debug(data.toString()); onNext((OUTPUT)data); }); @@ -73,7 +73,7 @@ public abstract class AggregationWorker private void aggregate(INPUT message) { dataCache.writing(); if (dataCache.containsKey(message.getId())) { - message.mergeData(dataCache.get(message.getId())); + dataCache.get(message.getId()).mergeData(message); } else { dataCache.put(message.getId(), message); } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/PersistenceWorker.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/PersistenceWorker.java index c6e8063cd..33248e73c 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/PersistenceWorker.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/PersistenceWorker.java @@ -86,8 +86,8 @@ public abstract class PersistenceWorker } } - if (dataCache.getLast().asMap() != null) { - batchCollection = prepareBatch(dataCache.getLast().asMap()); + if (dataCache.getLast().collection() != null) { + batchCollection = prepareBatch(dataCache.getLast().collection()); } } finally { dataCache.finishReadingLast(); @@ -104,7 +104,7 @@ public abstract class PersistenceWorker if (ObjectUtils.isNotEmpty(dbData)) { dbData.mergeData(data); try { - updateBatchCollection.add(persistenceDAO().prepareBatchUpdate(data)); + updateBatchCollection.add(persistenceDAO().prepareBatchUpdate(dbData)); } catch (Throwable t) { logger.error(t.getMessage(), t); } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataCache.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataCache.java index b8877b084..855f2952d 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataCache.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataCache.java @@ -18,15 +18,20 @@ package org.skywalking.apm.collector.stream.worker.impl.data; +import org.skywalking.apm.collector.core.cache.Window; import org.skywalking.apm.collector.core.data.Data; /** * @author peng-yongsheng */ -public class DataCache extends Window { +public class DataCache extends Window { private DataCollection lockedDataCollection; + @Override public DataCollection collectionInstance() { + return new DataCollection(); + } + public boolean containsKey(String id) { return lockedDataCollection.containsKey(id); } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataCollection.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataCollection.java index eff0e468c..843ac385d 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataCollection.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataCollection.java @@ -20,12 +20,13 @@ package org.skywalking.apm.collector.stream.worker.impl.data; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import org.skywalking.apm.collector.core.cache.Collection; import org.skywalking.apm.collector.core.data.Data; /** * @author peng-yongsheng */ -public class DataCollection { +public class DataCollection implements Collection> { private Map data; private volatile boolean writing; private volatile boolean reading; @@ -40,23 +41,23 @@ public class DataCollection { writing = false; } - public void writing() { + @Override public void writing() { writing = true; } - public boolean isWriting() { + @Override public boolean isWriting() { return writing; } - public void finishReading() { + @Override public void finishReading() { reading = false; } - public void reading() { + @Override public void reading() { reading = true; } - public boolean isReading() { + @Override public boolean isReading() { return reading; } @@ -72,15 +73,15 @@ public class DataCollection { return data.get(key); } - public int size() { + @Override public int size() { return data.size(); } - public void clear() { + @Override public void clear() { data.clear(); } - public Map asMap() { + public Map collection() { return data; } } diff --git a/apm-collector/pom.xml b/apm-collector/pom.xml index 47914d4d8..45b42ec42 100644 --- a/apm-collector/pom.xml +++ b/apm-collector/pom.xml @@ -70,11 +70,6 @@ log4j-core 2.9.0 - - com.google.guava - guava - 22.0 - org.apache.logging.log4j log4j-slf4j-impl diff --git a/apm-commons/apm-datacarrier/src/main/java/org/skywalking/apm/commons/datacarrier/consumer/ConsumerThread.java b/apm-commons/apm-datacarrier/src/main/java/org/skywalking/apm/commons/datacarrier/consumer/ConsumerThread.java index 17473a4d5..231bafa95 100644 --- a/apm-commons/apm-datacarrier/src/main/java/org/skywalking/apm/commons/datacarrier/consumer/ConsumerThread.java +++ b/apm-commons/apm-datacarrier/src/main/java/org/skywalking/apm/commons/datacarrier/consumer/ConsumerThread.java @@ -91,12 +91,14 @@ public class ConsumerThread extends Thread { consumeList.add(element); } hasData = true; - } - try { - consumer.consume(consumeList); - } catch (Throwable t) { - consumer.onError(consumeList, t); + + if (consumeList.size() > 0) { + try { + consumer.consume(consumeList); + } catch (Throwable t) { + consumer.onError(consumeList, t); + } } return hasData; } diff --git a/apm-network/pom.xml b/apm-network/pom.xml index 34d12090f..f2b789728 100644 --- a/apm-network/pom.xml +++ b/apm-network/pom.xml @@ -31,8 +31,7 @@ UTF-8 - 1.4.0 - 4.1.12.Final + 1.7.0 1.6 @@ -41,16 +40,6 @@ io.grpc grpc-netty ${grpc.version} - - - io.netty - netty-codec-http2 - - - io.netty - netty-handler-proxy - - io.grpc @@ -62,16 +51,6 @@ grpc-stub ${grpc.version} - - io.netty - netty-codec-http2 - ${netty.version} - - - io.netty - netty-handler-proxy - ${netty.version} - @@ -104,7 +83,7 @@ com.google.protobuf:protoc:3.3.0:exe:${os.detected.classifier} grpc-java - io.grpc:protoc-gen-grpc-java:1.4.0:exe:${os.detected.classifier} + io.grpc:protoc-gen-grpc-java:1.7.0:exe:${os.detected.classifier} diff --git a/apm-sniffer/apm-agent-core/pom.xml b/apm-sniffer/apm-agent-core/pom.xml index 095cede21..86b18ec99 100644 --- a/apm-sniffer/apm-agent-core/pom.xml +++ b/apm-sniffer/apm-agent-core/pom.xml @@ -35,7 +35,7 @@ UTF-8 9.4.2.v20170220 - 1.4.0 + 1.7.0 1.7.6 org.skywalking.apm.dependencies @@ -105,7 +105,7 @@ io.grpc grpc-testing - 1.4.0 + ${grpc.version} mockito-core diff --git a/apm-sniffer/apm-sdk-plugin/pom.xml b/apm-sniffer/apm-sdk-plugin/pom.xml index c513de7cd..fe2335214 100644 --- a/apm-sniffer/apm-sdk-plugin/pom.xml +++ b/apm-sniffer/apm-sdk-plugin/pom.xml @@ -164,9 +164,12 @@ - - + +