Scheduled Events
Understanding the Bot/Automation Scheduled Events Capability
InsightCloudSec Bots can spawn jobs that run at specified times with specified arguments. These jobs are called scheduled events.
Database & Backend Naming Conventions
Rebranding for DivvyCloud, now InsightCloudSec is ongoing. Logos, URLs, text, and images may reference either InsightCloudSec or DivvyCloud. This obviously extends to the internal portion of our product as well.
The most important thing to note is that the product functionality has remained the same. If you have any questions or concerns reach out to us through the Customer Support Portal.
Registration
A job is registered as a scheduled event using the register
method of the ScheduledEventManager
class, which acts as a class decorator. For example, this is how the StartResourceJob
scheduled event is registered in InsightCloudSec:
from DivvyWorkers.Processors.ScheduledEvents import ScheduledEventManager
@ScheduledEventManager.register('divvy.start_resource')
class StartResourceJob(MultiResourceScheduledEventJob):
...
The scheduled event can then be referred to using the unique identifier defined in the register
decorator, in this case divvy.start_resource
.
Here is an example of how to spawn a scheduled event within a bot action:
from DivvyBotfactory.scheduling import ScheduledEventTracker
@registry.action(
uid='divvy.action.start_resource_example',
bulk_action=True,
accepts_complement=True,
...
)
def start_resource_example(bot, settings, matches, non_matches):
with ScheduledEventTracker() as context:
for resource in matches:
context.schedule_bot_event(
bot=bot, resource=resource,
description='Start a resource.',
event_type='divvy.start_resource'
schedule_data=schedule.Once(when=datetime.utcnow() + timedelta(hours=12))
)
Dynamic Loading and Unloading
When developing a plugin that InsightCloudSec dynamically loads and unloads, it is necessary to unload the plugin’s scheduled events. This can be conveniently done using the ScheduledEventRegistryWrapper
class, which requires only a minor variation upon the pattern shown above.
An example:
from DivvyWorkers.Processors.ScheduledEvents import ScheduledEventRegistryWrapper
# Initialize the registry wrapper
events = ScheduledEventRegistryWrapper()
# Register a scheduled event job
events.register('divvy.start_resource')
class StartResourceJob(MultiResourceScheduledEventJob):
...
# Handle plugin loading/unloading
def load():
events.load()
def unload():
events.unload()
A scheduled event registered this way is created and referred to in the same way as before. The only difference is in the specific syntax of how that event is registered.
Updated over 1 year ago