Refactor trigger management and settings configuration

- Updated the Settings model to replace `trigger_ahead_time` with `trigger_workers`, allowing for configurable worker count for trigger processing.
- Added a new status `TRIGGERING` to the TriggerStatusEnum to better represent the trigger lifecycle.
- Enhanced the `trigger_cron` function to utilize the new worker configuration, enabling concurrent processing of due triggers.
- Refined the `create_next_triggers` function to calculate the next trigger time based on the cron expression, improving trigger scheduling logic.

Author: NiveditJain

state-manager/app/config/settings.py

@@ -12,7 +12,7 @@ class Settings(BaseModel):
     mongo_database_name: str = Field(default="exosphere-state-manager", description="MongoDB database name")
     state_manager_secret: str = Field(..., description="Secret key for API authentication")
     secrets_encryption_key: str = Field(..., description="Key for encrypting secrets")
-    trigger_ahead_time: int = Field(default=10, description="Time in minutes to trigger the graph ahead of the current time")
+    trigger_workers: int = Field(default=1, description="Number of workers to run the trigger cron")
 
     @classmethod
     def from_env(cls) -> "Settings":
@@ -21,7 +21,7 @@ def from_env(cls) -> "Settings":
             mongo_database_name=os.getenv("MONGO_DATABASE_NAME", "exosphere-state-manager"), # type: ignore
             state_manager_secret=os.getenv("STATE_MANAGER_SECRET"), # type: ignore
             secrets_encryption_key=os.getenv("SECRETS_ENCRYPTION_KEY"), # type: ignore
-            trigger_ahead_time=int(os.getenv("TRIGGER_AHEAD_TIME", 10)) # type: ignore
+            trigger_workers=int(os.getenv("TRIGGER_WORKERS", 1)) # type: ignore
         )
 
 

state-manager/app/models/trigger_models.py

@@ -11,6 +11,7 @@ class TriggerStatusEnum(str, Enum):
     FAILED = "FAILED"
     CANCELLED = "CANCELLED"
     TRIGGERED = "TRIGGERED"
+    TRIGGERING = "TRIGGERING"
 
 class CronTrigger(BaseModel):
     expression: str = Field(..., description="Cron expression for the trigger")

state-manager/app/tasks/trigger_cron.py

@@ -1,11 +1,14 @@
 from datetime import datetime
 from uuid import uuid4
 from app.models.db.trigger import DatabaseTriggers
-from app.models.trigger_models import TriggerStatusEnum
+from app.models.trigger_models import TriggerStatusEnum, TriggerTypeEnum
 from app.singletons.logs_manager import LogsManager
 from app.controller.trigger_graph import trigger_graph
 from app.models.trigger_graph_model import TriggerGraphRequestModel
 from pymongo import ReturnDocument
+from app.config.settings import get_settings
+import croniter
+import asyncio
 
 logger = LogsManager().get_logger()
 
@@ -16,7 +19,7 @@ async def get_due_triggers(cron_time: datetime) -> DatabaseTriggers | None:
             "trigger_status": TriggerStatusEnum.PENDING
         },
         {
-            "$set": {"trigger_status": TriggerStatusEnum.TRIGGERED}
+            "$set": {"trigger_status": TriggerStatusEnum.TRIGGERING}
         },
         return_document=ReturnDocument.AFTER
     )
@@ -30,13 +33,33 @@ async def call_trigger_graph(trigger: DatabaseTriggers):
         x_exosphere_request_id=str(uuid4())
     )
 
-async def create_next_triggers(trigger: DatabaseTriggers):
-    pass
+async def create_next_triggers(trigger: DatabaseTriggers, cron_time: datetime):
+    assert trigger.expression is not None
+    iter = croniter.croniter(trigger.expression, cron_time)
+    next_trigger_time = iter.get_next(datetime)
 
-async def trigger_cron():
-    cron_time = datetime.now()
-    logger.info(f"starting trigger_cron: {cron_time}")
+    await DatabaseTriggers(
+        type=TriggerTypeEnum.CRON,
+        expression=trigger.expression,
+        graph_name=trigger.graph_name,
+        namespace=trigger.namespace,
+        trigger_time=next_trigger_time,
+        trigger_status=TriggerStatusEnum.PENDING
+    ).insert()
+
+async def mark_as_triggered(trigger: DatabaseTriggers):
+    await DatabaseTriggers.get_pymongo_collection().update_one(
+        {"_id": trigger.id},
+        {"$set": {"trigger_status": TriggerStatusEnum.TRIGGERED}}
+    )
 
+async def handle_trigger(cron_time: datetime):
     while(trigger:= await get_due_triggers(cron_time)):
         await call_trigger_graph(trigger)
-        await create_next_triggers(trigger)
+        await create_next_triggers(trigger, cron_time)
+        await mark_as_triggered(trigger)
+
+async def trigger_cron():
+    cron_time = datetime.now()
+    logger.info(f"starting trigger_cron: {cron_time}")
+    await asyncio.gather(*[handle_trigger(cron_time) for _ in range(get_settings().trigger_workers)])

state-manager/app/tasks/verify_graph.py

@@ -127,30 +127,24 @@ async def create_crons(graph_template: GraphTemplate, old_triggers: list[Trigger
 
     crons_to_create = new_crons - old_crons
 
-    trigger_ahead_time = get_settings().trigger_ahead_time
     current_time = datetime.now()
-    limit_time = datetime.now() + timedelta(minutes=trigger_ahead_time)
 
     new_db_triggers = []
     for cron in crons_to_create:
         iter = croniter.croniter(cron.expression, current_time)
 
-        while(True):
-            next_trigger_time = iter.get_next(datetime)
+        next_trigger_time = iter.get_next(datetime)
 
-            # at least one event should be inserted for the cron
-            new_db_triggers.append(
-                DatabaseTriggers(
-                    type=TriggerTypeEnum.CRON,
-                    expression=cron.expression,
-                    graph_name=graph_template.name,
-                    namespace=graph_template.namespace,
-                    trigger_status=TriggerStatusEnum.PENDING,
-                    trigger_time=next_trigger_time
-                )
+        new_db_triggers.append(
+            DatabaseTriggers(
+                type=TriggerTypeEnum.CRON,
+                expression=cron.expression,
+                graph_name=graph_template.name,
+                namespace=graph_template.namespace,
+                trigger_status=TriggerStatusEnum.PENDING,
+                trigger_time=next_trigger_time
             )
-            if next_trigger_time > limit_time:
-                break
+        )
 
     if len(new_db_triggers) > 0:
         await DatabaseTriggers.insert_many(new_db_triggers)