Merge pull request #6902 from vrothberg/events-endpoint

events endpoint: fix panic and race condition

2 files changed, 123 insertions, 60 deletions

diff --git a/pkg/api/handlers/compat/events.go b/pkg/api/handlers/compat/events.go
index 5acc94153..9d5cb5045 100644
--- a/pkg/api/handlers/compat/events.go
+++ b/pkg/api/handlers/compat/events.go
@@ -1,9 +1,10 @@
 package compat
 
 import (
-	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
+	"sync"
 
 	"github.com/containers/libpod/v2/libpod"
 	"github.com/containers/libpod/v2/libpod/events"
@@ -15,77 +16,132 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+// filtersFromRequests extracts the "filters" parameter from the specified
+// http.Request.  The paramater can either be a `map[string][]string` as done
+// in new versions of Docker and libpod, or a `map[string]map[string]bool` as
+// done in older versions of Docker.  We have to do a bit of Yoga to support
+// both - just as Docker does as well.
+//
+// Please refer to https://github.com/containers/podman/issues/6899 for some
+// background.
+func filtersFromRequest(r *http.Request) ([]string, error) {
+	var (
+		compatFilters map[string]map[string]bool
+		filters       map[string][]string
+		libpodFilters []string
+	)
+	raw := []byte(r.Form.Get("filters"))
+
+	// Backwards compat with older versions of Docker.
+	if err := json.Unmarshal(raw, &compatFilters); err == nil {
+		for filterKey, filterMap := range compatFilters {
+			for filterValue, toAdd := range filterMap {
+				if toAdd {
+					libpodFilters = append(libpodFilters, fmt.Sprintf("%s=%s", filterKey, filterValue))
+				}
+			}
+		}
+		return libpodFilters, nil
+	}
+
+	if err := json.Unmarshal(raw, &filters); err != nil {
+		return nil, err
+	}
+
+	for filterKey, filterSlice := range filters {
+		for _, filterValue := range filterSlice {
+			libpodFilters = append(libpodFilters, fmt.Sprintf("%s=%s", filterKey, filterValue))
+		}
+	}
+
+	return libpodFilters, nil
+}
+
+// NOTE: this endpoint serves both the docker-compatible one and the new libpod
+// one.
 func GetEvents(w http.ResponseWriter, r *http.Request) {
 	var (
-		fromStart   bool
-		eventsError error
-		decoder     = r.Context().Value("decoder").(*schema.Decoder)
-		runtime     = r.Context().Value("runtime").(*libpod.Runtime)
+		fromStart bool
+		decoder   = r.Context().Value("decoder").(*schema.Decoder)
+		runtime   = r.Context().Value("runtime").(*libpod.Runtime)
+		json      = jsoniter.ConfigCompatibleWithStandardLibrary // FIXME: this should happen on the package level
 	)
 
+	// NOTE: the "filters" parameter is extracted separately for backwards
+	// compat via `fitlerFromRequest()`.
 	query := struct {
-		Since   string              `schema:"since"`
-		Until   string              `schema:"until"`
-		Filters map[string][]string `schema:"filters"`
-		Stream  bool                `schema:"stream"`
+		Since  string `schema:"since"`
+		Until  string `schema:"until"`
+		Stream bool   `schema:"stream"`
 	}{
 		Stream: true,
 	}
 	if err := decoder.Decode(&query, r.URL.Query()); err != nil {
 		utils.Error(w, "Failed to parse parameters", http.StatusBadRequest, errors.Wrapf(err, "Failed to parse parameters for %s", r.URL.String()))
-	}
-
-	var libpodFilters = []string{}
-	if _, found := r.URL.Query()["filters"]; found {
-		for k, v := range query.Filters {
-			libpodFilters = append(libpodFilters, fmt.Sprintf("%s=%s", k, v[0]))
-		}
+		return
 	}
 
 	if len(query.Since) > 0 || len(query.Until) > 0 {
 		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(eventCtx, readOpts)
-	}()
-	if eventsError != nil {
-		utils.InternalServerError(w, eventsError)
-		eventCancel()
-		close(eventChannel)
+	libpodFilters, err := filtersFromRequest(r)
+	if err != nil {
+		utils.Error(w, "Failed to parse parameters", http.StatusBadRequest, errors.Wrapf(err, "Failed to parse parameters for %s", r.URL.String()))
 		return
 	}
 
-	// If client disappears we need to stop listening for events
-	go func(done <-chan struct{}) {
-		<-done
-		eventCancel()
-		if _, ok := <-eventChannel; ok {
-			close(eventChannel)
+	eventChannel := make(chan *events.Event)
+	errorChannel := make(chan error)
+
+	// Start reading events.
+	go func() {
+		readOpts := events.ReadOptions{
+			FromStart:    fromStart,
+			Stream:       query.Stream,
+			Filters:      libpodFilters,
+			EventChannel: eventChannel,
+			Since:        query.Since,
+			Until:        query.Until,
 		}
-	}(r.Context().Done())
+		errorChannel <- runtime.Events(r.Context(), readOpts)
+	}()
 
-	// Headers need to be written out before turning Writer() over to json encoder
-	w.Header().Set("Content-Type", "application/json")
-	w.WriteHeader(http.StatusOK)
-	if flusher, ok := w.(http.Flusher); ok {
-		flusher.Flush()
-	}
+	var coder *jsoniter.Encoder
+	var writeHeader sync.Once
 
-	json := jsoniter.ConfigCompatibleWithStandardLibrary
-	coder := json.NewEncoder(w)
-	coder.SetEscapeHTML(true)
+	for stream := true; stream; stream = query.Stream {
+		select {
+		case err := <-errorChannel:
+			if err != nil {
+				utils.InternalServerError(w, err)
+				return
+			}
+		case evt := <-eventChannel:
+			writeHeader.Do(func() {
+				// Use a sync.Once so that we write the header
+				// only once.
+				w.Header().Set("Content-Type", "application/json")
+				w.WriteHeader(http.StatusOK)
+				if flusher, ok := w.(http.Flusher); ok {
+					flusher.Flush()
+				}
+				coder = json.NewEncoder(w)
+				coder.SetEscapeHTML(true)
+			})
 
-	for event := range eventChannel {
-		e := entities.ConvertToEntitiesEvent(*event)
-		if err := coder.Encode(e); err != nil {
-			logrus.Errorf("unable to write json: %q", err)
-		}
-		if flusher, ok := w.(http.Flusher); ok {
-			flusher.Flush()
+			if evt == nil {
+				continue
+			}
+
+			e := entities.ConvertToEntitiesEvent(*evt)
+			if err := coder.Encode(e); err != nil {
+				logrus.Errorf("unable to write json: %q", err)
+			}
+			if flusher, ok := w.(http.Flusher); ok {
+				flusher.Flush()
+			}
 		}
+
 	}
 }
diff --git a/pkg/bindings/test/system_test.go b/pkg/bindings/test/system_test.go
index 93141400b..430184f4a 100644
--- a/pkg/bindings/test/system_test.go
+++ b/pkg/bindings/test/system_test.go
@@ -1,6 +1,7 @@
 package test_bindings
 
 import (
+	"sync"
 	"time"
 
 	"github.com/containers/libpod/v2/pkg/bindings"
@@ -38,22 +39,28 @@ var _ = Describe("Podman system", func() {
 	})
 
 	It("podman events", func() {
-		eChan := make(chan entities.Event, 1)
-		var messages []entities.Event
-		cancelChan := make(chan bool, 1)
+		var name = "top"
+		_, err := bt.RunTopContainer(&name, bindings.PFalse, nil)
+		Expect(err).To(BeNil())
+
+		filters := make(map[string][]string)
+		filters["container"] = []string{name}
+
+		binChan := make(chan entities.Event)
+		done := sync.Mutex{}
+		done.Lock()
+		eventCounter := 0
 		go func() {
-			for e := range eChan {
-				messages = append(messages, e)
+			defer done.Unlock()
+			for range binChan {
+				eventCounter++
 			}
 		}()
-		go func() {
-			system.Events(bt.conn, eChan, cancelChan, nil, nil, nil, bindings.PFalse)
-		}()
 
-		_, err := bt.RunTopContainer(nil, nil, nil)
+		err = system.Events(bt.conn, binChan, nil, nil, nil, filters, bindings.PFalse)
 		Expect(err).To(BeNil())
-		cancelChan <- true
-		Expect(len(messages)).To(BeNumerically("==", 5))
+		done.Lock()
+		Expect(eventCounter).To(BeNumerically(">", 0))
 	})
 
 	It("podman system prune - pod,container stopped", func() {