Documentation

XE005: Distributed Metrics Aggregator (collectors, reducers, rollups, channels, tasks, multi‑module, supervision)

Back to documentation index
---
{
  "@context": "https://nxdlang.org/schema",
  "doc_id": "XE005",
  "title": "",
  "description": "",
  "layer": "Examples",
  "category": "XL Examples",
  "keywords": [],
  "doc_version": "1.0",
  "status": "active"
}
---


# XE005: Distributed Metrics Aggregator  (collectors, reducers, rollups, channels, tasks, multi‑module, supervision)

## Canonical NXD (XL‑Layer)

```nxd
MODULE metrics.types

TYPE RESULT UNION { OK(any), ERR(string) }
TYPE OPTION UNION { SOME(any), NONE }

TYPE METRIC {
    NAME: string,
    VALUE: int,
    TS: int
}

TYPE ROLLUP {
    NAME: string,
    AVG: int,
    MIN: int,
    MAX: int,
    COUNT: int,
    TS: int
}

TYPE CHANNEL<METRIC> { }
TYPE CHANNEL<ROLLUP> { }


MODULE metrics.window

TYPE WINDOW { ITEMS: LIST<METRIC> }

FUNC NEW_WINDOW(): WINDOW:
    RETURN WINDOW { ITEMS: [] }

FUNC ADD(W: WINDOW, M: METRIC):
    PUSH W.ITEMS, M

FUNC TRIM(W: WINDOW, AGE: int):
    LET NOW SET NOW()
    LET NEW SET []
    LOOP X IN W.ITEMS:
        IF NOW SUB X.TS LT AGE:
            PUSH NEW, X
    W.ITEMS SET NEW

FUNC GET_BY_NAME(W: WINDOW, NAME: string): LIST<METRIC>:
    LET OUT SET []
    LOOP X IN W.ITEMS:
        IF X.NAME EQ NAME:
            PUSH OUT, X
    RETURN OUT


MODULE metrics.rollup
IMPORT metrics.types
IMPORT metrics.window

FUNC COMPUTE(NAME: string, W: WINDOW): OPTION:
    LET LIST SET GET_BY_NAME(W, NAME)
    IF LEN(LIST) EQ 0:
        RETURN NONE

    LET SUM SET 0
    LET MINV SET LIST[0].VALUE
    LET MAXV SET LIST[0].VALUE

    LOOP X IN LIST:
        SUM SET SUM ADD X.VALUE
        IF X.VALUE LT MINV: MINV SET X.VALUE
        IF X.VALUE GT MAXV: MAXV SET X.VALUE

    LET AVG SET SUM DIV LEN(LIST)

    RETURN SOME(ROLLUP {
        NAME: NAME,
        AVG: AVG,
        MIN: MINV,
        MAX: MAXV,
        COUNT: LEN(LIST),
        TS: NOW()
    })


MODULE metrics.collector
IMPORT metrics.types

FUNC EMIT(IN: CHANNEL<METRIC>, NAME: string, VALUE: int):
    LET M SET METRIC {
        NAME: NAME,
        VALUE: VALUE,
        TS: NOW()
    }
    SEND M TO IN


MODULE metrics.reducer
IMPORT metrics.types
IMPORT metrics.window
IMPORT metrics.rollup

FUNC REDUCE(IN: CHANNEL<METRIC>, OUT: CHANNEL<ROLLUP>, NAMES: LIST<string>):
    LET W SET NEW_WINDOW()

    LOOP:
        LET M SET RECV IN
        ADD(W, M)
        TRIM(W, 5000)

        LOOP N IN NAMES:
            LET R SET COMPUTE(N, W)
            MATCH R:
                CASE SOME(ROLL):
                    SEND ROLL TO OUT
                CASE NONE:
                    NONE


MODULE metrics.sink
IMPORT metrics.types

FUNC PRINT_ROLLUPS(CH: CHANNEL<ROLLUP>):
    LOOP:
        LET R SET RECV CH
        PRINTLN(
            "[ROLLUP] " ADD R.NAME ADD
            " avg=" ADD R.AVG ADD
            " min=" ADD R.MIN ADD
            " max=" ADD R.MAX ADD
            " count=" ADD R.COUNT
        )


MODULE app.main
IMPORT metrics.types
IMPORT metrics.collector
IMPORT metrics.reducer
IMPORT metrics.sink

FUNC MAIN():
    LET IN SET CHANNEL<METRIC>()
    LET OUT SET CHANNEL<ROLLUP>()

    LET NAMES SET ["cpu", "mem", "disk"]

    SPAWN REDUCE(IN, OUT, NAMES)
    SPAWN PRINT_ROLLUPS(OUT)

    # simulate collectors
    SPAWN fn():
        LOOP:
            EMIT(IN, "cpu", RANDOM(0, 100))
            SLEEP(1)

    SPAWN fn():
        LOOP:
            EMIT(IN, "mem", RANDOM(0, 100))
            SLEEP(1)

    SPAWN fn():
        LOOP:
            EMIT(IN, "disk", RANDOM(0, 100))
            SLEEP(1)

    # allow rollups to accumulate
    SLEEP(10)

    RETURN NONE
```



# What this XL example demonstrates

### Multi‑module metrics architecture  
- `metrics.types`  
- `metrics.window`  
- `metrics.rollup`  
- `metrics.collector`  
- `metrics.reducer`  
- `metrics.sink`  
- `app.main`

### Rolling metric windows  
- Time‑based trimming  
- Per‑metric grouping  
- Dynamic rollup computation

### Rollup engine  
- AVG / MIN / MAX / COUNT  
- Multiple metric names  
- Continuous streaming rollups

### Channels + processes  
- Collectors → Reducer → Sink  
- Fully asynchronous  
- Real distributed‑system semantics

### Result + Option  
- Optional rollups  
- Safe evaluation  
- Pattern matching

### Realistic metrics aggregator  
- CPU, memory, disk  
- Randomized values  
- Continuous rollups  
- Multi‑sink output

This is a **full subsystem**, suitable for real NXD agent training and backend mapping.