Archive Processed Output Files on a Local Filesystem Mount

Batch jobs that write to a shared local disk (a bind-mounted volume, an on-prem worker's local storage, an edge server) tend to leave two kinds of mess behind: finished output files that need to move somewhere durable, and partial .tmp files from runs that got interrupted. This blueprint handles both with the io.kestra.plugin.fs.local plugin: it watches the processing directory for finished files, archives each one into a folder named for the day it landed, and sweeps any stale scratch files out of the way, all without a cron job or a shell script babysitting the disk.

How it works

1. io.kestra.plugin.fs.local.Trigger polls /data/pipeline/output every two minutes (interval: PT2M), matching new files with regExp: ".*\\.csv$" and firing on CREATE.
2. log_new_files (io.kestra.plugin.core.log.Log) records {{ trigger.count }}, the number of files the trigger picked up on this poll.
3. archive_each_file (io.kestra.plugin.core.flow.ForEach) iterates over {{ trigger.files }}. For each file, move_to_dated_archive (io.kestra.plugin.fs.local.Move) reads the source path with {{ taskrun.value | jq('.path') | first }} and moves it to /data/pipeline/archive/{{ now(dateFormat='yyyy/MM/dd') }}/, keeping the original filename via {{ taskrun.value | jq('.name') | first }}.
4. purge_stale_scratch_files (io.kestra.plugin.fs.local.Delete) then removes any .tmp files still sitting in /data/pipeline/output, recursively, with errorOnMissing: false so a clean directory never fails the flow.
5. notify_archive_complete posts a Slack summary; the errors block posts a separate alert if any step fails, most commonly because the archive destination already has a same-named file (Move defaults to overwrite: false) or the worker's allowed-paths configuration doesn't cover the directory.

What you get

• Files archived within two minutes of landing, with no separate scheduler or shell script.
• A dated archive layout (yyyy/MM/dd) that makes retention and manual lookups straightforward.
• Automatic cleanup of .tmp leftovers from interrupted runs, so scratch space doesn't quietly grow.
• A clear Slack trail of what got archived and when, plus a distinct alert if a move or delete fails.

Who it's for

• Data engineering teams running batch jobs (ETL scripts, report generators, ML training jobs) that write to a local or bind-mounted disk.
• Platform teams who inherited a directory full of output files and partial .tmp artifacts with no retention process.
• Anyone replacing an ad hoc cron mv and find -delete script with something observable and alertable.

Why orchestrate this with Kestra

A cron job can move files on a schedule, but it can't react the moment a file appears, doesn't give you a run history to check when something goes wrong, and silently swallows errors like a destination conflict. Kestra's fs.local.Trigger reacts to file creation directly, ForEach gives every file its own tracked task run, and the errors block guarantees a failed move or delete surfaces in Slack instead of getting lost in a cron log nobody reads.

Prerequisites

• A Kestra worker with /data/pipeline/output and /data/pipeline/archive reachable on its local filesystem (a bind mount or local disk).
• The Kestra instance administrator must add both directories to allowed-paths for io.kestra.plugin.fs.local in the server configuration; without it, every local filesystem task in this flow is rejected.
• A Slack incoming webhook for notifications.

Secrets

• SLACK_WEBHOOK_URL: Slack incoming webhook URL.

Quick start

1. Add SLACK_WEBHOOK_URL to your Kestra namespace.
2. Confirm /data/pipeline/output and /data/pipeline/archive are on the worker's allowed-paths list.
3. Drop a .csv file into /data/pipeline/output and confirm the trigger fires within two minutes.
4. Check /data/pipeline/archive/<today's date> for the moved file, and confirm the Slack notification arrived.

How to extend

• Swap the regExp filter for the extension your pipeline actually writes (Parquet, JSON, ION).
• Add a io.kestra.plugin.core.storage.Size style size check, or read {{ taskrun.value | jq('.size') | first }} directly from the trigger payload, to skip zero-byte files before archiving.
• Replace the local archive destination with a call to io.kestra.plugin.fs.sftp.Upload or a cloud object store task to ship the archive off the host entirely.
• Add a second Schedule trigger that runs purge_stale_scratch_files independently on a nightly cadence, in case no new files arrive to trigger a cleanup for several days.

Links

• Plugin docs: https://kestra.io/plugins/plugin-fs
• File access concepts: https://kestra.io/docs/concepts/file-access