summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--libpod/events/journal_linux.go7
-rw-r--r--libpod/events/logfile.go8
-rw-r--r--pkg/api/handlers/compat/events.go154
-rw-r--r--pkg/bindings/test/system_test.go29
-rw-r--r--test/e2e/events_test.go4
5 files changed, 140 insertions, 62 deletions
diff --git a/libpod/events/journal_linux.go b/libpod/events/journal_linux.go
index d341ca7b5..7c2a3e0f2 100644
--- a/libpod/events/journal_linux.go
+++ b/libpod/events/journal_linux.go
@@ -90,6 +90,13 @@ func (e EventJournalD) Read(ctx context.Context, options ReadOptions) error {
return err
}
for {
+ select {
+ case <-ctx.Done():
+ // the consumer has cancelled
+ return nil
+ default:
+ // fallthrough
+ }
if _, err := j.Next(); err != nil {
return err
}
diff --git a/libpod/events/logfile.go b/libpod/events/logfile.go
index 28d0dc07e..b70102450 100644
--- a/libpod/events/logfile.go
+++ b/libpod/events/logfile.go
@@ -63,6 +63,14 @@ func (e EventLogFile) Read(ctx context.Context, options ReadOptions) error {
}
}()
for line := range t.Lines {
+ select {
+ case <-ctx.Done():
+ // the consumer has cancelled
+ return nil
+ default:
+ // fallthrough
+ }
+
event, err := newEventFromJSONString(line.Text)
if err != nil {
return err
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() {
diff --git a/test/e2e/events_test.go b/test/e2e/events_test.go
index 93c51098f..9b527b88e 100644
--- a/test/e2e/events_test.go
+++ b/test/e2e/events_test.go
@@ -136,6 +136,7 @@ var _ = Describe("Podman events", func() {
Expect(ec).To(Equal(0))
test := podmanTest.Podman([]string{"events", "--stream=false", "--format", "json"})
test.WaitWithDefaultTimeout()
+ Expect(test.ExitCode()).To(BeZero())
jsonArr := test.OutputToStringArray()
Expect(len(jsonArr)).To(Not(BeZero()))
eventsMap := make(map[string]string)
@@ -143,10 +144,10 @@ var _ = Describe("Podman events", func() {
Expect(err).To(BeNil())
_, exist := eventsMap["Status"]
Expect(exist).To(BeTrue())
- Expect(test.ExitCode()).To(BeZero())
test = podmanTest.Podman([]string{"events", "--stream=false", "--format", "{{json.}}"})
test.WaitWithDefaultTimeout()
+ Expect(test.ExitCode()).To(BeZero())
jsonArr = test.OutputToStringArray()
Expect(len(jsonArr)).To(Not(BeZero()))
eventsMap = make(map[string]string)
@@ -154,6 +155,5 @@ var _ = Describe("Podman events", func() {
Expect(err).To(BeNil())
_, exist = eventsMap["Status"]
Expect(exist).To(BeTrue())
- Expect(test.ExitCode()).To(BeZero())
})
})
generated by cgit v1.2.3-54-g00ecf (git 2.39.0) at 2024-11-06 16:21:17 +0000