Enricher: Make audit line source pluggable

internal/pkg/cli/runner/impl.go

@@ -34,7 +34,7 @@ import (
 	"sigs.k8s.io/yaml"
 
 	"sigs.k8s.io/security-profiles-operator/internal/pkg/cli/command"
-	"sigs.k8s.io/security-profiles-operator/internal/pkg/daemon/enricher"
+	"sigs.k8s.io/security-profiles-operator/internal/pkg/daemon/enricher/auditsource"
 	"sigs.k8s.io/security-profiles-operator/internal/pkg/daemon/enricher/types"
 )
 
@@ -101,11 +101,11 @@ func (*defaultImpl) Lines(tailFile *tail.Tail) chan *tail.Line {
 }
 
 func (*defaultImpl) IsAuditLine(line string) bool {
-	return enricher.IsAuditLine(line)
+	return auditsource.IsAuditLine(line)
 }
 
 func (*defaultImpl) ExtractAuditLine(line string) (*types.AuditLine, error) {
-	return enricher.ExtractAuditLine(line)
+	return auditsource.ExtractAuditLine(line)
 }
 
 func (*defaultImpl) GetName(s libseccomp.ScmpSyscall) (string, error) {

internal/pkg/daemon/enricher/audit.go → internal/pkg/daemon/enricher/auditsource/audit.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2020 The Kubernetes Authors.
+Copyright 2025 The Kubernetes Authors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -14,18 +14,84 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package enricher
+package auditsource
 
 import (
 	"errors"
 	"fmt"
+	"io"
 	"regexp"
 	"strconv"
 	"strings"
 
+	"github.com/go-logr/logr"
+	"github.com/nxadm/tail"
+
+	"sigs.k8s.io/security-profiles-operator/internal/pkg/config"
+	"sigs.k8s.io/security-profiles-operator/internal/pkg/daemon/common"
 	"sigs.k8s.io/security-profiles-operator/internal/pkg/daemon/enricher/types"
 )
 
+type AuditdSource struct {
+	logger logr.Logger
+	file   *tail.Tail
+}
+
+func NewAuditdSource(logger logr.Logger) *AuditdSource {
+	return &AuditdSource{
+		logger: logger,
+	}
+}
+
+func (a *AuditdSource) StartTail() (log chan *types.AuditLine, err error) {
+	// Use auditd logs as main source or syslog as fallback.
+	filePath := common.LogFilePath()
+
+	// If the file does not exist, then tail will wait for it to appear
+	a.file, err = tail.TailFile(filePath, tail.Config{
+		ReOpen: true,
+		Follow: true,
+		Location: &tail.SeekInfo{
+			Offset: 0,
+			Whence: io.SeekEnd,
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	log = make(chan *types.AuditLine, 32)
+	go func() {
+		for l := range a.file.Lines {
+			line := l.Text
+			a.logger.V(config.VerboseLevel).Info("Got line: " + line)
+
+			if !IsAuditLine(line) {
+				a.logger.V(config.VerboseLevel).Info("Not an audit line")
+
+				continue
+			}
+
+			auditLine, err := ExtractAuditLine(line)
+			if err != nil {
+				a.logger.Error(err, "extract audit line")
+
+				continue
+			}
+
+			log <- auditLine
+		}
+
+		close(log)
+	}()
+
+	return log, nil
+}
+
+func (a *AuditdSource) TailErr() error {
+	return a.file.Err()
+}
+
 // type IDs are defined at https://elixir.bootlin.com/linux/latest/source/include/uapi/linux/audit.h
 var (
 	seccompLineRegex = regexp.MustCompile(

internal/pkg/daemon/enricher/audit_test.go → internal/pkg/daemon/enricher/auditsource/audit_test.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2020 The Kubernetes Authors.
+Copyright 2025 The Kubernetes Authors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package enricher
+package auditsource
 
 import (
 	"fmt"

internal/pkg/daemon/enricher/auditsource/source.go

@@ -0,0 +1,24 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package auditsource
+
+import "sigs.k8s.io/security-profiles-operator/internal/pkg/daemon/enricher/types"
+
+type AuditLineSource interface {
+	StartTail() (chan *types.AuditLine, error)
+	TailErr() error
+}

internal/pkg/daemon/enricher/enricher.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"os"
 	"strconv"
 	"strings"
@@ -29,7 +28,6 @@ import (
 
 	"github.com/go-logr/logr"
 	"github.com/jellydator/ttlcache/v3"
-	"github.com/nxadm/tail"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/protobuf/encoding/protojson"
@@ -39,7 +37,7 @@ import (
 	apienricher "sigs.k8s.io/security-profiles-operator/api/grpc/enricher"
 	apimetrics "sigs.k8s.io/security-profiles-operator/api/grpc/metrics"
 	"sigs.k8s.io/security-profiles-operator/internal/pkg/config"
-	"sigs.k8s.io/security-profiles-operator/internal/pkg/daemon/common"
+	"sigs.k8s.io/security-profiles-operator/internal/pkg/daemon/enricher/auditsource"
 	"sigs.k8s.io/security-profiles-operator/internal/pkg/daemon/enricher/types"
 	"sigs.k8s.io/security-profiles-operator/internal/pkg/util"
 )
@@ -59,6 +57,7 @@ const (
 type Enricher struct {
 	apienricher.UnimplementedEnricherServer
 	impl
+	source           auditsource.AuditLineSource
 	logger           logr.Logger
 	containerIDCache *ttlcache.Cache[string, string]
 	infoCache        *ttlcache.Cache[string, *types.ContainerInfo]
@@ -72,6 +71,7 @@ type Enricher struct {
 func New(logger logr.Logger) *Enricher {
 	return &Enricher{
 		impl:   &defaultImpl{},
+		source: auditsource.NewAuditdSource(logger),
 		logger: logger,
 		containerIDCache: ttlcache.New(
 			ttlcache.WithTTL[string, string](defaultCacheTimeout),
@@ -157,50 +157,12 @@ func (e *Enricher) Run() error {
 		return fmt.Errorf("start GRPC server: %w", err)
 	}
 
-	// Use auditd logs as main source or syslog as fallback.
-	filePath := common.LogFilePath()
-
-	// If the file does not exist, then tail will wait for it to appear
-	tailFile, err := e.TailFile(
-		filePath,
-		tail.Config{
-			ReOpen: true,
-			Follow: true,
-			Location: &tail.SeekInfo{
-				Offset: 0,
-				Whence: io.SeekEnd,
-			},
-		},
-	)
+	log, err := e.StartTail(e.source)
 	if err != nil {
-		return fmt.Errorf("tailing file: %w", err)
+		return fmt.Errorf("tail audit log: %w", err)
 	}
 
-	e.logger.Info("Reading from file " + filePath)
-
-	for l := range e.Lines(tailFile) {
-		if l.Err != nil {
-			e.logger.Error(l.Err, "failed to tail")
-
-			continue
-		}
-
-		line := l.Text
-		e.logger.V(config.VerboseLevel).Info("Got line: " + line)
-
-		if !IsAuditLine(line) {
-			e.logger.V(config.VerboseLevel).Info("Not an audit line")
-
-			continue
-		}
-
-		auditLine, err := ExtractAuditLine(line)
-		if err != nil {
-			e.logger.Error(err, "extract audit line")
-
-			continue
-		}
-
+	for auditLine := range log {
 		e.logger.V(config.VerboseLevel).Info(fmt.Sprintf("Get container ID for PID: %d", auditLine.ProcessID))
 
 		cID, err := e.ContainerIDForPID(e.containerIDCache, auditLine.ProcessID)
@@ -255,7 +217,7 @@ func (e *Enricher) Run() error {
 		}
 	}
 
-	return fmt.Errorf("enricher failed: %w", e.Reason(tailFile))
+	return fmt.Errorf("enricher failed: %w", e.source.TailErr())
 }
 
 func (e *Enricher) startGrpcServer() error {