support reloadable EnvoyGateway configuration (#4451)

* support reloadable EnvoyGateway configuration

Signed-off-by: zirain <[email protected]>

* lint

Signed-off-by: zirain <[email protected]>

* shutdown wasm http server

Signed-off-by: zirain <[email protected]>

---------

Signed-off-by: zirain <[email protected]>

internal/cmd/server.go

@@ -6,12 +6,15 @@
 package cmd
 
 import (
+	"context"
+
 	"github.com/spf13/cobra"
 	ctrl "sigs.k8s.io/controller-runtime"
 
 	egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1"
 	"github.com/envoyproxy/gateway/internal/admin"
 	"github.com/envoyproxy/gateway/internal/envoygateway/config"
+	"github.com/envoyproxy/gateway/internal/envoygateway/config/loader"
 	extensionregistry "github.com/envoyproxy/gateway/internal/extension/registry"
 	"github.com/envoyproxy/gateway/internal/extension/types"
 	gatewayapirunner "github.com/envoyproxy/gateway/internal/gatewayapi/runner"
@@ -51,6 +54,20 @@ func server() error {
 		return err
 	}
 
+	ctx := ctrl.SetupSignalHandler()
+	hook := func(c context.Context, cfg *config.Server) error {
+		cfg.Logger.Info("Setup runners")
+		if err := setupRunners(c, cfg); err != nil {
+			cfg.Logger.Error(err, "failed to setup runners")
+			return err
+		}
+		return nil
+	}
+	l := loader.New(cfgPath, cfg, hook)
+	if err := l.Start(ctx); err != nil {
+		return err
+	}
+
 	// Init eg admin servers.
 	if err := admin.Init(cfg); err != nil {
 		return err
@@ -60,10 +77,10 @@ func server() error {
 		return err
 	}
 
-	// init eg runners.
-	if err := setupRunners(cfg); err != nil {
-		return err
-	}
+	// Wait exit signal
+	<-ctx.Done()
+
+	cfg.Logger.Info("shutting down")
 
 	return nil
 }
@@ -110,11 +127,7 @@ func getConfigByPath(cfgPath string) (*config.Server, error) {
 
 // setupRunners starts all the runners required for the Envoy Gateway to
 // fulfill its tasks.
-func setupRunners(cfg *config.Server) (err error) {
-	// TODO - Setup a Config Manager
-	// https://github.com/envoyproxy/gateway/issues/43
-	ctx := ctrl.SetupSignalHandler()
-
+func setupRunners(ctx context.Context, cfg *config.Server) (err error) {
 	// Setup the Extension Manager
 	var extMgr types.Manager
 	if cfg.EnvoyGateway.Provider.Type == egv1a1.ProviderTypeKubernetes {
@@ -212,7 +225,7 @@ func setupRunners(cfg *config.Server) (err error) {
 	infraIR.Close()
 	xds.Close()
 
-	cfg.Logger.Info("shutting down")
+	cfg.Logger.Info("runners are shutting down")
 
 	if extMgr != nil {
 		// Close connections to extension services

internal/envoygateway/config/loader/configloader.go

@@ -0,0 +1,113 @@
+// Copyright Envoy Gateway Authors
+// SPDX-License-Identifier: Apache-2.0
+// The full text of the Apache license is available in the LICENSE file at
+// the root of the repo.
+
+package loader
+
+import (
+	"context"
+	"time"
+
+	"github.com/envoyproxy/gateway/internal/envoygateway/config"
+	"github.com/envoyproxy/gateway/internal/filewatcher"
+	"github.com/envoyproxy/gateway/internal/logging"
+)
+
+type HookFunc func(c context.Context, cfg *config.Server) error
+
+type Loader struct {
+	cfgPath string
+	cfg     *config.Server
+	logger  logging.Logger
+	cancel  context.CancelFunc
+	hook    HookFunc
+
+	w filewatcher.FileWatcher
+}
+
+func New(cfgPath string, cfg *config.Server, f HookFunc) *Loader {
+	return &Loader{
+		cfgPath: cfgPath,
+		cfg:     cfg,
+		logger:  cfg.Logger.WithName("config-loader"),
+		hook:    f,
+		w:       filewatcher.NewWatcher(),
+	}
+}
+
+func (r *Loader) Start(ctx context.Context) error {
+	r.runHook()
+
+	if r.cfgPath == "" {
+		r.logger.Info("no config file provided, skipping config watcher")
+		return nil
+	}
+
+	r.logger.Info("watching for changes to the EnvoyGateway configuration", "path", r.cfgPath)
+	if err := r.w.Add(r.cfgPath); err != nil {
+		r.logger.Error(err, "failed to add config file to watcher")
+		return err
+	}
+
+	go func() {
+		defer func() {
+			_ = r.w.Close()
+		}()
+		for {
+			select {
+			case e := <-r.w.Events(r.cfgPath):
+				r.logger.Info("received fsnotify events", "name", e.Name, "op", e.Op.String())
+
+				// Load the config file.
+				eg, err := config.Decode(r.cfgPath)
+				if err != nil {
+					r.logger.Info("failed to decode config file", "name", r.cfgPath, "error", err)
+					// TODO: add a metric for this?
+					continue
+				}
+				// Set defaults for unset fields
+				eg.SetEnvoyGatewayDefaults()
+				r.cfg.EnvoyGateway = eg
+				// update cfg logger
+				eg.Logging.SetEnvoyGatewayLoggingDefaults()
+				r.cfg.Logger = logging.NewLogger(eg.Logging)
+
+				// cancel last
+				if r.cancel != nil {
+					r.cancel()
+				}
+
+				// TODO: we need to make sure that all runners are stopped, before we start the new ones
+				// Otherwise we might end up with error listening on:8081
+				time.Sleep(3 * time.Second)
+
+				r.runHook()
+			case err := <-r.w.Errors(r.cfgPath):
+				r.logger.Error(err, "watcher error")
+			case <-ctx.Done():
+				if r.cancel != nil {
+					r.cancel()
+				}
+				return
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (r *Loader) runHook() {
+	if r.hook == nil {
+		return
+	}
+
+	r.logger.Info("running hook")
+	c, cancel := context.WithCancel(context.TODO())
+	r.cancel = cancel
+	go func(ctx context.Context) {
+		if err := r.hook(ctx, r.cfg); err != nil {
+			r.logger.Error(err, "hook error")
+		}
+	}(c)
+}

internal/envoygateway/config/loader/configloader_test.go

@@ -0,0 +1,59 @@
+// Copyright Envoy Gateway Authors
+// SPDX-License-Identifier: Apache-2.0
+// The full text of the Apache license is available in the LICENSE file at
+// the root of the repo.
+
+package loader
+
+import (
+	"context"
+	_ "embed"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/envoyproxy/gateway/internal/envoygateway/config"
+)
+
+var (
+	//go:embed testdata/default.yaml
+	defaultConfig string
+	//go:embed testdata/enable-redis.yaml
+	redisConfig string
+)
+
+func TestConfigLoader(t *testing.T) {
+	tmpDir, err := os.MkdirTemp("", "envoy-gateway-configloader-test")
+	require.NoError(t, err)
+	defer func(path string) {
+		_ = os.RemoveAll(path)
+	}(tmpDir)
+
+	cfgPath := tmpDir + "/config.yaml"
+	require.NoError(t, os.WriteFile(cfgPath, []byte(defaultConfig), 0o600))
+	s, err := config.New()
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(context.TODO())
+	defer func() {
+		cancel()
+	}()
+
+	changed := 0
+	loader := New(cfgPath, s, func(_ context.Context, cfg *config.Server) error {
+		changed++
+		t.Logf("config changed %d times", changed)
+		if changed > 1 {
+			cancel()
+		}
+		return nil
+	})
+
+	require.NoError(t, loader.Start(ctx))
+	go func() {
+		_ = os.WriteFile(cfgPath, []byte(redisConfig), 0o600)
+	}()
+
+	<-ctx.Done()
+}

internal/envoygateway/config/loader/testdata/default.yaml

@@ -0,0 +1,24 @@
+apiVersion: gateway.envoyproxy.io/v1alpha1
+kind: EnvoyGateway
+gateway:
+  controllerName: gateway.envoyproxy.io/gatewayclass-controller
+logging:
+  level:
+    default: info
+provider:
+  kubernetes:
+    rateLimitDeployment:
+      container:
+        image: docker.io/envoyproxy/ratelimit:master
+      patch:
+        type: StrategicMerge
+        value:
+          spec:
+            template:
+              spec:
+                containers:
+                  - imagePullPolicy: IfNotPresent
+                    name: envoy-ratelimit
+    shutdownManager:
+      image: docker.io/envoyproxy/gateway-dev:latest
+  type: Kubernetes

internal/envoygateway/config/loader/testdata/enable-redis.yaml

@@ -0,0 +1,14 @@
+apiVersion: gateway.envoyproxy.io/v1alpha1
+kind: EnvoyGateway
+provider:
+  type: Kubernetes
+gateway:
+  controllerName: gateway.envoyproxy.io/gatewayclass-controller
+extensionApis:
+  enableEnvoyPatchPolicy: true
+  enableBackend: true
+rateLimit:
+  backend:
+    type: Redis
+    redis:
+      url: redis.redis-system.svc.cluster.local:6379