Add a new sampler (`possibility`) in LAL (#8591)

CHANGES.md

@@ -64,6 +64,7 @@ Release Notes.
 * Support ElasticSearch 8 and add it into E2E tests.
 * Disable indexing for field `alarm_record.tags_raw_data` of binary type in ElasticSearch storage.
 * Fix Zipkin receiver wrong condition for decoding `gzip`. 
+* Add a new sampler (`possibility`) in LAL.
 
 #### UI
 

docs/en/concepts-and-designs/lal.md

@@ -248,14 +248,16 @@ extracted useful information, such as metrics.
 
 #### Sampler
 
-Sampler allows you to save the logs in a sampling manner. Currently, the sampling strategy `rateLimit` is supported. We welcome
-contributions on more sampling strategies. If multiple samplers are specified, the last one determines the final sampling
-result. See examples in [Enforcer](#enforcer).
+Sampler allows you to save the logs in a sampling manner. Currently, the following sampling strategies are supported:
 
-`rateLimit` samples `n` logs at a maximum rate of 1 minute. `rateLimit("SamplerID")` requires an ID for the sampler. Sampler
-declarations with the same ID share the same sampler instance, thus sharing the same `rpm` and resetting logic.
+- `rateLimit`: samples `n` logs at a maximum rate of 1 minute. `rateLimit("SamplerID")` requires an ID for the sampler.
+Sampler declarations with the same ID share the same sampler instance, thus sharing the same `rpm` and resetting logic.
+- `possibility`: every piece of log has a pseudo possibility of `percentage` to be sampled, the possibility was generated by Java random number generator and compare to the given `percentage` option.
 
-Examples:
+We welcome contributions on more sampling strategies. If multiple samplers are specified, the last one determines the
+final sampling result. See examples in [Enforcer](#enforcer).
+
+Examples 1, `rateLimit`:
 
 ```groovy
 filter {
@@ -277,6 +279,26 @@ filter {
 }
 ```
 
+Examples 2, `possibility`:
+
+```groovy
+filter {
+    // ... parser
+
+    sink {
+        sampler {
+            if (parsed.service == "ImportantApp") {
+                possibility(80) { // samples 80% of the logs for service "ImportantApp"
+                }
+            } else {
+                possibility(30) { // samples 30% of the logs for other services than "ImportantApp"
+                }
+            }
+        }
+    }
+}
+```
+
 #### Dropper
 
 Dropper is a special sink, meaning that all logs are dropped without any exception. This is useful when you want to

oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/dsl/spec/sink/SamplerSpec.java

@@ -24,20 +24,23 @@ import groovy.lang.GString;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import org.apache.skywalking.oap.log.analyzer.dsl.spec.AbstractSpec;
+import org.apache.skywalking.oap.log.analyzer.dsl.spec.sink.sampler.PossibilitySampler;
 import org.apache.skywalking.oap.log.analyzer.dsl.spec.sink.sampler.RateLimitingSampler;
 import org.apache.skywalking.oap.log.analyzer.dsl.spec.sink.sampler.Sampler;
 import org.apache.skywalking.oap.log.analyzer.provider.LogAnalyzerModuleConfig;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
 
 public class SamplerSpec extends AbstractSpec {
-    private final Map<GString, Sampler> samplers;
+    private final Map<GString, Sampler> rateLimitSamplers;
+    private final Map<Integer, Sampler> possibilitySamplers;
     private final RateLimitingSampler.ResetHandler rlsResetHandler;
 
     public SamplerSpec(final ModuleManager moduleManager,
                        final LogAnalyzerModuleConfig moduleConfig) {
         super(moduleManager, moduleConfig);
 
-        samplers = new ConcurrentHashMap<>();
+        rateLimitSamplers = new ConcurrentHashMap<>();
+        possibilitySamplers = new ConcurrentHashMap<>();
         rlsResetHandler = new RateLimitingSampler.ResetHandler();
     }
 
@@ -47,7 +50,21 @@ public class SamplerSpec extends AbstractSpec {
             return;
         }
 
-        final Sampler sampler = samplers.computeIfAbsent(id, $ -> new RateLimitingSampler(rlsResetHandler).start());
+        final Sampler sampler = rateLimitSamplers.computeIfAbsent(id, $ -> new RateLimitingSampler(rlsResetHandler).start());
+
+        cl.setDelegate(sampler);
+        cl.call();
+
+        sampleWith(sampler);
+    }
+
+    @SuppressWarnings("unused")
+    public void possibility(final int percentage, @DelegatesTo(PossibilitySampler.class) final Closure<?> cl) {
+        if (BINDING.get().shouldAbort()) {
+            return;
+        }
+
+        final Sampler sampler = possibilitySamplers.computeIfAbsent(percentage, $ -> new PossibilitySampler(percentage).start());
 
         cl.setDelegate(sampler);
         cl.call();

oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/dsl/spec/sink/sampler/PossibilitySampler.java

@@ -0,0 +1,54 @@
+/*
+ * 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.oap.log.analyzer.dsl.spec.sink.sampler;
+
+import io.netty.util.internal.ThreadLocalRandom;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.experimental.Accessors;
+
+@RequiredArgsConstructor
+@Accessors(fluent = true)
+@EqualsAndHashCode(of = {"percentage"})
+public class PossibilitySampler implements Sampler {
+    @Getter
+    private final int percentage;
+
+    private final ThreadLocalRandom random = ThreadLocalRandom.current();
+
+    @Override
+    public PossibilitySampler start() {
+        return this;
+    }
+
+    @Override
+    public void close() {
+    }
+
+    @Override
+    public boolean sample() {
+        return random.nextInt(100) < percentage;
+    }
+
+    @Override
+    public PossibilitySampler reset() {
+        return this;
+    }
+}

oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AccessLogAnalyzer.java

@@ -70,7 +70,7 @@ public interface AccessLogAnalyzer<E> {
 
     @Data
     @Builder
-    class Result {
+    static class Result {
         /**
          * The service representing the Envoy node.
          */