E2SM-KPM — Key Performance Measurement

E2SM-KPM is the measurement axis: aggregated performance counters — throughput, PRB usage, delay — reported per cell or per UE. This guide shows how to discover what a node measures, subscribe to it, read the indications, and take a one-shot snapshot, in Python, Go, and C.

Spec anchors

O-RAN.WG3.E2SM-KPM v03.00 — the contract implemented here.
OID 1.3.6.1.4.1.53148.1.3.2.2, default RAN Function ID 2.
REPORT-only per spec (no INSERT / CONTROL / POLICY).

What KPM gives you

Service	Implemented	Use
REPORT	Styles 1–5	Periodic cell- or UE-level measurement reports
One-shot read	`kpm_snapshot`	Subscribe, take the first report, unsubscribe — a client-side convenience, not a wire procedure

The five REPORT styles

All five v03.00 styles are implemented. A style fixes the reporting scope and what conditions it requires; per-style cross-field validation runs at submission time, so an invalid combination fails on subscribe, not on the wire.

Style	Scope	UE selection	Conditions
1	Cell-level	—	—
2	Single-UE	exactly 1 UE ID	—
3	Per-measurement, UE-level	matched by condition	per-measurement
4	Common condition, UE-level	matched by condition	≥1 global condition
5	Multi-UE	≥2 UE IDs	—

Style 4 (per-UE, filtered by a common condition such as an S-NSSAI slice) is the workhorse; Style 1 (cell-level counters) is the other common choice.

Step 1 — discover what the node measures

A KPM node advertises its report styles and named measurements in its RAN Function Definition. Resolve the function by OID — never by a hard-coded RAN Function ID — then decode the definition and subscribe only to what the node actually offers. This is what makes an xApp work against a gNB it was not written for.

PythonGoC

from airpuls_ric_sdk import KPM_OID, decode_kpm_ran_function

def on_e2_node_available(self, client, node):
    fn = node.find_ran_function_by_oid(KPM_OID)
    if fn is None:
        return                       # this node has no KPM

    desc = decode_kpm_ran_function(fn.definition_bytes)
    if desc is None:
        return                       # definition did not decode

    for style in desc.report_styles:
        print(style.style_type, style.style_name, len(style.measurements))
    measurements = desc.all_measurements     # union across styles

fn := node.FindRanFunctionByOID(ric.KpmOID)
if fn == nil {
    return
}
desc, err := ric.DecodeRanFunctionDefinition(ric.KpmPluginGet(),
    fn.DefinitionBytes())
if err != nil {
    return
}
defer desc.Free()

for i := 0; i < desc.ReportStyleCount(); i++ {
    style := desc.ReportStyleAt(i)
    for j := 0; j < style.MeasCount(); j++ {
        log.Printf("style %d: %s", style.Type(), style.MeasNameAt(j))
    }
}

DecodeRanFunctionDefinition is generic — pass any SM's plugin.

#include "e2sm-kpm.h"

const ric_ran_function_t *fn =
    ric_e2_node_find_ran_function_by_oid(node, KPM_OID);
if (!fn) return;

size_t len = 0;
const uint8_t *def = ric_ran_function_definition_bytes(fn, &len);

sm_ran_function_description_t *desc = NULL;
if (kpm_ran_function_definition_decode(def, len, &desc) != 0) return;

for (size_t i = 0; i < sm_ran_function_description_report_style_count(desc); i++) {
    const sm_report_style_info_t *st =
        sm_ran_function_description_report_style_at(desc, i);
    for (size_t j = 0; j < sm_report_style_meas_count(st); j++)
        printf("style %lld: %s\n", (long long)sm_report_style_type(st),
               sm_report_style_meas_name_at(st, j));
}
sm_ran_function_description_free(desc);

The bundled kpm-monitor subscribes to every advertised measurement this way — a good starting pattern.

What the airpuls E2 agent reports

KPM names are not fixed by the SDK — they are whatever the node advertises, which is why the discovery step above is not optional. For orientation, the airpuls E2 agent and the E2 emulator both advertise this set (3GPP TS 28.552 names):

Measurement	Scope	Meaning
`DRB.UEThpDl` / `DRB.UEThpUl`	per UE	UE throughput, downlink / uplink
`DRB.PdcpSduVolumeDL` / `…UL`	per UE	PDCP SDU volume
`DRB.RlcSduTransmittedVolumeDL` / `…UL`	per UE	RLC SDU transmitted volume
`DRB.RlcSduDelayDl`	style-dependent	RLC SDU delay
`RRU.PrbTotDl` / `RRU.PrbTotUl`	style-dependent	PRB utilisation, as a percentage
`RRU.PrbUsedDl` / `RRU.PrbUsedUl`	style-dependent	Mean PRBs used per slot

The last three rows carry the same measurement name at two scopes — the REPORT style decides which one you receive. TS 28.552 defines them cell-wide (object class NRCellDU); E2SM-KPM §7.9.0 defines their UE-level counterparts as the same formulas "with restriction to the individual UE", and Styles 2–4 are UE-level. So a Style 4 subscription returns this UE's PRB share, per-slot PRB usage and mean SDU delay on every UE block, while a Style 1 subscription returns the cell-wide originals. The airpuls agent advertises both styles with per-style measurement lists (RRU.PrbAvail* is Style 1-only), and a Style 1 subscription must name a served cell with set_cell_global_id() — a missing or foreign CGI is rejected at subscribe time.

A different gNB advertises a different set — never hard-code this table, subscribe to what the definition decode returned. A name the node does not advertise is rejected at subscribe time with a MEASUREMENT_NOT_ADVERTISED notification, not silently dropped.

Build a subscription with the KPM builder: pick a style, set the report period and granularity (ms), add measurement names, and — for Style 4 — add at least one condition. The builder is single-use: the subscribe call consumes it.

PythonGoC

from airpuls_ric_sdk import KpmSubscriptionBuilder, KpmStyle, KpmConditionExpr

# Cell-level (Style 1)
b = KpmSubscriptionBuilder(KpmStyle.STYLE_1)
b.set_report_period(1000)
b.set_granularity(1000)
b.add_measurement("RRU.PrbTotDl")
b.add_measurement("RRU.PrbTotUl")
sub_id = client.subscribe_kpm(node, b)

# Per-UE (Style 4 — requires ≥1 condition)
b = KpmSubscriptionBuilder(KpmStyle.STYLE_4)
b.set_report_period(1000)
b.set_granularity(1000)
b.add_measurement("DRB.UEThpDl")
b.add_measurement("DRB.UEThpUl")
b.add_condition_snssai_sst(KpmConditionExpr.EQUAL, 1)
sub_id = client.subscribe_kpm(node, b)

// Cell-level (Style 1)
b := ric.NewKpmSubscriptionBuilder(ric.KpmStyle1)
b.SetReportPeriod(1000).
    SetGranularity(1000).
    AddMeasurement("RRU.PrbTotDl").
    AddMeasurement("RRU.PrbTotUl")
subID, err := client.SubscribeKPM(node, b)

// Per-UE (Style 4 — requires ≥1 condition)
b = ric.NewKpmSubscriptionBuilder(ric.KpmStyle4)
b.SetReportPeriod(1000).
    SetGranularity(1000).
    AddMeasurement("DRB.UEThpDl").
    AddConditionSnssaiSST(ric.KpmExprEqual, 1)
subID, err = client.SubscribeKPM(node, b)

Builders are consumed by SubscribeKPM; further use returns ric.ErrBuilderConsumed.

#include "e2sm-kpm.h"

/* Cell-level (Style 1) */
kpm_subscription_builder_t *b = kpm_subscription_builder_new(KPM_STYLE_1);
kpm_sub_set_report_period(b, 1000);
kpm_sub_set_granularity(b, 1000);
kpm_sub_add_measurement(b, "RRU.PrbTotDl");
kpm_sub_add_measurement(b, "RRU.PrbTotUl");
uint32_t sub_id = 0;
ric_client_subscribe_kpm(client, node, b, &sub_id);   /* consumes b */

/* Per-UE (Style 4 — requires ≥1 condition) */
kpm_subscription_builder_t *b4 = kpm_subscription_builder_new(KPM_STYLE_4);
kpm_sub_set_report_period(b4, 1000);
kpm_sub_set_granularity(b4, 1000);
kpm_sub_add_measurement(b4, "DRB.UEThpDl");
kpm_sub_add_condition_snssai_sst(b4, KPM_EXPR_EQUAL, 1);
ric_client_subscribe_kpm(client, node, b4, &sub_id);

Selecting UEs by identity (Styles 2 and 5)

Styles 2 and 5 name the UEs instead of matching them by condition — Style 2 takes exactly one, Style 5 two or more. There is one adder per UEID flavour, matching the E2 node type the UE is known to:

UEID	Python	Go	C
gNB (AMF UE NGAP ID)	`add_ue_id(id, plmn)`	`AddUeID(id, plmn)`	`kpm_sub_add_ue_id(b, id, plmn)`
gNB-DU (F1AP ID)	`add_ue_id_gnb_du(id)`	`AddUeIDGnbDu(id)`	`kpm_sub_add_ue_id_gnb_du(b, id)`
gNB-CU-UP (E1AP ID)	`add_ue_id_gnb_cu_up(id)`	`AddUeIDGnbCuUp(id)`	`kpm_sub_add_ue_id_gnb_cu_up(b, id)`
ng-eNB / ng-eNB-DU / en-gNB / eNB	`add_ue_id_ng_enb(…)`, `…_ng_enb_du(…)`, `…_en_gnb(…)`, `…_enb(…)`	same names, Go-cased	`kpm_sub_add_ue_id_*`

b = KpmSubscriptionBuilder(KpmStyle.STYLE_2)
b.set_report_period(1000)
b.add_measurement("DRB.UEThpDl")
b.add_ue_id(amf_ue_ngap_id=12345, plmn=b"\x00\xf1\x10")   # exactly one
sub_id = client.subscribe_kpm(node, b)

Getting a UE ID to name is the bootstrap problem an RC QUERY solves: query the node for the UEs it already knows, then subscribe by identity.

Test conditions

Conditions filter which UEs a UE-level style reports on. Style 4 applies them globally; Style 3 applies them per measurement, so one subscription can report different measurements over different UE populations.

Condition type	Constant
S-NSSAI (slice)	`S_NSSAI` — use the `_snssai_sst` / `_snssai_sst_sd` helpers
5QI / QCI	`FIVE_QI`, `QCI`
Radio quality	`RSRP`, `RSRQ`, `UL_RSRP`, `CQI`
Bitrate	`GBR`, `AMBR`
UE class	`IS_STAT`, `IS_CAT_M`

Expressions are EQUAL, GREATER, LESS, CONTAINS, PRESENT.

from airpuls_ric_sdk import KpmConditionType, KpmConditionExpr

# Style 4 — one condition for the whole subscription
b.add_condition_int(KpmConditionType.RSRP, KpmConditionExpr.LESS, -100)

# Style 3 — a condition attached to one measurement
b.add_measurement_condition_snssai_sst("DRB.UEThpDl", KpmConditionExpr.EQUAL, 1)
b.add_measurement_condition_int("DRB.UEThpUl",
                                KpmConditionType.CQI, KpmConditionExpr.LESS, 7)

Subcounter labels

A MeasurementLabel selects a subcounter dimension — report DRB.UEThpDl broken down per 5QI, per slice, or as min/max/avg rather than the aggregate. Add the same measurement several times with different labels and each produces its own subcounter; labelled entries are deliberately not deduplicated.

PythonGoC

from airpuls_ric_sdk import KpmLabel

lbl = KpmLabel()
lbl.set_five_qi(9)
b.add_measurement_label("DRB.UEThpDl", lbl)      # takes ownership

# Several dimensions of the same measurement in one call
b.add_measurement_labels("DRB.UEThpDl", [_label_avg(), _label_max()])

lbl := ric.NewKpmLabel()
lbl.SetFiveQI(9)
b.AddMeasurementWithLabel("DRB.UEThpDl", lbl)

kpm_label_t *lbl = kpm_label_new();
kpm_label_set_five_qi(lbl, 9);
kpm_sub_add_measurement_label(b, "DRB.UEThpDl", lbl);  /* takes ownership */

Available dimensions: PLMN, S-NSSAI, 5QI, QFI, QCI (and QCI/ARP ranges), bitrate range, MU-MIMO layer, SSB index, beam ID, distribution bin, and the aggregates sum / min / max / avg. On the indication side the labels come back on each measurement — see below.

Step 3 — read the indications

KPM indications lower into the generic measurement-block tree: one block per UE (or one cell-scoped block), each holding named integer/real measurements. See Building xApps → Reading indication data for the full accessor list.

PythonGoC

def on_indication(self, client, sub_id, node, ran_func_id, header, data):
    for block in data.blocks:
        ue = block.ue_id.display        # "ue:amf_ue_ngap_id:12345" or cell-scoped
        for m in block.measurements:
            print(ue, m.name, m.value)  # m.value: int | float | None
            for label in m.labels:      # subcounter dimensions, if any
                print("   ", label.to_dict())

func (h *handler) OnIndication(c *ric.Client, subID uint32, node *ric.E2Node,
    ranFuncID uint16, hdr *ric.IndicationHeader, data *ric.IndicationData) {
    for _, block := range data.Blocks() {
        for _, m := range block.Measurements() {
            fmt.Printf("%s = %d\n", m.Name(), m.Integer())
        }
    }
}

for (size_t bi = 0; bi < sm_indication_data_block_count(data); bi++) {
    const sm_measurement_block_t *blk = sm_indication_data_block_at(data, bi);
    for (size_t j = 0; j < sm_measurement_block_count(blk); j++) {
        const sm_measurement_t *m = sm_measurement_block_at(blk, j);
        printf("%s = %lld\n", sm_measurement_name(m),
               (long long)sm_measurement_integer(m));
    }
}

KPM timestamps are RFC 5905 NTP

The KPM indication header carries a colletStartTime timestamp, decoded per O-RAN.WG3.E2SM-KPM-v03.00 §8.3.12 (RFC 5905 NTP: seconds-since-1900 + a 2⁻³² s fraction) and surfaced as header.timestamp_ms. Carry it into your sinks rather than stamping at dispatch time — see Telemetry Sinks → Preserving an external timestamp.

Every Style 4 measurement is per-UE

Style 4 is UE-level, so E2SM-KPM §7.9.0's per-UE conversions apply to every measurement it carries — including RRU.PrbTot* (this UE's share of the cell's available PRBs) and DRB.RlcSduDelayDl (this UE's mean SDU sojourn). Emit them under per-UE tags; per-UE aggregates such as SUM(RRU.PrbTotDl) BY (node) are meaningful. For the cell-wide originals, subscribe REPORT Style 1 instead — one name, two scopes, chosen by style.

One-shot: `kpm_snapshot`

A late-starting xApp that wants one reading — not a stream — can take a snapshot. It is a client-side state machine, not a wire procedure: it subscribes with the builder you give it, waits for the first indication, unsubscribes, and returns the decoded result. Cost is one full Subscribe round-trip plus at least one report period.

PythonGoC

b = KpmSubscriptionBuilder(KpmStyle.STYLE_1)
b.set_report_period(1000)
b.add_measurement("RRU.PrbTotDl")
# Style 1 subscribes cell-confined measurements: name the cell
# (E2SM-KPM §7.4.2.2) — 3-byte BCD PLMN + 36-bit NR Cell Identity.
b.set_cell_global_id(bytes([0x00, 0xF1, 0x10]), 0xA00000003)

snap = client.kpm_snapshot(node, b, timeout_ms=5000)   # consumes b
print(snap.timestamp_ms)
for block in snap.blocks:
    for m in block.measurements:
        print(m.name, m.value)

The result is pure Python — every value is copied out, so there is no C lifetime to manage.

hdr, data, err := client.KpmSnapshot(node, b, 5000)
if err != nil {
    return err
}
defer data.Free()

sm_indication_header_t *hdr = NULL;
sm_indication_data_t *data = NULL;
ric_client_kpm_snapshot(client, node, b, /*timeout_ms=*/5000, &hdr, &data);
/* caller owns hdr and data */

An empty snapshot is a valid result

A Style 5 (or condition-filtered) snapshot may legitimately return zero blocks — that is a successful read with no matching UEs, not an error.

Reference

Concept	Python	Go	C
Plugin	`kpm_plugin_get()`	`ric.KpmPluginGet()`	`kpm_plugin_get()`
RAN Function ID	`KPM_RAN_FUNC_ID` (2)	`ric.KpmRanFuncID`	`KPM_RAN_FUNC_ID`
OID	`KPM_OID`	`ric.KpmOID`	`KPM_OID`
Decode definition	`decode_kpm_ran_function(bytes)`	`ric.DecodeRanFunctionDefinition(plugin, bytes)`	`kpm_ran_function_definition_decode(...)`
Builder	`KpmSubscriptionBuilder(KpmStyle.STYLE_4)`	`ric.NewKpmSubscriptionBuilder(ric.KpmStyle4)`	`kpm_subscription_builder_new(KPM_STYLE_4)`
Period / granularity	`set_report_period` / `set_granularity`	`SetReportPeriod` / `SetGranularity`	`kpm_sub_set_report_period` / `_set_granularity`
Measurement	`add_measurement(name)`	`AddMeasurement(name)`	`kpm_sub_add_measurement(b, name)`
Labelled measurement	`add_measurement_label(name, label)`	`AddMeasurementWithLabel(name, label)`	`kpm_sub_add_measurement_label(b, name, label)`
Condition	`add_condition_snssai_sst(expr, sst)`, `add_condition_int(type, expr, v)`	`AddConditionSnssaiSST`, `AddConditionInt`	`kpm_sub_add_condition_snssai_sst`, `_add_condition_int`
Per-measurement condition	`add_measurement_condition_int(...)`	`AddMeasurementConditionInt(...)`	`kpm_sub_add_measurement_condition_int(...)`
UE ID (Style 2/5)	`add_ue_id(id, plmn)`	`AddUeID(id, plmn)`	`kpm_sub_add_ue_id(b, id, plmn)`
Subscribe	`client.subscribe_kpm(node, b)`	`client.SubscribeKPM(node, b)`	`ric_client_subscribe_kpm(client, node, b, &id)`
One-shot	`client.kpm_snapshot(node, b, timeout_ms)`	`client.KpmSnapshot(node, b, ms)`	`ric_client_kpm_snapshot(...)`
Unsubscribe	`client.unsubscribe(sub_id)`	`client.Unsubscribe(subID)`	`ric_client_unsubscribe(client, id)`

Bundled examples

kpm-monitor (Python, Go, and C) subscribes Style 4 to every advertised measurement and streams samples to the sinks; ho-trigger (Python) drives a handover from a KPM throughput threshold; y1-termination serves KPM-derived analytics over a Y1 REST API. See xApps → Monitoring and Control.

airpuls-sdk xapp new --sm kpm generates a working version of the discover-then-subscribe pattern above — see the CLI.

E2SM-KPM — Key Performance Measurement

What KPM gives you

The five REPORT styles

Step 1 — discover what the node measures

What the airpuls E2 agent reports

Step 2 — subscribe

Selecting UEs by identity (Styles 2 and 5)

Test conditions

Subcounter labels

Step 3 — read the indications

One-shot: kpm_snapshot

Reference

Bundled examples

One-shot: `kpm_snapshot`