Support Tail Based Sampling Processor From OTEL Collector Extension

.github/workflows/ci-e2e-tailsampling-processor.yml

@@ -0,0 +1,41 @@
+name: Test Tail Sampling Processor
+
+on:
+  push:
+    branches: [main]
+
+  pull_request:
+    branches: [main]
+
+concurrency:
+  group: ${{ github.workflow }}-${{ (github.event.pull_request && github.event.pull_request.number) || github.ref || github.run_id }}
+  cancel-in-progress: true
+
+# See https://github.com/ossf/scorecard/blob/main/docs/checks.md#token-permissions
+permissions:  # added using https://github.com/step-security/secure-workflows
+  contents: read
+
+jobs:
+  tailsampling-processor:
+    runs-on: ubuntu-latest
+    steps:
+    - name: Harden Runner
+      uses: step-security/harden-runner@0d381219ddf674d61a7572ddd19d7941e271515c # v2.9.0
+      with:
+        egress-policy: audit # TODO: change to 'egress-policy: block' after couple of runs
+
+    - uses: actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332 # v4.1.7
+
+    - uses: actions/setup-go@0a12ed9d6a96ab950c8f026ed9f722fe0da7ef32 # v5.0.2
+      with:
+        go-version: 1.23.x
+
+    - name: Run Tail Sampling Processor Integration Test
+      run: |
+        make tail-sampling-integration-test
+
+    - name: Upload coverage to codecov
+      uses: ./.github/actions/upload-codecov
+      with:
+        files: cover.out
+        flags: tailsampling-processor

Makefile

@@ -160,6 +160,10 @@ index-cleaner-integration-test: docker-images-elastic
 index-rollover-integration-test: docker-images-elastic
 	$(MAKE) storage-integration-test COVEROUT=cover-index-rollover.out
 
+.PHONY: tail-sampling-integration-test
+tail-sampling-integration-test:
+	SAMPLING=tail $(MAKE) jaeger-v2-storage-integration-test
+
 .PHONY: cover
 cover: nocover
 	bash -c "set -e; set -o pipefail; STORAGE=memory $(GOTEST) -timeout 5m -coverprofile $(COVEROUT) ./... | tee test-results.json"

cmd/jaeger/config-tail-sampling-always-sample.yaml

@@ -0,0 +1,38 @@
+service:
+  extensions: [jaeger_storage, jaeger_query, healthcheckv2]
+  pipelines:
+    traces:
+      receivers: [otlp]
+      processors: [tail_sampling]
+      exporters: [jaeger_storage_exporter]
+  telemetry:
+    logs:
+      level: DEBUG
+
+extensions:
+  healthcheckv2:
+    use_v2: true
+    http:
+  jaeger_query:
+    trace_storage: some_storage
+  jaeger_storage:
+    backends:
+      some_storage:
+        memory:
+          max_traces: 100000
+
+receivers:
+  otlp:
+    protocols:
+      grpc:
+      http:
+        endpoint: "0.0.0.0:4318"
+
+processors:
+  tail_sampling:
+    decision_wait: 5s
+    policies: [{ name: test-policy-1, type: always_sample }]
+
+exporters:
+  jaeger_storage_exporter:
+    trace_storage: some_storage

cmd/jaeger/config-tail-sampling-service-name-policy.yaml

@@ -0,0 +1,46 @@
+service:
+  extensions: [jaeger_storage, jaeger_query, healthcheckv2]
+  pipelines:
+    traces:
+      receivers: [otlp]
+      processors: [tail_sampling]
+      exporters: [jaeger_storage_exporter]
+  telemetry:
+    logs:
+      level: DEBUG
+
+extensions:
+  healthcheckv2:
+    use_v2: true
+    http:
+  jaeger_query:
+    trace_storage: some_storage
+  jaeger_storage:
+    backends:
+      some_storage:
+        memory:
+          max_traces: 100000
+
+receivers:
+  otlp:
+    protocols:
+      grpc:
+      http:
+        endpoint: "0.0.0.0:4318"
+
+processors:
+  tail_sampling:
+    decision_wait: 5s
+    policies:
+      [
+        {
+          name: filter-by-attribute,
+          type: string_attribute,
+          string_attribute:
+            { key: service.name, values: [tracegen-00, tracegen-03] },
+        },
+      ]
+
+exporters:
+  jaeger_storage_exporter:
+    trace_storage: some_storage

cmd/jaeger/internal/components.go

@@ -8,6 +8,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusexporter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/healthcheckv2extension"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/jaegerreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/zipkinreceiver"
@@ -104,6 +105,7 @@ func (b builders) build() (otelcol.Factories, error) {
 		// standard
 		batchprocessor.NewFactory(),
 		memorylimiterprocessor.NewFactory(),
+		tailsamplingprocessor.NewFactory(),
 		// add-ons
 		adaptivesampling.NewFactory(),
 	)

cmd/jaeger/internal/integration/tailsampling_test.go

@@ -0,0 +1,97 @@
+// // Copyright (c) 2024 The Jaeger Authors.
+// // SPDX-License-Identifier: Apache-2.0
+
+package integration
+
+import (
+	"context"
+	"os"
+	"os/exec"
+	"sort"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/jaegertracing/jaeger/plugin/storage/integration"
+)
+
+// TailSamplingIntegration contains the test components to perform an integration test
+// for the Tail Sampling Processor.
+type TailSamplingIntegration struct {
+	E2EStorageIntegration
+
+	// expectedServices contains a list of services that should be sampled in the test case.
+	expectedServices []string
+}
+
+// TestTailSamplingProcessor_EnforcesPolicies runs an A/B test to perform an integration test
+// for the Tail Sampling Processor.
+//   - Test A uses a Jaeger config file with a tail sampling processor that has a policy for sampling
+//     all traces. In this test, we check that all services that are samples are stored.
+//   - Test B uses a Jaeger config file with a tail sampling processor that has a policy to sample
+//     traces using on the `service.name` attribute. In this test, we check that only the services
+//     listed as part of the policy in the config file are stored.
+func TestTailSamplingProcessor_EnforcesPolicies(t *testing.T) {
+	if env := os.Getenv("SAMPLING"); env != "tail" {
+		t.Skipf("This test requires environment variable SAMPLING=tail")
+	}
+
+	expectedServicesA := []string{"tracegen-00", "tracegen-01", "tracegen-02", "tracegen-03", "tracegen-04"}
+	tailSamplingA := &TailSamplingIntegration{
+		E2EStorageIntegration: E2EStorageIntegration{
+			ConfigFile: "../../config-tail-sampling-always-sample.yaml",
+			StorageIntegration: integration.StorageIntegration{
+				CleanUp: purge,
+			},
+		},
+		expectedServices: expectedServicesA,
+	}
+
+	expectedServicesB := []string{"tracegen-00", "tracegen-03"}
+	tailSamplingB := &TailSamplingIntegration{
+		E2EStorageIntegration: E2EStorageIntegration{
+			ConfigFile: "../../config-tail-sampling-service-name-policy.yaml",
+			StorageIntegration: integration.StorageIntegration{
+				CleanUp: purge,
+			},
+		},
+		expectedServices: expectedServicesB,
+	}
+
+	t.Run("sample_all", tailSamplingA.testTailSamplingProccessor)
+	t.Run("sample_some", tailSamplingB.testTailSamplingProccessor)
+}
+
+// testTailSamplingProccessor performs the following steps:
+//  1. Initialize the test case by starting the Jaeger V2 collector
+//  2. Generate 5 traces using `tracegen` with one service per trace
+//  3. Read the stored services from the memory store
+//  4. Check that the sampled services match what is expected
+func (ts *TailSamplingIntegration) testTailSamplingProccessor(t *testing.T) {
+	ts.e2eInitialize(t, "memory")
+	ts.generateTraces(t)
+
+	var actual []string
+	found := assert.Eventually(t, func() bool {
+		var err error
+		actual, err = ts.SpanReader.GetServices(context.Background())
+		require.NoError(t, err)
+		sort.Strings(actual)
+		return assert.ObjectsAreEqualValues(ts.expectedServices, actual)
+	}, 100*time.Second, 15*time.Second)
+
+	if !found {
+		t.Log("\t Expected:", ts.expectedServices)
+		t.Log("\t Actual  :", actual)
+	}
+}
+
+// generateTraces generates 5 traces using `tracegen` with one service per trace
+func (*TailSamplingIntegration) generateTraces(t *testing.T) {
+	tracegenCmd := exec.Command("go", "run", "../../../../cmd/tracegen", "-traces", "5", "-services", "5")
+	stdout, err := tracegenCmd.Output()
+	require.NoError(t, err)
+	t.Logf("tracegen completed: %s", stdout)
+}

docker-compose/tail-sampling/Makefile

@@ -0,0 +1,30 @@
+# Copyright (c) 2024 The Jaeger Authors.
+# SPDX-License-Identifier: Apache-2.0
+
+BINARY ?= jaeger
+
+.PHONY: build
+build: clean-jaeger
+	cd ../../ && make build-$(BINARY) GOOS=linux 
+	cd ../../ && make create-baseimg PLATFORMS=linux/$(shell go env GOARCH)
+	cd ../../ && docker buildx build --target release \
+		--tag jaegertracing/$(BINARY):dev \
+		--build-arg base_image=localhost:5000/baseimg_alpine:latest \
+		--build-arg debug_image=not-used \
+		--build-arg TARGETARCH=$(shell go env GOARCH) \
+		--load \
+		cmd/$(BINARY)
+
+.PHONY: dev
+dev: export JAEGER_IMAGE_TAG = dev
+dev: build 
+	docker compose -f docker-compose.yml up $(DOCKER_COMPOSE_ARGS)
+
+.PHONY: clean-jaeger
+clean-jaeger:
+	# Also cleans up intermediate cached containers.
+	docker system prune -f
+
+.PHONY: clean-all
+clean-all: clean-jaeger
+	docker rmi -f otel/opentelemetry-collector-contrib:latest ;

docker-compose/tail-sampling/docker-compose.yml

@@ -0,0 +1,35 @@
+services:
+  jaeger:
+    networks:
+      backend:
+    image: jaegertracing/jaeger:${JAEGER_IMAGE_TAG:-latest}
+    volumes:
+      - "./jaeger-v2-config.yml:/etc/jaeger/config.yml"
+    command: ["--config", "/etc/jaeger/config.yml"]
+    ports:
+      - "16686:16686"
+
+  otel_collector:
+    networks:
+      backend:
+    image: otel/opentelemetry-collector-contrib:${OTEL_IMAGE_TAG:-0.108.0}
+    volumes:
+      - ${OTEL_CONFIG_SRC:-./otel-collector-config-connector.yml}:/etc/otelcol/otel-collector-config.yml
+    command: --config /etc/otelcol/otel-collector-config.yml
+    depends_on:
+      - jaeger
+    ports:
+      - "8889:8889"
+
+  tracegen:
+    networks:
+      - backend
+    image: jaegertracing/jaeger-tracegen:latest
+    environment:
+      - OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://jaeger:4318/v1/traces
+    command: ["-workers", "3", "-pause", "250ms", "-services", "5", "-duration", "10s"]
+    depends_on:
+      - jaeger
+
+networks:
+  backend:

docker-compose/tail-sampling/jaeger-v2-config.yml

@@ -0,0 +1,46 @@
+service:
+  extensions: [jaeger_storage, jaeger_query, healthcheckv2]
+  pipelines:
+    traces:
+      receivers: [otlp]
+      processors: [tail_sampling]
+      exporters: [jaeger_storage_exporter]
+  telemetry:
+    logs:
+      level: DEBUG
+
+extensions:
+  healthcheckv2:
+    use_v2: true
+    http:
+  jaeger_query:
+    trace_storage: some_storage
+  jaeger_storage:
+    backends:
+      some_storage:
+        memory:
+          max_traces: 100000
+
+receivers:
+  otlp:
+    protocols:
+      grpc:
+      http:
+        endpoint: "0.0.0.0:4318"
+
+processors:
+  tail_sampling:
+    decision_wait: 15s
+    policies:
+      [
+        {
+          name: filter-by-attribute,
+          type: string_attribute,
+          string_attribute:
+            { key: service.name, values: [tracegen-02, tracegen-04] },
+        },
+      ]
+
+exporters:
+  jaeger_storage_exporter:
+    trace_storage: some_storage

docker-compose/tail-sampling/otel-collector-config-connector.yml

@@ -0,0 +1,29 @@
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: 0.0.0.0:4317
+      http:
+        endpoint: 0.0.0.0:4318
+
+exporters:
+  loadbalancing:
+    routing_key: "traceID"
+    protocol:
+      otlp:
+        timeout: 1s
+        tls:
+          insecure: true
+    resolver:
+      static:
+        hostnames:
+        - 0.0.0.0:4317
+
+service:
+  pipelines:
+    traces:
+      receivers:
+        - otlp
+      processors: []
+      exporters:
+        - loadbalancing