Fix sampling and Kafka reporter conflicting. (#5408)

* Fix sampling and Kafka reporter conflicting.

* Remove legacy codes.
Co-authored-by: Nkezhenxu94 <kezhenxu94@apache.org>

apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/ContextManagerExtendService.java

@@ -65,7 +65,7 @@ public class ContextManagerExtendService implements BootService, GRPCChannelList
             context = new IgnoredTracerContext();
         } else {
             SamplingService samplingService = ServiceManager.INSTANCE.findService(SamplingService.class);
-            if (forceSampling || samplingService.trySampling()) {
+            if (forceSampling || samplingService.trySampling(operationName)) {
                 context = new TracingContext(operationName);
             } else {
                 context = new IgnoredTracerContext();

apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/TracingContext.java

@@ -40,7 +40,6 @@ import org.apache.skywalking.apm.agent.core.logging.api.ILog;
 import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
 import org.apache.skywalking.apm.agent.core.profile.ProfileStatusReference;
 import org.apache.skywalking.apm.agent.core.profile.ProfileTaskExecutionService;
-import org.apache.skywalking.apm.agent.core.sampling.SamplingService;
 import org.apache.skywalking.apm.util.StringUtil;
 
 /**
@@ -63,11 +62,6 @@ public class TracingContext implements AbstractTracerContext {
      */
     private static ProfileTaskExecutionService PROFILE_TASK_EXECUTION_SERVICE;
 
-    /**
-     * @see SamplingService
-     */
-    private static SamplingService SAMPLING_SERVICE;
-
     /**
      * The final {@link TraceSegment}, which includes all finished spans.
      */
@@ -123,10 +117,6 @@ public class TracingContext implements AbstractTracerContext {
         createTime = System.currentTimeMillis();
         running = true;
 
-        if (SAMPLING_SERVICE == null) {
-            SAMPLING_SERVICE = ServiceManager.INSTANCE.findService(SamplingService.class);
-        }
-
         // profiling status
         if (PROFILE_TASK_EXECUTION_SERVICE == null) {
             PROFILE_TASK_EXECUTION_SERVICE = ServiceManager.INSTANCE.findService(ProfileTaskExecutionService.class);
@@ -300,10 +290,6 @@ public class TracingContext implements AbstractTracerContext {
         }
         AbstractSpan parentSpan = peek();
         final int parentSpanId = parentSpan == null ? -1 : parentSpan.getSpanId();
-        /*
-         * From v6.0.0-beta, local span doesn't do op name register.
-         * All op name register is related to entry and exit spans only.
-         */
         AbstractTracingSpan span = new LocalSpan(spanIdGenerator++, parentSpanId, operationName, this);
         span.start();
         return push(span);
@@ -438,20 +424,7 @@ public class TracingContext implements AbstractTracerContext {
 
             if (isFinishedInMainThread && (!isRunningInAsyncMode || asyncSpanCounter == 0)) {
                 TraceSegment finishedSegment = segment.finish(isLimitMechanismWorking());
-                /*
-                 * Recheck the segment if the segment contains only one span.
-                 * Because in the runtime, can't sure this segment is part of distributed trace.
-                 *
-                 * @see {@link #createSpan(String, long, boolean)}
-                 */
-                if (!segment.hasRef() && segment.isSingleSpanSegment()) {
-                    if (!SAMPLING_SERVICE.trySampling()) {
-                        finishedSegment.setIgnore(true);
-                    }
-                }
-
                 TracingContext.ListenerManager.notifyFinish(finishedSegment);
-
                 running = false;
             }
         } finally {

apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractTracingSpan.java

@@ -39,7 +39,13 @@ import org.apache.skywalking.apm.network.trace.component.Component;
  * distributed trace.
  */
 public abstract class AbstractTracingSpan implements AbstractSpan {
+    /**
+     * Span id starts from 0.
+     */
     protected int spanId;
+    /**
+     * Parent span id starts from 0. -1 means no parent span.
+     */
     protected int parentSpanId;
     protected List<TagValuePair> tags;
     protected String operationName;

apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/sampling/SamplingService.java

@@ -86,9 +86,10 @@ public class SamplingService implements BootService {
     }
 
     /**
+     * @param operationName The first operation name of the new tracing context.
      * @return true, if sampling mechanism is on, and getDefault the sampling factor successfully.
      */
-    public boolean trySampling() {
+    public boolean trySampling(String operationName) {
         if (on) {
             int factor = samplingFactorHolder.get();
             if (factor < Config.Agent.SAMPLE_N_PER_3_SECS) {

apm-sniffer/optional-plugins/trace-ignore-plugin/src/main/java/org/apache/skywalking/apm/plugin/trace/ignore/TraceIgnoreExtendService.java

@@ -19,26 +19,20 @@
 package org.apache.skywalking.apm.plugin.trace.ignore;
 
 import org.apache.skywalking.apm.agent.core.boot.OverrideImplementor;
-import org.apache.skywalking.apm.agent.core.context.AbstractTracerContext;
-import org.apache.skywalking.apm.agent.core.context.ContextManagerExtendService;
-import org.apache.skywalking.apm.agent.core.context.IgnoredTracerContext;
 import org.apache.skywalking.apm.agent.core.logging.api.ILog;
 import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
+import org.apache.skywalking.apm.agent.core.sampling.SamplingService;
 import org.apache.skywalking.apm.plugin.trace.ignore.conf.IgnoreConfig;
 import org.apache.skywalking.apm.plugin.trace.ignore.conf.IgnoreConfigInitializer;
 import org.apache.skywalking.apm.plugin.trace.ignore.matcher.FastPathMatcher;
 import org.apache.skywalking.apm.plugin.trace.ignore.matcher.TracePathMatcher;
 import org.apache.skywalking.apm.util.StringUtil;
 
-@OverrideImplementor(ContextManagerExtendService.class)
-public class TraceIgnoreExtendService extends ContextManagerExtendService {
-
+@OverrideImplementor(SamplingService.class)
+public class TraceIgnoreExtendService extends SamplingService {
     private static final ILog LOGGER = LogManager.getLogger(TraceIgnoreExtendService.class);
-
     private static final String PATTERN_SEPARATOR = ",";
-
     private TracePathMatcher pathMatcher = new FastPathMatcher();
-
     private String[] patterns = new String[] {};
 
     @Override
@@ -50,15 +44,31 @@ public class TraceIgnoreExtendService extends ContextManagerExtendService {
     }
 
     @Override
-    public AbstractTracerContext createTraceContext(String operationName, boolean forceSampling) {
-        if (patterns.length > 0 && !forceSampling) {
+    public void prepare() {
+    }
+
+    @Override
+    public void onComplete() {
+    }
+
+    @Override
+    public void shutdown() {
+    }
+
+    @Override
+    public boolean trySampling(final String operationName) {
+        if (patterns.length > 0) {
             for (String pattern : patterns) {
                 if (pathMatcher.match(pattern, operationName)) {
                     LOGGER.debug("operationName : " + operationName + " Ignore tracking");
-                    return new IgnoredTracerContext();
+                    return false;
                 }
             }
         }
-        return super.createTraceContext(operationName, forceSampling);
+        return true;
+    }
+
+    @Override
+    public void forceSampled() {
     }
 }

apm-sniffer/optional-plugins/trace-ignore-plugin/src/test/java/org/apache/skywalking/apm/plugin/trace/ignore/TraceIgnoreTest.java

@@ -18,11 +18,9 @@
 
 package org.apache.skywalking.apm.plugin.trace.ignore;
 
+import java.util.Properties;
 import org.apache.skywalking.apm.agent.core.boot.ServiceManager;
-import org.apache.skywalking.apm.agent.core.context.AbstractTracerContext;
-import org.apache.skywalking.apm.agent.core.context.ContextManagerExtendService;
-import org.apache.skywalking.apm.agent.core.context.IgnoredTracerContext;
-import org.apache.skywalking.apm.agent.core.context.TracingContext;
+import org.apache.skywalking.apm.agent.core.sampling.SamplingService;
 import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule;
 import org.apache.skywalking.apm.plugin.trace.ignore.conf.IgnoreConfig;
 import org.apache.skywalking.apm.util.ConfigInitializer;
@@ -32,42 +30,40 @@ import org.junit.Rule;
 import org.junit.Test;
 import org.junit.contrib.java.lang.system.EnvironmentVariables;
 
-import java.util.Properties;
-
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
 
 public class TraceIgnoreTest {
 
     @Rule
-    public final EnvironmentVariables environmentVariables = new EnvironmentVariables().set("SW_AGENT_TRACE_IGNORE_PATH", "path_test");
+    public final EnvironmentVariables environmentVariables = new EnvironmentVariables().set(
+        "SW_AGENT_TRACE_IGNORE_PATH", "path_test");
 
     @Rule
     public AgentServiceRule serviceRule = new AgentServiceRule();
 
     @Test
     public void testServiceOverrideFromPlugin() {
-        ContextManagerExtendService service = ServiceManager.INSTANCE.findService(ContextManagerExtendService.class);
+        SamplingService service = ServiceManager.INSTANCE.findService(SamplingService.class);
         Assert.assertEquals(TraceIgnoreExtendService.class, service.getClass());
     }
 
     @Test
     public void testTraceIgnore() {
-        ContextManagerExtendService service = ServiceManager.INSTANCE.findService(ContextManagerExtendService.class);
+        SamplingService service = ServiceManager.INSTANCE.findService(SamplingService.class);
         IgnoreConfig.Trace.IGNORE_PATH = "/eureka/**";
         service.boot();
-        AbstractTracerContext ignoredTracerContext = service.createTraceContext("/eureka/apps", false);
-        Assert.assertEquals(IgnoredTracerContext.class, ignoredTracerContext.getClass());
 
-        AbstractTracerContext traceContext = service.createTraceContext("/consul/apps", false);
-        Assert.assertEquals(TracingContext.class, traceContext.getClass());
+        Assert.assertFalse(service.trySampling("/eureka/apps"));
+        Assert.assertTrue(service.trySampling("/consul/apps"));
     }
 
     @Test
     public void testTraceIgnoreConfigOverridingFromSystemEnv() throws IllegalAccessException {
         Properties properties = new Properties();
         properties.put("trace.ignore_path", "${SW_AGENT_TRACE_IGNORE_PATH:/path/eureka/**}");
-        properties.put("trace.ignore_path", PropertyPlaceholderHelper.INSTANCE.replacePlaceholders((String) properties.get("trace.ignore_path"), properties));
+        properties.put("trace.ignore_path", PropertyPlaceholderHelper.INSTANCE.replacePlaceholders(
+            (String) properties.get("trace.ignore_path"), properties));
         ConfigInitializer.initialize(properties, IgnoreConfig.class);
         assertThat(IgnoreConfig.Trace.IGNORE_PATH, is("path_test"));
     }