Merge pull request #125 from gatewayd-io/collect-and-merge-prometheus…

…-metrics-from-plugins

Collect and merge Prometheus metrics from plugins

cmd/run.go

@@ -7,11 +7,14 @@ import (
 	"net/url"
 	"os"
 	"os/signal"
+	"strconv"
 	"syscall"
 
+	"github.com/NYTimes/gziphandler"
 	"github.com/gatewayd-io/gatewayd/config"
 	gerr "github.com/gatewayd-io/gatewayd/errors"
 	"github.com/gatewayd-io/gatewayd/logging"
+	"github.com/gatewayd-io/gatewayd/metrics"
 	"github.com/gatewayd-io/gatewayd/network"
 	"github.com/gatewayd-io/gatewayd/plugin"
 	"github.com/gatewayd-io/gatewayd/pool"
@@ -20,6 +23,7 @@ import (
 	"github.com/knadh/koanf/providers/confmap"
 	"github.com/knadh/koanf/providers/file"
 	"github.com/panjf2000/gnet/v2"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/rs/zerolog"
 	"github.com/spf13/cobra"
@@ -76,6 +80,15 @@ var runCmd = &cobra.Command{
 		// Load plugins and register their hooks.
 		pluginRegistry.LoadPlugins(pConfig.Plugins)
 
+		// Start the metrics merger.
+		metricsMerger := metrics.NewMerger(pConfig.MetricsMergerPeriod, DefaultLogger)
+		pluginRegistry.ForEach(func(_ plugin.Identifier, plugin *plugin.Plugin) {
+			if metricsEnabled, err := strconv.ParseBool(plugin.Config["metricsEnabled"]); err == nil && metricsEnabled {
+				metricsMerger.Add(plugin.ID.Name, plugin.Config["metricsUnixDomainSocket"])
+			}
+		})
+		metricsMerger.Start()
+
 		// Load default global configuration.
 		config.LoadGlobalConfigDefaults(globalConfig)
 
@@ -126,39 +139,62 @@ var runCmd = &cobra.Command{
 			os.Exit(gerr.FailedToLoadGlobalConfig)
 		}
 
-		if gConfig.Metrics[config.Default].Enabled {
-			// Start the metrics server if enabled.
-			go func(
-				gConfig config.GlobalConfig,
-				logger zerolog.Logger,
-				pluginRegistry *plugin.Registry,
-			) {
-				http.Handle(gConfig.Metrics[config.Default].Path, promhttp.Handler())
+		// Start the metrics server if enabled.
+		go func(metricsConfig config.Metrics, logger zerolog.Logger) {
+			// TODO: refactor this to a separate function.
+			if !metricsConfig.Enabled {
+				logger.Info().Msg("Metrics server is disabled")
+				return
+			}
 
-				fqdn, err := url.Parse("http://" + gConfig.Metrics[config.Default].Address)
-				if err != nil {
-					logger.Fatal().Err(err).Msg("Failed to parse metrics address")
-					pluginRegistry.Shutdown()
-					os.Exit(gerr.FailedToStartMetricsServer)
-				}
+			fqdn, err := url.Parse("http://" + metricsConfig.Address)
+			if err != nil {
+				logger.Error().Err(err).Msg("Failed to parse metrics address")
+				return
+			}
 
-				address, err := url.JoinPath(fqdn.String(), gConfig.Metrics[config.Default].Path)
-				if err != nil {
-					logger.Fatal().Err(err).Msg("Failed to parse metrics path")
-					pluginRegistry.Shutdown()
-					os.Exit(gerr.FailedToStartMetricsServer)
-				}
+			address, err := url.JoinPath(fqdn.String(), metricsConfig.Path)
+			if err != nil {
+				logger.Error().Err(err).Msg("Failed to parse metrics path")
+				return
+			}
 
-				logger.Info().Str("address", address).Msg("Metrics are exposed")
-				//nolint:gosec
-				if err = http.ListenAndServe(
-					gConfig.Metrics[config.Default].Address, nil); err != nil {
-					logger.Fatal().Err(err).Msg("Failed to start metrics server")
-					pluginRegistry.Shutdown()
-					os.Exit(gerr.FailedToStartMetricsServer)
+			// Merge the metrics from the plugins with the ones from GatewayD.
+			mergedMetricsHandler := func(next http.Handler) http.Handler {
+				handler := func(w http.ResponseWriter, r *http.Request) {
+					if _, err := w.Write(metricsMerger.OutputMetrics); err != nil {
+						logger.Error().Err(err).Msg("Failed to write metrics")
+					}
+					next.ServeHTTP(w, r)
 				}
-			}(gConfig, DefaultLogger, pluginRegistry)
-		}
+				return http.HandlerFunc(handler)
+			}
+
+			decompressedGatewayDMetricsHandler := func() http.Handler {
+				return promhttp.InstrumentMetricHandler(
+					prometheus.DefaultRegisterer,
+					promhttp.HandlerFor(prometheus.DefaultGatherer, promhttp.HandlerOpts{
+						DisableCompression: true,
+					}),
+				)
+			}
+
+			logger.Info().Str("address", address).Msg("Metrics are exposed")
+			http.Handle(
+				metricsConfig.Path,
+				gziphandler.GzipHandler(
+					mergedMetricsHandler(
+						decompressedGatewayDMetricsHandler(),
+					),
+				),
+			)
+
+			//nolint:gosec
+			if err = http.ListenAndServe(
+				metricsConfig.Address, nil); err != nil {
+				logger.Error().Err(err).Msg("Failed to start metrics server")
+			}
+		}(gConfig.Metrics[config.Default], DefaultLogger)
 
 		// Create a new logger from the config.
 		loggerCfg := gConfig.Loggers[config.Default]
@@ -172,6 +208,7 @@ var runCmd = &cobra.Command{
 
 		// Replace the default logger with the new one from the config.
 		pluginRegistry.Logger = logger
+		metricsMerger.Logger = logger
 
 		// This is a notification hook, so we don't care about the result.
 		data := map[string]interface{}{
@@ -364,6 +401,7 @@ var runCmd = &cobra.Command{
 							logger.Error().Err(err).Msg("Failed to run OnSignal hooks")
 						}
 
+						metricsMerger.Stop()
 						server.Shutdown()
 						pluginRegistry.Shutdown()
 						os.Exit(0)
@@ -375,6 +413,7 @@ var runCmd = &cobra.Command{
 		// Run the server.
 		if err := server.Run(); err != nil {
 			logger.Error().Err(err).Msg("Failed to start server")
+			metricsMerger.Stop()
 			server.Shutdown()
 			pluginRegistry.Shutdown()
 			os.Exit(gerr.FailedToStartServer)

config/constants.go

@@ -61,10 +61,11 @@ const (
 	DefaultCompress    = true
 
 	// Plugin constants.
-	DefaultMinPort      = 50000
-	DefaultMaxPort      = 60000
-	PluginPriorityStart = 1000
-	LoggerName          = "plugin"
+	DefaultMinPort             = 50000
+	DefaultMaxPort             = 60000
+	PluginPriorityStart        = 1000
+	LoggerName                 = "plugin"
+	DefaultMetricsMergerPeriod = 5 * time.Second
 
 	// Client constants.
 	DefaultChunkSize          = 4096

config/types.go

@@ -30,10 +30,11 @@ type Plugin struct {
 }
 
 type PluginConfig struct {
-	VerificationPolicy  string   `koanf:"verificationPolicy"`
-	CompatibilityPolicy string   `koanf:"compatibilityPolicy"`
-	AcceptancePolicy    string   `koanf:"acceptancePolicy"`
-	Plugins             []Plugin `koanf:"plugins"`
+	VerificationPolicy  string        `koanf:"verificationPolicy"`
+	CompatibilityPolicy string        `koanf:"compatibilityPolicy"`
+	AcceptancePolicy    string        `koanf:"acceptancePolicy"`
+	MetricsMergerPeriod time.Duration `koanf:"metricsMergerPeriod"`
+	Plugins             []Plugin      `koanf:"plugins"`
 }
 
 type Client struct {
@@ -175,6 +176,8 @@ func LoadPluginConfigDefaults(cfg *koanf.Koanf) {
 		"plugins": map[string]interface{}{
 			"verificationPolicy":  "passdown",
 			"compatibilityPolicy": "strict",
+			"acceptancePolicy":    "accept",
+			"metricsMergerPeriod": DefaultMetricsMergerPeriod.String(),
 		},
 	}, "")
 

errors/errors.go

@@ -97,9 +97,8 @@ var (
 )
 
 const (
-	FailedToLoadPluginConfig   = 1
-	FailedToLoadGlobalConfig   = 2
-	FailedToInitializePool     = 3
-	FailedToStartMetricsServer = 4
-	FailedToStartServer        = 5
+	FailedToLoadPluginConfig = 1
+	FailedToLoadGlobalConfig = 2
+	FailedToInitializePool   = 3
+	FailedToStartServer      = 4
 )

gatewayd_plugins.yaml

@@ -6,6 +6,7 @@ verificationPolicy: "passdown"
 compatibilityPolicy: "strict"
 # Possible values: "accept" (default) and "reject" 
 acceptancePolicy: "accept"
+metricsMergerPeriod: 5s
 
 plugins:
   # Plugin name
@@ -25,7 +26,7 @@ plugins:
       - MAGIC_COOKIE_KEY=GATEWAYD_PLUGIN
       - MAGIC_COOKIE_VALUE=5712b87aa5d7e9f9e9ab643e6603181c5b796015cb1c09d6f5ada882bf2a1872
     # Checksum hash to verify the binary before loading
-    checksum: b1cfe3c6b268d904abe405b106ef737da5373912586ff8a16695939c38caf30d
+    checksum: 87ec8ea49300f2f5848532ebece5c70ac0aacbdc92b0b909cc5ad72d58e236db
   - name: gatewayd-plugin-cache
     enabled: True
     localPath: ../gatewayd-plugin-cache/gatewayd-plugin-cache
@@ -34,4 +35,4 @@ plugins:
       - MAGIC_COOKIE_KEY=GATEWAYD_PLUGIN
       - MAGIC_COOKIE_VALUE=5712b87aa5d7e9f9e9ab643e6603181c5b796015cb1c09d6f5ada882bf2a1872
       - REDIS_ADDR=localhost:6379
-    checksum: 0c55ed414432afd6db2fbe284112046b6ce4d70b7beb5b99d3703f4bbbcb5a84
+    checksum: c6cf26a4ec2d2a4496e1634692d306c518e36cf55f139d2d47ff554594c7252b

go.mod

@@ -4,27 +4,31 @@ go 1.19
 
 require (
 	github.com/Masterminds/semver/v3 v3.2.0
+	github.com/NYTimes/gziphandler v1.1.1
 	github.com/fergusstrange/embedded-postgres v1.19.0
 	github.com/go-co-op/gocron v1.18.0
 	github.com/google/go-cmp v0.5.9
 	github.com/hashicorp/go-hclog v1.4.0
 	github.com/hashicorp/go-plugin v1.4.8
-	github.com/knadh/koanf v1.4.5
+	github.com/knadh/koanf v1.5.0
 	github.com/mitchellh/mapstructure v1.5.0
-	github.com/panjf2000/gnet/v2 v2.2.2
+	github.com/panjf2000/gnet/v2 v2.2.4
 	github.com/prometheus/client_golang v1.14.0
+	github.com/prometheus/client_model v0.3.0
+	github.com/prometheus/common v0.39.0
 	github.com/rs/zerolog v1.28.0
 	github.com/spf13/cobra v1.6.1
 	github.com/stretchr/testify v1.8.1
 	github.com/zenizh/go-capturer v0.0.0-20211219060012-52ea6c8fed04
-	google.golang.org/grpc v1.51.0
+	golang.org/x/exp v0.0.0-20230118134722-a68e582fa157
+	google.golang.org/grpc v1.52.0
 	google.golang.org/protobuf v1.28.1
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 )
 
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
-	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/fatih/color v1.13.0 // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
@@ -34,25 +38,23 @@ require (
 	github.com/lib/pq v1.10.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.17 // indirect
-	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/go-testing-interface v1.14.1 // 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
-	github.com/prometheus/client_model v0.3.0 // indirect
-	github.com/prometheus/common v0.37.0 // indirect
-	github.com/prometheus/procfs v0.8.0 // indirect
+	github.com/prometheus/procfs v0.9.0 // indirect
 	github.com/robfig/cron/v3 v3.0.1 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/multierr v1.9.0 // indirect
 	go.uber.org/zap v1.24.0 // indirect
 	golang.org/x/net v0.5.0 // indirect
-	golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f // indirect
+	golang.org/x/sync v0.1.0 // indirect
 	golang.org/x/sys v0.4.0 // indirect
 	golang.org/x/text v0.6.0 // indirect
-	google.golang.org/genproto v0.0.0-20230104163317-caabf589fcbf // indirect
+	google.golang.org/genproto v0.0.0-20230119192704-9d59e20e5cd1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )