fix(kuma-dp) pass query parameters through the metrics hijacker (#2124)

The Envoy Prometheus endpoint supports passing a filtering regex in
the query parameters to limit the set of metrics that are returned.
This feature lets a metrics scraper scrape different sets of metrics
at different intervals.

This updates #1755.

Signed-off-by: James Peach <james.peach@konghq.com>

app/kuma-dp/pkg/dataplane/metrics/merge_test.go

@@ -1,4 +1,4 @@
-package metrics_test
+package metrics
 
 import (
 	"bufio"
@@ -9,8 +9,6 @@ import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/ginkgo/extensions/table"
 	. "github.com/onsi/gomega"
-
-	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/metrics"
 )
 
 func toLines(r io.Reader) (lines []string) {
@@ -34,7 +32,7 @@ var _ = Describe("Merge", func() {
 			Expect(err).ToNot(HaveOccurred())
 
 			actual := new(bytes.Buffer)
-			err = metrics.MergeClusters(input, actual)
+			err = MergeClusters(input, actual)
 			Expect(err).ToNot(HaveOccurred())
 			Expect(toLines(actual)).To(ConsistOf(toLines(expected)))
 		},

app/kuma-dp/pkg/dataplane/metrics/metrics_suite_test.go

@@ -1,4 +1,4 @@
-package metrics_test
+package metrics
 
 import (
 	"testing"

app/kuma-dp/pkg/dataplane/metrics/server.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net"
 	"net/http"
+	"net/url"
 	"os"
 
 	"github.com/pkg/errors"
@@ -23,32 +24,32 @@ var _ component.Component = &Hijacker{}
 
 type Hijacker struct {
 	envoyAdminPort uint32
-	address        string
+	socketPath     string
 }
 
 func New(dataplane kumadp.Dataplane, envoyAdminPort uint32) *Hijacker {
 	return &Hijacker{
 		envoyAdminPort: envoyAdminPort,
-		address:        envoy.MetricsHijackerSocketName(dataplane.Name, dataplane.Mesh),
+		socketPath:     envoy.MetricsHijackerSocketName(dataplane.Name, dataplane.Mesh),
 	}
 }
 
 func (s *Hijacker) Start(stop <-chan struct{}) error {
-	_, err := os.Stat(s.address)
+	_, err := os.Stat(s.socketPath)
 	if err == nil {
 		// File is accessible try to rename it to verify it is not open
-		newName := s.address + ".bak"
-		err := os.Rename(s.address, newName)
+		newName := s.socketPath + ".bak"
+		err := os.Rename(s.socketPath, newName)
 		if err != nil {
-			return errors.Errorf("file %s exists and probably opened by another kuma-dp instance", s.address)
+			return errors.Errorf("file %s exists and probably opened by another kuma-dp instance", s.socketPath)
 		}
 		err = os.Remove(newName)
 		if err != nil {
 			return errors.Errorf("not able the delete the backup file %s", newName)
 		}
 	}
 
-	lis, err := net.Listen("unix", s.address)
+	lis, err := net.Listen("unix", s.socketPath)
 	if err != nil {
 		return err
 	}
@@ -57,7 +58,10 @@ func (s *Hijacker) Start(stop <-chan struct{}) error {
 		lis.Close()
 	}()
 
-	logger.Info("starting Metrics Hijacker Server", "address", fmt.Sprintf("unix://%s", s.address))
+	logger.Info("starting Metrics Hijacker Server",
+		"socketPath", fmt.Sprintf("unix://%s", s.socketPath),
+		"adminPort", s.envoyAdminPort,
+	)
 
 	server := &http.Server{
 		Handler: s,
@@ -79,8 +83,22 @@ func (s *Hijacker) Start(stop <-chan struct{}) error {
 	}
 }
 
-func (s *Hijacker) ServeHTTP(writer http.ResponseWriter, _ *http.Request) {
-	resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/stats/prometheus", s.envoyAdminPort))
+// The Envoy stats endpoint recognizes the "used_only" and "filter" query
+// parameters. We squash the path to enforce Prometheus metrics format, but
+// forward the query parameters so that the scraper can do partial scrapes.
+func rewriteMetricsURL(port uint32, in *url.URL) string {
+	u := url.URL{
+		Scheme:   "http",
+		Host:     fmt.Sprintf("127.0.0.1:%d", port),
+		Path:     "/stats/prometheus",
+		RawQuery: in.RawQuery,
+	}
+
+	return u.String()
+}
+
+func (s *Hijacker) ServeHTTP(writer http.ResponseWriter, req *http.Request) {
+	resp, err := http.Get(rewriteMetricsURL(s.envoyAdminPort, req.URL))
 	if err != nil {
 		http.Error(writer, err.Error(), 500)
 		return

app/kuma-dp/pkg/dataplane/metrics/server_test.go

@@ -0,0 +1,35 @@
+package metrics
+
+import (
+	"net/url"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/ginkgo/extensions/table"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Rewriting the metrics URL", func() {
+	type testCase struct {
+		input     string
+		adminPort uint32
+		expected  string
+	}
+	DescribeTable("should",
+		func(given testCase) {
+			u, err := url.Parse(given.input)
+			Expect(err).ToNot(HaveOccurred())
+
+			Expect(rewriteMetricsURL(given.adminPort, u)).Should(Equal(given.expected))
+		},
+		Entry("use the admin port", testCase{
+			input:     "http://foo/bar",
+			adminPort: 99,
+			expected:  "http://127.0.0.1:99/stats/prometheus",
+		}),
+		Entry("preserve query parameters", testCase{
+			input:     "http://foo/bar?one=two&three=four",
+			adminPort: 80,
+			expected:  "http://127.0.0.1:80/stats/prometheus?one=two&three=four",
+		}),
+	)
+})