feat: configurable event filter (#610)

* feat: watch network connect / disconnect event

* refactor: use go-dockerclient builtin event filtering

* refactor: make event filter configurable

Author: buchdag

README.md

@@ -91,6 +91,11 @@ Options:
       config files with template directives. Config files will be merged if this option is specified multiple times. (default [])
   -endpoint string
       docker api endpoint (tcp|unix://..). Default unix:///var/run/docker.sock
+  -event-filter value
+      additional filter for event watched by docker-gen (e.g -event-filter event=connect -event-filter event=disconnect).
+      You can pass this option multiple times to combine filters.
+      By default docker-gen listen for container events start, stop, die and health_status.
+      https://docs.docker.com/engine/reference/commandline/events/#filtering-events
   -interval int
       notify command interval (secs)
   -keep-blank-lines

cmd/docker-gen/main.go

@@ -36,6 +36,7 @@ var (
 	includeStopped        bool
 	configFiles           stringslice
 	configs               config.ConfigFile
+	eventFilter           mapstringslice = mapstringslice{"event": {"start", "stop", "die", "health_status"}}
 	interval              int
 	keepBlankLines        bool
 	endpoint              string
@@ -129,6 +130,9 @@ func initFlags() {
 	flag.StringVar(&tlsCaCert, "tlscacert", filepath.Join(certPath, "ca.pem"), "path to TLS CA certificate file")
 	flag.BoolVar(&tlsVerify, "tlsverify", os.Getenv("DOCKER_TLS_VERIFY") != "", "verify docker daemon's TLS certicate")
 
+	flag.Var(&eventFilter, "event-filter",
+		"additional filter for event watched by docker-gen (e.g -event-filter event=connect -event-filter event=disconnect). You can pass this option multiple times to combine filters. By default docker-gen listen for container events start, stop, die and health_status. https://docs.docker.com/engine/reference/commandline/events/#filtering-events")
+
 	flag.Usage = usage
 	flag.Parse()
 }
@@ -202,13 +206,14 @@ func main() {
 	}
 
 	generator, err := generator.NewGenerator(generator.GeneratorConfig{
-		Endpoint:   endpoint,
-		TLSKey:     tlsKey,
-		TLSCert:    tlsCert,
-		TLSCACert:  tlsCaCert,
-		TLSVerify:  tlsVerify,
-		All:        all,
-		ConfigFile: configs,
+		Endpoint:    endpoint,
+		TLSKey:      tlsKey,
+		TLSCert:     tlsCert,
+		TLSCACert:   tlsCaCert,
+		TLSVerify:   tlsVerify,
+		All:         all,
+		EventFilter: eventFilter,
+		ConfigFile:  configs,
 	})
 
 	if err != nil {

internal/generator/generator.go

@@ -26,6 +26,7 @@ type generator struct {
 	TLSVerify                  bool
 	TLSCert, TLSCaCert, TLSKey string
 	All                        bool
+	EventFilter                map[string][]string
 
 	wg    sync.WaitGroup
 	retry bool
@@ -40,6 +41,8 @@ type GeneratorConfig struct {
 	TLSVerify bool
 	All       bool
 
+	EventFilter map[string][]string
+
 	ConfigFile config.ConfigFile
 }
 
@@ -63,15 +66,16 @@ func NewGenerator(gc GeneratorConfig) (*generator, error) {
 	context.SetDockerEnv(apiVersion)
 
 	return &generator{
-		Client:    client,
-		Endpoint:  gc.Endpoint,
-		TLSVerify: gc.TLSVerify,
-		TLSCert:   gc.TLSCert,
-		TLSCaCert: gc.TLSCACert,
-		TLSKey:    gc.TLSKey,
-		All:       gc.All,
-		Configs:   gc.ConfigFile,
-		retry:     true,
+		Client:      client,
+		Endpoint:    gc.Endpoint,
+		TLSVerify:   gc.TLSVerify,
+		TLSCert:     gc.TLSCert,
+		TLSCaCert:   gc.TLSCACert,
+		TLSKey:      gc.TLSKey,
+		All:         gc.All,
+		EventFilter: gc.EventFilter,
+		Configs:     gc.ConfigFile,
+		retry:       true,
 	}, nil
 }
 
@@ -249,7 +253,11 @@ func (g *generator) generateFromEvents() {
 					break
 				}
 				if !watching {
-					err := client.AddEventListener(eventChan)
+					options := docker.EventsOptions{
+						Filters: g.EventFilter,
+					}
+
+					err := client.AddEventListenerWithOptions(options, eventChan)
 					if err != nil && err != docker.ErrListenerAlreadyExists {
 						log.Printf("Error registering docker event listener: %s", err)
 						time.Sleep(10 * time.Second)
@@ -281,12 +289,11 @@ func (g *generator) generateFromEvents() {
 						time.Sleep(10 * time.Second)
 						break
 					}
-					if event.Status == "start" || event.Status == "stop" || event.Status == "die" || strings.Contains(event.Status, "health_status:") {
-						log.Printf("Received event %s for container %s", event.Status, event.ID[:12])
-						// fanout event to all watchers
-						for _, watcher := range watchers {
-							watcher <- event
-						}
+
+					log.Printf("Received event %s for %s %s", event.Action, event.Type, event.Actor.ID[:12])
+					// fanout event to all watchers
+					for _, watcher := range watchers {
+						watcher <- event
 					}
 				case <-time.After(10 * time.Second):
 					// check for docker liveness

internal/generator/generator_test.go

@@ -26,12 +26,11 @@ func TestGenerateFromEvents(t *testing.T) {
 	var counter atomic.Int32
 
 	eventsResponse := `
-{"status":"start","id":"8dfafdbc3a40","from":"base:latest","time":1374067924}
-{"status":"stop","id":"8dfafdbc3a40","from":"base:latest","time":1374067966}
-{"status":"start","id":"8dfafdbc3a40","from":"base:latest","time":1374067970}
-{"status":"destroy","id":"8dfafdbc3a40","from":"base:latest","time":1374067990}`
+{"Type":"container","Action":"start","Actor": {"ID":"8dfafdbc3a40"},"Time":1374067924}
+{"Type":"container","Action":"stop","Actor": {"ID":"8dfafdbc3a40"},"Time":1374067966}
+{"Type":"container","Action":"start","Actor": {"ID":"8dfafdbc3a40"},"Time":1374067970}`
 	infoResponse := `{"Containers":1,"Images":1,"Debug":false,"NFd":11,"NGoroutines":21,"MemoryLimit":true,"SwapLimit":false}`
-	versionResponse := `{"Version":"1.8.0","Os":"Linux","KernelVersion":"3.18.5-tinycore64","GoVersion":"go1.4.1","GitCommit":"a8a31ef","Arch":"amd64","ApiVersion":"1.19"}`
+	versionResponse := `{"Version":"19.03.12","Os":"Linux","KernelVersion":"4.19.76-linuxkit","GoVersion":"go1.13.14","GitCommit":"48a66213fe","Arch":"amd64","ApiVersion":"1.40"}`
 
 	server, _ := dockertest.NewServer("127.0.0.1:0", nil, nil)
 	server.CustomHandler("/events", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {