system-logs input ignores folders and add tests (#41296) (#41382)

The system-logs input now does not count folders as an "existing file"
when looking for files to decide between the journald and log inputs.

Unit and integration tests are added for the system-logs input.

(cherry picked from commit e82c43c)

Co-authored-by: Tiago Queiroz <tiago.queiroz@elastic.co>

filebeat/input/systemlogs/input.go

@@ -20,6 +20,7 @@ package systemlogs
 import (
 	"errors"
 	"fmt"
+	"os"
 	"path/filepath"
 
 	"github.com/elastic/beats/v7/filebeat/channel"
@@ -145,17 +146,29 @@ func useJournald(c *conf.C) (bool, error) {
 		if err != nil {
 			return false, fmt.Errorf("cannot resolve glob: %w", err)
 		}
-		if len(paths) != 0 {
-			// We found at least one system log file,
-			// journald will not be used, return early
-			logger.Info(
+
+		for _, p := range paths {
+			stat, err := os.Stat(p)
+			if err != nil {
+				return false, fmt.Errorf("cannot stat '%s': %w", p, err)
+			}
+
+			// Ignore directories
+			if stat.IsDir() {
+				continue
+			}
+
+			// We found one file, return early
+			logger.Infof(
 				"using log input because file(s) was(were) found when testing glob '%s'",
 				g)
 			return false, nil
 		}
 	}
 
 	// if no system log files are found, then use jounrald
+	logger.Info("no files were found, using journald input")
+
 	return true, nil
 }
 

filebeat/input/systemlogs/input_linux_test.go

@@ -0,0 +1,54 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you 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.
+
+//go:build linux
+
+package systemlogs
+
+import (
+	"testing"
+
+	conf "github.com/elastic/elastic-agent-libs/config"
+)
+
+func TestJournaldInputIsCreated(t *testing.T) {
+	c := map[string]any{
+		"files.paths": []string{"/file/does/not/exist"},
+		// The 'journald' object needs to exist for the input to be instantiated
+		"journald.enabled": true,
+	}
+
+	cfg := conf.MustNewConfigFrom(c)
+
+	_, inp, err := configure(cfg)
+	if err != nil {
+		t.Fatalf("did not expect an error calling newV1Input: %s", err)
+	}
+
+	type namer interface {
+		Name() string
+	}
+
+	i, isNamer := inp.(namer)
+	if !isNamer {
+		t.Fatalf("expecting an instance of *log.Input, got '%T' instead", inp)
+	}
+
+	if got, expected := i.Name(), "journald"; got != expected {
+		t.Fatalf("expecting '%s' input, got '%s'", expected, got)
+	}
+}

filebeat/input/systemlogs/input_test.go

@@ -0,0 +1,145 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you 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 systemlogs
+
+import (
+	"os"
+	"testing"
+
+	"github.com/elastic/beats/v7/filebeat/channel"
+	"github.com/elastic/beats/v7/filebeat/input"
+	"github.com/elastic/beats/v7/filebeat/input/log"
+	"github.com/elastic/beats/v7/libbeat/beat"
+	conf "github.com/elastic/elastic-agent-libs/config"
+)
+
+func generateFile(t *testing.T) string {
+	// Create a know file for testing, the content is not relevant
+	// it just needs to exist
+	knwonFile, err := os.CreateTemp(t.TempDir(), t.Name()+"knwonFile*")
+	if err != nil {
+		t.Fatalf("cannot create temporary file: %s", err)
+	}
+
+	if _, err := knwonFile.WriteString("Bowties are cool"); err != nil {
+		t.Fatalf("cannot write to temporary file '%s': %s", knwonFile.Name(), err)
+	}
+	knwonFile.Close()
+
+	return knwonFile.Name()
+}
+
+func TestUseJournald(t *testing.T) {
+	filename := generateFile(t)
+
+	testCases := map[string]struct {
+		cfg         map[string]any
+		useJournald bool
+		expectErr   bool
+	}{
+		"No files found": {
+			cfg: map[string]any{
+				"files.paths": []string{"/file/does/not/exist"},
+			},
+			useJournald: true,
+		},
+		"File exists": {
+			cfg: map[string]any{
+				"files.paths": []string{filename},
+			},
+			useJournald: false,
+		},
+		"use_journald is true": {
+			cfg: map[string]any{
+				"use_journald": true,
+				"journald":     struct{}{},
+			},
+			useJournald: true,
+		},
+		"use_files is true": {
+			cfg: map[string]any{
+				"use_files": true,
+				"journald":  nil,
+				"files":     struct{}{},
+			},
+			useJournald: false,
+		},
+		"use_journald and use_files are true": {
+			cfg: map[string]any{
+				"use_files":    true,
+				"use_journald": true,
+				"journald":     struct{}{},
+			},
+			useJournald: false,
+			expectErr:   true,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			cfg := conf.MustNewConfigFrom(tc.cfg)
+
+			useJournald, err := useJournald(cfg)
+			if !tc.expectErr && err != nil {
+				t.Fatalf("did not expect an error calling 'useJournald': %s", err)
+			}
+			if tc.expectErr && err == nil {
+				t.Fatal("expecting an error when calling 'userJournald', got none")
+			}
+
+			if useJournald != tc.useJournald {
+				t.Fatalf("expecting 'useJournald' to be %t, got %t",
+					tc.useJournald, useJournald)
+			}
+		})
+	}
+}
+
+func TestLogInputIsInstantiated(t *testing.T) {
+	filename := generateFile(t)
+	c := map[string]any{
+		"files.paths": []string{filename},
+	}
+
+	cfg := conf.MustNewConfigFrom(c)
+
+	inp, err := newV1Input(cfg, connectorMock{}, input.Context{})
+	if err != nil {
+		t.Fatalf("did not expect an error calling newV1Input: %s", err)
+	}
+	_, isLogInput := inp.(*log.Input)
+	if !isLogInput {
+		t.Fatalf("expecting an instance of *log.Input, got '%T' instead", inp)
+	}
+}
+
+type connectorMock struct{}
+
+func (mock connectorMock) Connect(c *conf.C) (channel.Outleter, error) {
+	return outleterMock{}, nil
+}
+
+func (mock connectorMock) ConnectWith(c *conf.C, clientConfig beat.ClientConfig) (channel.Outleter, error) {
+	return outleterMock{}, nil
+}
+
+type outleterMock struct{}
+
+func (o outleterMock) Close() error            { return nil }
+func (o outleterMock) Done() <-chan struct{}   { return make(chan struct{}) }
+func (o outleterMock) OnEvent(beat.Event) bool { return false }

filebeat/tests/integration/systemlogs_test.go

@@ -0,0 +1,108 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you 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.
+
+//go:build integration && linux
+
+package integration
+
+import (
+	_ "embed"
+	"fmt"
+	"os"
+	"path"
+	"path/filepath"
+	"testing"
+	"time"
+
+	cp "github.com/otiai10/copy"
+
+	"github.com/elastic/beats/v7/libbeat/tests/integration"
+)
+
+//go:embed testdata/filebeat_system_module.yml
+var systemModuleCfg string
+
+// TestSystemLogsCanUseJournald aims to ensure the system-logs input can
+// correctly choose and start a journald input when the globs defined in
+// var.paths do not resolve to any file.
+func TestSystemLogsCanUseJournaldInput(t *testing.T) {
+	filebeat := integration.NewBeat(
+		t,
+		"filebeat",
+		"../../filebeat.test",
+	)
+	workDir := filebeat.TempDir()
+	copyModulesDir(t, workDir)
+
+	// As the name says, we want this folder to exist bu t be empty
+	globWithoutFiles := filepath.Join(filebeat.TempDir(), "this-folder-does-not-exist")
+	yamlCfg := fmt.Sprintf(systemModuleCfg, globWithoutFiles, workDir)
+
+	filebeat.WriteConfigFile(yamlCfg)
+	filebeat.Start()
+
+	filebeat.WaitForLogs(
+		"no files were found, using journald input",
+		10*time.Second,
+		"system-logs did not select journald input")
+	filebeat.WaitForLogs(
+		"journalctl started with PID",
+		10*time.Second,
+		"system-logs did not start journald input")
+}
+
+func TestSystemLogsCanUseLogInput(t *testing.T) {
+	filebeat := integration.NewBeat(
+		t,
+		"filebeat",
+		"../../filebeat.test",
+	)
+	workDir := filebeat.TempDir()
+	copyModulesDir(t, workDir)
+
+	logFilePath := path.Join(workDir, "syslog")
+	integration.GenerateLogFile(t, logFilePath, 5, false)
+	yamlCfg := fmt.Sprintf(systemModuleCfg, logFilePath, workDir)
+
+	filebeat.WriteConfigFile(yamlCfg)
+	filebeat.Start()
+
+	filebeat.WaitForLogs(
+		"using log input because file(s) was(were) found",
+		10*time.Second,
+		"system-logs did not select the log input")
+	filebeat.WaitForLogs(
+		"Harvester started for paths:",
+		10*time.Second,
+		"system-logs did not start the log input")
+}
+
+func copyModulesDir(t *testing.T, dst string) {
+	pwd, err := os.Getwd()
+	if err != nil {
+		t.Fatalf("cannot get the current directory: %s", err)
+	}
+	localModules := filepath.Join(pwd, "../", "../", "module")
+	localModulesD := filepath.Join(pwd, "../", "../", "modules.d")
+
+	if err := cp.Copy(localModules, filepath.Join(dst, "module")); err != nil {
+		t.Fatalf("cannot copy 'module' folder to test folder: %s", err)
+	}
+	if err := cp.Copy(localModulesD, filepath.Join(dst, "modules.d")); err != nil {
+		t.Fatalf("cannot copy 'modules.d' folder to test folder: %s", err)
+	}
+}

filebeat/tests/integration/testdata/filebeat_system_module.yml

@@ -0,0 +1,16 @@
+filebeat.modules:
+  - module: system
+    syslog:
+      enabled: true
+      var.paths:
+        - "%s"
+
+path.home: %s
+
+queue.mem:
+  flush.timeout: 0
+
+output:
+  file:
+    path: ${path.home}
+    filename: "output"