Repository: MatterMiners/lapis

Issue:
Improve logging scheme
The logging scheme should be improved, including

* [ ] generation of log messages from the job itself
* [ ] parameters being logged to be put on next lines as mentioned in #23 

Discussion:

Patch:
diff --git a/lapis/drone.py b/lapis/drone.py
index d217993..9b33dde 100644
--- a/lapis/drone.py
+++ b/lapis/drone.py
@@ -1,5 +1,3 @@
-import logging
-
 from cobald import interfaces
 from usim import time, Scope, instant, Capacities, ResourcesUnavailable
 
@@ -55,7 +53,7 @@ class Drone(interfaces.Pool):
         await (time + self.scheduling_duration)
         self._supply = 1
         self.scheduler.register_drone(self)
-        await sampling_required.set(True)
+        await sampling_required.put(self)
 
     @property
     def supply(self) -> float:
@@ -94,9 +92,8 @@ class Drone(interfaces.Pool):
         from lapis.monitor import sampling_required
         self._supply = 0
         self.scheduler.unregister_drone(self)
-        await sampling_required.set(True)
+        await sampling_required.put(self)  # TODO: introduce state of drone
         await (time + 1)
-        # print("[drone %s] has been shut down" % self)
 
     async def start_job(self, job: Job, kill: bool = False):
         """
@@ -109,17 +106,17 @@ class Drone(interfaces.Pool):
                      requested resources
         :return:
         """
+        job.drone = self
         async with Scope() as scope:
             from lapis.monitor import sampling_required
             self._utilisation = self._allocation = None
 
             job_execution = scope.do(job.run())
-            await instant  # waiting just a moment to enable job to set parameters
+            self.jobs += 1
             try:
                 async with self.resources.claim(**job.resources), \
                         self.used_resources.claim(**job.used_resources):
-                    self.jobs += 1
-                    await sampling_required.set(True)
+                    await sampling_required.put(self)
                     if kill:
                         for resource_key in job.resources:
                             try:
@@ -132,30 +129,15 @@ class Drone(interfaces.Pool):
                     self.scheduler.update_drone(self)
                     await job_execution.done
             except ResourcesUnavailable:
+                await instant
                 job_execution.cancel()
             except AssertionError:
+                await instant
                 job_execution.cancel()
-            else:
-                self.jobs -= 1
-
-            if not job.successful:
-                for resource_key in job.resources:
-                    usage = job.used_resources.get(
-                        resource_key,
-                        job.resources.get(resource_key, None),
-                    )
-                    value = usage / job.resources.get(
-                        resource_key, self.pool_resources[resource_key]
-                    )
-                    if value > 1:
-                        logging.info("job_status", {
-                            "job_exceeds_%s" % resource_key: {
-                                repr(job): value
-                            }
-                        })
+            self.jobs -= 1
             self._utilisation = self._allocation = None
             self.scheduler.update_drone(self)
-            await sampling_required.set(True)
+            await sampling_required.put(self)
 
     def __repr__(self):
         return '<%s: %s>' % (self.__class__.__name__, id(self))
diff --git a/lapis/job.py b/lapis/job.py
index 758fd0c..096f167 100644
--- a/lapis/job.py
+++ b/lapis/job.py
@@ -1,17 +1,22 @@
 import logging
+from typing import Optional, TYPE_CHECKING
 
 from usim import time
-from usim import TaskCancelled
+from usim import CancelTask
 
 from lapis.monitor import sampling_required
 
+if TYPE_CHECKING:
+    from lapis.drone import Drone
+
 
 class Job(object):
     __slots__ = ("resources", "used_resources", "walltime", "requested_walltime",
-                 "queue_date", "in_queue_since", "in_queue_until", "_name", "_success")
+                 "queue_date", "in_queue_since", "in_queue_until", "_name", "drone",
+                 "_success")
 
     def __init__(self, resources: dict, used_resources: dict, in_queue_since: float = 0,
-                 queue_date: float = 0, name: str = None):
+                 queue_date: float = 0, name: str = None, drone: "Drone" = None):
         """
         Definition of a job that uses a specified amount of resources `used_resources`
         over a given amount of time, `walltime`. A job is described by its user
@@ -24,14 +29,15 @@ class Job(object):
                                simulation scheduler
         :param queue_date: Time when job was inserted into queue in real life
         :param name: Name of the job
+        :param drone: Drone where the job is running on
         """
         self.resources = resources
         self.used_resources = used_resources
         for key in used_resources:
             if key not in resources:
-                logging.getLogger("implementation")\
-                    .info("job uses different resources than specified, added",
-                          key, self.used_resources[key])
+                logging.getLogger("implementation").info(
+                    "job uses different resources than specified, added %s: %s",
+                    key, self.used_resources[key])
                 self.resources[key] = self.used_resources[key]
         self.walltime = used_resources.pop("walltime")
         self.requested_walltime = resources.pop("walltime", None)
@@ -39,15 +45,16 @@ class Job(object):
         assert in_queue_since >= 0, "Queue time cannot be negative"
         self.in_queue_since = in_queue_since
         self.in_queue_until = None
+        self.drone = drone
         self._name = name
-        self._success = False
+        self._success: Optional[bool] = None
 
     @property
     def name(self) -> str:
         return self._name or id(self)
 
     @property
-    def successful(self) -> bool:
+    def successful(self) -> Optional[bool]:
         return self._success
 
     @property
@@ -64,27 +71,18 @@ class Job(object):
 
     async def run(self):
         self.in_queue_until = time.now
-        logging.info("job_status", {
-            "job_queue_time": {
-                repr(self): self.queue_date
-            }, "job_waiting_time": {
-                repr(self): self.waiting_time
-            }
-        })
+        self._success = None
+        await sampling_required.put(self)
         try:
             await (time + self.walltime)
-        except TaskCancelled:
+        except CancelTask:
             self._success = False
         except BaseException:
             self._success = False
             raise
         else:
-            logging.info("job_status", {
-                "job_wall_time": {
-                    repr(self): self.walltime
-                }
-            })
             self._success = True
+        await sampling_required.put(self)
 
     def __repr__(self):
         return '<%s: %s>' % (self.__class__.__name__, self._name or id(self))
@@ -97,7 +95,6 @@ async def job_to_queue_scheduler(job_generator, job_queue):
             base_date = job.queue_date
         current_time = job.queue_date - base_date
         if time.now < current_time:
-            await sampling_required.set(True)
             await (time >= current_time)
         job.in_queue_since = time.now
         await job_queue.put(job)
diff --git a/lapis/job_io/htcondor.py b/lapis/job_io/htcondor.py
index 1115f53..6a8637d 100644
--- a/lapis/job_io/htcondor.py
+++ b/lapis/job_io/htcondor.py
@@ -31,7 +31,7 @@ def htcondor_job_reader(iterable, resource_name_mapping={  # noqa: B006
     for row in htcondor_reader:
         if float(row[used_resource_name_mapping["walltime"]]) <= 0:
             logging.getLogger("implementation").warning(
-                "removed job from htcondor import", row)
+                "removed job from htcondor import (%s)", row)
             continue
         resources = {}
         for key, original_key in resource_name_mapping.items():
diff --git a/lapis/monitor/__init__.py b/lapis/monitor/__init__.py
index 01e6855..fb3faeb 100644
--- a/lapis/monitor/__init__.py
+++ b/lapis/monitor/__init__.py
@@ -1,13 +1,10 @@
 import copy
 import logging
 import logging.handlers
-from typing import Callable, TYPE_CHECKING
+from typing import Callable
 
 from cobald.monitor.format_json import JsonFormatter
-from usim import time, Flag, delay
-
-if TYPE_CHECKING:
-    from lapis.simulator import Simulator
+from usim import time, Queue
 
 
 class LoggingSocketHandler(logging.handlers.SocketHandler):
@@ -29,28 +26,53 @@ class SimulationTimeFilter(logging.Filter):
         return True
 
 
-sampling_required = Flag()
+sampling_required = Queue()
 
 
 class Monitoring(object):
-    def __init__(self, simulator: "Simulator"):
-        self.simulator = simulator
-        self._statistics = []
+    """
+    Enable monitoring of a simulation. Objects that change during simulation are
+    registered in a queue. Whenever objects in the queue become available, the
+    monitoring object takes care to dispatch the object to registered statistic
+    callables taking care to generate relevant monitoring output.
+    """
+    def __init__(self):
+        self._statistics = {}
 
     async def run(self):
-        async for _ in delay(1):
-            await sampling_required
-            await sampling_required.set(False)
-            for statistic in self._statistics:
+        async for log_object in sampling_required:
+            for statistic in self._statistics.get(type(log_object), set()):
                 # do the logging
-                for record in statistic(self.simulator):
+                for record in statistic(log_object):
                     logging.getLogger(statistic.name).info(
                         statistic.name, record
                     )
 
-    def register_statistic(self, statistic: Callable):
+    def register_statistic(self, statistic: Callable) -> None:
+        """
+        Register a callable that takes an object for logging and generates a list
+        of records. The callable should have the following accessible attributes:
+
+        name:
+            The identifying name of the statistic for logging
+        logging_formatter:
+            Pre-defined formatters for the different supported logging formats
+            including socket, stream, and telegraf logging.
+        whitelist:
+            A tuple of objects the statistic callable is interested in to create
+            the required logging messages.
+
+        :param statistic: Callable that returns a list of records for logging
+        """
         assert hasattr(statistic, "name") and hasattr(statistic, "logging_formatter")
-        self._statistics.append(statistic)
+        try:
+            for element in statistic.whitelist:
+                self._statistics.setdefault(element, set()).add(statistic)
+        except AttributeError:
+            logging.getLogger("implementation").warning(
+                f"Removing statistic {statistic.name} as no whitelist has been defined."
+            )
+            return
 
         # prepare the logger
         logger = logging.getLogger(statistic.name)
diff --git a/lapis/monitor/cobald.py b/lapis/monitor/cobald.py
index 3c6c879..48b27bc 100644
--- a/lapis/monitor/cobald.py
+++ b/lapis/monitor/cobald.py
@@ -2,37 +2,34 @@ import logging
 
 from cobald.monitor.format_json import JsonFormatter
 from cobald.monitor.format_line import LineProtocolFormatter
-from typing import TYPE_CHECKING
 
+from lapis.drone import Drone
 from lapis.monitor import LoggingSocketHandler, LoggingUDPSocketHandler
+from lapis.pool import Pool
 
-if TYPE_CHECKING:
-    from lapis.simulator import Simulator
 
-
-def drone_statistics(simulator: "Simulator") -> list:
+def drone_statistics(drone: Drone) -> list:
     """
     Collect allocation, utilisation, demand and supply of drones.
 
-    :param simulator: the simulator
+    :param drone: the drone
     :return: list of records for logging
     """
-    results = []
-    for drone in simulator.job_scheduler.drone_list:
-        results.append({
-            "pool_configuration": "None",
-            "pool_type": "drone",
-            "pool": repr(drone),
-            "allocation": drone.allocation,
-            "utilisation": drone.utilisation,
-            "demand": drone.demand,
-            "supply": drone.supply,
-            "job_count": drone.jobs
-        })
+    results = [{
+        "pool_configuration": "None",
+        "pool_type": "drone",
+        "pool": repr(drone),
+        "allocation": drone.allocation,
+        "utilisation": drone.utilisation,
+        "demand": drone.demand,
+        "supply": drone.supply,
+        "job_count": drone.jobs
+    }]
     return results
 
 
 drone_statistics.name = "cobald_status"
+drone_statistics.whitelist = Drone,
 drone_statistics.logging_formatter = {
     LoggingSocketHandler.__name__: JsonFormatter(),
     logging.StreamHandler.__name__: JsonFormatter(),
@@ -43,28 +40,27 @@ drone_statistics.logging_formatter = {
 }
 
 
-def pool_statistics(simulator: "Simulator") -> list:
+def pool_statistics(pool: Pool) -> list:
     """
     Collect allocation, utilisation, demand and supply of pools.
 
-    :param simulator: the simulator
+    :param pool: the pool
     :return: list of records to log
     """
-    results = []
-    for pool in simulator.pools:
-        results.append({
-            "pool_configuration": "None",
-            "pool_type": "pool",
-            "pool": repr(pool),
-            "allocation": pool.allocation,
-            "utilisation": pool.utilisation,
-            "demand": pool.demand,
-            "supply": pool.supply,
-        })
+    results = [{
+        "pool_configuration": "None",
+        "pool_type": "pool",
+        "pool": repr(pool),
+        "allocation": pool.allocation,
+        "utilisation": pool.utilisation,
+        "demand": pool.demand,
+        "supply": pool.supply,
+    }]
     return results
 
 
 pool_statistics.name = "cobald_status"
+pool_statistics.whitelist = Pool,
 pool_statistics.logging_formatter = {
     LoggingSocketHandler.__name__: JsonFormatter(),
     logging.StreamHandler.__name__: JsonFormatter(),
diff --git a/lapis/monitor/general.py b/lapis/monitor/general.py
index ae8f0a3..4291ce8 100644
--- a/lapis/monitor/general.py
+++ b/lapis/monitor/general.py
@@ -5,39 +5,43 @@ import logging.handlers
 from cobald.monitor.format_json import JsonFormatter
 from cobald.monitor.format_line import LineProtocolFormatter
 
+from lapis.drone import Drone
+from lapis.job import Job
 from lapis.monitor import LoggingSocketHandler, LoggingUDPSocketHandler
+from lapis.pool import Pool
+from lapis.scheduler import CondorJobScheduler, JobQueue
 
 if TYPE_CHECKING:
     from lapis.simulator import Simulator
 
 
-def resource_statistics(simulator: "Simulator") -> list:
+def resource_statistics(drone: Drone) -> list:
     """
     Log ratio of used and requested resources for drones.
 
-    :param simulator: the simulator
+    :param drone: the drone
     :return: list of records for logging
     """
     results = []
-    for drone in simulator.job_scheduler.drone_list:
-        resources = drone.theoretical_available_resources
-        used_resources = drone.available_resources
-        for resource_type in resources:
-            results.append({
-                "resource_type": resource_type,
-                "pool_configuration": "None",
-                "pool_type": "drone",
-                "pool": repr(drone),
-                "used_ratio":
-                    1 - used_resources[resource_type]
-                    / drone.pool_resources[resource_type],
-                "requested_ratio":
-                    1 - resources[resource_type] / drone.pool_resources[resource_type]
-            })
+    resources = drone.theoretical_available_resources
+    used_resources = drone.available_resources
+    for resource_type in resources:
+        results.append({
+            "resource_type": resource_type,
+            "pool_configuration": "None",
+            "pool_type": "drone",
+            "pool": repr(drone),
+            "used_ratio":
+                1 - used_resources[resource_type]
+                / drone.pool_resources[resource_type],
+            "requested_ratio":
+                1 - resources[resource_type] / drone.pool_resources[resource_type]
+        })
     return results
 
 
 resource_statistics.name = "resource_status"
+resource_statistics.whitelist = Drone,
 resource_statistics.logging_formatter = {
     LoggingSocketHandler.__name__: JsonFormatter(),
     logging.StreamHandler.__name__: JsonFormatter(),
@@ -48,19 +52,21 @@ resource_statistics.logging_formatter = {
 }
 
 
-def user_demand(simulator: "Simulator") -> list:
+def user_demand(job_queue: JobQueue) -> list:
     """
     Log global user demand.
 
-    :param simulator: the simulator
+    :param scheduler: the scheduler
     :return: list of records for logging
     """
-    return [{
-        "value": len(simulator.job_scheduler.job_queue)
+    result = [{
+        "value": len(job_queue)
     }]
+    return result
 
 
 user_demand.name = "user_demand"
+user_demand.whitelist = JobQueue,
 user_demand.logging_formatter = {
     LoggingSocketHandler.__name__: JsonFormatter(),
     logging.StreamHandler.__name__: JsonFormatter(),
@@ -71,22 +77,33 @@ user_demand.logging_formatter = {
 }
 
 
-def job_statistics(simulator: "Simulator") -> list:
+def job_statistics(scheduler: CondorJobScheduler) -> list:
     """
-    Log number of jobs running in a drone.
+    Log number of jobs running in all drones.
 
-    :param simulator: the simulator
+    .. Note::
+
+        The logging is currently synchronised with the frequency of the
+        scheduler. If a finer resolution is required, the update of drones
+        can be considered additionally.
+
+    :param scheduler: the scheduler
     :return: list of records for logging
     """
     result = 0
-    for drone in simulator.job_scheduler.drone_list:
-        result += drone.jobs
+    for cluster in scheduler.drone_cluster.copy():
+        for drone in cluster:
+            result += drone.jobs
     return [{
+        "pool_configuration": "None",
+        "pool_type": "obs",
+        "pool": repr(scheduler),
         "job_count": result
     }]
 
 
 job_statistics.name = "cobald_status"
+job_statistics.whitelist = CondorJobScheduler,
 job_statistics.logging_formatter = {
     LoggingSocketHandler.__name__: JsonFormatter(),
     logging.StreamHandler.__name__: JsonFormatter(),
@@ -97,7 +114,69 @@ job_statistics.logging_formatter = {
 }
 
 
-def pool_status(simulator: "Simulator") -> list:
+def job_events(job: Job) -> list:
+    """
+    Log relevant events for jobs. Relevant events are
+
+    * start of a job,
+    * finishing of a job, either successful or not.
+
+    Information about the start of a job are relevant to enable timely analysis
+    of waiting times. For finishing of jobs information about the success itself,
+    but also additional information on exceeded resources or refusal by the
+    drone are added.
+
+    .. Warning::
+
+        The logging format includes the name / identifier of a job. This might
+        result in a huge index of the grafana database. The job is currently
+        included to enable better lookup and analysis of related events.
+
+    :param job: the job to log information for
+    :return: list of records for logging
+    """
+    result = {
+        "pool_configuration": "None",
+        "pool_type": "drone",
+        "pool": repr(job.drone),
+        "job": repr(job),
+    }
+    if job.successful is None:
+        result["queue_time"] = job.queue_date
+        result["waiting_time"] = job.waiting_time
+    elif job.successful:
+        result["wall_time"] = job.walltime
+        result["success"] = 1
+    else:
+        result["success"] = 0
+        error_logged = False
+        for resource_key in job.resources:
+            usage = job.used_resources.get(
+                resource_key, job.resources.get(resource_key, None))
+            value = usage / job.resources.get(
+                resource_key, job.drone.pool_resources[resource_key]
+            )
+            if value > 1:
+                result[f"exceeded_{resource_key}"] = value
+                error_logged = True
+        if not error_logged:
+            result["refused_by"] = repr(job.drone)
+    return [result]
+
+
+job_events.name = "job_event"
+job_events.whitelist = Job,
+job_events.logging_formatter = {
+    LoggingSocketHandler.__name__: JsonFormatter(),
+    logging.StreamHandler.__name__: JsonFormatter(),
+    LoggingUDPSocketHandler.__name__: LineProtocolFormatter(
+        tags={"tardis", "pool_configuration", "pool_type", "pool", "job"},
+        resolution=1
+    )
+}
+
+
+def pool_status(pool: Pool) -> list:
     """
     Log state changes of pools and drones.
 
@@ -108,6 +187,7 @@ def pool_status(simulator: "Simulator") -> list:
 
 
 pool_status.name = "pool_status"
+pool_status.whitelist = Pool,
 pool_status.logging_formatter = {
     LoggingSocketHandler.__name__: JsonFormatter(),
     logging.StreamHandler.__name__: JsonFormatter(),
diff --git a/lapis/scheduler.py b/lapis/scheduler.py
index a9a6815..02f9e03 100644
--- a/lapis/scheduler.py
+++ b/lapis/scheduler.py
@@ -2,6 +2,11 @@ from typing import Dict
 from usim import Scope, interval
 
 from lapis.drone import Drone
+from lapis.monitor import sampling_required
+
+
+class JobQueue(list):
+    pass
 
 
 class CondorJobScheduler(object):
@@ -25,7 +30,7 @@ class CondorJobScheduler(object):
         self._stream_queue = job_queue
         self.drone_cluster = []
         self.interval = 60
-        self.job_queue = []
+        self.job_queue = JobQueue()
         self._collecting = True
 
     @property
@@ -86,6 +91,7 @@ class CondorJobScheduler(object):
                     if best_match:
                         scope.do(best_match.start_job(job))
                         self.job_queue.remove(job)
+                        await sampling_required.put(self.job_queue)
                         self.unregister_drone(best_match)
                         left_resources = best_match.theoretical_available_resources
                         left_resources = {
@@ -95,10 +101,13 @@ class CondorJobScheduler(object):
                         self._add_drone(best_match, left_resources)
                 if not self._collecting and not self.job_queue:
                     break
+                await sampling_required.put(self)
 
     async def _collect_jobs(self):
         async for job in self._stream_queue:
             self.job_queue.append(job)
+            # TODO: logging happens with each job
+            await sampling_required.put(self.job_queue)
         self._collecting = False
 
     def _schedule_job(self, job) -> Drone:
diff --git a/lapis/simulator.py b/lapis/simulator.py
index 584bc12..cf615a9 100644
--- a/lapis/simulator.py
+++ b/lapis/simulator.py
@@ -1,3 +1,4 @@
+import logging
 import random
 from functools import partial
 
@@ -6,11 +7,14 @@ from usim import run, time, until, Scope, Queue
 from lapis.drone import Drone
 from lapis.job import job_to_queue_scheduler
 from lapis.monitor.general import user_demand, job_statistics, \
-    resource_statistics, pool_status, configuration_information
+    resource_statistics, pool_status, configuration_information, job_events
 from lapis.monitor import Monitoring
 from lapis.monitor.cobald import drone_statistics, pool_statistics
 
 
+logging.getLogger("implementation").propagate = False
+
+
 class Simulator(object):
     def __init__(self, seed=1234):
         random.seed(seed)
@@ -25,9 +29,10 @@ class Simulator(object):
         self.enable_monitoring()
 
     def enable_monitoring(self):
-        self.monitoring = Monitoring(self)
+        self.monitoring = Monitoring()
         self.monitoring.register_statistic(user_demand)
         self.monitoring.register_statistic(job_statistics)
+        self.monitoring.register_statistic(job_events)
         self.monitoring.register_statistic(pool_statistics)
         self.monitoring.register_statistic(drone_statistics)
         self.monitoring.register_statistic(resource_statistics)
@@ -49,11 +54,11 @@ class Simulator(object):
         self.job_scheduler = scheduler_type(job_queue=self.job_queue)
 
     def run(self, until=None):
-        print("running until", until)
+        print(f"running until {until}")
         run(self._simulate(until))
 
     async def _simulate(self, end):
-        print("Starting simulation at %s" % time.now)
+        print(f"Starting simulation at {time.now}")
         async with until(time == end) if end else Scope() as while_running:
             for pool in self.pools:
                 while_running.do(pool.run(), volatile=True)


Test patch:
diff --git a/lapis_tests/utility/test_monitor.py b/lapis_tests/utility/test_monitor.py
index 6d048fa..0e2cf9e 100644
--- a/lapis_tests/utility/test_monitor.py
+++ b/lapis_tests/utility/test_monitor.py
@@ -5,7 +5,7 @@ from time import time as pytime
 from cobald.monitor.format_line import LineProtocolFormatter
 from usim import Scope, time
 
-from lapis_tests import via_usim, DummyScheduler
+from lapis_tests import via_usim
 
 from . import make_test_logger
 
@@ -62,16 +62,23 @@ def dummy_statistics():
 
 class TestMonitoring(object):
     def test_registration(self):
-        scheduler = DummyScheduler()
-        monitoring = Monitoring(scheduler)
+        monitoring = Monitoring()
         statistics = resource_statistics
         monitoring.register_statistic(statistics)
-        assert statistics in monitoring._statistics
+        for element in statistics.whitelist:
+            assert statistics in monitoring._statistics.get(element)
 
     def test_registration_failure(self):
-        scheduler = DummyScheduler()
-        monitoring = Monitoring(scheduler)
+        monitoring = Monitoring()
         statistics = dummy_statistics
         with pytest.raises(AssertionError):
             monitoring.register_statistic(statistics)
-        assert statistics not in monitoring._statistics
+        assert all(statistics not in stat for stat in monitoring._statistics.values())
+        # define required attributes except whitelist
+        statistics.name = "test"
+        statistics.logging_formatter = {}
+        monitoring.register_statistic(statistics)
+        assert all(statistics not in stat for stat in monitoring._statistics.values())
+        statistics.whitelist = str,
+        monitoring.register_statistic(statistics)
+        assert all(statistics in stat for stat in monitoring._statistics.values())