better worker logs (#307)

Author: JKL98ISR

backend/deepchecks_monitoring/bgtasks/alert_task.py

@@ -21,7 +21,7 @@
 
 from deepchecks_monitoring.api.v1.alert import AlertCreationSchema
 from deepchecks_monitoring.logic.check_logic import SingleCheckRunOptions, reduce_check_window, run_check_window
-from deepchecks_monitoring.monitoring_utils import DataFilterList, make_oparator_func
+from deepchecks_monitoring.monitoring_utils import DataFilterList, configure_logger, make_oparator_func
 from deepchecks_monitoring.public_models import Organization
 from deepchecks_monitoring.public_models.task import BackgroundWorker, Task
 from deepchecks_monitoring.resources import ResourcesProvider
@@ -32,28 +32,31 @@
 from deepchecks_monitoring.utils import database
 from deepchecks_monitoring.utils.mixpanel import AlertTriggeredEvent
 
-__all__ = ["AlertsTask"]
+__all__ = ['AlertsTask']
 
 
 class AlertsTask(BackgroundWorker):
     """Worker to calculate alerts"""
 
     def __init__(self):
         super().__init__()
-        self._logger = logging.getLogger(__name__)
+        self.logger = configure_logger(self.__class__.__name__)
 
     @classmethod
     def queue_name(cls) -> str:
-        return "alerts"
+        return 'alerts'
 
     @classmethod
     def delay_seconds(cls) -> int:
         return 0
 
     async def run(self, task: Task, session: AsyncSession, resources_provider: ResourcesProvider, lock):
-        organization_id = task.params["organization_id"]
-        monitor_id = task.params["monitor_id"]
-        timestamp = task.params["timestamp"]
+        organization_id = task.params['organization_id']
+        monitor_id = task.params['monitor_id']
+        timestamp = task.params['timestamp']
+
+        self.logger.info({'message': 'starting job', 'worker name': str(type(self)),
+                          'task': task.id, 'monitor_id': monitor_id, 'org_id': organization_id})
 
         organization_schema = (await session.execute(
             select(Organization.schema_name).where(Organization.id == organization_id)
@@ -63,14 +66,14 @@ async def run(self, task: Task, session: AsyncSession, resources_provider: Resou
         if organization_schema is not None:
             await database.attach_schema_switcher_listener(
                 session=session,
-                schema_search_path=[organization_schema, "public"]
+                schema_search_path=[organization_schema, 'public']
             )
             alerts = await execute_monitor(
                 session=session,
                 resources_provider=resources_provider,
                 monitor_id=monitor_id,
                 timestamp=timestamp,
-                logger=self._logger,
+                logger=self.logger,
                 organization_id=organization_id,
             )
         else:
@@ -87,7 +90,7 @@ async def run(self, task: Task, session: AsyncSession, resources_provider: Resou
                 organization_id=organization_id,
                 session=session,
                 resources_provider=resources_provider,
-                logger=self._logger.getChild("alert-notificator")
+                logger=self.logger.getChild('alert-notificator')
             )
             await resources_provider.report_mixpanel_event(
                 AlertTriggeredEvent.create_event,
@@ -96,6 +99,9 @@ async def run(self, task: Task, session: AsyncSession, resources_provider: Resou
             )
             await notificator.notify()
 
+        self.logger.info({'message': 'finished job', 'worker name': str(type(self)),
+                          'task': task.id, 'monitor_id': monitor_id, 'org_id': organization_id})
+
 
 async def execute_monitor(
         monitor_id: int,
@@ -106,8 +112,8 @@ async def execute_monitor(
         logger: t.Optional[logging.Logger] = None,
 ) -> t.List[Alert]:
     """Execute monitor alert rules."""
-    logger = logger or logging.getLogger("monitor-executor")
-    logger.info("Execution of Monitor(id:%s) for timestamp %s", monitor_id, timestamp)
+    logger = logger or logging.getLogger('monitor-executor')
+    logger.info('Execution of Monitor(id:%s) for timestamp %s', monitor_id, timestamp)
 
     monitor = t.cast(Monitor, await session.scalar(
         sa.select(Monitor)
@@ -119,13 +125,13 @@ async def execute_monitor(
     ))
 
     if monitor is None:
-        raise ValueError(f"Did not find monitor with the id {monitor_id}")
+        raise ValueError(f'Did not find monitor with the id {monitor_id}')
 
     check = monitor.check
     alert_rules = monitor.alert_rules
 
     if len(alert_rules) == 0:
-        logger.info("Monitor(id:%s) does not have alert rules", monitor_id)
+        logger.info('Monitor(id:%s) does not have alert rules', monitor_id)
         return []
 
     monitor_frequency = t.cast(Frequency, monitor.frequency).to_pendulum_duration()
@@ -141,7 +147,7 @@ async def execute_monitor(
     )).all())
 
     if not model_versions:
-        logger.info("Model(id:%s) is empty (does not have versions)", check.model_id)
+        logger.info('Model(id:%s) is empty (does not have versions)', check.model_id)
         return []
 
     # First looking for results in cache if already calculated
@@ -154,7 +160,7 @@ async def execute_monitor(
             cache_results[model_version] = cache_result.value
         else:
             model_versions_without_cache.append(model_version)
-        logger.debug("Cache result: %s", cache_results)
+        logger.debug('Cache result: %s', cache_results)
 
     # For model versions without result in cache running calculation
     if model_versions_without_cache:
@@ -178,7 +184,7 @@ async def execute_monitor(
             resources_provider.cache_functions.set_monitor_cache(
                 organization_id, version.id, monitor_id, start_time, end_time, result)
 
-        logger.debug("Check execution result: %s", result_per_version)
+        logger.debug('Check execution result: %s', result_per_version)
     else:
         result_per_version = {}
 
@@ -192,20 +198,20 @@ async def execute_monitor(
             await session.execute(update(AlertRule).where(AlertRule.id == alert_rule.id)
                                   .values({AlertRule.start_time: func.least(AlertRule.start_time, end_time)}))
         if not alert_rule.is_active:
-            logger.info("AlertRule(id:%s) is not active, skipping it", alert_rule.id)
+            logger.info('AlertRule(id:%s) is not active, skipping it', alert_rule.id)
         elif alert := assert_check_results(alert_rule, check_results):
             alert.start_time = start_time
             alert.end_time = end_time
             AlertCreationSchema.validate(alert)
             session.add(alert)
-            logger.info("Alert(id:%s) instance created for monitor(id:%s)", alert.id, monitor.id)
+            logger.info('Alert(id:%s) instance created for monitor(id:%s)', alert.id, monitor.id)
             alerts.append(alert)
 
     if (n_of_alerts := len(alerts)) > 0:
-        logger.info("%s alerts raised for Monitor(id:%s)", n_of_alerts, monitor.id)
+        logger.info('%s alerts raised for Monitor(id:%s)', n_of_alerts, monitor.id)
         return alerts
 
-    logger.info("No alerts were raised for Monitor(id:%s)", monitor.id)
+    logger.info('No alerts were raised for Monitor(id:%s)', monitor.id)
     return []
 
 

backend/deepchecks_monitoring/bgtasks/delete_db_table_task.py

@@ -15,6 +15,7 @@
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from deepchecks_monitoring.monitoring_utils import configure_logger
 from deepchecks_monitoring.public_models.task import UNIQUE_NAME_TASK_CONSTRAINT, BackgroundWorker, Task
 
 __all__ = ['DeleteDbTableTask', 'insert_delete_db_table_task']
@@ -23,9 +24,11 @@
 
 
 class DeleteDbTableTask(BackgroundWorker):
-    """Worker to delete any database tables.
+    """Worker to delete any database tables."""
 
-    """
+    def __init__(self):
+        super().__init__()
+        self.logger = configure_logger(self.__class__.__name__)
 
     @classmethod
     def queue_name(cls) -> str:
@@ -36,10 +39,12 @@ def delay_seconds(cls) -> int:
         return 0
 
     async def run(self, task: 'Task', session: AsyncSession, resources_provider, lock):
+        self.logger.info({'message': 'started job', 'worker name': str(type(self))})
         for table in task.params['full_table_paths']:
             await session.execute(text(f'DROP TABLE IF EXISTS {table}'))
         # Deleting the task
         await session.execute(delete(Task).where(Task.id == task.id))
+        self.logger.info({'message': 'finished job', 'worker name': str(type(self))})
 
 
 async def insert_delete_db_table_task(session: AsyncSession, full_table_paths: t.List[str]):

backend/deepchecks_monitoring/bgtasks/mixpanel_system_state_event.py

@@ -14,21 +14,26 @@
 from sqlalchemy.dialects.postgresql import insert as pginsert
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from deepchecks_monitoring.monitoring_utils import configure_logger
 from deepchecks_monitoring.public_models.organization import Organization
 from deepchecks_monitoring.public_models.task import UNIQUE_NAME_TASK_CONSTRAINT, BackgroundWorker, Task
 from deepchecks_monitoring.resources import ResourcesProvider
 from deepchecks_monitoring.utils import database, mixpanel
 
-__all__ = ["MixpanelSystemStateEvent"]
+__all__ = ['MixpanelSystemStateEvent']
 
 
-QUEUE_NAME = "mixpanel system state event"
+QUEUE_NAME = 'mixpanel system state event'
 DELAY = 0
 
 
 class MixpanelSystemStateEvent(BackgroundWorker):
     """Worker that sends a system state event to the mixpanel."""
 
+    def __init__(self):
+        super().__init__()
+        self.logger = configure_logger(self.__class__.__name__)
+
     @classmethod
     def queue_name(cls) -> str:
         """REturn queue name."""
@@ -46,38 +51,40 @@ def retry_seconds(cls) -> int:
 
     async def run(
         self,
-        task: "Task",
+        task: 'Task',
         session: AsyncSession,
         resources_provider: ResourcesProvider,
         lock: Lock
     ):
         """Run task."""
+
         if not resources_provider.is_analytics_enabled:
             return
         if not resources_provider.settings.is_on_prem or resources_provider.settings.is_cloud:
             return
-
+        self.logger.info({'message': 'started job', 'worker name': str(type(self))})
         organizations = (await session.scalars(
             sa.select(Organization))
         ).all()
 
         for org in organizations:
             async with database.attach_schema_switcher(
                 session=session,
-                schema_search_path=[org.schema_name, "public"]
+                schema_search_path=[org.schema_name, 'public']
             ):
                 await resources_provider.report_mixpanel_event(
                     mixpanel.HealthcheckEvent.create_event,
                     organization=org
                 )
+        self.logger.info({'message': 'finished job', 'worker name': str(type(self))})
 
     @classmethod
     async def enqueue_task(cls, session: AsyncSession):
         """Enqueue task."""
         values = {
-            "name": "system-state",
-            "bg_worker_task": cls.queue_name(),
-            "params": {}
+            'name': 'system-state',
+            'bg_worker_task': cls.queue_name(),
+            'params': {}
         }
         await session.execute(
             pginsert(Task)

backend/deepchecks_monitoring/bgtasks/model_data_ingestion_alerter.py

@@ -14,7 +14,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import joinedload
 
-from deepchecks_monitoring.monitoring_utils import make_oparator_func
+from deepchecks_monitoring.monitoring_utils import configure_logger, make_oparator_func
 from deepchecks_monitoring.public_models.organization import Organization
 from deepchecks_monitoring.public_models.task import BackgroundWorker, Task
 from deepchecks_monitoring.resources import ResourcesProvider
@@ -26,16 +26,20 @@
 from deepchecks_monitoring.utils import database
 from deepchecks_monitoring.utils.alerts import Condition
 
-__all__ = ["ModelDataIngestionAlerter"]
+__all__ = ['ModelDataIngestionAlerter']
 
 
-QUEUE_NAME = "model data ingestion alerter"
+QUEUE_NAME = 'model data ingestion alerter'
 DELAY = 60
 
 
 class ModelDataIngestionAlerter(BackgroundWorker):
     """Worker that alerts about data ingestion stats in relation to a model."""
 
+    def __init__(self):
+        super().__init__()
+        self.logger = configure_logger(self.__class__.__name__)
+
     @classmethod
     def queue_name(cls) -> str:
         return QUEUE_NAME
@@ -44,14 +48,17 @@ def queue_name(cls) -> str:
     def delay_seconds(cls) -> int:
         return DELAY
 
-    async def run(self, task: "Task",
+    async def run(self, task: 'Task',
                   session: AsyncSession,  # pylint: disable=unused-argument
                   resources_provider: ResourcesProvider,
                   lock):
-        alert_rule_id = task.params["alert_rule_id"]
-        org_id = task.params["organization_id"]
-        end_time = task.params["end_time"]
-        start_time = task.params["start_time"]
+        alert_rule_id = task.params['alert_rule_id']
+        org_id = task.params['organization_id']
+        end_time = task.params['end_time']
+        start_time = task.params['start_time']
+
+        self.logger.info({'message': 'entered job', 'worker name': str(type(self)),
+                          'task': task.id, 'alert_rule_id': alert_rule_id, 'org_id': org_id})
 
         organization_schema = (await session.execute(
             sa.select(Organization.schema_name).where(
@@ -66,7 +73,7 @@ async def run(self, task: "Task",
 
         await database.attach_schema_switcher_listener(
             session=session,
-            schema_search_path=[organization_schema, "public"]
+            schema_search_path=[organization_schema, 'public']
         )
 
         alert_rule: DataIngestionAlertRule = (
@@ -86,8 +93,8 @@ async def run(self, task: "Task",
         pdl_start_time = as_pendulum_datetime(start_time)
         pdl_end_time = as_pendulum_datetime(end_time)
 
-        def truncate_date(col, agg_time_unit: str = "day"):
-            return sa.func.cast(sa.func.extract("epoch", sa.func.date_trunc(agg_time_unit, col)), sa.Integer)
+        def truncate_date(col, agg_time_unit: str = 'day'):
+            return sa.func.cast(sa.func.extract('epoch', sa.func.date_trunc(agg_time_unit, col)), sa.Integer)
 
         def sample_id(columns):
             return getattr(columns, SAMPLE_ID_COL)
@@ -103,33 +110,36 @@ def sample_label(columns):
         if not tables:
             return
 
+        self.logger.info({'message': 'starting job', 'worker name': str(type(self)),
+                          'task': task.id, 'alert_rule_id': alert_rule_id, 'org_id': org_id})
+
         labels_table = model.get_sample_labels_table(session)
         # Get all samples within time window from all the versions
         data_query = sa.union_all(*(
             sa.select(
-                sample_id(table.c).label("sample_id"),
+                sample_id(table.c).label('sample_id'),
                 truncate_date(sample_timestamp(table.c),
-                              freq.value.lower()).label("timestamp")
+                              freq.value.lower()).label('timestamp')
             ).where(
                 sample_timestamp(table.c) <= pdl_end_time,
                 sample_timestamp(table.c) > pdl_start_time
             ).distinct()
             for table in tables)
         )
-        joined_query = sa.select(sa.literal(model.id).label("model_id"),
+        joined_query = sa.select(sa.literal(model.id).label('model_id'),
                                  data_query.c.sample_id,
                                  data_query.c.timestamp,
-                                 sa.func.cast(sample_label(labels_table.c), sa.String).label("label")) \
+                                 sa.func.cast(sample_label(labels_table.c), sa.String).label('label')) \
             .join(labels_table, onclause=data_query.c.sample_id == sample_id(labels_table.c), isouter=True)
 
         rows = (await session.execute(
             sa.select(
                 joined_query.c.model_id,
                 joined_query.c.timestamp,
-                sa.func.count(joined_query.c.sample_id).label("count"),
-                sa.func.count(sa.func.cast(joined_query.c.label, sa.String)).label("label_count"))
+                sa.func.count(joined_query.c.sample_id).label('count'),
+                sa.func.count(sa.func.cast(joined_query.c.label, sa.String)).label('label_count'))
             .group_by(joined_query.c.model_id, joined_query.c.timestamp)
-            .order_by(joined_query.c.model_id, joined_query.c.timestamp, "count"),
+            .order_by(joined_query.c.model_id, joined_query.c.timestamp, 'count'),
         )).fetchall()
 
         pendulum_freq = freq.to_pendulum_duration()
@@ -156,3 +166,6 @@ def sample_label(columns):
 
         await session.execute(sa.delete(Task).where(Task.id == task.id))
         await session.commit()
+
+        self.logger.info({'message': 'finished job', 'worker name': str(type(self)),
+                          'task': task.id, 'alert_rule_id': alert_rule_id, 'org_id': org_id})