| 1 | # framework/functions manager |
| 2 | |
| 3 | This document describes how the functions manager works **today** (current implementation). |
| 4 | |
| 5 | ## Scope |
| 6 | |
| 7 | - Package: `src/go/plugin/framework/functions` |
| 8 | - Main implementation: |
| 9 | - `manager.go` |
| 10 | - `manager_worker.go` |
| 11 | - `scheduler.go` |
| 12 | - `parser.go` |
| 13 | - `finalizer.go` |
| 14 | |
| 15 | ## Input protocol handled by parser |
| 16 | |
| 17 | The parser recognizes these line types: |
| 18 | |
| 19 | - `FUNCTION ...` |
| 20 | - `FUNCTION_PAYLOAD ...` + payload body + `FUNCTION_PAYLOAD_END` |
| 21 | - `FUNCTION_CANCEL <transaction_id>` |
| 22 | - `FUNCTION_PROGRESS ...` (recognized/no-op event for manager) |
| 23 | - `QUIT` |
| 24 | |
| 25 | Payload-mode control behavior: |
| 26 | |
| 27 | - `FUNCTION_CANCEL <same payload uid>`: |
| 28 | - abort payload frame |
| 29 | - emit pre-admission cancel event |
| 30 | - `FUNCTION_CANCEL <different uid>`: |
| 31 | - emit cancel event |
| 32 | - continue payload accumulation |
| 33 | - `FUNCTION_PROGRESS ...`: |
| 34 | - emit progress event |
| 35 | - continue payload accumulation |
| 36 | - `QUIT`: |
| 37 | - abort payload frame |
| 38 | - emit quit event |
| 39 | - Any other `FUNCTION*` control line: |
| 40 | - abort current payload frame |
| 41 | - never dispatch partial payload |
| 42 | |
| 43 | ## Runtime architecture |
| 44 | |
| 45 | ### Dispatcher |
| 46 | |
| 47 | The dispatcher loop: |
| 48 | |
| 49 | - reads input lines |
| 50 | - parses events |
| 51 | - handles cancel/quit/progress |
| 52 | - resolves handler + route-aware schedule key |
| 53 | - admits calls into keyed scheduler |
| 54 | |
| 55 | Admission checks: |
| 56 | |
| 57 | - manager stopping -> reject `503` |
| 58 | - unknown/nil handler -> reject `501` |
| 59 | - duplicate active/tombstoned UID -> ignore duplicate input (debug/warn log, no terminal output) |
| 60 | - queue full -> blocks on `scheduler.enqueue` until space frees (back-pressures stdin reader -> netdata via OS pipe). The only errors returned from this path are manager-stopping (`503`) on shutdown and invalid-request (`500`) for malformed input. |
| 61 | |
| 62 | ### Keyed scheduler + worker pool |
| 63 | |
| 64 | - fixed-size worker pool (`defaultWorkerCount = 1`) |
| 65 | - bounded pending budget (`defaultQueueSize = 64`) |
| 66 | - per-key serialization: |
| 67 | - same schedule key executes sequentially |
| 68 | - different schedule keys execute concurrently (up to worker count) |
| 69 | - schedule key is route-aware: |
| 70 | - direct registration: `fn.Name` |
| 71 | - prefix registration: `fn.Name|<matched-prefix>` |
| 72 | - prefix registration guard: |
| 73 | - overlapped prefixes for the same function name are rejected at registration time |
| 74 | - manager logs an error and keeps the previously registered prefix set unchanged |
| 75 | - worker transitions lifecycle: |
| 76 | - `queued -> running -> awaiting_result` |
| 77 | - worker return is **not** terminal completion |
| 78 | - panic path finalizes terminal `500` |
| 79 | |
| 80 | ### Tracking and finalization |
| 81 | |
| 82 | Active requests are tracked by UID: |
| 83 | |
| 84 | - `invState` map (active entries) |
| 85 | - tombstones (`defaultTombstoneTTL = 60s`) to block immediate UID reuse |
| 86 | |
| 87 | All terminal outputs go through: |
| 88 | |
| 89 | - manager-bound terminal finalizer (`m.finalizeTerminal`) |
| 90 | - dyncfg responders receive manager finalizer wiring at component construction time |
| 91 | |
| 92 | `tryFinalize` guarantees: |
| 93 | |
| 94 | - first terminal writer wins |
| 95 | - late terminal duplicates are dropped |
| 96 | - fallback timer is stopped on finalization |
| 97 | - awaiting-result warning timer is stopped on finalization |
| 98 | - UID becomes tombstoned for a short window |
| 99 | |
| 100 | Awaiting-result observability: |
| 101 | |
| 102 | - when a worker returns without terminal output, manager moves UID to `awaiting_result` |
| 103 | - manager starts a warning timer (`defaultAwaitingWarnDelay = 30s`, capped by function timeout if lower) |
| 104 | - timer emits a warning log if UID is still `awaiting_result` (diagnostic only, no forced finalize) |
| 105 | |
| 106 | ## Runtime metrics |
| 107 | |
| 108 | Functions manager owns an internal runtime store (`metrix.NewRuntimeStore()`), and |
| 109 | can register it as a runtime component when runtime service is injected via |
| 110 | `SetRuntimeService(...)`. |
| 111 | |
| 112 | Registered component metadata: |
| 113 | |
| 114 | - component name: `functions.manager` |
| 115 | - module: `functions` |
| 116 | - job: `manager` |
| 117 | - autogen charts: enabled |
| 118 | |
| 119 | Pathology-focused metrics currently exposed: |
| 120 | |
| 121 | - gauges: |
| 122 | - `netdata.go.plugin.framework.functions.manager.invocations_active` |
| 123 | - `netdata.go.plugin.framework.functions.manager.invocations_awaiting_result` |
| 124 | - `netdata.go.plugin.framework.functions.manager.scheduler_pending` |
| 125 | - counters: |
| 126 | - `netdata.go.plugin.framework.functions.manager.cancel_fallback_total` |
| 127 | - `netdata.go.plugin.framework.functions.manager.late_terminal_dropped_total` |
| 128 | - `netdata.go.plugin.framework.functions.manager.duplicate_uid_ignored_total` |
| 129 | |
| 130 | ## Cancellation semantics |
| 131 | |
| 132 | ### 1) Queued request |
| 133 | |
| 134 | - mark cancel requested |
| 135 | - cancel internal context |
| 136 | - finalize exactly once with `499` |
| 137 | - worker skips execution if it dequeues a canceled request |
| 138 | |
| 139 | ### 2) Running / awaiting_result request |
| 140 | |
| 141 | - mark cancel requested |
| 142 | - call internal cancel func |
| 143 | - start fallback timer (`defaultCancelFallbackDelay = 5s`) |
| 144 | - if no terminal output arrives before timer, manager finalizes with `499` |
| 145 | |
| 146 | Important limitation: |
| 147 | |
| 148 | - handlers are currently `func(Function)` (no `context.Context` parameter) |
| 149 | - manager cannot force-stop handler code directly |
| 150 | - fallback `499` is the deterministic safety net |
| 151 | |
| 152 | ### 3) Unknown / already completed request |
| 153 | |
| 154 | - no-op |
| 155 | - debug log only |
| 156 | |
| 157 | ## Shutdown behavior |
| 158 | |
| 159 | Shutdown uses one bounded path for `ctx.Done()`, `QUIT`, and input close (EOF): |
| 160 | |
| 161 | - set stopping |
| 162 | - stop scheduler admission |
| 163 | - wait up to `defaultShutdownDrainTimeout = 8s` for natural drain |
| 164 | - if drain times out **or** unresolved active UIDs remain after worker drain: |
| 165 | - cancel in-flight |
| 166 | - force-finalize unresolved UIDs with `499` |
| 167 | - hard-stop scheduler waiters |
| 168 | |
| 169 | Input close still enters the same bounded path above: |
| 170 | |
| 171 | - input close/EOF: |
| 172 | - stop scheduler admission |
| 173 | - attempt bounded drain first |
| 174 | - escalate to cancel/force-finalize on timeout or if active UIDs remain unresolved |
| 175 | |
| 176 | ## Flow diagram |
| 177 | |
| 178 | ```mermaid |
| 179 | flowchart TD |
| 180 | A["Input line"] --> B["Parser.parseEvent()"] |
| 181 | B -->|call| C["dispatchInvocation()"] |
| 182 | B -->|cancel| D["handleCancelEvent()"] |
| 183 | B -->|progress| E["No-op"] |
| 184 | B -->|quit| F["Shutdown(canceling)"] |
| 185 | B -->|parse error| G["Warn + continue"] |
| 186 | C --> C1{"Admission checks"} |
| 187 | C1 -->|stopping| R503["respf 503"] |
| 188 | C1 -->|unregistered/nil handler| R501["respf 501"] |
| 189 | C1 -->|duplicate/tombstoned UID| DUP["ignore duplicate + log"] |
| 190 | C1 -->|accepted| Q["scheduler.enqueue by route key + state=queued (blocks if queue full)"] |
| 191 | Q --> S["keyScheduler"] |
| 192 | S -->|same key busy| SQ["lane queue (serialized)"] |
| 193 | S -->|key free| W["Worker"] |
| 194 | W --> W1{"Start allowed?"} |
| 195 | W1 -->|ctx canceled / cancelRequested| X["skip"] |
| 196 | W1 -->|yes| W2["state=running; run handler"] |
| 197 | W2 -->|panic| R500["respf 500"] |
| 198 | W2 -->|return| AWAIT["state=awaiting_result"] |
| 199 | D --> D1{"Cancel target state"} |
| 200 | D1 -->|pre - admission payload UID| C499["respf 499"] |
| 201 | D1 -->|queued| C499 |
| 202 | D1 -->|running/awaiting| TMR["cancel() + fallback timer"] |
| 203 | D1 -->|unknown/done| NOP["debug no-op"] |
| 204 | TMR -->|timer fires & still unresolved| C499 |
| 205 | R501 --> FIN["manager finalizer"] |
| 206 | R503 --> FIN |
| 207 | R500 --> FIN |
| 208 | C499 --> FIN |
| 209 | HRESP["Handler/dyncfg responder terminal output"] --> FIN |
| 210 | FIN --> TF["tryFinalize(): first wins, tombstone set, emit FUNCRESULT"] |
| 211 | TF --> SREL["scheduler.complete(key, uid)"] |
| 212 | SREL -->|promote next same-key request| W |
| 213 | TF --> OUT["stdout FUNCRESULT"] |
| 214 | HRESP -->|late duplicate| DROP["drop + debug log"] |
| 215 | A -->|ctx . Done| F |
| 216 | A -->|input close| F2["Shutdown(bounded canceling path)"] |
| 217 | ``` |