O-RAN Telemetry Reference
This checked-in job spec is what expanso-cli job deploy sends to Expanso Cloud. It passes expanso-edge validate, which checks the config schema, every Bloblang mapping, and the component types. A site-specific exporter or adapter is still required before real DU data reaches it; the built-in simulator runs alongside the real inputs until you remove it as described in the run path.
Complete pipeline
static/pipelines/oran-telco-pipeline.yaml
# O-RAN DU telemetry on Single-Node OpenShift (SNO)
#
# This is an Expanso job spec. Deploy it through Expanso Cloud:
#
# expanso-cli job deploy oran-telco-pipeline.yaml
#
# The job targets the Expanso Edge agent that runs on the SNO node as a
# restricted-SCC Deployment (see examples/integrations/oran-telco-pipeline/openshift/).
# Every path, host, and credential below is satisfied by those manifests:
#
# /data PersistentVolumeClaim oran-telemetry-data
# /etc/expanso/kafka-ca/ca.crt Secret oran-kafka-cluster-ca-cert (AMQ Streams)
# KAFKA_USERNAME/KAFKA_PASSWORD Secret expanso-oran (KafkaUser, SCRAM-SHA-512)
# OTEL_ENDPOINT Service oran-collector:4318 (OpenTelemetry Collector)
#
# Validate offline before deploying:
#
# expanso-edge validate oran-telco-pipeline.yaml
name: oran-edge-telemetry
description: Normalize DU telemetry on an SNO node and fan out to OpenTelemetry, Parquet on a PVC, and AMQ Streams Kafka
type: pipeline
namespace: default
labels:
example: oran-telco-pipeline
platform: openshift-sno
# Node labels are set in the edge agent's config (edge-config.yaml in the ConfigMap).
selector:
match_labels:
platform: openshift-sno
role: oran-du-telemetry
config:
rate_limit_resources:
- label: du_api_polling
local:
count: 1
interval: 1s
input:
broker:
inputs:
# 1. Adapter push. The DU exporter pod POSTs decoded telemetry to the
# agent's in-cluster Service (expanso-edge:8080). No hostPath is needed.
- label: du_adapter_push
http_server:
address: 0.0.0.0:8080
path: /telemetry
allowed_verbs: [POST]
timeout: 30s
processors:
- mapping: |
root = this
root.collection_method = "adapter_push"
root.collector_timestamp = now()
root.du_id = this.du_id.or(env("DU_ID")).or("du-unknown")
root.cell_id = this.cell_id.or(env("CELL_ID")).or("cell-unknown")
root.gnb_id = this.gnb_id.or(env("GNB_ID")).or("gnb-unknown")
# 2. DU API polling over the cluster network.
- label: du_api_realtime
http_client:
url: '${DU_ENDPOINT:http://oran-du.oran-du.svc:8080}/api/v1/metrics/realtime'
verb: GET
rate_limit: du_api_polling
headers:
Authorization: 'Bearer ${DU_API_KEY}'
Accept: application/json
User-Agent: expanso-edge-oran-collector/2.0.0
timeout: 30s
retries: 3
retry_period: 1s
max_retry_backoff: 30s
processors:
- mapping: |
root = this
root.collection_method = "api_realtime"
root.collector_timestamp = now()
root.du_id = this.du_id.or(env("DU_ID")).or("du-unknown")
root.cell_id = this.cell_id.or(env("CELL_ID")).or("cell-unknown")
root.gnb_id = this.gnb_id.or(env("GNB_ID")).or("gnb-unknown")
# 3. Simulated DU telemetry so the job produces data before a DU is wired in.
- label: du_simulator
generate:
interval: '${SIMULATION_INTERVAL:10s}'
count: 0
mapping: |
root.timestamp = now()
root.collection_method = "simulation"
let du_number = random_int(min: 1, max: 8).string()
root.du_id = "du-00" + $du_number
root.cell_id = "cell-" + $du_number
root.gnb_id = env("GNB_ID").or("gnb-seattle-01")
# PTP timing: +/-100 ns baseline with +/-10 ns jitter
root.ptp4l_offset_ns = (random_int(max: 200) - 100) + (random_int(max: 20) - 10)
# PRB utilization with a business-hours bump
let hour = now().ts_format("15").number()
let business_hour_factor = if $hour >= 8 && $hour <= 18 { 1.5 } else { 0.7 }
root.prb_dl_pct = [(random_int(min: 20, max: 70) * $business_hour_factor).round(), 95].min()
root.prb_ul_pct = [(random_int(min: 10, max: 40) * $business_hour_factor).round(), 85].min()
root.cpu_pct = random_int(min: 25, max: 85)
root.memory_pct = random_int(min: 40, max: 80)
root.disk_pct = random_int(min: 15, max: 60)
root.rsrp_dbm = 0 - random_int(min: 70, max: 110)
root.sinr_db = random_int(max: 27) - 2
root.active_ue_count = random_int(min: 5, max: 150)
root.network_rx_mbps = random_int(min: 50, max: 800)
root.network_tx_mbps = random_int(min: 20, max: 400)
pipeline:
processors:
# Drop records that cannot be attributed to a cell.
- label: input_validation
mapping: |
root = if this.du_id.or("") != "" && this.cell_id.or("") != "" && this.timestamp != null {
this
} else {
deleted()
}
# Normalize timestamps. RFC 3339 strings and Unix seconds are both accepted.
- label: timestamp_processing
mapping: |
let ts = this.timestamp.or(now())
root = this
root.timestamp_ns = $ts.ts_unix_nano()
root.timestamp_iso = $ts.ts_format("2006-01-02T15:04:05.000000000Z07:00")
root.processed_at = now()
root.hour_of_day = $ts.ts_format("15").number()
root.day_of_week = $ts.ts_format("Monday")
root.is_business_hours = root.hour_of_day >= 8 && root.hour_of_day <= 18
# Site metadata from the agent's environment (ConfigMap oran-site).
- label: site_enrichment
mapping: |
root = this
root.region = env("REGION").or("us-west-2")
root.market = env("MARKET").or("seattle")
root.site_type = env("SITE_TYPE").or("urban-macro")
root.gnb_id = if this.gnb_id.or("gnb-unknown") == "gnb-unknown" {
"gnb-" + root.region + "-" + this.du_id
} else {
this.gnb_id
}
root.sector = match this.cell_id {
"cell-1" => "alpha"
"cell-2" => "beta"
"cell-3" => "gamma"
_ => "omni"
}
root.network_hierarchy = {
"region": root.region,
"market": root.market,
"site_id": root.gnb_id,
"du_id": this.du_id,
"cell_id": this.cell_id,
"sector": root.sector
}
# PTP review bands (ITU-T G.8271.1 offsets). Authored thresholds, not a timing diagnosis.
- label: ptp_analysis
mapping: |
let abs_offset = this.ptp4l_offset_ns.or(0).abs()
root = this
root.ptp_review = if this.ptp4l_offset_ns == null {
deleted()
} else if $abs_offset < 100 {
"within_band"
} else if $abs_offset < 1000 {
"degraded"
} else if $abs_offset < 10000 {
"critical"
} else {
"outside_band"
}
root.ptp_score = if this.ptp4l_offset_ns == null {
deleted()
} else if $abs_offset < 100 {
100
} else if $abs_offset < 1000 {
(100 - (($abs_offset - 100) / 9)).round()
} else if $abs_offset < 10000 {
[(90 - (($abs_offset - 1000) / 100)).round(), 0].max()
} else {
0
}
root.ptp_alert_level = match root.ptp_review {
"within_band" => "none"
"degraded" => "warning"
"critical" => "major"
"outside_band" => "critical"
_ => deleted()
}
# PRB utilization and capacity bands.
- label: capacity_analysis
mapping: |
root = this
root = if this.prb_dl_pct == null || this.prb_ul_pct == null {
this
} else {
this.merge({
"prb_total_utilization": (this.prb_dl_pct + this.prb_ul_pct) / 2,
"prb_efficiency": ((this.prb_dl_pct + this.prb_ul_pct) / 2).round(),
"congestion_level": match {
this.prb_dl_pct < 50 => "low"
this.prb_dl_pct < 80 => "medium"
this.prb_dl_pct < 95 => "high"
_ => "critical"
},
"capacity_headroom_pct": 100 - this.prb_dl_pct,
"spectral_efficiency_dl": match {
this.prb_dl_pct < 30 => 2.4
this.prb_dl_pct < 70 => 3.2
this.prb_dl_pct < 90 => 2.8
_ => 1.8
},
"time_to_congestion_min": if this.prb_dl_pct > 70 {
((95 - this.prb_dl_pct) / 0.5).round()
} else {
null
}
})
}
# RF quality bands (3GPP TS 36.133 RSRP ranges, authored SINR ranges).
- label: rf_analysis
mapping: |
let rsrp_score = match {
this.rsrp_dbm == null => 0
this.rsrp_dbm > -80 => 100
this.rsrp_dbm > -90 => 80
this.rsrp_dbm > -100 => 60
this.rsrp_dbm > -110 => 40
_ => 20
}
let sinr_score = match {
this.sinr_db == null => 0
this.sinr_db > 20 => 100
this.sinr_db > 13 => 80
this.sinr_db > 0 => 60
this.sinr_db > -6 => 40
_ => 20
}
root = this
root = if this.rsrp_dbm == null || this.sinr_db == null {
this
} else {
this.merge({
"rsrp_quality": match {
this.rsrp_dbm > -80 => "excellent"
this.rsrp_dbm > -90 => "good"
this.rsrp_dbm > -100 => "fair"
this.rsrp_dbm > -110 => "poor"
_ => "very_poor"
},
"sinr_quality": match {
this.sinr_db > 20 => "excellent"
this.sinr_db > 13 => "good"
this.sinr_db > 0 => "fair"
this.sinr_db > -6 => "poor"
_ => "very_poor"
},
"rf_health_score": (($rsrp_score + $sinr_score) / 2).round(),
"estimated_coverage_radius_m": match {
this.rsrp_dbm > -80 => 500
this.rsrp_dbm > -90 => 1000
this.rsrp_dbm > -100 => 2000
this.rsrp_dbm > -110 => 3000
_ => 5000
}
})
}
# Data quality flags and pipeline metadata.
- label: data_quality
mapping: |
let quality = {
"ptp_valid": this.ptp4l_offset_ns != null,
"prb_valid": this.prb_dl_pct != null && this.prb_ul_pct != null,
"rf_valid": this.rsrp_dbm != null && this.sinr_db != null,
"system_valid": this.cpu_pct != null,
"timestamp_valid": this.timestamp_ns != null
}
let valid_count = $quality.values().filter(v -> v == true).length()
root = this
root.data_quality = $quality
root.data_quality_score = ($valid_count / 5) * 100
root.pipeline_metadata = {
"pipeline_id": "oran-edge-telemetry",
"pipeline_version": "2.0.0",
"processing_node": env("NODE_ID").or("unknown"),
"processing_timestamp": now().ts_format("2006-01-02T15:04:05Z07:00"),
"data_quality_score": root.data_quality_score
}
# Agent-side metrics. The edge agent exports these over OTLP to the same
# collector; see the telemetry block in edge-config.yaml.
- label: pipeline_metrics
metric:
type: counter
name: oran_telemetry_messages_processed
labels:
pipeline_id: oran-edge-telemetry
du_id: '${! this.du_id }'
region: '${! this.region }'
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 }'
labels:
du_id: '${! this.du_id }'
output:
broker:
pattern: fan_out
outputs:
# 1. Observability: OTLP/HTTP JSON to the OpenTelemetry Collector on the
# node. OpenShift user-workload monitoring scrapes the collector's
# Prometheus exporter (ServiceMonitor in the manifests). Records that
# cannot be delivered land in a dead-letter file on the PVC.
- 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_offset_nanoseconds",
"description": "PTP timing offset from master clock",
"unit": "ns",
"gauge": {"dataPoints": [{
"timeUnixNano": this.timestamp_ns.string(),
"asInt": this.ptp4l_offset_ns
}]}
},
{
"name": "oran_ptp_review_score",
"description": "PTP review score (0-100)",
"unit": "1",
"gauge": {"dataPoints": [{
"timeUnixNano": this.timestamp_ns.string(),
"asDouble": this.ptp_score
}]}
},
{
"name": "oran_prb_utilization_percent",
"description": "PRB utilization percentage",
"unit": "%",
"gauge": {"dataPoints": [
{
"timeUnixNano": this.timestamp_ns.string(),
"asDouble": this.prb_dl_pct,
"attributes": [{"key": "direction", "value": {"stringValue": "downlink"}}]
},
{
"timeUnixNano": this.timestamp_ns.string(),
"asDouble": this.prb_ul_pct,
"attributes": [{"key": "direction", "value": {"stringValue": "uplink"}}]
}
]}
},
{
"name": "oran_cpu_utilization_percent",
"description": "DU CPU utilization",
"unit": "%",
"gauge": {"dataPoints": [{
"timeUnixNano": this.timestamp_ns.string(),
"asDouble": this.cpu_pct
}]}
},
{
"name": "oran_rf_health_score",
"description": "RF quality health score",
"unit": "1",
"gauge": {"dataPoints": [{
"timeUnixNano": this.timestamp_ns.string(),
"asDouble": this.rf_health_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
}
# 2. Long-term storage: Parquet files on the PVC mounted at /data.
# Each batch is encoded in memory and written as one file.
- label: parquet_pvc
processors:
- mapping: |
root = {
"timestamp_ns": this.timestamp_ns,
"date": this.timestamp_iso.slice(0, 10),
"hour": this.hour_of_day,
"du_id": this.du_id,
"cell_id": this.cell_id,
"gnb_id": this.gnb_id,
"region": this.region,
"sector": this.sector,
"ptp_offset_ns": this.ptp4l_offset_ns,
"ptp_review": this.ptp_review,
"ptp_score": this.ptp_score,
"prb_dl_pct": this.prb_dl_pct,
"prb_ul_pct": this.prb_ul_pct,
"prb_efficiency": this.prb_efficiency,
"cpu_pct": this.cpu_pct,
"memory_pct": this.memory_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: hour, type: INT32 }
- { name: du_id, type: UTF8 }
- { name: cell_id, type: UTF8 }
- { name: gnb_id, type: UTF8 }
- { name: region, type: UTF8 }
- { name: sector, type: UTF8 }
- { name: ptp_offset_ns, type: INT64, optional: true }
- { name: ptp_review, type: UTF8, optional: true }
- { name: ptp_score, type: INT32, optional: true }
- { name: prb_dl_pct, type: DOUBLE, optional: true }
- { name: prb_ul_pct, type: DOUBLE, optional: true }
- { name: prb_efficiency, type: DOUBLE, optional: true }
- { name: cpu_pct, type: DOUBLE, optional: true }
- { name: memory_pct, type: DOUBLE, optional: true }
- { name: rsrp_dbm, type: DOUBLE, optional: true }
- { name: sinr_db, type: DOUBLE, optional: true }
- { name: rf_health_score, type: DOUBLE, optional: true }
- { name: data_quality_score, type: DOUBLE }
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
# 3. Analytics: AMQ Streams (Strimzi) Kafka on the cluster, TLS listener
# on 9093 with SCRAM-SHA-512. The cluster CA is mounted from the
# operator-managed Secret. Cloudera consumes from this topic.
- label: amq_streams_kafka
processors:
- mapping: |
root = {
"schema": {"type": "oran_telemetry", "version": "2.0.0"},
"metadata": {
"event_time": this.timestamp_iso,
"ingestion_time": now().ts_format("2006-01-02T15:04:05Z07:00"),
"pipeline_version": "2.0.0",
"data_quality_score": this.data_quality_score
},
"network": this.network_hierarchy,
"telemetry": {
"timing": {
"ptp_offset_ns": this.ptp4l_offset_ns,
"review": this.ptp_review,
"score": this.ptp_score
},
"resources": {
"prb_dl_percent": this.prb_dl_pct,
"prb_ul_percent": this.prb_ul_pct,
"efficiency": this.prb_efficiency,
"congestion": this.congestion_level
},
"system": {
"cpu_percent": this.cpu_pct,
"memory_percent": this.memory_pct
},
"radio": {
"rsrp_dbm": this.rsrp_dbm,
"sinr_db": this.sinr_db,
"health_score": this.rf_health_score
}
}
}
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
processors:
- mapping: |
root = {
"error_timestamp": now(),
"error_destination": "amq_streams_kafka",
"error_message": meta("fallback_error"),
"original_payload": this
}
Deployment manifests
The agent that runs this job is a restricted-SCC Deployment on the SNO node. The full set of manifests, including the AMQ Streams cluster, the OpenTelemetry Collector, the ServiceMonitor, and both PersistentVolumeClaims, is in examples/integrations/oran-telco-pipeline/openshift and applied with oc apply -k. The run path walks through the order and the Secrets created by hand.
openshift/kustomization.yaml
# Single-Node OpenShift deployment for the O-RAN telemetry example.
#
# oc apply -k examples/integrations/oran-telco-pipeline/openshift
#
# Operators that must already be installed on the cluster:
# - AMQ Streams (Strimzi) for the Kafka, KafkaNodePool, KafkaTopic, KafkaUser kinds
# - Red Hat build of OpenTelemetry for the OpenTelemetryCollector kind
#
# Secrets are created with `oc create secret` (see the setup page), never committed.
# cluster/user-workload-monitoring.yaml is cluster-scoped and applied separately.
apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
namespace: oran-telemetry
resources:
- 00-namespace.yaml
- 10-serviceaccount.yaml
- 20-pvc-data.yaml
- 21-pvc-edge-state.yaml
- 30-site-configmap.yaml
- 31-edge-config.yaml
- 40-kafka-nodepool.yaml
- 41-kafka.yaml
- 42-kafka-topic-structured.yaml
- 43-kafka-topic-alerts.yaml
- 44-kafka-user.yaml
- 45-kafka-consumer.yaml
- 50-otel-collector.yaml
- 51-servicemonitor.yaml
- 60-expanso-edge-deployment.yaml
- 61-expanso-edge-service.yaml
- 62-telemetry-networkpolicy.yaml
openshift/60-expanso-edge-deployment.yaml
# Expanso Edge agent on the SNO node, running under the restricted-v2 SCC.
#
# OpenShift assigns the UID and fsGroup from the namespace range, so there is
# no runAsUser here; both PersistentVolumeClaims are group-writable to that
# UID. There is no hostPath and no hostNetwork. DU telemetry reaches the agent
# over the expanso-edge Service, not through a mounted host directory.
#
# The pipeline itself is not in this manifest. It is deployed through Expanso
# Cloud (expanso-cli job deploy) and scheduled onto this agent by its labels.
apiVersion: apps/v1
kind: Deployment
metadata:
name: expanso-edge
namespace: oran-telemetry
labels:
app.kubernetes.io/name: expanso-edge
app.kubernetes.io/part-of: oran-telco-pipeline
spec:
replicas: 1
# ReadWriteOnce claims: never run two agents against the same state.
strategy:
type: Recreate
selector:
matchLabels:
app.kubernetes.io/name: expanso-edge
template:
metadata:
labels:
app.kubernetes.io/name: expanso-edge
app.kubernetes.io/part-of: oran-telco-pipeline
spec:
serviceAccountName: expanso-edge
automountServiceAccountToken: false
securityContext:
runAsNonRoot: true
seccompProfile:
type: RuntimeDefault
containers:
- name: expanso-edge
image: ghcr.io/expanso-io/expanso-edge:latest
args:
- run
- --no-watch
- --config
- /etc/expanso/edge-config.yaml
env:
- name: EXPANSO_EDGE_BOOTSTRAP_TOKEN
valueFrom:
secretKeyRef:
name: expanso-bootstrap
key: token
- name: NODE_ID
valueFrom:
fieldRef:
fieldPath: spec.nodeName
- name: KAFKA_PASSWORD
valueFrom:
secretKeyRef:
name: expanso-oran # created by the KafkaUser
key: password
- name: DU_API_KEY
valueFrom:
secretKeyRef:
name: oran-du-api
key: token
envFrom:
- configMapRef:
name: oran-site
ports:
- name: telemetry
containerPort: 8080
protocol: TCP
securityContext:
allowPrivilegeEscalation: false
capabilities:
drop: [ALL]
resources:
requests:
cpu: 250m
memory: 256Mi
limits:
cpu: '1'
memory: 1Gi
volumeMounts:
- name: edge-config
mountPath: /etc/expanso/edge-config.yaml
subPath: edge-config.yaml
readOnly: true
- name: edge-state
mountPath: /var/lib/expanso
- name: data
mountPath: /data
- name: kafka-ca
mountPath: /etc/expanso/kafka-ca
readOnly: true
volumes:
- name: edge-config
configMap:
name: expanso-edge-config
- name: edge-state
persistentVolumeClaim:
claimName: expanso-edge-state
- name: data
persistentVolumeClaim:
claimName: oran-telemetry-data
- name: kafka-ca
secret:
secretName: oran-kafka-cluster-ca-cert # created by the Kafka CR
items:
- key: ca.crt
path: ca.crt
Adaptation boundary
Review the exporter contract, units, time semantics, thresholds, Parquet schema, Kafka envelope, retention, storage sizing, and credentials before execution. The manifests assume the cluster's default StorageClass and the operator-managed Kafka CA; change both if your SNO node differs.
