Method subscribeEvents

pkg/acquisition/modules/docker/run.go:381–418

subscribeEvents will loop until it can successfully call d.Client.Events() without immediately receiving an error. It applies exponential backoff on failures. Returns the new (eventsChan, errChan) pair or an error if context/tomb is done.

(ctx context.Context)

// without immediately receiving an error. It applies exponential backoff on failures.
// Returns the new (eventsChan, errChan) pair or an error if context/tomb is done.
func (d *Source) subscribeEvents(ctx context.Context) (*subscription, error) {
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case <-d.t.Dying():
		return nil, errors.New("connection aborted, shutting down docker watcher")
	default:
	}

	d.logger.Infof("Subscribing to Docker events")

	operation := func() (*subscription, error) {
		select {
		case <-ctx.Done():
			return nil, backoff.Permanent(ctx.Err())
		case <-d.t.Dying():
			return nil, backoff.Permanent(errors.New("connection aborted, shutting down docker watcher"))
		default:
		}

		return d.trySubscribeEvents(ctx)
	}

	notify := func(err error, wait time.Duration) {
		d.logger.Warnf("failed to subscribe to Docker events: %v; retrying in %s", err, wait)
	}

	bo := d.backoffFactory()

	sub, err := backoff.Retry(ctx, operation, backoff.WithBackOff(bo), backoff.WithNotify(notify))
	if err != nil {
		return nil, err
	}

	d.logger.Info("successfully subscribed to Docker events")

	return sub, nil
}

func (d *Source) Watch(ctx context.Context, containerChan chan *ContainerConfig, containerDeleteChan chan *ContainerConfig, serviceChan chan *ContainerConfig, serviceDeleteChan chan *ContainerConfig) error {
	err := d.checkContainers(ctx, containerChan, containerDeleteChan)