feat(processing_engine): Add cron plugins and triggers to the processing engine. #25852

jacksonrnewhouse · 2025-01-17T01:03:06Z

This adds the CronSchedule plugin type to the processing engine, and an accompanying variant of TriggerSpecificationDefinition.

When a cron schedule runs it looks at the current time to construct an iterator, using the cron library. Calls to the plugin proceed sequentially, and there is not currently any catchup. The time passed down to the plugin is the time the cron schedule triggered for, rather than the current time.

I've added an endpoint to test cron plugins, similar to the existing one for the wal_plugin.

Some semantics questions I'd appreciate feedback on:

Right now we will fall behind if the plugin consistently takes longer to run than the time between triggers. Do we want to have a different default or expose this as an option?
The cron library we're using includes both second and year fields. The year fields allow for a cron schedule that ends. Right now the code exits, which gives an error if you disable. Should it be marked as disabled? Something else?
Should I hook this into the TimeProvider?

I'm planning to write up tests tomorrow, but wanted to get the PR published.

pauldix

Some comments and changes. On your questions:

Right now we will fall behind if the plugin consistently takes longer to run than the time between triggers. Do we want to have a different default or expose this as an option?

I think it should be an option. But this means that every execution should spawn its own background task? That way if they've specified that the behavior is noskip, multiple executions of the trigger should be able to run at the same time.

The cron library we're using includes both second and year fields. The year fields allow for a cron schedule that ends. Right now the code exits, which gives an error if you disable. Should it be marked as disabled? Something else?

Yeah, they give one that has an end on the schedule, it makes sense that we'd mark it as disabled.

Should I hook this into the TimeProvider?

Yes, if we need to get the "now" time.

pauldix · 2025-01-17T16:00:33Z

influxdb3/src/commands/test.rs

@@ -1,7 +1,7 @@
 use crate::commands::common::{InfluxDb3Config, SeparatedKeyValue, SeparatedList};
 use anyhow::Context;
 use hashbrown::HashMap;
-use influxdb3_client::plugin_development::WalPluginTestRequest;
+use influxdb3_client::plugin_development::{CronPluginTestRequest, WalPluginTestRequest};


It should be called Schedule plugin, rather than cron. Even though the cron string is what we're using, we want a more user friendly name.

pauldix · 2025-01-17T17:23:55Z

influxdb3/src/commands/test.rs

+    influxdb3_config: InfluxDb3Config,
+    /// If given pass this map of string key/value pairs as input arguments
+    #[clap(long = "input-arguments")]
+    pub input_arguments: Option<SeparatedList<SeparatedKeyValue<String, String>>>,


input_arguments should be on the trigger, not on the plugin itself.

I'm mirroring the behavior of WalPluginConfig above, where this gets passed up to the server and then passed down into python via the final argument of the process_scheduled_call() method.

Ah, this is for the test endpoint?

Yeah, this is the CLI arguments, mirroring the test wal_plugin.

The existing plugin and trigger creation APIs were sufficient for this, once I added a new TriggerSpecificationDefinition.

influxdb3_client/src/lib.rs

pauldix · 2025-01-17T18:01:29Z

influxdb3_processing_engine/src/plugins.rs

+                Arc::clone(&plugin.query_executor),
+                &plugin.trigger_definition.trigger_arguments,
+            )
+            .expect("should be able to run trigger");


any reason this might error, can we catch this and log it?

Yeah, any plugin error would trigger, switching to just have it exit with ?.

pauldix · 2025-01-17T18:02:27Z

influxdb3_processing_engine/src/plugins.rs

+            db_schema: Arc<DatabaseSchema>,
+        ) -> Result<(), Error> {
+            let Some(trigger_time) = self.next_trigger_time else {
+                return Err(anyhow!("running a cron trigger that is finished.").into());


This language could be a little more clear. Not sure what finished means in this context.

pauldix · 2025-01-17T18:06:26Z

influxdb3_py_api/src/system_py.rs

+        let local_api = api.into_pyobject(py)?;
+
+        // turn args into an optional dict to pass into python
+        let args = args.as_ref().map(|args| {


since we also have this block in the execute_wal_plugin, can we pull into a function?

factored it out.

…ing engine.

…n', use TimeProvider.

pauldix · 2025-01-18T12:15:07Z

influxdb3_processing_engine/src/plugins.rs

+                                break;
+                            }
+                            Some(PluginEvent::WriteWalContents(_)) => {
+                                debug!("ignoring wal contents in cron plugin.")


This is unexpected so maybe should be a warn? Can fix in a followup.

Actually this will be happening. Right now we send the Wal contents to all plugins, without doing any up-front work to figure out whether they actually need the message. For WalRows, this means we don't check if there is data on a relevant table, and for scheduled plugins all of it is irrelevant.

pauldix · 2025-01-18T12:18:15Z

Going to merge this in so I can stack my new plugin type on it.

jacksonrnewhouse requested review from pauldix and mgattozzi January 17, 2025 01:03

pauldix reviewed Jan 17, 2025

View reviewed changes

feat(processing_engine): Add cron plugins and triggers to the process…

ec64143

…ing engine.

jacksonrnewhouse force-pushed the cron_plugin branch 2 times, most recently from ac6ef95 to e59a2f1 Compare January 18, 2025 03:23

jacksonrnewhouse added 2 commits January 17, 2025 19:26

feat(processing_engine): switch from 'cron plugin' to 'schedule plugi…

7cd494a

…n', use TimeProvider.

feat(processing_engine): add test for test scheduled plugin.

f3baede

jacksonrnewhouse force-pushed the cron_plugin branch from e59a2f1 to f3baede Compare January 18, 2025 03:26

pauldix approved these changes Jan 18, 2025

View reviewed changes

pauldix merged commit 1d8d3d6 into main Jan 18, 2025
13 checks passed

pauldix deleted the cron_plugin branch January 18, 2025 12:18

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(processing_engine): Add cron plugins and triggers to the processing engine. #25852

feat(processing_engine): Add cron plugins and triggers to the processing engine. #25852

jacksonrnewhouse commented Jan 17, 2025

pauldix left a comment

pauldix Jan 17, 2025

pauldix Jan 17, 2025

jacksonrnewhouse Jan 17, 2025

pauldix Jan 17, 2025

jacksonrnewhouse Jan 18, 2025

jacksonrnewhouse Jan 18, 2025

pauldix Jan 17, 2025

jacksonrnewhouse Jan 17, 2025

pauldix Jan 17, 2025

pauldix Jan 17, 2025

jacksonrnewhouse Jan 18, 2025

pauldix Jan 18, 2025

jacksonrnewhouse Jan 21, 2025

pauldix commented Jan 18, 2025

feat(processing_engine): Add cron plugins and triggers to the processing engine. #25852

feat(processing_engine): Add cron plugins and triggers to the processing engine. #25852

Conversation

jacksonrnewhouse commented Jan 17, 2025

pauldix left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pauldix commented Jan 18, 2025