Plugin system

buf.gen.yaml

@@ -3,12 +3,12 @@ version: v1
 #   enabled: true
 plugins:
   - name: go-grpc
-    out: gen/proto/go
+    out: .
     opt: paths=source_relative
   - name: go
-    out: gen/proto/go
+    out: .
     opt: paths=source_relative
   # - name: python_betterproto
-  #   out: gen/proto/python_betterproto
+  #   out: .
   # - name: java
-  #   out: gen/proto/java
+  #   out: .

cmd/config_parser.go

@@ -15,6 +15,9 @@ import (
 // Global koanf instance. Using "." as the key path delimiter.
 var globalConfig = koanf.New(".")
 
+// Plugin koanf instance. Using "." as the key path delimiter.
+var pluginConfig = koanf.New(".")
+
 func getPath(path string) string {
 	ref := globalConfig.String(path)
 	if globalConfig.Exists(path) && globalConfig.StringMap(ref) != nil {

cmd/run.go

@@ -1,40 +1,66 @@
 package cmd
 
 import (
+	"context"
 	"os"
 	"os/signal"
 	"syscall"
 	"time"
 
+	gerr "github.com/gatewayd-io/gatewayd/errors"
 	"github.com/gatewayd-io/gatewayd/logging"
 	"github.com/gatewayd-io/gatewayd/network"
 	"github.com/gatewayd-io/gatewayd/plugin"
 	"github.com/gatewayd-io/gatewayd/pool"
-	"github.com/knadh/koanf"
+	goplugin "github.com/hashicorp/go-plugin"
 	"github.com/knadh/koanf/parsers/yaml"
 	"github.com/knadh/koanf/providers/confmap"
 	"github.com/knadh/koanf/providers/file"
 	"github.com/panjf2000/gnet/v2"
+	"github.com/rs/zerolog"
 	"github.com/spf13/cobra"
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
 const (
 	DefaultTCPKeepAlive = 3 * time.Second
 )
 
 var (
-	configFile  string
-	hooksConfig = plugin.NewHookConfig()
+	globalConfigFile string
+	pluginConfigFile string
+)
+
+var (
+	hooksConfig    = plugin.NewHookConfig()
+	DefaultLogger  = logging.NewLogger(logging.LoggerConfig{Level: zerolog.DebugLevel})
+	pluginRegistry = plugin.NewRegistry(hooksConfig)
 )
 
 // runCmd represents the run command.
 var runCmd = &cobra.Command{
 	Use:   "run",
 	Short: "Run a gatewayd instance",
 	Run: func(cmd *cobra.Command, args []string) {
+		// The plugins are loaded and hooks registered
+		// before the configuration is loaded.
+		hooksConfig.Logger = DefaultLogger
+
+		// Load the plugin configuration file
+		if f, err := cmd.Flags().GetString("plugin-config"); err == nil {
+			if err := pluginConfig.Load(file.Provider(f), yaml.Parser()); err != nil {
+				DefaultLogger.Fatal().Err(err).Msg("Failed to load plugin configuration")
+				os.Exit(gerr.FailedToLoadPluginConfig)
+			}
+		}
+
+		// Load plugins and register their hooks
+		pluginRegistry.LoadPlugins(pluginConfig)
+
 		if f, err := cmd.Flags().GetString("config"); err == nil {
 			if err := globalConfig.Load(file.Provider(f), yaml.Parser()); err != nil {
-				panic(err)
+				DefaultLogger.Fatal().Err(err).Msg("Failed to load configuration")
+				os.Exit(gerr.FailedToLoadGlobalConfig)
 			}
 		}
 
@@ -43,39 +69,47 @@ var runCmd = &cobra.Command{
 
 		// The config will be passed to the hooks, and in turn to the plugins that
 		// register to this hook.
-		result := hooksConfig.Run(
-			plugin.OnConfigLoaded,
-			plugin.Signature{"config": globalConfig.All()},
-			hooksConfig.Verification)
-		if result != nil {
-			var config map[string]interface{}
-			if cfg, ok := result["config"].(map[string]interface{}); ok {
-				config = cfg
-			}
-
-			if config != nil {
-				// Load the config from the map emitted by the hook
-				var hookEmittedConfig *koanf.Koanf
-				if err := hookEmittedConfig.Load(confmap.Provider(config, "."), nil); err != nil {
-					// Since the logger is not yet initialized, we can't log the error.
-					// So we panic. Same happens in the next if statement.
-					panic(err)
-				}
+		currentGlobalConfig, err := structpb.NewStruct(globalConfig.All())
+		if err != nil {
+			DefaultLogger.Error().Err(err).Msg("Failed to convert configuration to structpb")
+		} else {
+			updatedGlobalConfig, _ := hooksConfig.Run(
+				context.Background(),
+				currentGlobalConfig,
+				plugin.OnConfigLoaded,
+				hooksConfig.Verification)
 
+			if updatedGlobalConfig != nil && plugin.Verify(updatedGlobalConfig, currentGlobalConfig) {
 				// Merge the config with the one loaded from the file (in memory).
 				// The changes won't be persisted to disk.
-				if err := globalConfig.Merge(hookEmittedConfig); err != nil {
-					panic(err)
+				if err := globalConfig.Load(
+					confmap.Provider(updatedGlobalConfig.AsMap(), "."), nil); err != nil {
+					DefaultLogger.Fatal().Err(err).Msg("Failed to merge configuration")
 				}
 			}
 		}
 
 		// Create a new logger from the config
-		logger := logging.NewLogger(loggerConfig())
-		hooksConfig.Logger = logger
+		loggerCfg := loggerConfig()
+		logger := logging.NewLogger(loggerCfg)
+		// TODO: Use https://github.com/dcarbone/zadapters to adapt hclog to zerolog
 		// This is a notification hook, so we don't care about the result.
-		hooksConfig.Run(
-			plugin.OnNewLogger, plugin.Signature{"logger": logger}, hooksConfig.Verification)
+		data, err := structpb.NewStruct(map[string]interface{}{
+			"timeFormat": loggerCfg.TimeFormat,
+			"level":      loggerCfg.Level,
+			"output":     loggerCfg.Output,
+			"noColor":    loggerCfg.NoColor,
+		})
+		if err != nil {
+			logger.Error().Err(err).Msg("Failed to convert logger config to structpb")
+		} else {
+			// TODO: Use a context with a timeout
+			_, err := hooksConfig.Run(
+				context.Background(), data, plugin.OnNewLogger, hooksConfig.Verification)
+			if err != nil {
+				logger.Error().Err(err).Msg("Failed to run OnNewLogger hooks")
+			}
+		}
 
 		// Create and initialize a pool of connections
 		pool := pool.NewPool()
@@ -90,15 +124,26 @@ var runCmd = &cobra.Command{
 				logger,
 			)
 
-			hooksConfig.Run(
-				plugin.OnNewClient,
-				plugin.Signature{
-					"client": client,
-				},
-				hooksConfig.Verification,
-			)
-
 			if client != nil {
+				clientCfg, err := structpb.NewStruct(map[string]interface{}{
+					"id":                client.ID,
+					"network":           clientConfig.Network,
+					"address":           clientConfig.Address,
+					"receiveBufferSize": clientConfig.ReceiveBufferSize,
+				})
+				if err != nil {
+					logger.Error().Err(err).Msg("Failed to convert client config to structpb")
+				} else {
+					_, err := hooksConfig.Run(
+						context.Background(),
+						clientCfg,
+						plugin.OnNewClient,
+						hooksConfig.Verification)
+					if err != nil {
+						logger.Error().Err(err).Msg("Failed to run OnNewClient hooks")
+					}
+				}
+
 				pool.Put(client.ID, client)
 			}
 		}
@@ -113,15 +158,38 @@ var runCmd = &cobra.Command{
 			os.Exit(1)
 		}
 
-		hooksConfig.Run(
-			plugin.OnNewPool, plugin.Signature{"pool": pool}, hooksConfig.Verification)
+		poolCfg, err := structpb.NewStruct(map[string]interface{}{
+			"size": poolSize,
+		})
+		if err != nil {
+			logger.Error().Err(err).Msg("Failed to convert pool config to structpb")
+		} else {
+			_, err := hooksConfig.Run(
+				context.Background(), poolCfg, plugin.OnNewPool, hooksConfig.Verification)
+			if err != nil {
+				logger.Error().Err(err).Msg("Failed to run OnNewPool hooks")
+			}
+		}
 
 		// Create a prefork proxy with the pool of clients
 		elastic, reuseElasticClients, elasticClientConfig := proxyConfig()
 		proxy := network.NewProxy(
 			pool, hooksConfig, elastic, reuseElasticClients, elasticClientConfig, logger)
-		hooksConfig.Run(
-			plugin.OnNewProxy, plugin.Signature{"proxy": proxy}, hooksConfig.Verification)
+
+		proxyCfg, err := structpb.NewStruct(map[string]interface{}{
+			"elastic":             elastic,
+			"reuseElasticClients": reuseElasticClients,
+			"clientConfig":        elasticClientConfig,
+		})
+		if err != nil {
+			logger.Error().Err(err).Msg("Failed to convert proxy config to structpb")
+		} else {
+			_, err := hooksConfig.Run(
+				context.Background(), proxyCfg, plugin.OnNewProxy, hooksConfig.Verification)
+			if err != nil {
+				logger.Error().Err(err).Msg("Failed to run OnNewProxy hooks")
+			}
+		}
 
 		// Create a server
 		serverConfig := serverConfig()
@@ -166,9 +234,35 @@ var runCmd = &cobra.Command{
 			logger,
 			hooksConfig,
 		)
-		hooksConfig.Run(
-			plugin.OnNewServer, plugin.Signature{"server": server}, hooksConfig.Verification)
 
+		serverCfg, err := structpb.NewStruct(map[string]interface{}{
+			"network":          serverConfig.Network,
+			"address":          serverConfig.Address,
+			"softLimit":        serverConfig.SoftLimit,
+			"hardLimit":        serverConfig.HardLimit,
+			"tickInterval":     serverConfig.TickInterval,
+			"multiCore":        serverConfig.MultiCore,
+			"lockOSThread":     serverConfig.LockOSThread,
+			"enableTicker":     serverConfig.EnableTicker,
+			"loadBalancer":     serverConfig.LoadBalancer,
+			"readBufferCap":    serverConfig.ReadBufferCap,
+			"writeBufferCap":   serverConfig.WriteBufferCap,
+			"socketRecvBuffer": serverConfig.SocketRecvBuffer,
+			"socketSendBuffer": serverConfig.SocketSendBuffer,
+			"reuseAddress":     serverConfig.ReuseAddress,
+			"reusePort":        serverConfig.ReusePort,
+			"tcpKeepAlive":     serverConfig.TCPKeepAlive,
+			"tcpNoDelay":       serverConfig.TCPNoDelay,
+		})
+		if err != nil {
+			logger.Error().Err(err).Msg("Failed to convert server config to structpb")
+		} else {
+			_, err := hooksConfig.Run(
+				context.Background(), serverCfg, plugin.OnNewServer, hooksConfig.Verification)
+			if err != nil {
+				logger.Error().Err(err).Msg("Failed to run OnNewServer hooks")
+			}
+		}
 		// Shutdown the server gracefully
 		var signals []os.Signal
 		signals = append(signals,
@@ -187,10 +281,25 @@ var runCmd = &cobra.Command{
 				for _, s := range signals {
 					if sig != s {
 						// Notify the hooks that the server is shutting down
-						hooksConfig.Run(
-							plugin.OnSignal, plugin.Signature{"signal": sig}, hooksConfig.Verification)
+						signalCfg, err := structpb.NewStruct(map[string]interface{}{"signal": sig})
+						if err != nil {
+							logger.Error().Err(err).Msg(
+								"Failed to convert signal config to structpb")
+						} else {
+							_, err := hooksConfig.Run(
+								context.Background(),
+								signalCfg,
+								plugin.OnSignal,
+								hooksConfig.Verification,
+							)
+							if err != nil {
+								logger.Error().Err(err).Msg("Failed to run OnSignal hooks")
+							}
+						}
 
 						server.Shutdown()
+						pluginRegistry.Shutdown()
+						goplugin.CleanupClients()
 						os.Exit(0)
 					}
 				}
@@ -208,5 +317,11 @@ func init() {
 	rootCmd.AddCommand(runCmd)
 
 	runCmd.PersistentFlags().StringVarP(
-		&configFile, "config", "c", "./gatewayd.yaml", "config file (default is ./gatewayd.yaml)")
+		&globalConfigFile,
+		"config", "c", "./gatewayd.yaml",
+		"config file (default is ./gatewayd.yaml)")
+	runCmd.PersistentFlags().StringVarP(
+		&pluginConfigFile,
+		"plugin-config", "p", "./gatewayd_plugins.yaml",
+		"plugin config file (default is ./gatewayd_plugins.yaml)")
 }

network/errors.go → errors/errors.go

@@ -1,4 +1,4 @@
-package network
+package errors
 
 import "errors"
 
@@ -7,4 +7,12 @@ var (
 	ErrNetworkNotSupported = errors.New("network is not supported")
 	ErrClientNotConnected  = errors.New("client is not connected")
 	ErrPoolExhausted       = errors.New("pool is exhausted")
+
+	ErrPluginNotFound = errors.New("plugin not found")
+	ErrPluginNotReady = errors.New("plugin is not ready")
+)
+
+const (
+	FailedToLoadPluginConfig = 1
+	FailedToLoadGlobalConfig = 2
 )

gatewayd_plugins.yaml

@@ -0,0 +1,4 @@
+gatewayd-plugin-test:
+  enabled: True
+  localPath: ../gatewayd-plugin-test/gatewayd-plugin-test
+  checksum: 9c12267332a609ed7c739098a0aa7614f580ba174f35f611fc1cc53de5647bff

go.mod

@@ -3,17 +3,17 @@ module github.com/gatewayd-io/gatewayd
 go 1.19
 
 require (
-	github.com/Masterminds/semver/v3 v3.2.0
 	github.com/fergusstrange/embedded-postgres v1.19.0
-	github.com/gatewayd-io/gatewayd-plugin-test v0.0.0-00010101000000-000000000000
 	github.com/google/go-cmp v0.5.9
 	github.com/hashicorp/go-plugin v1.4.8
 	github.com/knadh/koanf v1.4.4
+	github.com/mitchellh/mapstructure v1.5.0
 	github.com/panjf2000/gnet/v2 v2.2.0
 	github.com/rs/zerolog v1.28.0
 	github.com/spf13/cobra v1.6.1
 	github.com/stretchr/testify v1.8.1
 	google.golang.org/grpc v1.51.0
+	google.golang.org/protobuf v1.28.1
 )
 
 require (
@@ -29,7 +29,6 @@ require (
 	github.com/mattn/go-isatty v0.0.16 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/go-testing-interface v1.14.1 // indirect
-	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/oklog/run v1.1.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
@@ -42,7 +41,6 @@ require (
 	golang.org/x/sys v0.3.0 // indirect
 	golang.org/x/text v0.5.0 // indirect
 	google.golang.org/genproto v0.0.0-20221207170731-23e4bf6bdc37 // indirect
-	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -2,8 +2,6 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMT
 cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
-github.com/Masterminds/semver/v3 v3.2.0 h1:3MEsd0SM6jqZojhjLWWeBY+Kcjy9i6MQAeY7YgDP83g=
-github.com/Masterminds/semver/v3 v3.2.0/go.mod h1:qvl/7zhW3nngYb5+80sSMF+FG2BjYrf8m9wsX0PNOMQ=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=