| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package functions |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "log/slog" |
| 10 | "maps" |
| 11 | "strconv" |
| 12 | "strings" |
| 13 | "sync" |
| 14 | "sync/atomic" |
| 15 | "time" |
| 16 | |
| 17 | "github.com/netdata/netdata/go/plugins/logger" |
| 18 | "github.com/netdata/netdata/go/plugins/pkg/metrix" |
| 19 | "github.com/netdata/netdata/go/plugins/pkg/netdataapi" |
| 20 | "github.com/netdata/netdata/go/plugins/pkg/safewriter" |
| 21 | "github.com/netdata/netdata/go/plugins/plugin/framework/runtimecomp" |
| 22 | ) |
| 23 | |
| 24 | type functionSet struct { |
| 25 | direct func(Function) // for globally-unique names |
| 26 | prefixes map[string]func(Function) // for prefix-multiplexed names |
| 27 | } |
| 28 | |
| 29 | const ( |
| 30 | // defaultWorkerCount stays at 1 for now. |
| 31 | // Working theory: remaining concurrency risk is in collector MethodHandler |
| 32 | // implementations (shared mutable state and lifecycle races), not in manager |
| 33 | // queue/worker logic. |
| 34 | // TODO: establish and document a MethodHandler goroutine-safety contract |
| 35 | // before increasing this default. |
| 36 | defaultWorkerCount = 1 |
| 37 | // defaultQueueSize is intentionally 1 (not removed: the keyed scheduler is |
| 38 | // still the dispatch primitive, we just don't want it to absorb bursts). |
| 39 | // Rationale: every downstream stage is single-threaded today (1 worker, 1 |
| 40 | // jobmgr loop, serial Check()), so admitting more than 1 extra request |
| 41 | // only buys wedge surface (a queued request that is later cancelled by |
| 42 | // netdata cannot reach jobmgr, so the dyncfg wait gate stays in |
| 43 | // 'accepted' forever). With queue=1 the stdin reader back-pressures |
| 44 | // earlier through the OS pipe instead of |
| 45 | // piling up admitted-but-unprocessed work in stateQueued. If/when we add |
| 46 | // real downstream concurrency, raise this again. |
| 47 | defaultQueueSize = 1 |
| 48 | defaultCancelFallbackDelay = 5 * time.Second |
| 49 | defaultShutdownDrainTimeout = 8 * time.Second |
| 50 | defaultTombstoneTTL = 60 * time.Second |
| 51 | defaultAwaitingWarnDelay = 30 * time.Second |
| 52 | ) |
| 53 | |
| 54 | type invocationState uint8 |
| 55 | |
| 56 | const ( |
| 57 | stateQueued invocationState = iota + 1 |
| 58 | stateRunning |
| 59 | stateAwaitingResult |
| 60 | ) |
| 61 | |
| 62 | type invocationAdmission uint8 |
| 63 | |
| 64 | const ( |
| 65 | invocationAdmissionAccepted invocationAdmission = iota + 1 |
| 66 | invocationAdmissionDuplicateActive |
| 67 | invocationAdmissionDuplicateTombstone |
| 68 | invocationAdmissionInvalid |
| 69 | ) |
| 70 | |
| 71 | type invocationRequest struct { |
| 72 | fn *Function |
| 73 | handler func(Function) |
| 74 | ctx context.Context |
| 75 | scheduleKey string |
| 76 | } |
| 77 | |
| 78 | type invocationRecord struct { |
| 79 | state invocationState |
| 80 | cancel context.CancelFunc |
| 81 | cancelRequested bool |
| 82 | fallbackTimer *time.Timer |
| 83 | awaitingTimer *time.Timer |
| 84 | awaitingSince time.Time |
| 85 | scheduleKey string |
| 86 | } |
| 87 | |
| 88 | func NewManager() *Manager { |
| 89 | runtimeStore := metrix.NewRuntimeStore() |
| 90 | return &Manager{ |
| 91 | Logger: logger.New().With( |
| 92 | slog.String("component", "functions manager"), |
| 93 | ), |
| 94 | api: netdataapi.New(safewriter.Stdout), |
| 95 | input: newStdinInput(), |
| 96 | mux: &sync.Mutex{}, |
| 97 | functionRegistry: make(map[string]*functionSet), |
| 98 | workerCount: defaultWorkerCount, |
| 99 | queueSize: defaultQueueSize, |
| 100 | invStateMux: &sync.Mutex{}, |
| 101 | invState: make(map[string]*invocationRecord), |
| 102 | tombstones: make(map[string]time.Time), |
| 103 | tombstoneTTL: defaultTombstoneTTL, |
| 104 | cancelFallbackDelay: defaultCancelFallbackDelay, |
| 105 | shutdownDrainTimeout: defaultShutdownDrainTimeout, |
| 106 | awaitingWarnDelay: defaultAwaitingWarnDelay, |
| 107 | runtimeStore: runtimeStore, |
| 108 | runtimeMetrics: newManagerRuntimeMetrics(runtimeStore), |
| 109 | } |
| 110 | } |
| 111 | |
| 112 | type Manager struct { |
| 113 | *logger.Logger |
| 114 | |
| 115 | api *netdataapi.API |
| 116 | |
| 117 | input input |
| 118 | |
| 119 | mux *sync.Mutex |
| 120 | functionRegistry map[string]*functionSet |
| 121 | |
| 122 | workerCount int |
| 123 | queueSize int |
| 124 | |
| 125 | scheduler *keyScheduler |
| 126 | |
| 127 | invStateMux *sync.Mutex |
| 128 | invState map[string]*invocationRecord |
| 129 | tombstones map[string]time.Time |
| 130 | tombstoneTTL time.Duration |
| 131 | cancelFallbackDelay time.Duration |
| 132 | shutdownDrainTimeout time.Duration |
| 133 | awaitingWarnDelay time.Duration |
| 134 | stopping atomic.Bool |
| 135 | |
| 136 | runtimeService runtimecomp.Service |
| 137 | runtimeStore metrix.RuntimeStore |
| 138 | runtimeMetrics *managerRuntimeMetrics |
| 139 | runtimeComponentName string |
| 140 | runtimeComponentRegistered bool |
| 141 | } |
| 142 | |
| 143 | func (m *Manager) Run(ctx context.Context, quitCh chan struct{}) { |
| 144 | m.Info("instance is started") |
| 145 | defer func() { m.Info("instance is stopped") }() |
| 146 | |
| 147 | if err := m.registerRuntimeComponent(); err != nil { |
| 148 | m.Warningf("runtime metrics registration failed: %v", err) |
| 149 | } else { |
| 150 | defer m.unregisterRuntimeComponent() |
| 151 | } |
| 152 | |
| 153 | m.run(ctx, quitCh) |
| 154 | } |
| 155 | |
| 156 | func (m *Manager) run(ctx context.Context, quitCh chan struct{}) { |
| 157 | parser := newInputParser() |
| 158 | m.scheduler = newKeyScheduler(m.queueSize) |
| 159 | m.observeSchedulerPending() |
| 160 | var workersWG sync.WaitGroup |
| 161 | |
| 162 | m.startWorkers(&workersWG) |
| 163 | |
| 164 | // Wake any enqueue() blocked on a full scheduler when the parent context |
| 165 | // is canceled. Necessary because the reader loop below may itself be |
| 166 | // blocked inside dispatchInvocation -> scheduler.enqueue waiting for |
| 167 | // space, and would otherwise miss the ctx.Done signal. |
| 168 | stopWatcher := make(chan struct{}) |
| 169 | defer close(stopWatcher) |
| 170 | go func() { |
| 171 | select { |
| 172 | case <-ctx.Done(): |
| 173 | if m.scheduler != nil { |
| 174 | m.scheduler.stop() |
| 175 | } |
| 176 | case <-stopWatcher: |
| 177 | } |
| 178 | }() |
| 179 | |
| 180 | for { |
| 181 | select { |
| 182 | case <-ctx.Done(): |
| 183 | m.shutdown(false, true, quitCh, &workersWG) |
| 184 | return |
| 185 | case line, ok := <-m.input.lines(): |
| 186 | if !ok { |
| 187 | m.shutdown(false, true, quitCh, &workersWG) |
| 188 | return |
| 189 | } |
| 190 | event, err := parser.parseEvent(line) |
| 191 | if err != nil { |
| 192 | m.Warningf("parse function: %v ('%s')", err, line) |
| 193 | continue |
| 194 | } |
| 195 | |
| 196 | switch event.kind { |
| 197 | case inputEventNone, inputEventProgress: |
| 198 | continue |
| 199 | case inputEventQuit: |
| 200 | m.shutdown(true, true, quitCh, &workersWG) |
| 201 | return |
| 202 | case inputEventCancel: |
| 203 | m.handleCancelEvent(event) |
| 204 | continue |
| 205 | case inputEventCall: |
| 206 | m.observeFunctionCall() |
| 207 | m.dispatchInvocation(ctx, event.fn) |
| 208 | } |
| 209 | } |
| 210 | } |
| 211 | } |
| 212 | |
| 213 | func (m *Manager) startWorkers(workersWG *sync.WaitGroup) { |
| 214 | if workersWG == nil { |
| 215 | return |
| 216 | } |
| 217 | |
| 218 | for range m.workerCount { |
| 219 | workersWG.Go(m.runWorker) |
| 220 | } |
| 221 | } |
| 222 | |
| 223 | func (m *Manager) shutdown(signalQuit, cancelInflight bool, quitCh chan struct{}, workersWG *sync.WaitGroup) { |
| 224 | m.setStopping(true) |
| 225 | m.signalQuitIfRequested(signalQuit, quitCh) |
| 226 | m.stopSchedulerAdmission() |
| 227 | timedOut := m.waitWorkers(workersWG) |
| 228 | m.finalizeUnresolvedOnShutdown(cancelInflight, timedOut) |
| 229 | } |
| 230 | |
| 231 | func (m *Manager) signalQuitIfRequested(signalQuit bool, quitCh chan struct{}) { |
| 232 | if signalQuit && quitCh != nil { |
| 233 | quitCh <- struct{}{} |
| 234 | } |
| 235 | } |
| 236 | |
| 237 | func (m *Manager) stopSchedulerAdmission() { |
| 238 | if m.scheduler != nil { |
| 239 | m.scheduler.stopAccepting() |
| 240 | } |
| 241 | } |
| 242 | |
| 243 | func (m *Manager) waitWorkers(workersWG *sync.WaitGroup) bool { |
| 244 | if workersWG == nil { |
| 245 | return false |
| 246 | } |
| 247 | |
| 248 | drainCtx, cancelDrain := context.WithTimeout(context.Background(), m.shutdownDrainTimeout) |
| 249 | defer cancelDrain() |
| 250 | |
| 251 | done := make(chan struct{}) |
| 252 | go func() { |
| 253 | defer close(done) |
| 254 | workersWG.Wait() |
| 255 | }() |
| 256 | |
| 257 | select { |
| 258 | case <-done: |
| 259 | return false |
| 260 | case <-drainCtx.Done(): |
| 261 | return true |
| 262 | } |
| 263 | } |
| 264 | |
| 265 | func (m *Manager) finalizeUnresolvedOnShutdown(cancelInflight, timedOut bool) { |
| 266 | if !cancelInflight { |
| 267 | return |
| 268 | } |
| 269 | if !timedOut && !m.hasActiveInvocations() { |
| 270 | return |
| 271 | } |
| 272 | |
| 273 | m.cancelAllInvocations() |
| 274 | m.forceFinalizeAll(499, "request canceled during shutdown") |
| 275 | if m.scheduler != nil { |
| 276 | m.scheduler.stop() |
| 277 | } |
| 278 | } |
| 279 | |
| 280 | func (m *Manager) dispatchInvocation(parentCtx context.Context, fn *Function) { |
| 281 | if fn == nil { |
| 282 | return |
| 283 | } |
| 284 | if m.isStopping() { |
| 285 | m.respf(fn, 503, "functions manager is stopping") |
| 286 | return |
| 287 | } |
| 288 | |
| 289 | handler, scheduleKey, ok := m.lookupFunctionRoute(*fn) |
| 290 | if !ok { |
| 291 | m.Infof("skipping execution of '%s': unregistered function", fn.Name) |
| 292 | m.respf(fn, 501, "unregistered function: %s", fn.Name) |
| 293 | return |
| 294 | } |
| 295 | if handler == nil { |
| 296 | m.Warningf("skipping execution of '%s': nil function registered", fn.Name) |
| 297 | m.respf(fn, 501, "nil function: %s", fn.Name) |
| 298 | return |
| 299 | } |
| 300 | |
| 301 | reqCtx, cancel := context.WithCancel(parentCtx) |
| 302 | switch m.trySetInvocationState(fn.UID, stateQueued, cancel, scheduleKey) { |
| 303 | case invocationAdmissionAccepted: |
| 304 | // admitted |
| 305 | case invocationAdmissionDuplicateActive: |
| 306 | cancel() |
| 307 | // Do not emit terminal output for duplicates of an active UID. Emitting |
| 308 | // via tryFinalize would mutate active tracking for the original invocation. |
| 309 | m.Warningf("ignoring duplicate active transaction id: %s", fn.UID) |
| 310 | m.observeDuplicateUIDIgnored() |
| 311 | return |
| 312 | case invocationAdmissionDuplicateTombstone: |
| 313 | cancel() |
| 314 | m.Warningf("ignoring duplicate recently finalized transaction id: %s", fn.UID) |
| 315 | m.observeDuplicateUIDIgnored() |
| 316 | return |
| 317 | case invocationAdmissionInvalid: |
| 318 | cancel() |
| 319 | m.Warningf("ignoring invalid transaction id: %q", fn.UID) |
| 320 | return |
| 321 | default: |
| 322 | cancel() |
| 323 | m.Warningf("ignoring transaction id '%s': unsupported admission state", fn.UID) |
| 324 | return |
| 325 | } |
| 326 | |
| 327 | req := &invocationRequest{ |
| 328 | fn: fn, |
| 329 | handler: handler, |
| 330 | ctx: reqCtx, |
| 331 | scheduleKey: scheduleKey, |
| 332 | } |
| 333 | |
| 334 | if m.scheduler == nil { |
| 335 | cancel() |
| 336 | m.respf(fn, 503, "functions manager is stopping") |
| 337 | return |
| 338 | } |
| 339 | |
| 340 | // scheduler.enqueue blocks when the queue is full; it only returns an |
| 341 | // error for invalid inputs or when the scheduler is stopping (shutdown). |
| 342 | // The blocking is intentional: by stalling here, the stdin reader slows |
| 343 | // down as well, which propagates back-pressure to netdata through the OS |
| 344 | // pipe instead of allowing unbounded buffering or dropping requests. |
| 345 | if err := m.scheduler.enqueue(req); err != nil { |
| 346 | cancel() |
| 347 | switch { |
| 348 | case errors.Is(err, errSchedulerStopping): |
| 349 | m.respf(fn, 503, "functions manager is stopping") |
| 350 | case errors.Is(err, errSchedulerInvalid): |
| 351 | m.respf(fn, 500, "invalid scheduler request") |
| 352 | default: |
| 353 | // Should be unreachable: enqueue's other return points block. |
| 354 | // If a new error variant is added in the future, surface it |
| 355 | // loudly instead of swallowing it as a generic 5xx. |
| 356 | m.Warningf("unexpected scheduler enqueue error for '%s': %v", fn.Name, err) |
| 357 | m.respf(fn, 500, "unexpected scheduler error: %v", err) |
| 358 | } |
| 359 | return |
| 360 | } |
| 361 | m.observeSchedulerPending() |
| 362 | } |
| 363 | |
| 364 | func (m *Manager) trySetInvocationState(uid string, state invocationState, cancel context.CancelFunc, scheduleKey string) invocationAdmission { |
| 365 | if uid == "" { |
| 366 | return invocationAdmissionInvalid |
| 367 | } |
| 368 | |
| 369 | m.invStateMux.Lock() |
| 370 | defer m.invStateMux.Unlock() |
| 371 | |
| 372 | m.pruneExpiredTombstonesLocked(time.Now()) |
| 373 | |
| 374 | if _, ok := m.tombstones[uid]; ok { |
| 375 | return invocationAdmissionDuplicateTombstone |
| 376 | } |
| 377 | |
| 378 | if _, ok := m.invState[uid]; ok { |
| 379 | return invocationAdmissionDuplicateActive |
| 380 | } |
| 381 | m.invState[uid] = &invocationRecord{ |
| 382 | state: state, |
| 383 | cancel: cancel, |
| 384 | scheduleKey: scheduleKey, |
| 385 | } |
| 386 | m.observeInvocationsLocked() |
| 387 | return invocationAdmissionAccepted |
| 388 | } |
| 389 | |
| 390 | func (m *Manager) setAwaitingResultState(uid string, fnTimeout time.Duration) { |
| 391 | if uid == "" { |
| 392 | return |
| 393 | } |
| 394 | |
| 395 | m.invStateMux.Lock() |
| 396 | defer m.invStateMux.Unlock() |
| 397 | |
| 398 | rec, ok := m.invState[uid] |
| 399 | if !ok || rec == nil { |
| 400 | return |
| 401 | } |
| 402 | |
| 403 | rec.state = stateAwaitingResult |
| 404 | rec.awaitingSince = time.Now() |
| 405 | m.startAwaitingTimerLocked(uid, rec, fnTimeout) |
| 406 | m.observeInvocationsLocked() |
| 407 | } |
| 408 | |
| 409 | func (m *Manager) logAwaitingResult(uid string) { |
| 410 | if uid == "" { |
| 411 | return |
| 412 | } |
| 413 | |
| 414 | m.invStateMux.Lock() |
| 415 | rec, ok := m.invState[uid] |
| 416 | if !ok || rec == nil || rec.state != stateAwaitingResult { |
| 417 | m.invStateMux.Unlock() |
| 418 | return |
| 419 | } |
| 420 | age := time.Since(rec.awaitingSince) |
| 421 | m.invStateMux.Unlock() |
| 422 | |
| 423 | m.Warningf("transaction uid '%s' is still awaiting terminal response after %s", uid, age) |
| 424 | } |
| 425 | |
| 426 | func (m *Manager) startInvocation(uid string) bool { |
| 427 | if uid == "" { |
| 428 | return false |
| 429 | } |
| 430 | |
| 431 | m.invStateMux.Lock() |
| 432 | defer m.invStateMux.Unlock() |
| 433 | |
| 434 | rec, ok := m.invState[uid] |
| 435 | if !ok || rec == nil || rec.cancelRequested { |
| 436 | return false |
| 437 | } |
| 438 | rec.state = stateRunning |
| 439 | m.observeInvocationsLocked() |
| 440 | return true |
| 441 | } |
| 442 | |
| 443 | func (m *Manager) handleCancelEvent(event inputEvent) { |
| 444 | uid := event.uid |
| 445 | if uid == "" { |
| 446 | return |
| 447 | } |
| 448 | |
| 449 | if event.preAdmission { |
| 450 | m.respUID(uid, 499, "request canceled") |
| 451 | return |
| 452 | } |
| 453 | |
| 454 | if _, ok := m.requestCancellation(uid); !ok { |
| 455 | m.Debugf("ignoring cancel for unknown transaction id: %s", uid) |
| 456 | return |
| 457 | } |
| 458 | // No immediate terminal response. requestCancellation handled state |
| 459 | // transitions: for queued functions the cancel is intentionally ignored |
| 460 | // (function will run to completion); for running/awaiting functions a |
| 461 | // fallback timer was armed and will emit 499 + tombstone after |
| 462 | // cancelFallbackDelay. |
| 463 | } |
| 464 | |
| 465 | func (m *Manager) requestCancellation(uid string) (invocationState, bool) { |
| 466 | m.invStateMux.Lock() |
| 467 | defer m.invStateMux.Unlock() |
| 468 | |
| 469 | rec, ok := m.invState[uid] |
| 470 | if !ok || rec == nil { |
| 471 | return 0, false |
| 472 | } |
| 473 | |
| 474 | // For stateQueued we deliberately ignore the cancel entirely: no |
| 475 | // cancelRequested flag, no ctx cancellation, no fallback timer, no |
| 476 | // tombstone. Reason: dyncfg commands (enable/disable/update/restart) |
| 477 | // carry side-effects that must reach jobmgr, otherwise the wait gate |
| 478 | // stays in 'accepted' forever (since the wait-decision timeout was |
| 479 | // removed). Setting cancelRequested would make startInvocation skip the |
| 480 | // handler when the worker pulls; the fallback timer's tryFinalize would |
| 481 | // also tombstone+remove from invState before the worker gets there. So |
| 482 | // either path wedges the function. We let it run to completion as if |
| 483 | // nothing happened; netdata already considers the transaction done (it |
| 484 | // 504'd before sending CANCEL), so the eventual terminal response just |
| 485 | // produces a benign "transaction not found" log on the netdata side. |
| 486 | if rec.state == stateQueued { |
| 487 | return rec.state, true |
| 488 | } |
| 489 | |
| 490 | if rec.cancelRequested { |
| 491 | return rec.state, true |
| 492 | } |
| 493 | rec.cancelRequested = true |
| 494 | if rec.cancel != nil { |
| 495 | rec.cancel() |
| 496 | } |
| 497 | |
| 498 | m.startCancelFallbackTimerLocked(uid, rec) |
| 499 | |
| 500 | return rec.state, true |
| 501 | } |
| 502 | |
| 503 | func (m *Manager) setStopping(v bool) { |
| 504 | m.stopping.Store(v) |
| 505 | } |
| 506 | |
| 507 | func (m *Manager) isStopping() bool { |
| 508 | return m.stopping.Load() |
| 509 | } |
| 510 | |
| 511 | func (m *Manager) hasActiveInvocations() bool { |
| 512 | m.invStateMux.Lock() |
| 513 | defer m.invStateMux.Unlock() |
| 514 | return len(m.invState) > 0 |
| 515 | } |
| 516 | |
| 517 | // TerminalFinalizer returns the manager-bound terminal finalizer for responder wiring. |
| 518 | func (m *Manager) TerminalFinalizer() TerminalFinalizer { |
| 519 | return m.finalizeTerminal |
| 520 | } |
| 521 | |
| 522 | func (m *Manager) finalizeTerminal(uid, source string, emit func()) bool { |
| 523 | return m.tryFinalize(uid, source, emit) |
| 524 | } |
| 525 | |
| 526 | func (m *Manager) cancelAllInvocations() { |
| 527 | m.invStateMux.Lock() |
| 528 | defer m.invStateMux.Unlock() |
| 529 | |
| 530 | for _, rec := range m.invState { |
| 531 | if rec == nil { |
| 532 | continue |
| 533 | } |
| 534 | rec.cancelRequested = true |
| 535 | if rec.cancel != nil { |
| 536 | rec.cancel() |
| 537 | } |
| 538 | } |
| 539 | } |
| 540 | |
| 541 | func (m *Manager) forceFinalizeAll(code int, message string) { |
| 542 | m.invStateMux.Lock() |
| 543 | uids := make([]string, 0, len(m.invState)) |
| 544 | for uid := range m.invState { |
| 545 | uids = append(uids, uid) |
| 546 | } |
| 547 | m.invStateMux.Unlock() |
| 548 | |
| 549 | for _, uid := range uids { |
| 550 | m.respUID(uid, code, "%s", message) |
| 551 | } |
| 552 | } |
| 553 | |
| 554 | // markCancelled performs the same bookkeeping as tryFinalize (tombstone + |
| 555 | // scheduler.complete + remove from invState) but does NOT emit anything to |
| 556 | // netdata. Used by the cancel fallback path: by the time CANCEL is sent, |
| 557 | // netdata has already timed out the transaction and removed its inflight |
| 558 | // entry, so any response we emit just produces a "transaction not found" |
| 559 | // log on the netdata side. We still need the bookkeeping so the lane |
| 560 | // advances and any later terminal response from the handler is dropped. |
| 561 | func (m *Manager) markCancelled(uid string) { |
| 562 | if uid == "" { |
| 563 | return |
| 564 | } |
| 565 | |
| 566 | m.invStateMux.Lock() |
| 567 | now := time.Now() |
| 568 | m.pruneExpiredTombstonesLocked(now) |
| 569 | if _, ok := m.tombstones[uid]; ok { |
| 570 | m.invStateMux.Unlock() |
| 571 | return |
| 572 | } |
| 573 | |
| 574 | var scheduleKey string |
| 575 | if rec, ok := m.invState[uid]; ok && rec != nil { |
| 576 | m.stopTimersLocked(rec) |
| 577 | scheduleKey = rec.scheduleKey |
| 578 | } |
| 579 | delete(m.invState, uid) |
| 580 | m.tombstones[uid] = now.Add(m.tombstoneTTL) |
| 581 | m.observeInvocationsLocked() |
| 582 | m.invStateMux.Unlock() |
| 583 | |
| 584 | if scheduleKey != "" && m.scheduler != nil { |
| 585 | m.scheduler.complete(scheduleKey, uid) |
| 586 | m.observeSchedulerPending() |
| 587 | } |
| 588 | } |
| 589 | |
| 590 | // tryFinalize emits a terminal response once per transaction UID. |
| 591 | // Later terminal attempts for the same UID are dropped while tombstone is active. |
| 592 | func (m *Manager) tryFinalize(uid, source string, emit func()) bool { |
| 593 | if uid == "" || emit == nil { |
| 594 | return false |
| 595 | } |
| 596 | |
| 597 | m.invStateMux.Lock() |
| 598 | now := time.Now() |
| 599 | m.pruneExpiredTombstonesLocked(now) |
| 600 | if _, ok := m.tombstones[uid]; ok { |
| 601 | m.invStateMux.Unlock() |
| 602 | m.Debugf("dropping late terminal response for uid '%s' (source=%s)", uid, source) |
| 603 | m.observeLateTerminalDropped() |
| 604 | return false |
| 605 | } |
| 606 | |
| 607 | var scheduleKey string |
| 608 | if rec, ok := m.invState[uid]; ok && rec != nil { |
| 609 | m.stopTimersLocked(rec) |
| 610 | scheduleKey = rec.scheduleKey |
| 611 | } |
| 612 | delete(m.invState, uid) |
| 613 | m.tombstones[uid] = now.Add(m.tombstoneTTL) |
| 614 | m.observeInvocationsLocked() |
| 615 | m.invStateMux.Unlock() |
| 616 | |
| 617 | if scheduleKey != "" && m.scheduler != nil { |
| 618 | m.scheduler.complete(scheduleKey, uid) |
| 619 | m.observeSchedulerPending() |
| 620 | } |
| 621 | |
| 622 | emit() |
| 623 | return true |
| 624 | } |
| 625 | |
| 626 | func (m *Manager) pruneExpiredTombstonesLocked(now time.Time) { |
| 627 | for uid, expiresAt := range m.tombstones { |
| 628 | if !expiresAt.After(now) { |
| 629 | delete(m.tombstones, uid) |
| 630 | } |
| 631 | } |
| 632 | } |
| 633 | |
| 634 | // startAwaitingTimerLocked starts/refreshes awaiting-result warning timer. |
| 635 | // Caller must hold m.invStateMux. |
| 636 | func (m *Manager) startAwaitingTimerLocked(uid string, rec *invocationRecord, fnTimeout time.Duration) { |
| 637 | if uid == "" || rec == nil { |
| 638 | return |
| 639 | } |
| 640 | |
| 641 | delay := m.awaitingWarnDelay |
| 642 | if fnTimeout > 0 && fnTimeout < delay { |
| 643 | delay = fnTimeout |
| 644 | } |
| 645 | if delay <= 0 { |
| 646 | return |
| 647 | } |
| 648 | |
| 649 | if rec.awaitingTimer != nil { |
| 650 | rec.awaitingTimer.Stop() |
| 651 | } |
| 652 | uidCopy := uid |
| 653 | rec.awaitingTimer = time.AfterFunc(delay, func() { |
| 654 | m.logAwaitingResult(uidCopy) |
| 655 | }) |
| 656 | } |
| 657 | |
| 658 | // startCancelFallbackTimerLocked starts cancel fallback timer once. |
| 659 | // Caller must hold m.invStateMux. |
| 660 | func (m *Manager) startCancelFallbackTimerLocked(uid string, rec *invocationRecord) { |
| 661 | if uid == "" || rec == nil || rec.fallbackTimer != nil { |
| 662 | return |
| 663 | } |
| 664 | |
| 665 | uidCopy := uid |
| 666 | rec.fallbackTimer = time.AfterFunc(m.cancelFallbackDelay, func() { |
| 667 | m.observeCancelFallback() |
| 668 | // Don't emit a 499: by the time CANCEL was sent, netdata has already |
| 669 | // 504'd the transaction and removed its inflight entry, so any |
| 670 | // response we send just produces a "transaction not found" log. |
| 671 | // markCancelled does the bookkeeping (tombstone + lane advance) so |
| 672 | // the late real response from the handler is dropped. |
| 673 | m.markCancelled(uidCopy) |
| 674 | }) |
| 675 | } |
| 676 | |
| 677 | // stopTimersLocked stops invocation timers and clears timer references. |
| 678 | // Caller must hold m.invStateMux. |
| 679 | func (m *Manager) stopTimersLocked(rec *invocationRecord) { |
| 680 | if rec == nil { |
| 681 | return |
| 682 | } |
| 683 | |
| 684 | if rec.fallbackTimer != nil { |
| 685 | rec.fallbackTimer.Stop() |
| 686 | rec.fallbackTimer = nil |
| 687 | } |
| 688 | if rec.awaitingTimer != nil { |
| 689 | rec.awaitingTimer.Stop() |
| 690 | rec.awaitingTimer = nil |
| 691 | } |
| 692 | } |
| 693 | |
| 694 | type functionSnapshot struct { |
| 695 | direct func(Function) |
| 696 | prefixes map[string]func(Function) |
| 697 | } |
| 698 | |
| 699 | func (m *Manager) snapshotFunction(name string) (functionSnapshot, bool) { |
| 700 | m.mux.Lock() |
| 701 | fs, ok := m.functionRegistry[name] |
| 702 | snap := functionSnapshot{} |
| 703 | if ok && fs != nil { |
| 704 | snap.direct = fs.direct |
| 705 | if len(fs.prefixes) > 0 { |
| 706 | snap.prefixes = make(map[string]func(Function), len(fs.prefixes)) |
| 707 | maps.Copy(snap.prefixes, fs.prefixes) |
| 708 | } |
| 709 | } |
| 710 | m.mux.Unlock() |
| 711 | |
| 712 | if !ok || fs == nil { |
| 713 | return functionSnapshot{}, false |
| 714 | } |
| 715 | return snap, true |
| 716 | } |
| 717 | |
| 718 | func matchPrefix(prefixes map[string]func(Function), id string) (string, func(Function), bool) { |
| 719 | if len(prefixes) == 0 || id == "" { |
| 720 | return "", nil, false |
| 721 | } |
| 722 | |
| 723 | for prefix, handler := range prefixes { |
| 724 | if strings.HasPrefix(id, prefix) { |
| 725 | return prefix, handler, true |
| 726 | } |
| 727 | } |
| 728 | |
| 729 | return "", nil, false |
| 730 | } |
| 731 | |
| 732 | func (m *Manager) lookupFunctionRoute(fn Function) (handler func(Function), scheduleKey string, ok bool) { |
| 733 | snap, ok := m.snapshotFunction(fn.Name) |
| 734 | if !ok { |
| 735 | return nil, "", false |
| 736 | } |
| 737 | unknownHandler := m.unknownFunctionHandler() |
| 738 | |
| 739 | if len(snap.prefixes) > 0 { |
| 740 | if len(fn.Args) > 0 { |
| 741 | id := fn.Args[0] |
| 742 | if prefix, routeHandler, matched := matchPrefix(snap.prefixes, id); matched && routeHandler != nil { |
| 743 | return routeHandler, routeScheduleKey(fn.Name, prefix), true |
| 744 | } |
| 745 | } |
| 746 | |
| 747 | return unknownHandler, routeScheduleKey(fn.Name, scheduleKeyUnmatched), true |
| 748 | } |
| 749 | |
| 750 | if snap.direct != nil { |
| 751 | return snap.direct, routeScheduleKey(fn.Name, ""), true |
| 752 | } |
| 753 | |
| 754 | return unknownHandler, routeScheduleKey(fn.Name, scheduleKeyDirectMissing), true |
| 755 | } |
| 756 | |
| 757 | // lookupFunction returns a snapshot handler used by existing tests to verify |
| 758 | // registry snapshot semantics at lookup time. |
| 759 | func (m *Manager) lookupFunction(name string) (func(Function), bool) { |
| 760 | snap, ok := m.snapshotFunction(name) |
| 761 | if !ok { |
| 762 | return nil, false |
| 763 | } |
| 764 | unknownHandler := m.unknownFunctionHandler() |
| 765 | |
| 766 | return func(f Function) { |
| 767 | if len(snap.prefixes) > 0 { |
| 768 | if len(f.Args) > 0 { |
| 769 | id := f.Args[0] |
| 770 | if _, handler, matched := matchPrefix(snap.prefixes, id); matched && handler != nil { |
| 771 | handler(f) |
| 772 | return |
| 773 | } |
| 774 | } |
| 775 | unknownHandler(f) |
| 776 | return |
| 777 | } |
| 778 | |
| 779 | if snap.direct != nil { |
| 780 | snap.direct(f) |
| 781 | return |
| 782 | } |
| 783 | |
| 784 | unknownHandler(f) |
| 785 | }, true |
| 786 | } |
| 787 | |
| 788 | func (m *Manager) unknownFunctionHandler() func(Function) { |
| 789 | return func(f Function) { |
| 790 | m.respf(&f, 503, "unknown function '%s' (%v)", f.Name, f.Args) |
| 791 | } |
| 792 | } |
| 793 | |
| 794 | const ( |
| 795 | scheduleKeyUnmatched = "__unmatched__" |
| 796 | scheduleKeyDirectMissing = "__direct_missing__" |
| 797 | ) |
| 798 | |
| 799 | func routeScheduleKey(name, discriminator string) string { |
| 800 | if discriminator == "" { |
| 801 | return name |
| 802 | } |
| 803 | return name + "|" + discriminator |
| 804 | } |
| 805 | |
| 806 | func (m *Manager) respUID(uid string, code int, msgf string, a ...any) { |
| 807 | if uid == "" { |
| 808 | return |
| 809 | } |
| 810 | m.respf(&Function{UID: uid}, code, msgf, a...) |
| 811 | } |
| 812 | |
| 813 | func (m *Manager) respf(fn *Function, code int, msgf string, a ...any) { |
| 814 | if fn == nil || fn.UID == "" { |
| 815 | return |
| 816 | } |
| 817 | |
| 818 | msg := fmt.Sprintf(msgf, a...) |
| 819 | bs := BuildJSONPayload(code, msg) |
| 820 | |
| 821 | res := netdataapi.FunctionResult{ |
| 822 | UID: fn.UID, |
| 823 | ContentType: "application/json", |
| 824 | Payload: string(bs), |
| 825 | Code: strconv.Itoa(code), |
| 826 | ExpireTimestamp: strconv.FormatInt(time.Now().Unix(), 10), |
| 827 | } |
| 828 | |
| 829 | m.finalizeTerminal(fn.UID, "functions.manager.respf", func() { |
| 830 | m.api.FUNCRESULT(res) |
| 831 | }) |
| 832 | } |