123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317 |
- // +build !windows
- package libcontainerd
- import (
- "context"
- "fmt"
- "io"
- "io/ioutil"
- "os"
- "os/exec"
- "path/filepath"
- "strconv"
- "strings"
- "sync"
- "syscall"
- "time"
- "github.com/BurntSushi/toml"
- "github.com/containerd/containerd"
- "github.com/containerd/containerd/server"
- "github.com/docker/docker/pkg/system"
- "github.com/pkg/errors"
- "github.com/sirupsen/logrus"
- )
- const (
- maxConnectionRetryCount = 3
- healthCheckTimeout = 3 * time.Second
- shutdownTimeout = 15 * time.Second
- configFile = "containerd.toml"
- binaryName = "docker-containerd"
- pidFile = "docker-containerd.pid"
- )
- type pluginConfigs struct {
- Plugins map[string]interface{} `toml:"plugins"`
- }
- type remote struct {
- sync.RWMutex
- server.Config
- daemonPid int
- logger *logrus.Entry
- daemonWaitCh chan struct{}
- clients []*client
- shutdownContext context.Context
- shutdownCancel context.CancelFunc
- shutdown bool
- // Options
- startDaemon bool
- rootDir string
- stateDir string
- snapshotter string
- pluginConfs pluginConfigs
- }
- // New creates a fresh instance of libcontainerd remote.
- func New(rootDir, stateDir string, options ...RemoteOption) (rem Remote, err error) {
- defer func() {
- if err != nil {
- err = errors.Wrap(err, "Failed to connect to containerd")
- }
- }()
- r := &remote{
- rootDir: rootDir,
- stateDir: stateDir,
- Config: server.Config{
- Root: filepath.Join(rootDir, "daemon"),
- State: filepath.Join(stateDir, "daemon"),
- },
- pluginConfs: pluginConfigs{make(map[string]interface{})},
- daemonPid: -1,
- logger: logrus.WithField("module", "libcontainerd"),
- }
- r.shutdownContext, r.shutdownCancel = context.WithCancel(context.Background())
- rem = r
- for _, option := range options {
- if err = option.Apply(r); err != nil {
- return
- }
- }
- r.setDefaults()
- if err = system.MkdirAll(stateDir, 0700, ""); err != nil {
- return
- }
- if r.startDaemon {
- os.Remove(r.GRPC.Address)
- if err = r.startContainerd(); err != nil {
- return
- }
- defer func() {
- if err != nil {
- r.Cleanup()
- }
- }()
- }
- // This connection is just used to monitor the connection
- client, err := containerd.New(r.GRPC.Address)
- if err != nil {
- return
- }
- if _, err := client.Version(context.Background()); err != nil {
- system.KillProcess(r.daemonPid)
- return nil, errors.Wrapf(err, "unable to get containerd version")
- }
- go r.monitorConnection(client)
- return r, nil
- }
- func (r *remote) NewClient(ns string, b Backend) (Client, error) {
- c := &client{
- stateDir: r.stateDir,
- logger: r.logger.WithField("namespace", ns),
- namespace: ns,
- backend: b,
- containers: make(map[string]*container),
- }
- rclient, err := containerd.New(r.GRPC.Address, containerd.WithDefaultNamespace(ns))
- if err != nil {
- return nil, err
- }
- c.remote = rclient
- go c.processEventStream(r.shutdownContext)
- r.Lock()
- r.clients = append(r.clients, c)
- r.Unlock()
- return c, nil
- }
- func (r *remote) Cleanup() {
- if r.daemonPid != -1 {
- r.shutdownCancel()
- r.stopDaemon()
- }
- // cleanup some files
- os.Remove(filepath.Join(r.stateDir, pidFile))
- r.platformCleanup()
- }
- func (r *remote) getContainerdPid() (int, error) {
- pidFile := filepath.Join(r.stateDir, pidFile)
- f, err := os.OpenFile(pidFile, os.O_RDWR, 0600)
- if err != nil {
- if os.IsNotExist(err) {
- return -1, nil
- }
- return -1, err
- }
- defer f.Close()
- b := make([]byte, 8)
- n, err := f.Read(b)
- if err != nil && err != io.EOF {
- return -1, err
- }
- if n > 0 {
- pid, err := strconv.ParseUint(string(b[:n]), 10, 64)
- if err != nil {
- return -1, err
- }
- if system.IsProcessAlive(int(pid)) {
- return int(pid), nil
- }
- }
- return -1, nil
- }
- func (r *remote) getContainerdConfig() (string, error) {
- path := filepath.Join(r.stateDir, configFile)
- f, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0600)
- if err != nil {
- return "", errors.Wrapf(err, "failed to open containerd config file at %s", path)
- }
- defer f.Close()
- enc := toml.NewEncoder(f)
- if err = enc.Encode(r.Config); err != nil {
- return "", errors.Wrapf(err, "failed to encode general config")
- }
- if err = enc.Encode(r.pluginConfs); err != nil {
- return "", errors.Wrapf(err, "failed to encode plugin configs")
- }
- return path, nil
- }
- func (r *remote) startContainerd() error {
- pid, err := r.getContainerdPid()
- if err != nil {
- return err
- }
- if pid != -1 {
- r.daemonPid = pid
- logrus.WithField("pid", pid).
- Infof("libcontainerd: %s is still running", binaryName)
- return nil
- }
- configFile, err := r.getContainerdConfig()
- if err != nil {
- return err
- }
- args := []string{"--config", configFile}
- cmd := exec.Command(binaryName, args...)
- // redirect containerd logs to docker logs
- cmd.Stdout = os.Stdout
- cmd.Stderr = os.Stderr
- cmd.SysProcAttr = containerdSysProcAttr()
- // clear the NOTIFY_SOCKET from the env when starting containerd
- cmd.Env = nil
- for _, e := range os.Environ() {
- if !strings.HasPrefix(e, "NOTIFY_SOCKET") {
- cmd.Env = append(cmd.Env, e)
- }
- }
- if err := cmd.Start(); err != nil {
- return err
- }
- r.daemonWaitCh = make(chan struct{})
- go func() {
- // Reap our child when needed
- if err := cmd.Wait(); err != nil {
- r.logger.WithError(err).Errorf("containerd did not exit successfully")
- }
- close(r.daemonWaitCh)
- }()
- r.daemonPid = cmd.Process.Pid
- err = ioutil.WriteFile(filepath.Join(r.stateDir, pidFile), []byte(fmt.Sprintf("%d", r.daemonPid)), 0660)
- if err != nil {
- system.KillProcess(r.daemonPid)
- return errors.Wrap(err, "libcontainerd: failed to save daemon pid to disk")
- }
- logrus.WithField("pid", r.daemonPid).
- Infof("libcontainerd: started new %s process", binaryName)
- return nil
- }
- func (r *remote) monitorConnection(client *containerd.Client) {
- var transientFailureCount = 0
- ticker := time.NewTicker(500 * time.Millisecond)
- defer ticker.Stop()
- for {
- <-ticker.C
- ctx, cancel := context.WithTimeout(r.shutdownContext, healthCheckTimeout)
- _, err := client.IsServing(ctx)
- cancel()
- if err == nil {
- transientFailureCount = 0
- continue
- }
- select {
- case <-r.shutdownContext.Done():
- r.logger.Info("stopping healthcheck following graceful shutdown")
- client.Close()
- return
- default:
- }
- r.logger.WithError(err).WithField("binary", binaryName).Debug("daemon is not responding")
- if r.daemonPid != -1 {
- transientFailureCount++
- if transientFailureCount >= maxConnectionRetryCount || !system.IsProcessAlive(r.daemonPid) {
- transientFailureCount = 0
- if system.IsProcessAlive(r.daemonPid) {
- r.logger.WithField("pid", r.daemonPid).Info("killing and restarting containerd")
- // Try to get a stack trace
- syscall.Kill(r.daemonPid, syscall.SIGUSR1)
- <-time.After(100 * time.Millisecond)
- system.KillProcess(r.daemonPid)
- }
- <-r.daemonWaitCh
- var err error
- client.Close()
- os.Remove(r.GRPC.Address)
- if err = r.startContainerd(); err != nil {
- r.logger.WithError(err).Error("failed restarting containerd")
- } else {
- newClient, err := containerd.New(r.GRPC.Address)
- if err != nil {
- r.logger.WithError(err).Error("failed connect to containerd")
- } else {
- client = newClient
- }
- }
- }
- }
- }
- }
|