diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/graph/TraceStreamGraph.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/graph/TraceStreamGraph.java index a988493b9..29118d530 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/graph/TraceStreamGraph.java +++ b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/graph/TraceStreamGraph.java @@ -20,11 +20,11 @@ package org.apache.skywalking.apm.collector.agent.stream.graph; import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SegmentStandardizationWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.application.ApplicationComponentPersistenceWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.application.ApplicationComponentRemoteWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.application.ApplicationMappingAggregationWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.segment.SegmentPersistenceWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceEntryRemoteWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationComponentPersistenceWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationComponentRemoteWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMappingAggregationWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment.SegmentPersistenceWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceEntryRemoteWorker; import org.apache.skywalking.apm.collector.core.graph.Graph; import org.apache.skywalking.apm.collector.core.graph.GraphManager; import org.apache.skywalking.apm.collector.queue.QueueModule; @@ -33,17 +33,17 @@ import org.apache.skywalking.apm.collector.storage.table.application.Application import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric; import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener; import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SegmentStandardization; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.application.ApplicationComponentAggregationWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.application.ApplicationMappingPersistenceWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.application.ApplicationMappingRemoteWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.global.GlobalTracePersistenceWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.instance.InstanceMetricPersistenceWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.segment.SegmentCostPersistenceWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceEntryAggregationWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceEntryPersistenceWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceReferenceMetricAggregationWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceReferenceMetricPersistenceWorker; -import org.apache.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceReferenceMetricRemoteWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationComponentAggregationWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMappingPersistenceWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMappingRemoteWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global.GlobalTracePersistenceWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance.InstanceMetricPersistenceWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment.SegmentCostPersistenceWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceEntryAggregationWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceEntryPersistenceWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceReferenceMetricAggregationWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceReferenceMetricPersistenceWorker; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceReferenceMetricRemoteWorker; import org.apache.skywalking.apm.collector.core.module.ModuleManager; import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; import org.apache.skywalking.apm.collector.remote.RemoteModule; diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationReferenceMetricSpanListener.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationReferenceMetricSpanListener.java deleted file mode 100644 index 5e02297b3..000000000 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationReferenceMetricSpanListener.java +++ /dev/null @@ -1,138 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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. - * - */ - - -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application; - -import java.util.LinkedList; -import java.util.List; -import org.apache.skywalking.apm.collector.agent.stream.parser.EntrySpanListener; -import org.apache.skywalking.apm.collector.agent.stream.parser.ExitSpanListener; -import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator; -import org.apache.skywalking.apm.collector.cache.CacheModule; -import org.apache.skywalking.apm.collector.cache.service.InstanceCacheService; -import org.apache.skywalking.apm.collector.configuration.ConfigurationModule; -import org.apache.skywalking.apm.collector.configuration.service.IApdexThresholdService; -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.core.util.ApdexThresholdUtils; -import org.apache.skywalking.apm.collector.core.util.CollectionUtils; -import org.apache.skywalking.apm.collector.core.util.Const; -import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils; -import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * @author peng-yongsheng - */ -public class ApplicationReferenceMetricSpanListener implements EntrySpanListener, ExitSpanListener { - - private final Logger logger = LoggerFactory.getLogger(ApplicationReferenceMetricSpanListener.class); - - private final IApdexThresholdService apdexThresholdService; - private final InstanceCacheService instanceCacheService; - private final List applicationReferenceMetrics; - private final List references; - - public ApplicationReferenceMetricSpanListener(ModuleManager moduleManager) { - this.applicationReferenceMetrics = new LinkedList<>(); - this.references = new LinkedList<>(); - this.instanceCacheService = moduleManager.find(CacheModule.NAME).getService(InstanceCacheService.class); - this.apdexThresholdService = moduleManager.find(ConfigurationModule.NAME).getService(IApdexThresholdService.class); - } - - @Override - public void parseExit(SpanDecorator spanDecorator, int applicationId, int instanceId, String segmentId) { - ApplicationReferenceMetric applicationReferenceMetric = new ApplicationReferenceMetric(Const.EMPTY_STRING); - applicationReferenceMetric.setFrontApplicationId(applicationId); - applicationReferenceMetric.setBehindApplicationId(spanDecorator.getPeerId()); - applicationReferenceMetric.setTimeBucket(TimeBucketUtils.INSTANCE.getMinuteTimeBucket(spanDecorator.getStartTime())); - - String idBuilder = String.valueOf(applicationReferenceMetric.getTimeBucket()) + Const.ID_SPLIT + applicationId + - Const.ID_SPLIT + spanDecorator.getPeerId(); - - applicationReferenceMetric.setId(idBuilder); - applicationReferenceMetrics.add(buildApplicationRefSum(applicationReferenceMetric, spanDecorator.getStartTime(), spanDecorator.getEndTime(), spanDecorator.getIsError())); - } - - @Override - public void parseEntry(SpanDecorator spanDecorator, int applicationId, int instanceId, - String segmentId) { - if (CollectionUtils.isNotEmpty(references)) { - references.forEach(applicationReference -> { - applicationReference.setTimeBucket(TimeBucketUtils.INSTANCE.getMinuteTimeBucket(spanDecorator.getStartTime())); - String idBuilder = String.valueOf(applicationReference.getTimeBucket()) + Const.ID_SPLIT + applicationReference.getFrontApplicationId() + - Const.ID_SPLIT + applicationReference.getBehindApplicationId(); - - applicationReference.setId(idBuilder); - applicationReferenceMetrics.add(buildApplicationRefSum(applicationReference, spanDecorator.getStartTime(), spanDecorator.getEndTime(), spanDecorator.getIsError())); - }); - } else { - ApplicationReferenceMetric applicationReferenceMetric = new ApplicationReferenceMetric(Const.EMPTY_STRING); - applicationReferenceMetric.setFrontApplicationId(Const.USER_ID); - applicationReferenceMetric.setBehindApplicationId(applicationId); - applicationReferenceMetric.setTimeBucket(TimeBucketUtils.INSTANCE.getMinuteTimeBucket(spanDecorator.getStartTime())); - - String idBuilder = String.valueOf(applicationReferenceMetric.getTimeBucket()) + Const.ID_SPLIT + applicationReferenceMetric.getFrontApplicationId() + - Const.ID_SPLIT + applicationReferenceMetric.getBehindApplicationId(); - - applicationReferenceMetric.setId(idBuilder); - applicationReferenceMetrics.add(buildApplicationRefSum(applicationReferenceMetric, spanDecorator.getStartTime(), spanDecorator.getEndTime(), spanDecorator.getIsError())); - } - } - -// @Override public void parseRef(ReferenceDecorator referenceDecorator, int applicationId, int instanceId, -// String segmentId) { -// int parentApplicationId = instanceCacheService.get(referenceDecorator.getParentApplicationInstanceId()); -// -// ApplicationReferenceMetric referenceSum = new ApplicationReferenceMetric(Const.EMPTY_STRING); -// referenceSum.setFrontApplicationId(parentApplicationId); -// referenceSum.setBehindApplicationId(applicationId); -// references.add(referenceSum); -// } - - @Override public void build() { - logger.debug("node reference summary listener build"); -// Graph graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.APPLICATION_REFERENCE_METRIC_GRAPH_ID, ApplicationReferenceMetric.class); -// for (ApplicationReferenceMetric applicationReferenceMetric : applicationReferenceMetrics) { -// graph.start(applicationReferenceMetric); -// } - } - - private ApplicationReferenceMetric buildApplicationRefSum(ApplicationReferenceMetric reference, - long startTime, long endTime, boolean isError) { - long duration = endTime - startTime; - - reference.setTransactionCalls(1L); - reference.setTransactionDurationSum(duration); - if (isError) { - reference.setTransactionErrorCalls(1L); - reference.setTransactionErrorDurationSum(duration); - } - - ApdexThresholdUtils.Apdex apdex = ApdexThresholdUtils.compute(apdexThresholdService.getApplicationApdexThreshold(reference.getBehindApplicationId()), duration); - if (ApdexThresholdUtils.Apdex.Satisfied.equals(apdex)) { - reference.setSatisfiedCount(1L); - } else if (ApdexThresholdUtils.Apdex.Tolerating.equals(apdex)) { - reference.setToleratingCount(1L); - } else { - reference.setFrustratedCount(1L); - } - return reference; - } -} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceMetricSpanListener.java b/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceMetricSpanListener.java deleted file mode 100644 index 53e037abe..000000000 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceMetricSpanListener.java +++ /dev/null @@ -1,78 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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. - * - */ - - -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.instance; - -import org.apache.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph; -import org.apache.skywalking.apm.collector.agent.stream.parser.EntrySpanListener; -import org.apache.skywalking.apm.collector.agent.stream.parser.FirstSpanListener; -import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator; -import org.apache.skywalking.apm.collector.core.graph.Graph; -import org.apache.skywalking.apm.collector.core.graph.GraphManager; -import org.apache.skywalking.apm.collector.core.util.Const; -import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils; -import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * @author peng-yongsheng - */ -public class InstanceMetricSpanListener implements EntrySpanListener, FirstSpanListener { - - private final Logger logger = LoggerFactory.getLogger(InstanceMetricSpanListener.class); - - private int applicationId; - private int instanceId; - private boolean isError; - private long duration; - private long timeBucket; - - @Override - public void parseEntry(SpanDecorator spanDecorator, int applicationId, int instanceId, - String segmentId) { - } - - @Override - public void parseFirst(SpanDecorator spanDecorator, int applicationId, int instanceId, - String segmentId) { - this.applicationId = applicationId; - this.instanceId = instanceId; - this.isError = spanDecorator.getIsError(); - this.duration = spanDecorator.getEndTime() - spanDecorator.getStartTime(); - timeBucket = TimeBucketUtils.INSTANCE.getSecondTimeBucket(spanDecorator.getStartTime()); - } - - @Override public void build() { - InstanceMetric instanceMetric = new InstanceMetric(timeBucket + Const.ID_SPLIT + instanceId); - instanceMetric.setApplicationId(applicationId); - instanceMetric.setInstanceId(instanceId); - instanceMetric.setTransactionCalls(1L); - instanceMetric.setTransactionDurationSum(duration); - - if (isError) { - instanceMetric.setTransactionErrorCalls(1L); - instanceMetric.setTransactionErrorDurationSum(duration); - } - instanceMetric.setTimeBucket(timeBucket); - - Graph graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.INSTANCE_METRIC_GRAPH_ID, InstanceMetric.class); - graph.start(instanceMetric); - } -} diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/GraphIdDefine.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/GraphIdDefine.java new file mode 100644 index 000000000..356742db4 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/GraphIdDefine.java @@ -0,0 +1,38 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.define.graph; + +/** + * @author peng-yongsheng + */ +public class GraphIdDefine { + public static final int SERVICE_REFERENCE_METRIC_GRAPH_ID = 400; + public static final int INSTANCE_REFERENCE_METRIC_GRAPH_ID = 401; + public static final int APPLICATION_REFERENCE_METRIC_GRAPH_ID = 402; + + public static final int SERVICE_METRIC_GRAPH_ID = 403; + public static final int INSTANCE_METRIC_GRAPH_ID = 404; + public static final int APPLICATION_METRIC_GRAPH_ID = 405; + + public static final int APPLICATION_COMPONENT_GRAPH_ID = 406; + public static final int APPLICATION_MAPPING_GRAPH_ID = 407; + public static final int SERVICE_ENTRY_GRAPH_ID = 408; + public static final int GLOBAL_TRACE_GRAPH_ID = 409; + public static final int SEGMENT_COST_GRAPH_ID = 410; +} diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/WorkerIdDefine.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/WorkerIdDefine.java new file mode 100644 index 000000000..e2877d83a --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/WorkerIdDefine.java @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.define.graph; + +/** + * @author peng-yongsheng + */ +public class WorkerIdDefine { + public static final int SERVICE_REFERENCE_METRIC_AGGREGATION_WORKER_ID = 400; + public static final int SERVICE_REFERENCE_METRIC_REMOTE_WORKER_ID = 401; + public static final int SERVICE_REFERENCE_METRIC_PERSISTENCE_WORKER_ID = 402; + + public static final int INSTANCE_REFERENCE_METRIC_AGGREGATION_WORKER_ID = 403; + public static final int INSTANCE_REFERENCE_METRIC_REMOTE_WORKER_ID = 404; + public static final int INSTANCE_REFERENCE_METRIC_PERSISTENCE_WORKER_ID = 405; + + public static final int APPLICATION_REFERENCE_METRIC_AGGREGATION_WORKER_ID = 406; + public static final int APPLICATION_REFERENCE_METRIC_REMOTE_WORKER_ID = 407; + public static final int APPLICATION_REFERENCE_METRIC_PERSISTENCE_WORKER_ID = 408; + + public static final int SERVICE_METRIC_AGGREGATION_WORKER_ID = 409; + public static final int SERVICE_METRIC_REMOTE_WORKER_ID = 410; + public static final int SERVICE_METRIC_PERSISTENCE_WORKER_ID = 411; + + public static final int INSTANCE_METRIC_AGGREGATION_WORKER_ID = 412; + public static final int INSTANCE_METRIC_REMOTE_WORKER_ID = 413; + public static final int INSTANCE_METRIC_PERSISTENCE_WORKER_ID = 414; + + public static final int APPLICATION_METRIC_AGGREGATION_WORKER_ID = 415; + public static final int APPLICATION_METRIC_REMOTE_WORKER_ID = 416; + public static final int APPLICATION_METRIC_PERSISTENCE_WORKER_ID = 417; + + public static final int APPLICATION_COMPONENT_AGGREGATION_WORKER_ID = 418; + public static final int APPLICATION_COMPONENT_REMOTE_WORKER_ID = 419; + public static final int APPLICATION_COMPONENT_PERSISTENCE_WORKER_ID = 420; + + public static final int APPLICATION_MAPPING_AGGREGATION_WORKER_ID = 421; + public static final int APPLICATION_MAPPING_REMOTE_WORKER_ID = 422; + public static final int APPLICATION_MAPPING_PERSISTENCE_WORKER_ID = 423; + + public static final int SERVICE_ENTRY_AGGREGATION_WORKER_ID = 424; + public static final int SERVICE_ENTRY_REMOTE_WORKER_ID = 425; + public static final int SERVICE_ENTRY_PERSISTENCE_WORKER_ID = 426; + + public static final int GLOBAL_TRACE_PERSISTENCE_WORKER_ID = 427; + public static final int SEGMENT_COST_PERSISTENCE_WORKER_ID = 428; + + public static final int INSTANCE_REFERENCE_GRAPH_BRIDGE_WORKER_ID = 429; + public static final int APPLICATION_REFERENCE_GRAPH_BRIDGE_WORKER_ID = 430; + public static final int SERVICE_METRIC_GRAPH_BRIDGE_WORKER_ID = 431; + public static final int INSTANCE_METRIC_GRAPH_BRIDGE_WORKER_ID = 432; +} diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/pom.xml b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/pom.xml index 65083dd29..da1815f39 100644 --- a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/pom.xml +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/pom.xml @@ -36,6 +36,26 @@ metric-define ${project.version} + + org.apache.skywalking + segment-parser-define + ${project.version} + + + org.apache.skywalking + collector-configuration-define + ${project.version} + + + org.apache.skywalking + collector-cache-define + ${project.version} + + + org.apache.skywalking + collector-storage-define + ${project.version} + org.apache.skywalking analysis-worker-model diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/AnalysisMetricModuleProvider.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/AnalysisMetricModuleProvider.java index c78d76532..5d8dbb930 100644 --- a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/AnalysisMetricModuleProvider.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/AnalysisMetricModuleProvider.java @@ -20,6 +20,25 @@ package org.apache.skywalking.apm.collector.analysis.metric.provider; import java.util.Properties; import org.apache.skywalking.apm.collector.analysis.metric.define.AnalysisMetricModule; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationComponentGraph; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationComponentSpanListener; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMappingGraph; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMappingSpanListener; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMetricGraph; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationReferenceMetricGraph; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global.GlobalTraceGraph; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global.GlobalTraceSpanListener; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance.InstanceMetricGraph; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance.InstanceReferenceMetricGraph; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment.SegmentCostGraph; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment.SegmentCostSpanListener; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceEntryGraph; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceEntrySpanListener; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceMetricGraph; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceReferenceMetricGraph; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceReferenceMetricSpanListener; +import org.apache.skywalking.apm.collector.analysis.segment.parser.define.AnalysisSegmentParserModule; +import org.apache.skywalking.apm.collector.analysis.segment.parser.define.service.ISegmentParserListenerRegister; import org.apache.skywalking.apm.collector.core.module.Module; import org.apache.skywalking.apm.collector.core.module.ModuleProvider; import org.apache.skywalking.apm.collector.core.module.ServiceNotProvidedException; @@ -43,7 +62,40 @@ public class AnalysisMetricModuleProvider extends ModuleProvider { } @Override public void start(Properties config) throws ServiceNotProvidedException { + segmentParserListenerRegister(); + ServiceReferenceMetricGraph serviceReferenceMetricGraph = new ServiceReferenceMetricGraph(getManager()); + serviceReferenceMetricGraph.create(); + + InstanceReferenceMetricGraph instanceReferenceMetricGraph = new InstanceReferenceMetricGraph(getManager()); + instanceReferenceMetricGraph.create(); + + ApplicationReferenceMetricGraph applicationReferenceMetricGraph = new ApplicationReferenceMetricGraph(getManager()); + applicationReferenceMetricGraph.create(); + + ServiceMetricGraph serviceMetricGraph = new ServiceMetricGraph(getManager()); + serviceMetricGraph.create(); + + InstanceMetricGraph instanceMetricGraph = new InstanceMetricGraph(getManager()); + instanceMetricGraph.create(); + + ApplicationMetricGraph applicationMetricGraph = new ApplicationMetricGraph(getManager()); + applicationMetricGraph.create(); + + ApplicationComponentGraph applicationComponentGraph = new ApplicationComponentGraph(getManager()); + applicationComponentGraph.create(); + + ApplicationMappingGraph applicationMappingGraph = new ApplicationMappingGraph(getManager()); + applicationMappingGraph.create(); + + ServiceEntryGraph serviceEntryGraph = new ServiceEntryGraph(getManager()); + serviceEntryGraph.create(); + + GlobalTraceGraph globalTraceGraph = new GlobalTraceGraph(getManager()); + globalTraceGraph.create(); + + SegmentCostGraph segmentCostGraph = new SegmentCostGraph(getManager()); + segmentCostGraph.create(); } @Override public void notifyAfterCompleted() throws ServiceNotProvidedException { @@ -53,4 +105,14 @@ public class AnalysisMetricModuleProvider extends ModuleProvider { @Override public String[] requiredModules() { return new String[0]; } + + private void segmentParserListenerRegister() { + ISegmentParserListenerRegister segmentParserListenerRegister = getManager().find(AnalysisSegmentParserModule.NAME).getService(ISegmentParserListenerRegister.class); + segmentParserListenerRegister.register(new ServiceReferenceMetricSpanListener()); + segmentParserListenerRegister.register(new ApplicationComponentSpanListener()); + segmentParserListenerRegister.register(new ApplicationMappingSpanListener()); + segmentParserListenerRegister.register(new ServiceEntrySpanListener(getManager())); + segmentParserListenerRegister.register(new GlobalTraceSpanListener()); + segmentParserListenerRegister.register(new SegmentCostSpanListener(getManager())); + } } diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-define/src/main/java/org/apache/skywalking/apm/collector/agent/stream/service/trace/MetricSource.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/MetricSource.java similarity index 92% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-define/src/main/java/org/apache/skywalking/apm/collector/agent/stream/service/trace/MetricSource.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/MetricSource.java index 01719985a..f65969627 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-define/src/main/java/org/apache/skywalking/apm/collector/agent/stream/service/trace/MetricSource.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/MetricSource.java @@ -17,7 +17,7 @@ */ -package org.apache.skywalking.apm.collector.agent.stream.service.trace; +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker; /** * @author peng-yongsheng diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationComponentAggregationWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentAggregationWorker.java similarity index 84% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationComponentAggregationWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentAggregationWorker.java index 21d113c89..6dfac7a4e 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationComponentAggregationWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentAggregationWorker.java @@ -16,14 +16,13 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application; - -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; -import org.apache.skywalking.apm.collector.storage.table.application.ApplicationComponent; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationComponent; /** * @author peng-yongsheng @@ -35,13 +34,13 @@ public class ApplicationComponentAggregationWorker extends AggregationWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ApplicationComponentAggregationWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentGraph.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentGraph.java new file mode 100644 index 000000000..2c2629fad --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentGraph.java @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.core.graph.GraphManager; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.RemoteModule; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationComponent; + +/** + * @author peng-yongsheng + */ +public class ApplicationComponentGraph { + + private final ModuleManager moduleManager; + + public ApplicationComponentGraph(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + public void create() { + RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class); + + GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_COMPONENT_GRAPH_ID, ApplicationComponent.class) + .addNode(new ApplicationComponentAggregationWorker.Factory(moduleManager).create(null)) + .addNext(new ApplicationComponentRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_COMPONENT_GRAPH_ID).create(null)) + .addNext(new ApplicationComponentPersistenceWorker.Factory(moduleManager).create(null)); + } +} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationComponentPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentPersistenceWorker.java similarity index 85% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationComponentPersistenceWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentPersistenceWorker.java index e0e71ca8e..a1e23a0c7 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationComponentPersistenceWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentPersistenceWorker.java @@ -16,17 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application; - +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; import org.apache.skywalking.apm.collector.storage.StorageModule; import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationComponentPersistenceDAO; import org.apache.skywalking.apm.collector.storage.table.application.ApplicationComponent; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; -import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; /** * @author peng-yongsheng @@ -38,7 +37,7 @@ public class ApplicationComponentPersistenceWorker extends PersistenceWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ApplicationComponentPersistenceWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationComponentRemoteWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentRemoteWorker.java similarity index 90% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationComponentRemoteWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentRemoteWorker.java index 2c46c1877..2190876dd 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationComponentRemoteWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentRemoteWorker.java @@ -16,16 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application; - -import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; import org.apache.skywalking.apm.collector.remote.service.Selector; import org.apache.skywalking.apm.collector.storage.table.application.ApplicationComponent; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; /** * @author peng-yongsheng @@ -37,7 +37,7 @@ public class ApplicationComponentRemoteWorker extends AbstractRemoteWorker graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.APPLICATION_COMPONENT_GRAPH_ID, ApplicationComponent.class); + Graph graph = GraphManager.INSTANCE.findGraph(GraphIdDefine.APPLICATION_COMPONENT_GRAPH_ID, ApplicationComponent.class); applicationComponents.forEach(applicationComponent -> { applicationComponent.setId(timeBucket + Const.ID_SPLIT + applicationComponent.getId()); diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMappingAggregationWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingAggregationWorker.java similarity index 84% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMappingAggregationWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingAggregationWorker.java index 47a681a2d..b56c99d04 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMappingAggregationWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingAggregationWorker.java @@ -16,14 +16,13 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application; - -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; -import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMapping; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMapping; /** * @author peng-yongsheng @@ -35,13 +34,13 @@ public class ApplicationMappingAggregationWorker extends AggregationWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ApplicationMappingAggregationWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingGraph.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingGraph.java new file mode 100644 index 000000000..a7265608b --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingGraph.java @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.core.graph.GraphManager; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.RemoteModule; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMapping; + +/** + * @author peng-yongsheng + */ +public class ApplicationMappingGraph { + + private final ModuleManager moduleManager; + + public ApplicationMappingGraph(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + public void create() { + RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class); + + GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_MAPPING_GRAPH_ID, ApplicationMapping.class) + .addNode(new ApplicationMappingAggregationWorker.Factory(moduleManager).create(null)) + .addNext(new ApplicationMappingRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_MAPPING_GRAPH_ID).create(null)) + .addNext(new ApplicationMappingPersistenceWorker.Factory(moduleManager).create(null)); + } +} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMappingPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingPersistenceWorker.java similarity index 85% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMappingPersistenceWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingPersistenceWorker.java index 7ae066df6..8e2d2ced2 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMappingPersistenceWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingPersistenceWorker.java @@ -16,17 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application; - +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; +import org.apache.skywalking.apm.collector.storage.StorageModule; import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationMappingPersistenceDAO; import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMapping; -import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; -import org.apache.skywalking.apm.collector.storage.StorageModule; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; /** * @author peng-yongsheng @@ -38,7 +37,7 @@ public class ApplicationMappingPersistenceWorker extends PersistenceWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ApplicationMappingPersistenceWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMappingRemoteWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingRemoteWorker.java similarity index 90% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMappingRemoteWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingRemoteWorker.java index 1f57022d8..9ac4faeec 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMappingRemoteWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingRemoteWorker.java @@ -16,16 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application; - +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; import org.apache.skywalking.apm.collector.core.module.ModuleManager; import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; -import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMapping; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider; import org.apache.skywalking.apm.collector.remote.service.Selector; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMapping; /** * @author peng-yongsheng @@ -37,7 +37,7 @@ public class ApplicationMappingRemoteWorker extends AbstractRemoteWorker graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.APPLICATION_MAPPING_GRAPH_ID, ApplicationMapping.class); + Graph graph = GraphManager.INSTANCE.findGraph(GraphIdDefine.APPLICATION_MAPPING_GRAPH_ID, ApplicationMapping.class); applicationMappings.forEach(applicationMapping -> { applicationMapping.setId(timeBucket + Const.ID_SPLIT + applicationMapping.getId()); applicationMapping.setTimeBucket(timeBucket); diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMetricAggregationWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricAggregationWorker.java similarity index 62% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMetricAggregationWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricAggregationWorker.java index 89ee41c5b..d55a3eca4 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMetricAggregationWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricAggregationWorker.java @@ -16,16 +16,15 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application; - -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.core.util.Const; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; -import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; -import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.core.util.Const; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric; /** * @author peng-yongsheng @@ -37,7 +36,7 @@ public class ApplicationMetricAggregationWorker extends AggregationWorker { - public Factory(ModuleManager moduleManager, - QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ApplicationMetricAggregationWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricGraph.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricGraph.java new file mode 100644 index 000000000..3de3f68df --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricGraph.java @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.core.graph.GraphManager; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.RemoteModule; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric; + +/** + * @author peng-yongsheng + */ +public class ApplicationMetricGraph { + + private final ModuleManager moduleManager; + + public ApplicationMetricGraph(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + public void create() { + RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class); + + GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_METRIC_GRAPH_ID, ApplicationReferenceMetric.class) + .addNode(new ApplicationMetricAggregationWorker.Factory(moduleManager).create(null)) + .addNext(new ApplicationMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_METRIC_GRAPH_ID).create(null)) + .addNext(new ApplicationMetricPersistenceWorker.Factory(moduleManager).create(null)); + } +} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMetricPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricPersistenceWorker.java similarity index 87% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMetricPersistenceWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricPersistenceWorker.java index 5e54b7c6e..bb17cc9ec 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMetricPersistenceWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricPersistenceWorker.java @@ -16,17 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application; - -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; -import org.apache.skywalking.apm.collector.storage.dao.IApplicationMetricPersistenceDAO; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; -import org.apache.skywalking.apm.collector.storage.StorageModule; -import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.storage.StorageModule; +import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.IApplicationMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; /** * @author peng-yongsheng @@ -38,7 +37,7 @@ public class ApplicationMetricPersistenceWorker extends PersistenceWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ApplicationMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMetricRemoteWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricRemoteWorker.java similarity index 90% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMetricRemoteWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricRemoteWorker.java index 0f68599a4..9748626ae 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMetricRemoteWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricRemoteWorker.java @@ -16,16 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application; - +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; import org.apache.skywalking.apm.collector.core.module.ModuleManager; import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; import org.apache.skywalking.apm.collector.remote.service.Selector; import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; /** * @author peng-yongsheng @@ -37,7 +37,7 @@ public class ApplicationMetricRemoteWorker extends AbstractRemoteWorker { private final InstanceCacheService instanceCacheService; + private final IApdexThresholdService apdexThresholdService; public ApplicationReferenceMetricAggregationWorker(ModuleManager moduleManager) { super(moduleManager); this.instanceCacheService = moduleManager.find(CacheModule.NAME).getService(InstanceCacheService.class); + this.apdexThresholdService = moduleManager.find(ConfigurationModule.NAME).getService(IApdexThresholdService.class); } @Override public int id() { - return ApplicationGraphNodeIdDefine.APPLICATION_REFERENCE_METRIC_AGGREGATION_NODE_ID; + return WorkerIdDefine.APPLICATION_REFERENCE_METRIC_AGGREGATION_WORKER_ID; } @Override protected ApplicationReferenceMetric transform(InstanceReferenceMetric instanceReferenceMetric) { @@ -71,14 +74,21 @@ public class ApplicationReferenceMetricAggregationWorker extends AggregationWork applicationReferenceMetric.setMqTransactionDurationSum(instanceReferenceMetric.getMqTransactionDurationSum()); applicationReferenceMetric.setMqTransactionErrorDurationSum(instanceReferenceMetric.getMqTransactionErrorDurationSum()); + ApdexThresholdUtils.Apdex apdex = ApdexThresholdUtils.compute(apdexThresholdService.getApplicationApdexThreshold(behindApplicationId), instanceReferenceMetric.getTransactionDurationSum()); + if (ApdexThresholdUtils.Apdex.Satisfied.equals(apdex)) { + applicationReferenceMetric.setSatisfiedCount(1L); + } else if (ApdexThresholdUtils.Apdex.Tolerating.equals(apdex)) { + applicationReferenceMetric.setToleratingCount(1L); + } else { + applicationReferenceMetric.setFrustratedCount(1L); + } return applicationReferenceMetric; } public static class Factory extends AbstractLocalAsyncWorkerProvider { - public Factory(ModuleManager moduleManager, - QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ApplicationReferenceMetricAggregationWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricGraph.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricGraph.java new file mode 100644 index 000000000..043c2d7b7 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricGraph.java @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.core.graph.Graph; +import org.apache.skywalking.apm.collector.core.graph.GraphManager; +import org.apache.skywalking.apm.collector.core.graph.Next; +import org.apache.skywalking.apm.collector.core.graph.NodeProcessor; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.RemoteModule; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric; + +/** + * @author peng-yongsheng + */ +public class ApplicationReferenceMetricGraph { + + private final ModuleManager moduleManager; + + public ApplicationReferenceMetricGraph(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + public void create() { + RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class); + + Graph graph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_REFERENCE_METRIC_GRAPH_ID, InstanceReferenceMetric.class); + + graph.addNode(new ApplicationReferenceMetricAggregationWorker.Factory(moduleManager).create(null)) + .addNext(new ApplicationReferenceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_REFERENCE_METRIC_GRAPH_ID).create(null)) + .addNext(new ApplicationReferenceMetricPersistenceWorker.Factory(moduleManager).create(null)); + + link(graph); + } + + private void link(Graph graph) { + GraphManager.INSTANCE.findGraph(GraphIdDefine.INSTANCE_REFERENCE_METRIC_GRAPH_ID, InstanceReferenceMetric.class) + .toFinder().findNode(WorkerIdDefine.INSTANCE_REFERENCE_METRIC_AGGREGATION_WORKER_ID, InstanceReferenceMetric.class) + .addNext(new NodeProcessor() { + @Override public int id() { + return WorkerIdDefine.APPLICATION_REFERENCE_GRAPH_BRIDGE_WORKER_ID; + } + + @Override public void process(InstanceReferenceMetric instanceReferenceMetric, + Next next) { + graph.start(instanceReferenceMetric); + } + }); + } +} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationReferenceMetricPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricPersistenceWorker.java similarity index 82% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationReferenceMetricPersistenceWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricPersistenceWorker.java index 3113ee19e..faa6127b2 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationReferenceMetricPersistenceWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricPersistenceWorker.java @@ -16,18 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application; - -import org.apache.skywalking.apm.collector.agent.stream.service.graph.ApplicationGraphNodeIdDefine; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; import org.apache.skywalking.apm.collector.storage.StorageModule; import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; -import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; /** * @author peng-yongsheng @@ -39,7 +37,7 @@ public class ApplicationReferenceMetricPersistenceWorker extends PersistenceWork } @Override public int id() { - return ApplicationGraphNodeIdDefine.APPLICATION_REFERENCE_METRIC_PERSISTENCE_NODE_ID; + return WorkerIdDefine.APPLICATION_REFERENCE_METRIC_PERSISTENCE_WORKER_ID; } @Override protected boolean needMergeDBData() { @@ -52,9 +50,8 @@ public class ApplicationReferenceMetricPersistenceWorker extends PersistenceWork public static class Factory extends AbstractLocalAsyncWorkerProvider { - public Factory(ModuleManager moduleManager, - QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ApplicationReferenceMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationReferenceMetricRemoteWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricRemoteWorker.java similarity index 90% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationReferenceMetricRemoteWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricRemoteWorker.java index ca0352664..51994a1ed 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationReferenceMetricRemoteWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricRemoteWorker.java @@ -16,15 +16,14 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application; - -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider; import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; -import org.apache.skywalking.apm.collector.agent.stream.service.graph.ApplicationGraphNodeIdDefine; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; import org.apache.skywalking.apm.collector.remote.service.Selector; import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric; @@ -38,7 +37,7 @@ public class ApplicationReferenceMetricRemoteWorker extends AbstractRemoteWorker } @Override public int id() { - return ApplicationGraphNodeIdDefine.APPLICATION_REFERENCE_METRIC_REMOTE_NODE_ID; + return WorkerIdDefine.APPLICATION_REFERENCE_METRIC_REMOTE_WORKER_ID; } @Override protected void onWork(ApplicationReferenceMetric applicationReferenceMetric) throws WorkerException { diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/GlobalTraceGraph.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/GlobalTraceGraph.java new file mode 100644 index 000000000..d193046fe --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/GlobalTraceGraph.java @@ -0,0 +1,41 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.core.graph.GraphManager; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.storage.table.global.GlobalTrace; + +/** + * @author peng-yongsheng + */ +public class GlobalTraceGraph { + + private final ModuleManager moduleManager; + + public GlobalTraceGraph(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + public void create() { + GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.GLOBAL_TRACE_GRAPH_ID, GlobalTrace.class) + .addNode(new GlobalTracePersistenceWorker.Factory(moduleManager).create(null)); + } +} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/global/GlobalTracePersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/GlobalTracePersistenceWorker.java similarity index 87% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/global/GlobalTracePersistenceWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/GlobalTracePersistenceWorker.java index 45f145274..ad4fdff7d 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/global/GlobalTracePersistenceWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/GlobalTracePersistenceWorker.java @@ -16,17 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.global; - +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; import org.apache.skywalking.apm.collector.storage.StorageModule; import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IGlobalTracePersistenceDAO; import org.apache.skywalking.apm.collector.storage.table.global.GlobalTrace; -import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; /** * @author peng-yongsheng @@ -38,7 +37,7 @@ public class GlobalTracePersistenceWorker extends PersistenceWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public GlobalTracePersistenceWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/global/GlobalTraceSpanListener.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/GlobalTraceSpanListener.java similarity index 82% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/global/GlobalTraceSpanListener.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/GlobalTraceSpanListener.java index 525b4590c..1a61d7dd3 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/global/GlobalTraceSpanListener.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/GlobalTraceSpanListener.java @@ -16,20 +16,19 @@ * */ - -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.global; +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global; import java.util.ArrayList; import java.util.List; -import org.apache.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph; -import org.apache.skywalking.apm.collector.agent.stream.parser.FirstSpanListener; -import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator.SpanDecorator; +import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.FirstSpanListener; +import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.GlobalTraceIdsListener; import org.apache.skywalking.apm.collector.core.graph.Graph; import org.apache.skywalking.apm.collector.core.graph.GraphManager; import org.apache.skywalking.apm.collector.core.util.Const; -import org.apache.skywalking.apm.collector.storage.table.global.GlobalTrace; -import org.apache.skywalking.apm.collector.agent.stream.parser.GlobalTraceIdsListener; import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils; +import org.apache.skywalking.apm.collector.storage.table.global.GlobalTrace; import org.apache.skywalking.apm.network.proto.UniqueId; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -67,7 +66,7 @@ public class GlobalTraceSpanListener implements FirstSpanListener, GlobalTraceId @Override public void build() { logger.debug("global trace listener build"); - Graph graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.GLOBAL_TRACE_GRAPH_ID, GlobalTrace.class); + Graph graph = GraphManager.INSTANCE.findGraph(GraphIdDefine.GLOBAL_TRACE_GRAPH_ID, GlobalTrace.class); for (String globalTraceId : globalTraceIds) { GlobalTrace globalTrace = new GlobalTrace(segmentId + Const.ID_SPLIT + globalTraceId); globalTrace.setGlobalTraceId(globalTraceId); diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricAggregationWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricAggregationWorker.java new file mode 100644 index 000000000..b95599a73 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricAggregationWorker.java @@ -0,0 +1,81 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.core.util.Const; +import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric; +import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric; + +/** + * @author peng-yongsheng + */ +public class InstanceMetricAggregationWorker extends AggregationWorker { + + public InstanceMetricAggregationWorker(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public int id() { + return WorkerIdDefine.INSTANCE_METRIC_AGGREGATION_WORKER_ID; + } + + @Override protected InstanceMetric transform(InstanceReferenceMetric instanceReferenceMetric) { + String id = instanceReferenceMetric.getTimeBucket() + Const.ID_SPLIT + instanceReferenceMetric.getBehindInstanceId(); + + InstanceMetric instanceMetric = new InstanceMetric(id); + instanceMetric.setInstanceId(instanceReferenceMetric.getBehindInstanceId()); + + instanceMetric.setTransactionCalls(instanceReferenceMetric.getTransactionCalls()); + instanceMetric.setTransactionErrorCalls(instanceReferenceMetric.getTransactionErrorCalls()); + instanceMetric.setTransactionDurationSum(instanceReferenceMetric.getTransactionDurationSum()); + instanceMetric.setTransactionErrorDurationSum(instanceReferenceMetric.getTransactionErrorDurationSum()); + + instanceMetric.setBusinessTransactionCalls(instanceReferenceMetric.getBusinessTransactionCalls()); + instanceMetric.setBusinessTransactionErrorCalls(instanceReferenceMetric.getBusinessTransactionErrorCalls()); + instanceMetric.setBusinessTransactionDurationSum(instanceReferenceMetric.getBusinessTransactionDurationSum()); + instanceMetric.setBusinessTransactionErrorDurationSum(instanceReferenceMetric.getBusinessTransactionErrorDurationSum()); + + instanceMetric.setMqTransactionCalls(instanceReferenceMetric.getMqTransactionCalls()); + instanceMetric.setMqTransactionErrorCalls(instanceReferenceMetric.getMqTransactionErrorCalls()); + instanceMetric.setMqTransactionDurationSum(instanceReferenceMetric.getMqTransactionDurationSum()); + instanceMetric.setMqTransactionErrorDurationSum(instanceReferenceMetric.getMqTransactionErrorDurationSum()); + + return instanceMetric; + } + + public static class Factory extends AbstractLocalAsyncWorkerProvider { + + public Factory(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public InstanceMetricAggregationWorker workerInstance(ModuleManager moduleManager) { + return new InstanceMetricAggregationWorker(moduleManager); + } + + @Override + public int queueSize() { + return 1024; + } + } +} diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricGraph.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricGraph.java new file mode 100644 index 000000000..f3a664a44 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricGraph.java @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.core.graph.Graph; +import org.apache.skywalking.apm.collector.core.graph.GraphManager; +import org.apache.skywalking.apm.collector.core.graph.Next; +import org.apache.skywalking.apm.collector.core.graph.NodeProcessor; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.RemoteModule; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric; + +/** + * @author peng-yongsheng + */ +public class InstanceMetricGraph { + + private final ModuleManager moduleManager; + + public InstanceMetricGraph(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + public void create() { + RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class); + + Graph graph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.INSTANCE_METRIC_GRAPH_ID, InstanceReferenceMetric.class); + + graph.addNode(new InstanceMetricAggregationWorker.Factory(moduleManager).create(null)) + .addNext(new InstanceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.INSTANCE_METRIC_GRAPH_ID).create(null)) + .addNext(new InstanceMetricPersistenceWorker.Factory(moduleManager).create(null)); + + link(graph); + } + + private void link(Graph graph) { + GraphManager.INSTANCE.findGraph(GraphIdDefine.INSTANCE_REFERENCE_METRIC_GRAPH_ID, InstanceReferenceMetric.class) + .toFinder().findNode(WorkerIdDefine.INSTANCE_REFERENCE_METRIC_AGGREGATION_WORKER_ID, InstanceReferenceMetric.class) + .addNext(new NodeProcessor() { + + @Override public int id() { + return WorkerIdDefine.INSTANCE_METRIC_GRAPH_BRIDGE_WORKER_ID; + } + + @Override + public void process(InstanceReferenceMetric instanceReferenceMetric, + Next next) { + graph.start(instanceReferenceMetric); + } + }); + } +} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceMetricPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricPersistenceWorker.java similarity index 87% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceMetricPersistenceWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricPersistenceWorker.java index 6b83b221a..7f154df42 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceMetricPersistenceWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricPersistenceWorker.java @@ -16,17 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.instance; - +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.storage.StorageModule; import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IInstanceMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; -import org.apache.skywalking.apm.collector.storage.StorageModule; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; -import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; /** * @author peng-yongsheng @@ -38,7 +37,7 @@ public class InstanceMetricPersistenceWorker extends PersistenceWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public InstanceMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricRemoteWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricRemoteWorker.java new file mode 100644 index 000000000..bd500340f --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricRemoteWorker.java @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.remote.service.Selector; +import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric; + +/** + * @author peng-yongsheng + */ +public class InstanceMetricRemoteWorker extends AbstractRemoteWorker { + + public InstanceMetricRemoteWorker(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public int id() { + return WorkerIdDefine.INSTANCE_METRIC_REMOTE_WORKER_ID; + } + + @Override public Selector selector() { + return Selector.HashCode; + } + + @Override protected void onWork(InstanceMetric instanceMetric) throws WorkerException { + onNext(instanceMetric); + } + + public static class Factory extends AbstractRemoteWorkerProvider { + + public Factory(ModuleManager moduleManager, RemoteSenderService remoteSenderService, int graphId) { + super(moduleManager, remoteSenderService, graphId); + } + + @Override public InstanceMetricRemoteWorker workerInstance(ModuleManager moduleManager) { + return new InstanceMetricRemoteWorker(moduleManager); + } + } +} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceReferenceMetricAggregationWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferenceMetricAggregationWorker.java similarity index 89% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceReferenceMetricAggregationWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferenceMetricAggregationWorker.java index 223cceccf..24a21e96f 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceReferenceMetricAggregationWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferenceMetricAggregationWorker.java @@ -16,17 +16,15 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.instance; - -import org.apache.skywalking.apm.collector.agent.stream.service.graph.InstanceGraphNodeIdDefine; -import org.apache.skywalking.apm.collector.core.util.Const; -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; -import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric; -import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.core.util.Const; +import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric; /** * @author peng-yongsheng @@ -38,7 +36,7 @@ public class InstanceReferenceMetricAggregationWorker extends AggregationWorker< } @Override public int id() { - return InstanceGraphNodeIdDefine.INSTANCE_REFERENCE_METRIC_AGGREGATION_NODE_ID; + return WorkerIdDefine.INSTANCE_REFERENCE_METRIC_AGGREGATION_WORKER_ID; } @Override protected InstanceReferenceMetric transform(ServiceReferenceMetric serviceReferenceMetric) { @@ -69,8 +67,8 @@ public class InstanceReferenceMetricAggregationWorker extends AggregationWorker< public static class Factory extends AbstractLocalAsyncWorkerProvider { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public InstanceReferenceMetricAggregationWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferenceMetricGraph.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferenceMetricGraph.java new file mode 100644 index 000000000..3a04086be --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferenceMetricGraph.java @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.core.graph.Graph; +import org.apache.skywalking.apm.collector.core.graph.GraphManager; +import org.apache.skywalking.apm.collector.core.graph.Next; +import org.apache.skywalking.apm.collector.core.graph.NodeProcessor; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.RemoteModule; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric; + +/** + * @author peng-yongsheng + */ +public class InstanceReferenceMetricGraph { + + private final ModuleManager moduleManager; + + public InstanceReferenceMetricGraph(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + public void create() { + RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class); + + Graph graph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.INSTANCE_REFERENCE_METRIC_GRAPH_ID, ServiceReferenceMetric.class); + + graph.addNode(new InstanceReferenceMetricAggregationWorker.Factory(moduleManager).create(null)) + .addNext(new InstanceReferenceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.INSTANCE_REFERENCE_METRIC_GRAPH_ID).create(null)) + .addNext(new InstanceReferencePersistenceWorker.Factory(moduleManager).create(null)); + + link(graph); + } + + private void link(Graph graph) { + GraphManager.INSTANCE.findGraph(GraphIdDefine.SERVICE_REFERENCE_METRIC_GRAPH_ID, ServiceReferenceMetric.class) + .toFinder().findNode(WorkerIdDefine.SERVICE_REFERENCE_METRIC_AGGREGATION_WORKER_ID, ServiceReferenceMetric.class) + .addNext(new NodeProcessor() { + @Override public int id() { + return WorkerIdDefine.INSTANCE_REFERENCE_GRAPH_BRIDGE_WORKER_ID; + } + + @Override + public void process(ServiceReferenceMetric serviceReferenceMetric, Next next) { + graph.start(serviceReferenceMetric); + } + }); + } +} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceReferenceMetricRemoteWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferenceMetricRemoteWorker.java similarity index 90% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceReferenceMetricRemoteWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferenceMetricRemoteWorker.java index 3ea3719e6..d21f44b6b 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceReferenceMetricRemoteWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferenceMetricRemoteWorker.java @@ -16,17 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.instance; - -import org.apache.skywalking.apm.collector.agent.stream.service.graph.InstanceGraphNodeIdDefine; -import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; -import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider; -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.remote.service.Selector; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider; import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.remote.service.Selector; +import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric; /** * @author peng-yongsheng @@ -38,7 +37,7 @@ public class InstanceReferenceMetricRemoteWorker extends AbstractRemoteWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public InstanceReferencePersistenceWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostGraph.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostGraph.java new file mode 100644 index 000000000..e02077649 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostGraph.java @@ -0,0 +1,41 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.core.graph.GraphManager; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.storage.table.segment.SegmentCost; + +/** + * @author peng-yongsheng + */ +public class SegmentCostGraph { + + private final ModuleManager moduleManager; + + public SegmentCostGraph(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + public void create() { + GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SEGMENT_COST_GRAPH_ID, SegmentCost.class) + .addNode(new SegmentCostPersistenceWorker.Factory(moduleManager).create(null)); + } +} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/segment/SegmentCostPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostPersistenceWorker.java similarity index 87% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/segment/SegmentCostPersistenceWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostPersistenceWorker.java index cc4155513..a58565462 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/segment/SegmentCostPersistenceWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostPersistenceWorker.java @@ -16,17 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.segment; - -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; -import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostPersistenceDAO; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; -import org.apache.skywalking.apm.collector.storage.StorageModule; -import org.apache.skywalking.apm.collector.storage.table.segment.SegmentCost; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.storage.StorageModule; +import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.segment.SegmentCost; /** * @author peng-yongsheng @@ -38,7 +37,7 @@ public class SegmentCostPersistenceWorker extends PersistenceWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public SegmentCostPersistenceWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/segment/SegmentCostSpanListener.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostSpanListener.java similarity index 83% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/segment/SegmentCostSpanListener.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostSpanListener.java index fc9e91acb..502de5da7 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/segment/SegmentCostSpanListener.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostSpanListener.java @@ -16,17 +16,16 @@ * */ - -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.segment; +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment; import java.util.ArrayList; import java.util.List; -import org.apache.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph; -import org.apache.skywalking.apm.collector.agent.stream.parser.EntrySpanListener; -import org.apache.skywalking.apm.collector.agent.stream.parser.ExitSpanListener; -import org.apache.skywalking.apm.collector.agent.stream.parser.FirstSpanListener; -import org.apache.skywalking.apm.collector.agent.stream.parser.LocalSpanListener; -import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator.SpanDecorator; +import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.EntrySpanListener; +import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.ExitSpanListener; +import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.FirstSpanListener; +import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.LocalSpanListener; import org.apache.skywalking.apm.collector.cache.CacheModule; import org.apache.skywalking.apm.collector.cache.service.ServiceNameCacheService; import org.apache.skywalking.apm.collector.core.graph.Graph; @@ -95,7 +94,7 @@ public class SegmentCostSpanListener implements EntrySpanListener, ExitSpanListe } @Override public void build() { - Graph graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SEGMENT_COST_GRAPH_ID, SegmentCost.class); + Graph graph = GraphManager.INSTANCE.findGraph(GraphIdDefine.SEGMENT_COST_GRAPH_ID, SegmentCost.class); logger.debug("segment cost listener build"); for (SegmentCost segmentCost : segmentCosts) { segmentCost.setIsError(isError); diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntryAggregationWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryAggregationWorker.java similarity index 84% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntryAggregationWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryAggregationWorker.java index ca53cfd51..3296dad7c 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntryAggregationWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryAggregationWorker.java @@ -16,14 +16,13 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service; - -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; -import org.apache.skywalking.apm.collector.storage.table.service.ServiceEntry; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceEntry; /** * @author peng-yongsheng @@ -35,13 +34,13 @@ public class ServiceEntryAggregationWorker extends AggregationWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ServiceEntryAggregationWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryGraph.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryGraph.java new file mode 100644 index 000000000..41e925442 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryGraph.java @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.core.graph.GraphManager; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.RemoteModule; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceEntry; + +/** + * @author peng-yongsheng + */ +public class ServiceEntryGraph { + + private final ModuleManager moduleManager; + + public ServiceEntryGraph(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + public void create() { + RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class); + + GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SERVICE_ENTRY_GRAPH_ID, ServiceEntry.class) + .addNode(new ServiceEntryAggregationWorker.Factory(moduleManager).create(null)) + .addNext(new ServiceEntryRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_ENTRY_GRAPH_ID).create(null)) + .addNext(new ServiceEntryPersistenceWorker.Factory(moduleManager).create(null)); + } +} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntryPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryPersistenceWorker.java similarity index 87% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntryPersistenceWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryPersistenceWorker.java index d142d8dc5..419befdc7 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntryPersistenceWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryPersistenceWorker.java @@ -16,17 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service; - -import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; -import org.apache.skywalking.apm.collector.storage.StorageModule; -import org.apache.skywalking.apm.collector.storage.dao.IServiceEntryPersistenceDAO; -import org.apache.skywalking.apm.collector.storage.table.service.ServiceEntry; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.storage.StorageModule; +import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.IServiceEntryPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceEntry; /** * @author peng-yongsheng @@ -38,7 +37,7 @@ public class ServiceEntryPersistenceWorker extends PersistenceWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ServiceEntryPersistenceWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntryRemoteWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryRemoteWorker.java similarity index 90% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntryRemoteWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryRemoteWorker.java index f6630cff5..e16f4a7bd 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntryRemoteWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryRemoteWorker.java @@ -16,16 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service; - -import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; import org.apache.skywalking.apm.collector.remote.service.Selector; import org.apache.skywalking.apm.collector.storage.table.service.ServiceEntry; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; /** * @author peng-yongsheng @@ -37,7 +37,7 @@ public class ServiceEntryRemoteWorker extends AbstractRemoteWorker graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SERVICE_ENTRY_GRAPH_ID, ServiceEntry.class); + Graph graph = GraphManager.INSTANCE.findGraph(GraphIdDefine.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/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceMetricAggregationWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricAggregationWorker.java similarity index 61% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceMetricAggregationWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricAggregationWorker.java index 4af12b68c..a6a090af3 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceMetricAggregationWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricAggregationWorker.java @@ -16,16 +16,15 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service; - -import org.apache.skywalking.apm.collector.core.util.Const; -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; -import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; -import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.core.util.Const; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric; /** * @author peng-yongsheng @@ -37,7 +36,7 @@ public class ServiceMetricAggregationWorker extends AggregationWorker { - public Factory(ModuleManager moduleManager, - QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ServiceMetricAggregationWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricGraph.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricGraph.java new file mode 100644 index 000000000..00b6e377a --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricGraph.java @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.core.graph.Graph; +import org.apache.skywalking.apm.collector.core.graph.GraphManager; +import org.apache.skywalking.apm.collector.core.graph.Next; +import org.apache.skywalking.apm.collector.core.graph.NodeProcessor; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.RemoteModule; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric; + +/** + * @author peng-yongsheng + */ +public class ServiceMetricGraph { + + private final ModuleManager moduleManager; + + public ServiceMetricGraph(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + public void create() { + RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class); + + Graph graph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SERVICE_METRIC_GRAPH_ID, ServiceReferenceMetric.class); + + graph.addNode(new ServiceMetricAggregationWorker.Factory(moduleManager).create(null)) + .addNext(new ServiceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_METRIC_GRAPH_ID).create(null)) + .addNext(new ServiceMetricPersistenceWorker.Factory(moduleManager).create(null)); + + link(graph); + } + + private void link(Graph graph) { + GraphManager.INSTANCE.findGraph(GraphIdDefine.SERVICE_REFERENCE_METRIC_GRAPH_ID, ServiceReferenceMetric.class) + .toFinder().findNode(WorkerIdDefine.SERVICE_REFERENCE_METRIC_AGGREGATION_WORKER_ID, ServiceReferenceMetric.class) + .addNext(new NodeProcessor() { + @Override public int id() { + return WorkerIdDefine.SERVICE_METRIC_GRAPH_BRIDGE_WORKER_ID; + } + + @Override + public void process(ServiceReferenceMetric serviceReferenceMetric, Next next) { + graph.start(serviceReferenceMetric); + } + }); + } +} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceMetricPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricPersistenceWorker.java similarity index 87% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceMetricPersistenceWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricPersistenceWorker.java index 1f1858979..d6038a03a 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceMetricPersistenceWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricPersistenceWorker.java @@ -16,17 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service; - -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; -import org.apache.skywalking.apm.collector.storage.StorageModule; -import org.apache.skywalking.apm.collector.storage.dao.IServiceMetricPersistenceDAO; -import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.storage.StorageModule; +import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.IServiceMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; /** * @author peng-yongsheng @@ -38,7 +37,7 @@ public class ServiceMetricPersistenceWorker extends PersistenceWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ServiceMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceMetricRemoteWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricRemoteWorker.java similarity index 90% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceMetricRemoteWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricRemoteWorker.java index f26306f50..4552213c9 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceMetricRemoteWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricRemoteWorker.java @@ -16,16 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service; - -import org.apache.skywalking.apm.collector.core.module.ModuleManager; -import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; -import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider; import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; import org.apache.skywalking.apm.collector.remote.service.Selector; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; /** * @author peng-yongsheng @@ -37,7 +37,7 @@ public class ServiceMetricRemoteWorker extends AbstractRemoteWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ServiceReferenceMetricAggregationWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricGraph.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricGraph.java new file mode 100644 index 000000000..5e3ae2869 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricGraph.java @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.core.graph.GraphManager; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.RemoteModule; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric; + +/** + * @author peng-yongsheng + */ +public class ServiceReferenceMetricGraph { + + private final ModuleManager moduleManager; + + public ServiceReferenceMetricGraph(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + public void create() { + RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class); + + GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SERVICE_REFERENCE_METRIC_GRAPH_ID, ServiceReferenceMetric.class) + .addNode(new ServiceReferenceMetricAggregationWorker.Factory(moduleManager).create(null)) + .addNext(new ServiceReferenceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_REFERENCE_METRIC_GRAPH_ID).create(null)) + .addNext(new ServiceReferenceMetricAggregationWorker.Factory(moduleManager).create(null)); + } +} diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceReferenceMetricPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricPersistenceWorker.java similarity index 82% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceReferenceMetricPersistenceWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricPersistenceWorker.java index 2e41e6f71..167afe11f 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceReferenceMetricPersistenceWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricPersistenceWorker.java @@ -16,18 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service; - -import org.apache.skywalking.apm.collector.agent.stream.service.graph.ServiceGraphNodeIdDefine; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.storage.StorageModule; import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceReferenceMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric; -import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService; -import org.apache.skywalking.apm.collector.storage.StorageModule; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; -import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; /** * @author peng-yongsheng @@ -39,7 +37,7 @@ public class ServiceReferenceMetricPersistenceWorker extends PersistenceWorker { - public Factory(ModuleManager moduleManager, QueueCreatorService queueCreatorService) { - super(moduleManager, queueCreatorService); + public Factory(ModuleManager moduleManager) { + super(moduleManager); } @Override public ServiceReferenceMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { diff --git a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceReferenceMetricRemoteWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricRemoteWorker.java similarity index 90% rename from apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceReferenceMetricRemoteWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricRemoteWorker.java index 6970149d0..c582c8894 100644 --- a/apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceReferenceMetricRemoteWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricRemoteWorker.java @@ -16,17 +16,16 @@ * */ +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; -package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service; - -import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider; -import org.apache.skywalking.apm.collector.agent.stream.service.graph.ServiceGraphNodeIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; import org.apache.skywalking.apm.collector.remote.service.Selector; import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; -import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; /** * @author peng-yongsheng @@ -38,7 +37,7 @@ public class ServiceReferenceMetricRemoteWorker extends AbstractRemoteWorker graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SERVICE_REFERENCE_GRAPH_ID, ServiceReferenceMetric.class); + Graph graph = GraphManager.INSTANCE.findGraph(GraphIdDefine.SERVICE_REFERENCE_METRIC_GRAPH_ID, ServiceReferenceMetric.class); entryReferenceMetric.forEach(serviceReferenceMetric -> { String id = timeBucket + Const.ID_SPLIT + serviceReferenceMetric.getEntryServiceId() + Const.ID_SPLIT + serviceReferenceMetric.getFrontServiceId() + Const.ID_SPLIT + serviceReferenceMetric.getBehindServiceId(); diff --git a/apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/graph/GraphNodeFinder.java b/apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/graph/GraphNodeFinder.java index dd52a3f7d..1876072ed 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/graph/GraphNodeFinder.java +++ b/apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/graph/GraphNodeFinder.java @@ -16,7 +16,6 @@ * */ - package org.apache.skywalking.apm.collector.core.graph; import java.util.concurrent.ConcurrentHashMap; @@ -41,4 +40,23 @@ public class GraphNodeFinder { } return node.getNext(); } + + /** + * Find an exist node to build the graph. + * + * @param handlerId of specific node in graph. + * @param outputClass of the found node + * @param type of given output class + * @return Node instance. + */ + public Node findNode(int handlerId, Class outputClass) { + ConcurrentHashMap graphNodeIndex = graph.getNodeIndex(); + Node node = graphNodeIndex.get(handlerId); + if (node == null) { + throw new NodeNotFoundException("Can't find node with handlerId=" + + handlerId + + " in graph[" + graph.getId() + "]"); + } + return node; + } }