feat(odin): Upload fetch_logs files to a data set as a new copy each run (EDG-1037)

CHANGELOG.md

@@ -17,7 +17,10 @@ Changes are grouped as follows
 ### Added
 * In the `unstable` package: New action status `completed_with_errors` in `ActionStatus`, reported when an action finishes but non-fatal (`error`-level) errors were reported during the run.
 
+* In the `unstable` package: New `Extractor.get_log_upload_data_set_id()` hook that tells the built-in `fetch_logs` action which data set to upload log files into. Resolution order: integration "data domains" field (not available yet, marked TODO), then the extractor's own configured data set, then no data set. The default uses `application_config.data_set` when the config derives from `WithDataSetId` and otherwise returns `None`; extractors that keep their data set elsewhere (for example the File and DB extractors) should override it in their own repositories.
+
 ### Changed
+* In the `unstable` package: `fetch_logs` now uploads a new copy of the log files on every invocation instead of overwriting: the file external ID is now `extractor-logs-<integration>-<date>-<action external id>` (the `extractor-logs-<integration>-<date>` prefix is unchanged). The per-file status `skipped` for a date without a log file is now called `missing`, matching the existing `missing_files` counter (all counter keys are unchanged). When no file was uploaded at all the action now ends as `completed_with_errors` with the message "No log files were uploaded ..." instead of `succeeded`. Custom actions can end as `completed_with_errors` with `ActionContext.set_result(..., completed_with_errors=True)`.
 * In the `unstable` package: `Stop <task>` actions now stay `running` until the task has actually ended and only then report `succeeded`. If the task has not ended within 300 seconds (`STOP_TASK_ACTION_TIMEOUT_SECONDS`) the action is reported as `failed` ("Task '<name>' did not stop within 300 seconds"); a waiting Stop is reported as `canceled` on extractor shutdown.
 
 ### Fixed

cognite/extractorutils/unstable/core/_log_upload_action.py

@@ -1,13 +1,23 @@
-"""Built-in ``fetch_logs`` action: streams rotated log files to CDF Files."""
+"""
+Built-in ``fetch_logs`` action: streams rotated log files to CDF Files.
 
+Every invocation uploads a new copy of each file (nothing is overwritten): the CDF external ID is
+``extractor-logs-<integration>-<log date>-<action external id>``. The files go into the data set returned by
+``Extractor.get_log_upload_data_set_id`` (none by default).
+
+Final status: ``failed`` for invalid parameters, ``completed_with_errors`` when no file at all was uploaded
+(everything missing, too large or failed), ``succeeded`` otherwise, including when only some days had a file.
+"""
+
+import hashlib
 import json
 import logging
 from collections import Counter
 from dataclasses import dataclass
 from datetime import date, timedelta, timezone
 from datetime import datetime as dt
 from pathlib import Path
-from typing import BinaryIO, Literal
+from typing import Any, BinaryIO, Literal
 
 from cognite.client import CogniteClient
 
@@ -81,23 +91,23 @@ def _build_candidate_files(
     today: date,
 ) -> tuple[list[LogFileCandidate], list[date]]:
     """
-    Enumerate log files for [start_date, end_date] and partition into candidates vs skipped.
+    Enumerate log files for [start_date, end_date] and partition into candidates vs missing.
 
     Rotated files follow the naming convention ``<base_path>.YYYY-MM-DD``.
     The live file (``base_path``) is used for ``today``; all other dates use the rotated name.
 
-    A file is skipped when it does not exist, is empty, or is inaccessible (any ``OSError``).
+    A file is missing when it does not exist, is empty, or is inaccessible (any ``OSError``).
     This covers the brief rotation race window: if the action is dispatched right after midnight
     before ``TimedRotatingFileHandler`` has renamed ``file.log`` to ``file.log.YYYY-MM-DD``,
-    the rotated file will not be found and that date will appear in ``skipped``. Retrying after
+    the rotated file will not be found and that date will be reported as ``missing``. Retrying after
     rotation completes (within seconds) will pick it up.
 
     Returns:
-        (candidates, skipped): candidates are files that exist and have content;
-        skipped contains dates for which no usable file was found.
+        (candidates, missing): candidates are files that exist and have content;
+        missing contains dates for which no usable file was found.
     """
     candidates: list[LogFileCandidate] = []
-    skipped: list[date] = []
+    missing: list[date] = []
 
     current = start_date
     while current <= end_date:
@@ -111,30 +121,55 @@ def _build_candidate_files(
         try:
             size = path.stat().st_size
         except OSError as e:
-            _logger.warning("fetch_logs: skipping %s for %s — %s", path, current, e)
-            skipped.append(current)
+            _logger.warning("fetch_logs: no usable log file %s for %s — %s", path, current, e)
+            missing.append(current)
         else:
             if size > 0:
                 candidates.append(LogFileCandidate(log_date=current, path=path, is_current=is_current))
             else:
-                skipped.append(current)
+                missing.append(current)
 
         current += timedelta(days=1)
 
-    return candidates, skipped
+    return candidates, missing
+
 
+MAX_EXTERNAL_ID_LENGTH = 255
+"""CDF's limit for file external IDs."""
+
+
+def _file_external_id(integration_external_id: str, log_date: date, run_id: str) -> str:
+    """
+    Build the CDF external ID of one uploaded log file.
 
-def _file_external_id(integration_external_id: str, log_date: date) -> str:
-    return f"extractor-logs-{integration_external_id}-{log_date.isoformat()}"
+    ``run_id`` (the external ID of the action invocation) makes the ID unique per run so repeated runs never
+    overwrite each other, while the ``extractor-logs-<integration>-<date>`` prefix keeps the files findable. IDs
+    are deterministic for a given run. If the result would exceed CDF's limit, ``run_id`` is replaced by a hash of
+    itself and, as a last resort, the integration part is shortened.
+    """
+    suffix = f"{log_date.isoformat()}-{run_id}"
+    external_id = f"extractor-logs-{integration_external_id}-{suffix}"
+    if len(external_id) <= MAX_EXTERNAL_ID_LENGTH:
+        return external_id
+    suffix = f"{log_date.isoformat()}-{hashlib.sha256(run_id.encode('utf-8')).hexdigest()[:16]}"
+    room = MAX_EXTERNAL_ID_LENGTH - len("extractor-logs--") - len(suffix)
+    return f"extractor-logs-{integration_external_id[:room]}-{suffix}"
 
 
 def _upload_candidate(
     candidate: LogFileCandidate,
     integration_external_id: str,
     cdf_client: CogniteClient,
+    *,
+    run_id: str,
+    data_set_id: int | None = None,
 ) -> _FileUploadResult:
-    """Upload one candidate log file to CDF Files. Returns a result regardless of success or failure."""
-    external_id = _file_external_id(integration_external_id, candidate.log_date)
+    """
+    Upload one candidate log file to CDF Files as a new file. Returns a result regardless of success or failure.
+
+    The file is never overwritten: ``run_id`` makes the external ID unique per ``fetch_logs`` invocation.
+    """
+    external_id = _file_external_id(integration_external_id, candidate.log_date, run_id)
 
     try:
         actual_size = candidate.path.stat().st_size
@@ -157,6 +192,8 @@ def _upload_candidate(
             size_bytes=actual_size,
         )
 
+    # Only send data_set_id when there is one, so extractors without a data set upload exactly as before.
+    extra_upload_args: dict[str, Any] = {} if data_set_id is None else {"data_set_id": data_set_id}
     try:
         with open(candidate.path, "rb") as f:
             reader: BinaryIO = BoundedReader(f, actual_size) if candidate.is_current else f  # type: ignore[assignment]
@@ -165,7 +202,7 @@ def _upload_candidate(
                 name=f"{external_id}.log",
                 external_id=external_id,
                 mime_type="text/plain",
-                overwrite=True,
+                **extra_upload_args,
             )
         _logger.info("fetch_logs: uploaded %s (%d bytes)", external_id, actual_size)
         return _FileUploadResult(
@@ -229,27 +266,32 @@ def fetch_logs_action(ctx: ActionContext) -> None:
             error_type="no_file_handler_configured",
         )
 
-    candidates, skipped_dates = _build_candidate_files(log_file_path, start_date, end_date, today)
+    candidates, missing_dates = _build_candidate_files(log_file_path, start_date, end_date, today)
     _logger.info(
-        "fetch_logs: %d candidate file(s) for %s to %s; %d date(s) skipped",
+        "fetch_logs: %d candidate file(s) for %s to %s; %d date(s) missing",
         len(candidates),
         start_date,
         end_date,
-        len(skipped_dates),
+        len(missing_dates),
     )
 
     integration_external_id = ctx.integration_external_id
     cdf_client = ctx.cdf_client
+    data_set_id = ctx.log_upload_data_set_id()
 
     total_candidates = len(candidates)
     upload_results: list[_FileUploadResult] = []
     for i, candidate in enumerate(candidates, 1):
-        upload_results.append(_upload_candidate(candidate, integration_external_id, cdf_client))
+        upload_results.append(
+            _upload_candidate(
+                candidate, integration_external_id, cdf_client, run_id=ctx.external_id, data_set_id=data_set_id
+            )
+        )
         ctx.report_progress(f"Uploading: {i}/{total_candidates} files complete")
 
     counts = Counter(r.status for r in upload_results)
 
-    # Per-file entries: upload results (sorted by date) + missing dates (skipped by candidate builder).
+    # Per-file entries: upload results (sorted by date) + missing dates (no usable file found by the candidate builder).
     files_list: list[dict[str, str]] = []
     for r in upload_results:
         entry: dict[str, str] = {"date": str(r.log_date), "status": r.status}
@@ -260,14 +302,17 @@ def fetch_logs_action(ctx: ActionContext) -> None:
         if r.error:
             entry["error"] = r.error
         files_list.append(entry)
-    files_list.extend({"date": str(skipped_date), "status": "skipped"} for skipped_date in skipped_dates)
+    files_list.extend({"date": str(missing_date), "status": "missing"} for missing_date in missing_dates)
     files_list.sort(key=lambda e: e["date"])
 
-    result_message = f"{counts['uploaded']} of {num_days} log files uploaded to CDF Files"
+    uploaded = counts["uploaded"]
+    result_message = f"{uploaded} of {num_days} log files uploaded to CDF Files"
+    if uploaded == 0:
+        result_message = f"No log files were uploaded ({result_message})"
     metadata: dict[str, str] = {
         "total_files": str(num_days),
         "uploaded_files": str(counts["uploaded"]),
-        "missing_files": str(len(skipped_dates)),
+        "missing_files": str(len(missing_dates)),
         "skipped_too_large_files": str(counts["skipped_too_large"]),
         "failed_files": str(counts["failed"]),
     }
@@ -290,4 +335,4 @@ def fetch_logs_action(ctx: ActionContext) -> None:
         )
         result_message += " (per-file detail omitted from result metadata — see extractor logs for full detail)"
 
-    ctx.set_result(result_message, metadata=metadata)
+    ctx.set_result(result_message, metadata=metadata, completed_with_errors=uploaded == 0)

cognite/extractorutils/unstable/core/actions.py

@@ -50,6 +50,7 @@ def __init__(
         self.cancellation_token = cancellation_token
         self._result_message: str | None = None
         self._result_metadata: dict[str, str] | None = None
+        self._completed_with_errors = False
 
         self._logger = logging.getLogger(f"{self._extractor.EXTERNAL_ID}.action.{self._action.name.replace(' ', '')}")
 
@@ -83,15 +84,30 @@ def _new_error(
             task_name=task_name if task_name is not None else self._action.name,
         )
 
-    def set_result(self, message: str, *, metadata: dict[str, str] | None = None) -> None:
-        """Record the result for a successful action completion."""
+    def log_upload_data_set_id(self) -> int | None:
+        """The data set the ``fetch_logs`` action uploads into, see ``Extractor.get_log_upload_data_set_id``."""
+        return self._extractor.get_log_upload_data_set_id()
+
+    def set_result(
+        self, message: str, *, metadata: dict[str, str] | None = None, completed_with_errors: bool = False
+    ) -> None:
+        """
+        Record the result for an action completion.
+
+        Args:
+            message: The result message.
+            metadata: Optional result metadata.
+            completed_with_errors: Finish the action as ``completed_with_errors`` instead of ``succeeded``, for an
+                action that ran to the end but did not achieve its goal. Ignored if the action was cancelled.
+        """
         if self._result_message is not None:
             raise RuntimeError(
                 f"set_result() has already been called for this action invocation; "
                 f"existing message: {self._result_message!r}"
             )
         self._result_message = message
         self._result_metadata = metadata
+        self._completed_with_errors = completed_with_errors
 
     def report_progress(self, message: str) -> None:
         """Queue an intermediate progress update while the action is still running."""

cognite/extractorutils/unstable/core/base.py

@@ -110,6 +110,7 @@ def my_task_function(self, task_context: TaskContext) -> None:
     ExtractorConfig,
     LogConsoleHandlerConfig,
     LogFileHandlerConfig,
+    WithDataSetId,
     create_state_store,
 )
 from cognite.extractorutils.unstable.core._dto import (
@@ -442,6 +443,28 @@ def __init_actions__(self) -> None:
         """
         pass
 
+    def get_log_upload_data_set_id(self) -> int | None:
+        """
+        Return the id of the CDF data set the built-in ``fetch_logs`` action uploads log files into, or ``None``.
+
+        Resolution order:
+
+        1. TODO(data-domains): the integration's "data domains" field. It does not exist in the Integrations API yet;
+           when it does, resolve it here and prefer it over everything below. Intentionally not implemented.
+        2. The extractor's own configured data set. By default this is ``application_config.data_set`` if the
+           application config derives from ``WithDataSetId``. Extractors whose config keeps the data set elsewhere
+           (for example per destination) should override this method.
+        3. ``None``: the files are uploaded without a data set.
+
+        Called on every ``fetch_logs`` invocation. An exception raised here (for example the data set cannot be
+        found) fails the action.
+        """
+        config = self.application_config
+        if isinstance(config, WithDataSetId):
+            data_set = config.get_data_set(self.cognite_client)
+            return data_set.id if data_set is not None else None
+        return None
+
     def add_action(self, action: CustomAction) -> None:
         """
         Register a custom action that Odin can invoke on this extractor.
@@ -1005,7 +1028,12 @@ def _handle_custom_action(self, action: Action) -> None:
 
             custom.target(ctx)
             filtered_metadata, oversized_fields = drop_oversized_metadata_fields(ctx._result_metadata)
-            completed_status = ActionStatus.canceled if action_token.is_cancelled else ActionStatus.succeeded
+            if action_token.is_cancelled:
+                completed_status = ActionStatus.canceled
+            elif ctx._completed_with_errors:
+                completed_status = ActionStatus.completed_with_errors
+            else:
+                completed_status = ActionStatus.succeeded
             if oversized_fields:
                 # The action itself ran to completion (or was cancelled) — only reporting the full
                 # result back to Odin is affected, because a metadata value is too large to send.
@@ -1014,7 +1042,10 @@ def _handle_custom_action(self, action: Action) -> None:
                 # bundles all pending updates together and requeues the whole batch on any rejection).
                 # Non-oversized fields are still reported, and the action's real outcome (succeeded or
                 # canceled) is preserved rather than being overwritten by this reporting issue.
-                outcome = "was canceled" if completed_status == ActionStatus.canceled else "completed successfully"
+                outcome = {
+                    ActionStatus.canceled: "was canceled",
+                    ActionStatus.completed_with_errors: "completed with errors",
+                }.get(completed_status, "completed successfully")
                 self._checkin_worker.queue_action_update(
                     ActionUpdate(
                         external_id=action.external_id,