Repository: firecracker-microvm/firecracker-containerd
Pull request #606 (2022-04-13 23:06:05)

feat: add snapshotter metrics proxy and service discovery mechanism
Signed-off-by: Gavin Inglis <giinglis@amazon.com>

*Issue #, if available:*

Addresses tasks 2 and 3 (metrics proxy / service discovery) of #602 


*Description of changes:*

Snapshotters handled by the the demux snapshotter may expose metrics
endpoints. These changes modify the newProxySnapshotterFunc to
optionally start a metrics proxy server when a new proxy snapshotter is
initialized. The demux snapshotter's cache has been modified to contain
a map of `RemoteSnapshotter`s, which are simply wrappers around the
containerd snapshotter construct and a new type for metrics proxies.

These changes also introduce a mechanism for discovering these proxied
snapshotter metrics for the purposes of aggregating these metrics.


By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.



Patch:
diff --git a/Makefile b/Makefile
index 4b5bd86..65afcbd 100644
--- a/Makefile
+++ b/Makefile
@@ -294,12 +294,12 @@ install-test-cni-bins: test-cni-bins $(CNI_BIN_ROOT)
 	install -D -o root -g root -m755 -t $(CNI_BIN_ROOT) $(TEST_BRIDGED_TAP_BIN)
 
 FCNET_CONFIG?=/etc/cni/conf.d/fcnet.conflist
-$(FCNET_CONFIG): tools/demo/fcnet.conflist
+$(FCNET_CONFIG):
 	mkdir -p $(dir $(FCNET_CONFIG))
 	install -o root -g root -m644 tools/demo/fcnet.conflist $(FCNET_CONFIG)
 
 FCNET_BRIDGE_CONFIG?=/etc/network/interfaces.d/fc-br0
-$(FCNET_BRIDGE_CONFIG): tools/demo/fc-br0.interface
+$(FCNET_BRIDGE_CONFIG):
 	mkdir -p $(dir $(FCNET_BRIDGE_CONFIG))
 	install -o root -g root -m644 tools/demo/fc-br0.interface $(FCNET_BRIDGE_CONFIG)
 
diff --git a/snapshotter/app/service.go b/snapshotter/app/service.go
index e05bc86..8734769 100644
--- a/snapshotter/app/service.go
+++ b/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
 }
diff --git a/snapshotter/config/config.go b/snapshotter/config/config.go
index fd047f4..3b42b0f 100644
--- a/snapshotter/config/config.go
+++ b/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.
diff --git a/snapshotter/config/config.toml.example b/snapshotter/config/config.toml.example
index 5110600..4abd0f8 100644
--- a/snapshotter/config/config.toml.example
+++ b/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"
diff --git a/snapshotter/demux/cache/cache.go b/snapshotter/demux/cache/cache.go
index 4b811e6..1250144 100644
--- a/snapshotter/demux/cache/cache.go
+++ b/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
diff --git a/snapshotter/demux/cache/snapshotter_cache.go b/snapshotter/demux/cache/snapshotter_cache.go
index 5f039a9..761cbf8 100644
--- a/snapshotter/demux/cache/snapshotter_cache.go
+++ b/snapshotter/demux/cache/snapshotter_cache.go
@@ -18,7 +18,7 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/containerd/containerd/snapshots"
+	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/proxy"
 	"github.com/hashicorp/go-multierror"
 )
 
@@ -26,16 +26,16 @@ import (
 // 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,7 +68,7 @@ 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
 }
@@ -76,8 +76,8 @@ func (cache *SnapshotterCache) Evict(key string) error {
 // 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)
 		}
 	}
diff --git a/snapshotter/demux/metrics/discovery/service_discovery.go b/snapshotter/demux/metrics/discovery/service_discovery.go
new file mode 100644
index 0000000..5cbd707
--- /dev/null
+++ b/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)
+}
diff --git a/snapshotter/demux/metrics/proxy.go b/snapshotter/demux/metrics/proxy.go
new file mode 100644
index 0000000..c86398e
--- /dev/null
+++ b/snapshotter/demux/metrics/proxy.go
@@ -0,0 +1,104 @@
+// 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 metrics
+
+import (
+	"context"
+	"io"
+	"net"
+	"net/http"
+	"strconv"
+
+	"github.com/containerd/containerd/log"
+)
+
+const (
+	// RequestPath is the path in the microVM hosting remote snapshotter metrics.
+	RequestPath = "http://localhost/metrics"
+	// MaxMetricsResponseSize is the limit in bytes for size of a metrics response from a remote snapshotter.
+	MaxMetricsResponseSize = 32768 // 32 KB
+)
+
+// Proxy represents a metrics proxy server for getting and serving remote snapshotter metrics.
+type Proxy struct {
+	// client is the HTTP client used to send metrics requests to a remote snapshotter.
+	client *http.Client
+	// server is the proxy server on host for serving remote snapshotter metrics.
+	server *http.Server
+	// host is the nonroutable address listening for metrics requests.
+	host string
+	// Port is the port that the metrics proxy server listens on.
+	Port int
+	// Labels are labels used to apply to metrics.
+	Labels map[string]string
+}
+
+// NewProxy creates a new Proxy with initialized HTTP client and port.
+// dialer is used as the DialContext underlying the metrics HTTP client's RoundTripper.
+//
+// Reference: https://pkg.go.dev/net/http#Transport
+func NewProxy(host string, port int, labels map[string]string, dialer func(context.Context, string, string) (net.Conn, error)) (*Proxy, error) {
+	return &Proxy{
+		client: &http.Client{
+			Transport: &http.Transport{
+				DialContext: dialer,
+			},
+		},
+		server: nil,
+		host:   host,
+		Port:   port,
+		Labels: labels,
+	}, nil
+}
+
+// Serve starts the metrics proxy server for a single in-VM snapshotter.
+func (mp *Proxy) Serve(ctx context.Context) error {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/metrics", mp.metrics)
+
+	mp.server = &http.Server{
+		Addr:    mp.host + ":" + strconv.Itoa(mp.Port),
+		Handler: mux,
+	}
+
+	err := mp.server.ListenAndServe()
+	if err != http.ErrServerClosed {
+		return err
+	}
+
+	return nil
+}
+
+// Shutdown shuts down the remote snapshotter metrics proxy server.
+func (mp *Proxy) Shutdown(ctx context.Context) error {
+	return mp.server.Shutdown(ctx)
+}
+
+// metrics is an HTTP handler that returns metrics for a given VM/snapshotter.
+func (mp *Proxy) metrics(w http.ResponseWriter, req *http.Request) {
+	ctx := req.Context()
+
+	// Pull metrics and copy to HTTP response.
+	res, err := mp.client.Get(RequestPath)
+	if err != nil {
+		log.G(ctx).Errorf("error reading response body: %v\n", err)
+		w.WriteHeader(500)
+		return
+	}
+
+	defer res.Body.Close()
+	response := io.LimitReader(res.Body, MaxMetricsResponseSize)
+
+	io.Copy(w, response)
+}
diff --git a/snapshotter/demux/proxy/address/resolver.go b/snapshotter/demux/proxy/address/resolver.go
index 1bc3f9f..e302594 100644
--- a/snapshotter/demux/proxy/address/resolver.go
+++ b/snapshotter/demux/proxy/address/resolver.go
@@ -28,6 +28,9 @@ type Response struct {
 
 	// MetricsPort is the port used in vsock.DialContext for sending metrics requests to the remote snapshotter.
 	MetricsPort string `json:"metrics_port"`
+
+	// Labels is a map used for applying labels to metrics.
+	Labels map[string]string `json:"labels"`
 }
 
 // Resolver for the proxy network address.
diff --git a/snapshotter/demux/proxy/snapshotter.go b/snapshotter/demux/proxy/snapshotter.go
index 3831688..ed1cd84 100644
--- a/snapshotter/demux/proxy/snapshotter.go
+++ b/snapshotter/demux/proxy/snapshotter.go
@@ -20,6 +20,8 @@ import (
 	snapshotsapi "github.com/containerd/containerd/api/services/snapshots/v1"
 	"github.com/containerd/containerd/snapshots"
 	"github.com/containerd/containerd/snapshots/proxy"
+	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/metrics"
+	"github.com/hashicorp/go-multierror"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
@@ -28,8 +30,31 @@ import (
 // SnapshotterDialer defines an interface for establishing a network connection.
 type SnapshotterDialer = func(context.Context, *logrus.Entry, string, uint32)
 
+// RemoteSnapshotter embeds a snapshots.Snapshotter and its metrics proxy.
+type RemoteSnapshotter struct {
+	snapshots.Snapshotter
+	metricsProxy *metrics.Proxy
+}
+
+// Close closes the remote snapshotter's snapshotter and shuts down its metrics proxy server.
+func (rs *RemoteSnapshotter) Close() error {
+	var compiledErr error
+	if err := rs.Snapshotter.Close(); err != nil {
+		compiledErr = multierror.Append(compiledErr, err)
+	}
+	if rs.metricsProxy != nil {
+		if err := rs.metricsProxy.Shutdown(context.Background()); err != nil {
+			compiledErr = multierror.Append(compiledErr, err)
+		}
+	}
+
+	return compiledErr
+}
+
 // NewProxySnapshotter creates a proxy snapshotter using gRPC over vsock connection.
-func NewProxySnapshotter(ctx context.Context, address string, dialer func(context.Context, string) (net.Conn, error)) (snapshots.Snapshotter, error) {
+func NewProxySnapshotter(ctx context.Context, address string,
+	dialer func(context.Context, string) (net.Conn, error), metricsProxy *metrics.Proxy) (*RemoteSnapshotter, error) {
+
 	opts := []grpc.DialOption{
 		grpc.WithContextDialer(dialer),
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
@@ -39,5 +64,24 @@ func NewProxySnapshotter(ctx context.Context, address string, dialer func(contex
 	if err != nil {
 		return nil, err
 	}
-	return proxy.NewSnapshotter(snapshotsapi.NewSnapshotsClient(gRPCConn), address), nil
+
+	// TODO (ginglis13)
+	// we should be monitoring this goroutine's status.
+	// https://github.com/firecracker-microvm/firecracker-containerd/issues/607
+	if metricsProxy != nil {
+		go metricsProxy.Serve(ctx)
+	}
+
+	return &RemoteSnapshotter{proxy.NewSnapshotter(snapshotsapi.NewSnapshotsClient(gRPCConn), address), metricsProxy}, nil
+}
+
+// MetricsProxyPort returns the metrics proxy port for a remote snapshotter.
+func (rs *RemoteSnapshotter) MetricsProxyPort() int {
+	return rs.metricsProxy.Port
+
+}
+
+// MetricsProxyLabels returns the metrics labels for a remote snapshotter.
+func (rs *RemoteSnapshotter) MetricsProxyLabels() map[string]string {
+	return rs.metricsProxy.Labels
 }


Test patch:
diff --git a/snapshotter/config/config_test.go b/snapshotter/config/config_test.go
index 6ecbef3..911a4a5 100644
--- a/snapshotter/config/config_test.go
+++ b/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{
diff --git a/snapshotter/demux/cache/snapshotter_cache_test.go b/snapshotter/demux/cache/snapshotter_cache_test.go
index d683c0c..46405eb 100644
--- a/snapshotter/demux/cache/snapshotter_cache_test.go
+++ b/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 {
diff --git a/snapshotter/demux/proxy/address/http_resolver_test.go b/snapshotter/demux/proxy/address/http_resolver_test.go
index 94b60c6..f2bf9de 100644
--- a/snapshotter/demux/proxy/address/http_resolver_test.go
+++ b/snapshotter/demux/proxy/address/http_resolver_test.go
@@ -58,12 +58,23 @@ func returnErrorOnJSONParserError() error {
 }
 
 func happyPath() error {
-	reader := mockReader{response: []byte(`{"network": "unix", "address": "/path/to/snapshotter.vsock", "snapshotter_port": "10000", "metrics_port": "10001"}`), err: io.EOF}
+	reader := mockReader{response: []byte(
+		`{
+			"network": "unix", 
+			"address": "/path/to/snapshotter.vsock", 
+			"snapshotter_port": "10000", 
+			"metrics_port": "10001",
+			"labels": {
+				"test1": "label1",
+				"test2": "label2"
+				}
+			}`), err: io.EOF}
 	client := mockClient{getError: nil, getResponse: http.Response{Body: &reader}}
 	uut := HTTPResolver{url: "localhost:10001", client: &client}
 
 	actual, err := uut.Get("namespace")
 	if err != nil {
+		fmt.Println(err)
 		return errors.New("expected no error from HTTP resolver")
 	}
 
@@ -79,6 +90,12 @@ func happyPath() error {
 	if actual.MetricsPort != "10001" {
 		return fmt.Errorf("Expected metrics port '10001' but actual %s", actual.MetricsPort)
 	}
+	if actual.Labels["test1"] != "label1" {
+		return fmt.Errorf("Expected metrics label key='test1' value='label1' but actual %s", actual.Labels["test1"])
+	}
+	if actual.Labels["test2"] != "label2" {
+		return fmt.Errorf("Expected metrics label key='test2' value='label2' but actual %s", actual.Labels["test1"])
+	}
 	return nil
 }
 
diff --git a/snapshotter/demux/snapshotter_test.go b/snapshotter/demux/snapshotter_test.go
index b3894a4..70d20bd 100644
--- a/snapshotter/demux/snapshotter_test.go
+++ b/snapshotter/demux/snapshotter_test.go
@@ -24,14 +24,15 @@ import (
 
 	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/cache"
 	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/internal"
+	"github.com/firecracker-microvm/firecracker-containerd/snapshotter/demux/proxy"
 )
 
-func fetchOkSnapshotter(ctx context.Context, key string) (snapshots.Snapshotter, error) {
+func fetchOkSnapshotter(ctx context.Context, key string) (*proxy.RemoteSnapshotter, error) {
 	var snapshotter internal.SuccessfulSnapshotter = internal.SuccessfulSnapshotter{}
-	return &snapshotter, nil
+	return &proxy.RemoteSnapshotter{Snapshotter: &snapshotter}, nil
 }
 
-func fetchSnapshotterNotFound(ctx context.Context, key string) (snapshots.Snapshotter, error) {
+func fetchSnapshotterNotFound(ctx context.Context, key string) (*proxy.RemoteSnapshotter, error) {
 	return nil, errors.New("mock snapshotter not found")
 }
 
@@ -41,9 +42,9 @@ func createSnapshotterCacheWithSuccessfulSnapshotter(namespace string) cache.Cac
 	return cache
 }
 
-func fetchFailingSnapshotter(ctx context.Context, key string) (snapshots.Snapshotter, error) {
+func fetchFailingSnapshotter(ctx context.Context, key string) (*proxy.RemoteSnapshotter, error) {
 	var snapshotter internal.FailingSnapshotter = internal.FailingSnapshotter{}
-	return &snapshotter, nil
+	return &proxy.RemoteSnapshotter{Snapshotter: &snapshotter}, nil
 }
 
 func createSnapshotterCacheWithFailingSnapshotter(namespace string) cache.Cache {