OSIDB-4738: Add context to async process in history.

osidb/mixins.py

@@ -72,12 +72,14 @@ def save(self, *args, auto_timestamps=True, **kwargs):
             # updated_dt should never change from the DB version
             # otherwise assume that there was a conflicting parallel change
             if db_self is not None and db_self.updated_dt != self.updated_dt:
-                raise DataInconsistencyException(
-                    "Save operation based on an outdated model instance: "
-                    f"Updated datetime in the request {self.updated_dt} "
-                    f"differes from the DB {db_self.updated_dt}. "
-                    "You need to refresh."
+                actual_event_user = self._get_actual_user()
+
+                message = (
+                    f"Save operation based on an outdated model instance by "
+                    f"{actual_event_user} with updated_dt {self.updated_dt} "
+                    f"differs from the DB {db_self.updated_dt} check in history for details."
                 )
+                raise DataInconsistencyException(f"{message} You need to refresh.")
 
             # auto-set updated_dt as now on any change
             # cut off the microseconds to allow mid-air
@@ -86,6 +88,13 @@ def save(self, *args, auto_timestamps=True, **kwargs):
 
         super().save(*args, **kwargs)
 
+    def _get_actual_user(self):
+        ctx = getattr(
+            pghistory.runtime._tracker, "value", None
+        )  # active context for this thread, if any
+        actual_user = (ctx.metadata.get("user") if ctx else None) or "unknown"
+        return actual_user
+
 
 class TrackingMixinManager(models.Manager):
     """

osidb/sync_manager.py

@@ -1,7 +1,9 @@
 import json
+from contextlib import contextmanager
 from datetime import timedelta
-from typing import Optional, Type
+from typing import Any, Iterator, Optional, Type
 
+import pghistory
 from celery.exceptions import Ignore
 from celery.utils.log import get_task_logger
 from django.conf import settings
@@ -15,6 +17,22 @@
 logger = get_task_logger(__name__)
 
 
+@contextmanager
+def pghistory_context(
+    action: str,
+    celery_task_id: str,
+    *,
+    source: str = "celery",
+    user: str = "celery_task",
+    **extra_context: Any,
+) -> Iterator[None]:
+    ctx = {"source": source, "user": user, "action": action, **extra_context}
+    if celery_task_id:
+        ctx["celery_task_id"] = celery_task_id
+    with pghistory.context(**ctx):
+        yield
+
+
 class SyncManager(models.Model):
     """
     Model to handle synchronization of some OSIDB data with external system like Bugzilla
@@ -486,7 +504,7 @@ def __str__(self):
         return result
 
     @staticmethod
-    def link_tracker_with_affects(tracker_id):
+    def link_tracker_with_affects(tracker_id, task="UNKNOWN"):
         # Code adapted from collectors.bzimport.convertors.BugzillaTrackerConvertor.affects
 
         from osidb.models import Affect, Flaw, Tracker
@@ -561,9 +579,13 @@ def link_tracker_with_affects(tracker_id):
         affects = list(set(affects))
 
         with transaction.atomic():
-            tracker.affects.clear()
-            tracker.affects.add(*affects)
-            tracker.save(raise_validation_error=False, auto_timestamps=False)
+            with pghistory_context(
+                action="link_tracker_with_affects",
+                celery_task_id=getattr(getattr(task, "request", None), "id", None),
+            ):
+                tracker.affects.clear()
+                tracker.affects.add(*affects)
+                tracker.save(raise_validation_error=False, auto_timestamps=False)
 
         return affects, failed_flaws, failed_affects
 
@@ -580,7 +602,9 @@ def sync_task(task, tracker_id, **kwargs):
         collector = collectors.BugzillaTrackerCollector()
         try:
             collector.sync_tracker(tracker_id)
-            result = BZTrackerDownloadManager.link_tracker_with_affects(tracker_id)
+            result = BZTrackerDownloadManager.link_tracker_with_affects(
+                tracker_id, task
+            )
             # Handle link failures
             affects, failed_flaws, failed_affects = result
             if failed_flaws:
@@ -819,8 +843,12 @@ def sync_task(task, flaw_id, **kwargs):
         set_user_acls(settings.ALL_GROUPS)
 
         try:
-            flaw = Flaw.objects.get(uuid=flaw_id)
-            flaw._create_or_update_task()
+            with pghistory_context(
+                action="jira_task_sync",
+                celery_task_id=getattr(getattr(task, "request", None), "id", None),
+            ):
+                flaw = Flaw.objects.get(uuid=flaw_id)
+                flaw._create_or_update_task()
 
         except Exception as e:
             JiraTaskSyncManager.failed(flaw_id, e)
@@ -869,8 +897,12 @@ def sync_task(task, flaw_id, **kwargs):
         set_user_acls(settings.ALL_GROUPS)
 
         try:
-            flaw = Flaw.objects.get(uuid=flaw_id)
-            flaw._transition_task()
+            with pghistory_context(
+                action="jira_task_transition",
+                celery_task_id=getattr(getattr(task, "request", None), "id", None),
+            ):
+                flaw = Flaw.objects.get(uuid=flaw_id)
+                flaw._transition_task()
 
         except Exception as e:
             JiraTaskTransitionManager.failed(flaw_id, e)
@@ -990,7 +1022,11 @@ def sync_task(task, tracker_id, **kwargs):
             tracker_data = JiraQuerier().get_issue(tracker_id)
             tracker = JiraTrackerConvertor(tracker_data).tracker
             if tracker:
-                tracker.save()
+                with pghistory_context(
+                    action="jira_tracker_download",
+                    celery_task_id=getattr(getattr(task, "request", None), "id", None),
+                ):
+                    tracker.save()
 
             # Link this Jira tracker to OSIDB Affects as part of the sync process.
             result = JiraTrackerDownloadManager.link_tracker_with_affects(tracker_id)