Prototype to stream Envoy metrics to Ambassador's backend

Signed-off-by: Douglas Camata <[email protected]>

Author: douglascamata

pkg/agent/agent.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	envoyMetrics "github.com/datawire/ambassador/v2/pkg/api/envoy/service/metrics/v2"
 	"io/ioutil"
 	"net/http"
 	"net/url"
@@ -599,6 +600,24 @@ func (a *Agent) ProcessSnapshot(ctx context.Context, snapshot *snapshotTypes.Sna
 	return nil
 }
 
+func (a *Agent) MetricsRelayHandler(in *envoyMetrics.StreamMetricsMessage) {
+	ctx := context.Background()
+	dlog.Debugf(ctx, "received %d metrics", len(in.GetEnvoyMetrics()))
+	if a.comm != nil && !a.reportingStopped {
+		a.ambassadorAPIKeyMutex.Lock()
+		apikey := a.ambassadorAPIKey
+		a.ambassadorAPIKeyMutex.Unlock()
+		outMessage := &agent.StreamMetricsMessage{
+			Identity:     a.agentID,
+			EnvoyMetrics: in.EnvoyMetrics,
+		}
+		dlog.Debugf(ctx, "relaying %d metrics", len(outMessage.GetEnvoyMetrics()))
+		if err := a.comm.StreamMetrics(ctx, outMessage, apikey); err != nil {
+			dlog.Errorf(ctx, "Error streaming metrics: %+v", err)
+		}
+	}
+}
+
 // ClearComm ends the current connection to the Director, if it exists, thereby
 // forcing a new connection to be created when needed.
 func (a *Agent) ClearComm() {

pkg/agent/comm.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"net/url"
+	"sync"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -19,12 +20,13 @@ import (
 const APIKeyMetadataKey = "x-ambassador-api-key"
 
 type RPCComm struct {
-	conn       io.Closer
-	client     agent.DirectorClient
-	rptWake    chan struct{}
-	retCancel  context.CancelFunc
-	agentID    *agent.Identity
-	directives chan *agent.Directive
+	conn                     io.Closer
+	client                   agent.DirectorClient
+	rptWake                  chan struct{}
+	retCancel                context.CancelFunc
+	agentID                  *agent.Identity
+	directives               chan *agent.Directive
+	metricsStreamWriterMutex sync.Mutex
 }
 
 const (
@@ -189,6 +191,19 @@ func (c *RPCComm) Report(ctx context.Context, report *agent.Snapshot, apiKey str
 	return nil
 }
 
+func (c *RPCComm) StreamMetrics(ctx context.Context, metrics *agent.StreamMetricsMessage, apiKey string) error {
+	ctx = dlog.WithField(ctx, "agent", "streammetrics")
+
+	c.metricsStreamWriterMutex.Lock()
+	defer c.metricsStreamWriterMutex.Unlock()
+	ctx = metadata.AppendToOutgoingContext(ctx, APIKeyMetadataKey, apiKey)
+	streamClient, err := c.client.StreamMetrics(ctx)
+	if err != nil {
+		return err
+	}
+	return streamClient.Send(metrics)
+}
+
 func (c *RPCComm) Directives() <-chan *agent.Directive {
 	return c.directives
 }

pkg/agent/envoy_metrics_server.go

@@ -0,0 +1,56 @@
+package agent
+
+import (
+	"context"
+	envoyMetrics "github.com/datawire/ambassador/v2/pkg/api/envoy/service/metrics/v2"
+	"github.com/datawire/dlib/dlog"
+	"google.golang.org/grpc"
+	"io"
+	"net"
+)
+
+type streamHandler func(in *envoyMetrics.StreamMetricsMessage)
+
+type metricsServer struct {
+	envoyMetrics.MetricsServiceServer
+	handler streamHandler
+}
+
+// NewMetricsServer is the main metricsServer constructor.
+func NewMetricsServer(handler streamHandler) *metricsServer {
+	return &metricsServer{
+		handler: handler,
+	}
+}
+
+// StartServer will start the metrics gRPC server, listening on :8123
+// It is a blocking call until grpcServer.Serve returns.
+func (s *metricsServer) StartServer(ctx context.Context) error {
+	grpcServer := grpc.NewServer()
+	envoyMetrics.RegisterMetricsServiceServer(grpcServer, s)
+
+	listener, err := net.Listen("tcp", ":8123")
+	if err != nil {
+		dlog.Errorf(ctx, "metrics service failed to listen: %v", err)
+	}
+
+	dlog.Infof(ctx, "metrics service listening on %s", listener.Addr().String())
+	return grpcServer.Serve(listener)
+}
+
+// StreamMetrics implements the StreamMetrics rpc call by calling the stream handler on each
+// message received.
+func (s *metricsServer) StreamMetrics(stream envoyMetrics.MetricsService_StreamMetricsServer) error {
+	ctx := stream.Context()
+	dlog.Debug(ctx, "started stream")
+	for {
+		in, err := stream.Recv()
+		if err == io.EOF {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		s.handler(in)
+	}
+}