add trace

pom.xml

@@ -93,6 +93,7 @@
         <sonar.host.url>https://sonarcloud.io</sonar.host.url>
         <okhttp.version>4.12.0</okhttp.version>
         <jacoco.version>0.8.11</jacoco.version>
+        <skywalking.version>8.12.0</skywalking.version>
     </properties>
 
     <dependencyManagement>
@@ -277,6 +278,12 @@
             </dependency>
             <!--swagger end-->
 
+            <dependency>
+                <groupId>org.apache.skywalking</groupId>
+                <artifactId>apm-toolkit-trace</artifactId>
+                <version>${skywalking.version}</version>
+            </dependency>
+
             <!--log start-->
             <dependency>
                 <groupId>org.apache.logging.log4j</groupId>

rill-flow-dag/olympicene-traversal/pom.xml

@@ -81,6 +81,10 @@
             <groupId>io.opentelemetry</groupId>
             <artifactId>opentelemetry-sdk-extension-autoconfigure-spi</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>apm-toolkit-trace</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

rill-flow-dag/olympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/DAGOperations.java

@@ -44,6 +44,8 @@
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.tuple.Pair;
+import org.apache.skywalking.apm.toolkit.trace.ActiveSpan;
+import org.apache.skywalking.apm.toolkit.trace.Trace;
 
 import java.util.*;
 import java.util.concurrent.ExecutorService;
@@ -192,7 +194,11 @@ public void finishTaskAsync(String executionId, String taskCategory, NotifyInfo
         }));
     }
 
+    @Trace(operationName = "DAGOperations.finishTaskSync")
     public void finishTaskSync(String executionId, String taskCategory, NotifyInfo notifyInfo, Map<String, Object> output) {
+        ActiveSpan.tag("executionId", executionId);
+        ActiveSpan.tag("taskName", notifyInfo.getTaskInfoName());
+        ActiveSpan.tag("taskCategory", taskCategory);
         log.info("finishTask task begin to execute executionId:{} notifyInfo:{}", executionId, notifyInfo);
         Map<String, Object> params = Maps.newHashMap();
         params.put(EXECUTION_ID, executionId);
@@ -231,15 +237,20 @@ public void redoTask(String executionId, List<String> taskNames, Map<String, Obj
         dagTraversal.submitTraversal(executionId, null);
     }
 
+
+    @Trace(operationName = "DAGOperations.submitDAG")
     public void submitDAG(String executionId, DAG dag, DAGSettings settings, Map<String, Object> data, NotifyInfo notifyInfo) {
+        ActiveSpan.tag("executionId", executionId);
         log.info("submitDAG task begin to execute executionId:{} notifyInfo:{}", executionId, notifyInfo);
         ExecutionResult executionResult = dagRunner.submitDAG(executionId, dag, settings, data, notifyInfo);
         Optional.ofNullable(getTimeoutSeconds(new HashMap<>(), executionResult.getContext(), dag.getTimeline()))
                 .ifPresent(timeoutSeconds -> timeCheckRunner.addDAGToTimeoutCheck(executionId, timeoutSeconds));
         dagTraversal.submitTraversal(executionId, null);
     }
 
+    @Trace(operationName = "DAGOperations.finishDAG")
     public void finishDAG(String executionId, DAGInfo dagInfo, DAGStatus dagStatus, DAGInvokeMsg dagInvokeMsg) {
+        ActiveSpan.tag("executionId", executionId);
         log.info("finishDAG task begin to execute executionId:{} dagStatus:{}", executionId, dagStatus);
         Map<String, Object> params = Maps.newHashMap();
         params.put(EXECUTION_ID, executionId);

rill-flow-dag/olympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/Olympicene.java

@@ -35,6 +35,8 @@
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.skywalking.apm.toolkit.trace.ActiveSpan;
+import org.apache.skywalking.apm.toolkit.trace.Trace;
 
 import java.util.*;
 import java.util.concurrent.ExecutorService;
@@ -111,7 +113,9 @@ public void finish(String executionId, DAGSettings settings, Map<String, Object>
      * 唤醒suspense任务
      */
     @Override
+    @Trace(operationName = "wakeup suspense")
     public void wakeup(String executionId, Map<String, Object> data, NotifyInfo notifyInfo) {
+        ActiveSpan.tag("executionId", executionId);
         runNotify(executionId, NotifyType.WAKEUP, notifyInfo,
                 () -> {
                     Optional.ofNullable(notifyInfo.getTaskInfoName())

rill-flow-dag/olympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/helper/Stasher.java

@@ -16,9 +16,9 @@
 
 package com.weibo.rill.flow.olympicene.traversal.helper;
 
+import com.weibo.rill.flow.interfaces.model.task.TaskInfo;
 import com.weibo.rill.flow.olympicene.core.model.dag.DAGInfo;
 import com.weibo.rill.flow.olympicene.core.model.dag.DAGStatus;
-import com.weibo.rill.flow.interfaces.model.task.TaskInfo;
 import org.apache.commons.lang3.tuple.Pair;
 
 import java.util.Map;

rill-flow-dag/olympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/runners/AbstractTaskRunner.java

@@ -46,6 +46,8 @@
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.skywalking.apm.toolkit.trace.ActiveSpan;
+import org.apache.skywalking.apm.toolkit.trace.Trace;
 
 import java.util.*;
 
@@ -99,9 +101,13 @@ public void inputMappings(Map<String, Object> context, Map<String, Object> input
     }
 
     @Override
+    @Trace(operationName = "TaskRunner.run")
     public ExecutionResult run(String executionId, TaskInfo taskInfo, Map<String, Object> context) {
         try {
+            ActiveSpan.tag("taskName", taskInfo.getName());
+            ActiveSpan.tag("taskCategory", taskInfo.getTask().getCategory());
             if (needNormalSkip(executionId, taskInfo)) {
+                ActiveSpan.tag("taskStatus", "skipped current and following tasks");
                 skipCurrentAndFollowingTasks(executionId, taskInfo);
                 return ExecutionResult.builder().taskStatus(taskInfo.getTaskStatus()).build();
             }
@@ -130,11 +136,13 @@ public ExecutionResult run(String executionId, TaskInfo taskInfo, Map<String, Ob
             }
 
             updateProgressArgs(executionId, taskInfo, input);
+            ActiveSpan.tag("taskStatusBeforeRun", taskInfo.getTaskStatus().toString());
 
             ExecutionResult ret = doRun(executionId, taskInfo, input);
             if (MapUtils.isEmpty(ret.getInput())) {
                 ret.setInput(input);
             }
+            ActiveSpan.tag("taskStatusAfterRun", taskInfo.getTaskStatus().toString());
             return ret;
         } catch (Exception e) {
             log.warn("run task fails, executionId:{}, taskName:{}", executionId, taskInfo.getName(), e);
@@ -146,6 +154,7 @@ public ExecutionResult run(String executionId, TaskInfo taskInfo, Map<String, Ob
 
             boolean tolerance = Optional.ofNullable(taskInfo.getTask()).map(BaseTask::isTolerance).orElse(false);
             taskInfo.setTaskStatus(tolerance ? TaskStatus.SKIPPED : TaskStatus.FAILED);
+            ActiveSpan.tag("taskStatusAfterRun", taskInfo.getTaskStatus().toString());
 
             Map<String, TaskInfo> subTasks = taskInfo.getChildren();
             taskInfo.setChildren(new LinkedHashMap<>());

rill-flow-dag/olympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/runners/TaskRunner.java

@@ -16,9 +16,9 @@
 
 package com.weibo.rill.flow.olympicene.traversal.runners;
 
-import com.weibo.rill.flow.olympicene.core.model.NotifyInfo;
 import com.weibo.rill.flow.interfaces.model.mapping.Mapping;
 import com.weibo.rill.flow.interfaces.model.task.TaskInfo;
+import com.weibo.rill.flow.olympicene.core.model.NotifyInfo;
 import com.weibo.rill.flow.olympicene.core.model.task.ExecutionResult;
 
 import java.util.List;

rill-flow-dag/olympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/strategy/SimpleRetryPolicy.java

@@ -22,6 +22,7 @@
 import com.weibo.rill.flow.interfaces.model.task.TaskStatus;
 import com.weibo.rill.flow.olympicene.core.model.strategy.RetryContext;
 import com.weibo.rill.flow.olympicene.traversal.utils.ConditionsUtil;
+import org.apache.skywalking.apm.toolkit.trace.ActiveSpan;
 import org.springframework.util.CollectionUtils;
 
 import java.util.List;
@@ -41,6 +42,7 @@ public boolean needRetry(RetryContext context) {
                 .map(TaskInvokeMsg::getInvokeTimeInfos)
                 .map(List::size)
                 .orElse(1);
+        ActiveSpan.tag("invokeTimes", String.valueOf(invokeTimes + 1));
         int maxRetryTimes = Optional.ofNullable(context.getRetryConfig()).map(Retry::getMaxRetryTimes).orElse(0);
 
         return invokeTimes <= maxRetryTimes;

rill-flow-dag/pom.xml

@@ -64,6 +64,12 @@
                 <groupId>org.yaml</groupId>
                 <artifactId>snakeyaml</artifactId>
             </dependency>
+
+            <dependency>
+                <groupId>org.apache.skywalking</groupId>
+                <artifactId>apm-toolkit-trace</artifactId>
+                <version>8.12.0</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 </project>

rill-flow-service/pom.xml

@@ -160,5 +160,9 @@
             <groupId>io.micrometer</groupId>
             <artifactId>micrometer-registry-prometheus</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>apm-toolkit-trace</artifactId>
+        </dependency>
     </dependencies>
 </project>

rill-flow-service/src/main/java/com/weibo/rill/flow/service/facade/OlympiceneFacade.java

@@ -58,6 +58,7 @@
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.tuple.Pair;
+import org.apache.skywalking.apm.toolkit.trace.ActiveSpan;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;