log API: add context to allow for cancelling

Add a `context.Context` to the log APIs to allow for cancelling
streaming (e.g., via `podman logs -f`).  This fixes issues for
the remote API where some go routines of the server will continue
writing and produce nothing but heat and waste CPU cycles.

Signed-off-by: Valentin Rothberg <rothberg@redhat.com>

4 files changed, 34 insertions, 11 deletions

diff --git a/libpod/container_api.go b/libpod/container_api.go
index d19dc651b..487f75e67 100644
--- a/libpod/container_api.go
+++ b/libpod/container_api.go
@@ -353,7 +353,7 @@ func (c *Container) HTTPAttach(httpCon net.Conn, httpBuf *bufio.ReadWriter, stre
 			logOpts.WaitGroup.Wait()
 			close(logChan)
 		}()
-		if err := c.ReadLog(logOpts, logChan); err != nil {
+		if err := c.ReadLog(context.Background(), logOpts, logChan); err != nil {
 			return err
 		}
 		logrus.Debugf("Done reading logs for container %s, %d bytes", c.ID(), logSize)
diff --git a/libpod/container_log.go b/libpod/container_log.go
index 97936c683..80f8e6e50 100644
--- a/libpod/container_log.go
+++ b/libpod/container_log.go
@@ -1,6 +1,7 @@
 package libpod
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"time"
@@ -13,9 +14,9 @@ import (
 )
 
 // Log is a runtime function that can read one or more container logs.
-func (r *Runtime) Log(containers []*Container, options *logs.LogOptions, logChannel chan *logs.LogLine) error {
+func (r *Runtime) Log(ctx context.Context, containers []*Container, options *logs.LogOptions, logChannel chan *logs.LogLine) error {
 	for _, ctr := range containers {
-		if err := ctr.ReadLog(options, logChannel); err != nil {
+		if err := ctr.ReadLog(ctx, options, logChannel); err != nil {
 			return err
 		}
 	}
@@ -23,25 +24,25 @@ func (r *Runtime) Log(containers []*Container, options *logs.LogOptions, logChan
 }
 
 // ReadLog reads a containers log based on the input options and returns loglines over a channel.
-func (c *Container) ReadLog(options *logs.LogOptions, logChannel chan *logs.LogLine) error {
+func (c *Container) ReadLog(ctx context.Context, options *logs.LogOptions, logChannel chan *logs.LogLine) error {
 	switch c.LogDriver() {
 	case define.NoLogging:
 		return errors.Wrapf(define.ErrNoLogs, "this container is using the 'none' log driver, cannot read logs")
 	case define.JournaldLogging:
 		// TODO Skip sending logs until journald logs can be read
-		return c.readFromJournal(options, logChannel)
+		return c.readFromJournal(ctx, options, logChannel)
 	case define.JSONLogging:
 		// TODO provide a separate implementation of this when Conmon
 		// has support.
 		fallthrough
 	case define.KubernetesLogging, "":
-		return c.readFromLogFile(options, logChannel)
+		return c.readFromLogFile(ctx, options, logChannel)
 	default:
 		return errors.Wrapf(define.ErrInternal, "unrecognized log driver %q, cannot read logs", c.LogDriver())
 	}
 }
 
-func (c *Container) readFromLogFile(options *logs.LogOptions, logChannel chan *logs.LogLine) error {
+func (c *Container) readFromLogFile(ctx context.Context, options *logs.LogOptions, logChannel chan *logs.LogLine) error {
 	t, tailLog, err := logs.GetLogFile(c.LogPath(), options)
 	if err != nil {
 		// If the log file does not exist, this is not fatal.
@@ -62,8 +63,17 @@ func (c *Container) readFromLogFile(options *logs.LogOptions, logChannel chan *l
 	}
 
 	go func() {
+		defer options.WaitGroup.Done()
+
 		var partial string
 		for line := range t.Lines {
+			select {
+			case <-ctx.Done():
+				// the consumer has cancelled
+				return
+			default:
+				// fallthrough
+			}
 			nll, err := logs.NewLogLine(line.Text)
 			if err != nil {
 				logrus.Error(err)
@@ -82,7 +92,6 @@ func (c *Container) readFromLogFile(options *logs.LogOptions, logChannel chan *l
 				logChannel <- nll
 			}
 		}
-		options.WaitGroup.Done()
 	}()
 	// Check if container is still running or paused
 	if options.Follow {
diff --git a/libpod/container_log_linux.go b/libpod/container_log_linux.go
index fad3bf87c..00b2039a9 100644
--- a/libpod/container_log_linux.go
+++ b/libpod/container_log_linux.go
@@ -4,6 +4,7 @@
 package libpod
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"math"
@@ -29,7 +30,7 @@ const (
 	bufLen = 16384
 )
 
-func (c *Container) readFromJournal(options *logs.LogOptions, logChannel chan *logs.LogLine) error {
+func (c *Container) readFromJournal(ctx context.Context, options *logs.LogOptions, logChannel chan *logs.LogLine) error {
 	var config journal.JournalReaderConfig
 	if options.Tail < 0 {
 		config.NumFromTail = math.MaxUint64
@@ -65,13 +66,24 @@ func (c *Container) readFromJournal(options *logs.LogOptions, logChannel chan *l
 
 	if options.Follow {
 		go func() {
+			done := make(chan bool)
+			until := make(chan time.Time)
+			go func() {
+				select {
+				case <-ctx.Done():
+					until <- time.Time{}
+				case <-done:
+					// nothing to do anymore
+				}
+			}()
 			follower := FollowBuffer{logChannel}
-			err := r.Follow(nil, follower)
+			err := r.Follow(until, follower)
 			if err != nil {
 				logrus.Debugf(err.Error())
 			}
 			r.Close()
 			options.WaitGroup.Done()
+			done <- true
 			return
 		}()
 		return nil
diff --git a/libpod/container_log_unsupported.go b/libpod/container_log_unsupported.go
index 18882720a..f3b36619e 100644
--- a/libpod/container_log_unsupported.go
+++ b/libpod/container_log_unsupported.go
@@ -3,11 +3,13 @@
 package libpod
 
 import (
+	"context"
+
 	"github.com/containers/libpod/v2/libpod/define"
 	"github.com/containers/libpod/v2/libpod/logs"
 	"github.com/pkg/errors"
 )
 
-func (c *Container) readFromJournal(options *logs.LogOptions, logChannel chan *logs.LogLine) error {
+func (c *Container) readFromJournal(_ context.Context, _ *logs.LogOptions, _ chan *logs.LogLine) error {
 	return errors.Wrapf(define.ErrOSNotSupported, "Journald logging only enabled with systemd on linux")
 }