Step 3: Multi-Destination Routing
Route each processed record to the three destinations provisioned on the SNO cluster. The fragments below abbreviate the complete job spec; use that spec for deployment.
The Fan-Out Pattern
Each output keeps its own retry and batching policy. The collector and Kafka outputs fall back to dead-letter files after delivery fails. Parquet has no fallback, and a full or unavailable claim can block progress across the fan-out.
output:
broker:
pattern: fan_out
outputs:
- label: otel_collector # OTLP/HTTP, scraped by user-workload monitoring
drop: {}
- label: parquet_pvc # Parquet files under /data
drop: {}
- label: amq_streams_kafka # TLS Kafka on the cluster
drop: {}
Output-level processors shape the record for each destination before it is sent.
Output 1: OpenTelemetry Collector and User-Workload Monitoring
OpenShift runs no Prometheus Pushgateway, and a Pushgateway would not accept JSON anyway. The pipeline posts OTLP/HTTP JSON to the OpenTelemetry Collector in the namespace. The collector's Prometheus exporter is what user-workload monitoring scrapes, through the ServiceMonitor in the manifests.
- label: otel_collector
processors:
- mapping: |
root = {
"resourceMetrics": [{
"resource": {
"attributes": [
{"key": "du_id", "value": {"stringValue": this.du_id}},
{"key": "cell_id", "value": {"stringValue": this.cell_id}},
{"key": "gnb_id", "value": {"stringValue": this.gnb_id}},
{"key": "region", "value": {"stringValue": this.region}}
]
},
"scopeMetrics": [{
"scope": {"name": "oran.telemetry", "version": "2.0.0"},
"metrics": [
{
"name": "oran_ptp_review_score",
"unit": "1",
"gauge": {"dataPoints": [{
"timeUnixNano": this.timestamp_ns.string(),
"asDouble": this.ptp_score
}]}
}
]
}]
}]
}
fallback:
- http_client:
url: '${OTEL_ENDPOINT:http://oran-collector.oran-telemetry.svc:4318}/v1/metrics'
verb: POST
headers:
Content-Type: application/json
timeout: 30s
retries: 3
retry_period: 1s
max_retry_backoff: 30s
- file:
path: '/data/dead-letter/otel-${! now().ts_format("2006-01-02") }.jsonl'
codec: lines
processors:
- mapping: |
root = {
"error_timestamp": now(),
"error_destination": "otel_collector",
"error_message": meta("fallback_error"),
"original_payload": this
}
The full mapping also emits oran_ptp_offset_nanoseconds, oran_prb_utilization_percent (labelled by direction), oran_cpu_utilization_percent, and oran_rf_health_score. The collector keeps resource attributes as labels and leaves metric names unchanged (add_metric_suffixes: false), so the queries in Step 4 use these names as written. The hop to oran-collector:4318 never leaves the node.
Output 2: Parquet on the PersistentVolumeClaim
The file output takes only a path and a codec. A nested broker batches records, parquet_encode turns each batch into one Parquet file in memory, and file writes that single binary message with codec: all-bytes.
- label: parquet_pvc
processors:
- mapping: |
root = {
"timestamp_ns": this.timestamp_ns,
"date": this.timestamp_ns.ts_format("2006-01-02"),
"du_id": this.du_id,
"cell_id": this.cell_id,
"gnb_id": this.gnb_id,
"region": this.region,
"ptp_offset_ns": this.ptp4l_offset_ns,
"ptp_review": this.ptp_review,
"prb_dl_pct": this.prb_dl_pct,
"prb_ul_pct": this.prb_ul_pct,
"cpu_pct": this.cpu_pct,
"rsrp_dbm": this.rsrp_dbm,
"sinr_db": this.sinr_db,
"rf_health_score": this.rf_health_score,
"data_quality_score": this.data_quality_score
}
broker:
batching:
count: 10000
period: 5m
byte_size: 10000000
processors:
- parquet_encode:
default_compression: zstd
schema:
- { name: timestamp_ns, type: INT64 }
- { name: date, type: UTF8 }
- { name: du_id, type: UTF8 }
- { name: ptp_offset_ns, type: INT64, optional: true }
- { name: ptp_review, type: UTF8, optional: true }
- { name: prb_dl_pct, type: DOUBLE, optional: true }
# ... one entry per column above
outputs:
- file:
path: '/data/oran-telemetry/date=${! now().ts_format("2006-01-02") }/node=${! env("NODE_ID").or("unknown") }/part-${! uuid_v4() }.parquet'
codec: all-bytes
/data is the oran-telemetry-data claim. The restricted-v2 SCC forbids host paths, and the claim is what survives a pod reschedule. Step 5 covers the schema, read-back, and retention.
Output 3: AMQ Streams Kafka over TLS
The Kafka resource exposes one listener: TLS on 9093 with SCRAM-SHA-512. The operator publishes the cluster CA in Secret oran-kafka-cluster-ca-cert, mounted at /etc/expanso/kafka-ca, and the KafkaUser writes the password the agent reads as KAFKA_PASSWORD.
- label: amq_streams_kafka
processors:
- mapping: |
root = {
"schema": {"type": "oran_telemetry", "version": "2.0.0"},
"metadata": {
"event_time": this.timestamp_iso,
"data_quality_score": this.data_quality_score
},
"network": this.network_hierarchy,
"telemetry": {
"timing": {"ptp_offset_ns": this.ptp4l_offset_ns, "review": this.ptp_review},
"resources": {"prb_dl_percent": this.prb_dl_pct, "prb_ul_percent": this.prb_ul_pct},
"system": {"cpu_percent": this.cpu_pct, "memory_percent": this.memory_pct},
"radio": {"rsrp_dbm": this.rsrp_dbm, "sinr_db": this.sinr_db}
}
}
fallback:
- kafka:
addresses:
[
'${KAFKA_BOOTSTRAP_SERVERS:oran-kafka-kafka-bootstrap.oran-telemetry.svc:9093}',
]
topic: oran-telemetry-structured
key: '${! this.network.site_id }-${! this.network.cell_id }'
client_id: expanso-oran-pipeline
tls:
enabled: true
root_cas_file: /etc/expanso/kafka-ca/ca.crt
sasl:
mechanism: SCRAM-SHA-512
user: '${KAFKA_USERNAME}'
password: '${KAFKA_PASSWORD}'
compression: lz4
idempotent_write: false
ack_replicas: true
max_msg_bytes: 10485760
batching:
count: 500
period: 5s
- file:
path: '/data/dead-letter/kafka-${! now().ts_format("2006-01-02") }.jsonl'
codec: lines
The producer ACLs permit Write and Describe on the two telemetry topics. Idempotent writes are disabled to avoid requiring cluster-level IdempotentWrite permission. Plaintext 9092 is not configured on the broker.
Dead-Letter Files and Replay
When a first tier fails after its retries, the record moves to the file tier with fallback_error set in metadata and lands under /data/dead-letter/. Replay through a separate Cloud-deployed job that extracts original_payload and sends it to the corresponding transport. The payload already has the destination shape; do not apply the original output mapping again. Delete the file only after confirming delivery.
Monitoring Pipeline Health
Three metric processors count records, time processing latency, and gauge the data quality score. The agent exports them, with its own process metrics, over OTLP gRPC to the same collector through the telemetry block of expanso-edge-config.
- label: pipeline_metrics
metric:
type: counter
name: oran_telemetry_messages_processed
labels:
du_id: '${! this.du_id }'
collection_method: '${! this.collection_method }'
- metric:
type: timing
name: oran_telemetry_processing_latency_ns
value: '${! now().ts_unix_nano() - this.timestamp_ns }'
- metric:
type: gauge
name: oran_telemetry_data_quality_score
value: '${! this.data_quality_score }'
Next Steps
- Step 4: Grafana dashboards against the user-workload Prometheus
- Step 5: Parquet and Cloudera for the schema, read-back, and the Cloudera consumer
- Run path to apply the manifests and deploy the job
