refactor(k8saudit): use event-driven file watching with fsnotify

Replace polling-based file watching with fsnotify for better efficiency.
Key changes:
- Use fsnotify to watch parent directory (per maintainer recommendation)
- Rename scheme from tail:// to file://
- Remove watchPollIntervalMs config (no longer needed)
- Rename test package from tail to filewatch

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: RichardoC

plugins/k8saudit/README.md

@@ -9,7 +9,7 @@ Audit events are logged by the API server when almost every cluster management t
 
 This plugin supports consuming Kubernetes Audit Events coming from the [Webhook backend](https://kubernetes.io/docs/tasks/debug/debug-cluster/audit/#webhook-backend) or from a file. For webhooks, the plugin embeds a web server that listens on a configurable port and accepts POST requests. The posted JSON object comprises one or more events. The web server of the plugin can be configured as part of the plugin's init configuration and open parameters. For files, the plugin expects content to be in [JSONL format](https://jsonlines.org/), where each line represents a JSON object, containing one or more audit events.
 
-The expected way of using the plugin with Falco is through a Webhook. File reading support can be used with Stratoshark or testing and development. The `tail://` scheme enables continuous file watching with log rotation support, useful for reading audit logs written to disk by the API server.
+The expected way of using the plugin with Falco is through a Webhook. File reading support can be used with Stratoshark or testing and development. The `file://` scheme enables continuous file watching with log rotation support, useful for reading audit logs written to disk by the API server.
 
 ## Capabilities
 
@@ -131,12 +131,11 @@ load_plugins: [k8saudit, json]
 - `maxEventSize`: Maximum size of single audit event (Default: 262144)
 - `webhookMaxBatchSize`: Maximum size of incoming webhook POST request bodies (Default: 12582912)
 - `useAsync`: If true, then async extraction optimization is enabled (Default: true)
-- `watchPollIntervalMs`: Polling interval in milliseconds when watching a file with the `tail://` scheme (Default: 250)
 
 **Open Parameters**:
 - `http://<host>:<port>/<endpoint>`: Opens an event stream by listening on an HTTP web server
 - `https://<host>:<port>/<endpoint>`: Opens an event stream by listening on an HTTPS web server
-- `tail://<filepath>`: Opens an event stream by continuously watching a file for new audit events, similar to `tail -f`. Handles log rotation (inode changes) and file truncation automatically. Example: `tail:///var/log/kube-apiserver/audit.log`
+- `file://<filepath>`: Opens an event stream by continuously watching a file for new audit events. Handles log rotation automatically. Example: `file:///var/log/kube-apiserver/audit.log`
 - `no scheme`: Opens an event stream by reading the events from a file on the local filesystem. The params string is interpreted as a filepath
 
 

plugins/k8saudit/go.mod

@@ -5,6 +5,7 @@ go 1.15
 require (
 	github.com/alecthomas/jsonschema v0.0.0-20220216202328-9eeeec9d044b
 	github.com/falcosecurity/plugin-sdk-go v0.8.3
+	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/iancoleman/orderedmap v0.3.0 // indirect
 	github.com/valyala/fastjson v1.6.4
 )

plugins/k8saudit/go.sum

@@ -5,6 +5,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/falcosecurity/plugin-sdk-go v0.8.3 h1:KsX7qt83dzC57qcNpZKaBrCjTXqpXgvxDcEXs6Z5sHI=
 github.com/falcosecurity/plugin-sdk-go v0.8.3/go.mod h1:gEgxjvuopv5VF4wc8s0EHnmT9qrIKBtcJVBnRlEPU1A=
+github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k=
+github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
 github.com/geraldcombs/fastjson v0.0.0-20250801170450-bf39244e60b8 h1:S2FAMWjJKPRR9fvtgYVWQ5joNsl0qQoRxmxYHKDDtx4=
 github.com/geraldcombs/fastjson v0.0.0-20250801170450-bf39244e60b8/go.mod h1:CLCAqky6SMuOcxStkYQvblddUtoRxhYMGLrsQns1aXY=
 github.com/iancoleman/orderedmap v0.0.0-20190318233801-ac98e3ecb4b0/go.mod h1:N0Wam8K1arqPXNWjMo21EXnBPOPp36vB07FNRdD2geA=
@@ -25,6 +27,8 @@ github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2
 github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
 github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
 github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
+golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
+golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

plugins/k8saudit/pkg/k8saudit/config.go

@@ -24,7 +24,6 @@ type PluginConfig struct {
 	UseAsync            bool   `json:"useAsync"             jsonschema:"title=Use async extraction,description=If true then async extraction optimization is enabled (Default: true),default=true"`
 	MaxEventSize        uint64 `json:"maxEventSize"         jsonschema:"title=Maximum event size,description=Maximum size of single audit event (Default: 262144),default=262144"`
 	WebhookMaxBatchSize uint64 `json:"webhookMaxBatchSize"  jsonschema:"title=Maximum webhook request size,description=Maximum size of incoming webhook POST request bodies (Default: 12582912),default=12582912"`
-	WatchPollIntervalMs uint64 `json:"watchPollIntervalMs"  jsonschema:"title=Watch poll interval,description=Polling interval in milliseconds when watching a file with tail:// scheme (Default: 250),default=250"`
 }
 
 // Resets sets the configuration to its default values
@@ -39,5 +38,4 @@ func (k *PluginConfig) Reset() {
 	// The following values have been chosen by increasing by ~20% the default
 	// values of the K8S docs
 	k.WebhookMaxBatchSize = 12 * 1024 * 1024
-	k.WatchPollIntervalMs = 250
 }

…/k8saudit/pkg/k8saudit/tail/tail_test.go …pkg/k8saudit/filewatch/filewatch_test.goplugins/k8saudit/pkg/k8saudit/tail/tail_test.go renamed to plugins/k8saudit/pkg/k8saudit/filewatch/filewatch_test.go plugins/k8saudit/pkg/k8saudit/tail/tail_test.go renamed to plugins/k8saudit/pkg/k8saudit/filewatch/filewatch_test.go

@@ -15,7 +15,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package tail_test
+package filewatch_test
 
 import (
 	"os"
@@ -32,30 +32,10 @@ const testAuditEvent = `{"kind":"Event","apiVersion":"audit.k8s.io/v1","level":"
 func newTestPlugin() *k8saudit.Plugin {
 	p := &k8saudit.Plugin{}
 	p.Config.Reset()
-	p.Config.WatchPollIntervalMs = 50
 	return p
 }
 
-func TestOpenFileTail_ReadsExistingContent(t *testing.T) {
-	tmpDir := t.TempDir()
-	filePath := filepath.Join(tmpDir, "audit.log")
-
-	if err := os.WriteFile(filePath, []byte(testAuditEvent+"\n"), 0644); err != nil {
-		t.Fatal(err)
-	}
-
-	p := newTestPlugin()
-	inst, err := p.OpenFileTail(filePath)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer inst.(sdk.Closer).Close()
-
-	// Allow time for initial read
-	time.Sleep(100 * time.Millisecond)
-}
-
-func TestOpenFileTail_DetectsNewContent(t *testing.T) {
+func TestOpenFileWatch_DetectsNewContent(t *testing.T) {
 	tmpDir := t.TempDir()
 	filePath := filepath.Join(tmpDir, "audit.log")
 
@@ -64,13 +44,12 @@ func TestOpenFileTail_DetectsNewContent(t *testing.T) {
 	}
 
 	p := newTestPlugin()
-	inst, err := p.OpenFileTail(filePath)
+	inst, err := p.OpenFileWatch(filePath)
 	if err != nil {
 		t.Fatal(err)
 	}
 	defer inst.(sdk.Closer).Close()
 
-	// Write new content after opening
 	time.Sleep(100 * time.Millisecond)
 	f, err := os.OpenFile(filePath, os.O_APPEND|os.O_WRONLY, 0644)
 	if err != nil {
@@ -79,11 +58,10 @@ func TestOpenFileTail_DetectsNewContent(t *testing.T) {
 	f.WriteString(testAuditEvent + "\n")
 	f.Close()
 
-	// Allow time for polling to detect new content
 	time.Sleep(200 * time.Millisecond)
 }
 
-func TestOpenFileTail_HandlesRotation(t *testing.T) {
+func TestOpenFileWatch_HandlesRotation(t *testing.T) {
 	tmpDir := t.TempDir()
 	filePath := filepath.Join(tmpDir, "audit.log")
 
@@ -92,51 +70,23 @@ func TestOpenFileTail_HandlesRotation(t *testing.T) {
 	}
 
 	p := newTestPlugin()
-	inst, err := p.OpenFileTail(filePath)
+	inst, err := p.OpenFileWatch(filePath)
 	if err != nil {
 		t.Fatal(err)
 	}
 	defer inst.(sdk.Closer).Close()
 
 	time.Sleep(100 * time.Millisecond)
 
-	// Simulate rotation: remove and recreate with new content
 	os.Remove(filePath)
 	if err := os.WriteFile(filePath, []byte(testAuditEvent+"\n"), 0644); err != nil {
 		t.Fatal(err)
 	}
 
-	// Allow time for polling to detect rotation
-	time.Sleep(200 * time.Millisecond)
-}
-
-func TestOpenFileTail_HandlesTruncation(t *testing.T) {
-	tmpDir := t.TempDir()
-	filePath := filepath.Join(tmpDir, "audit.log")
-
-	if err := os.WriteFile(filePath, []byte(testAuditEvent+"\n"+testAuditEvent+"\n"), 0644); err != nil {
-		t.Fatal(err)
-	}
-
-	p := newTestPlugin()
-	inst, err := p.OpenFileTail(filePath)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer inst.(sdk.Closer).Close()
-
-	time.Sleep(100 * time.Millisecond)
-
-	// Truncate the file
-	if err := os.WriteFile(filePath, []byte(testAuditEvent+"\n"), 0644); err != nil {
-		t.Fatal(err)
-	}
-
-	// Allow time for polling to detect truncation
 	time.Sleep(200 * time.Millisecond)
 }
 
-func TestOpen_TailScheme(t *testing.T) {
+func TestOpen_FileScheme(t *testing.T) {
 	tmpDir := t.TempDir()
 	filePath := filepath.Join(tmpDir, "audit.log")
 
@@ -145,7 +95,7 @@ func TestOpen_TailScheme(t *testing.T) {
 	}
 
 	p := newTestPlugin()
-	inst, err := p.Open("tail://" + filePath)
+	inst, err := p.Open("file://" + filePath)
 	if err != nil {
 		t.Fatal(err)
 	}

plugins/k8saudit/pkg/k8saudit/source.go

@@ -26,23 +26,17 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"path/filepath"
 	"sort"
 	"strings"
-	"syscall"
 	"time"
 
 	"github.com/falcosecurity/plugin-sdk-go/pkg/sdk"
 	"github.com/falcosecurity/plugin-sdk-go/pkg/sdk/plugins/source"
+	"github.com/fsnotify/fsnotify"
 	"github.com/valyala/fastjson"
 )
 
-func fileInode(info os.FileInfo) uint64 {
-	if stat, ok := info.Sys().(*syscall.Stat_t); ok {
-		return stat.Ino
-	}
-	return 0
-}
-
 const (
 	webServerShutdownTimeoutSecs = 5
 	webServerEventChanBufSize    = 50
@@ -59,8 +53,8 @@ func (k *Plugin) Open(params string) (source.Instance, error) {
 		return k.OpenWebServer(u.Host, u.Path, false)
 	case "https":
 		return k.OpenWebServer(u.Host, u.Path, true)
-	case "tail":
-		return k.OpenFileTail(u.Path)
+	case "file":
+		return k.OpenFileWatch(u.Path)
 	case "": // by default, fallback to opening a filepath
 		trimmed := strings.TrimSpace(params)
 
@@ -135,72 +129,112 @@ func (k *Plugin) OpenReader(r io.ReadCloser) (source.Instance, error) {
 		source.WithInstanceEventSize(uint32(k.Config.MaxEventSize)))
 }
 
-// OpenFileTail opens a source.Instance that continuously watches a file for
-// new K8S Audit Events, similar to "tail -f". It handles log rotation by
-// detecting file truncation or inode changes.
-func (k *Plugin) OpenFileTail(path string) (source.Instance, error) {
+// OpenFileWatch opens a source.Instance that continuously watches a file for
+// new K8S Audit Events using fsnotify. It watches the parent directory (as
+// recommended by fsnotify) to handle atomic file replacements and log rotation.
+func (k *Plugin) OpenFileWatch(path string) (source.Instance, error) {
+	absPath, err := filepath.Abs(path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get absolute path: %w", err)
+	}
+	dir := filepath.Dir(absPath)
+
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create file watcher: %w", err)
+	}
+
 	ctx, cancelCtx := context.WithCancel(context.Background())
 	evtC := make(chan source.PushEvent)
 
 	go func() {
 		defer close(evtC)
+		defer watcher.Close()
+
 		var parser fastjson.Parser
+		var file *os.File
 		var offset int64
-		var lastInode uint64
-
-		pollInterval := time.Duration(k.Config.WatchPollIntervalMs) * time.Millisecond
-
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			default:
-			}
 
-			file, err := os.Open(path)
-			if err != nil {
-				select {
-				case <-ctx.Done():
-					return
-				case <-time.After(pollInterval):
-					continue
-				}
+		openFile := func(seekEnd bool) bool {
+			if file != nil {
+				file.Close()
+				file = nil
 			}
-
-			info, err := file.Stat()
+			f, err := os.Open(absPath)
 			if err != nil {
-				file.Close()
-				continue
+				return false
 			}
-
-			inode := fileInode(info)
-			if lastInode != 0 && inode != lastInode {
-				offset = 0 // file rotated (new inode)
-			} else if info.Size() < offset {
-				offset = 0 // file truncated
+			file = f
+			if seekEnd {
+				offset, _ = file.Seek(0, io.SeekEnd)
+			} else {
+				offset = 0
 			}
-			lastInode = inode
+			return true
+		}
 
-			if offset > 0 {
-				file.Seek(offset, io.SeekStart)
+		readNewLines := func() {
+			if file == nil {
+				return
 			}
-
+			file.Seek(offset, io.SeekStart)
 			scanner := bufio.NewScanner(file)
 			for scanner.Scan() {
 				line := scanner.Text()
+				offset += int64(len(line)) + 1
 				if len(line) > 0 {
 					k.parseAuditEventsAndPush(&parser, []byte(line), evtC)
 				}
 			}
+		}
+
+		openFile(true)
 
-			pos, _ := file.Seek(0, io.SeekCurrent)
-			offset = pos
-			file.Close()
+		if err := watcher.Add(dir); err != nil {
+			if file != nil {
+				file.Close()
+			}
+			evtC <- source.PushEvent{Err: err}
+			return
+		}
 
+		for {
 			select {
 			case <-ctx.Done():
+				if file != nil {
+					file.Close()
+				}
 				return
-			case <-time.After(pollInterval):
+			case event, ok := <-watcher.Events:
+				if !ok {
+					if file != nil {
+						file.Close()
+					}
+					return
+				}
+				if event.Name != absPath {
+					continue
+				}
+				if event.Op&fsnotify.Write == fsnotify.Write {
+					readNewLines()
+				}
+				if event.Op&fsnotify.Create == fsnotify.Create {
+					openFile(false)
+					readNewLines()
+				}
+				if event.Op&(fsnotify.Rename|fsnotify.Remove) != 0 {
+					if file != nil {
+						file.Close()
+						file = nil
+					}
+				}
+			case _, ok := <-watcher.Errors:
+				if !ok {
+					if file != nil {
+						file.Close()
+					}
+					return
+				}
 			}
 		}
 	}()