Merge pull request #606 from ginglis13/metrics-proxy

feat: add snapshotter metrics proxy and service discovery mechanism

Author: ginglis13

snapshotter/app/service.go

@@ -21,6 +21,7 @@ import (
 	"os"
 	"os/signal"
 	"strconv"
+	"strings"
 	"syscall"
 
 	snapshotsapi "github.com/containerd/containerd/api/services/snapshots/v1"
@@ -35,6 +36,8 @@ import (
 	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/config"
 	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux"
 	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/cache"
+	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/metrics"
+	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/metrics/discovery"
 	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/proxy"
 	proxyaddress "github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/proxy/address"
 )
@@ -52,7 +55,8 @@ func Run(config config.Config) error {
 
 	group, ctx := errgroup.WithContext(ctx)
 
-	snapshotter, err := initSnapshotter(ctx, config)
+	cache := cache.NewSnapshotterCache()
+	snapshotter, err := initSnapshotter(ctx, config, cache)
 	if err != nil {
 		log.G(ctx).WithFields(
 			logrus.Fields{"resolver": config.Snapshotter.Proxy.Address.Resolver.Type},
@@ -76,6 +80,15 @@ func Run(config config.Config) error {
 		return err
 	}
 
+	var serviceDiscovery *discovery.ServiceDiscovery
+	if config.Snapshotter.Metrics.Enable {
+		sdPort := config.Snapshotter.Metrics.ServiceDiscoveryPort
+		serviceDiscovery = discovery.NewServiceDiscovery(config.Snapshotter.Metrics.Host, sdPort, cache)
+		group.Go(func() error {
+			return serviceDiscovery.Serve()
+		})
+	}
+
 	group.Go(func() error {
 		return grpcServer.Serve(listener)
 	})
@@ -88,6 +101,11 @@ func Run(config config.Config) error {
 			if err := snapshotter.Close(); err != nil {
 				log.G(ctx).WithError(err).Error("failed to close snapshotter")
 			}
+			if serviceDiscovery != nil {
+				if err := serviceDiscovery.Shutdown(ctx); err != nil {
+					log.G(ctx).WithError(err).Error("failed to shutdown service discovery server")
+				}
+			}
 		}()
 
 		for {
@@ -123,13 +141,13 @@ func initResolver(config config.Config) (proxyaddress.Resolver, error) {
 const base10 = 10
 const bits32 = 32
 
-func initSnapshotter(ctx context.Context, config config.Config) (snapshots.Snapshotter, error) {
+func initSnapshotter(ctx context.Context, config config.Config, cache cache.Cache) (snapshots.Snapshotter, error) {
 	resolver, err := initResolver(config)
 	if err != nil {
 		return nil, err
 	}
 
-	newProxySnapshotterFunc := func(ctx context.Context, namespace string) (snapshots.Snapshotter, error) {
+	newProxySnapshotterFunc := func(ctx context.Context, namespace string) (*proxy.RemoteSnapshotter, error) {
 		r := resolver
 		response, err := r.Get(namespace)
 		if err != nil {
@@ -148,22 +166,58 @@ func initSnapshotter(ctx context.Context, config config.Config) (snapshots.Snaps
 			return vsock.DialContext(ctx, host, uint32(port), vsock.WithLogger(log.G(ctx)))
 		}
 
+		var metricsProxy *metrics.Proxy
+		// TODO (ginglis13): port management and lifecycle ties in to overall metrics proxy
+		// server lifecycle. tracked here: https://github.com/firecracker-microvm/firecracker-containerd/issues/607
+		portMap := make(map[int]bool)
 		if config.Snapshotter.Metrics.Enable {
 			metricsPort, err := strconv.ParseUint(response.MetricsPort, base10, bits32)
 			if err != nil {
 				return nil, err
 			}
 
-			// TODO (ginglis13) metricsDialer func to be defined here using metricsPort. It will dial
-			// the same host but connect via its own port. The metrics proxy will be configured in NewProxySnapshotter
-			// task 2 of https://github.com/firecracker-microvm/firecracker-containerd/issues/602
-			_ = func(ctx context.Context, _ string) (net.Conn, error) {
+			metricsDialer := func(ctx context.Context, _, _ string) (net.Conn, error) {
 				return vsock.DialContext(ctx, host, uint32(metricsPort), vsock.WithLogger(log.G(ctx)))
 			}
+
+			portRange := config.Snapshotter.Metrics.PortRange
+			metricsHost := config.Snapshotter.Metrics.Host
+
+			// Assign a port for metrics proxy server.
+			ports := strings.Split(portRange, "-")
+			portRangeError := fmt.Errorf("invalid port range %s", portRange)
+			if len(ports) < 2 {
+				return nil, portRangeError
+			}
+			lower, err := strconv.Atoi(ports[0])
+			if err != nil {
+				return nil, portRangeError
+			}
+			upper, err := strconv.Atoi(ports[1])
+			if err != nil {
+				return nil, portRangeError
+			}
+			port := -1
+			for p := lower; p <= upper; p++ {
+				if _, ok := portMap[p]; !ok {
+					port = p
+					portMap[p] = true
+					break
+				}
+			}
+			if port < 0 {
+				return nil, fmt.Errorf("invalid port: %d", port)
+			}
+
+			metricsProxy, err = metrics.NewProxy(metricsHost, port, response.Labels, metricsDialer)
+			if err != nil {
+				return nil, err
+			}
+
 		}
 
-		return proxy.NewProxySnapshotter(ctx, host, snapshotterDialer)
+		return proxy.NewProxySnapshotter(ctx, host, snapshotterDialer, metricsProxy)
 	}
 
-	return demux.NewSnapshotter(cache.NewSnapshotterCache(), newProxySnapshotterFunc), nil
+	return demux.NewSnapshotter(cache, newProxySnapshotterFunc), nil
 }

snapshotter/config/config.go

@@ -59,7 +59,10 @@ type debug struct {
 }
 
 type metrics struct {
-	Enable bool `toml:"enable" default:"false"`
+	Enable               bool   `toml:"enable" default:"false"`
+	Host                 string `toml:"host"`
+	PortRange            string `toml:"port_range"`
+	ServiceDiscoveryPort int    `toml:"service_discovery_port"`
 }
 
 // Load parses application configuration from a specified file path.

snapshotter/config/config.toml.example

@@ -8,6 +8,10 @@
 
 [snapshotter.metrics]
   enable = true
+  port_range = "9000-9999"
+  [snapshotter.metrics.service_discovery]
+    enable = true
+    port = 8080
 
 [debug]
   logLevel = "info"

snapshotter/config/config_test.go

@@ -84,7 +84,9 @@ func parseExampleConfig() error {
 	    address = "localhost:10001"
 	  [snapshotter.metrics]
         enable = true
-
+		port_range = "9000-9999"
+		host = "0.0.0.0"
+		service_discovery_port = 8080
 	[debug]
 	  logLevel = "debug"
 	`)
@@ -103,7 +105,10 @@ func parseExampleConfig() error {
 				},
 			},
 			Metrics: metrics{
-				Enable: true,
+				Enable:               true,
+				PortRange:            "9000-9999",
+				Host:                 "0.0.0.0",
+				ServiceDiscoveryPort: 8080,
 			},
 		},
 		Debug: debug{

snapshotter/demux/cache/cache.go

@@ -16,17 +16,17 @@ package cache
 import (
 	"context"
 
-	"github.com/containerd/containerd/snapshots"
+	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/proxy"
 )
 
 // SnapshotterProvider defines a snapshotter fetch function.
-type SnapshotterProvider = func(context.Context, string) (snapshots.Snapshotter, error)
+type SnapshotterProvider = func(context.Context, string) (*proxy.RemoteSnapshotter, error)
 
 // Cache defines the interface for a snapshotter caching mechanism.
 type Cache interface {
 	// Retrieves the snapshotter from the underlying cache using the provided
 	// fetch function if the snapshotter is not currently cached.
-	Get(ctx context.Context, key string, fetch SnapshotterProvider) (snapshots.Snapshotter, error)
+	Get(ctx context.Context, key string, fetch SnapshotterProvider) (*proxy.RemoteSnapshotter, error)
 
 	// Closes the snapshotter and removes it from the cache.
 	Evict(key string) error

snapshotter/demux/cache/snapshotter_cache.go

@@ -18,24 +18,24 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/containerd/containerd/snapshots"
+	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/proxy"
 	"github.com/hashicorp/go-multierror"
 )
 
 // SnapshotterCache implements a read, write protected cache mechanism
 // for keyed snapshotters.
 type SnapshotterCache struct {
 	mutex        *sync.RWMutex
-	snapshotters map[string]snapshots.Snapshotter
+	snapshotters map[string]*proxy.RemoteSnapshotter
 }
 
 // NewSnapshotterCache creates a new instance with an empty cache.
 func NewSnapshotterCache() *SnapshotterCache {
-	return &SnapshotterCache{&sync.RWMutex{}, make(map[string]snapshots.Snapshotter)}
+	return &SnapshotterCache{&sync.RWMutex{}, make(map[string]*proxy.RemoteSnapshotter)}
 }
 
 // Get fetches and caches the snapshotter for a given key.
-func (cache *SnapshotterCache) Get(ctx context.Context, key string, fetch SnapshotterProvider) (snapshots.Snapshotter, error) {
+func (cache *SnapshotterCache) Get(ctx context.Context, key string, fetch SnapshotterProvider) (*proxy.RemoteSnapshotter, error) {
 	snapshotter, ok := cache.snapshotters[key]
 
 	if !ok {
@@ -59,7 +59,7 @@ func (cache *SnapshotterCache) Get(ctx context.Context, key string, fetch Snapsh
 // Evict removes a cached snapshotter for a given key.
 func (cache *SnapshotterCache) Evict(key string) error {
 	cache.mutex.RLock()
-	snapshotter, ok := cache.snapshotters[key]
+	remoteSnapshotter, ok := cache.snapshotters[key]
 	cache.mutex.RUnlock()
 
 	if !ok {
@@ -68,16 +68,16 @@ func (cache *SnapshotterCache) Evict(key string) error {
 	cache.mutex.Lock()
 	defer cache.mutex.Unlock()
 
-	err := snapshotter.Close()
+	err := remoteSnapshotter.Close()
 	delete(cache.snapshotters, key)
 	return err
 }
 
 // Close calls Close on all cached remote snapshotters.
 func (cache *SnapshotterCache) Close() error {
 	var compiledErr error
-	for _, snapshotter := range cache.snapshotters {
-		if err := snapshotter.Close(); err != nil {
+	for _, remoteSnapshotter := range cache.snapshotters {
+		if err := remoteSnapshotter.Close(); err != nil {
 			compiledErr = multierror.Append(compiledErr, err)
 		}
 	}

snapshotter/demux/cache/snapshotter_cache_test.go

@@ -17,23 +17,23 @@ import (
 	"context"
 	"testing"
 
-	"github.com/containerd/containerd/snapshots"
 	"github.com/hashicorp/go-multierror"
 	"github.com/pkg/errors"
 
 	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/internal"
+	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/proxy"
 )
 
-func getSnapshotterOkFunction(ctx context.Context, key string) (snapshots.Snapshotter, error) {
-	return &internal.SuccessfulSnapshotter{}, nil
+func getSnapshotterOkFunction(ctx context.Context, key string) (*proxy.RemoteSnapshotter, error) {
+	return &proxy.RemoteSnapshotter{Snapshotter: &internal.SuccessfulSnapshotter{}}, nil
 }
 
-func getSnapshotterErrorFunction(ctx context.Context, key string) (snapshots.Snapshotter, error) {
-	return &internal.SuccessfulSnapshotter{}, errors.New("MOCK ERROR")
+func getSnapshotterErrorFunction(ctx context.Context, key string) (*proxy.RemoteSnapshotter, error) {
+	return &proxy.RemoteSnapshotter{Snapshotter: &internal.SuccessfulSnapshotter{}}, errors.New("MOCK ERROR")
 }
 
-func getFailingSnapshotterOkFunction(ctx context.Context, key string) (snapshots.Snapshotter, error) {
-	return &internal.FailingSnapshotter{}, nil
+func getFailingSnapshotterOkFunction(ctx context.Context, key string) (*proxy.RemoteSnapshotter, error) {
+	return &proxy.RemoteSnapshotter{Snapshotter: &internal.FailingSnapshotter{}}, nil
 }
 
 func getSnapshotterFromEmptyCache(uut *SnapshotterCache) error {

snapshotter/demux/metrics/discovery/service_discovery.go

@@ -0,0 +1,105 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"). You may
+// not use this file except in compliance with the License. A copy of the
+// License is located at
+//
+//	http://aws.amazon.com/apache2.0/
+//
+// or in the "license" file accompanying this file. This file is distributed
+// on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+// express or implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package discovery
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strconv"
+
+	"github.com/containerd/containerd/log"
+	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/cache"
+)
+
+// ServiceDiscovery discovers metrics services in the snapshotter cache
+// and serves a []MetricsTarget on an HTTP server.
+type ServiceDiscovery struct {
+	// cache is the shared host-level cache of snapshotters also used by the demux snapshotter.
+	cache cache.Cache
+	// server is the HTTP server that returns a list of targets to pull metrics from and apply labels to those metrics.
+	server *http.Server
+}
+
+// NewServiceDiscovery returns a ServiceDiscovery with configured HTTP server and provided cache.
+func NewServiceDiscovery(host string, port int, c cache.Cache) *ServiceDiscovery {
+	return &ServiceDiscovery{
+		server: &http.Server{
+			Addr: host + ":" + strconv.Itoa(port),
+		},
+		cache: c,
+	}
+}
+
+// Serve starts the HTTP server for receiving service discovery requests.
+func (sd *ServiceDiscovery) Serve() error {
+	sd.server.Handler = http.HandlerFunc(sd.serviceDiscoveryHandler)
+	err := sd.server.ListenAndServe()
+	if err != http.ErrServerClosed {
+		return err
+	}
+
+	return nil
+}
+
+// Shutdown shuts down the service discovery HTTP server.
+func (sd *ServiceDiscovery) Shutdown(ctx context.Context) error {
+	return sd.server.Shutdown(ctx)
+}
+
+// metricsTarget represents the underlying JSON structure
+// that a Prometheus HTTP Service Discovery server is
+// expected to return.
+//
+// https://prometheus.io/docs/prometheus/latest/http_sd/
+type metricsTarget struct {
+	Targets []string          `json:"targets"`
+	Labels  map[string]string `json:"labels"`
+}
+
+// serviceDiscoveryHandler scans the cache for snapshotters and their proxy server information,
+// and builds and returns a JSON response in the format of a Prometheus service discovery endpoint.
+func (sd *ServiceDiscovery) serviceDiscoveryHandler(w http.ResponseWriter, req *http.Request) {
+	ctx := req.Context()
+
+	namespaces := sd.cache.List()
+	services := []metricsTarget{}
+	for _, ns := range namespaces {
+		cachedSnapshotter, err := sd.cache.Get(ctx, ns, nil)
+		if err != nil {
+			log.G(ctx).Error("unable to retrieve from snapshotter cache: ", err)
+			continue
+		}
+
+		// make target "localhost:{PORT}" -> the metrics proxy for given snapshotter
+		targetString := fmt.Sprintf("localhost:%v", cachedSnapshotter.MetricsProxyPort())
+		target := []string{targetString}
+
+		// build list of discovered services
+		mt := metricsTarget{Targets: target, Labels: cachedSnapshotter.MetricsProxyLabels()}
+		services = append(services, mt)
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+
+	response, err := json.Marshal(services)
+	if err != nil {
+		log.G(ctx).Error("unable to marshal service discovery response: ", err)
+		w.WriteHeader(500)
+		return
+	}
+
+	w.Write(response)
+}