Add rate limiter for Zipkin trace receiver (#13481)

Author: kezhenxu94

docs/en/changes/changes.md

@@ -94,6 +94,7 @@
 * Fix otlp log handler reponse error and otlp span convert error.
 * Fix service_relation source layer in mq entry span analyse.
 * Fix metrics comparison in promql with bool modifier.
+* Add rate limiter for Zipkin trace receiver to limit maximum spans per second.
 
 #### UI
 

docs/en/setup/backend/configuration-vocabulary.md

@@ -198,6 +198,7 @@ It divided into several modules, each of which has its own settings. The followi
 | -                       | -             | enabledOtelMetricsRules                                                                                                                                                  | Enabled metric rules for OTLP handler.                                                                                                                                                                                                                                                                                                                                                                                                                                     | SW_OTEL_RECEIVER_ENABLED_OTEL_METRICS_RULES             | -                                                                                            |
 | receiver-zipkin         | default       | A receiver for Zipkin traces.                                                                                                                                            | -                                                                                                                                                                                                                                                                                                                                                                                                                                                                          | -                                                       |                                                                                              |
 | -                       | -             | sampleRate                                                                                                                                                               | The sample rate precision is 1/10000, should be between 0 and 10000                                                                                                                                                                                                                                                                                                                                                                                                        | SW_ZIPKIN_SAMPLE_RATE                                   | 10000                                                                                        |
+| -                       | -             | maxSpansPerSecond                                                                                                                                                        | The maximum spans to be collected per second. 0 means no limit. Spans exceeding this threshold will be dropped.                                                                                                                                                                                                                                                                                                                                                            | SW_ZIPKIN_MAX_SPANS_PER_SECOND                          | 0                                                                                            |
 | -                       | -             | searchableTracesTags                                                                                                                                                     | Defines a set of span tag keys which are searchable. Multiple values are separated by commas. The max length of key=value should be less than 256 or will be dropped.                                                                                                                                                                                                                                                                                                      | SW_ZIPKIN_SEARCHABLE_TAG_KEYS                           | http.method                                                                                  |
 | -                       | -             | enableHttpCollector                                                                                                                                                      | Enable Http Collector.                                                                                                                                                                                                                                                                                                                                                                                                                                                     | SW_ZIPKIN_HTTP_COLLECTOR_ENABLED                        | true                                                                                         |
 | -                       | -             | restHost                                                                                                                                                                 | Binding IP of RESTful services.                                                                                                                                                                                                                                                                                                                                                                                                                                            | SW_RECEIVER_ZIPKIN_REST_HOST                            | 0.0.0.0                                                                                      |

docs/en/setup/backend/zipkin-trace.md

@@ -16,6 +16,8 @@ receiver-zipkin:
     searchableTracesTags: ${SW_ZIPKIN_SEARCHABLE_TAG_KEYS:http.method}
     # The sample rate precision is 1/10000, should be between 0 and 10000
     sampleRate: ${SW_ZIPKIN_SAMPLE_RATE:10000}
+    # The maximum spans to be collected per second. 0 means no limit. Spans exceeding this threshold will be dropped.
+    maxSpansPerSecond: ${SW_ZIPKIN_MAX_SPANS_PER_SECOND:0}
     ## The below configs are for OAP collect zipkin trace from HTTP
     enableHttpCollector: ${SW_ZIPKIN_HTTP_COLLECTOR_ENABLED:true}
     restHost: ${SW_RECEIVER_ZIPKIN_REST_HOST:0.0.0.0}
@@ -71,4 +73,4 @@ Lens UI is Zipkin native UI. SkyWalking webapp has bundled it in the binary dist
 `{webapp IP}:{webapp port}/zipkin` is exposed and accessible for the browser.
 Meanwhile, `Iframe` UI component could be used to host Zipkin Lens UI on the SkyWalking booster UI dashboard.(link=/zipkin) 
 
-Zipkin Lens UI source codes could be found [here](https://github.com/openzipkin/zipkin/tree/master/zipkin-lens).
+Zipkin Lens UI source codes could be found [here](https://github.com/openzipkin/zipkin/tree/master/zipkin-lens).

oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryTraceHandler.java

@@ -42,6 +42,7 @@
 import org.apache.skywalking.oap.server.receiver.zipkin.SpanForwardService;
 import org.apache.skywalking.oap.server.receiver.zipkin.ZipkinReceiverModule;
 import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
+import org.apache.skywalking.oap.server.telemetry.api.CounterMetrics;
 import org.apache.skywalking.oap.server.telemetry.api.HistogramMetrics;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
@@ -78,6 +79,13 @@ public class OpenTelemetryTraceHandler
         MetricsTag.EMPTY_KEY,
         MetricsTag.EMPTY_VALUE
     );
+    @Getter(lazy = true)
+    private final CounterMetrics droppedSpans = getMetricsCreator().createCounter(
+        "otel_spans_dropped",
+        "The count of spans that were dropped due to rate limit",
+        MetricsTag.EMPTY_KEY,
+        MetricsTag.EMPTY_VALUE
+    );
 
     @Override
     public void init(ModuleManager manager, OtelMetricReceiverConfig config) {
@@ -127,7 +135,10 @@ public void export(ExportTraceServiceRequest request, StreamObserver<ExportTrace
                     log.warn("convert span error, discarding the span: {}", e.getMessage());
                 }
             });
-            getForwardService().send(result);
+            final var processedSpans = getForwardService().send(result);
+            if (processedSpans.size() < result.size()) {
+                getDroppedSpans().inc(result.size() - processedSpans.size());
+            }
         }
 
         responseObserver.onNext(ExportTraceServiceResponse.getDefaultInstance());

oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/SpanForwardService.java

@@ -25,7 +25,9 @@
 
 public interface SpanForwardService extends Service {
     /**
-     * Forward and process zipkin span
+     * Forward and process zipkin span.
+     *
+     * @return the spans that are successfully processed.
      */
-    void send(List<Span> spanList);
+    List<Span> send(List<Span> spanList);
 }

oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverConfig.java

@@ -35,6 +35,7 @@ public class ZipkinReceiverConfig extends ModuleConfig {
     private int restAcceptQueueSize = 0;
     private String searchableTracesTags = DEFAULT_SEARCHABLE_TAG_KEYS;
     private int sampleRate = 10000;
+    private int maxSpansPerSecond = 0;
 
     private static final String DEFAULT_SEARCHABLE_TAG_KEYS = String.join(
         Const.COMMA,

oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/ZipkinSpanHTTPHandler.java

@@ -43,6 +43,7 @@
 public class ZipkinSpanHTTPHandler {
     private final HistogramMetrics histogram;
     private final CounterMetrics errorCounter;
+    private final CounterMetrics droppedCounter;
     private final SpanForward spanForward;
 
     public ZipkinSpanHTTPHandler(SpanForward forward, ModuleManager manager) {
@@ -58,6 +59,10 @@ public ZipkinSpanHTTPHandler(SpanForward forward, ModuleManager manager) {
             "trace_analysis_error_count", "The error number of trace analysis",
             new MetricsTag.Keys("protocol"), new MetricsTag.Values("zipkin-http")
         );
+        droppedCounter = metricsCreator.createCounter(
+            "spans_dropped_count", "The count of spans that were dropped due to rate limit",
+            new MetricsTag.Keys("protocol"), new MetricsTag.Values("zipkin-http")
+        );
     }
 
     @Post("/api/v2/spans")
@@ -100,7 +105,10 @@ HttpResponse doCollectSpans(final SpanBytesDecoder decoder,
         final HttpResponse response = HttpResponse.from(req.aggregate().thenApply(request -> {
             try (final HttpData httpData = request.content()) {
                 final List<Span> spanList = decoder.decodeList(httpData.byteBuf().nioBuffer());
-                spanForward.send(spanList);
+                final var processedSpans = spanForward.send(spanList);
+                if (processedSpans.size() < spanList.size()) {
+                    droppedCounter.inc(spanList.size() - processedSpans.size());
+                }
                 return HttpResponse.of(HttpStatus.ACCEPTED);
             }
         }));

oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/trace/SpanForward.java

@@ -18,9 +18,11 @@
 
 package org.apache.skywalking.oap.server.receiver.zipkin.trace;
 
+import com.google.common.util.concurrent.RateLimiter;
 import com.google.gson.JsonObject;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 
 import java.util.Map;
@@ -55,20 +57,25 @@ public class SpanForward implements SpanForwardService {
     private final long samplerBoundary;
     private NamingControl namingControl;
     private SourceReceiver receiver;
+    private RateLimiter rateLimiter;
 
     public SpanForward(final ZipkinReceiverConfig config, final ModuleManager manager) {
         this.config = config;
         this.moduleManager = manager;
         this.searchTagKeys = Arrays.asList(config.getSearchableTracesTags().split(Const.COMMA));
         float sampleRate = (float) config.getSampleRate() / 10000;
         samplerBoundary = (long) (Long.MAX_VALUE * sampleRate);
+        if (config.getMaxSpansPerSecond() > 0) {
+            this.rateLimiter = RateLimiter.create(config.getMaxSpansPerSecond());
+        }
     }
 
-    public void send(List<Span> spanList) {
+    public List<Span> send(List<Span> spanList) {
         if (CollectionUtils.isEmpty(spanList)) {
-            return;
+            return Collections.emptyList();
         }
-        getSampledTraces(spanList).forEach(span -> {
+        final var sampledTraces = getSampledTraces(spanList);
+        sampledTraces.forEach(span -> {
             ZipkinSpan zipkinSpan = new ZipkinSpan();
             String serviceName = span.localServiceName();
             if (StringUtil.isEmpty(serviceName)) {
@@ -153,6 +160,7 @@ public void send(List<Span> spanList) {
                 toServiceRelation(zipkinSpan, minuteTimeBucket);
             }
         });
+        return sampledTraces;
     }
 
     private void addAutocompleteTags(final long minuteTimeBucket, final String key, final String value) {
@@ -188,8 +196,8 @@ private void toServiceRelation(ZipkinSpan zipkinSpan, final long minuteTimeBucke
     }
 
     private List<Span> getSampledTraces(List<Span> input) {
-        //100% sampleRate
-        if (config.getSampleRate() == 10000) {
+        // 100% sampleRate and no rateLimiter, return all spans
+        if (config.getSampleRate() == 10000 && rateLimiter == null) {
             return input;
         }
         List<Span> sampledTraces = new ArrayList<>(input.size());
@@ -198,10 +206,23 @@ private List<Span> getSampledTraces(List<Span> input) {
                 sampledTraces.add(span);
                 continue;
             }
-            long traceId = HexCodec.lowerHexToUnsignedLong(span.traceId());
-            traceId = traceId == Long.MIN_VALUE ? Long.MAX_VALUE : Math.abs(traceId);
-            if (traceId <= samplerBoundary) {
+            
+            // Apply maximum spans per minute sampling first
+            if (rateLimiter != null && !rateLimiter.tryAcquire()) {
+                log.debug("Span dropped due to maximum spans per minute limit: {}", span.id());
+                continue;
+            }
+            
+            // Apply percentage-based sampling
+            if (config.getSampleRate() == 10000) {
+                // 100% sample rate - include all spans that passed the maximum spans check
                 sampledTraces.add(span);
+            } else {
+                long traceId = HexCodec.lowerHexToUnsignedLong(span.traceId());
+                traceId = traceId == Long.MIN_VALUE ? Long.MAX_VALUE : Math.abs(traceId);
+                if (traceId <= samplerBoundary) {
+                    sampledTraces.add(span);
+                }
             }
         }
         return sampledTraces;

oap-server/server-starter/src/main/resources/application.yml

@@ -378,6 +378,8 @@ receiver-zipkin:
     searchableTracesTags: ${SW_ZIPKIN_SEARCHABLE_TAG_KEYS:http.method}
     # The sample rate precision is 1/10000, should be between 0 and 10000
     sampleRate: ${SW_ZIPKIN_SAMPLE_RATE:10000}
+    # The maximum spans to be collected per second. 0 means no limit. Spans exceeding this threshold will be dropped.
+    maxSpansPerSecond: ${SW_ZIPKIN_MAX_SPANS_PER_SECOND:0}
     ## The below configs are for OAP collect zipkin trace from HTTP
     enableHttpCollector: ${SW_ZIPKIN_HTTP_COLLECTOR_ENABLED:true}
     restHost: ${SW_RECEIVER_ZIPKIN_REST_HOST:0.0.0.0}

oap-server/server-starter/src/main/resources/otel-rules/oap.yaml

@@ -118,6 +118,8 @@ metricsRules:
     exp: otel_logs_latency_count.sum(['service', 'host_name']).increase('PT1M')
   - name: otel_spans_received
     exp: otel_spans_latency_count.sum(['service', 'host_name']).increase('PT1M')
+  - name: otel_spans_dropped
+    exp: otel_spans_dropped.sum(['service', 'host_name']).increase('PT1M')
   - name: otel_metrics_latency_percentile
     exp: otel_metrics_latency.sum(['le', 'service', 'host_name']).increase('PT1M').histogram().histogram_percentile([50,70,90,99])
   - name: otel_logs_latency_percentile