Idempotent start time with CronWorkflows

[Mathias1701]

Hi there!
Compared to Airflow I see no option to implement idempotent behaviour of CronWorkflows. (data_interval_start, data_interval_end)
To be more specific:
Lets assume I would like to pull data from a database with given timeframe (a -> b) into my MinIO storage.
a and b can be derived from the moment, when the executor starts the specific step. But what happens, when a retry is needed? The timeframe gets moved und unprecise and so I will loose data in the end.
So is there an option that Argo Workflows provides the 'calculated' trigger moment as input parameter? And so the retry would get the same trigger moment as parameter again?

[Joibel]

Relying on start time of a step is a risky behavior at all times. I would not recommend doing this anyway, and instead calculate a suitable start time based upon other available information.

If you'd still like to rely on it you could run a step to store this. You could use a short lived memoized step to store this information and output it as a parameter, and then the actual step you'd like to use this timestamp for would take it as an input parameter. As it is memoized the next time it runs it will skip it and output the same timestamp for the next step to consume.

[agilgur5]

Yea, the general practice I've seen in such batch jobs is typically to start from the "last processed time" through "24 hours after it" (if your interval is daily, for example).

Or you can do something more precise than your current method, such as "yesterday at 2am" through "today at 2am" -- within the course of a day, it will be the same.

Typically you may also have environment variables to override the start and end time as well. Then if you need a manual rerun, you can always override based on the gap in your data.

[agilgur5]

Also if you really want to do what you're saying, you could use the Workflow's metadata.creationTimestamp which is available as the workflow.creationTimestamp variable.

For CronWorkflows, you can also use the annotation workflows.argoproj.io/scheduled-time which is available as the workflow.scheduledTime variable.