Mark OCI charts as succeed when OCI is disabled.

Add ability to track hanging background tasks.

Author: allenporter

flux_local/helm_controller/controller.py

@@ -91,7 +91,8 @@ def _added_listener(self, resource_id: NamedResource, obj: BaseManifest) -> None
         if resource_id.kind == "HelmRelease":
             self._tasks.append(
                 self._task_service.create_task(
-                    self.on_helm_release_added(resource_id, obj)
+                    self.on_helm_release_added(resource_id, obj),
+                    f"helm-controller-{resource_id}",
                 )
             )
 
@@ -386,9 +387,7 @@ async def _resolve_helm_chart(self, helm_release: HelmRelease) -> None:
             # Create a new chart from the HelmChartSource instead of mutating the existing one
             helm_release.chart = HelmChart.from_helm_chart_source(helm_chart)
         else:
-            _LOGGER.debug(
-                "HelmChart %s has no version specified", chart_resource
-            )
+            _LOGGER.debug("HelmChart %s has no version specified", chart_resource)
 
     async def _apply(self, manifests: list[dict[str, Any]]) -> None:
         """Apply the manifests to the cluster."""

flux_local/image.py

@@ -25,9 +25,9 @@
     "Alertmanager",  # monitoring.coreos.com/v1
     "Prometheus",  # monitoring.coreos.com/v1
     "AutoscalingRunnerSet",  # actions.github.com/v1alpha1
-    "EnvoyProxy", # gateway.envoyproxy.io/v1alpha1
-    "Plan", # upgrade.cattle.io/v1
-    "Grafana", # grafana.integreatly.org/v1beta1
+    "EnvoyProxy",  # gateway.envoyproxy.io/v1alpha1
+    "Plan",  # upgrade.cattle.io/v1
+    "Grafana",  # grafana.integreatly.org/v1beta1
 ]
 
 # Default image key for most object types.
@@ -45,7 +45,9 @@ def _extract_images(kind: str, doc: dict[str, Any]) -> set[str]:
     for key, value in doc.items():
         if key == image_key:
             if isinstance(value, dict) and "reference" in value:
-                value = value.get("reference")  # https://kubernetes.io/blog/2024/08/16/kubernetes-1-31-image-volume-source/
+                value = value.get(
+                    "reference"
+                )  # https://kubernetes.io/blog/2024/08/16/kubernetes-1-31-image-volume-source/
             if not isinstance(value, str):
                 raise ValueError(
                     f"Expected string for image key '{image_key}', got type {type(value).__name__}: {value}"

flux_local/kustomize_controller/controller.py

@@ -118,7 +118,10 @@ async def _watch_kustomizations(self) -> None:
                 )
                 continue
             self._tasks.append(
-                self._task_service.create_task(self.reconcile(resource_id, obj))
+                self._task_service.create_task(
+                    self.reconcile(resource_id, obj),
+                    f"kustomize-controller-{resource_id}",
+                )
             )
         _LOGGER.info("Stopped watching for Kustomization objects")
 

flux_local/orchestrator/orchestrator.py

@@ -110,6 +110,7 @@ def __init__(
         self.store = store
         self.config = config or OrchestratorConfig()
         self.controllers: dict[str, Any] = {}
+        self._tasks: list[asyncio.Task[None]] = []
 
     def _create_controllers(self) -> None:
         """Create and initialize all controllers."""
@@ -141,8 +142,12 @@ async def start(self) -> None:
         if self.controllers:
             return
 
-        _LOGGER.info("Starting orchestrator")
         self._create_controllers()
+        self._tasks.append(
+            get_task_service().create_background_task(
+                self._log_active_tasks(), "log-active-tasks"
+            )
+        )
 
     async def stop(self) -> None:
         """Stop the orchestrator and all controllers."""
@@ -152,9 +157,15 @@ async def stop(self) -> None:
         _LOGGER.info("Stopping orchestrator")
 
         # Stop controllers in reverse order
+        close_tasks = []
         for name, controller in reversed(self.controllers.items()):
             _LOGGER.debug("Stopping controller: %s", name)
-            await controller.close()
+            close_tasks.append(controller.close())
+        await asyncio.gather(*close_tasks, return_exceptions=True)
+
+        for task in self._tasks:
+            task.cancel()
+        await asyncio.gather(*self._tasks, return_exceptions=True)
 
         # Wait for all tasks to complete
         await get_task_service().block_till_done()
@@ -332,3 +343,14 @@ async def run(self) -> None:
             raise FluxException(f"Uncaught error in orchestrator: {err}") from err
         finally:
             await self.stop()
+
+    async def _log_active_tasks(self) -> None:
+        """Log active tasks periodically."""
+        task_service = get_task_service()
+        while True:
+            await asyncio.sleep(5)
+            active_tasks = task_service.get_active_tasks()
+            if active_tasks:
+                _LOGGER.debug("Active tasks (%d):", len(active_tasks))
+                for task in active_tasks:
+                    _LOGGER.debug("  %s", task.get_name())

flux_local/source_controller/controller.py

@@ -90,7 +90,8 @@ def listener(resource_id: NamedResource, obj: BaseManifest) -> None:
             if resource_id.kind in self.SUPPORTED_KINDS:
                 self._tasks.append(
                     self._task_service.create_task(
-                        self.on_source_added(resource_id, obj)
+                        self.on_source_added(resource_id, obj),
+                        f"source-controller-{resource_id}",
                     )
                 )
 
@@ -131,6 +132,9 @@ async def on_source_added(
             return
         if isinstance(obj, OCIRepository) and not self._config.enable_oci:
             _LOGGER.info("OCI support is disabled, skipping %s", resource_id)
+            self._store.update_status(
+                resource_id, Status.READY, "OCI support is disabled"
+            )
             return
         if isinstance(obj, GitRepository) and self._store.get_artifact(
             resource_id, GitArtifact

flux_local/store/store.py

@@ -16,15 +16,17 @@
 V = TypeVar("V", bound=BaseManifest | StatusInfo | Artifact)
 
 
-SUPPORTS_STATUS: set[str] = set({
+SUPPORTS_STATUS: set[str] = set(
+    {
         "Kustomization",
         "GitRepository",
         "HelmRelease",
         "HelmRepository",
         "Bucket",
         "OCIRepository",
         "ImageRepository",
-})
+    }
+)
 
 
 class StoreEvent(str, Enum):

flux_local/store/watcher.py

@@ -258,7 +258,8 @@ async def watch(self) -> AsyncGenerator[DependencyResolutionEvent, None]:
         for dep_id in list(self._dependencies.keys()):
             if self._dependencies[dep_id] is None:
                 self._dependencies[dep_id] = self._task_service.create_task(
-                    self._watch_single_dependency(dep_id)
+                    self._watch_single_dependency(dep_id),
+                    f"dependency-watcher-{dep_id}",
                 )
 
         active_watches = len(self._dependencies)
@@ -292,7 +293,9 @@ async def watch(self) -> AsyncGenerator[DependencyResolutionEvent, None]:
                         "DependencyWaiter.watch for %s cancelled. Cleaning up.",
                         self._parent_resource_id,
                     )
-                    await self.cancel_pending_watches()  # Ensure all sub-tasks are cancelled
+                    await (
+                        self.cancel_pending_watches()
+                    )  # Ensure all sub-tasks are cancelled
                     # Yield any remaining events that might have been queued before cancellation
                     while not self._event_queue.empty():
                         try:

flux_local/task/service.py

@@ -67,6 +67,10 @@ async def wait_for_task(self, task: asyncio.Task[None]) -> None:
     def get_num_active_tasks(self) -> int:
         """Get the number of active non-background tasks."""
 
+    @abstractmethod
+    def get_active_tasks(self) -> list[asyncio.Task[Any]]:
+        """Get the active non-background tasks."""
+
 
 class TaskServiceImpl(TaskService):
     """Service for tracking and waiting for asynchronous tasks.
@@ -107,6 +111,7 @@ def create_background_task(
             The created task
         """
         task = asyncio.create_task(coro, name=name)
+        _LOGGER.debug("Created background task: %s", name)
         self._background_tasks.add(task)
         task.add_done_callback(partial(self._task_done, self._background_tasks))
         return task
@@ -170,3 +175,7 @@ async def wait_for_task(self, task: asyncio.Task[None]) -> None:
     def get_num_active_tasks(self) -> int:
         """Get the number of active tasks."""
         return len(self._active_tasks)
+
+    def get_active_tasks(self) -> list[asyncio.Task[Any]]:
+        """Get the active tasks."""
+        return list(self._active_tasks)

flux_local/tool/shell/action.py

@@ -21,7 +21,8 @@ class ShellAction:
 
     @classmethod
     def register(
-        cls, subparsers: SubParsersAction  # type: ignore[type-arg]
+        cls,
+        subparsers: SubParsersAction,  # type: ignore[type-arg]
     ) -> ArgumentParser:
         """Register the shell subcommand."""
         parser = cast(
@@ -54,7 +55,9 @@ async def run(self, **kwargs: Any) -> None:
 
         try:
             # Bootstrap the system with the specified path
-            task = asyncio.create_task(orchestrator.bootstrap(bootstrap_options))
+            task = asyncio.create_task(
+                orchestrator.bootstrap(bootstrap_options), name="bootstrap"
+            )
             _LOGGER.debug("Orchestrator started, bootstrapping with path: %s", path)
 
             # Create the shell with the bootstrapped store

flux_local/values.py

@@ -199,7 +199,6 @@ def _update_helmrelease_values(
     """Expand value references in the HelmRelease."""
 
     if ref.target_path:
-
         raw_parts = re.split(r"(?<!\\)\.", ref.target_path)
         parts = [re.sub(r"\\(.)", r"\1", raw_part) for raw_part in raw_parts]