EDG-881: Wire fetch_logs into BaseExtractor and upload via CDF Files by vikramlc-cognite · Pull Request #719 · cognitedata/dotnet-extractor-utils

vikramlc-cognite · 2026-10-06T07:50:29Z

Summary

Second of two PRs for fetch_logs (EDG-881). Builds on the groundwork primitives in #718 (log file naming/enumeration, date validation, BoundedFileStream) to complete the action: orchestrates validate → enumerate → open → upload → aggregate result, uploads rotated log files to CDF Files, and registers the action in BaseExtractor so it's actually advertised and callable. This is the PR that makes fetch_logs live.

Replaces #717 (closed as part of re-splitting this work into groundwork + wiring, rather than code + tests).

Type of change

Bug fix (non-breaking change that fixes an issue)
New feature (non-breaking change that adds functionality)
Breaking change (fix or feature that changes existing behavior)
Refactor (no functional change)
Documentation update
Chore / tooling / CI

What changed

FetchLogsAction.RunAsync: validates config, enumerates candidate files, uploads each via CDF Files, aggregates fileCount/totalBytes/per-file-size metadata, and reports progress; skipped (missing) and failed (network/CDF error) files are tracked separately so a partial failure doesn't void files already uploaded.
FetchLogsAction.UploadLogFileAsync: the actual CDF Files upload call, using a pooled IHttpClientFactory client when available and falling back to a shared static HttpClient otherwise.
BaseExtractor.RegisterBuiltInActions(), called right after InitTasks() and before InitActions(), so a user-defined custom action still wins any name collision on fetch_logs (built-ins register first, so a colliding user registration fails loudly at startup rather than being silently shadowed).
Renamed a pre-existing test's custom action from "fetch_logs" to "my_custom_action" (BaseExtractorTest), since it now collides with this built-in; updated its advertised-actions-count assertion from 3 to 4.

Why it changed

Related issue: https://cognitedata.atlassian.net/browse/EDG-881

Giving Odin operators a way to pull recent logs without needing host/container access.

What to focus on during review

Error aggregation in RunAsync: a succeeded result with zero files only happens when every candidate was genuinely missing/skipped, not when every upload failed — a run where every upload fails throws ActionError("upload_failed", ...) instead of silently reporting success with 0 files.
fetch_logs is unconditionally registered for every extractor; if LoggerConfig.File isn't configured, the action correctly reports an ActionError rather than being hidden — but this does mean the action is always advertised even when it can never succeed.

Test evidence

FetchLogsActionTest.cs: 9 additional RunAsync-level tests (missing file-handler/CDF-client config, skipped vs. failed dates, successful upload with progress/metadata, partial-failure handling, all-uploads-fail case).
BaseExtractorTest.cs: updated for the new built-in's registration and the renamed collision.
Full non-integration suite: 65/65 passing (groundwork + wiring tests combined). Builds clean on netstandard2.0, netstandard2.1, and net8.0/net9.0.

Risks and unknowns

fetch_logs is unconditionally registered for every extractor; if LoggerConfig.File isn't configured, the action correctly reports an ActionError rather than being hidden — not fixed here since suppressing an advertised action based on runtime config wasn't in scope; worth a follow-up if this proves confusing to operators.

Rollout and rollback

No config or migration changes. Purely additive.

Checklist

Self-reviewed the diff
Tests added or updated (or N/A with reason)
Docs updated (or N/A)
No secrets, credentials, or PII committed
Breaking changes called out above and communicated to affected teams

RegisteredTask.FinishTask could not distinguish a deliberate cancellation (a Stop action, or the scheduler itself shutting down) from a genuine unexpected failure. Any cancellation of a task with ErrorIsFatal=true was rethrown, which -- combined with ExtractorRestartPolicy.OnError, as opcua-extractor-net uses on 3 of its 4 tasks -- crashes and restarts the entire extractor process. This was reachable today with no Actions feature involved: ordinary Shutdown()/DisposeAsync() cancels the scheduler's internal token before the outer token, hitting the same fatal path for any ErrorIsFatal continuous task still running at shutdown time. Track intentional cancellation via two signals: a per-task flag set by RegisteredTask.Cancel() (the Stop-action case), and the scheduler's own internal cancellation-requested state (the shutdown case). FinishTask now only escalates to a fatal, scheduler-terminating exception when neither applies -- a genuine failure still propagates exactly as before. Both signals are checked explicitly rather than treating "any cancellation is intentional" as an axiom, so this stays correct if a future change introduces another cancellation source. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

…query, and Actions wire-contract regression tests The dispatch engine that will handle Start/Stop actions (EDG-878) needs to distinguish "successfully cancelled/scheduled" from "there was nothing to do" -- CancelTask/ScheduleTaskNow's void return can't express this. Add TryCancelTask/TryScheduleTaskNow, returning that outcome explicitly. CancelTask now delegates to TryCancelTask; ScheduleTaskNow is left as-is since reScheduleIfRunning has no equivalent in the Try* variant. Also add CanTaskRunNow, exposing BaseSchedulableTask.CanRunNow() by task name. A successful TryScheduleTaskNow only means a task was queued -- CanRunNow independently and unboundedly gates whether it actually starts (e.g. opcua-extractor-net's BrowseTask/SubscriptionsTask/ HistoryTask can stay un-runnable for extended periods before a session connects). Without this, a Start handler would block on WaitForNextEndOfTask with no way to fail fast. Separately, add ActionsWireContractTests pinning JSON/enum behaviors the whole Actions feature depends on: ActionStatus/ActionType serialize to their exact wire strings, and CheckInResponse/ActionUpdate tolerate unknown properties -- currently true only because System.Text.Json's default is permissive and nothing overrides it, which is exactly the kind of implicit correctness that silently breaks if a future change elsewhere adds strict deserialization options. Confirmed via the actual cognite-sdk-dotnet v5.13.0 source that CheckInResponse.PendingActions is List<IntegrationAction>, which does carry a Status field on redelivered entries -- resolving an open question from ACTIONS_FEATURE_DESIGN.md's re-verification pass about whether cancel_pending redelivery is even detectable client-side (it is, no SDK follow-up needed). Deliberately out of scope here: a guard against the client emitting `pending`/`cancel_pending` in an outgoing ActionUpdate. There is nothing to guard yet -- no code in this repo constructs ActionUpdate for sending -- so add that guard where the dispatcher actually builds one (EDG-879), not speculatively here. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

…test TestTryScheduleTaskNowReturnsTrueAndRunsWhenIdle called TryScheduleTaskNow *then* WaitForNextEndOfTask -- the same ordering bug found and fixed in BaseExtractor.RunStartTaskAction while working on EDG-878 (a fast-completing task can finish and flush its waiters before WaitForNextEndOfTask's AddWaiter call registers, hanging until the timeout). This test just hadn't hit it yet. Confirmed by running it 40 times after the fix with zero failures, where it had failed intermittently in a full-suite run before. Fixed by registering the wait first, then triggering -- matching TestScheduler's existing convention in this same file. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

…spatcher registration Extends CheckInWorker (rather than introducing a parallel channel, per the Actions design doc's recommendation) with everything needed to turn "an action finished" into a wire-level checkin request, and "Odin wants an action run" into a callback invocation: - QueueActionUpdate(ActionUpdate): thread-safe, callable from any action-handler task once one exists (EDG-879). - SetActionDispatcher(Func<IReadOnlyList<IntegrationAction>, Task>): registers the callback invoked whenever a checkin *or* startup response carries pendingActions (both StartupAsync and CheckInAsync return CheckInResponse, so both go through the same HandleCheckInResponse path). Wrapped in try/catch that logs, never swallows silently and never lets a throwing dispatcher take down the check-in loop. - MAX_ACTION_UPDATES_PER_CHECKIN = 100, matching odin's confirmed limit. Both new methods are added to IIntegrationSink, so BaseExtractor can call them without casting once it owns an action registry (EDG-877). Updated the other three implementers (LogIntegrationSink, BootstrapErrorReporter, and the test DummySink) accordingly -- BootstrapErrorReporter throws, matching its existing behavior for task-related calls during the too-early bootstrap phase; LogIntegrationSink no-ops/logs, since it never talks to a real backend and so can never observe a pending action in the first place. Real-world correction found while implementing the batch-splitting logic: the design doc described this as needing to become a "three-way time-sorted merge" with errors and task updates. That's not actually possible -- CogniteSdk.Alpha.ActionUpdate (v5.13.0) has no timestamp field to sort by, unlike ErrorWithTask/TaskUpdate. Action updates are instead sent in FIFO queue order, batched independently of the existing error/task time-based merge, capped at 100 per checkin regardless of how many errors/tasks are batched alongside them. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Cross-checked against python-extractor-utils' actual checkin_worker.py (not just the design doc's porting guide) to validate the FIFO-batching approach from the previous commit. Confirmed: Python's ActionUpdate also has no timestamp field, and its real batching logic independently FIFO-slices action updates exactly like this port does -- the design doc's "three-way time-sorted merge" was never what Python actually does, not something this port diverged from. But the same check surfaced a real gap: Python's requeue-on-failure prepends unsent action updates back to the front of the queue (`self._action_updates = action_updates + self._action_updates`), this port appended them to the end instead. That distinction doesn't matter for errors/task updates, since both are re-sorted by timestamp on every send regardless of queue order -- but ActionUpdate has no timestamp and is always sent in raw queue order, so append-vs-prepend on retry actually changes wire order. A `running` progress update that failed to send could end up ordered *after* a newer `succeeded` update for the same action queued in the meantime, reversing their effective order at the server. Added a regression test that specifically queues the newer update while the failing send is still in flight (via a one-shot test hook), since a naive version of this test -- queuing the newer update only after Flush() had already returned -- can't distinguish insert-at-front from append-at-end (the queue is empty either way at requeue time) and would pass against the bug. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

…ctionError SDK surface Adds the registration surface extractor authors use to expose actions, and the per-invocation API surface a custom action's target receives: - BaseExtractor<TConfig>.InitActions(): new virtual hook, called after InitTasks() (see Init()) so that auto-generated Start/Stop actions for already-registered actionable tasks are known before custom actions register. - RegisterAction(CustomAction<TConfig>): registers a custom action, enforcing two guards at registration time: no two custom actions share a name, and no custom action collides with the auto-generated "Start <task>"/"Stop <task>" name of an already-registered actionable task. Only checked in this one direction, deliberately -- InitTasks() always running before InitActions() means every actionable task is already known by the time any custom action registers, so there's no practical "task registered after a colliding action name" case to guard against within the normal startup flow. - GetStartupRequest() now populates AvailableActions: an auto Start/Stop pair per actionable task (in task-registration order), then custom actions (in registration order) -- this is the actual wire-up point between "actions are registered" and "Odin learns about them". - ActionContext<TConfig>/ActionError/CustomAction<TConfig> (new Actions.cs): SetResult (throws InvalidOperationException if called twice, only records state for the dispatcher to read later -- doesn't itself queue anything), ReportProgress (queues a `running` update via IIntegrationSink.QueueActionUpdate immediately), and ActionError's structured errorType/errorDetail metadata. - ActionNaming: centralizes the "Start <task>"/"Stop <task>" string convention in one place, since both this ticket's advertising code and the future dispatch engine's routing code need to construct/parse the exact same strings. Note for whoever picks up the dispatch engine (EDG-878/879): while testing this, found a pre-existing, unrelated flaky test (BaseExtractorTest.TestBaseExtractor, intermittent, ~1 in 5-10 runs) that traces to RegisteredTask.Run's linked CancellationTokenSource -- it flips IsCancellationRequested on shutdown regardless of whether its task already completed naturally, so a task that finished cleanly just before shutdown can still be misreported as a cancellation warning. This predates all Actions work (confirmed: the racing task there has ErrorIsFatal=false, so EDG-874's fatal-rethrow guard is not involved) and is out of scope for these tickets, but worth its own follow-up. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Adds the engine that turns a checkin/startup response's pendingActions into real Start/Stop task operations, registered as the extractor's action dispatcher (IIntegrationSink.SetActionDispatcher) before the check-in worker starts. - DispatchActions/DispatchAction: routes by name ("Start <task>" / "Stop <task>", via EDG-877's ActionNaming), one independent Task.Run per action -- overlapping dispatch across actions and checkin cycles is intentional. Unrecognized names report `failed`; this currently includes registered custom actions too, since custom-action dispatch isn't implemented yet (EDG-879's scope) -- routing only distinguishes "matches Start/Stop" from "doesn't", not yet "doesn't, but is a known custom action". - Start handler: TryScheduleTaskNow, then CanTaskRunNow (fails fast rather than hanging if the task can't actually start yet), then blocks on WaitForNextEndOfTask with no timeout -- a dedicated Task.Run per action means blocking costs nothing else, and there's no principled timeout for how long a legitimate task should run; Stop/cancel_pending is the intended way to end it early. - Stop handler: TryCancelTask, reports `succeeded` on success (not `canceled` -- matches python-extractor-utils' convention). - A `cancel_pending` redelivery for a Start action resolves via TryCancelTask on the underlying task, reusing the Stop mechanism -- no custom cancellation infrastructure needed for this case (that's only required for custom actions, EDG-879). - Every dispatched unit of work has a mandatory outer try/catch: an unhandled exception anywhere always produces a `failed` update, never a silently dropped action. Two real bugs found and fixed while writing tests against actual dispatch (not just unit-testing the scheduler primitives in isolation): 1. RunStartTaskAction called TryScheduleTaskNow *then* WaitForNextEndOfTask. For a fast-completing task, the scheduler could finish and flush its waiters before the handler started listening, hanging it forever (this handler's wait has no timeout). Fixed by registering the wait first, then triggering -- matching this codebase's own existing convention in TaskSchedulerTest.TestScheduler. 2. A cancellation surfacing through WaitForNextEndOfTask can arrive as a bare TaskCanceledException *or* as AggregateException wrapping one, depending on which of two independent paths in ExtractorTaskScheduler resolves the waiter first (FinishTask's normal completion path vs. Cancel's immediate TrySetCanceled-on-any-current-waiter path -- the latter is also documented by the existing TestWaitWhenCancel test, which asserts AggregateException for exactly this reason). The Start handler now unwraps once before deciding `canceled` vs `failed`. Also documented (not fixed -- accepted as out of proportion to fix) one more narrow race: TryCancelTask/TryCancelTask's underlying Cancel() is a no-op while ActiveTask is still null, which it briefly is between TryScheduleTaskNow queuing a run and the scheduler's own loop actually starting it. In production this window is microseconds against a ~30s checkin interval; it only mattered in tests that dispatch a cancel_pending redelivery with no natural delay at all, which is why the corresponding test waits for the task to actually start (not just for the `running` report) before cancelling it. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

… wiring Completes the dispatch engine started in EDG-878 with custom action support: - DispatchAction now checks _customActions (from EDG-877's registry) after the Start/Stop routing, before falling through to "unknown action". A cancel_pending redelivery for a name that doesn't match Start/Stop is treated generically as a cancel-in-flight request for a custom action, looked up by externalId. - In-flight tracking (_inFlightCustomActions, ConcurrentDictionary<string, CancellationTokenSource>): added before invoking a custom action's target, atomically via TryAdd so a redelivery of the same externalId before the first dispatch completes is skipped rather than invoking the target twice concurrently. The same map backs cancel-in-flight: a cancel_pending redelivery looks up the CancellationTokenSource and cancels it -- the in-flight unit of work's own try/catch/finally is what reports the eventual terminal status, nothing is queued from the cancel path itself. - RunCustomAction: reports `running`, invokes the target with a per-action CancellationToken (not the general dispatcher/shutdown token), and on the happy path queues `succeeded` using whatever ActionContext.SetResult recorded (or an empty result if it was never called -- ownership clarified in EDG-877). On any exception, disambiguates `canceled` vs `failed` by token state (cts.IsCancellationRequested), not exception type -- a target that throws while its own cancellation was requested is reporting that cancellation's outcome, whether or not it threw a "clean" OperationCanceledException. ActionError exceptions surface their structured errorType/errorDetail metadata on a `failed` outcome. try/catch/finally throughout: the finally removes the in-flight entry and disposes the CancellationTokenSource unconditionally, not tied to the happy-path return, so a callback that throws before reaching any status-recording code can't leak the dedup entry or silently drop the action. - BaseExtractor now resolves an IntegrationExternalId (from ConnectionConfig, when registered via the full Runtime/Builder DI setup) to pass through ActionContext -- best-effort, since it's only actually observable in production, where a real integration is always present (nothing can invoke DispatchActions without one). Found and fixed two test bugs while verifying this (not production bugs): an exception-path test asserted on a single ActionUpdate for an externalId that actually has two (the initial `running` report plus the terminal one), and a dedup test asserted `canceled` for a scenario where nothing actually cancelled the token (a same-status redelivery is deduped/ignored, not cancelled, so the target completes normally once unblocked -- fixed the assertion to expect `succeeded`). Both new custom-action-dispatch and cancel-in-flight code paths were verified against sabotaged versions to confirm the tests catch real regressions, not just pass by construction. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Enforces odin's payload constraints centrally in CheckInWorker.QueueActionUpdate -- the single choke point every outgoing ActionUpdate passes through (both the dispatch engine's Start/Stop/custom handlers and ActionContext.ReportProgress), so callers get this for free without duplicating truncation/limit logic at each call site. - ResultMessage truncated to 1000 chars with a trailing "..." -- matches python-extractor-utils' MAX_MESSAGE_LENGTH constant. odin enforces no server-side limit here (confirmed unchanged: no length validator, unconstrained TEXT column), so this is for cross-language parity and payload-size hygiene, not a technical necessity. - ResultMetadata checked against all four of odin's verified limits (16 keys, 32-byte keys, 512-byte values, 4096-byte total -- read directly from cognite-fastapi's _design_guide_types.py). On violation: sanitized (reduced to fit) and a note appended to the result message, but Status is left untouched -- matches python-extractor-utils' actual shipped drop-and-preserve-status behavior, not the drop-and-fail policy an earlier design draft assumed by analogy with the original porting guide. Reused Cognite.Extensions/Sanitation.cs's existing SanitizeMetadata/ VerifyMetadata helpers rather than writing new limit-checking logic -- they already implement exactly this shape (per-key-bytes, key-count, per-value-bytes, total-bytes) for other CDF API sanitization elsewhere in this codebase, so this ticket ended up needing no new algorithm, just wiring. Verified against sabotaged code (guardrail call commented out) that all three new tests (oversized value, too many keys, over-length message) fail without the fix and pass with it. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

…'s own lifetime A custom action's CancellationTokenSource was a bare `new CancellationTokenSource()`, unlinked to anything -- unlike every other cancellation source in this class (RegisteredTask's per-run token is a child of the scheduler's, which is itself a child of Source). This diverges from python-extractor-utils, where every task/action token is a create_child_token() of the single root cancellation_token, so a shutdown cancels everything transitively and consistently. Concretely: an in-flight custom action had no way of being told to stop on Shutdown()/DisposeAsync() -- only an explicit cancel_pending redelivery could cancel it. It would keep running past the extractor's own lifetime. Two changes, both needed: - The custom-action CancellationTokenSource is now a proper linked child of Source (CancellationTokenSource.CreateLinkedTokenSource), for structural consistency with the rest of this class's token hierarchy. - ShutdownInternal now explicitly cancels every in-flight custom action's token as its first step. Linking to Source alone isn't enough for *prompt* cancellation: Source itself is only cancelled at the very end of Shutdown()/DisposeAsyncCore(), after the task scheduler's own (up to 20s) graceful-wait window -- an in-flight custom action needs to be signalled at the same time as the task scheduler, not after it. Added TestShutdownCancelsInFlightCustomActions, and sabotage-verified it: reverting either half of the fix (the bare CancellationTokenSource, or removing ShutdownInternal's explicit cancel loop) makes the test fail with a timeout waiting for the canceled status, confirming this isn't just passing by construction. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>