1package plugin // import "github.com/docker/docker/plugin" 2 3import ( 4 "encoding/json" 5 "io" 6 "io/ioutil" 7 "os" 8 "path/filepath" 9 "reflect" 10 "regexp" 11 "sort" 12 "strings" 13 "sync" 14 15 "github.com/docker/distribution/reference" 16 "github.com/docker/docker/api/types" 17 "github.com/docker/docker/image" 18 "github.com/docker/docker/layer" 19 "github.com/docker/docker/pkg/authorization" 20 "github.com/docker/docker/pkg/ioutils" 21 "github.com/docker/docker/pkg/mount" 22 "github.com/docker/docker/pkg/pubsub" 23 "github.com/docker/docker/pkg/system" 24 "github.com/docker/docker/plugin/v2" 25 "github.com/docker/docker/registry" 26 "github.com/opencontainers/go-digest" 27 "github.com/opencontainers/runtime-spec/specs-go" 28 "github.com/pkg/errors" 29 "github.com/sirupsen/logrus" 30) 31 32const configFileName = "config.json" 33const rootFSFileName = "rootfs" 34 35var validFullID = regexp.MustCompile(`^([a-f0-9]{64})$`) 36 37// Executor is the interface that the plugin manager uses to interact with for starting/stopping plugins 38type Executor interface { 39 Create(id string, spec specs.Spec, stdout, stderr io.WriteCloser) error 40 IsRunning(id string) (bool, error) 41 Restore(id string, stdout, stderr io.WriteCloser) (alive bool, err error) 42 Signal(id string, signal int) error 43} 44 45func (pm *Manager) restorePlugin(p *v2.Plugin, c *controller) error { 46 if p.IsEnabled() { 47 return pm.restore(p, c) 48 } 49 return nil 50} 51 52type eventLogger func(id, name, action string) 53 54// ManagerConfig defines configuration needed to start new manager. 55type ManagerConfig struct { 56 Store *Store // remove 57 RegistryService registry.Service 58 LiveRestoreEnabled bool // TODO: remove 59 LogPluginEvent eventLogger 60 Root string 61 ExecRoot string 62 CreateExecutor ExecutorCreator 63 AuthzMiddleware *authorization.Middleware 64} 65 66// ExecutorCreator is used in the manager config to pass in an `Executor` 67type ExecutorCreator func(*Manager) (Executor, error) 68 69// Manager controls the plugin subsystem. 70type Manager struct { 71 config ManagerConfig 72 mu sync.RWMutex // protects cMap 73 muGC sync.RWMutex // protects blobstore deletions 74 cMap map[*v2.Plugin]*controller 75 blobStore *basicBlobStore 76 publisher *pubsub.Publisher 77 executor Executor 78} 79 80// controller represents the manager's control on a plugin. 81type controller struct { 82 restart bool 83 exitChan chan bool 84 timeoutInSecs int 85} 86 87// pluginRegistryService ensures that all resolved repositories 88// are of the plugin class. 89type pluginRegistryService struct { 90 registry.Service 91} 92 93func (s pluginRegistryService) ResolveRepository(name reference.Named) (repoInfo *registry.RepositoryInfo, err error) { 94 repoInfo, err = s.Service.ResolveRepository(name) 95 if repoInfo != nil { 96 repoInfo.Class = "plugin" 97 } 98 return 99} 100 101// NewManager returns a new plugin manager. 102func NewManager(config ManagerConfig) (*Manager, error) { 103 if config.RegistryService != nil { 104 config.RegistryService = pluginRegistryService{config.RegistryService} 105 } 106 manager := &Manager{ 107 config: config, 108 } 109 for _, dirName := range []string{manager.config.Root, manager.config.ExecRoot, manager.tmpDir()} { 110 if err := os.MkdirAll(dirName, 0700); err != nil { 111 return nil, errors.Wrapf(err, "failed to mkdir %v", dirName) 112 } 113 } 114 var err error 115 manager.executor, err = config.CreateExecutor(manager) 116 if err != nil { 117 return nil, err 118 } 119 120 manager.blobStore, err = newBasicBlobStore(filepath.Join(manager.config.Root, "storage/blobs")) 121 if err != nil { 122 return nil, err 123 } 124 125 manager.cMap = make(map[*v2.Plugin]*controller) 126 if err := manager.reload(); err != nil { 127 return nil, errors.Wrap(err, "failed to restore plugins") 128 } 129 130 manager.publisher = pubsub.NewPublisher(0, 0) 131 return manager, nil 132} 133 134func (pm *Manager) tmpDir() string { 135 return filepath.Join(pm.config.Root, "tmp") 136} 137 138// HandleExitEvent is called when the executor receives the exit event 139// In the future we may change this, but for now all we care about is the exit event. 140func (pm *Manager) HandleExitEvent(id string) error { 141 p, err := pm.config.Store.GetV2Plugin(id) 142 if err != nil { 143 return err 144 } 145 146 if err := os.RemoveAll(filepath.Join(pm.config.ExecRoot, id)); err != nil && !os.IsNotExist(err) { 147 logrus.WithError(err).WithField("id", id).Error("Could not remove plugin bundle dir") 148 } 149 150 pm.mu.RLock() 151 c := pm.cMap[p] 152 if c.exitChan != nil { 153 close(c.exitChan) 154 c.exitChan = nil // ignore duplicate events (containerd issue #2299) 155 } 156 restart := c.restart 157 pm.mu.RUnlock() 158 159 if restart { 160 pm.enable(p, c, true) 161 } else { 162 if err := mount.RecursiveUnmount(filepath.Join(pm.config.Root, id)); err != nil { 163 return errors.Wrap(err, "error cleaning up plugin mounts") 164 } 165 } 166 return nil 167} 168 169func handleLoadError(err error, id string) { 170 if err == nil { 171 return 172 } 173 logger := logrus.WithError(err).WithField("id", id) 174 if os.IsNotExist(errors.Cause(err)) { 175 // Likely some error while removing on an older version of docker 176 logger.Warn("missing plugin config, skipping: this may be caused due to a failed remove and requires manual cleanup.") 177 return 178 } 179 logger.Error("error loading plugin, skipping") 180} 181 182func (pm *Manager) reload() error { // todo: restore 183 dir, err := ioutil.ReadDir(pm.config.Root) 184 if err != nil { 185 return errors.Wrapf(err, "failed to read %v", pm.config.Root) 186 } 187 plugins := make(map[string]*v2.Plugin) 188 for _, v := range dir { 189 if validFullID.MatchString(v.Name()) { 190 p, err := pm.loadPlugin(v.Name()) 191 if err != nil { 192 handleLoadError(err, v.Name()) 193 continue 194 } 195 plugins[p.GetID()] = p 196 } else { 197 if validFullID.MatchString(strings.TrimSuffix(v.Name(), "-removing")) { 198 // There was likely some error while removing this plugin, let's try to remove again here 199 if err := system.EnsureRemoveAll(v.Name()); err != nil { 200 logrus.WithError(err).WithField("id", v.Name()).Warn("error while attempting to clean up previously removed plugin") 201 } 202 } 203 } 204 } 205 206 pm.config.Store.SetAll(plugins) 207 208 var wg sync.WaitGroup 209 wg.Add(len(plugins)) 210 for _, p := range plugins { 211 c := &controller{exitChan: make(chan bool)} 212 pm.mu.Lock() 213 pm.cMap[p] = c 214 pm.mu.Unlock() 215 216 go func(p *v2.Plugin) { 217 defer wg.Done() 218 if err := pm.restorePlugin(p, c); err != nil { 219 logrus.WithError(err).WithField("id", p.GetID()).Error("Failed to restore plugin") 220 return 221 } 222 223 if p.Rootfs != "" { 224 p.Rootfs = filepath.Join(pm.config.Root, p.PluginObj.ID, "rootfs") 225 } 226 227 // We should only enable rootfs propagation for certain plugin types that need it. 228 for _, typ := range p.PluginObj.Config.Interface.Types { 229 if (typ.Capability == "volumedriver" || typ.Capability == "graphdriver") && typ.Prefix == "docker" && strings.HasPrefix(typ.Version, "1.") { 230 if p.PluginObj.Config.PropagatedMount != "" { 231 propRoot := filepath.Join(filepath.Dir(p.Rootfs), "propagated-mount") 232 233 // check if we need to migrate an older propagated mount from before 234 // these mounts were stored outside the plugin rootfs 235 if _, err := os.Stat(propRoot); os.IsNotExist(err) { 236 rootfsProp := filepath.Join(p.Rootfs, p.PluginObj.Config.PropagatedMount) 237 if _, err := os.Stat(rootfsProp); err == nil { 238 if err := os.Rename(rootfsProp, propRoot); err != nil { 239 logrus.WithError(err).WithField("dir", propRoot).Error("error migrating propagated mount storage") 240 } 241 } 242 } 243 244 if err := os.MkdirAll(propRoot, 0755); err != nil { 245 logrus.Errorf("failed to create PropagatedMount directory at %s: %v", propRoot, err) 246 } 247 } 248 } 249 } 250 251 pm.save(p) 252 requiresManualRestore := !pm.config.LiveRestoreEnabled && p.IsEnabled() 253 254 if requiresManualRestore { 255 // if liveRestore is not enabled, the plugin will be stopped now so we should enable it 256 if err := pm.enable(p, c, true); err != nil { 257 logrus.WithError(err).WithField("id", p.GetID()).Error("failed to enable plugin") 258 } 259 } 260 }(p) 261 } 262 wg.Wait() 263 return nil 264} 265 266// Get looks up the requested plugin in the store. 267func (pm *Manager) Get(idOrName string) (*v2.Plugin, error) { 268 return pm.config.Store.GetV2Plugin(idOrName) 269} 270 271func (pm *Manager) loadPlugin(id string) (*v2.Plugin, error) { 272 p := filepath.Join(pm.config.Root, id, configFileName) 273 dt, err := ioutil.ReadFile(p) 274 if err != nil { 275 return nil, errors.Wrapf(err, "error reading %v", p) 276 } 277 var plugin v2.Plugin 278 if err := json.Unmarshal(dt, &plugin); err != nil { 279 return nil, errors.Wrapf(err, "error decoding %v", p) 280 } 281 return &plugin, nil 282} 283 284func (pm *Manager) save(p *v2.Plugin) error { 285 pluginJSON, err := json.Marshal(p) 286 if err != nil { 287 return errors.Wrap(err, "failed to marshal plugin json") 288 } 289 if err := ioutils.AtomicWriteFile(filepath.Join(pm.config.Root, p.GetID(), configFileName), pluginJSON, 0600); err != nil { 290 return errors.Wrap(err, "failed to write atomically plugin json") 291 } 292 return nil 293} 294 295// GC cleans up unreferenced blobs. This is recommended to run in a goroutine 296func (pm *Manager) GC() { 297 pm.muGC.Lock() 298 defer pm.muGC.Unlock() 299 300 whitelist := make(map[digest.Digest]struct{}) 301 for _, p := range pm.config.Store.GetAll() { 302 whitelist[p.Config] = struct{}{} 303 for _, b := range p.Blobsums { 304 whitelist[b] = struct{}{} 305 } 306 } 307 308 pm.blobStore.gc(whitelist) 309} 310 311type logHook struct{ id string } 312 313func (logHook) Levels() []logrus.Level { 314 return logrus.AllLevels 315} 316 317func (l logHook) Fire(entry *logrus.Entry) error { 318 entry.Data = logrus.Fields{"plugin": l.id} 319 return nil 320} 321 322func makeLoggerStreams(id string) (stdout, stderr io.WriteCloser) { 323 logger := logrus.New() 324 logger.Hooks.Add(logHook{id}) 325 return logger.WriterLevel(logrus.InfoLevel), logger.WriterLevel(logrus.ErrorLevel) 326} 327 328func validatePrivileges(requiredPrivileges, privileges types.PluginPrivileges) error { 329 if !isEqual(requiredPrivileges, privileges, isEqualPrivilege) { 330 return errors.New("incorrect privileges") 331 } 332 333 return nil 334} 335 336func isEqual(arrOne, arrOther types.PluginPrivileges, compare func(x, y types.PluginPrivilege) bool) bool { 337 if len(arrOne) != len(arrOther) { 338 return false 339 } 340 341 sort.Sort(arrOne) 342 sort.Sort(arrOther) 343 344 for i := 1; i < arrOne.Len(); i++ { 345 if !compare(arrOne[i], arrOther[i]) { 346 return false 347 } 348 } 349 350 return true 351} 352 353func isEqualPrivilege(a, b types.PluginPrivilege) bool { 354 if a.Name != b.Name { 355 return false 356 } 357 358 return reflect.DeepEqual(a.Value, b.Value) 359} 360 361func configToRootFS(c []byte) (*image.RootFS, error) { 362 var pluginConfig types.PluginConfig 363 if err := json.Unmarshal(c, &pluginConfig); err != nil { 364 return nil, err 365 } 366 // validation for empty rootfs is in distribution code 367 if pluginConfig.Rootfs == nil { 368 return nil, nil 369 } 370 371 return rootFSFromPlugin(pluginConfig.Rootfs), nil 372} 373 374func rootFSFromPlugin(pluginfs *types.PluginConfigRootfs) *image.RootFS { 375 rootFS := image.RootFS{ 376 Type: pluginfs.Type, 377 DiffIDs: make([]layer.DiffID, len(pluginfs.DiffIds)), 378 } 379 for i := range pluginfs.DiffIds { 380 rootFS.DiffIDs[i] = layer.DiffID(pluginfs.DiffIds[i]) 381 } 382 383 return &rootFS 384} 385