vendor/github.com/containerd/containerd/supervisor/worker.go

   1 package supervisor
   2
   3 import (
   4         "sync"
   5         "time"
   6
   7         "github.com/Sirupsen/logrus"
   8         "github.com/containerd/containerd/runtime"
   9         "golang.org/x/net/context"
  10 )
  11
  12 // Worker interface
  13 type Worker interface {
  14         Start()
  15 }
  16
  17 type startTask struct {
  18         Container      runtime.Container
  19         CheckpointPath string
  20         Stdin          string
  21         Stdout         string
  22         Stderr         string
  23         Err            chan error
  24         StartResponse  chan StartResponse
  25         Ctx            context.Context
  26 }
  27
  28 // NewWorker return a new initialized worker
  29 func NewWorker(s *Supervisor, wg *sync.WaitGroup) Worker {
  30         return &worker{
  31                 s:  s,
  32                 wg: wg,
  33         }
  34 }
  35
  36 type worker struct {
  37         wg *sync.WaitGroup
  38         s  *Supervisor
  39 }
  40
  41 // Start runs a loop in charge of starting new containers
  42 func (w *worker) Start() {
  43         defer w.wg.Done()
  44         for t := range w.s.startTasks {
  45                 started := time.Now()
  46                 process, err := t.Container.Start(t.Ctx, t.CheckpointPath, runtime.NewStdio(t.Stdin, t.Stdout, t.Stderr))
  47                 if err != nil {
  48                         logrus.WithFields(logrus.Fields{
  49                                 "error": err,
  50                                 "id":    t.Container.ID(),
  51                         }).Error("containerd: start container")
  52                         t.Err <- err
  53                         evt := &DeleteTask{
  54                                 ID:      t.Container.ID(),
  55                                 NoEvent: true,
  56                                 Process: process,
  57                         }
  58                         w.s.SendTask(evt)
  59                         continue
  60                 }
  61                 if err := w.s.monitor.MonitorOOM(t.Container); err != nil && err != runtime.ErrContainerExited {
  62                         if process.State() != runtime.Stopped {
  63                                 logrus.WithField("error", err).Error("containerd: notify OOM events")
  64                         }
  65                 }
  66                 if err := w.s.monitorProcess(process); err != nil {
  67                         logrus.WithField("error", err).Error("containerd: add process to monitor")
  68                         t.Err <- err
  69                         evt := &DeleteTask{
  70                                 ID:      t.Container.ID(),
  71                                 NoEvent: true,
  72                                 Process: process,
  73                         }
  74                         w.s.SendTask(evt)
  75                         continue
  76                 }
  77                 // only call process start if we aren't restoring from a checkpoint
  78                 // if we have restored from a checkpoint then the process is already started
  79                 if t.CheckpointPath == "" {
  80                         if err := process.Start(); err != nil {
  81                                 logrus.WithField("error", err).Error("containerd: start init process")
  82                                 t.Err <- err
  83                                 evt := &DeleteTask{
  84                                         ID:      t.Container.ID(),
  85                                         NoEvent: true,
  86                                         Process: process,
  87                                 }
  88                                 w.s.SendTask(evt)
  89                                 continue
  90                         }
  91                 }
  92                 ContainerStartTimer.UpdateSince(started)
  93                 w.s.newExecSyncMap(t.Container.ID())
  94                 t.Err <- nil
  95                 t.StartResponse <- StartResponse{
  96                         Container: t.Container,
  97                 }
  98                 w.s.notifySubscribers(Event{
  99                         Timestamp: time.Now(),
 100                         ID:        t.Container.ID(),
 101                         Type:      StateStart,
 102                 })
 103         }
 104 }