summaryrefslogtreecommitdiff
path: root/libpod
diff options
context:
space:
mode:
Diffstat (limited to 'libpod')
-rw-r--r--libpod/container_api.go2
-rw-r--r--libpod/container_log.go23
-rw-r--r--libpod/container_log_linux.go16
-rw-r--r--libpod/container_log_unsupported.go4
4 files changed, 34 insertions, 11 deletions
diff --git a/libpod/container_api.go b/libpod/container_api.go
index b37b05ff2..c7df9d66c 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")
}