Fix system service panic from early hangup in events

We weren't actually halting the goroutine that sent events, so it
would continue sending even when the channel closed (the most
notable cause being early hangup - e.g. Control-c on a curl
session). Use a context to cancel the events goroutine and stop
sending events.

Fixes containers#6805

Signed-off-by: Matthew Heon <matthew.heon@pm.me>

libpod/events.go

@@ -1,6 +1,7 @@
 package libpod
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/containers/libpod/libpod/events"
@@ -75,16 +76,16 @@ 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(options events.ReadOptions) error {
+func (r *Runtime) Events(ctx context.Context, options events.ReadOptions) error {
 	eventer, err := r.newEventer()
 	if err != nil {
 		return err
 	}
-	return eventer.Read(options)
+	return eventer.Read(ctx, options)
 }
 
 // GetEvents reads the event log and returns events based on input filters
-func (r *Runtime) GetEvents(filters []string) ([]*events.Event, error) {
+func (r *Runtime) GetEvents(ctx context.Context, filters []string) ([]*events.Event, error) {
 	var readErr error
 	eventChannel := make(chan *events.Event)
 	options := events.ReadOptions{
@@ -98,7 +99,7 @@ func (r *Runtime) GetEvents(filters []string) ([]*events.Event, error) {
 		return nil, err
 	}
 	go func() {
-		readErr = eventer.Read(options)
+		readErr = eventer.Read(ctx, options)
 	}()
 	if readErr != nil {
 		return nil, readErr
@@ -112,7 +113,7 @@ func (r *Runtime) GetEvents(filters []string) ([]*events.Event, error) {
 
 // GetLastContainerEvent takes a container name or ID and an event status and returns
 // the last occurrence of the container event
-func (r *Runtime) GetLastContainerEvent(nameOrID string, containerEvent events.Status) (*events.Event, error) {
+func (r *Runtime) GetLastContainerEvent(ctx context.Context, nameOrID string, containerEvent events.Status) (*events.Event, error) {
 	// check to make sure the event.Status is valid
 	if _, err := events.StringToStatus(containerEvent.String()); err != nil {
 		return nil, err
@@ -122,7 +123,7 @@ func (r *Runtime) GetLastContainerEvent(nameOrID string, containerEvent events.S
 		fmt.Sprintf("event=%s", containerEvent),
 		"type=container",
 	}
-	containerEvents, err := r.GetEvents(filters)
+	containerEvents, err := r.GetEvents(ctx, filters)
 	if err != nil {
 		return nil, err
 	}

libpod/events/config.go

@@ -1,6 +1,7 @@
 package events
 
 import (
+	"context"
 	"time"
 
 	"github.com/pkg/errors"
@@ -52,7 +53,7 @@ type Eventer interface {
 	// Write an event to a backend
 	Write(event Event) error
 	// Read an event from the backend
-	Read(options ReadOptions) error
+	Read(ctx context.Context, options ReadOptions) error
 	// String returns the type of event logger
 	String() string
 }

libpod/events/journal_linux.go

@@ -3,6 +3,7 @@
 package events
 
 import (
+	"context"
 	"fmt"
 	"strconv"
 	"time"
@@ -53,7 +54,7 @@ func (e EventJournalD) Write(ee Event) error {
 }
 
 // Read reads events from the journal and sends qualified events to the event channel
-func (e EventJournalD) Read(options ReadOptions) error {
+func (e EventJournalD) Read(ctx context.Context, options ReadOptions) error {
 	defer close(options.EventChannel)
 	eventOptions, err := generateEventOptions(options.Filters, options.Since, options.Until)
 	if err != nil {

libpod/events/logfile.go

@@ -1,6 +1,7 @@
 package events
 
 import (
+	"context"
 	"fmt"
 	"os"
 
@@ -40,7 +41,7 @@ func (e EventLogFile) Write(ee Event) error {
 }
 
 // Reads from the log file
-func (e EventLogFile) Read(options ReadOptions) error {
+func (e EventLogFile) Read(ctx context.Context, options ReadOptions) error {
 	defer close(options.EventChannel)
 	eventOptions, err := generateEventOptions(options.Filters, options.Since, options.Until)
 	if err != nil {
@@ -50,6 +51,17 @@ func (e EventLogFile) Read(options ReadOptions) error {
 	if err != nil {
 		return err
 	}
+	funcDone := make(chan bool)
+	copy := true
+	go func() {
+		select {
+		case <-funcDone:
+			// Do nothing
+		case <-ctx.Done():
+			copy = false
+			t.Kill(errors.New("hangup by client"))
+		}
+	}()
 	for line := range t.Lines {
 		event, err := newEventFromJSONString(line.Text)
 		if err != nil {
@@ -65,10 +77,11 @@ func (e EventLogFile) Read(options ReadOptions) error {
 		for _, filter := range eventOptions {
 			include = include && filter(event)
 		}
-		if include {
+		if include && copy {
 			options.EventChannel <- event
 		}
 	}
+	funcDone <- true
 	return nil
 }
 

libpod/events/nullout.go

@@ -1,5 +1,9 @@
 package events
 
+import (
+	"context"
+)
+
 // EventToNull is an eventer type that only performs write operations
 // and only writes to /dev/null. It is meant for unittests only
 type EventToNull struct{}
@@ -10,7 +14,7 @@ func (e EventToNull) Write(ee Event) error {
 }
 
 // Read does nothing. Do not use it.
-func (e EventToNull) Read(options ReadOptions) error {
+func (e EventToNull) Read(ctx context.Context, options ReadOptions) error {
 	return nil
 }
 

pkg/api/handlers/compat/events.go

@@ -1,6 +1,7 @@
 package compat
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 
@@ -45,20 +46,23 @@ func GetEvents(w http.ResponseWriter, r *http.Request) {
 		fromStart = true
 	}
 
+	eventCtx, eventCancel := context.WithCancel(r.Context())
 	eventChannel := make(chan *events.Event)
 	go func() {
 		readOpts := events.ReadOptions{FromStart: fromStart, Stream: query.Stream, Filters: libpodFilters, EventChannel: eventChannel, Since: query.Since, Until: query.Until}
-		eventsError = runtime.Events(readOpts)
+		eventsError = runtime.Events(eventCtx, readOpts)
 	}()
 	if eventsError != nil {
 		utils.InternalServerError(w, eventsError)
+		eventCancel()
 		close(eventChannel)
 		return
 	}
 
 	// If client disappears we need to stop listening for events
 	go func(done <-chan struct{}) {
 		<-done
+		eventCancel()
 		if _, ok := <-eventChannel; ok {
 			close(eventChannel)
 		}

pkg/domain/infra/abi/containers.go

@@ -741,7 +741,7 @@ func (ic *ContainerEngine) ContainerStart(ctx context.Context, namesOrIds []stri
 			if ecode, err := ctr.Wait(); err != nil {
 				if errors.Cause(err) == define.ErrNoSuchCtr {
 					// Check events
-					event, err := ic.Libpod.GetLastContainerEvent(ctr.ID(), events.Exited)
+					event, err := ic.Libpod.GetLastContainerEvent(ctx, ctr.ID(), events.Exited)
 					if err != nil {
 						logrus.Errorf("Cannot get exit code: %v", err)
 						exitCode = define.ExecErrorCodeNotFound
@@ -871,7 +871,7 @@ func (ic *ContainerEngine) ContainerRun(ctx context.Context, opts entities.Conta
 	if ecode, err := ctr.Wait(); err != nil {
 		if errors.Cause(err) == define.ErrNoSuchCtr {
 			// Check events
-			event, err := ic.Libpod.GetLastContainerEvent(ctr.ID(), events.Exited)
+			event, err := ic.Libpod.GetLastContainerEvent(ctx, ctr.ID(), events.Exited)
 			if err != nil {
 				logrus.Errorf("Cannot get exit code: %v", err)
 				report.ExitCode = define.ExecErrorCodeNotFound

pkg/domain/infra/abi/events.go

@@ -9,5 +9,5 @@ import (
 
 func (ic *ContainerEngine) Events(ctx context.Context, opts entities.EventsOptions) error {
 	readOpts := events.ReadOptions{FromStart: opts.FromStart, Stream: opts.Stream, Filters: opts.Filter, EventChannel: opts.EventChan, Since: opts.Since, Until: opts.Until}
-	return ic.Libpod.Events(readOpts)
+	return ic.Libpod.Events(ctx, readOpts)
 }

pkg/varlinkapi/attach.go

@@ -4,6 +4,7 @@ package varlinkapi
 
 import (
 	"bufio"
+	"context"
 	"io"
 
 	"github.com/containers/libpod/libpod"
@@ -89,7 +90,7 @@ func (i *VarlinkAPI) Attach(call iopodman.VarlinkCall, name string, detachKeys s
 		if ecode, err := ctr.Wait(); err != nil {
 			if errors.Cause(err) == define.ErrNoSuchCtr {
 				// Check events
-				event, err := i.Runtime.GetLastContainerEvent(ctr.ID(), events.Exited)
+				event, err := i.Runtime.GetLastContainerEvent(context.Background(), ctr.ID(), events.Exited)
 				if err != nil {
 					logrus.Errorf("Cannot get exit code: %v", err)
 					exitCode = define.ExecErrorCodeNotFound

pkg/varlinkapi/events.go

@@ -3,6 +3,7 @@
 package varlinkapi
 
 import (
+	"context"
 	"time"
 
 	"github.com/containers/libpod/libpod/events"
@@ -27,7 +28,7 @@ func (i *VarlinkAPI) GetEvents(call iopodman.VarlinkCall, filter []string, since
 	eventChannel := make(chan *events.Event)
 	go func() {
 		readOpts := events.ReadOptions{FromStart: fromStart, Stream: stream, Filters: filter, EventChannel: eventChannel}
-		eventsError = i.Runtime.Events(readOpts)
+		eventsError = i.Runtime.Events(context.Background(), readOpts)
 	}()
 	if eventsError != nil {
 		return call.ReplyErrorOccurred(eventsError.Error())