Merge pull request #133 from blacklanternsecurity/rename-watchdog

Rename watchdog -> worker

Author: TheTechromancer

README.md

@@ -48,7 +48,7 @@ bbctl server start
 
 ## Deploy with Helm (Kubernetes)
 
-BBOT Server can be deployed to Kubernetes using its official Helm chart. The chart deploys the API server, watchdog, MongoDB, and Redis.
+BBOT Server can be deployed to Kubernetes using its official Helm chart. The chart deploys the API server, worker, MongoDB, and Redis.
 
 ### Quick Start
 

bbot_server/applets/base.py

@@ -49,14 +49,14 @@ def decorator(fn):
     return decorator
 
 
-def watchdog_task(**kwargs):
+def worker_task(**kwargs):
     """
-    Decorate your applet method with this to make it a watchdog task
+    Decorate your applet method with this to make it a worker task
     """
 
     def decorator(fn):
         fn._kwargs = kwargs
-        fn._watchdog_task = True
+        fn._worker_task = True
         return fn
 
     return decorator
@@ -240,8 +240,8 @@ async def _native_setup(self):
             self.task_broker = await self.message_queue.make_taskiq_broker()
             await self.task_broker.startup()
 
-        # register watchdog tasks
-        await self.register_watchdog_tasks(self.task_broker)
+        # register worker tasks
+        await self.register_worker_tasks(self.task_broker)
 
         if self.name != "Root Applet":
             try:
@@ -290,15 +290,15 @@ async def reconcile_all_indexes(self):
             diff = compute_index_diff(desired, desired_text, existing, existing_text)
             await apply_index_diff(collection, diff, existing)
 
-    async def register_watchdog_tasks(self, broker):
-        # register watchdog tasks
+    async def register_worker_tasks(self, broker):
+        # register worker tasks
         methods = {name: member for name, member in getmembers(self) if callable(member)}
         for method_name, method in methods.items():
             # handle case where tasks have already been registered
             method = getattr(method, "original_func", method)
 
-            _watchdog_task = getattr(method, "_watchdog_task", None)
-            if _watchdog_task is None:
+            _worker_task = getattr(method, "_worker_task", None)
+            if _worker_task is None:
                 continue
             kwargs = getattr(method, "_kwargs", {})
             # crontab handling

bbot_server/modules/emails/emails_api.py

@@ -1,4 +1,4 @@
-# from bbot_server.watchdogs.emails import EmailWatchdog
+# from bbot_server.workers.emails import EmailWorker
 from bbot_server.applets.base import BaseApplet, api_endpoint, BaseModel, Field
 
 
@@ -7,7 +7,7 @@ class EmailsApplet(BaseApplet):
     description = "emails discovered during scans"
     # watched_events = ["EMAIL_ADDRESS"]
     route_prefix = ""
-    # watchdogs = [EmailWatchdog]
+    # workers = [EmailWorker]
     attach_to = "assets"
 
     class AssetFields(BaseModel):

bbot_server/modules/events/events_api.py

@@ -29,7 +29,7 @@ async def insert_event(self, event: Event):
         Insert a BBOT event into the asset database
         """
         # publish event to the message queue
-        # it will be picked up by the watchdog and ingested
+        # it will be picked up by the worker and ingested
         await self.root.message_queue.publish_event(event)
 
     @api_endpoint("/get/{uuid}", methods=["GET"], summary="Get an event by its UUID")

bbot_server/modules/server/server_cli.py

@@ -28,8 +28,8 @@ def start(
         api_only: Annotated[
             bool, Option("--api-only", "-a", help="Only start the REST API, without Docker Compose")
         ] = False,
-        watchdog_only: Annotated[
-            bool, Option("--watchdog-only", "-w", help="Only start the watchdog, without Docker Compose")
+        worker_only: Annotated[
+            bool, Option("--worker-only", "-w", help="Only start the worker, without Docker Compose")
         ] = False,
         listen: Annotated[str, Option("--listen", "-l", help="Listen address", metavar="IP_ADDRESS")] = "127.0.0.1",
         port: Annotated[int, Option("--port", "-p", help="Port to run the server on", metavar="PORT")] = 8807,
@@ -60,30 +60,30 @@ def start(
                 workers=1,
             )
 
-        elif watchdog_only:
-            print("Starting watchdog")
+        elif worker_only:
+            print("Starting worker")
 
-            async def run_watchdog():
+            async def run_worker():
                 try:
                     from bbot_server import BBOTServer
-                    from bbot_server.watchdog import BBOTWatchdog
+                    from bbot_server.worker import BBOTWorker
 
                     bbot_server = BBOTServer()
                     await bbot_server.setup()
 
-                    watchdog = BBOTWatchdog(bbot_server)
-                    await watchdog.start()
-                    print("Watchdog successfully started")
+                    worker = BBOTWorker(bbot_server)
+                    await worker.start()
+                    print("Worker successfully started")
 
                     # sleep for infinity
                     event = asyncio.Event()
                     await event.wait()
 
                 except KeyboardInterrupt:
                     with suppress(Exception):
-                        await watchdog.stop()
+                        await worker.stop()
 
-            asyncio.run(run_watchdog())
+            asyncio.run(run_worker())
 
         else:
             # initialize the config if not already

bbot_server/watchdog/__init__.py

@@ -0,0 +1,3 @@
+from .worker import BBOTWorker
+
+__all__ = ["BBOTWorker"]

bbot_server/worker/__init__.py

@@ -12,7 +12,7 @@
 from bbot_server.modules.activity.activity_models import Activity
 
 
-class BBOTWatchdog:
+class BBOTWorker:
     """
     Contains:
         - taskiq worker
@@ -40,9 +40,9 @@ async def startup(state: TaskiqState) -> None:
 
         await self.broker.startup()
 
-        # register watchdog tasks
+        # register worker tasks
         for app in self.bbot_server.all_child_applets(include_self=True):
-            await app.register_watchdog_tasks(self.broker)
+            await app.register_worker_tasks(self.broker)
 
         # taskiq worker tasks
         self.taskiq_worker_task = asyncio.create_task(run_receiver_task(self.broker))
@@ -157,7 +157,7 @@ async def _get_or_create_asset(self, host: str, event: Event = None, parent_acti
         return asset, activities
 
     async def stop(self) -> None:
-        self.log.info("Stopping watchdog")
+        self.log.info("Stopping worker")
         await self.bbot_server.message_queue.unsubscribe(self.event_listener)
         self.taskiq_worker_task.cancel()
         self.taskiq_scheduler_task.cancel()

bbot_server/watchdog/worker.py bbot_server/worker/worker.pybbot_server/watchdog/worker.py renamed to bbot_server/worker/worker.py

@@ -25,9 +25,9 @@ services:
       - mongodb
       - redis
 
-  watchdog:
+  worker:
     <<: *bbot-server-base
-    command: ["bbctl", "server", "start", "--watchdog-only"]
+    command: ["bbctl", "server", "start", "--worker-only"]
     depends_on:
       server:
         condition: service_healthy

compose.yml

@@ -1,6 +1,6 @@
 apiVersion: v2
 name: bbot-server
-description: Helm chart for BBOT Server (API + Watchdog)
+description: Helm chart for BBOT Server (API + Worker)
 type: application
 version: 0.1.0
 appVersion: "0.1.0"