Add a forceIgnoring mechanism to the agent kernel (#689)

IgnoredTracerContext can propagate the unsampled flag into the following tracing context through the invalid snapshot.
apache · May 17, 2024 · b608d74 · b608d74
1 parent 4dfc1e8
commit b608d74
Show file tree

Hide file tree

Showing 16 changed files with 202 additions and 15 deletions.
diff --git a/CHANGES.md b/CHANGES.md
@@ -12,6 +12,7 @@ Release Notes.
 * Fix typos in `URLParser`.
 * Add support for `Derby`/`Sybase`/`SQLite`/`DB2`/`OceanBase` jdbc url format in `URLParser`.
 * Optimize spring-plugins:scheduled-annotation-plugin compatibility about Spring 6.1.x support.
+* Add a forceIgnoring mechanism in a CROSS_THREAD scenario.
 
 All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/213?closed=1)
 

diff --git a/...ore/src/main/java/org/apache/skywalking/apm/agent/core/context/AbstractTracerContext.java b/...ore/src/main/java/org/apache/skywalking/apm/agent/core/context/AbstractTracerContext.java
@@ -138,4 +138,10 @@ public interface AbstractTracerContext {
      * Get current primary endpoint name
      */
     String getPrimaryEndpointName();
+
+    /**
+     *  Change the current context to be in ignoring status.
+     */
+    AbstractTracerContext forceIgnoring();
+
 }
diff --git a/...agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/ContextManager.java b/...agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/ContextManager.java
@@ -163,7 +163,14 @@ public static void continued(ContextSnapshot snapshot) {
             throw new IllegalArgumentException("ContextSnapshot can't be null.");
         }
         if (!snapshot.isFromCurrent()) {
-            get().continued(snapshot);
+            // Invalid snapshot is only created by {@link IgnoredTracerContext#capture()}.
+            // When the snapshot is not valid, need to force ignoring the current tracing context.
+            if (snapshot.isValid()) {
+                get().continued(snapshot);
+            } else {
+                AbstractTracerContext context = get().forceIgnoring();
+                CONTEXT.set(context);
+            }
         }
     }
 

diff --git a/...core/src/main/java/org/apache/skywalking/apm/agent/core/context/IgnoredTracerContext.java b/...core/src/main/java/org/apache/skywalking/apm/agent/core/context/IgnoredTracerContext.java
@@ -47,6 +47,13 @@ public IgnoredTracerContext() {
         this.profileStatusContext = ProfileStatusContext.createWithNone();
     }
 
+    public IgnoredTracerContext(int stackDepth) {
+        this.stackDepth = stackDepth;
+        this.correlationContext = new CorrelationContext();
+        this.extensionContext = new ExtensionContext();
+        this.profileStatusContext = ProfileStatusContext.createWithNone();
+    }
+
     @Override
     public void inject(ContextCarrier carrier) {
         this.correlationContext.inject(carrier);
@@ -134,6 +141,11 @@ public String getPrimaryEndpointName() {
         return null;
     }
 
+    @Override
+    public AbstractTracerContext forceIgnoring() {
+        return this;
+    }
+
     public static class ListenerManager {
         private static List<IgnoreTracerContextListener> LISTENERS = new LinkedList<>();
 

diff --git a/...agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/TracingContext.java b/...agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/TracingContext.java
@@ -419,6 +419,14 @@ public String getPrimaryEndpointName() {
         return primaryEndpoint.getName();
     }
 
+    @Override
+    public AbstractTracerContext forceIgnoring() {
+        for (AbstractSpan span: activeSpanStack) {
+            span.forceIgnoring();
+        }
+        return new IgnoredTracerContext(activeSpanStack.size());
+    }
+
     /**
      * Re-check current trace need profiling, encase third part plugin change the operation name.
      *

diff --git a/...t-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractSpan.java b/...t-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractSpan.java
@@ -124,4 +124,9 @@ public interface AbstractSpan extends AsyncSpan {
      * Should skip analysis in the backend.
      */
     void skipAnalysis();
+
+    /**
+     * Set to ignored status.
+     */
+    void forceIgnoring();
 }
diff --git a/...src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractTracingSpan.java b/...src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractTracingSpan.java
@@ -63,7 +63,8 @@ public abstract class AbstractTracingSpan implements AbstractSpan {
     private volatile boolean isAsyncStopped = false;
 
     /**
-     * The context to which the span belongs
+     * The context to which the span belongs.
+     * This should not be called when {@link #ignored} is true.
      */
     protected final TracingContext owner;
 
@@ -99,6 +100,11 @@ public abstract class AbstractTracingSpan implements AbstractSpan {
      */
     protected boolean skipAnalysis;
 
+    /**
+     * The ignore flag of this span.
+     */
+    protected boolean ignored;
+
     protected AbstractTracingSpan(int spanId, int parentSpanId, String operationName, TracingContext owner) {
         this.operationName = operationName;
         this.spanId = spanId;
@@ -318,7 +324,9 @@ public AbstractSpan prepareForAsync() {
         if (isInAsyncMode) {
             throw new RuntimeException("Prepare for async repeatedly. Span is already in async mode.");
         }
-        ContextManager.awaitFinishAsync(this);
+        if (!ignored) {
+            ContextManager.awaitFinishAsync(this);
+        }
         isInAsyncMode = true;
         return this;
     }
@@ -332,18 +340,28 @@ public AbstractSpan asyncFinish() {
             throw new RuntimeException("Can not do async finish for the span repeatedly.");
         }
         this.endTime = System.currentTimeMillis();
-        owner.asyncStop(this);
+        if (!ignored) {
+            owner.asyncStop(this);
+        }
         isAsyncStopped = true;
         return this;
     }
 
     @Override
     public boolean isProfiling() {
+        if (ignored) {
+            return false;
+        }
         return this.owner.profileStatus().isProfiling();
     }
 
     @Override
     public void skipAnalysis() {
         this.skipAnalysis = true;
     }
+
+    @Override
+    public void forceIgnoring() {
+        this.ignored = true;
+    }
 }
diff --git a/...agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/ExitSpan.java b/...agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/ExitSpan.java
@@ -111,7 +111,9 @@ public String getPeer() {
 
     @Override
     public ExitSpan inject(final ContextCarrier carrier) {
-        this.owner.inject(this, carrier);
+        if (!ignored) {
+            this.owner.inject(this, carrier);
+        }
         return this;
     }
 

diff --git a/...agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/NoopSpan.java b/...agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/NoopSpan.java
@@ -123,6 +123,10 @@ public boolean isProfiling() {
     public void skipAnalysis() {
     }
 
+    @Override
+    public void forceIgnoring() {
+    }
+
     @Override
     public AbstractSpan prepareForAsync() {
         return this;

diff --git a/...core/src/test/java/org/apache/skywalking/apm/agent/core/context/ContinuedContextTest.java b/...core/src/test/java/org/apache/skywalking/apm/agent/core/context/ContinuedContextTest.java
@@ -0,0 +1,98 @@
+/*
+ * 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.agent.core.context;
+
+import org.apache.skywalking.apm.agent.core.boot.ServiceManager;
+import org.apache.skywalking.apm.agent.core.conf.Config;
+import org.apache.skywalking.apm.agent.core.context.ids.NewDistributedTraceId;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.LocalSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.NoopSpan;
+import org.apache.skywalking.apm.agent.core.profile.ProfileStatusContext;
+import org.apache.skywalking.apm.agent.core.test.tools.AgentServiceRule;
+import org.apache.skywalking.apm.agent.core.test.tools.SegmentStorage;
+import org.apache.skywalking.apm.agent.core.test.tools.SegmentStoragePoint;
+import org.apache.skywalking.apm.agent.core.test.tools.TracingSegmentRunner;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.Assert;
+
+@RunWith(TracingSegmentRunner.class)
+public class ContinuedContextTest {
+
+    @SegmentStoragePoint
+    private SegmentStorage tracingData;
+
+    @Rule
+    public AgentServiceRule agentServiceRule = new AgentServiceRule();
+
+    @BeforeClass
+    public static void beforeClass() {
+        Config.Agent.KEEP_TRACING = true;
+    }
+
+    @AfterClass
+    public static void afterClass() {
+        Config.Agent.KEEP_TRACING = false;
+        ServiceManager.INSTANCE.shutdown();
+    }
+
+    @Test
+    public void testContinued() {
+
+        NewDistributedTraceId distributedTraceId = new NewDistributedTraceId();
+        ContextSnapshot snapshot = new ContextSnapshot(
+                "1, 2, 3",
+                1,
+                distributedTraceId,
+                "/for-test-continued",
+                new CorrelationContext(),
+                new ExtensionContext(),
+                ProfileStatusContext.createWithNone()
+        );
+
+        AbstractSpan span = ContextManager.createLocalSpan("test-span");
+        ContextManager.continued(snapshot);
+
+        Assert.assertEquals(distributedTraceId.getId(), ContextManager.getGlobalTraceId());
+        ContextManager.stopSpan();
+    }
+
+    @Test
+    public void testContinuedWithIgnoredSnapshot() {
+
+        ContextSnapshot snapshot =
+                new ContextSnapshot(null, -1, null, null, new CorrelationContext(), new ExtensionContext(), ProfileStatusContext.createWithNone());
+
+        AbstractSpan span = ContextManager.createLocalSpan("test-span");
+        ContextManager.continued(snapshot);
+
+        Assert.assertTrue(span instanceof LocalSpan);
+
+        AbstractSpan span2 = ContextManager.createLocalSpan("test-span2");
+        Assert.assertTrue(span2 instanceof NoopSpan);
+
+        ContextManager.stopSpan();
+        ContextManager.stopSpan();
+    }
+
+}
diff --git a/...rc/test/java/org/apache/skywalking/apm/plugin/rocketMQ/v3/OnExceptionInterceptorTest.java b/...rc/test/java/org/apache/skywalking/apm/plugin/rocketMQ/v3/OnExceptionInterceptorTest.java
@@ -23,6 +23,7 @@
 import static org.mockito.Mockito.when;
 import java.util.List;
 import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
+import org.apache.skywalking.apm.agent.core.context.MockContextSnapshot;
 import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
 import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
@@ -55,7 +56,6 @@ public class OnExceptionInterceptorTest {
     @Rule
     public MockitoRule rule = MockitoJUnit.rule();
 
-    @Mock
     private ContextSnapshot contextSnapshot;
     private SendCallBackEnhanceInfo enhanceInfo;
 
@@ -65,6 +65,7 @@ public class OnExceptionInterceptorTest {
     @Before
     public void setUp() {
         exceptionInterceptor = new OnExceptionInterceptor();
+        contextSnapshot = MockContextSnapshot.INSTANCE.mockContextSnapshot();
 
         enhanceInfo = new SendCallBackEnhanceInfo("test", contextSnapshot);
         when(enhancedInstance.getSkyWalkingDynamicField()).thenReturn(enhanceInfo);

diff --git a/.../src/test/java/org/apache/skywalking/apm/plugin/rocketMQ/v3/OnSuccessInterceptorTest.java b/.../src/test/java/org/apache/skywalking/apm/plugin/rocketMQ/v3/OnSuccessInterceptorTest.java
@@ -23,6 +23,7 @@
 import static org.mockito.Mockito.when;
 import java.util.List;
 import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
+import org.apache.skywalking.apm.agent.core.context.MockContextSnapshot;
 import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
 import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
@@ -57,7 +58,6 @@ public class OnSuccessInterceptorTest {
     @Rule
     public MockitoRule rule = MockitoJUnit.rule();
 
-    @Mock
     private ContextSnapshot contextSnapshot;
     @Mock
     private SendResult sendResult;
@@ -70,6 +70,7 @@ public class OnSuccessInterceptorTest {
     @Before
     public void setUp() {
         successInterceptor = new OnSuccessInterceptor();
+        contextSnapshot = MockContextSnapshot.INSTANCE.mockContextSnapshot();
 
         enhanceInfo = new SendCallBackEnhanceInfo("test", contextSnapshot);
         when(enhancedInstance.getSkyWalkingDynamicField()).thenReturn(enhanceInfo);

diff --git a/...rc/test/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/OnExceptionInterceptorTest.java b/...rc/test/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/OnExceptionInterceptorTest.java
@@ -23,6 +23,7 @@
 import static org.mockito.Mockito.when;
 import java.util.List;
 import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
+import org.apache.skywalking.apm.agent.core.context.MockContextSnapshot;
 import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
 import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
@@ -55,7 +56,6 @@ public class OnExceptionInterceptorTest {
     @Rule
     public MockitoRule rule = MockitoJUnit.rule();
 
-    @Mock
     private ContextSnapshot contextSnapshot;
     private SendCallBackEnhanceInfo enhanceInfo;
 
@@ -69,6 +69,7 @@ public void setUp() {
 
     @Test
     public void testOnException() throws Throwable {
+        contextSnapshot = MockContextSnapshot.INSTANCE.mockContextSnapshot();
         enhanceInfo = new SendCallBackEnhanceInfo("test", contextSnapshot);
         when(enhancedInstance.getSkyWalkingDynamicField()).thenReturn(enhanceInfo);
 

diff --git a/.../src/test/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/OnSuccessInterceptorTest.java b/.../src/test/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/OnSuccessInterceptorTest.java
@@ -25,6 +25,7 @@
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.client.producer.SendStatus;
 import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
+import org.apache.skywalking.apm.agent.core.context.MockContextSnapshot;
 import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
 import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
@@ -57,7 +58,6 @@ public class OnSuccessInterceptorTest {
     @Rule
     public MockitoRule rule = MockitoJUnit.rule();
 
-    @Mock
     private ContextSnapshot contextSnapshot;
     @Mock
     private SendResult sendResult;
@@ -70,6 +70,7 @@ public class OnSuccessInterceptorTest {
     @Before
     public void setUp() {
         successInterceptor = new OnSuccessInterceptor();
+        contextSnapshot = MockContextSnapshot.INSTANCE.mockContextSnapshot();
 
         enhanceInfo = new SendCallBackEnhanceInfo("test", contextSnapshot);
         when(enhancedInstance.getSkyWalkingDynamicField()).thenReturn(enhanceInfo);

diff --git a/...tools/src/main/java/org/apache/skywalking/apm/agent/core/context/MockContextSnapshot.java b/...tools/src/main/java/org/apache/skywalking/apm/agent/core/context/MockContextSnapshot.java
@@ -26,6 +26,8 @@ public enum MockContextSnapshot {
 
     private ContextSnapshot contextSnapshot;
 
+    private ContextSnapshot ignoreContextSnapshot;
+
     MockContextSnapshot() {
         contextSnapshot = new ContextSnapshot(
             "1, 2, 3",
@@ -36,9 +38,21 @@ public enum MockContextSnapshot {
             new ExtensionContext(),
             ProfileStatusContext.createWithNone()
         );
+        ignoreContextSnapshot = new ContextSnapshot(
+                null,
+                -1,
+                null,
+                null,
+                new CorrelationContext(),
+                new ExtensionContext(),
+                ProfileStatusContext.createWithNone());
     }
 
     public ContextSnapshot mockContextSnapshot() {
         return contextSnapshot;
     }
+
+    public ContextSnapshot mockIgnoreContextSnapshot() {
+        return ignoreContextSnapshot;
+    }
 }