Merge pull request #15717 from Luap99/events

fix several podman events issues

6 files changed, 87 insertions, 76 deletions

diff --git a/libpod/events.go b/libpod/events.go
index 60142cb60..2f9799114 100644
--- a/libpod/events.go
+++ b/libpod/events.go
@@ -3,6 +3,7 @@ package libpod
 import (
 	"context"
 	"fmt"
+	"path/filepath"
 	"sync"
 
 	"github.com/containers/podman/v4/libpod/events"
@@ -11,6 +12,10 @@ import (
 
 // newEventer returns an eventer that can be used to read/write events
 func (r *Runtime) newEventer() (events.Eventer, error) {
+	if r.config.Engine.EventsLogFilePath == "" {
+		// default, use path under tmpdir when none was explicitly set by the user
+		r.config.Engine.EventsLogFilePath = filepath.Join(r.config.Engine.TmpDir, "events", "events.log")
+	}
 	options := events.EventerOptions{
 		EventerType:    r.config.Engine.EventsLogger,
 		LogFilePath:    r.config.Engine.EventsLogFilePath,
@@ -133,11 +138,7 @@ func (v *Volume) newVolumeEvent(status events.Status) {
 // Events is a wrapper function for everyone to begin tailing the events log
 // with options
 func (r *Runtime) Events(ctx context.Context, options events.ReadOptions) error {
-	eventer, err := r.newEventer()
-	if err != nil {
-		return err
-	}
-	return eventer.Read(ctx, options)
+	return r.eventer.Read(ctx, options)
 }
 
 // GetEvents reads the event log and returns events based on input filters
@@ -149,10 +150,6 @@ func (r *Runtime) GetEvents(ctx context.Context, filters []string) ([]*events.Ev
 		FromStart:    true,
 		Stream:       false,
 	}
-	eventer, err := r.newEventer()
-	if err != nil {
-		return nil, err
-	}
 
 	logEvents := make([]*events.Event, 0, len(eventChannel))
 	readLock := sync.Mutex{}
@@ -164,7 +161,7 @@ func (r *Runtime) GetEvents(ctx context.Context, filters []string) ([]*events.Ev
 		readLock.Unlock()
 	}()
 
-	readErr := eventer.Read(ctx, options)
+	readErr := r.eventer.Read(ctx, options)
 	readLock.Lock() // Wait for the events to be consumed.
 	return logEvents, readErr
 }
diff --git a/libpod/events/events_linux.go b/libpod/events/events_linux.go
index e7801af5b..66b125dd5 100644
--- a/libpod/events/events_linux.go
+++ b/libpod/events/events_linux.go
@@ -18,9 +18,9 @@ func NewEventer(options EventerOptions) (Eventer, error) {
 		}
 		return eventer, nil
 	case strings.ToUpper(LogFile.String()):
-		return EventLogFile{options}, nil
+		return newLogFileEventer(options)
 	case strings.ToUpper(Null.String()):
-		return NewNullEventer(), nil
+		return newNullEventer(), nil
 	case strings.ToUpper(Memory.String()):
 		return NewMemoryEventer(), nil
 	default:
diff --git a/libpod/events/journal_linux.go b/libpod/events/journal_linux.go
index 16ef6504f..4986502a2 100644
--- a/libpod/events/journal_linux.go
+++ b/libpod/events/journal_linux.go
@@ -112,57 +112,16 @@ func (e EventJournalD) Read(ctx context.Context, options ReadOptions) error {
 		}
 	}
 
-	// the api requires a next|prev before getting a cursor
-	if _, err := j.Next(); err != nil {
-		return fmt.Errorf("failed to move journal cursor to next entry: %w", err)
-	}
-
-	prevCursor, err := j.GetCursor()
-	if err != nil {
-		return fmt.Errorf("failed to get journal cursor: %w", err)
-	}
 	for {
-		select {
-		case <-ctx.Done():
-			// the consumer has cancelled
-			return nil
-		default:
-			// fallthrough
-		}
-
-		if _, err := j.Next(); err != nil {
-			return fmt.Errorf("failed to move journal cursor to next entry: %w", err)
-		}
-		newCursor, err := j.GetCursor()
+		entry, err := getNextEntry(ctx, j, options.Stream, untilTime)
 		if err != nil {
-			return fmt.Errorf("failed to get journal cursor: %w", err)
+			return err
 		}
-		if prevCursor == newCursor {
-			if !options.Stream || (len(options.Until) > 0 && time.Now().After(untilTime)) {
-				break
-			}
-
-			// j.Wait() is blocking, this would cause the goroutine to hang forever
-			// if no more journal entries are generated and thus if the client
-			// has closed the connection in the meantime to leak memory.
-			// Waiting only 5 seconds makes sure we can check if the client closed in the
-			// meantime at least every 5 seconds.
-			t := 5 * time.Second
-			if len(options.Until) > 0 {
-				until := time.Until(untilTime)
-				if until < t {
-					t = until
-				}
-			}
-			_ = j.Wait(t)
-			continue
+		// no entry == we hit the end
+		if entry == nil {
+			return nil
 		}
-		prevCursor = newCursor
 
-		entry, err := j.GetEntry()
-		if err != nil {
-			return fmt.Errorf("failed to read journal entry: %w", err)
-		}
 		newEvent, err := newEventFromJournalEntry(entry)
 		if err != nil {
 			// We can't decode this event.
@@ -177,7 +136,6 @@ func (e EventJournalD) Read(ctx context.Context, options ReadOptions) error {
 			options.EventChannel <- newEvent
 		}
 	}
-	return nil
 }
 
 func newEventFromJournalEntry(entry *sdjournal.JournalEntry) (*Event, error) {
@@ -238,3 +196,51 @@ func newEventFromJournalEntry(entry *sdjournal.JournalEntry) (*Event, error) {
 func (e EventJournalD) String() string {
 	return Journald.String()
 }
+
+// getNextEntry returns the next entry in the journal. If the end  of the
+// journal is reached and stream is not set or the current time is after
+// the until time this function return nil,nil.
+func getNextEntry(ctx context.Context, j *sdjournal.Journal, stream bool, untilTime time.Time) (*sdjournal.JournalEntry, error) {
+	for {
+		select {
+		case <-ctx.Done():
+			// the consumer has cancelled
+			return nil, nil
+		default:
+			// fallthrough
+		}
+		// the api requires a next|prev before reading the event
+		ret, err := j.Next()
+		if err != nil {
+			return nil, fmt.Errorf("failed to move journal cursor to next entry: %w", err)
+		}
+		// ret == 0 equals EOF, see sd_journal_next(3)
+		if ret == 0 {
+			if !stream || (!untilTime.IsZero() && time.Now().After(untilTime)) {
+				// we hit the end and should not keep streaming
+				return nil, nil
+			}
+			// keep waiting for the next entry
+			// j.Wait() is blocking, this would cause the goroutine to hang forever
+			// if no more journal entries are generated and thus if the client
+			// has closed the connection in the meantime to leak memory.
+			// Waiting only 5 seconds makes sure we can check if the client closed in the
+			// meantime at least every 5 seconds.
+			t := 5 * time.Second
+			if !untilTime.IsZero() {
+				until := time.Until(untilTime)
+				if until < t {
+					t = until
+				}
+			}
+			_ = j.Wait(t)
+			continue
+		}
+
+		entry, err := j.GetEntry()
+		if err != nil {
+			return nil, fmt.Errorf("failed to read journal entry: %w", err)
+		}
+		return entry, nil
+	}
+}
diff --git a/libpod/events/logfile.go b/libpod/events/logfile.go
index 519e16629..d749a0d4d 100644
--- a/libpod/events/logfile.go
+++ b/libpod/events/logfile.go
@@ -12,6 +12,7 @@ import (
 	"io/ioutil"
 	"os"
 	"path"
+	"path/filepath"
 	"time"
 
 	"github.com/containers/podman/v4/pkg/util"
@@ -27,6 +28,21 @@ type EventLogFile struct {
 	options EventerOptions
 }
 
+// newLogFileEventer creates a new EventLogFile eventer
+func newLogFileEventer(options EventerOptions) (*EventLogFile, error) {
+	// Create events log dir
+	if err := os.MkdirAll(filepath.Dir(options.LogFilePath), 0700); err != nil {
+		return nil, fmt.Errorf("creating events dirs: %w", err)
+	}
+	// We have to make sure the file is created otherwise reading events will hang.
+	// https://github.com/containers/podman/issues/15688
+	fd, err := os.OpenFile(options.LogFilePath, os.O_RDONLY|os.O_CREATE, 0700)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create event log file: %w", err)
+	}
+	return &EventLogFile{options: options}, fd.Close()
+}
+
 // Writes to the log file
 func (e EventLogFile) Write(ee Event) error {
 	// We need to lock events file
@@ -108,6 +124,8 @@ func (e EventLogFile) Read(ctx context.Context, options ReadOptions) error {
 			}
 		}()
 	}
+	logrus.Debugf("Reading events from file %q", e.options.LogFilePath)
+
 	var line *tail.Line
 	var ok bool
 	for {
diff --git a/libpod/events/nullout.go b/libpod/events/nullout.go
index 587a1b98b..da3820c23 100644
--- a/libpod/events/nullout.go
+++ b/libpod/events/nullout.go
@@ -2,10 +2,11 @@ package events
 
 import (
 	"context"
+	"errors"
 )
 
-// EventToNull is an eventer type that only performs write operations
-// and only writes to /dev/null. It is meant for unittests only
+// EventToNull is an eventer type that does nothing.
+// It is meant for unittests only
 type EventToNull struct{}
 
 // Write eats the event and always returns nil
@@ -13,14 +14,14 @@ func (e EventToNull) Write(ee Event) error {
 	return nil
 }
 
-// Read does nothing. Do not use it.
+// Read does nothing and returns an error.
 func (e EventToNull) Read(ctx context.Context, options ReadOptions) error {
-	return nil
+	return errors.New("cannot read events with the \"none\" backend")
 }
 
-// NewNullEventer returns a new null eventer.  You should only do this for
+// newNullEventer returns a new null eventer.  You should only do this for
 // the purposes of internal libpod testing.
-func NewNullEventer() Eventer {
+func newNullEventer() Eventer {
 	return EventToNull{}
 }
 
diff --git a/libpod/runtime.go b/libpod/runtime.go
index fe90b6df1..83c9f53e2 100644
--- a/libpod/runtime.go
+++ b/libpod/runtime.go
@@ -466,14 +466,6 @@ func makeRuntime(runtime *Runtime) (retErr error) {
 		}
 	}
 
-	// Create events log dir
-	if err := os.MkdirAll(filepath.Dir(runtime.config.Engine.EventsLogFilePath), 0700); err != nil {
-		// The directory is allowed to exist
-		if !errors.Is(err, os.ErrExist) {
-			return fmt.Errorf("creating events dirs: %w", err)
-		}
-	}
-
 	// Get us at least one working OCI runtime.
 	runtime.ociRuntimes = make(map[string]OCIRuntime)
 
@@ -1038,9 +1030,6 @@ func (r *Runtime) mergeDBConfig(dbConfig *DBConfig) {
 			logrus.Debugf("Overriding tmp dir %q with %q from database", c.TmpDir, dbConfig.LibpodTmp)
 		}
 		c.TmpDir = dbConfig.LibpodTmp
-		if c.EventsLogFilePath == "" {
-			c.EventsLogFilePath = filepath.Join(dbConfig.LibpodTmp, "events", "events.log")
-		}
 	}
 
 	if !r.storageSet.VolumePathSet && dbConfig.VolumePath != "" {