system-logs input ignores folders and add tests

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"