summaryrefslogtreecommitdiff
path: root/vendor/github.com/fsouza/go-dockerclient/event.go
blob: ce1fb5021b28b4199e2485790a6253cd7209bad9 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
// Copyright 2014 go-dockerclient authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package docker

import (
	"encoding/json"
	"errors"
	"io"
	"math"
	"net"
	"net/http"
	"net/http/httputil"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
)

// EventsOptions to filter events
// See https://docs.docker.com/engine/api/v1.41/#operation/SystemEvents for more details.
type EventsOptions struct {
	// Show events created since this timestamp then stream new events.
	Since string

	// Show events created until this timestamp then stop streaming.
	Until string

	// Filter for events. For example:
	//  map[string][]string{"type": {"container"}, "event": {"start", "die"}}
	// will return events when container was started and stopped or killed
	//
	// Available filters:
	//  config=<string> config name or ID
	//  container=<string> container name or ID
	//  daemon=<string> daemon name or ID
	//  event=<string> event type
	//  image=<string> image name or ID
	//  label=<string> image or container label
	//  network=<string> network name or ID
	//  node=<string> node ID
	//  plugin= plugin name or ID
	//  scope= local or swarm
	//  secret=<string> secret name or ID
	//  service=<string> service name or ID
	//  type=<string> container, image, volume, network, daemon, plugin, node, service, secret or config
	//  volume=<string> volume name
	Filters map[string][]string
}

// APIEvents represents events coming from the Docker API
// The fields in the Docker API changed in API version 1.22, and
// events for more than images and containers are now fired off.
// To maintain forward and backward compatibility, go-dockerclient
// replicates the event in both the new and old format as faithfully as possible.
//
// For events that only exist in 1.22 in later, `Status` is filled in as
// `"Type:Action"` instead of just `Action` to allow for older clients to
// differentiate and not break if they rely on the pre-1.22 Status types.
//
// The transformEvent method can be consulted for more information about how
// events are translated from new/old API formats
type APIEvents struct {
	// New API Fields in 1.22
	Action string   `json:"action,omitempty"`
	Type   string   `json:"type,omitempty"`
	Actor  APIActor `json:"actor,omitempty"`

	// Old API fields for < 1.22
	Status string `json:"status,omitempty"`
	ID     string `json:"id,omitempty"`
	From   string `json:"from,omitempty"`

	// Fields in both
	Time     int64 `json:"time,omitempty"`
	TimeNano int64 `json:"timeNano,omitempty"`
}

// APIActor represents an actor that accomplishes something for an event
type APIActor struct {
	ID         string            `json:"id,omitempty"`
	Attributes map[string]string `json:"attributes,omitempty"`
}

type eventMonitoringState struct {
	// `sync/atomic` expects the first word in an allocated struct to be 64-bit
	// aligned on both ARM and x86-32. See https://goo.gl/zW7dgq for more details.
	lastSeen int64
	sync.RWMutex
	sync.WaitGroup
	enabled   bool
	C         chan *APIEvents
	errC      chan error
	listeners []chan<- *APIEvents
	closeConn func()
}

const (
	maxMonitorConnRetries = 5
	retryInitialWaitTime  = 10.
)

var (
	// ErrNoListeners is the error returned when no listeners are available
	// to receive an event.
	ErrNoListeners = errors.New("no listeners present to receive event")

	// ErrListenerAlreadyExists is the error returned when the listerner already
	// exists.
	ErrListenerAlreadyExists = errors.New("listener already exists for docker events")

	// ErrTLSNotSupported is the error returned when the client does not support
	// TLS (this applies to the Windows named pipe client).
	ErrTLSNotSupported = errors.New("tls not supported by this client")

	// EOFEvent is sent when the event listener receives an EOF error.
	EOFEvent = &APIEvents{
		Type:   "EOF",
		Status: "EOF",
	}
)

// AddEventListener adds a new listener to container events in the Docker API.
//
// The parameter is a channel through which events will be sent.
func (c *Client) AddEventListener(listener chan<- *APIEvents) error {
	return c.AddEventListenerWithOptions(EventsOptions{}, listener)
}

// AddEventListener adds a new listener to container events in the Docker API.
// See https://docs.docker.com/engine/api/v1.41/#operation/SystemEvents for more details.
//
// The listener parameter is a channel through which events will be sent.
func (c *Client) AddEventListenerWithOptions(options EventsOptions, listener chan<- *APIEvents) error {
	var err error
	if !c.eventMonitor.isEnabled() {
		err = c.eventMonitor.enableEventMonitoring(c, options)
		if err != nil {
			return err
		}
	}
	return c.eventMonitor.addListener(listener)
}

// RemoveEventListener removes a listener from the monitor.
func (c *Client) RemoveEventListener(listener chan *APIEvents) error {
	err := c.eventMonitor.removeListener(listener)
	if err != nil {
		return err
	}
	if c.eventMonitor.listernersCount() == 0 {
		c.eventMonitor.disableEventMonitoring()
	}
	return nil
}

func (eventState *eventMonitoringState) addListener(listener chan<- *APIEvents) error {
	eventState.Lock()
	defer eventState.Unlock()
	if listenerExists(listener, &eventState.listeners) {
		return ErrListenerAlreadyExists
	}
	eventState.Add(1)
	eventState.listeners = append(eventState.listeners, listener)
	return nil
}

func (eventState *eventMonitoringState) removeListener(listener chan<- *APIEvents) error {
	eventState.Lock()
	defer eventState.Unlock()
	if listenerExists(listener, &eventState.listeners) {
		var newListeners []chan<- *APIEvents
		for _, l := range eventState.listeners {
			if l != listener {
				newListeners = append(newListeners, l)
			}
		}
		eventState.listeners = newListeners
		eventState.Add(-1)
	}
	return nil
}

func (eventState *eventMonitoringState) closeListeners() {
	for _, l := range eventState.listeners {
		close(l)
		eventState.Add(-1)
	}
	eventState.listeners = nil
}

func (eventState *eventMonitoringState) listernersCount() int {
	eventState.RLock()
	defer eventState.RUnlock()
	return len(eventState.listeners)
}

func listenerExists(a chan<- *APIEvents, list *[]chan<- *APIEvents) bool {
	for _, b := range *list {
		if b == a {
			return true
		}
	}
	return false
}

func (eventState *eventMonitoringState) enableEventMonitoring(c *Client, opts EventsOptions) error {
	eventState.Lock()
	defer eventState.Unlock()
	if !eventState.enabled {
		eventState.enabled = true
		atomic.StoreInt64(&eventState.lastSeen, 0)
		eventState.C = make(chan *APIEvents, 100)
		eventState.errC = make(chan error, 1)
		go eventState.monitorEvents(c, opts)
	}
	return nil
}

func (eventState *eventMonitoringState) disableEventMonitoring() {
	eventState.Lock()
	defer eventState.Unlock()

	eventState.closeListeners()

	eventState.Wait()

	if eventState.enabled {
		eventState.enabled = false
		close(eventState.C)
		close(eventState.errC)

		if eventState.closeConn != nil {
			eventState.closeConn()
			eventState.closeConn = nil
		}
	}
}

func (eventState *eventMonitoringState) monitorEvents(c *Client, opts EventsOptions) {
	const (
		noListenersTimeout  = 5 * time.Second
		noListenersInterval = 10 * time.Millisecond
		noListenersMaxTries = noListenersTimeout / noListenersInterval
	)

	var err error
	for i := time.Duration(0); i < noListenersMaxTries && eventState.noListeners(); i++ {
		time.Sleep(10 * time.Millisecond)
	}

	if eventState.noListeners() {
		// terminate if no listener is available after 5 seconds.
		// Prevents goroutine leak when RemoveEventListener is called
		// right after AddEventListener.
		eventState.disableEventMonitoring()
		return
	}

	if err = eventState.connectWithRetry(c, opts); err != nil {
		// terminate if connect failed
		eventState.disableEventMonitoring()
		return
	}
	for eventState.isEnabled() {
		timeout := time.After(100 * time.Millisecond)
		select {
		case ev, ok := <-eventState.C:
			if !ok {
				return
			}
			if ev == EOFEvent {
				eventState.disableEventMonitoring()
				return
			}
			eventState.updateLastSeen(ev)
			eventState.sendEvent(ev)
		case err = <-eventState.errC:
			if errors.Is(err, ErrNoListeners) {
				eventState.disableEventMonitoring()
				return
			} else if err != nil {
				defer func() { go eventState.monitorEvents(c, opts) }()
				return
			}
		case <-timeout:
			continue
		}
	}
}

func (eventState *eventMonitoringState) connectWithRetry(c *Client, opts EventsOptions) error {
	var retries int
	eventState.RLock()
	eventChan := eventState.C
	errChan := eventState.errC
	eventState.RUnlock()
	closeConn, err := c.eventHijack(opts, atomic.LoadInt64(&eventState.lastSeen), eventChan, errChan)
	for ; err != nil && retries < maxMonitorConnRetries; retries++ {
		waitTime := int64(retryInitialWaitTime * math.Pow(2, float64(retries)))
		time.Sleep(time.Duration(waitTime) * time.Millisecond)
		eventState.RLock()
		eventChan = eventState.C
		errChan = eventState.errC
		eventState.RUnlock()
		closeConn, err = c.eventHijack(opts, atomic.LoadInt64(&eventState.lastSeen), eventChan, errChan)
	}
	eventState.Lock()
	defer eventState.Unlock()
	eventState.closeConn = closeConn
	return err
}

func (eventState *eventMonitoringState) noListeners() bool {
	eventState.RLock()
	defer eventState.RUnlock()
	return len(eventState.listeners) == 0
}

func (eventState *eventMonitoringState) isEnabled() bool {
	eventState.RLock()
	defer eventState.RUnlock()
	return eventState.enabled
}

func (eventState *eventMonitoringState) sendEvent(event *APIEvents) {
	eventState.RLock()
	defer eventState.RUnlock()
	eventState.Add(1)
	defer eventState.Done()
	if eventState.enabled {
		if len(eventState.listeners) == 0 {
			eventState.errC <- ErrNoListeners
			return
		}

		for _, listener := range eventState.listeners {
			select {
			case listener <- event:
			default:
			}
		}
	}
}

func (eventState *eventMonitoringState) updateLastSeen(e *APIEvents) {
	eventState.Lock()
	defer eventState.Unlock()
	if atomic.LoadInt64(&eventState.lastSeen) < e.Time {
		atomic.StoreInt64(&eventState.lastSeen, e.Time)
	}
}

func (c *Client) eventHijack(opts EventsOptions, startTime int64, eventChan chan *APIEvents, errChan chan error) (closeConn func(), err error) {
	// on reconnect override initial Since with last event seen time
	if startTime != 0 {
		opts.Since = strconv.FormatInt(startTime, 10)
	}
	uri := "/events?" + queryString(opts)
	protocol := c.endpointURL.Scheme
	address := c.endpointURL.Path
	if protocol != "unix" && protocol != "npipe" {
		protocol = "tcp"
		address = c.endpointURL.Host
	}
	var dial net.Conn
	if c.TLSConfig == nil {
		dial, err = c.Dialer.Dial(protocol, address)
	} else {
		netDialer, ok := c.Dialer.(*net.Dialer)
		if !ok {
			return nil, ErrTLSNotSupported
		}
		dial, err = tlsDialWithDialer(netDialer, protocol, address, c.TLSConfig)
	}
	if err != nil {
		return nil, err
	}
	//lint:ignore SA1019 the alternative doesn't quite work, so keep using the deprecated thing.
	conn := httputil.NewClientConn(dial, nil)
	req, err := http.NewRequest(http.MethodGet, uri, nil)
	if err != nil {
		return nil, err
	}
	res, err := conn.Do(req)
	if err != nil {
		return nil, err
	}

	keepRunning := int32(1)
	//lint:ignore SA1019 the alternative doesn't quite work, so keep using the deprecated thing.
	go func(res *http.Response, conn *httputil.ClientConn) {
		defer conn.Close()
		defer res.Body.Close()
		decoder := json.NewDecoder(res.Body)
		for atomic.LoadInt32(&keepRunning) == 1 {
			var event APIEvents
			if err := decoder.Decode(&event); err != nil {
				if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
					c.eventMonitor.RLock()
					if c.eventMonitor.enabled && c.eventMonitor.C == eventChan {
						// Signal that we're exiting.
						eventChan <- EOFEvent
					}
					c.eventMonitor.RUnlock()
					break
				}
				errChan <- err
			}
			if event.Time == 0 {
				continue
			}
			transformEvent(&event)
			c.eventMonitor.RLock()
			if c.eventMonitor.enabled && c.eventMonitor.C == eventChan {
				eventChan <- &event
			}
			c.eventMonitor.RUnlock()
		}
	}(res, conn)
	return func() {
		atomic.StoreInt32(&keepRunning, 0)
	}, nil
}

// transformEvent takes an event and determines what version it is from
// then populates both versions of the event
func transformEvent(event *APIEvents) {
	// if event version is <= 1.21 there will be no Action and no Type
	if event.Action == "" && event.Type == "" {
		event.Action = event.Status
		event.Actor.ID = event.ID
		event.Actor.Attributes = map[string]string{}
		switch event.Status {
		case "delete", "import", "pull", "push", "tag", "untag":
			event.Type = "image"
		default:
			event.Type = "container"
			if event.From != "" {
				event.Actor.Attributes["image"] = event.From
			}
		}
	} else {
		if event.Status == "" {
			if event.Type == "image" || event.Type == "container" {
				event.Status = event.Action
			} else {
				// Because just the Status has been overloaded with different Types
				// if an event is not for an image or a container, we prepend the type
				// to avoid problems for people relying on actions being only for
				// images and containers
				event.Status = event.Type + ":" + event.Action
			}
		}
		if event.ID == "" {
			event.ID = event.Actor.ID
		}
		if event.From == "" {
			event.From = event.Actor.Attributes["image"]
		}
	}
}