Merge pull request #6474 from ipfs/feat/builtin-plugin

feat: make it easier to load custom plugins

cmd/ipfs/main.go

@@ -48,18 +48,20 @@ const (
 func loadPlugins(repoPath string) (*loader.PluginLoader, error) {
 	pluginpath := filepath.Join(repoPath, "plugins")
 
+	plugins, err := loader.NewPluginLoader()
+	if err != nil {
+		return nil, fmt.Errorf("error loading preloaded plugins: %s", err)
+	}
+
 	// check if repo is accessible before loading plugins
-	var plugins *loader.PluginLoader
 	ok, err := checkPermissions(repoPath)
 	if err != nil {
 		return nil, err
 	}
-	if !ok {
-		pluginpath = ""
-	}
-	plugins, err = loader.NewPluginLoader(pluginpath)
-	if err != nil {
-		return nil, fmt.Errorf("error loading plugins: %s", err)
+	if ok {
+		if err := plugins.LoadDirectory(pluginpath); err != nil {
+			return nil, err
+		}
 	}
 
 	if err := plugins.Initialize(); err != nil {

plugin/loader/loader.go

@@ -21,43 +21,123 @@ var loadPluginsFunc = func(string) ([]plugin.Plugin, error) {
 	return nil, nil
 }
 
-// PluginLoader keeps track of loaded plugins
+type loaderState int
+
+const (
+	loaderLoading loaderState = iota
+	loaderInitializing
+	loaderInitialized
+	loaderInjecting
+	loaderInjected
+	loaderStarting
+	loaderStarted
+	loaderClosing
+	loaderClosed
+	loaderFailed
+)
+
+func (ls loaderState) String() string {
+	switch ls {
+	case loaderLoading:
+		return "Loading"
+	case loaderInitializing:
+		return "Initializing"
+	case loaderInitialized:
+		return "Initialized"
+	case loaderInjecting:
+		return "Injecting"
+	case loaderInjected:
+		return "Injected"
+	case loaderStarting:
+		return "Starting"
+	case loaderStarted:
+		return "Started"
+	case loaderClosing:
+		return "Closing"
+	case loaderClosed:
+		return "Closed"
+	case loaderFailed:
+		return "Failed"
+	default:
+		return "Unknown"
+	}
+}
+
+// PluginLoader keeps track of loaded plugins.
+//
+// To use:
+// 1. Load any desired plugins with Load and LoadDirectory. Preloaded plugins
+//    will automatically be loaded.
+// 2. Call Initialize to run all initialization logic.
+// 3. Call Inject to register the plugins.
+// 4. Optionally call Start to start plugins.
+// 5. Call Close to close all plugins.
 type PluginLoader struct {
-	plugins []plugin.Plugin
+	state   loaderState
+	plugins map[string]plugin.Plugin
+	started []plugin.Plugin
 }
 
 // NewPluginLoader creates new plugin loader
-func NewPluginLoader(pluginDir string) (*PluginLoader, error) {
-	plMap := make(map[string]plugin.Plugin)
+func NewPluginLoader() (*PluginLoader, error) {
+	loader := &PluginLoader{plugins: make(map[string]plugin.Plugin, len(preloadPlugins))}
 	for _, v := range preloadPlugins {
-		plMap[v.Name()] = v
-	}
-
-	if pluginDir != "" {
-		newPls, err := loadDynamicPlugins(pluginDir)
-		if err != nil {
+		if err := loader.Load(v); err != nil {
 			return nil, err
 		}
+	}
+	return loader, nil
+}
 
-		for _, pl := range newPls {
-			if ppl, ok := plMap[pl.Name()]; ok {
-				// plugin is already preloaded
-				return nil, fmt.Errorf(
-					"plugin: %s, is duplicated in version: %s, "+
-						"while trying to load dynamically: %s",
-					ppl.Name(), ppl.Version(), pl.Version())
-			}
-			plMap[pl.Name()] = pl
-		}
+func (loader *PluginLoader) assertState(state loaderState) error {
+	if loader.state != state {
+		return fmt.Errorf("loader state must be %s, was %s", state, loader.state)
 	}
+	return nil
+}
 
-	loader := &PluginLoader{plugins: make([]plugin.Plugin, 0, len(plMap))}
+func (loader *PluginLoader) transition(from, to loaderState) error {
+	if err := loader.assertState(from); err != nil {
+		return err
+	}
+	loader.state = to
+	return nil
+}
 
-	for _, v := range plMap {
-		loader.plugins = append(loader.plugins, v)
+// Load loads a plugin into the plugin loader.
+func (loader *PluginLoader) Load(pl plugin.Plugin) error {
+	if err := loader.assertState(loaderLoading); err != nil {
+		return err
 	}
 
-	return loader, nil
+	name := pl.Name()
+	if ppl, ok := loader.plugins[name]; ok {
+		// plugin is already loaded
+		return fmt.Errorf(
+			"plugin: %s, is duplicated in version: %s, "+
+				"while trying to load dynamically: %s",
+			name, ppl.Version(), pl.Version())
+	}
+	loader.plugins[name] = pl
+	return nil
+}
+
+// LoadDirectory loads a directory of plugins into the plugin loader.
+func (loader *PluginLoader) LoadDirectory(pluginDir string) error {
+	if err := loader.assertState(loaderLoading); err != nil {
+		return err
+	}
+	newPls, err := loadDynamicPlugins(pluginDir)
+	if err != nil {
+		return err
+	}
+
+	for _, pl := range newPls {
+		if err := loader.Load(pl); err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 func loadDynamicPlugins(pluginDir string) ([]plugin.Plugin, error) {
@@ -74,63 +154,85 @@ func loadDynamicPlugins(pluginDir string) ([]plugin.Plugin, error) {
 
 // Initialize initializes all loaded plugins
 func (loader *PluginLoader) Initialize() error {
+	if err := loader.transition(loaderLoading, loaderInitializing); err != nil {
+		return err
+	}
 	for _, p := range loader.plugins {
 		err := p.Init()
 		if err != nil {
+			loader.state = loaderFailed
 			return err
 		}
 	}
 
-	return nil
+	return loader.transition(loaderInitializing, loaderInitialized)
 }
 
 // Inject hooks all the plugins into the appropriate subsystems.
 func (loader *PluginLoader) Inject() error {
+	if err := loader.transition(loaderInitialized, loaderInjecting); err != nil {
+		return err
+	}
+
 	for _, pl := range loader.plugins {
 		if pl, ok := pl.(plugin.PluginIPLD); ok {
 			err := injectIPLDPlugin(pl)
 			if err != nil {
+				loader.state = loaderFailed
 				return err
 			}
 		}
 		if pl, ok := pl.(plugin.PluginTracer); ok {
 			err := injectTracerPlugin(pl)
 			if err != nil {
+				loader.state = loaderFailed
 				return err
 			}
 		}
 		if pl, ok := pl.(plugin.PluginDatastore); ok {
 			err := injectDatastorePlugin(pl)
 			if err != nil {
+				loader.state = loaderFailed
 				return err
 			}
 		}
 	}
-	return nil
+
+	return loader.transition(loaderInjecting, loaderInjected)
 }
 
 // Start starts all long-running plugins.
 func (loader *PluginLoader) Start(iface coreiface.CoreAPI) error {
-	for i, pl := range loader.plugins {
+	if err := loader.transition(loaderInjected, loaderStarting); err != nil {
+		return err
+	}
+	for _, pl := range loader.plugins {
 		if pl, ok := pl.(plugin.PluginDaemon); ok {
 			err := pl.Start(iface)
 			if err != nil {
-				_ = closePlugins(loader.plugins[:i])
+				_ = loader.Close()
 				return err
 			}
+			loader.started = append(loader.started, pl)
 		}
 	}
-	return nil
+
+	return loader.transition(loaderStarting, loaderStarted)
 }
 
 // StopDaemon stops all long-running plugins.
 func (loader *PluginLoader) Close() error {
-	return closePlugins(loader.plugins)
-}
+	switch loader.state {
+	case loaderClosing, loaderFailed, loaderClosed:
+		// nothing to do.
+		return nil
+	}
+	loader.state = loaderClosing
 
-func closePlugins(plugins []plugin.Plugin) error {
 	var errs []string
-	for _, pl := range plugins {
+	started := loader.started
+	loader.started = nil
+	for _, pl := range started {
 		if pl, ok := pl.(plugin.PluginDaemon); ok {
 			err := pl.Close()
 			if err != nil {
@@ -143,8 +245,10 @@ func closePlugins(plugins []plugin.Plugin) error {
 		}
 	}
 	if errs != nil {
+		loader.state = loaderFailed
 		return fmt.Errorf(strings.Join(errs, "\n"))
 	}
+	loader.state = loaderClosed
 	return nil
 }
 

repo/fsrepo/config_test.go

@@ -75,7 +75,7 @@ var measureConfig = []byte(`{
 }`)
 
 func TestDefaultDatastoreConfig(t *testing.T) {
-	loader, err := loader.NewPluginLoader("")
+	loader, err := loader.NewPluginLoader()
 	if err != nil {
 		t.Fatal(err)
 	}