Verify OpenTelemetry Redaction Without Losing Useful Traces

id: otel-redaction-export-verification
namespace: company.observability
description: Run candidate redaction options through a real isolated
  OpenTelemetry Collector and verify exported trace span attributes without
  dropping useful data.
labels:
  team: platform
  use-case: telemetry-redaction
inputs:
  - id: policy_yaml
    type: STRING
    defaults: |
      allow_all_keys: false
      allowed_keys: [test.case, http.route, app.order_id, description]
      blocked_values:
        - '[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}'
        - 'CANARY_TOKEN_[A-Za-z0-9]+'
      summary: silent
    description: 'Redaction processor options only; 1–65536 bytes, no external
      references. Supported: allow_all_keys, allowed_keys, blocked_values,
      summary (silent).'
  - id: scenario
    type: SELECT
    defaults: POLICY
    values:
      - POLICY
      - LEAK
      - DROP_SAFE
    description: POLICY tests your options. LEAK and DROP_SAFE replace the real
      Collector policy to demonstrate failure detection.
  - id: timeout_seconds
    type: INT
    defaults: 30
    description: Readiness and export observation timeout, each 5–120 seconds;
      excludes image pulls.
  - id: notify_slack
    type: BOOL
    defaults: false
    description: Send failure notifications only when enabled; requires SLACK_WEBHOOK.
  - id: kestra_url
    type: STRING
    defaults: http://localhost:8080
    description: Kestra URL for failure notification execution links.
variables:
  python_image: python:3.13-slim@sha256:3dd7cc108ec1493442514f5c2a871af6af0ec31d768ff6e378a93340c3b3db5f
  collector_image: otel/opentelemetry-collector-contrib:0.163.2@sha256:7f22d2a6748110ffb6a59bc0df448bccbf501cdb667d610ede4c61d3863d5849
  docker_cli_image: docker:27.5.1-cli@sha256:851f91d241214e7c6db86513b270d58776379aacc5eb9c4a87e5b47115e3065c
  capture_script: |
    import gzip, json, os, threading
    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
    from pathlib import Path

    root = Path('/data')
    root.mkdir(exist_ok=True)
    lock = threading.Lock()

    class Capture(BaseHTTPRequestHandler):
        def log_message(self, *args):
            pass

        def do_GET(self):
            if self.path == '/batches':
                with lock:
                    path = root / 'batches.jsonl'
                    batches = [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else []
                    body = json.dumps({'batches': batches, 'invalid_export': (root / 'invalid-export').exists()}).encode()
                self.send_response(200)
                self.send_header('Content-Type', 'application/json')
                self.end_headers()
                self.wfile.write(body)
            else:
                self.send_response(200 if self.path == '/health' else 404)
                self.end_headers()

        def do_POST(self):
            if self.path != '/v1/traces':
                self.send_error(404)
                return
            size = int(self.headers.get('Content-Length', 0))
            if not 0 < size <= 1048576:
                self.send_error(413)
                return
            try:
                raw = self.rfile.read(size)
                if self.headers.get('Content-Encoding') == 'gzip':
                    raw = gzip.decompress(raw)
                data = json.loads(raw)
                if not isinstance(data, dict) or not isinstance(data.get('resourceSpans'), list):
                    raise ValueError('Invalid OTLP trace envelope')
                with lock:
                    with (root / 'batches.jsonl').open('a') as stream:
                        stream.write(json.dumps(data) + '\n')
                self.send_response(200)
                self.send_header('Content-Type', 'application/json')
                self.end_headers()
                self.wfile.write(b'{}')
            except Exception:
                (root / 'invalid-export').touch()
                self.send_error(400)

    ThreadingHTTPServer(('0.0.0.0', 4318), Capture).serve_forever()
tasks:
  - id: harness
    type: io.kestra.plugin.core.flow.Sequential
    tasks:
      - id: prepare_harness
        type: io.kestra.plugin.scripts.python.Script
        description: Validate policy options and configure the real Collector and
          persistent capture service. Probe logic runs in native Python tasks.
        containerImage: '{{ vars.python_image }}'
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        env:
          POLICY: '{{ inputs.policy_yaml }}'
          SCENARIO: '{{ inputs.scenario }}'
          TIMEOUT: '{{ inputs.timeout_seconds }}'
          PYTHON_IMAGE: '{{ vars.python_image }}'
          COLLECTOR_IMAGE: '{{ vars.collector_image }}'
          CAPTURE_SCRIPT: '{{ vars.capture_script }}'
        outputFiles:
          - compose.json
        script: |
          import hashlib, json, os
          from pathlib import Path
          import yaml

          text = os.environ['POLICY']
          if not 0 < len(text.encode()) <= 65536 or '${' in text:
              raise ValueError('Policy must be 1–65536 bytes, without external configuration references')
          policy = yaml.safe_load(text)
          allowed = {'allow_all_keys', 'allowed_keys', 'blocked_values', 'summary'}
          if not isinstance(policy, dict) or set(policy) - allowed:
              raise ValueError('Provide only allow_all_keys, allowed_keys, blocked_values and summary')
          if type(policy.get('allow_all_keys', False)) is not bool:
              raise ValueError('allow_all_keys must be a Boolean')
          for field in ['allowed_keys', 'blocked_values']:
              value = policy.get(field, [])
              if not isinstance(value, list) or any(not isinstance(v, str) or len(v) > 2048 for v in value):
                  raise ValueError(field + ' must be a list of bounded strings')
          if policy.get('summary', 'silent') != 'silent':
              raise ValueError('Use summary: silent to avoid exporting redaction audit metadata')
          scenario = os.environ['SCENARIO']
          if scenario == 'LEAK':
              policy = {'allow_all_keys': True, 'summary': 'silent'}
          elif scenario == 'DROP_SAFE':
              policy = {'allow_all_keys': False, 'allowed_keys': ['test.case'], 'summary': 'silent'}
          elif scenario != 'POLICY':
              raise ValueError('Unknown scenario')
          limit = int(os.environ['TIMEOUT'])
          if not 5 <= limit <= 120:
              raise ValueError('timeout_seconds must be 5–120')
          policy_hash = hashlib.sha256(json.dumps(policy, sort_keys=True).encode()).hexdigest()
          config = {'receivers': {'otlp': {'protocols': {'http': {'endpoint': '0.0.0.0:4318'}}}},
                    'processors': {'redaction': policy},
                    'exporters': {'otlp_http': {'endpoint': 'http://capture:4318', 'encoding': 'json',
                                              'compression': 'none', 'timeout': '5s',
                                              'retry_on_failure': {'enabled': True, 'initial_interval': '1s', 'max_interval': '2s', 'max_elapsed_time': '10s'}}},
                    'extensions': {'health_check': {'endpoint': '0.0.0.0:13133'}},
                    'service': {'extensions': ['health_check'], 'pipelines': {'traces': {
                        'receivers': ['otlp'], 'processors': ['redaction'], 'exporters': ['otlp_http']}}}}
          python = os.environ['PYTHON_IMAGE']
          services = {'capture': {'image': python, 'command': ['python', '-u', '-c', os.environ['CAPTURE_SCRIPT']], 'volumes': ['capture:/data']},
                      'collector': {'image': os.environ['COLLECTOR_IMAGE'], 'command': ['--config=env:HARNESS_CONFIG'],
                                    'environment': {'HARNESS_CONFIG': yaml.safe_dump(config)}}}
          # Escape Compose interpolation, including literal dollar anchors in candidate regexes.
          compose = json.dumps({'services': services, 'volumes': {'capture': {}}, 'networks': {'default': {'internal': True}}}).replace('$', '$$')
          Path('compose.json').write_text(compose)
          print('Prepared isolated harness with policy SHA256 ' + policy_hash)

          print('::' + json.dumps({'outputs': {'policy_sha256': policy_hash}}) + '::')
        beforeCommands:
          - python -m pip install --disable-pip-version-check --no-cache-dir
            PyYAML==6.0.2
      - id: start_harness
        type: io.kestra.plugin.docker.cli.Compose
        description: Start a real Collector and capture receiver on an
          execution-specific internal network without host ports.
        containerImage: '{{ vars.docker_cli_image }}'
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          user: 0:0
          volumes:
            - /var/run/docker.sock:/var/run/docker.sock
        composeFile: '{{ outputs.prepare_harness.outputFiles[''compose.json''] }}'
        composeArgs:
          - --project-name
          - otel-{{ execution.id | lower }}
          - up
          - --detach
          - capture
          - collector
        timeout: PT5M
        retry:
          type: constant
          interval: PT2S
          maxAttempts: 3
      - id: send_canaries
        type: io.kestra.plugin.scripts.python.Script
        description: Submit real OTLP traces from an inline Python task and persist
          expected identities as JSON. Reject partial-success responses.
        containerImage: '{{ vars.python_image }}'
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          networkMode: otel-{{ execution.id | lower }}_default
        env:
          TIMEOUT: '{{ inputs.timeout_seconds }}'
        script: |
          import json, os, time, urllib.request
          from pathlib import Path

          limit = int(os.environ['TIMEOUT'])
          deadline = time.monotonic() + limit
          while True:
              try:
                  with urllib.request.urlopen('http://collector:13133/', timeout=2) as response:
                      if response.status == 200:
                          break
              except Exception:
                  if time.monotonic() >= deadline:
                      raise RuntimeError('Collector did not become ready')
                  time.sleep(0.5)

          cases = []
          for number, extra in enumerate([
              {'user.email': 'canary.person@example.invalid'},
              {'description': 'contact canary.person@example.invalid'},
              {'description': 'credential CANARY_TOKEN_1234567890'},
          ]):
              cases.append({'traceId': f'{number+1:032x}', 'spanId': f'{number+1:016x}',
                            'safe': {'test.case': f'case-{number+1}', 'http.route': '/orders/{id}', 'app.order_id': 'order-42'},
                            'extra': extra})
          Path('expected.json').write_text(json.dumps(cases))
          spans = []
          now = time.time_ns()
          for case in cases:
              spans.append({'traceId': case['traceId'], 'spanId': case['spanId'], 'name': 'redaction-check',
                            'kind': 1, 'startTimeUnixNano': str(now), 'endTimeUnixNano': str(now+1000000),
                            'attributes': [{'key': key, 'value': {'stringValue': value}}
                                           for key, value in (case['safe'] | case['extra']).items()]})
          payload = json.dumps({'resourceSpans': [{'resource': {'attributes': []},
                                'scopeSpans': [{'scope': {'name': 'kestra-redaction-test'}, 'spans': spans}]}]}).encode()
          for attempt in range(3):
              try:
                  request = urllib.request.Request('http://collector:4318/v1/traces', data=payload,
                                                  headers={'Content-Type': 'application/json'})
                  with urllib.request.urlopen(request, timeout=5) as response:
                      result = json.load(response)
                  partial = result.get('partialSuccess', {})
                  if int(partial.get('rejectedSpans', 0)) or partial.get('errorMessage'):
                      raise ValueError('Collector reported partial success')
                  break
              except (OSError, TimeoutError):
                  if attempt == 2:
                      raise
                  time.sleep(1)
          print('Submitted three synthetic trace cases to the real Collector.')
        outputFiles:
          - expected.json
        timeout: PT3M
      - id: verify_export
        type: io.kestra.plugin.scripts.python.Script
        description: Read the actual exports over the private capture API; check leaks,
          missing spans and safe attributes, then persist reports before
          enforcing the verdict.
        containerImage: '{{ vars.python_image }}'
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          networkMode: otel-{{ execution.id | lower }}_default
        env:
          TIMEOUT: '{{ inputs.timeout_seconds }}'
          POLICY_SHA256: '{{ outputs.prepare_harness.vars.policy_sha256 }}'
          COLLECTOR_IMAGE: '{{ vars.collector_image }}'
          SCENARIO: '{{ inputs.scenario }}'
        inputFiles:
          expected.json: '{{ outputs.send_canaries.outputFiles[''expected.json''] }}'
        script: |
          import json, os, time, urllib.request
          from pathlib import Path

          root = Path('.')
          cases = json.loads((root / 'expected.json').read_text())
          expected = {(case['traceId'], case['spanId']): case for case in cases}
          deadline = time.monotonic() + int(os.environ['TIMEOUT'])
          settle = None
          while True:
              errors, seen, batches = [], set(), []
              try:
                  with urllib.request.urlopen('http://capture:4318/batches', timeout=2) as response:
                      capture = json.load(response)
                  batches = capture['batches']
                  if capture['invalid_export']:
                      errors.append({'kind': 'INVALID_EXPORT'})
              except (OSError, ValueError, KeyError):
                  errors.append({'kind': 'INVALID_EXPORT', 'detail': 'Capture service unavailable or malformed'})
              for batch in batches:
                  encoded = json.dumps(batch)
                  for marker in ['canary.person@example.invalid', 'CANARY_TOKEN_1234567890']:
                      if marker in encoded:
                          errors.append({'kind': 'LEAK', 'detail': 'Synthetic sensitive value reached exporter'})
                  for resource in batch.get('resourceSpans', []):
                      for scope in resource.get('scopeSpans', []):
                          for span in scope.get('spans', []):
                              identity = (span.get('traceId', '').lower(), span.get('spanId', '').lower())
                              if identity not in expected:
                                  errors.append({'kind': 'INVALID_EXPORT', 'detail': 'Unexpected span identity'})
                                  continue
                              seen.add(identity)
                              attributes = {a['key']: a['value'].get('stringValue') for a in span.get('attributes', [])}
                              if 'user.email' in attributes:
                                  errors.append({'kind': 'LEAK', 'case': expected[identity]['safe']['test.case'], 'detail': 'Forbidden key survived'})
                              for key, value in expected[identity]['safe'].items():
                                  if attributes.get(key) != value:
                                      errors.append({'kind': 'SAFE_FIELD_LOSS', 'key': key, 'case': expected[identity]['safe']['test.case']})
              if seen == set(expected) and settle is None:
                  settle = time.monotonic() + 2
              if (settle is not None and time.monotonic() >= settle) or time.monotonic() >= deadline:
                  break
              time.sleep(0.25)
          for identity in set(expected) - seen:
              errors.append({'kind': 'MISSING_SPANS', 'case': expected[identity]['safe']['test.case']})
          unique = {json.dumps(error, sort_keys=True): error for error in errors}
          report = {'verdict': 'PASS' if not unique else 'FAIL', 'expected_spans': len(expected),
                    'exported_spans': len(seen), 'errors': list(unique.values()),
                    'policy_sha256': os.environ['POLICY_SHA256'], 'collector_image': os.environ['COLLECTOR_IMAGE'],
                    'scenario': os.environ['SCENARIO']}
          (root / 'report.json').write_text(json.dumps(report, indent=2))
          (root / 'report.md').write_text('# OpenTelemetry redaction verification\n\n'
              + f"Verdict: **{report['verdict']}**\n\nSpans received: {len(seen)}/{len(expected)}\n\n"
              + '\n'.join('- ' + json.dumps(e) for e in unique.values())
              + '\n\nPolicy SHA256: ' + report['policy_sha256'] + '\n')
          print(f"Verified real exported traces: {report['verdict']}; report saved before policy enforcement.")

          print('::' + json.dumps({'outputs': {'passed': report['verdict'] == 'PASS',
                'verdict': report['verdict'], 'exported_spans': len(seen),
                'error_count': len(report['errors']), 'policy_sha256': report['policy_sha256']}}) + '::')
        outputFiles:
          - report.json
          - report.md
        timeout: PT3M
      - id: enforce_policy
        type: io.kestra.plugin.core.execution.Assert
        description: Block a leaking or over-aggressive redaction policy; a zero-span
          result cannot pass.
        conditions:
          - '{{ outputs.verify_export.vars.passed }}'
    finally:
      - id: cleanup
        type: io.kestra.plugin.docker.cli.Compose
        description: Remove only this execution’s test containers, network and volume,
          including after a failed assertion.
        containerImage: '{{ vars.docker_cli_image }}'
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          user: 0:0
          volumes:
            - /var/run/docker.sock:/var/run/docker.sock
        composeFile: '{"services":{"capture":{"image":"python:3.13-slim","volumes":["capture:/data"]},"collector":{"image":"otel/opentelemetry-collector-contrib:0.163.2"}},"volumes":{"capture":{}},"networks":{"default":{}}}'
        composeArgs:
          - --project-name
          - otel-{{ execution.id | lower }}
          - down
          - --volumes
          - --remove-orphans
        timeout: PT5M
        retry:
          type: constant
          interval: PT2S
          maxAttempts: 3
errors:
  - id: notify_failure
    type: io.kestra.plugin.core.flow.If
    condition: '{{ inputs.notify_slack }}'
    then:
      - id: slack_failure
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: '{{ secret(''SLACK_WEBHOOK'') }}'
        messageText: |-
          Redaction verification failed: {{ flow.namespace }}.{{ flow.id }}, execution {{ execution.id }}.
          Failed task: {{ (errorLogs() | length) > 0 ? errorLogs()[0].taskId : 'unknown' }}.
          Inspect the captured report and logs: {{ inputs.kestra_url }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}
        retry:
          type: constant
          interval: PT2S
          maxAttempts: 3
triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    cron: 0 6 * * *
    timezone: UTC
    disabled: true
outputs:
  - id: report
    type: FILE
    value: '{{ outputs.verify_export.outputFiles[''report.json''] ?? null }}'
  - id: readable_report
    type: FILE
    value: '{{ outputs.verify_export.outputFiles[''report.md''] ?? null }}'