containerd.go 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208
  1. package containerd // import "github.com/docker/docker/plugin/executor/containerd"
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "sync"
  7. "syscall"
  8. "github.com/containerd/containerd"
  9. "github.com/containerd/containerd/cio"
  10. "github.com/containerd/log"
  11. "github.com/docker/docker/errdefs"
  12. "github.com/docker/docker/libcontainerd"
  13. libcontainerdtypes "github.com/docker/docker/libcontainerd/types"
  14. specs "github.com/opencontainers/runtime-spec/specs-go"
  15. "github.com/pkg/errors"
  16. )
  17. // ExitHandler represents an object that is called when the exit event is received from containerd
  18. type ExitHandler interface {
  19. HandleExitEvent(id string) error
  20. }
  21. // New creates a new containerd plugin executor
  22. func New(ctx context.Context, rootDir string, cli *containerd.Client, ns string, exitHandler ExitHandler, shim string, shimOpts interface{}) (*Executor, error) {
  23. e := &Executor{
  24. rootDir: rootDir,
  25. exitHandler: exitHandler,
  26. shim: shim,
  27. shimOpts: shimOpts,
  28. plugins: make(map[string]*c8dPlugin),
  29. }
  30. client, err := libcontainerd.NewClient(ctx, cli, rootDir, ns, e)
  31. if err != nil {
  32. return nil, errors.Wrap(err, "error creating containerd exec client")
  33. }
  34. e.client = client
  35. return e, nil
  36. }
  37. // Executor is the containerd client implementation of a plugin executor
  38. type Executor struct {
  39. rootDir string
  40. client libcontainerdtypes.Client
  41. exitHandler ExitHandler
  42. shim string
  43. shimOpts interface{}
  44. mu sync.Mutex // Guards plugins map
  45. plugins map[string]*c8dPlugin
  46. }
  47. type c8dPlugin struct {
  48. log *log.Entry
  49. ctr libcontainerdtypes.Container
  50. tsk libcontainerdtypes.Task
  51. }
  52. // deleteTaskAndContainer deletes plugin task and then plugin container from containerd
  53. func (p c8dPlugin) deleteTaskAndContainer(ctx context.Context) {
  54. if p.tsk != nil {
  55. if err := p.tsk.ForceDelete(ctx); err != nil && !errdefs.IsNotFound(err) {
  56. p.log.WithError(err).Error("failed to delete plugin task from containerd")
  57. }
  58. }
  59. if p.ctr != nil {
  60. if err := p.ctr.Delete(ctx); err != nil && !errdefs.IsNotFound(err) {
  61. p.log.WithError(err).Error("failed to delete plugin container from containerd")
  62. }
  63. }
  64. }
  65. // Create creates a new container
  66. func (e *Executor) Create(id string, spec specs.Spec, stdout, stderr io.WriteCloser) error {
  67. ctx := context.Background()
  68. ctr, err := libcontainerd.ReplaceContainer(ctx, e.client, id, &spec, e.shim, e.shimOpts)
  69. if err != nil {
  70. return errors.Wrap(err, "error creating containerd container for plugin")
  71. }
  72. p := c8dPlugin{log: log.G(ctx).WithField("plugin", id), ctr: ctr}
  73. p.tsk, err = ctr.Start(ctx, "", false, attachStreamsFunc(stdout, stderr))
  74. if err != nil {
  75. p.deleteTaskAndContainer(ctx)
  76. return err
  77. }
  78. e.mu.Lock()
  79. defer e.mu.Unlock()
  80. e.plugins[id] = &p
  81. return nil
  82. }
  83. // Restore restores a container
  84. func (e *Executor) Restore(id string, stdout, stderr io.WriteCloser) (bool, error) {
  85. ctx := context.Background()
  86. p := c8dPlugin{log: log.G(ctx).WithField("plugin", id)}
  87. ctr, err := e.client.LoadContainer(ctx, id)
  88. if err != nil {
  89. if errdefs.IsNotFound(err) {
  90. return false, nil
  91. }
  92. return false, err
  93. }
  94. p.tsk, err = ctr.AttachTask(ctx, attachStreamsFunc(stdout, stderr))
  95. if err != nil {
  96. if errdefs.IsNotFound(err) {
  97. p.deleteTaskAndContainer(ctx)
  98. return false, nil
  99. }
  100. return false, err
  101. }
  102. s, err := p.tsk.Status(ctx)
  103. if err != nil {
  104. if errdefs.IsNotFound(err) {
  105. // Task vanished after attaching?
  106. p.tsk = nil
  107. p.deleteTaskAndContainer(ctx)
  108. return false, nil
  109. }
  110. return false, err
  111. }
  112. if s.Status == containerd.Stopped {
  113. p.deleteTaskAndContainer(ctx)
  114. return false, nil
  115. }
  116. e.mu.Lock()
  117. defer e.mu.Unlock()
  118. e.plugins[id] = &p
  119. return true, nil
  120. }
  121. // IsRunning returns if the container with the given id is running
  122. func (e *Executor) IsRunning(id string) (bool, error) {
  123. e.mu.Lock()
  124. p := e.plugins[id]
  125. e.mu.Unlock()
  126. if p == nil {
  127. return false, errdefs.NotFound(fmt.Errorf("unknown plugin %q", id))
  128. }
  129. status, err := p.tsk.Status(context.Background())
  130. return status.Status == containerd.Running, err
  131. }
  132. // Signal sends the specified signal to the container
  133. func (e *Executor) Signal(id string, signal syscall.Signal) error {
  134. e.mu.Lock()
  135. p := e.plugins[id]
  136. e.mu.Unlock()
  137. if p == nil {
  138. return errdefs.NotFound(fmt.Errorf("unknown plugin %q", id))
  139. }
  140. return p.tsk.Kill(context.Background(), signal)
  141. }
  142. // ProcessEvent handles events from containerd
  143. // All events are ignored except the exit event, which is sent of to the stored handler
  144. func (e *Executor) ProcessEvent(id string, et libcontainerdtypes.EventType, ei libcontainerdtypes.EventInfo) error {
  145. switch et {
  146. case libcontainerdtypes.EventExit:
  147. e.mu.Lock()
  148. p := e.plugins[id]
  149. e.mu.Unlock()
  150. if p == nil {
  151. log.G(context.TODO()).WithField("id", id).Warn("Received exit event for an unknown plugin")
  152. } else {
  153. p.deleteTaskAndContainer(context.Background())
  154. }
  155. return e.exitHandler.HandleExitEvent(ei.ContainerID)
  156. }
  157. return nil
  158. }
  159. type rio struct {
  160. cio.IO
  161. wg sync.WaitGroup
  162. }
  163. func (c *rio) Wait() {
  164. c.wg.Wait()
  165. c.IO.Wait()
  166. }
  167. func attachStreamsFunc(stdout, stderr io.WriteCloser) libcontainerdtypes.StdioCallback {
  168. return func(iop *cio.DirectIO) (cio.IO, error) {
  169. if iop.Stdin != nil {
  170. iop.Stdin.Close()
  171. // closing stdin shouldn't be needed here, it should never be open
  172. panic("plugin stdin shouldn't have been created!")
  173. }
  174. rio := &rio{IO: iop}
  175. rio.wg.Add(2)
  176. go func() {
  177. io.Copy(stdout, iop.Stdout)
  178. stdout.Close()
  179. rio.wg.Done()
  180. }()
  181. go func() {
  182. io.Copy(stderr, iop.Stderr)
  183. stderr.Close()
  184. rio.wg.Done()
  185. }()
  186. return rio, nil
  187. }
  188. }