Testing OpenTelemetry Collectors and Exporters in CI

# docker-compose.collector-test.yml
services:
  # Collector under test
  otelcol:
    image: otel/opentelemetry-collector-contrib:0.91.0
    command: ["--config", "/etc/otel/config.yaml"]
    volumes:
      - ./otel-collector-config.yaml:/etc/otel/config.yaml
    ports:
      - "4317:4317"   # OTLP gRPC receiver
      - "4318:4318"   # OTLP HTTP receiver
      - "8888:8888"   # Prometheus metrics (collector self-monitoring)
    depends_on:
      - prometheus
      - tempo
      - loki
    healthcheck:
      test: ["CMD", "wget", "--quiet", "--tries=1", "--spider", "http://localhost:13133/"]
      interval: 5s
      retries: 10

  # Backends for assertion
  prometheus:
    image: prom/prometheus:v2.48.0
    volumes:
      - ./prometheus-test.yml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"

  tempo:
    image: grafana/tempo:2.3.1
    command: ["-config.file=/etc/tempo.yaml"]
    volumes:
      - ./tempo-test.yaml:/etc/tempo.yaml
    ports:
      - "3200:3200"
      - "4319:4317"   # Tempo's OTLP port

  loki:
    image: grafana/loki:2.9.0
    ports:
      - "3100:3100"
    command: -config.file=/etc/loki/local-config.yaml

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: "0.0.0.0:4317"
      http:
        endpoint: "0.0.0.0:4318"

processors:
  # Remove sensitive attributes before exporting
  attributes/remove-pii:
    actions:
      - key: user.email
        action: delete
      - key: payment.card_number
        action: delete
  
  # Add environment label to all telemetry
  resource:
    attributes:
      - key: environment
        value: production
        action: upsert

  # Memory limit to prevent OOM
  memory_limiter:
    check_interval: 5s
    limit_mib: 400

  batch:
    timeout: 5s
    send_batch_size: 512

exporters:
  # Traces → Tempo
  otlp/tempo:
    endpoint: tempo:4317
    tls:
      insecure: true

  # Metrics → Prometheus
  prometheusremotewrite:
    endpoint: http://prometheus:9090/api/v1/write

  # Logs → Loki
  loki:
    endpoint: http://loki:3100/loki/api/v1/push
    labels:
      resource:
        - service.name

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, attributes/remove-pii, resource, batch]
      exporters: [otlp/tempo]
    
    metrics:
      receivers: [otlp]
      processors: [memory_limiter, resource, batch]
      exporters: [prometheusremotewrite]
    
    logs:
      receivers: [otlp]
      processors: [memory_limiter, attributes/remove-pii, resource, batch]
      exporters: [loki]

# Install telemetrygen
go install github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen@latest

# Or via Docker
docker pull ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen:latest

# Send 100 test traces to the collector
telemetrygen traces \
  --otlp-endpoint localhost:4317 \
  --otlp-insecure \
  --traces 100 \
  --service payments-api \
  --spans 5

# Generate test metrics
telemetrygen metrics \
  --otlp-endpoint localhost:4317 \
  --otlp-insecure \
  --metrics 50 \
  --metric-type Sum \
  --service payments-api

# Generate test logs
telemetrygen logs \
  --otlp-endpoint localhost:4317 \
  --otlp-insecure \
  --logs 200 \
  --service payments-api

# Send traces with a specific attribute you want to test filtering on
telemetrygen traces \
  --otlp-endpoint localhost:4317 \
  --otlp-insecure \
  --traces 10 \
  --service test-service \
  --otlp-attributes "user.email=secret@example.com,payment.card_number=4111111111111111"

# tests/collector/test_trace_routing.py
import time
import requests
import subprocess
import pytest

TEMPO_URL = "http://localhost:3200"
COLLECTOR_ENDPOINT = "localhost:4317"

def generate_traces(count: int, service: str, **attrs) -> None:
    attr_args = ",".join(f"{k}={v}" for k, v in attrs.items())
    cmd = [
        "telemetrygen", "traces",
        "--otlp-endpoint", COLLECTOR_ENDPOINT,
        "--otlp-insecure",
        "--traces", str(count),
        "--service", service,
    ]
    if attr_args:
        cmd += ["--otlp-attributes", attr_args]
    
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
    assert result.returncode == 0, f"telemetrygen failed: {result.stderr}"

def search_tempo(service: str) -> list:
    response = requests.get(
        f"{TEMPO_URL}/api/search",
        params={"q": f'{{resource.service.name="{service}"}}', "limit": 100}
    )
    response.raise_for_status()
    return response.json().get("traces", [])

class TestTraceRouting:
    def test_traces_reach_tempo(self):
        """Traces sent to collector should appear in Tempo"""
        service = f"test-svc-{int(time.time())}"
        
        generate_traces(5, service)
        time.sleep(10)  # allow collector to batch and export
        
        traces = search_tempo(service)
        assert len(traces) >= 1, (
            f"No traces for service '{service}' found in Tempo. "
            f"Collector may be dropping or misrouting traces."
        )

    def test_pii_attributes_stripped_before_tempo(self):
        """PII attributes should not appear in Tempo traces"""
        service = f"test-pii-{int(time.time())}"
        
        generate_traces(
            3, service,
            **{"user.email": "secret@example.com", "payment.card_number": "4111111111111111"}
        )
        time.sleep(10)
        
        traces = search_tempo(service)
        assert len(traces) >= 1, "No traces found"
        
        # Fetch the full trace and check attributes
        trace_id = traces[0]["traceID"]
        response = requests.get(f"{TEMPO_URL}/api/traces/{trace_id}")
        trace_data = response.json()
        
        # Walk all spans looking for PII attributes
        for batch in trace_data.get("batches", []):
            for scope_span in batch.get("scopeSpans", []):
                for span in scope_span.get("spans", []):
                    for attr in span.get("attributes", []):
                        assert attr["key"] != "user.email", (
                            "PII attribute 'user.email' found in Tempo — "
                            "attributes/remove-pii processor not working"
                        )
                        assert attr["key"] != "payment.card_number", (
                            "PII attribute 'payment.card_number' found in Tempo"
                        )

    def test_environment_attribute_added_to_traces(self):
        """Resource processor should add 'environment=production' to all traces"""
        service = f"test-env-{int(time.time())}"
        generate_traces(3, service)
        time.sleep(10)
        
        traces = search_tempo(service)
        assert len(traces) >= 1
        
        trace_id = traces[0]["traceID"]
        response = requests.get(f"{TEMPO_URL}/api/traces/{trace_id}")
        trace_data = response.json()
        
        # Check resource attributes on the first batch
        resource_attrs = {}
        for batch in trace_data.get("batches", []):
            for attr in batch.get("resource", {}).get("attributes", []):
                resource_attrs[attr["key"]] = attr["value"].get("stringValue")
        
        assert resource_attrs.get("environment") == "production", (
            f"Expected environment=production in resource attributes, "
            f"got: {resource_attrs.get('environment')}"
        )

# tests/collector/test_metric_routing.py
import time
import requests
import subprocess

PROMETHEUS_URL = "http://localhost:9090"

def generate_metrics(count: int, service: str) -> None:
    subprocess.run([
        "telemetrygen", "metrics",
        "--otlp-endpoint", "localhost:4317",
        "--otlp-insecure",
        "--metrics", str(count),
        "--metric-type", "Sum",
        "--service", service,
    ], check=True, timeout=30)

def query_prometheus(query: str) -> list:
    response = requests.get(
        f"{PROMETHEUS_URL}/api/v1/query",
        params={"query": query}
    )
    return response.json()["data"]["result"]

class TestMetricRouting:
    def test_metrics_reach_prometheus(self):
        """Metrics sent to collector should appear in Prometheus"""
        service = f"test-metric-svc-{int(time.time())}"
        
        generate_metrics(20, service)
        time.sleep(15)  # allow remote write
        
        # telemetrygen generates a metric called "gen"
        results = query_prometheus(
            f'gen{{service_name="{service}"}}'
        )
        
        assert len(results) > 0, (
            f"No metrics for service '{service}' found in Prometheus. "
            f"Metrics pipeline may be dropping data."
        )

    def test_environment_label_on_metrics(self):
        """Resource processor should add environment label to metrics"""
        service = f"test-env-metric-{int(time.time())}"
        generate_metrics(10, service)
        time.sleep(15)
        
        results = query_prometheus(
            f'gen{{service_name="{service}", environment="production"}}'
        )
        
        assert len(results) > 0, (
            "Metrics missing 'environment=production' label — "
            "resource processor not adding label to metrics pipeline"
        )

# otel-collector-with-sampling.yaml (partial)
processors:
  tail_sampling:
    decision_wait: 10s
    policies:
      - name: errors-policy
        type: status_code
        status_code:
          status_codes: [ERROR]
      - name: slow-traces-policy
        type: latency
        latency:
          threshold_ms: 500
      - name: probabilistic-fallback
        type: probabilistic
        probabilistic:
          sampling_percentage: 10

def test_error_traces_always_sampled(self):
    """Error traces should reach Tempo regardless of sampling rate"""
    service = f"test-sampling-error-{int(time.time())}"
    
    # Generate 50 traces with error status codes
    subprocess.run([
        "telemetrygen", "traces",
        "--otlp-endpoint", "localhost:4317",
        "--otlp-insecure",
        "--traces", "50",
        "--service", service,
        "--otlp-attributes", "otel.status_code=ERROR",
    ], check=True)
    
    time.sleep(20)  # allow tail sampling decision window
    
    traces = search_tempo(service)
    # All error traces should be kept
    assert len(traces) >= 40, (
        f"Expected ~50 error traces, got {len(traces)}. "
        f"Error sampling policy may not be working."
    )

def test_success_traces_sampled_at_10_percent(self):
    """Success traces should be sampled at ~10% (probabilistic fallback)"""
    service = f"test-sampling-success-{int(time.time())}"
    
    # Generate 200 success traces
    subprocess.run([
        "telemetrygen", "traces",
        "--otlp-endpoint", "localhost:4317",
        "--otlp-insecure",
        "--traces", "200",
        "--service", service,
    ], check=True)
    
    time.sleep(30)  # tail sampling needs time
    
    traces = search_tempo(service)
    # With 10% sampling, expect 15-25 traces (allowing for variance)
    assert 10 <= len(traces) <= 40, (
        f"Expected ~20 success traces (10% of 200), got {len(traces)}. "
        f"Probabilistic sampling may be misconfigured."
    )

# .github/workflows/collector-config-test.yml
name: Test OTel Collector Config

on:
  pull_request:
    paths:
      - 'otel-collector-config.yaml'
      - 'tests/collector/**'

jobs:
  validate-config:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Validate collector config syntax
        run: |
          docker run --rm \
            -v $(pwd)/otel-collector-config.yaml:/etc/otel/config.yaml \
            otel/opentelemetry-collector-contrib:0.91.0 \
            validate --config /etc/otel/config.yaml

  integration-tests:
    runs-on: ubuntu-latest
    needs: validate-config
    steps:
      - uses: actions/checkout@v4

      - name: Install telemetrygen
        run: |
          go install github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen@v0.91.0
          echo "$(go env GOPATH)/bin" >> $GITHUB_PATH

      - name: Start test infrastructure
        run: docker compose -f docker-compose.collector-test.yml up -d

      - name: Wait for all services
        run: |
          timeout 60 bash -c 'until curl -sf http://localhost:13133/; do sleep 2; done'
          echo "Collector ready"
          timeout 60 bash -c 'until curl -sf http://localhost:3200/ready; do sleep 2; done'
          echo "Tempo ready"
          timeout 60 bash -c 'until curl -sf http://localhost:9090/-/ready; do sleep 2; done'
          echo "Prometheus ready"

      - name: Run collector integration tests
        run: |
          pip install pytest requests
          pytest tests/collector/ -v --tb=short -x

      - name: Dump collector logs on failure
        if: failure()
        run: |
          docker compose -f docker-compose.collector-test.yml logs otelcol

      - name: Teardown
        if: always()
        run: docker compose -f docker-compose.collector-test.yml down -v

def test_collector_not_dropping_data(self):
    """Collector exporter should have zero failed sends"""
    # Generate some test data first
    generate_traces(50, "test-drop-check")
    time.sleep(15)
    
    # Check collector's own metrics
    response = requests.get("http://localhost:8888/metrics")
    metrics_text = response.text
    
    # Parse the exporter metrics
    # A non-zero value here means the collector is dropping data
    for line in metrics_text.split('\n'):
        if 'otelcol_exporter_send_failed_spans_total' in line and not line.startswith('#'):
            parts = line.split()
            value = float(parts[-1])
            assert value == 0.0, (
                f"Collector has {value} failed span exports — "
                f"check backend connectivity and exporter config"
            )