Delete service and node reference entity, rename service and node reference... (167dc818) · Commits · SUMMER2020 / students / proj-2005254

apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/handler/TraceSegmentServletHandler.java

+1 −0

Original line number	Diff line number	Diff line
		@@ -43,6 +43,7 @@ public class TraceSegmentServletHandler extends JettyHandler {

		private void read(BufferedReader bufferedReader) throws IOException {
		JsonReader reader = new JsonReader(bufferedReader);

		reader.beginArray();
		while (reader.hasNext()) {
		SegmentParse segmentParse = new SegmentParse();

apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/reference/NodeRefAggregationWorker.java→apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/NodeReferenceAggregationWorker.java

+10 −10

Original line number	Diff line number	Diff line
		package org.skywalking.apm.collector.agentstream.worker.noderef.reference;
		package org.skywalking.apm.collector.agentstream.worker.noderef;

		import org.skywalking.apm.collector.storage.define.noderef.NodeRefDataDefine;
		import org.skywalking.apm.collector.storage.define.noderef.NodeReferenceDataDefine;
		import org.skywalking.apm.collector.stream.worker.AbstractLocalAsyncWorkerProvider;
		import org.skywalking.apm.collector.stream.worker.ClusterWorkerContext;
		import org.skywalking.apm.collector.stream.worker.ProviderNotFoundException;
		@@ -15,9 +15,9 @@ import org.skywalking.apm.collector.stream.worker.selector.WorkerSelector;
		/**
		* @author pengys5
		*/
		public class NodeRefAggregationWorker extends AggregationWorker {
		public class NodeReferenceAggregationWorker extends AggregationWorker {

		public NodeRefAggregationWorker(Role role, ClusterWorkerContext clusterContext) {
		public NodeReferenceAggregationWorker(Role role, ClusterWorkerContext clusterContext) {
		super(role, clusterContext);
		}

		@@ -26,18 +26,18 @@ public class NodeRefAggregationWorker extends AggregationWorker {
		}

		@Override protected WorkerRefs nextWorkRef(String id) throws WorkerNotFoundException {
		return getClusterContext().lookup(NodeRefRemoteWorker.WorkerRole.INSTANCE);
		return getClusterContext().lookup(NodeReferenceRemoteWorker.WorkerRole.INSTANCE);
		}

		public static class Factory extends AbstractLocalAsyncWorkerProvider<NodeRefAggregationWorker> {
		public static class Factory extends AbstractLocalAsyncWorkerProvider<NodeReferenceAggregationWorker> {
		@Override
		public Role role() {
		return WorkerRole.INSTANCE;
		}

		@Override
		public NodeRefAggregationWorker workerInstance(ClusterWorkerContext clusterContext) {
		return new NodeRefAggregationWorker(role(), clusterContext);
		public NodeReferenceAggregationWorker workerInstance(ClusterWorkerContext clusterContext) {
		return new NodeReferenceAggregationWorker(role(), clusterContext);
		}

		@Override
		@@ -51,7 +51,7 @@ public class NodeRefAggregationWorker extends AggregationWorker {

		@Override
		public String roleName() {
		return NodeRefAggregationWorker.class.getSimpleName();
		return NodeReferenceAggregationWorker.class.getSimpleName();
		}

		@Override
		@@ -60,7 +60,7 @@ public class NodeRefAggregationWorker extends AggregationWorker {
		}

		@Override public DataDefine dataDefine() {
		return new NodeRefDataDefine();
		return new NodeReferenceDataDefine();
		}
		}
		}

apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/reference/NodeRefPersistenceWorker.java→apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/NodeReferencePersistenceWorker.java

+11 −11

Original line number	Diff line number	Diff line
		package org.skywalking.apm.collector.agentstream.worker.noderef.reference;
		package org.skywalking.apm.collector.agentstream.worker.noderef;

		import org.skywalking.apm.collector.agentstream.worker.noderef.reference.dao.INodeReferenceDAO;
		import org.skywalking.apm.collector.storage.define.noderef.NodeRefDataDefine;
		import org.skywalking.apm.collector.agentstream.worker.noderef.dao.INodeReferenceDAO;
		import org.skywalking.apm.collector.storage.dao.DAOContainer;
		import org.skywalking.apm.collector.storage.define.DataDefine;
		import org.skywalking.apm.collector.storage.define.noderef.NodeReferenceDataDefine;
		import org.skywalking.apm.collector.stream.worker.AbstractLocalAsyncWorkerProvider;
		import org.skywalking.apm.collector.stream.worker.ClusterWorkerContext;
		import org.skywalking.apm.collector.stream.worker.ProviderNotFoundException;
		import org.skywalking.apm.collector.stream.worker.Role;
		import org.skywalking.apm.collector.stream.worker.impl.PersistenceWorker;
		import org.skywalking.apm.collector.stream.worker.impl.dao.IPersistenceDAO;
		import org.skywalking.apm.collector.storage.define.DataDefine;
		import org.skywalking.apm.collector.stream.worker.selector.HashCodeSelector;
		import org.skywalking.apm.collector.stream.worker.selector.WorkerSelector;

		/**
		* @author pengys5
		*/
		public class NodeRefPersistenceWorker extends PersistenceWorker {
		public class NodeReferencePersistenceWorker extends PersistenceWorker {

		public NodeRefPersistenceWorker(Role role, ClusterWorkerContext clusterContext) {
		public NodeReferencePersistenceWorker(Role role, ClusterWorkerContext clusterContext) {
		super(role, clusterContext);
		}

		@@ -34,15 +34,15 @@ public class NodeRefPersistenceWorker extends PersistenceWorker {
		return (IPersistenceDAO)DAOContainer.INSTANCE.get(INodeReferenceDAO.class.getName());
		}

		public static class Factory extends AbstractLocalAsyncWorkerProvider<NodeRefPersistenceWorker> {
		public static class Factory extends AbstractLocalAsyncWorkerProvider<NodeReferencePersistenceWorker> {
		@Override
		public Role role() {
		return WorkerRole.INSTANCE;
		}

		@Override
		public NodeRefPersistenceWorker workerInstance(ClusterWorkerContext clusterContext) {
		return new NodeRefPersistenceWorker(role(), clusterContext);
		public NodeReferencePersistenceWorker workerInstance(ClusterWorkerContext clusterContext) {
		return new NodeReferencePersistenceWorker(role(), clusterContext);
		}

		@Override
		@@ -56,7 +56,7 @@ public class NodeRefPersistenceWorker extends PersistenceWorker {

		@Override
		public String roleName() {
		return NodeRefPersistenceWorker.class.getSimpleName();
		return NodeReferencePersistenceWorker.class.getSimpleName();
		}

		@Override
		@@ -65,7 +65,7 @@ public class NodeRefPersistenceWorker extends PersistenceWorker {
		}

		@Override public DataDefine dataDefine() {
		return new NodeRefDataDefine();
		return new NodeReferenceDataDefine();
		}
		}
		}

apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/summary/NodeRefSumRemoteWorker.java→apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/NodeReferenceRemoteWorker.java

+10 −10

Original line number	Diff line number	Diff line
		package org.skywalking.apm.collector.agentstream.worker.noderef.summary;
		package org.skywalking.apm.collector.agentstream.worker.noderef;

		import org.skywalking.apm.collector.storage.define.noderef.NodeRefSumDataDefine;
		import org.skywalking.apm.collector.storage.define.noderef.NodeReferenceDataDefine;
		import org.skywalking.apm.collector.stream.worker.AbstractRemoteWorker;
		import org.skywalking.apm.collector.stream.worker.AbstractRemoteWorkerProvider;
		import org.skywalking.apm.collector.stream.worker.ClusterWorkerContext;
		@@ -14,9 +14,9 @@ import org.skywalking.apm.collector.stream.worker.selector.WorkerSelector;
		/**
		* @author pengys5
		*/
		public class NodeRefSumRemoteWorker extends AbstractRemoteWorker {
		public class NodeReferenceRemoteWorker extends AbstractRemoteWorker {

		protected NodeRefSumRemoteWorker(Role role, ClusterWorkerContext clusterContext) {
		protected NodeReferenceRemoteWorker(Role role, ClusterWorkerContext clusterContext) {
		super(role, clusterContext);
		}

		@@ -25,18 +25,18 @@ public class NodeRefSumRemoteWorker extends AbstractRemoteWorker {
		}

		@Override protected void onWork(Object message) throws WorkerException {
		getClusterContext().lookup(NodeRefSumPersistenceWorker.WorkerRole.INSTANCE).tell(message);
		getClusterContext().lookup(NodeReferencePersistenceWorker.WorkerRole.INSTANCE).tell(message);
		}

		public static class Factory extends AbstractRemoteWorkerProvider<NodeRefSumRemoteWorker> {
		public static class Factory extends AbstractRemoteWorkerProvider<NodeReferenceRemoteWorker> {
		@Override
		public Role role() {
		return WorkerRole.INSTANCE;
		}

		@Override
		public NodeRefSumRemoteWorker workerInstance(ClusterWorkerContext clusterContext) {
		return new NodeRefSumRemoteWorker(role(), clusterContext);
		public NodeReferenceRemoteWorker workerInstance(ClusterWorkerContext clusterContext) {
		return new NodeReferenceRemoteWorker(role(), clusterContext);
		}
		}

		@@ -45,7 +45,7 @@ public class NodeRefSumRemoteWorker extends AbstractRemoteWorker {

		@Override
		public String roleName() {
		return NodeRefSumRemoteWorker.class.getSimpleName();
		return NodeReferenceRemoteWorker.class.getSimpleName();
		}

		@Override
		@@ -54,7 +54,7 @@ public class NodeRefSumRemoteWorker extends AbstractRemoteWorker {
		}

		@Override public DataDefine dataDefine() {
		return new NodeRefSumDataDefine();
		return new NodeReferenceDataDefine();
		}
		}
		}

apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/summary/NodeRefSumSpanListener.java→apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/NodeReferenceSpanListener.java

+13 −13

Original line number	Diff line number	Diff line
		package org.skywalking.apm.collector.agentstream.worker.noderef.summary;
		package org.skywalking.apm.collector.agentstream.worker.noderef;

		import java.util.ArrayList;
		import java.util.List;
		@@ -10,7 +10,7 @@ import org.skywalking.apm.collector.agentstream.worker.segment.RefsListener;
		import org.skywalking.apm.collector.core.framework.CollectorContextHelper;
		import org.skywalking.apm.collector.core.util.Const;
		import org.skywalking.apm.collector.core.util.TimeBucketUtils;
		import org.skywalking.apm.collector.storage.define.noderef.NodeRefSumDataDefine;
		import org.skywalking.apm.collector.storage.define.noderef.NodeReferenceDataDefine;
		import org.skywalking.apm.collector.stream.StreamModuleContext;
		import org.skywalking.apm.collector.stream.StreamModuleGroupDefine;
		import org.skywalking.apm.collector.stream.worker.WorkerInvokeException;
		@@ -23,13 +23,13 @@ import org.slf4j.LoggerFactory;
		/**
		* @author pengys5
		*/
		public class NodeRefSumSpanListener implements EntrySpanListener, ExitSpanListener, FirstSpanListener, RefsListener {
		public class NodeReferenceSpanListener implements EntrySpanListener, ExitSpanListener, FirstSpanListener, RefsListener {

		private final Logger logger = LoggerFactory.getLogger(NodeRefSumSpanListener.class);
		private final Logger logger = LoggerFactory.getLogger(NodeReferenceSpanListener.class);

		private List<NodeRefSumDataDefine.NodeReferenceSum> nodeExitReferences = new ArrayList<>();
		private List<NodeRefSumDataDefine.NodeReferenceSum> nodeEntryReferences = new ArrayList<>();
		private List<NodeRefSumDataDefine.NodeReferenceSum> nodeReferences = new ArrayList<>();
		private List<NodeReferenceDataDefine.NodeReferenceSum> nodeExitReferences = new ArrayList<>();
		private List<NodeReferenceDataDefine.NodeReferenceSum> nodeEntryReferences = new ArrayList<>();
		private List<NodeReferenceDataDefine.NodeReferenceSum> nodeReferences = new ArrayList<>();
		private long timeBucket;
		private boolean hasReference = false;
		private long startTime;
		@@ -38,7 +38,7 @@ public class NodeRefSumSpanListener implements EntrySpanListener, ExitSpanListen

		@Override
		public void parseExit(SpanObject spanObject, int applicationId, int applicationInstanceId, String segmentId) {
		NodeRefSumDataDefine.NodeReferenceSum referenceSum = new NodeRefSumDataDefine.NodeReferenceSum();
		NodeReferenceDataDefine.NodeReferenceSum referenceSum = new NodeReferenceDataDefine.NodeReferenceSum();
		referenceSum.setApplicationId(applicationId);
		referenceSum.setBehindApplicationId(spanObject.getPeerId());

		@@ -56,7 +56,7 @@ public class NodeRefSumSpanListener implements EntrySpanListener, ExitSpanListen

		@Override
		public void parseEntry(SpanObject spanObject, int applicationId, int applicationInstanceId, String segmentId) {
		NodeRefSumDataDefine.NodeReferenceSum referenceSum = new NodeRefSumDataDefine.NodeReferenceSum();
		NodeReferenceDataDefine.NodeReferenceSum referenceSum = new NodeReferenceDataDefine.NodeReferenceSum();
		referenceSum.setApplicationId(Const.USER_ID);
		referenceSum.setBehindApplicationId(applicationId);
		referenceSum.setBehindPeer(Const.EMPTY_STRING);
		@@ -66,7 +66,7 @@ public class NodeRefSumSpanListener implements EntrySpanListener, ExitSpanListen
		nodeEntryReferences.add(buildNodeRefSum(referenceSum, spanObject.getStartTime(), spanObject.getEndTime(), spanObject.getIsError()));
		}

		private NodeRefSumDataDefine.NodeReferenceSum buildNodeRefSum(NodeRefSumDataDefine.NodeReferenceSum referenceSum,
		private NodeReferenceDataDefine.NodeReferenceSum buildNodeRefSum(NodeReferenceDataDefine.NodeReferenceSum referenceSum,
		long startTime, long endTime, boolean isError) {
		long cost = endTime - startTime;
		if (cost <= 1000 && !isError) {
		@@ -96,7 +96,7 @@ public class NodeRefSumSpanListener implements EntrySpanListener, ExitSpanListen
		String segmentId) {
		int parentApplicationId = InstanceCache.get(reference.getParentApplicationInstanceId());

		NodeRefSumDataDefine.NodeReferenceSum referenceSum = new NodeRefSumDataDefine.NodeReferenceSum();
		NodeReferenceDataDefine.NodeReferenceSum referenceSum = new NodeReferenceDataDefine.NodeReferenceSum();
		referenceSum.setApplicationId(parentApplicationId);
		referenceSum.setBehindApplicationId(applicationId);
		referenceSum.setBehindPeer(Const.EMPTY_STRING);
		@@ -119,13 +119,13 @@ public class NodeRefSumSpanListener implements EntrySpanListener, ExitSpanListen
		});
		}

		for (NodeRefSumDataDefine.NodeReferenceSum referenceSum : nodeExitReferences) {
		for (NodeReferenceDataDefine.NodeReferenceSum referenceSum : nodeExitReferences) {
		referenceSum.setId(timeBucket + Const.ID_SPLIT + referenceSum.getId());
		referenceSum.setTimeBucket(timeBucket);

		try {
		logger.debug("send to node reference summary aggregation worker, id: {}", referenceSum.getId());
		context.getClusterWorkerContext().lookup(NodeRefSumAggregationWorker.WorkerRole.INSTANCE).tell(referenceSum.toData());
		context.getClusterWorkerContext().lookup(NodeReferenceAggregationWorker.WorkerRole.INSTANCE).tell(referenceSum.toData());
		} catch (WorkerInvokeException \| WorkerNotFoundException e) {
		logger.error(e.getMessage(), e);
		}