feature: support event service

Signed-off-by: Michael Wan <[email protected]>

Author: HusterWan

apis/filters/parse.go

@@ -0,0 +1,156 @@
+package filters
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+)
+
+// Args stores filter arguments as map key:{map key: bool}.
+// It contains an aggregation of the map of arguments (which are in the form
+// of -f 'key=value') based on the key, and stores values for the same key
+// in a map with string keys and boolean values.
+// e.g given -f 'label=label1=1' -f 'label=label2=2' -f 'image.name=ubuntu'
+// the args will be {"image.name":{"ubuntu":true},"label":{"label1=1":true,"label2=2":true}}
+type Args struct {
+	fields map[string]map[string]bool
+}
+
+// KeyValuePair is used to initialize a new Args
+type KeyValuePair struct {
+	Key   string
+	Value string
+}
+
+// Arg creates a new KeyValuePair for initializing Args
+func Arg(key, value string) KeyValuePair {
+	return KeyValuePair{Key: key, Value: value}
+}
+
+// NewArgs returns a new Args populated with the initial args
+func NewArgs(initialArgs ...KeyValuePair) Args {
+	args := Args{fields: map[string]map[string]bool{}}
+	for _, arg := range initialArgs {
+		args.Add(arg.Key, arg.Value)
+	}
+	return args
+}
+
+// Get returns the list of values associated with the key
+func (args Args) Get(key string) []string {
+	values := args.fields[key]
+	if values == nil {
+		return make([]string, 0)
+	}
+	slice := make([]string, 0, len(values))
+	for key := range values {
+		slice = append(slice, key)
+	}
+	return slice
+}
+
+// Add a new value to the set of values
+func (args Args) Add(key, value string) {
+	if _, ok := args.fields[key]; ok {
+		args.fields[key][value] = true
+	} else {
+		args.fields[key] = map[string]bool{value: true}
+	}
+}
+
+// Del removes a value from the set
+func (args Args) Del(key, value string) {
+	if _, ok := args.fields[key]; ok {
+		delete(args.fields[key], value)
+		if len(args.fields[key]) == 0 {
+			delete(args.fields, key)
+		}
+	}
+}
+
+// Len returns the number of fields in the arguments.
+func (args Args) Len() int {
+	return len(args.fields)
+}
+
+// ExactMatch returns true if the source matches exactly one of the filters.
+func (args Args) ExactMatch(field, source string) bool {
+	fieldValues, ok := args.fields[field]
+	//do not filter if there is no filter set or cannot determine filter
+	if !ok || len(fieldValues) == 0 {
+		return true
+	}
+
+	// try to match full name value to avoid O(N) regular expression matching
+	return fieldValues[source]
+}
+
+// MarshalJSON returns a JSON byte representation of the Args
+func (args Args) MarshalJSON() ([]byte, error) {
+	if len(args.fields) == 0 {
+		return []byte{}, nil
+	}
+	return json.Marshal(args.fields)
+}
+
+// UnmarshalJSON populates the Args from JSON encode bytes
+func (args Args) UnmarshalJSON(raw []byte) error {
+	if len(raw) == 0 {
+		return nil
+	}
+	return json.Unmarshal(raw, &args.fields)
+}
+
+// ErrBadFormat is an error returned when a filter is not in the form key=value
+//
+// Deprecated: this error will be removed in a future version
+var ErrBadFormat = errors.New("bad format of filter (expected name=value)")
+
+// ParseFlag parses a key=value string and adds it to an Args.
+//
+// Deprecated: Use Args.Add()
+func ParseFlag(arg string, prev Args) (Args, error) {
+	filters := prev
+	if len(arg) == 0 {
+		return filters, nil
+	}
+
+	if !strings.Contains(arg, "=") {
+		return filters, ErrBadFormat
+	}
+
+	f := strings.SplitN(arg, "=", 2)
+
+	name := strings.ToLower(strings.TrimSpace(f[0]))
+	value := strings.TrimSpace(f[1])
+
+	filters.Add(name, value)
+
+	return filters, nil
+}
+
+// ToParam packs the Args into a string for easy transport from client to server.
+func ToParam(a Args) (string, error) {
+	if a.Len() == 0 {
+		return "", nil
+	}
+
+	buf, err := json.Marshal(a)
+	return string(buf), err
+}
+
+// FromParam decodes a JSON encoded string into Args
+func FromParam(p string) (Args, error) {
+	args := NewArgs()
+
+	if p == "" {
+		return args, nil
+	}
+
+	raw := []byte(p)
+	err := json.Unmarshal(raw, &args)
+	if err != nil {
+		return args, err
+	}
+	return args, nil
+}

apis/filters/parse_test.go

@@ -0,0 +1,151 @@
+package filters
+
+import (
+	"testing"
+)
+
+func TestParseArgs(t *testing.T) {
+	// equivalent of `docker ps -f 'created=today' -f 'image.name=ubuntu*' -f 'image.name=*untu'`
+	flagArgs := []string{
+		"created=today",
+		"image.name=ubuntu*",
+		"image.name=*untu",
+	}
+	var (
+		args = NewArgs()
+		err  error
+	)
+
+	for i := range flagArgs {
+		args, err = ParseFlag(flagArgs[i], args)
+		if err != nil {
+			t.Fatalf("ParseFlag got err: %v", err)
+		}
+	}
+
+	if len(args.Get("created")) != 1 {
+		t.Fatalf("got unexpected created keys: %v", args.Get("created"))
+	}
+	if len(args.Get("image.name")) != 2 {
+		t.Fatalf("got unexpected image.name keys: %v", args.Get("image.name"))
+	}
+}
+
+func TestAdd(t *testing.T) {
+	f := NewArgs()
+	f.Add("status", "running")
+	v := f.fields["status"]
+	if len(v) != 1 || !v["running"] {
+		t.Fatalf("Expected to include a running status, got %v", v)
+	}
+
+	f.Add("status", "paused")
+	if len(v) != 2 || !v["paused"] {
+		t.Fatalf("Expected to include a paused status, got %v", v)
+	}
+}
+
+func TestDel(t *testing.T) {
+	f := NewArgs()
+	f.Add("status", "running")
+	f.Del("status", "running")
+	v := f.fields["status"]
+	if v["running"] {
+		t.Fatal("Expected to not include a running status filter, got true")
+	}
+}
+
+func TestLen(t *testing.T) {
+	f := NewArgs()
+	if f.Len() != 0 {
+		t.Fatal("Expected to not include any field")
+	}
+	f.Add("status", "running")
+	if f.Len() != 1 {
+		t.Fatal("Expected to include one field")
+	}
+}
+
+func TestExactMatch(t *testing.T) {
+	f := NewArgs()
+
+	if !f.ExactMatch("status", "running") {
+		t.Fatal("Expected to match `running` when there are no filters, got false")
+	}
+
+	f.Add("status", "running")
+	f.Add("status", "pause*")
+
+	if !f.ExactMatch("status", "running") {
+		t.Fatal("Expected to match `running` with one of the filters, got false")
+	}
+
+	if f.ExactMatch("status", "paused") {
+		t.Fatal("Expected to not match `paused` with one of the filters, got true")
+	}
+}
+
+func TestToParam(t *testing.T) {
+	fields := map[string]map[string]bool{
+		"created":    {"today": true},
+		"image.name": {"ubuntu*": true, "*untu": true},
+	}
+	a := Args{fields: fields}
+
+	_, err := ToParam(a)
+	if err != nil {
+		t.Errorf("failed to marshal the filters: %s", err)
+	}
+}
+
+func TestFromParam(t *testing.T) {
+	invalids := []string{
+		"anything",
+		"['a','list']",
+		"{'key': 'value'}",
+		`{"key": "value"}`,
+		`{"key": ["value"]}`,
+	}
+	valid := map[*Args][]string{
+		{fields: map[string]map[string]bool{"key": {"value": true}}}: {
+			`{"key": {"value": true}}`,
+		},
+		{fields: map[string]map[string]bool{"key": {"value1": true, "value2": true}}}: {
+			`{"key": {"value1": true, "value2": true}}`,
+		},
+		{fields: map[string]map[string]bool{"key1": {"value1": true}, "key2": {"value2": true}}}: {
+			`{"key1": {"value1": true}, "key2": {"value2": true}}`,
+		},
+	}
+
+	for _, invalid := range invalids {
+		if _, err := FromParam(invalid); err == nil {
+			t.Fatalf("Expected an error with %v, got nothing", invalid)
+		}
+	}
+
+	for expectedArgs, matchers := range valid {
+		for _, json := range matchers {
+			args, err := FromParam(json)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if args.Len() != expectedArgs.Len() {
+				t.Fatalf("Expected %v, go %v", expectedArgs, args)
+			}
+			for key, expectedValues := range expectedArgs.fields {
+				values := args.Get(key)
+
+				if len(values) != len(expectedValues) {
+					t.Fatalf("Expected %v, go %v", expectedArgs, args)
+				}
+
+				for _, v := range values {
+					if !expectedValues[v] {
+						t.Fatalf("Expected %v, go %v", expectedArgs, args)
+					}
+				}
+			}
+		}
+	}
+}

apis/server/router.go

@@ -28,6 +28,7 @@ func initRoute(s *Server) http.Handler {
 	s.addRoute(r, http.MethodGet, "/info", s.info)
 	s.addRoute(r, http.MethodGet, "/version", s.version)
 	s.addRoute(r, http.MethodPost, "/auth", s.auth)
+	s.addRoute(r, http.MethodGet, "/events", s.events)
 
 	// daemon, we still list this API into system manager.
 	s.addRoute(r, http.MethodPost, "/daemon/update", s.updateDaemon)

apis/server/system_bridge.go

@@ -5,8 +5,12 @@ import (
 	"encoding/json"
 	"net/http"
 
+	"github.com/alibaba/pouch/apis/filters"
 	"github.com/alibaba/pouch/apis/types"
 	"github.com/alibaba/pouch/pkg/httputils"
+	"github.com/alibaba/pouch/pkg/ioutils"
+
+	"github.com/pkg/errors"
 )
 
 func (s *Server) ping(context context.Context, rw http.ResponseWriter, req *http.Request) (err error) {
@@ -60,3 +64,34 @@ func (s *Server) auth(ctx context.Context, rw http.ResponseWriter, req *http.Req
 	}
 	return EncodeResponse(rw, http.StatusOK, authResp)
 }
+
+func (s *Server) events(ctx context.Context, rw http.ResponseWriter, req *http.Request) (err error) {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	rw.Header().Set("Content-Type", "application/json")
+	output := ioutils.NewWriteFlusher(rw)
+	defer output.Close()
+	output.Flush()
+	enc := json.NewEncoder(output)
+
+	ef, err := filters.FromParam(req.FormValue("filters"))
+	if err != nil {
+		return err
+	}
+
+	eventq, errq := s.SystemMgr.SubscribeToEvents(ctx, ef)
+	for {
+		select {
+		case ev := <-eventq:
+			if err := enc.Encode(ev); err != nil {
+				return err
+			}
+		case err := <-errq:
+			if err != nil {
+				return errors.Wrapf(err, "subscribe failed")
+			}
+			return nil
+		}
+	}
+}