From 4995c511e581723ee9d441d500eda2e6d2e8a839 Mon Sep 17 00:00:00 2001 From: Matthew Heon Date: Wed, 1 Jul 2020 16:21:57 -0400 Subject: [PATCH] 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 #6805 Signed-off-by: Matthew Heon --- libpod/events.go | 13 +++++++------ libpod/events/config.go | 3 ++- libpod/events/journal_linux.go | 3 ++- libpod/events/logfile.go | 17 +++++++++++++++-- libpod/events/nullout.go | 6 +++++- pkg/api/handlers/compat/events.go | 6 +++++- pkg/domain/infra/abi/containers.go | 4 ++-- pkg/domain/infra/abi/events.go | 2 +- pkg/varlinkapi/attach.go | 3 ++- pkg/varlinkapi/events.go | 3 ++- 10 files changed, 43 insertions(+), 17 deletions(-) diff --git a/libpod/events.go b/libpod/events.go index 74c910004a..7560940a50 100644 --- a/libpod/events.go +++ b/libpod/events.go @@ -1,6 +1,7 @@ package libpod import ( + "context" "fmt" "github.com/containers/libpod/v2/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 } diff --git a/libpod/events/config.go b/libpod/events/config.go index 8fe551c5df..c34408e639 100644 --- a/libpod/events/config.go +++ b/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 } diff --git a/libpod/events/journal_linux.go b/libpod/events/journal_linux.go index 482435038d..d341ca7b5f 100644 --- a/libpod/events/journal_linux.go +++ b/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 { diff --git a/libpod/events/logfile.go b/libpod/events/logfile.go index 93e6fa3c9e..28d0dc07ee 100644 --- a/libpod/events/logfile.go +++ b/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 } diff --git a/libpod/events/nullout.go b/libpod/events/nullout.go index f3b36e6091..3eca9e8dba 100644 --- a/libpod/events/nullout.go +++ b/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 } diff --git a/pkg/api/handlers/compat/events.go b/pkg/api/handlers/compat/events.go index 215d7c972e..5acc94153f 100644 --- a/pkg/api/handlers/compat/events.go +++ b/pkg/api/handlers/compat/events.go @@ -1,6 +1,7 @@ package compat import ( + "context" "fmt" "net/http" @@ -45,13 +46,15 @@ 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 } @@ -59,6 +62,7 @@ func GetEvents(w http.ResponseWriter, r *http.Request) { // If client disappears we need to stop listening for events go func(done <-chan struct{}) { <-done + eventCancel() if _, ok := <-eventChannel; ok { close(eventChannel) } diff --git a/pkg/domain/infra/abi/containers.go b/pkg/domain/infra/abi/containers.go index d5dce0b0ff..596fc2cc1e 100644 --- a/pkg/domain/infra/abi/containers.go +++ b/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 diff --git a/pkg/domain/infra/abi/events.go b/pkg/domain/infra/abi/events.go index 50d7727ced..7a81854451 100644 --- a/pkg/domain/infra/abi/events.go +++ b/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) } diff --git a/pkg/varlinkapi/attach.go b/pkg/varlinkapi/attach.go index 731d89b8f6..8acf2a1b65 100644 --- a/pkg/varlinkapi/attach.go +++ b/pkg/varlinkapi/attach.go @@ -4,6 +4,7 @@ package varlinkapi import ( "bufio" + "context" "io" "github.com/containers/libpod/v2/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 diff --git a/pkg/varlinkapi/events.go b/pkg/varlinkapi/events.go index 2e468b706c..910b64a571 100644 --- a/pkg/varlinkapi/events.go +++ b/pkg/varlinkapi/events.go @@ -3,6 +3,7 @@ package varlinkapi import ( + "context" "time" "github.com/containers/libpod/v2/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())