feat: Logging license type usage

src/dvsim/logging.py

@@ -64,7 +64,7 @@ def __init__(self, name: str) -> None:
 
     def verbose(self, msg: object, *args: object) -> None:
         """Log a verbose msg."""
-        self.log(self.VERBOSE, msg, *args)
+        self.log(self.VERBOSE, msg, *args, stacklevel=2)
 
     def set_logfile(
         self,

src/dvsim/scheduler/resources.py

@@ -10,6 +10,7 @@
 from typing import Protocol
 
 from dvsim.job.data import JobSpec, ResourceMapping
+from dvsim.job.status import JobStatus
 from dvsim.logging import log
 
 __all__ = (
@@ -19,6 +20,10 @@
 )
 
 
+# Used to format logs
+MIN_FIELD_WIDTH = 5
+
+
 class ResourceProvider(Protocol):
     """An abstraction of something that provides resource availability info to the scheduler."""
 
@@ -73,6 +78,13 @@ def __init__(
         self._provider = provider
         self._missing_policy = missing_policy
         self._usage = defaultdict(int)
+        # Index tracking number of jobs with each status, per resource
+        self._status_counts: defaultdict[str, dict[JobStatus, int]] = defaultdict(
+            lambda: defaultdict(int)
+        )
+        # The number of characters used to represent the largest field in the displayed table
+        self._int_field_width: int
+        self._str_field_width: int
 
     async def can_allocate(self, request: ResourceMapping) -> bool:
         """Check if a given resource request can be allocated, given current usage and limits."""
@@ -109,6 +121,29 @@ def release(self, request: ResourceMapping) -> None:
             if amount is not None:
                 self._usage[resource] -= amount
 
+    def update_status_counts(self, spec: JobSpec, old: JobStatus, new: JobStatus) -> None:
+        """Update the index that tracks job status counts per resource."""
+        for resource in spec.resources or {}:
+            if status_counts := self._status_counts.get(resource):
+                status_counts[old] = status_counts[old] - 1
+                status_counts[new] = status_counts[new] + 1
+            else:
+                log.warning("Unrecognised resource: %s", resource)
+
+    def log_status_counts(self) -> None:
+        """Log job status counts per resource.
+
+        This should be registered as a callback on job status change in Scheduler only when verbose
+        logging is enabled.
+        """
+        for resource, status_counts in self._status_counts.items():
+            status_count_str = ", ".join(
+                f"{status.value[0]}: {count:>{self._int_field_width}}"
+                for status, count in status_counts.items()
+            )
+            log_msg = f"[ Resource: %-{self._str_field_width}s ] [ %s ]"
+            log.verbose(log_msg, resource, status_count_str)
+
     def _log_usage(self, capacity: ResourceMapping, used: ResourceMapping) -> None:
         """Debug log individual job resource usage aggregates."""
         if log.isEnabledFor(log.DEBUG):
@@ -201,3 +236,22 @@ async def validate_jobs(self, jobs: Iterable[JobSpec]) -> None:
 
         self._log_usage(capacity, aggregate)
         self._emit_validation_errors(missing_resource_errors, limit_exceeded_errors)
+
+    def init_status_counts(self, jobs: Iterable[JobSpec]) -> None:
+        """Initialise an index tracking the number of jobs with each status, per resource."""
+        for job in jobs:
+            for resource in job.resources or {}:
+                if not self._status_counts.get(resource):
+                    self._status_counts[resource] = dict.fromkeys(JobStatus, 0)
+                self._status_counts[resource][JobStatus.SCHEDULED] += 1
+
+        self._int_field_width = max(
+            (
+                len(str(job_status_counts[JobStatus.SCHEDULED]))
+                for job_status_counts in self._status_counts.values()
+            ),
+            default=MIN_FIELD_WIDTH,
+        )
+        self._str_field_width = max(
+            (len(resource) for resource in self._status_counts), default=MIN_FIELD_WIDTH
+        )

src/dvsim/scheduler/runner.py

@@ -8,6 +8,7 @@
 
 import dvsim.instrumentation.runtime as instrumentation
 from dvsim.job.data import CompletedJobStatus, JobSpec
+from dvsim.logging import log
 from dvsim.runtime.backend import RuntimeBackend
 from dvsim.runtime.fake import FakePolicy, FakeRuntimeBackend
 from dvsim.runtime.registry import backend_registry
@@ -95,6 +96,9 @@ async def run_scheduler(
     # Convert to list so that first use doesn't consume the Iterable
     jobs = list(jobs)
 
+    if log.isEnabledFor(log.VERBOSE) and resource_manager:
+        resource_manager.init_status_counts(jobs)
+
     scheduler = Scheduler(
         jobs=jobs,
         backends={backend.name: backend},
@@ -112,6 +116,14 @@ async def run_scheduler(
         ),
     )
 
+    # Setup additional verbose logging
+    if log.isEnabledFor(log.VERBOSE) and resource_manager:
+        scheduler.add_job_status_change_callback(
+            lambda spec, old, new, resources=resource_manager: (
+                resources.update_status_counts(spec, old, new) or resources.log_status_counts()
+            )
+        )
+
     if not interactive:
         status_printer = create_status_printer(jobs)
 

tests/test_scheduler.py

@@ -4,8 +4,10 @@
 
 """Test the DVSim scheduler."""
 
+import logging
 import multiprocessing
 import os
+import random
 import sys
 import threading
 import time
@@ -22,10 +24,13 @@
 from dvsim.job.data import CompletedJobStatus, JobSpec, WorkspaceConfig
 from dvsim.job.status import JobStatus
 from dvsim.launcher.base import ErrorMessage, Launcher, LauncherBusyError, LauncherError
+from dvsim.logging import _VERBOSE_LOG_LEVEL
 from dvsim.report.data import IPMeta, ToolMeta
 from dvsim.runtime.legacy import LegacyLauncherAdapter
 from dvsim.scheduler.core import Scheduler
 from dvsim.scheduler.resources import ResourceManager, StaticResourceProvider
+from dvsim.scheduler.runner import run_scheduler
+from dvsim.tool.utils import _SUPPORTED_SIM_TOOLS
 
 __all__ = ()
 
@@ -34,6 +39,10 @@
 DEFAULT_TIMEOUT = 2
 SIGNAL_TEST_TIMEOUT = 5
 
+# Used for randomly selected simulation tool names
+SEED = 42
+random.seed(SEED)
+
 
 @dataclass
 class MockJob:
@@ -255,9 +264,14 @@ def ip_meta_factory(**overrides: str | None) -> IPMeta:
     return IPMeta(**meta)
 
 
-def tool_meta_factory(name: str = "test_tool", version: str = "test_version") -> ToolMeta:
+def tool_meta_factory(name: str = "", version: str = "test_version") -> ToolMeta:
     """Create a ToolMeta from a set of default values, for use in testing."""
-    return ToolMeta(name=name, version=version)
+    if name:
+        return ToolMeta(name=name, version=version)
+    return ToolMeta(
+        name=random.choice([*_SUPPORTED_SIM_TOOLS]),  # noqa: S311
+        version=version,
+    )
 
 
 def build_workspace(
@@ -1037,3 +1051,65 @@ def test_signal_kill(tmp_path: Path, *, sig: int, repeat: bool, long_poll: bool)
             proc.join()
             pytest.fail("Scheduler hung and was terminated")
         assert_that(proc.exitcode, equal_to(0))
+
+
+class TestLogging:
+    """Tests for the logging functionality of the scheduler."""
+
+    @staticmethod
+    @pytest.mark.asyncio
+    @pytest.mark.timeout(DEFAULT_TIMEOUT)
+    async def test_resource_logging(fxt: Fxt, caplog: pytest.LogCaptureFixture) -> None:
+        """Test logging of job status counts per tool.
+
+        This test modifies TestSchedulingPriority::test_blocked_weight_starvation
+        """
+        start_job = job_spec_factory(fxt.tmp_path, name="start")
+        short_blocker = job_spec_factory(fxt.tmp_path, name="short", dependencies=["start"])
+        long_blocker = job_spec_factory(fxt.tmp_path, name="long", dependencies=["start"])
+        high = job_spec_factory(fxt.tmp_path, name="high", dependencies=["long"], weight=1000000)
+        jobs = [start_job, short_blocker, long_blocker, high]
+        jobs += make_many_jobs(
+            fxt.tmp_path,
+            n=5,
+            weight=1,
+            dependencies=["short"],
+            vary_targets=True,
+        )
+        # Status counts are tracked per resource, so have each job use its tool as a resource
+        jobs = [job.model_copy(update={"resources": {job.tool.name: 1}}) for job in jobs]
+        fxt.mock_ctx.set_config(
+            short_blocker,
+            MockJob(status_thresholds=[(0, JobStatus.RUNNING), (1, JobStatus.PASSED)]),
+        )
+        fxt.mock_ctx.set_config(
+            long_blocker,
+            MockJob(status_thresholds=[(0, JobStatus.RUNNING), (5, JobStatus.PASSED)]),
+        )
+
+        # Capture logging and set debug level
+        dvsim_logger = logging.getLogger("dvsim")
+        dvsim_logger.addHandler(caplog.handler)
+        caplog.set_level(_VERBOSE_LOG_LEVEL, logger="dvsim")
+        try:
+            result = await run_scheduler(
+                jobs=jobs,
+                max_parallel=len(jobs),
+                interactive=True,
+                backend=fxt.mock_legacy_backend,
+                resource_manager=ResourceManager(StaticResourceProvider({"A": 5, "B": 10})),
+            )
+        finally:
+            dvsim_logger.removeHandler(caplog.handler)
+
+        _assert_result_status(result, len(jobs))
+
+        expected_tools = {job.tool.name for job in jobs}
+        logged_tools = set()
+        for record in caplog.records:
+            if record.module == "resources" and record.levelno == _VERBOSE_LOG_LEVEL:
+                args = record.args
+                if isinstance(args, tuple) and args:
+                    logged_tools.add(str(args[0]).strip())
+
+        assert_that(logged_tools, equal_to(expected_tools))