Method ServiceManager

pkg/acquisition/modules/docker/run.go:846–884

(ctx context.Context, in chan *ContainerConfig, deleteChan chan *ContainerConfig, outChan chan pipeline.Event)

}

func (d *Source) ServiceManager(ctx context.Context, in chan *ContainerConfig, deleteChan chan *ContainerConfig, outChan chan pipeline.Event) error {
	d.logger.Info("Service Manager started")

	for {
		select {
		case newService := <-in:
			if _, ok := d.runningServiceState.Get(newService.ID); !ok {
				newService.logger = d.logger.WithField("service_name", newService.Name)
				newService.t.Go(func() error {
					return d.TailService(ctx, newService, outChan, deleteChan)
				})

				d.runningServiceState.Set(newService.ID, newService)
			}
		case serviceToDelete := <-deleteChan:
			if serviceConfig, ok := d.runningServiceState.Get(serviceToDelete.ID); ok {
				d.logger.Infof("service acquisition stopped for service '%s'", serviceConfig.Name)
				serviceConfig.t.Kill(nil)
				d.runningServiceState.Delete(serviceToDelete.ID)
			}
		case <-d.t.Dying():
			for _, service := range d.runningServiceState.GetAll() {
				if service.t.Alive() {
					d.logger.Infof("killing tail for service %s", service.Name)
					service.t.Kill(nil)

					if err := service.t.Wait(); err != nil {
						d.logger.Infof("error while waiting for death of %s : %s", service.Name, err)
					}
				}
			}

			d.runningServiceState = nil
			d.logger.Debugf("service manager cleanup done, return")

			return nil
		}
	}
}