Do not stop collecting events when journal entries change (#9994)

Previously sd_journal_wait was not used. From now on all changes to journals are detected.

I also added custom seccomp policy to Journalbeat.

Closes #9533
(cherry picked from commit cead4b6)

CHANGELOG.next.asciidoc

@@ -95,6 +95,7 @@ https://github.com/elastic/beats/compare/v6.6.0...6.x[Check the HEAD diff]
 *Journalbeat*
 
 - Add missing journalbeat non breaking fixes. {pull}9106[9106]
+- Do not stop collecting events when journal entries change. {pull}9994[9994]
 
 *Metricbeat*
 

journalbeat/beater/journalbeat.go

@@ -31,6 +31,7 @@ import (
 	"github.com/elastic/beats/libbeat/logp"
 
 	conf "github.com/elastic/beats/journalbeat/config"
+	_ "github.com/elastic/beats/journalbeat/include"
 )
 
 // Journalbeat instance

journalbeat/config/config.go

@@ -27,6 +27,9 @@ import (
 	"github.com/elastic/beats/libbeat/common"
 )
 
+// SeekMode is specifies how a journal is read
+type SeekMode uint8
+
 // Config stores the configuration of Journalbeat
 type Config struct {
 	Inputs       []*common.Config `config:"inputs"`

journalbeat/input/config.go

@@ -18,9 +18,9 @@
 package input
 
 import (
-	"fmt"
 	"time"
 
+	"github.com/elastic/beats/journalbeat/config"
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/processors"
 )
@@ -56,19 +56,3 @@ var (
 		CursorSeekFallback: config.SeekHead,
 	}
 )
-
-// Validate check the configuration of the input.
-func (c *Config) Validate() error {
-	correctSeek := false
-	for _, s := range []string{"cursor", "head", "tail"} {
-		if c.Seek == s {
-			correctSeek = true
-		}
-	}
-
-	if !correctSeek {
-		return fmt.Errorf("incorrect value for seek: %s. possible values: cursor, head, tail", c.Seek)
-	}
-
-	return nil
-}

journalbeat/input/input.go

@@ -37,6 +37,7 @@ type Input struct {
 	done       chan struct{}
 	config     Config
 	pipeline   beat.Pipeline
+	client     beat.Client
 	states     map[string]checkpoint.JournalState
 	id         uuid.UUID
 	logger     *logp.Logger
@@ -122,7 +123,8 @@ func New(
 // Run connects to the output, collects entries from the readers
 // and then publishes the events.
 func (i *Input) Run() {
-	client, err := i.pipeline.ConnectWith(beat.ClientConfig{
+	var err error
+	i.client, err = i.pipeline.ConnectWith(beat.ClientConfig{
 		PublishMode:   beat.GuaranteedSend,
 		EventMetadata: i.eventMeta,
 		Meta:          nil,
@@ -135,13 +137,12 @@ func (i *Input) Run() {
 		i.logger.Error("Error connecting to output: %v", err)
 		return
 	}
-	defer client.Close()
 
-	i.publishAll(client)
+	i.publishAll()
 }
 
 // publishAll reads events from all readers and publishes them.
-func (i *Input) publishAll(client beat.Client) {
+func (i *Input) publishAll() {
 	out := make(chan *beat.Event)
 	defer close(out)
 
@@ -181,13 +182,14 @@ func (i *Input) publishAll(client beat.Client) {
 		case <-i.done:
 			return
 		case e := <-out:
-			client.Publish(*e)
+			i.client.Publish(*e)
 		}
 	}
 }
 
 // Stop stops all readers of the input.
 func (i *Input) Stop() {
+	i.client.Close()
 	for _, r := range i.readers {
 		r.Close()
 	}

journalbeat/reader/config.go

@@ -17,7 +17,11 @@
 
 package reader
 
-import "time"
+import (
+	"time"
+
+	"github.com/elastic/beats/journalbeat/config"
+)
 
 // Config stores the options of a reder.
 type Config struct {