manager_linux.go 9.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336
  1. package plugin
  2. import (
  3. "encoding/json"
  4. "net"
  5. "os"
  6. "path/filepath"
  7. "time"
  8. "github.com/docker/docker/api/types"
  9. "github.com/docker/docker/daemon/initlayer"
  10. "github.com/docker/docker/errdefs"
  11. "github.com/docker/docker/pkg/containerfs"
  12. "github.com/docker/docker/pkg/idtools"
  13. "github.com/docker/docker/pkg/mount"
  14. "github.com/docker/docker/pkg/plugins"
  15. "github.com/docker/docker/pkg/stringid"
  16. "github.com/docker/docker/plugin/v2"
  17. "github.com/opencontainers/go-digest"
  18. "github.com/pkg/errors"
  19. "github.com/sirupsen/logrus"
  20. "golang.org/x/sys/unix"
  21. )
  22. func (pm *Manager) enable(p *v2.Plugin, c *controller, force bool) (err error) {
  23. p.Rootfs = filepath.Join(pm.config.Root, p.PluginObj.ID, "rootfs")
  24. if p.IsEnabled() && !force {
  25. return errors.Wrap(enabledError(p.Name()), "plugin already enabled")
  26. }
  27. spec, err := p.InitSpec(pm.config.ExecRoot)
  28. if err != nil {
  29. return err
  30. }
  31. c.restart = true
  32. c.exitChan = make(chan bool)
  33. pm.mu.Lock()
  34. pm.cMap[p] = c
  35. pm.mu.Unlock()
  36. var propRoot string
  37. if p.PropagatedMount != "" {
  38. propRoot = filepath.Join(filepath.Dir(p.Rootfs), "propagated-mount")
  39. if err = os.MkdirAll(propRoot, 0755); err != nil {
  40. logrus.Errorf("failed to create PropagatedMount directory at %s: %v", propRoot, err)
  41. }
  42. if err = mount.MakeRShared(propRoot); err != nil {
  43. return errors.Wrap(err, "error setting up propagated mount dir")
  44. }
  45. if err = mount.Mount(propRoot, p.PropagatedMount, "none", "rbind"); err != nil {
  46. return errors.Wrap(err, "error creating mount for propagated mount")
  47. }
  48. }
  49. rootFS := containerfs.NewLocalContainerFS(filepath.Join(pm.config.Root, p.PluginObj.ID, rootFSFileName))
  50. if err := initlayer.Setup(rootFS, idtools.IDPair{0, 0}); err != nil {
  51. return errors.WithStack(err)
  52. }
  53. stdout, stderr := makeLoggerStreams(p.GetID())
  54. if err := pm.executor.Create(p.GetID(), *spec, stdout, stderr); err != nil {
  55. if p.PropagatedMount != "" {
  56. if err := mount.Unmount(p.PropagatedMount); err != nil {
  57. logrus.Warnf("Could not unmount %s: %v", p.PropagatedMount, err)
  58. }
  59. if err := mount.Unmount(propRoot); err != nil {
  60. logrus.Warnf("Could not unmount %s: %v", propRoot, err)
  61. }
  62. }
  63. }
  64. return pm.pluginPostStart(p, c)
  65. }
  66. func (pm *Manager) pluginPostStart(p *v2.Plugin, c *controller) error {
  67. sockAddr := filepath.Join(pm.config.ExecRoot, p.GetID(), p.GetSocket())
  68. client, err := plugins.NewClientWithTimeout("unix://"+sockAddr, nil, time.Duration(c.timeoutInSecs)*time.Second)
  69. if err != nil {
  70. c.restart = false
  71. shutdownPlugin(p, c, pm.executor)
  72. return errors.WithStack(err)
  73. }
  74. p.SetPClient(client)
  75. // Initial sleep before net Dial to allow plugin to listen on socket.
  76. time.Sleep(500 * time.Millisecond)
  77. maxRetries := 3
  78. var retries int
  79. for {
  80. // net dial into the unix socket to see if someone's listening.
  81. conn, err := net.Dial("unix", sockAddr)
  82. if err == nil {
  83. conn.Close()
  84. break
  85. }
  86. time.Sleep(3 * time.Second)
  87. retries++
  88. if retries > maxRetries {
  89. logrus.Debugf("error net dialing plugin: %v", err)
  90. c.restart = false
  91. // While restoring plugins, we need to explicitly set the state to disabled
  92. pm.config.Store.SetState(p, false)
  93. shutdownPlugin(p, c, pm.executor)
  94. return err
  95. }
  96. }
  97. pm.config.Store.SetState(p, true)
  98. pm.config.Store.CallHandler(p)
  99. return pm.save(p)
  100. }
  101. func (pm *Manager) restore(p *v2.Plugin) error {
  102. stdout, stderr := makeLoggerStreams(p.GetID())
  103. if err := pm.executor.Restore(p.GetID(), stdout, stderr); err != nil {
  104. return err
  105. }
  106. if pm.config.LiveRestoreEnabled {
  107. c := &controller{}
  108. if isRunning, _ := pm.executor.IsRunning(p.GetID()); !isRunning {
  109. // plugin is not running, so follow normal startup procedure
  110. return pm.enable(p, c, true)
  111. }
  112. c.exitChan = make(chan bool)
  113. c.restart = true
  114. pm.mu.Lock()
  115. pm.cMap[p] = c
  116. pm.mu.Unlock()
  117. return pm.pluginPostStart(p, c)
  118. }
  119. return nil
  120. }
  121. func shutdownPlugin(p *v2.Plugin, c *controller, executor Executor) {
  122. pluginID := p.GetID()
  123. err := executor.Signal(pluginID, int(unix.SIGTERM))
  124. if err != nil {
  125. logrus.Errorf("Sending SIGTERM to plugin failed with error: %v", err)
  126. } else {
  127. select {
  128. case <-c.exitChan:
  129. logrus.Debug("Clean shutdown of plugin")
  130. case <-time.After(time.Second * 10):
  131. logrus.Debug("Force shutdown plugin")
  132. if err := executor.Signal(pluginID, int(unix.SIGKILL)); err != nil {
  133. logrus.Errorf("Sending SIGKILL to plugin failed with error: %v", err)
  134. }
  135. select {
  136. case <-c.exitChan:
  137. logrus.Debug("SIGKILL plugin shutdown")
  138. case <-time.After(time.Second * 10):
  139. logrus.Debug("Force shutdown plugin FAILED")
  140. }
  141. }
  142. }
  143. }
  144. func setupRoot(root string) error {
  145. if err := mount.MakePrivate(root); err != nil {
  146. return errors.Wrap(err, "error setting plugin manager root to private")
  147. }
  148. return nil
  149. }
  150. func (pm *Manager) disable(p *v2.Plugin, c *controller) error {
  151. if !p.IsEnabled() {
  152. return errors.Wrap(errDisabled(p.Name()), "plugin is already disabled")
  153. }
  154. c.restart = false
  155. shutdownPlugin(p, c, pm.executor)
  156. pm.config.Store.SetState(p, false)
  157. return pm.save(p)
  158. }
  159. // Shutdown stops all plugins and called during daemon shutdown.
  160. func (pm *Manager) Shutdown() {
  161. plugins := pm.config.Store.GetAll()
  162. for _, p := range plugins {
  163. pm.mu.RLock()
  164. c := pm.cMap[p]
  165. pm.mu.RUnlock()
  166. if pm.config.LiveRestoreEnabled && p.IsEnabled() {
  167. logrus.Debug("Plugin active when liveRestore is set, skipping shutdown")
  168. continue
  169. }
  170. if pm.executor != nil && p.IsEnabled() {
  171. c.restart = false
  172. shutdownPlugin(p, c, pm.executor)
  173. }
  174. }
  175. mount.Unmount(pm.config.Root)
  176. }
  177. func (pm *Manager) upgradePlugin(p *v2.Plugin, configDigest digest.Digest, blobsums []digest.Digest, tmpRootFSDir string, privileges *types.PluginPrivileges) (err error) {
  178. config, err := pm.setupNewPlugin(configDigest, blobsums, privileges)
  179. if err != nil {
  180. return err
  181. }
  182. pdir := filepath.Join(pm.config.Root, p.PluginObj.ID)
  183. orig := filepath.Join(pdir, "rootfs")
  184. // Make sure nothing is mounted
  185. // This could happen if the plugin was disabled with `-f` with active mounts.
  186. // If there is anything in `orig` is still mounted, this should error out.
  187. if err := mount.RecursiveUnmount(orig); err != nil {
  188. return errdefs.System(err)
  189. }
  190. backup := orig + "-old"
  191. if err := os.Rename(orig, backup); err != nil {
  192. return errors.Wrap(errdefs.System(err), "error backing up plugin data before upgrade")
  193. }
  194. defer func() {
  195. if err != nil {
  196. if rmErr := os.RemoveAll(orig); rmErr != nil && !os.IsNotExist(rmErr) {
  197. logrus.WithError(rmErr).WithField("dir", backup).Error("error cleaning up after failed upgrade")
  198. return
  199. }
  200. if mvErr := os.Rename(backup, orig); mvErr != nil {
  201. err = errors.Wrap(mvErr, "error restoring old plugin root on upgrade failure")
  202. }
  203. if rmErr := os.RemoveAll(tmpRootFSDir); rmErr != nil && !os.IsNotExist(rmErr) {
  204. logrus.WithError(rmErr).WithField("plugin", p.Name()).Errorf("error cleaning up plugin upgrade dir: %s", tmpRootFSDir)
  205. }
  206. } else {
  207. if rmErr := os.RemoveAll(backup); rmErr != nil && !os.IsNotExist(rmErr) {
  208. logrus.WithError(rmErr).WithField("dir", backup).Error("error cleaning up old plugin root after successful upgrade")
  209. }
  210. p.Config = configDigest
  211. p.Blobsums = blobsums
  212. }
  213. }()
  214. if err := os.Rename(tmpRootFSDir, orig); err != nil {
  215. return errors.Wrap(errdefs.System(err), "error upgrading")
  216. }
  217. p.PluginObj.Config = config
  218. err = pm.save(p)
  219. return errors.Wrap(err, "error saving upgraded plugin config")
  220. }
  221. func (pm *Manager) setupNewPlugin(configDigest digest.Digest, blobsums []digest.Digest, privileges *types.PluginPrivileges) (types.PluginConfig, error) {
  222. configRC, err := pm.blobStore.Get(configDigest)
  223. if err != nil {
  224. return types.PluginConfig{}, err
  225. }
  226. defer configRC.Close()
  227. var config types.PluginConfig
  228. dec := json.NewDecoder(configRC)
  229. if err := dec.Decode(&config); err != nil {
  230. return types.PluginConfig{}, errors.Wrapf(err, "failed to parse config")
  231. }
  232. if dec.More() {
  233. return types.PluginConfig{}, errors.New("invalid config json")
  234. }
  235. requiredPrivileges := computePrivileges(config)
  236. if err != nil {
  237. return types.PluginConfig{}, err
  238. }
  239. if privileges != nil {
  240. if err := validatePrivileges(requiredPrivileges, *privileges); err != nil {
  241. return types.PluginConfig{}, err
  242. }
  243. }
  244. return config, nil
  245. }
  246. // createPlugin creates a new plugin. take lock before calling.
  247. func (pm *Manager) createPlugin(name string, configDigest digest.Digest, blobsums []digest.Digest, rootFSDir string, privileges *types.PluginPrivileges, opts ...CreateOpt) (p *v2.Plugin, err error) {
  248. if err := pm.config.Store.validateName(name); err != nil { // todo: this check is wrong. remove store
  249. return nil, errdefs.InvalidParameter(err)
  250. }
  251. config, err := pm.setupNewPlugin(configDigest, blobsums, privileges)
  252. if err != nil {
  253. return nil, err
  254. }
  255. p = &v2.Plugin{
  256. PluginObj: types.Plugin{
  257. Name: name,
  258. ID: stringid.GenerateRandomID(),
  259. Config: config,
  260. },
  261. Config: configDigest,
  262. Blobsums: blobsums,
  263. }
  264. p.InitEmptySettings()
  265. for _, o := range opts {
  266. o(p)
  267. }
  268. pdir := filepath.Join(pm.config.Root, p.PluginObj.ID)
  269. if err := os.MkdirAll(pdir, 0700); err != nil {
  270. return nil, errors.Wrapf(err, "failed to mkdir %v", pdir)
  271. }
  272. defer func() {
  273. if err != nil {
  274. os.RemoveAll(pdir)
  275. }
  276. }()
  277. if err := os.Rename(rootFSDir, filepath.Join(pdir, rootFSFileName)); err != nil {
  278. return nil, errors.Wrap(err, "failed to rename rootfs")
  279. }
  280. if err := pm.save(p); err != nil {
  281. return nil, err
  282. }
  283. pm.config.Store.Add(p) // todo: remove
  284. return p, nil
  285. }