[Feature] Recurring / rescheduling tasks

.github/dependabot.yml

@@ -0,0 +1,11 @@
+# To get started with Dependabot version updates, you'll need to specify which
+# package ecosystems to update and where the package manifests are located.
+# Please see the documentation for all configuration options:
+# https://docs.github.com/github/administering-a-repository/configuration-options-for-dependency-updates
+
+version: 2
+updates:
+  - package-ecosystem: "pip" # See documentation for possible values
+    directory: "/" # Location of package manifests
+    schedule:
+      interval: "weekly"

README.md

@@ -77,7 +77,7 @@ to use whatever you want.
 
 ### Prerequisites
 
-By the use of environment variables we load in the configuration of the
+By the use of environment variables we load in the configuration of the 
 scheduler. Look at the [.env-dist](.env-dist) file for the application
 configuration settings, to build a `.env` file. Refer to the
 [`configuration.md`](docs/configuration.md) file for more information on the

docs/design.md

@@ -48,14 +48,13 @@ graph TB
     Scheduler["Scheduler<br/>[system]"]
     TaskRunner["Task Runner<br/>[software system]"]
 
-    Rocky--"Create object"-->Octopoes
     Rocky--"Create scan job<br/>HTTP POST"--->Scheduler
 
-    Octopoes--"Get random oois<br/>HTTP GET"-->Scheduler
+    Octopoes--"Scan Profile Mutation<br/>AMQP"-->RabbitMQ
 
-    RabbitMQ--"Get latest created oois<br/>Get latest raw files<br/>AMQP"-->Scheduler
+    RabbitMQ--"1. Get scan profile mutation<br/>2. Get latest raw files<br/>AMQP"-->Scheduler
 
-    Katalogus--"Get available plugins<br/>HTTP GET"-->Scheduler
+    Katalogus--"1. Get organizations</br>2. Get available plugins<br/>HTTP GET"-->Scheduler
     Bytes--"Get last run boefje<br/>HTTP GET"-->Scheduler
 
     Scheduler--"Pop task of queue"-->TaskRunner
@@ -73,11 +72,19 @@ Following we review how different dataflows, from the `boefjes` and the
 `normalizers` are implemented within the `Scheduler` system. The following
 events within a KAT installation will trigger dataflows in the `Scheduler`:
 
+App:
+
 * When a plugin is enabled or disabled (`monitor_organisations`)
 
 * When an organisation is created or deleted (`monitor_organisations`)
 
-* When a scan level is increased (`get_latest_object`)
+Boefje scheduler:
+
+* When a new boefje has been added (`get_new_boefjes_by_org_id`)
+
+* When a scan level is increased (`get_scan_profile_mutation`)
+
+Normalizer scheduler:
 
 * When a raw file is created (`get_latest_raw_data`)
 
@@ -172,21 +179,25 @@ flowchart TB
 * The `BoefjeScheduler` implementation of the `populate_queue()` method will:
 
   - Continuously get the latest scan level changes of ooi's from a message
-    queue that was sent by octopoes (`get_latest_objects()`). The tasks
-    created from these ooi's (`tasks = ooi * boefjes`) from this queue will
+    queue that is sent from octopoes received on a rabbitmq message queue
+    (`get_scan_profile_mutations`). The tasks created from these ooi's
+    (`tasks = ooi * boefjes`) from this queue will
     get the priority of 2.
 
-  - To fill up the queue, and to enforce that we reschedule tasks we get
-    random ooi's from octopoes (`get_random_objects`). The tasks of from these
-    ooi's (`tasks = ooi * boefjes`) will get the priority that has been
+  - Check newly enabled/added boefjes, and check the ooi's in the datastore
+    that match the boefje and tasks that need to be created.
+
+  - To fill up the queue, and to enforce that we reschedule tasks we reference
+    the scheduler internal ooi datastore. We only consider oois that have been
+    processed by the scheduler after the set grace period. The tasks of from
+    these ooi's (`tasks = ooi * boefjes`) will get the priority that has been
     calculated by the ranker. At the moment a task will get the priority of 3,
-    when 7 days have gone by (e.g. how longer it hasn't been checked the
-    higher the priority it will get). For everything that hasn't been check
-    before the 7 days it will scale the priority appropriately.
+    when 7 days have gone by (e.g. how longer it hasn't been checked the higher
+    the priority it will get). For everything that hasn't been check before the
+    7 days it will scale the priority appropriately.
 
-  - In order for a  created tasks from `get_latest_objects()` and
-    `get_random_objects()` to be elligible for execution, the task adhere to
-    the following (`create_tasks_for_ooi()`):
+  - In order for a created tasks o be elligible for execution, the tasks adhere
+    to the following (`create_tasks_for_ooi()`):
 
     * Should not have run within the 'grace period', meaning a task should not
       be scheduled again within the last 24 hours (can be configured).

scheduler/config/settings.py

@@ -30,6 +30,7 @@ class Settings(BaseSettings):
     host_bytes_password: str = Field(..., env="BYTES_PASSWORD")
     host_octopoes: str = Field(..., env="OCTOPOES_API")
     host_scan_profile: str = Field(..., env="SCHEDULER_RABBITMQ_DSN")
+    host_mutation: str = Field(..., env="SCHEDULER_RABBITMQ_DSN")
     host_raw_data: str = Field(..., env="SCHEDULER_RABBITMQ_DSN")
     host_normalizer_meta: str = Field(..., env="SCHEDULER_RABBITMQ_DSN")
 

scheduler/connectors/listeners/__init__.py

@@ -1,4 +1,4 @@
 from .listeners import Listener, RabbitMQ
 from .normalizer_meta import NormalizerMeta
+from .ooi import ScanProfileMutation
 from .raw_data import RawData
-from .scan_profile import ScanProfile

scheduler/connectors/listeners/ooi.py

@@ -0,0 +1,31 @@
+from typing import List, Optional
+
+from scheduler.connectors.errors import exception_handler
+from scheduler.models import ScanProfileMutation as ScanProfileMutationModel
+
+from .listeners import RabbitMQ
+
+
+class ScanProfileMutation(RabbitMQ):
+    name = "scan_profile_mutation"
+
+    @exception_handler
+    def get_scan_profile_mutation(self, queue: str) -> Optional[ScanProfileMutationModel]:
+        response = self.get(queue)
+        if response is None:
+            return None
+
+        return ScanProfileMutationModel(**response)
+
+    @exception_handler
+    def get_scan_profile_mutations(self, queue: str, n: int) -> Optional[List[ScanProfileMutationModel]]:
+        oois: List[ScanProfileMutationModel] = []
+
+        for _ in range(n):
+            ooi = self.get_scan_profile_mutation(queue=queue)
+            if ooi is None:
+                break
+
+            oois.append(ooi)
+
+        return oois

scheduler/connectors/services/katalogus.py

@@ -1,4 +1,4 @@
-from typing import List
+from typing import Dict, List
 
 from scheduler.connectors.errors import exception_handler
 from scheduler.models import Boefje, Organisation, Plugin
@@ -13,10 +13,37 @@ class Katalogus(HTTPService):
     def __init__(self, host: str, source: str, timeout: int = 5):
         super().__init__(host, source, timeout)
 
+        # Example:
+        #
+        # {
+        #     "organisation_id": {
+        #          "plugin-id": {}
+        #     }
+        # }
         self.organisations_plugin_cache: dict_utils.ExpiringDict = dict_utils.ExpiringDict(lifetime=30)
+
+        # Example:
+        #
+        # {
+        #     "organisation_id": {
+        #         "plugin-type": {
+        #             "plugin-id": {}
+        #         }
+        #     }
+        # }
         self.organisations_boefje_type_cache: dict_utils.ExpiringDict = dict_utils.ExpiringDict(lifetime=30)
         self.organisations_normalizer_type_cache: dict_utils.ExpiringDict = dict_utils.ExpiringDict(lifetime=30)
 
+        # Example:
+        #
+        # {
+        #     "organisation_id": {
+        #          "plugin-id": {}
+        #     }
+        # }
+        self.organisations_new_boefjes_cache: Dict = {}
+
+        # Initialize the caches
         self._flush_organisations_plugin_cache()
         self._flush_organisations_normalizer_type_cache()
         self._flush_organisations_boefje_type_cache()
@@ -26,9 +53,19 @@ def _flush_organisations_plugin_cache(self) -> None:
         orgs = self.get_organisations()
 
         for org in orgs:
-            self.organisations_plugin_cache[org.id] = {
-                plugin.id: plugin for plugin in self.get_plugins_by_organisation(org.id)
-            }
+            if org.id not in self.organisations_plugin_cache:
+                self.organisations_plugin_cache[org.id] = {}
+
+            plugins = self.get_plugins_by_organisation(org.id)
+            for plugin in plugins:
+                if plugin.id in self.organisations_plugin_cache[org.id]:
+                    continue
+
+                # Add new boefje to organisation plugin cache and new boefjes cache
+                self.organisations_plugin_cache[org.id][plugin.id] = plugin
+
+                # Add new boefje to new boefjes cache
+                self.organisations_new_boefjes_cache.setdefault(org.id, {})[plugin.id] = plugin
 
     def _flush_organisations_boefje_type_cache(self) -> None:
         """boefje.consumes -> plugin type boefje"""
@@ -39,18 +76,14 @@ def _flush_organisations_boefje_type_cache(self) -> None:
             self.organisations_boefje_type_cache[org.id] = {}
 
             for plugin in self.get_plugins_by_organisation(org.id):
-                if plugin.type != "boefje":
-                    continue
-
-                # NOTE: backwards compatability, when it is a boefje the
-                # consumes field is a string field.
-                if isinstance(plugin.consumes, str):
-                    self.organisations_boefje_type_cache[org.id].setdefault(plugin.consumes, []).append(plugin)
+                if plugin.type != "boefje" and plugin.enabled is False: # TODO: CHECK THIS enabled
                     continue
 
                 for type_ in plugin.consumes:
                     self.organisations_boefje_type_cache[org.id].setdefault(type_, []).append(plugin)
 
+        self.logger.debug("flushed boefje cache [cache=%s]", self.organisations_boefje_type_cache.cache)
+
     def _flush_organisations_normalizer_type_cache(self) -> None:
         """normalizer.consumes -> plugin type normalizer"""
         self.logger.debug("flushing normalizer cache [cache=%s]", self.organisations_normalizer_type_cache.cache)
@@ -60,12 +93,14 @@ def _flush_organisations_normalizer_type_cache(self) -> None:
             self.organisations_normalizer_type_cache[org.id] = {}
 
             for plugin in self.get_plugins_by_organisation(org.id):
-                if plugin.type != "normalizer":
+                if plugin.type != "normalizer" and plugin.enabled is False:  # TODO: CHECK THIS enabled
                     continue
 
                 for type_ in plugin.consumes:
                     self.organisations_normalizer_type_cache[org.id].setdefault(type_, []).append(plugin)
 
+        self.logger.debug("flushed normalizer cache [cache=%s]", self.organisations_normalizer_type_cache.cache)
+
     @exception_handler
     def get_boefjes(self) -> List[Boefje]:
         url = f"{self.host}/boefjes"
@@ -115,3 +150,8 @@ def get_normalizers_by_org_id_and_type(self, organisation_id: str, normalizer_ty
         except dict_utils.ExpiredError:
             self._flush_organisations_normalizer_type_cache()
             return dict_utils.deep_get(self.organisations_normalizer_type_cache, [organisation_id, normalizer_type])
+
+    def get_new_boefjes_by_org_id(self, organisation_id: str) -> List[Plugin]:
+        new_boefjes = self.organisations_new_boefjes_cache[organisation_id].values()
+        self.organisations_new_boefjes_cache[organisation_id] = {}
+        return new_boefjes

scheduler/context/context.py

@@ -52,8 +52,8 @@ def __init__(self) -> None:
             orgs=svc_katalogus.get_organisations(),
         )
 
-        lst_scan_profile = listeners.ScanProfile(
-            dsn=self.config.host_scan_profile,
+        lst_mutations = listeners.ScanProfileMutation(
+            dsn=self.config.host_mutation,
         )
 
         lst_raw_data = listeners.RawData(
@@ -71,7 +71,7 @@ def __init__(self) -> None:
                 services.Katalogus.name: svc_katalogus,
                 services.Octopoes.name: svc_octopoes,
                 services.Bytes.name: svc_bytes,
-                listeners.ScanProfile.name: lst_scan_profile,
+                listeners.ScanProfileMutation.name: lst_mutations,
                 listeners.RawData.name: lst_raw_data,
                 listeners.NormalizerMeta.name: lst_normalizer_meta,
             }
@@ -83,3 +83,4 @@ def __init__(self) -> None:
         datastore = sqlalchemy.SQLAlchemy(self.config.database_dsn)
         self.task_store: stores.TaskStorer = sqlalchemy.TaskStore(datastore)
         self.pq_store: stores.PriorityQueueStorer = sqlalchemy.PriorityQueueStore(datastore)
+        self.job_store: stores.JobStorer = sqlalchemy.JobStore(datastore)

scheduler/models/__init__.py

@@ -2,8 +2,9 @@
 from .boefje import Boefje, BoefjeMeta
 from .events import NormalizerMetaReceivedEvent, RawData, RawDataReceivedEvent
 from .health import ServiceHealth
+from .jobs import ScheduledJob, ScheduledJobORM
 from .normalizer import Normalizer
-from .ooi import OOI
+from .ooi import OOI, OOIORM, MutationOperationType, ScanProfileMutation
 from .organisation import Organisation
 from .plugin import Plugin
 from .queue import Filter, PrioritizedItem, PrioritizedItemORM, Queue

scheduler/models/jobs.py

@@ -0,0 +1,52 @@
+import datetime
+import uuid
+from typing import List, Optional
+
+from pydantic import BaseModel, Field
+from sqlalchemy import JSON, Boolean, Column, DateTime, String
+from sqlalchemy.orm import relationship
+
+from scheduler.utils import GUID
+
+from .base import Base
+from .queue import PrioritizedItem
+from .tasks import Task
+
+
+class ScheduledJob(BaseModel):
+    id: uuid.UUID
+    hash: str
+    enabled: bool
+    crontab: Optional[str]
+    scheduler_id: str
+    p_item: PrioritizedItem
+    tasks: List[Task] = []
+
+    next_check: Optional[datetime.datetime] = None
+
+    checked_at: Optional[datetime.datetime] = None
+
+    created_at: datetime.datetime = Field(
+        default_factory=datetime.datetime.utcnow)
+
+    modified_at: datetime.datetime = Field(
+        default_factory=datetime.datetime.utcnow)
+
+    class Config:
+        orm_mode = True
+
+
+class ScheduledJobORM(Base):
+    __tablename__ = "scheduled_jobs"
+
+    id = Column(GUID, primary_key=True, default=uuid.uuid4)
+    hash = Column(String, nullable=False)  # TODO: unique
+    enabled = Column(Boolean, nullable=False)
+    crontab = Column(String, nullable=True)
+    scheduler_id = Column(String, nullable=False)
+    p_item = Column(JSON, nullable=False)
+    tasks = relationship("TaskORM", back_populates="scheduled_job")
+
+    checked_at = Column(DateTime, nullable=True)
+    created_at = Column(DateTime, nullable=False)
+    modified_at = Column(DateTime, nullable=False)