observability-stack/docker-compose/data-prepper/pipelines.template.yaml at main · opensearch-project/observability-stack · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
# Data Prepper Pipeline Configuration
# Processes and transforms logs, traces, and metrics before writing to OpenSearch

# Main OTLP pipeline - receives all telemetry and routes by type
otlp-pipeline:
  delay: 10
  source:
    # OTLP source receives telemetry via OpenTelemetry Protocol
    otlp:
      # Listen on port 21890 for gRPC connections from OpenTelemetry Collector
      port: 21890
      # Disable SSL for development
      ssl: false
  # Route telemetry by signal type
  route:
    - logs: "getEventType() == \"LOG\""
    - traces: "getEventType() == \"TRACE\""
  # Send to appropriate sub-pipelines
  sink:
    - pipeline:
        name: "otel-logs-pipeline"
        routes:
          - "logs"
    - pipeline:
        name: "otel-traces-pipeline"
        routes:
          - "traces"

# Log processing pipeline
# Receives logs from main pipeline and writes to OpenSearch
otel-logs-pipeline:
  workers: 5
  delay: 10
  source:
    pipeline:
      name: "otlp-pipeline"
  buffer:
    bounded_blocking:
  processor:
    - copy_values:
        entries:
          - from_key: "time"
            to_key: "@timestamp"
  # Write processed logs to OpenSearch
  sink:
    - opensearch:
        hosts: ["OPENSEARCH_PROTOCOL://OPENSEARCH_HOST:OPENSEARCH_PORT"]
        username: OPENSEARCH_USER
        password: OPENSEARCH_PASSWORD
        # Disable SSL verification for development
        insecure: true
        # Use log analytics index type for automatic index management
        index_type: log-analytics-plain

# Trace processing pipeline
# Receives traces from main pipeline and distributes to raw and service map pipelines
otel-traces-pipeline:
  delay: 100
  source:
    pipeline:
      name: "otlp-pipeline"
  # Send traces to both raw storage and service map generation
  sink:
    - pipeline:
        name: "traces-raw-pipeline"
    - pipeline:
        name: "service-map-pipeline"

# Raw trace storage pipeline
# Stores individual trace spans in OpenSearch
traces-raw-pipeline:
  source:
    pipeline:
      name: "otel-traces-pipeline"
  processor:
    # Process raw trace data for OpenSearch storage
    - otel_traces:
  sink:
    - opensearch:
        hosts: ["OPENSEARCH_PROTOCOL://OPENSEARCH_HOST:OPENSEARCH_PORT"]
        username: OPENSEARCH_USER
        password: OPENSEARCH_PASSWORD
        insecure: true
        # Use trace analytics index type for automatic index management
        index_type: trace-analytics-plain-raw

# Service map generation pipeline (APM)
# Builds service dependency maps and RED metrics from trace relationships
service-map-pipeline:
  delay: 100
  source:
    pipeline:
      name: "otel-traces-pipeline"
  processor:
    # APM service map processor - generates topology, service details, and RED metrics
    - otel_apm_service_map:
        group_by_attributes: [telemetry.sdk.language]
        window_duration: 10s
  route:
    - otel_apm_service_map_route: 'getEventType() == "SERVICE_MAP"'
    - service_processed_metrics: 'getEventType() == "METRIC"'
  sink:
    - opensearch:
        hosts: ["OPENSEARCH_PROTOCOL://OPENSEARCH_HOST:OPENSEARCH_PORT"]
        username: OPENSEARCH_USER
        password: OPENSEARCH_PASSWORD
        index_type: otel-v2-apm-service-map
        routes: [otel_apm_service_map_route]
        insecure: true
    # Route RED metrics to local Prometheus via remote write
    - prometheus:
        url: "http://PROMETHEUS_HOST:PROMETHEUS_PORT/api/v1/write"
        threshold:
          max_events: 500
          flush_interval: 5s
        routes: [service_processed_metrics]