manager_linux.go 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319
  1. // +build linux
  2. package plugin
  3. import (
  4. "encoding/json"
  5. "fmt"
  6. "net"
  7. "os"
  8. "path/filepath"
  9. "syscall"
  10. "time"
  11. "github.com/Sirupsen/logrus"
  12. "github.com/docker/docker/api/types"
  13. "github.com/docker/docker/daemon/initlayer"
  14. "github.com/docker/docker/libcontainerd"
  15. "github.com/docker/docker/pkg/mount"
  16. "github.com/docker/docker/pkg/plugins"
  17. "github.com/docker/docker/pkg/stringid"
  18. "github.com/docker/docker/plugin/v2"
  19. "github.com/opencontainers/go-digest"
  20. specs "github.com/opencontainers/runtime-spec/specs-go"
  21. "github.com/pkg/errors"
  22. )
  23. func (pm *Manager) enable(p *v2.Plugin, c *controller, force bool) error {
  24. p.Rootfs = filepath.Join(pm.config.Root, p.PluginObj.ID, "rootfs")
  25. if p.IsEnabled() && !force {
  26. return fmt.Errorf("plugin %s is already enabled", p.Name())
  27. }
  28. spec, err := p.InitSpec(pm.config.ExecRoot)
  29. if err != nil {
  30. return err
  31. }
  32. c.restart = true
  33. c.exitChan = make(chan bool)
  34. pm.mu.Lock()
  35. pm.cMap[p] = c
  36. pm.mu.Unlock()
  37. var propRoot string
  38. if p.PropagatedMount != "" {
  39. propRoot = filepath.Join(filepath.Dir(p.Rootfs), "propagated-mount")
  40. if err := os.MkdirAll(propRoot, 0755); err != nil {
  41. logrus.Errorf("failed to create PropagatedMount directory at %s: %v", propRoot, err)
  42. }
  43. if err := mount.MakeRShared(propRoot); err != nil {
  44. return errors.Wrap(err, "error setting up propagated mount dir")
  45. }
  46. if err := mount.Mount(propRoot, p.PropagatedMount, "none", "rbind"); err != nil {
  47. return errors.Wrap(err, "error creating mount for propagated mount")
  48. }
  49. }
  50. if err := initlayer.Setup(filepath.Join(pm.config.Root, p.PluginObj.ID, rootFSFileName), 0, 0); err != nil {
  51. return errors.WithStack(err)
  52. }
  53. if err := pm.containerdClient.Create(p.GetID(), "", "", specs.Spec(*spec), attachToLog(p.GetID())); err != nil {
  54. if p.PropagatedMount != "" {
  55. if err := mount.Unmount(p.PropagatedMount); err != nil {
  56. logrus.Warnf("Could not unmount %s: %v", p.PropagatedMount, err)
  57. }
  58. if err := mount.Unmount(propRoot); err != nil {
  59. logrus.Warnf("Could not unmount %s: %v", propRoot, err)
  60. }
  61. }
  62. return errors.WithStack(err)
  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, c.timeoutInSecs)
  69. if err != nil {
  70. c.restart = false
  71. shutdownPlugin(p, c, pm.containerdClient)
  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.containerdClient)
  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. if err := pm.containerdClient.Restore(p.GetID(), attachToLog(p.GetID())); err != nil {
  103. return err
  104. }
  105. if pm.config.LiveRestoreEnabled {
  106. c := &controller{}
  107. if pids, _ := pm.containerdClient.GetPidsForContainer(p.GetID()); len(pids) == 0 {
  108. // plugin is not running, so follow normal startup procedure
  109. return pm.enable(p, c, true)
  110. }
  111. c.exitChan = make(chan bool)
  112. c.restart = true
  113. pm.mu.Lock()
  114. pm.cMap[p] = c
  115. pm.mu.Unlock()
  116. return pm.pluginPostStart(p, c)
  117. }
  118. return nil
  119. }
  120. func shutdownPlugin(p *v2.Plugin, c *controller, containerdClient libcontainerd.Client) {
  121. pluginID := p.GetID()
  122. err := containerdClient.Signal(pluginID, int(syscall.SIGTERM))
  123. if err != nil {
  124. logrus.Errorf("Sending SIGTERM to plugin failed with error: %v", err)
  125. } else {
  126. select {
  127. case <-c.exitChan:
  128. logrus.Debug("Clean shutdown of plugin")
  129. case <-time.After(time.Second * 10):
  130. logrus.Debug("Force shutdown plugin")
  131. if err := containerdClient.Signal(pluginID, int(syscall.SIGKILL)); err != nil {
  132. logrus.Errorf("Sending SIGKILL to plugin failed with error: %v", err)
  133. }
  134. }
  135. }
  136. }
  137. func (pm *Manager) disable(p *v2.Plugin, c *controller) error {
  138. if !p.IsEnabled() {
  139. return fmt.Errorf("plugin %s is already disabled", p.Name())
  140. }
  141. c.restart = false
  142. shutdownPlugin(p, c, pm.containerdClient)
  143. pm.config.Store.SetState(p, false)
  144. return pm.save(p)
  145. }
  146. // Shutdown stops all plugins and called during daemon shutdown.
  147. func (pm *Manager) Shutdown() {
  148. plugins := pm.config.Store.GetAll()
  149. for _, p := range plugins {
  150. pm.mu.RLock()
  151. c := pm.cMap[p]
  152. pm.mu.RUnlock()
  153. if pm.config.LiveRestoreEnabled && p.IsEnabled() {
  154. logrus.Debug("Plugin active when liveRestore is set, skipping shutdown")
  155. continue
  156. }
  157. if pm.containerdClient != nil && p.IsEnabled() {
  158. c.restart = false
  159. shutdownPlugin(p, c, pm.containerdClient)
  160. }
  161. }
  162. }
  163. func (pm *Manager) upgradePlugin(p *v2.Plugin, configDigest digest.Digest, blobsums []digest.Digest, tmpRootFSDir string, privileges *types.PluginPrivileges) (err error) {
  164. config, err := pm.setupNewPlugin(configDigest, blobsums, privileges)
  165. if err != nil {
  166. return err
  167. }
  168. pdir := filepath.Join(pm.config.Root, p.PluginObj.ID)
  169. orig := filepath.Join(pdir, "rootfs")
  170. // Make sure nothing is mounted
  171. // This could happen if the plugin was disabled with `-f` with active mounts.
  172. // If there is anything in `orig` is still mounted, this should error out.
  173. if err := recursiveUnmount(orig); err != nil {
  174. return err
  175. }
  176. backup := orig + "-old"
  177. if err := os.Rename(orig, backup); err != nil {
  178. return errors.Wrap(err, "error backing up plugin data before upgrade")
  179. }
  180. defer func() {
  181. if err != nil {
  182. if rmErr := os.RemoveAll(orig); rmErr != nil && !os.IsNotExist(rmErr) {
  183. logrus.WithError(rmErr).WithField("dir", backup).Error("error cleaning up after failed upgrade")
  184. return
  185. }
  186. if mvErr := os.Rename(backup, orig); mvErr != nil {
  187. err = errors.Wrap(mvErr, "error restoring old plugin root on upgrade failure")
  188. }
  189. if rmErr := os.RemoveAll(tmpRootFSDir); rmErr != nil && !os.IsNotExist(rmErr) {
  190. logrus.WithError(rmErr).WithField("plugin", p.Name()).Errorf("error cleaning up plugin upgrade dir: %s", tmpRootFSDir)
  191. }
  192. } else {
  193. if rmErr := os.RemoveAll(backup); rmErr != nil && !os.IsNotExist(rmErr) {
  194. logrus.WithError(rmErr).WithField("dir", backup).Error("error cleaning up old plugin root after successful upgrade")
  195. }
  196. p.Config = configDigest
  197. p.Blobsums = blobsums
  198. }
  199. }()
  200. if err := os.Rename(tmpRootFSDir, orig); err != nil {
  201. return errors.Wrap(err, "error upgrading")
  202. }
  203. p.PluginObj.Config = config
  204. err = pm.save(p)
  205. return errors.Wrap(err, "error saving upgraded plugin config")
  206. }
  207. func (pm *Manager) setupNewPlugin(configDigest digest.Digest, blobsums []digest.Digest, privileges *types.PluginPrivileges) (types.PluginConfig, error) {
  208. configRC, err := pm.blobStore.Get(configDigest)
  209. if err != nil {
  210. return types.PluginConfig{}, err
  211. }
  212. defer configRC.Close()
  213. var config types.PluginConfig
  214. dec := json.NewDecoder(configRC)
  215. if err := dec.Decode(&config); err != nil {
  216. return types.PluginConfig{}, errors.Wrapf(err, "failed to parse config")
  217. }
  218. if dec.More() {
  219. return types.PluginConfig{}, errors.New("invalid config json")
  220. }
  221. requiredPrivileges, err := computePrivileges(config)
  222. if err != nil {
  223. return types.PluginConfig{}, err
  224. }
  225. if privileges != nil {
  226. if err := validatePrivileges(requiredPrivileges, *privileges); err != nil {
  227. return types.PluginConfig{}, err
  228. }
  229. }
  230. return config, nil
  231. }
  232. // createPlugin creates a new plugin. take lock before calling.
  233. func (pm *Manager) createPlugin(name string, configDigest digest.Digest, blobsums []digest.Digest, rootFSDir string, privileges *types.PluginPrivileges) (p *v2.Plugin, err error) {
  234. if err := pm.config.Store.validateName(name); err != nil { // todo: this check is wrong. remove store
  235. return nil, err
  236. }
  237. config, err := pm.setupNewPlugin(configDigest, blobsums, privileges)
  238. if err != nil {
  239. return nil, err
  240. }
  241. p = &v2.Plugin{
  242. PluginObj: types.Plugin{
  243. Name: name,
  244. ID: stringid.GenerateRandomID(),
  245. Config: config,
  246. },
  247. Config: configDigest,
  248. Blobsums: blobsums,
  249. }
  250. p.InitEmptySettings()
  251. pdir := filepath.Join(pm.config.Root, p.PluginObj.ID)
  252. if err := os.MkdirAll(pdir, 0700); err != nil {
  253. return nil, errors.Wrapf(err, "failed to mkdir %v", pdir)
  254. }
  255. defer func() {
  256. if err != nil {
  257. os.RemoveAll(pdir)
  258. }
  259. }()
  260. if err := os.Rename(rootFSDir, filepath.Join(pdir, rootFSFileName)); err != nil {
  261. return nil, errors.Wrap(err, "failed to rename rootfs")
  262. }
  263. if err := pm.save(p); err != nil {
  264. return nil, err
  265. }
  266. pm.config.Store.Add(p) // todo: remove
  267. return p, nil
  268. }