Stream Envoy metrics to the cloud

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

Author: douglascamata

api/agent/director.proto

@@ -8,6 +8,8 @@ syntax = "proto3";
 import "google/protobuf/duration.proto";
 import "google/protobuf/timestamp.proto";
 
+import "prometheus/metrics.proto";
+
 package agent;
 
 service Director {
@@ -20,6 +22,9 @@ service Director {
   // Report a consistent Snapshot of information to the CEPC.
   rpc ReportStream(stream RawSnapshotChunk) returns (SnapshotResponse) {}
 
+  // Stream metrics to the CEPC.
+  rpc StreamMetrics(stream StreamMetricsMessage) returns (StreamMetricsResponse) {}
+
   // Retrieve Directives from the CEPC
   rpc Retrieve(Identity) returns (stream Directive) {}
 
@@ -105,3 +110,14 @@ message Command {
   // Log this message if present
   string message = 1;
 }
+
+message StreamMetricsMessage {
+  Identity identity = 1;
+
+  // A list of metric entries
+  repeated io.prometheus.client.MetricFamily envoy_metrics = 2;
+}
+
+message StreamMetricsResponse {
+
+}

api/prometheus/metrics.proto

@@ -0,0 +1,92 @@
+// Copyright 2013 Prometheus Team
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License 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.
+
+syntax = "proto2";
+
+package io.prometheus.client;
+option java_package = "io.prometheus.client";
+option go_package = "github.com/prometheus/client_model/go;io_prometheus_client";
+
+import "google/protobuf/timestamp.proto";
+
+message LabelPair {
+  optional string name  = 1;
+  optional string value = 2;
+}
+
+enum MetricType {
+  COUNTER    = 0;
+  GAUGE      = 1;
+  SUMMARY    = 2;
+  UNTYPED    = 3;
+  HISTOGRAM  = 4;
+}
+
+message Gauge {
+  optional double value = 1;
+}
+
+message Counter {
+  optional double   value    = 1;
+  optional Exemplar exemplar = 2;
+}
+
+message Quantile {
+  optional double quantile = 1;
+  optional double value    = 2;
+}
+
+message Summary {
+  optional uint64   sample_count = 1;
+  optional double   sample_sum   = 2;
+  repeated Quantile quantile     = 3;
+}
+
+message Untyped {
+  optional double value = 1;
+}
+
+message Histogram {
+  optional uint64 sample_count = 1;
+  optional double sample_sum   = 2;
+  repeated Bucket bucket       = 3; // Ordered in increasing order of upper_bound, +Inf bucket is optional.
+}
+
+message Bucket {
+  optional uint64   cumulative_count = 1; // Cumulative in increasing order.
+  optional double   upper_bound      = 2; // Inclusive.
+  optional Exemplar exemplar         = 3;
+}
+
+message Exemplar {
+  repeated LabelPair                 label     = 1;
+  optional double                    value     = 2;
+  optional google.protobuf.Timestamp timestamp = 3; // OpenMetrics-style.
+}
+
+message Metric {
+  repeated LabelPair label        = 1;
+  optional Gauge     gauge        = 2;
+  optional Counter   counter      = 3;
+  optional Summary   summary      = 4;
+  optional Untyped   untyped      = 5;
+  optional Histogram histogram    = 7;
+  optional int64     timestamp_ms = 6;
+}
+
+message MetricFamily {
+  optional string     name   = 1;
+  optional string     help   = 2;
+  optional MetricType type   = 3;
+  repeated Metric     metric = 4;
+}

cmd/agent/main.go

@@ -50,6 +50,13 @@ func run(cmd *cobra.Command, args []string) error {
 		snapshotURL = fmt.Sprintf(DefaultSnapshotURLFmt, entrypoint.ExternalSnapshotPort)
 	}
 
+	metricsServer := agent.NewMetricsServer(ambAgent.MetricsRelayHandler)
+	go func() {
+		if err := metricsServer.StartServer(ctx); err != nil {
+			dlog.Error(ctx, err)
+		}
+	}()
+
 	if err := ambAgent.Watch(ctx, snapshotURL); err != nil {
 		return err
 	}

pkg/agent/agent.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	envoyMetrics "github.com/datawire/ambassador/v2/pkg/api/envoy/service/metrics/v2"
+	envoyMetrics "github.com/datawire/ambassador/v2/pkg/api/envoy/service/metrics/v3"
 	"io/ioutil"
 	"net/http"
 	"net/url"
@@ -28,6 +28,7 @@ type Comm interface {
 	Close() error
 	Report(context.Context, *agent.Snapshot, string) error
 	Directives() <-chan *agent.Directive
+	StreamMetrics(context.Context, *agent.StreamMetricsMessage, string) error
 }
 
 type atomicBool struct {

pkg/agent/envoy_metrics_server.go

@@ -2,7 +2,7 @@ package agent
 
 import (
 	"context"
-	envoyMetrics "github.com/datawire/ambassador/v2/pkg/api/envoy/service/metrics/v2"
+	envoyMetrics "github.com/datawire/ambassador/v2/pkg/api/envoy/service/metrics/v3"
 	"github.com/datawire/dlib/dlog"
 	"google.golang.org/grpc"
 	"io"
@@ -39,7 +39,7 @@ func (s *metricsServer) StartServer(ctx context.Context) error {
 }
 
 // StreamMetrics implements the StreamMetrics rpc call by calling the stream handler on each
-// message received.
+// message received. It's invoked whenever metrics arrive from Envoy.
 func (s *metricsServer) StreamMetrics(stream envoyMetrics.MetricsService_StreamMetricsServer) error {
 	ctx := stream.Context()
 	dlog.Debug(ctx, "started stream")