remote_daemon.go 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317
  1. // +build !windows
  2. package libcontainerd
  3. import (
  4. "context"
  5. "fmt"
  6. "io"
  7. "io/ioutil"
  8. "os"
  9. "os/exec"
  10. "path/filepath"
  11. "strconv"
  12. "strings"
  13. "sync"
  14. "syscall"
  15. "time"
  16. "github.com/BurntSushi/toml"
  17. "github.com/containerd/containerd"
  18. "github.com/containerd/containerd/server"
  19. "github.com/docker/docker/pkg/system"
  20. "github.com/pkg/errors"
  21. "github.com/sirupsen/logrus"
  22. )
  23. const (
  24. maxConnectionRetryCount = 3
  25. healthCheckTimeout = 3 * time.Second
  26. shutdownTimeout = 15 * time.Second
  27. configFile = "containerd.toml"
  28. binaryName = "docker-containerd"
  29. pidFile = "docker-containerd.pid"
  30. )
  31. type pluginConfigs struct {
  32. Plugins map[string]interface{} `toml:"plugins"`
  33. }
  34. type remote struct {
  35. sync.RWMutex
  36. server.Config
  37. daemonPid int
  38. logger *logrus.Entry
  39. daemonWaitCh chan struct{}
  40. clients []*client
  41. shutdownContext context.Context
  42. shutdownCancel context.CancelFunc
  43. shutdown bool
  44. // Options
  45. startDaemon bool
  46. rootDir string
  47. stateDir string
  48. snapshotter string
  49. pluginConfs pluginConfigs
  50. }
  51. // New creates a fresh instance of libcontainerd remote.
  52. func New(rootDir, stateDir string, options ...RemoteOption) (rem Remote, err error) {
  53. defer func() {
  54. if err != nil {
  55. err = errors.Wrap(err, "Failed to connect to containerd")
  56. }
  57. }()
  58. r := &remote{
  59. rootDir: rootDir,
  60. stateDir: stateDir,
  61. Config: server.Config{
  62. Root: filepath.Join(rootDir, "daemon"),
  63. State: filepath.Join(stateDir, "daemon"),
  64. },
  65. pluginConfs: pluginConfigs{make(map[string]interface{})},
  66. daemonPid: -1,
  67. logger: logrus.WithField("module", "libcontainerd"),
  68. }
  69. r.shutdownContext, r.shutdownCancel = context.WithCancel(context.Background())
  70. rem = r
  71. for _, option := range options {
  72. if err = option.Apply(r); err != nil {
  73. return
  74. }
  75. }
  76. r.setDefaults()
  77. if err = system.MkdirAll(stateDir, 0700, ""); err != nil {
  78. return
  79. }
  80. if r.startDaemon {
  81. os.Remove(r.GRPC.Address)
  82. if err = r.startContainerd(); err != nil {
  83. return
  84. }
  85. defer func() {
  86. if err != nil {
  87. r.Cleanup()
  88. }
  89. }()
  90. }
  91. // This connection is just used to monitor the connection
  92. client, err := containerd.New(r.GRPC.Address)
  93. if err != nil {
  94. return
  95. }
  96. if _, err := client.Version(context.Background()); err != nil {
  97. system.KillProcess(r.daemonPid)
  98. return nil, errors.Wrapf(err, "unable to get containerd version")
  99. }
  100. go r.monitorConnection(client)
  101. return r, nil
  102. }
  103. func (r *remote) NewClient(ns string, b Backend) (Client, error) {
  104. c := &client{
  105. stateDir: r.stateDir,
  106. logger: r.logger.WithField("namespace", ns),
  107. namespace: ns,
  108. backend: b,
  109. containers: make(map[string]*container),
  110. }
  111. rclient, err := containerd.New(r.GRPC.Address, containerd.WithDefaultNamespace(ns))
  112. if err != nil {
  113. return nil, err
  114. }
  115. c.remote = rclient
  116. go c.processEventStream(r.shutdownContext)
  117. r.Lock()
  118. r.clients = append(r.clients, c)
  119. r.Unlock()
  120. return c, nil
  121. }
  122. func (r *remote) Cleanup() {
  123. if r.daemonPid != -1 {
  124. r.shutdownCancel()
  125. r.stopDaemon()
  126. }
  127. // cleanup some files
  128. os.Remove(filepath.Join(r.stateDir, pidFile))
  129. r.platformCleanup()
  130. }
  131. func (r *remote) getContainerdPid() (int, error) {
  132. pidFile := filepath.Join(r.stateDir, pidFile)
  133. f, err := os.OpenFile(pidFile, os.O_RDWR, 0600)
  134. if err != nil {
  135. if os.IsNotExist(err) {
  136. return -1, nil
  137. }
  138. return -1, err
  139. }
  140. defer f.Close()
  141. b := make([]byte, 8)
  142. n, err := f.Read(b)
  143. if err != nil && err != io.EOF {
  144. return -1, err
  145. }
  146. if n > 0 {
  147. pid, err := strconv.ParseUint(string(b[:n]), 10, 64)
  148. if err != nil {
  149. return -1, err
  150. }
  151. if system.IsProcessAlive(int(pid)) {
  152. return int(pid), nil
  153. }
  154. }
  155. return -1, nil
  156. }
  157. func (r *remote) getContainerdConfig() (string, error) {
  158. path := filepath.Join(r.stateDir, configFile)
  159. f, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0600)
  160. if err != nil {
  161. return "", errors.Wrapf(err, "failed to open containerd config file at %s", path)
  162. }
  163. defer f.Close()
  164. enc := toml.NewEncoder(f)
  165. if err = enc.Encode(r.Config); err != nil {
  166. return "", errors.Wrapf(err, "failed to encode general config")
  167. }
  168. if err = enc.Encode(r.pluginConfs); err != nil {
  169. return "", errors.Wrapf(err, "failed to encode plugin configs")
  170. }
  171. return path, nil
  172. }
  173. func (r *remote) startContainerd() error {
  174. pid, err := r.getContainerdPid()
  175. if err != nil {
  176. return err
  177. }
  178. if pid != -1 {
  179. r.daemonPid = pid
  180. logrus.WithField("pid", pid).
  181. Infof("libcontainerd: %s is still running", binaryName)
  182. return nil
  183. }
  184. configFile, err := r.getContainerdConfig()
  185. if err != nil {
  186. return err
  187. }
  188. args := []string{"--config", configFile}
  189. cmd := exec.Command(binaryName, args...)
  190. // redirect containerd logs to docker logs
  191. cmd.Stdout = os.Stdout
  192. cmd.Stderr = os.Stderr
  193. cmd.SysProcAttr = containerdSysProcAttr()
  194. // clear the NOTIFY_SOCKET from the env when starting containerd
  195. cmd.Env = nil
  196. for _, e := range os.Environ() {
  197. if !strings.HasPrefix(e, "NOTIFY_SOCKET") {
  198. cmd.Env = append(cmd.Env, e)
  199. }
  200. }
  201. if err := cmd.Start(); err != nil {
  202. return err
  203. }
  204. r.daemonWaitCh = make(chan struct{})
  205. go func() {
  206. // Reap our child when needed
  207. if err := cmd.Wait(); err != nil {
  208. r.logger.WithError(err).Errorf("containerd did not exit successfully")
  209. }
  210. close(r.daemonWaitCh)
  211. }()
  212. r.daemonPid = cmd.Process.Pid
  213. err = ioutil.WriteFile(filepath.Join(r.stateDir, pidFile), []byte(fmt.Sprintf("%d", r.daemonPid)), 0660)
  214. if err != nil {
  215. system.KillProcess(r.daemonPid)
  216. return errors.Wrap(err, "libcontainerd: failed to save daemon pid to disk")
  217. }
  218. logrus.WithField("pid", r.daemonPid).
  219. Infof("libcontainerd: started new %s process", binaryName)
  220. return nil
  221. }
  222. func (r *remote) monitorConnection(client *containerd.Client) {
  223. var transientFailureCount = 0
  224. ticker := time.NewTicker(500 * time.Millisecond)
  225. defer ticker.Stop()
  226. for {
  227. <-ticker.C
  228. ctx, cancel := context.WithTimeout(r.shutdownContext, healthCheckTimeout)
  229. _, err := client.IsServing(ctx)
  230. cancel()
  231. if err == nil {
  232. transientFailureCount = 0
  233. continue
  234. }
  235. select {
  236. case <-r.shutdownContext.Done():
  237. r.logger.Info("stopping healthcheck following graceful shutdown")
  238. client.Close()
  239. return
  240. default:
  241. }
  242. r.logger.WithError(err).WithField("binary", binaryName).Debug("daemon is not responding")
  243. if r.daemonPid != -1 {
  244. transientFailureCount++
  245. if transientFailureCount >= maxConnectionRetryCount || !system.IsProcessAlive(r.daemonPid) {
  246. transientFailureCount = 0
  247. if system.IsProcessAlive(r.daemonPid) {
  248. r.logger.WithField("pid", r.daemonPid).Info("killing and restarting containerd")
  249. // Try to get a stack trace
  250. syscall.Kill(r.daemonPid, syscall.SIGUSR1)
  251. <-time.After(100 * time.Millisecond)
  252. system.KillProcess(r.daemonPid)
  253. }
  254. <-r.daemonWaitCh
  255. var err error
  256. client.Close()
  257. os.Remove(r.GRPC.Address)
  258. if err = r.startContainerd(); err != nil {
  259. r.logger.WithError(err).Error("failed restarting containerd")
  260. } else {
  261. newClient, err := containerd.New(r.GRPC.Address)
  262. if err != nil {
  263. r.logger.WithError(err).Error("failed connect to containerd")
  264. } else {
  265. client = newClient
  266. }
  267. }
  268. }
  269. }
  270. }
  271. }