kestra-io / kestra-io/plugin-fs

Add realtime WatchService-based trigger for local filesystem events

Open
#365 2 comments 0 reactions 0 assignees View on GitHub
area/plugin good first issue
Dominant language
Java
Stars
7
Forks
21
Avg merge
4d 4m
Merged PRs (30d)
8

Description

## Summary

`io.kestra.plugin.fs.local.Trigger` is polling-only: it wakes up on a fixed `interval` (default 60s) and re-lists the watched directory every time, even when nothing changed. Add a companion `io.kestra.plugin.fs.local.RealtimeTrigger` that registers a Java NIO `WatchService` on the target directory/directories and fires as soon as the filesystem reports a create/modify/delete event — no polling loop, no wasted list calls, near-instant reaction to file changes.

## Motivation

- Today, users who want fast reaction to new files must set a short `interval` (e.g. `PT5S`) on the polling `Trigger`, which means constant directory listing even when idle — wasted I/O and worker cycles.
- `WatchService` (`java.nio.file.WatchService`, part of the JDK since 7) delivers OS-level filesystem events (`inotify` on Linux, `ReadDirectoryChangesW` on Windows, `FSEvents`/kqueue on macOS) instead of re-scanning the directory.
- This plugin already has the realtime-trigger pattern for other transports (`io.kestra.plugin.fs.tcp.RealtimeTrigger`, `io.kestra.plugin.fs.udp.RealtimeTrigger`), both implementing `RealtimeTriggerInterface` with a reactive `Flux`. This issue extends that same pattern to local filesystem events, complementing the existing polling `io.kestra.plugin.fs.local.Trigger` (kept as-is for users who want interval-based batching/actions).

## Context

- Reference implementation to model the polling behavior after: `io.kestra.plugin.fs.local.Trigger` (`src/main/java/io/kestra/plugin/fs/local/Trigger.java`) — reuse its `on` (`CREATE_OR_UPDATE`/etc.), `regExp`, and `recursive` semantics where they make sense for a realtime source.
- Reference implementation to model the realtime/reactive plumbing after: `io.kestra.plugin.fs.tcp.RealtimeTrigger` and `io.kestra.plugin.fs.udp.RealtimeTrigger` (`RealtimeTriggerInterface`, `Flux`, `AtomicBoolean` stop flag, `Publisher`).
- Related closed issues on the polling trigger (do not need to be re-fixed here, just informative): [#208](https://github.com/kestra-io/plugin-fs/issues/208) (regex handling on MOVE), [#353](https://github.com/kestra-io/plugin-fs/issues/353) (count output), [#354](https://github.com/kestra-io/plugin-fs/issues/354) (MOVE returns pre-move path).

## API Reference

- **API**: `java.nio.file.WatchService` / `java.nio.file.WatchKey` / `java.nio.file.StandardWatchEventKinds` — JDK standard library, no external client.
- **Authentication**: none (local filesystem access only).
- **OS backing**: `inotify` (Linux), `ReadDirectoryChangesW` (Windows), polling-based `WatchService` fallback on macOS (JDK does not use native FSEvents for `WatchService`) — worth calling out as a known macOS caveat in the docs.

## Gradle Dependencies

None. `java.nio.file.WatchService` is part of the JDK standard library — no new dependency required.

## Plugin Structure

- **Repository**: `plugin-fs` (existing)
- **Namespace**: `io.kestra.plugin.fs.local`
- **New class**: `io.kestra.plugin.fs.local.RealtimeTrigger`
- **Sub-plugins**: `local` (existing sub-package, no new categories)
- **Categories**: unchanged — keep whatever `@PluginSubGroup` category `io.kestra.plugin.fs.local`'s `package-info.java` already declares.

> **Task class naming**: `RealtimeTrigger` matches the existing convention used by `fs.tcp` and `fs.udp` — do not suffix/prefix with `Local`.

## Suggested Tasks

1. Implement `io.kestra.plugin.fs.local.RealtimeTrigger` implementing `RealtimeTriggerInterface`, `TriggerOutput` (reuse `local.List.Output`/`local.models.File` shapes for parity with the polling trigger's output).
2. Register a `WatchService` on `from` (single directory); if `recursive` is true, register a `WatchKey` per subdirectory (walk tree on start, and register new subdirectories as `ENTRY_CREATE` events for directories arrive).
3. Map `StandardWatchEventKinds.ENTRY_CREATE` / `ENTRY_MODIFY` / `ENTRY_DELETE` to the same `on` filter (`CREATE`, `UPDATE`, `CREATE_OR_UPDATE`, `DELETE`) already used by `local.Trigger`.
4. Apply `regExp` filtering to matched file names, same semantics as `local.Trigger`.
5. Emit one `Execution` per matched event via the `Flux`/`Publisher` pattern from `tcp.RealtimeTrigger`; stop cleanly on trigger cancel/kill (close the `WatchService`, honor the `AtomicBoolean` stop flag).
6. Write unit + integration tests (create/modify/delete a file in a temp dir, assert an execution fires within a bounded time; test `recursive` picking up a new subdirectory).
7. Add/extend `package-info.java` for `io.kestra.plugin.fs.local` if a new category is needed (none expected — reuse existing).
8. Add YAML examples and plugin documentation; document the macOS `WatchService` polling-fallback caveat.

## YAML Examples

### Example 1 — React immediately when a file is created or updated in a directory

```yaml
id: local_realtime_trigger
namespace: company.team

triggers:
- id: watch
type: io.kestra.plugin.fs.local.RealtimeTrigger
from: /data/incoming
on: CREATE_OR_UPDATE

tasks:
- id: log_file
type: io.kestra.plugin.core.log.Log
message: "New file: {{ trigger.file.path }}"
```

### Example 2 — Watch recursively and filter by extension

```yaml
id: local_realtime_recursive
namespace: company.team

triggers:
- id: watch_csv
type: io.kestra.plugin.fs.local.RealtimeTrigger
from: /data/incoming
recursive: true
regExp: ".*\\.csv$"

tasks:
- id: process
type: io.kestra.plugin.core.log.Log
message: "CSV ready: {{ trigger.file.path }}"
```

### Example 3 — React to deletions for cleanup bookkeeping

```yaml
id: local_realtime_delete
namespace: company.team

triggers:
- id: watch_deletes
type: io.kestra.plugin.fs.local.RealtimeTrigger
from: /data/incoming
on: DELETE

tasks:
- id: log_removed
type: io.kestra.plugin.core.log.Log
message: "File removed: {{ trigger.file.path }}"
```

## Acceptance Criteria

### Functional
- [ ] `RealtimeTrigger` fires an execution as soon as a filesystem event is reported, no polling `interval`
- [ ] `recursive` picks up events from subdirectories, including subdirectories created after the trigger starts
- [ ] `regExp` and `on` filters behave the same as the polling `local.Trigger`
- [ ] `WatchService`/`WatchKey`s are closed and no thread is leaked when the trigger stops/is killed
- [ ] Unit + integration tests pass (`./gradlew test`)
- [ ] Build passes (`./gradlew build`)

### Kestra Plugin Coding Standards
- [ ] All new properties use `Property` — no legacy `@PluginProperty(dynamic = true)` on new code
- [ ] Every property and output has a `@Schema` annotation
- [ ] Trigger class carries the five mandatory Lombok annotations (`@SuperBuilder`, `@ToString`, `@EqualsAndHashCode`, `@Getter`, `@NoArgsConstructor`)
- [ ] Logging via `runContext.logger()` only
- [ ] All `Property` fields support Kestra expression language (template rendering)

### Documentation & Structure
- [ ] `@Plugin(examples = ...)` entries each set `full = true` with a complete runnable flow (id + namespace + triggers/tasks)
- [ ] `package-info.java` category unchanged/consistent for `io.kestra.plugin.fs.local`
- [ ] Docs call out the macOS `WatchService` polling-fallback caveat

---
*[View as Artifact](https://claude.ai/artifact/PWNirRoCWQWTNrPhSyU5oh)*

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with src/main/java/io/kestra/plugin/fs/local/Trigger.java and the tcp/udp RealtimeTrigger implementations to understand filtering, reactive plumbing, and lifecycle handling. Add the local RealtimeTrigger plus the requested unit and integration tests for filesystem events, recursion, and cleanup, then run ./gradlew test and ./gradlew build. Done includes YAML examples and documentation covering the macOS WatchService fallback.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend, operating-systems
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Clearly specified
Newbie friendliness
42/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.