logs: fix logs ordering, group by subtask

* closes CERNDocumentServer/cds-rdm#615

Author: jrcastro2

invenio_jobs/assets/semantic-ui/js/invenio_jobs/administration/RunsLogs.js

@@ -22,26 +22,63 @@ import { http, withCancel } from "react-invenio-forms";
 import { DateTime } from "luxon";
 import { i18next } from "@translations/invenio_jobs/i18next";
 
+/**
+ * TaskGroup component - displays logs for a single task group
+ * @param {Object} taskGroup - Task group containing taskId, taskName, parentTaskId, and logs
+ * @param {Object} levelClass - Mapping of log levels to CSS classes
+ */
+const TaskGroup = ({ taskGroup, levelClass }) => (
+  <div className={taskGroup.parentTaskId ? "subtask-container" : ""}>
+    {taskGroup.logs.map((log) => (
+      <div
+        key={`${log.timestamp}-${log.level}-${log.message}`}
+        className={`log-line ${log.level.toLowerCase()}`}
+      >
+        <span className="log-timestamp">[{log.formatted_timestamp}]</span>{" "}
+        <span className={levelClass[log.level] || ""}>{log.level}</span>{" "}
+        <span className="log-message">{log.message}</span>
+      </div>
+    ))}
+  </div>
+);
+
+TaskGroup.propTypes = {
+  taskGroup: PropTypes.shape({
+    taskId: PropTypes.string.isRequired,
+    parentTaskId: PropTypes.string,
+    logs: PropTypes.array.isRequired,
+  }).isRequired,
+  levelClass: PropTypes.object.isRequired,
+};
+
 export class RunsLogs extends Component {
   constructor(props) {
     super(props);
 
     const { logs, run, sort, warnings } = props;
 
+    const formattedLogs = logs.map((log) => ({
+      ...log,
+      formatted_timestamp: DateTime.fromISO(log.timestamp).toFormat(
+        "yyyy-MM-dd HH:mm"
+      ),
+    }));
+
     this.state = {
       error: null,
-      logs: logs.map((log) => ({
-        ...log,
-        formatted_timestamp: DateTime.fromISO(log.timestamp).toFormat(
-          "yyyy-MM-dd HH:mm"
-        ),
-      })),
+      logs: formattedLogs,
       run,
       sort,
       warnings: warnings || [],
       runDuration: this.getDurationInMinutes(run.started_at, run.finished_at),
       formatted_started_at: this.formatDatetime(run.started_at),
     };
+
+    // Cache for memoized log tree
+    this.logTreeCache = {
+      logs: null,
+      tree: null,
+    };
   }
 
   componentDidMount() {
@@ -60,6 +97,18 @@ export class RunsLogs extends Component {
     this.statusFetchCancel?.cancel();
   }
 
+  getLogTree() {
+    const { logs } = this.state;
+    // Return cached tree if logs haven't changed
+    if (this.logTreeCache.logs === logs) {
+      return this.logTreeCache.tree;
+    }
+    // Rebuild tree and update cache
+    const tree = this.buildLogTree(logs);
+    this.logTreeCache = { logs, tree };
+    return tree;
+  }
+
   getDurationInMinutes(startedAt, finishedAt) {
     if (!startedAt) return 0;
     const start = DateTime.fromISO(startedAt);
@@ -71,6 +120,32 @@ export class RunsLogs extends Component {
     return ts ? DateTime.fromISO(ts).toFormat("yyyy-MM-dd HH:mm") : null;
   }
 
+  buildLogTree = (logs) => {
+    /**
+     * Build flat task groups from log list.
+     * Returns array of task groups, each with taskId, parentTaskId, and logs.
+     * Root tasks have parentTaskId == null; subtasks have parentTaskId set.
+     */
+    const taskGroups = {};
+
+    logs.forEach((log) => {
+      const context = log.context || {};
+      const taskId = context.task_id || "unknown";
+
+      if (!taskGroups[taskId]) {
+        taskGroups[taskId] = {
+          taskId,
+          parentTaskId: context.parent_task_id || null,
+          logs: [],
+        };
+      }
+
+      taskGroups[taskId].logs.push(log);
+    });
+
+    return Object.values(taskGroups);
+  };
+
   fetchLogs = async (runId, sort) => {
     try {
       const searchAfterParams = (sort || [])
@@ -153,6 +228,7 @@ export class RunsLogs extends Component {
       formatted_started_at: formattedStartedAt,
       warnings,
     } = this.state;
+    const logTree = this.getLogTree();
     const levelClass = {
       DEBUG: "",
       INFO: "primary",
@@ -254,7 +330,7 @@ export class RunsLogs extends Component {
                     </List.Item>
                   </List>
                 </Grid.Column>
-                <Grid.Column className="log-table" width={13}>
+                <Grid.Column className="job-log-table" width={13}>
                   {/* Display error message for failed jobs */}
                   {(run.status === "FAILED" ||
                     run.status === "PARTIAL_SUCCESS") && (
@@ -278,19 +354,15 @@ export class RunsLogs extends Component {
                     </Message>
                   )}
                   <Segment>
-                    {logs.map((log) => (
-                      <div
-                        key={`${log.timestamp}-${log.level}-${log.message}`}
-                        className={`log-line ${log.level.toLowerCase()}`}
-                      >
-                        <span className="log-timestamp">
-                          [{log.formatted_timestamp}]
-                        </span>{" "}
-                        <span className={levelClass[log.level] || ""}>
-                          {log.level}
-                        </span>{" "}
-                        <span className="log-message">{log.message}</span>
-                      </div>
+                    {logTree.map((taskGroup) => (
+                      <React.Fragment key={taskGroup.taskId}>
+                        {!taskGroup.parentTaskId &&
+                          logTree.indexOf(taskGroup) > 0 && <Divider />}
+                        <TaskGroup
+                          taskGroup={taskGroup}
+                          levelClass={levelClass}
+                        />
+                      </React.Fragment>
                     ))}
                   </Segment>
                 </Grid.Column>

invenio_jobs/logging/celery_signals.py

@@ -23,7 +23,7 @@ def capture_context(sender=None, headers=None, body=None, **kwargs):
     if (
         "context" not in headers and job_context.get() is not EMPTY_JOB_CTX
     ):  # Ensure context is only added if missing
-        headers["context"] = job_context.get()
+        headers["context"] = dict(job_context.get())
 
 
 # Restore context when a task starts executing
@@ -35,6 +35,12 @@ def restore_context(task=None, **kwargs):
     if job_context.get() is EMPTY_JOB_CTX:
         task_context = getattr(task.request, "context", None)
         if task_context:
+            # Update context with current task's metadata from Celery's own request
+            task_context = dict(task_context)
+            task_context["task_id"] = str(task.request.id)
+            task_context["parent_task_id"] = (
+                str(task.request.parent_id) if task.request.parent_id else None
+            )
             token = job_context.set(task_context)
             # Store token in task.request
             task.request._job_context_token = token

invenio_jobs/logging/index_templates/os-v1/job-logs-v1.0.0.json

@@ -15,7 +15,9 @@
           "properties": {
             "job_id": { "type": "keyword" },
             "run_id": { "type": "keyword" },
-            "identity_id": { "type": "keyword" }
+            "identity_id": { "type": "keyword" },
+            "task_id": { "type": "keyword" },
+            "parent_task_id": { "type": "keyword" }
           },
           "dynamic": true
         }

invenio_jobs/logging/index_templates/os-v2/job-logs-v1.0.0.json

@@ -15,7 +15,9 @@
           "properties": {
             "job_id": { "type": "keyword" },
             "run_id": { "type": "keyword" },
-            "identity_id": { "type": "keyword" }
+            "identity_id": { "type": "keyword" },
+            "task_id": { "type": "keyword" },
+            "parent_task_id": { "type": "keyword" }
           },
           "dynamic": true
         }

invenio_jobs/logging/jobs.py

@@ -44,14 +44,16 @@ def emit(self, record):
 
     def enrich_log(self, record):
         """Enrich log record with contextvars' global context."""
+        context = dict(job_context.get())
+
         log_data = {
             "timestamp": datetime.now().isoformat(),
             "level": record.levelname,
             "message": record.getMessage(),
             "module": record.module,
             "function": record.funcName,
             "line": record.lineno,
-            "context": job_context.get(),
+            "context": context,
         }
         serialized_data = JobLogEntrySchema().load(log_data)
         return serialized_data

invenio_jobs/services/schema.py

@@ -389,6 +389,8 @@ class LogContextSchema(Schema):
     job_id = fields.Str(required=True)
     run_id = fields.Str(required=True)
     identity_id = fields.Str(required=True)
+    task_id = fields.Str(allow_none=True)
+    parent_task_id = fields.Str(allow_none=True)
 
 
 class JobLogEntrySchema(Schema):

invenio_jobs/services/services.py

@@ -556,6 +556,18 @@ def search(self, identity, params):
         # Track if we're truncating results
         truncated = total > max_docs
 
+        # Sort chronologically by timestamp and _id
+        # Note: Hierarchical grouping by task hierarchy is handled client-side
+        # in the UI via buildLogTree() function
+        # This is not done in the backend by updating the sort parameters due
+        # to 2 main reasons:
+        # 1 - For dynamically loading the logs while the jobs is being executed
+        # we use the search_after parameter that will return only new values,
+        # meaning that they cant be properly grouped in the backend without the
+        # risk of being splitted by this.
+        # 2 - By changing the sort order from timestamp to prioritize the
+        # task_id, the logs will loose the chronological order. There is no easy
+        # way to group by task_id and prioritize by timestamp.
         search = search.sort("@timestamp", "_id").extra(size=batch_size)
         if search_after:
             search = search.extra(search_after=search_after)

invenio_jobs/tasks.py

@@ -68,6 +68,10 @@ def execute_run(self, run_id, identity_id, kwargs=None):
             "run_id": str(run_id),
             "job_id": str(run.job.id),
             "identity_id": str(identity_id),
+            "task_id": str(self.request.id),
+            "parent_task_id": (
+                str(self.request.parent_id) if self.request.parent_id else None
+            ),
         }
     ):
         update_run(

tests/test_services_job_logs.py

@@ -6,8 +6,11 @@
 # Invenio-Jobs is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
+from datetime import datetime, timezone
+
 import pytest
 
+from invenio_jobs.api import AttrDict
 from invenio_jobs.proxies import current_jobs_logs_service
 
 
@@ -103,3 +106,74 @@ def fake_search(self, *args, **kwargs):
     assert "warnings" not in payload
 
     assert payload["hits"]["sort"] == hits[2].meta.sort  # Last hit (log-1)
+
+
+def _make_hit_with_lineage(idx, task_id, parent_task_id=None):
+    """Create a fake search hit with task lineage metadata."""
+    timestamp = datetime(2025, 1, 1, tzinfo=timezone.utc).timestamp() + idx
+    sort_value = [
+        task_id,
+        idx,
+        timestamp,
+        f"id-{idx}",
+    ]
+    hit = AttrDict(
+        {
+            "@timestamp": datetime.fromtimestamp(timestamp, timezone.utc).isoformat(),
+            "level": "INFO",
+            "message": f"log-{idx}",
+            "module": "tests",
+            "function": "fn",
+            "line": idx,
+            "context": {
+                "job_id": "job-123",
+                "run_id": "run-456",
+                "identity_id": "user-789",
+                "task_id": task_id,
+                "parent_task_id": parent_task_id,
+            },
+            "sort": sort_value,
+        }
+    )
+    hit.meta = AttrDict({"sort": sort_value})
+    return hit
+
+
+@pytest.mark.usefixtures("app")
+def test_job_logs_search_with_task_lineage(monkeypatch, anon_identity, app, FakeSearch):
+    """Service returns logs with task lineage metadata."""
+    service = current_jobs_logs_service
+
+    original_max = app.config.get("JOBS_LOGS_MAX_RESULTS")
+    original_batch = app.config.get("JOBS_LOGS_BATCH_SIZE")
+    app.config["JOBS_LOGS_MAX_RESULTS"] = 10
+    app.config["JOBS_LOGS_BATCH_SIZE"] = 10
+
+    # Create hits with task lineage
+    hits = [
+        _make_hit_with_lineage(1, "task-A"),
+        _make_hit_with_lineage(2, "task-A"),
+        _make_hit_with_lineage(1, "task-B", "task-A"),
+        _make_hit_with_lineage(2, "task-B", "task-A"),
+    ]
+
+    def fake_search(self, *args, **kwargs):
+        return FakeSearch(hits)
+
+    monkeypatch.setattr(service.__class__, "_search", fake_search)
+
+    payload = None
+    try:
+        with app.app_context():
+            result = service.search(anon_identity, {"q": "test"})
+            payload = result.to_dict()
+    finally:
+        app.config["JOBS_LOGS_MAX_RESULTS"] = original_max
+        app.config["JOBS_LOGS_BATCH_SIZE"] = original_batch
+
+    assert payload["hits"]["total"] == len(hits)
+    assert len(payload["hits"]["hits"]) == len(hits)
+    # Verify task lineage fields are present
+    first_hit = payload["hits"]["hits"][0]
+    assert "task_id" in first_hit["context"]
+    assert "parent_task_id" in first_hit["context"]