feat(metrics): support OpenMetrics from applications

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

@@ -11,7 +11,9 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"regexp"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -228,14 +230,41 @@ func processMetrics(metrices <-chan []byte, contentType expfmt.Format) []byte {
 		}
 	}
 
-	// if the content type is not OpenMetrics, we don't need to do any processing
+	str := processNewlineChars(buf.String(), true, true)
+	str = fmt.Sprintf("%s\n", str)
+
+	// if the content type is not OpenMetrics, we don't need to add EOF marker
 	if !(contentType == expfmt.FmtOpenMetrics_1_0_0 || contentType == expfmt.FmtOpenMetrics_0_0_1) {
-		return buf.Bytes()
+		return []byte(str)
 	}
 
 	// make metrics OpenMetrics compliant
-	expfmt.FinalizeOpenMetrics(buf)
-	return bytes.ReplaceAll(buf.Bytes(), []byte("\n\n"), []byte("\n"))
+	return []byte(fmt.Sprintf("%s# EOF\n", str))
+}
+
+// processNewlineChars processes the newline characters in the text.
+// If dedup is true, it replaces multiple newline characters with a single newline character.
+// If trim is true, it trims the leading and trailing newline characters.
+func processNewlineChars(text string, dedup, trim bool) string {
+	if dedup {
+		// Create a regular expression to match multiple newline characters.
+		reg, err := regexp.Compile(`(\r\n?|\n){2,}`)
+		if err != nil {
+			return text
+		}
+
+		// Replace all the matches with a single newline character.
+		text = reg.ReplaceAllString(text, "\n")
+	}
+
+	if trim {
+		// Trim the leading and trailing newline characters.
+		text = strings.TrimFunc(text, func(r rune) bool {
+			return r == '\n'
+		})
+	}
+
+	return text
 }
 
 // selectContentType selects the highest priority content type supported by the applications.

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

@@ -1,8 +1,11 @@
 package metrics
 
 import (
+	"io"
 	"net/http"
 	"net/url"
+	"os"
+	"path"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -123,3 +126,87 @@ var _ = Describe("Response Format", func() {
 		}),
 	)
 })
+
+var _ = Describe("Process Metrics", func() {
+	type testCase struct {
+		input       []string // input files containing metrics
+		contentType expfmt.Format
+		expected    string // expected output file
+	}
+	DescribeTable("should",
+		func(given testCase) {
+			inputMetrics := make(chan []byte, len(given.input))
+			for _, input := range given.input {
+				fo, err := os.Open(path.Join("testdata", input))
+				Expect(err).ToNot(HaveOccurred())
+				byteData, err := io.ReadAll(fo)
+				Expect(err).ToNot(HaveOccurred())
+				inputMetrics <- byteData
+			}
+			close(inputMetrics)
+
+			fo, err := os.Open(path.Join("testdata", given.expected))
+			Expect(err).ToNot(HaveOccurred())
+			expected, err := io.ReadAll(fo)
+			Expect(err).ToNot(HaveOccurred())
+
+			actual := processMetrics(inputMetrics, given.contentType)
+			Expect(string(actual)).To(Equal(string(expected)))
+		},
+		Entry("return OpenMetrics compliant metrics", testCase{
+			input:       []string{"openmetrics_0_1_1.in", "counter.out"},
+			contentType: expfmt.FmtOpenMetrics_0_0_1,
+			expected:    "openmetrics_0_0_1-counter.out",
+		}),
+		Entry("handle multiple # EOF", testCase{
+			input:       []string{"openmetrics_0_1_1.in", "openmetrics_0_1_1.in", "counter.out"},
+			contentType: expfmt.FmtOpenMetrics_0_0_1,
+			expected:    "multi-openmetrics-counter.out",
+		}),
+		Entry("return Prometheus text compliant metrics", testCase{
+			input:       []string{"prom-text.in", "counter.out"},
+			contentType: expfmt.FmtText,
+			expected:    "prom-text-counter.out",
+		}),
+	)
+})
+
+var _ = Describe("ProcessNewlineChars", func() {
+	type testCase struct {
+		input       string
+		deduplicate bool
+		trim        bool
+		expected    string
+	}
+
+	DescribeTable("should",
+		func(given testCase) {
+			actual := processNewlineChars(given.input, given.deduplicate, given.trim)
+			Expect(actual).To(Equal(given.expected))
+		},
+		Entry("should not deduplicate or trim newline characters", testCase{
+			input:       "This is a test.\n\nThis is another test.\n",
+			deduplicate: false,
+			trim:        false,
+			expected:    "This is a test.\n\nThis is another test.\n",
+		}),
+		Entry("should deduplicate newline characters", testCase{
+			input:       "This is a test.\n\n\nThis is another test.\n",
+			deduplicate: true,
+			trim:        false,
+			expected:    "This is a test.\nThis is another test.\n",
+		}),
+		Entry("should trim leading and trailing newline characters", testCase{
+			input:       "\nThis is a test.\n\nThis is another test\n\n",
+			deduplicate: false,
+			trim:        true,
+			expected:    "This is a test.\n\nThis is another test",
+		}),
+		Entry("should deduplicate and trim newline characters", testCase{
+			input:       "\nThis is a test.\n\n\nThis is another test\n",
+			deduplicate: true,
+			trim:        true,
+			expected:    "This is a test.\nThis is another test",
+		}),
+	)
+})

app/kuma-dp/pkg/dataplane/metrics/testdata/multi-openmetrics-counter.out

@@ -0,0 +1,61 @@
+# HELP python_gc_objects_collected Objects collected during gc
+# TYPE python_gc_objects_collected counter
+python_gc_objects_collected_total{generation="0"} 309.0
+python_gc_objects_collected_total{generation="1"} 53.0
+python_gc_objects_collected_total{generation="2"} 0.0
+# HELP python_gc_objects_uncollectable Uncollectable objects found during GC
+# TYPE python_gc_objects_uncollectable counter
+python_gc_objects_uncollectable_total{generation="0"} 0.0
+python_gc_objects_uncollectable_total{generation="1"} 0.0
+python_gc_objects_uncollectable_total{generation="2"} 0.0
+# HELP python_gc_collections Number of times this generation was collected
+# TYPE python_gc_collections counter
+python_gc_collections_total{generation="0"} 77.0
+python_gc_collections_total{generation="1"} 7.0
+python_gc_collections_total{generation="2"} 0.0
+# HELP python_info Python platform information
+# TYPE python_info gauge
+python_info{implementation="CPython",major="3",minor="10",patchlevel="8",version="3.10.8"} 1.0
+# HELP python_gc_objects_collected Objects collected during gc
+# TYPE python_gc_objects_collected counter
+python_gc_objects_collected_total{generation="0"} 309.0
+python_gc_objects_collected_total{generation="1"} 53.0
+python_gc_objects_collected_total{generation="2"} 0.0
+# HELP python_gc_objects_uncollectable Uncollectable objects found during GC
+# TYPE python_gc_objects_uncollectable counter
+python_gc_objects_uncollectable_total{generation="0"} 0.0
+python_gc_objects_uncollectable_total{generation="1"} 0.0
+python_gc_objects_uncollectable_total{generation="2"} 0.0
+# HELP python_gc_collections Number of times this generation was collected
+# TYPE python_gc_collections counter
+python_gc_collections_total{generation="0"} 77.0
+python_gc_collections_total{generation="1"} 7.0
+python_gc_collections_total{generation="2"} 0.0
+# HELP python_info Python platform information
+# TYPE python_info gauge
+python_info{implementation="CPython",major="3",minor="10",patchlevel="8",version="3.10.8"} 1.0
+# TYPE envoy_cluster_assignment_stale counter
+envoy_cluster_assignment_stale{envoy_cluster_name="access_log_sink"} 11
+envoy_cluster_assignment_stale{envoy_cluster_name="ads_cluster"} 12
+envoy_cluster_assignment_stale{envoy_cluster_name="demo-client",kuma_io_mesh_traffic="true"} 13
+envoy_cluster_assignment_stale{envoy_cluster_name="echo-server_kuma-test_svc_8080",kuma_io_mesh_traffic="true"} 306
+envoy_cluster_assignment_stale{envoy_cluster_name="inbound_passthrough_ipv4",kuma_io_mesh_traffic="true"} 14
+envoy_cluster_assignment_stale{envoy_cluster_name="inbound_passthrough_ipv6",kuma_io_mesh_traffic="true"} 15
+envoy_cluster_assignment_stale{envoy_cluster_name="kuma_envoy_admin"} 16
+envoy_cluster_assignment_stale{envoy_cluster_name="localhost_3000",kuma_io_mesh_traffic="true"} 17
+envoy_cluster_assignment_stale{envoy_cluster_name="outbound_passthrough_ipv4",kuma_io_mesh_traffic="true"} 18
+envoy_cluster_assignment_stale{envoy_cluster_name="outbound_passthrough_ipv6",kuma_io_mesh_traffic="true"} 19
+# TYPE envoy_http_downstream_rq_total counter
+envoy_http_downstream_rq_total{envoy_http_conn_manager_prefix="default_gateway",kuma_io_mesh_traffic="true"} 10
+# TYPE envoy_cluster_assignment_timeout_received counter
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="access_log_sink"} 1
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="ads_cluster"} 2
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="demo-client",kuma_io_mesh_traffic="true"} 3
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="echo-server_kuma-test_svc_8080",kuma_io_mesh_traffic="true"} 606
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="inbound_passthrough_ipv4",kuma_io_mesh_traffic="true"} 4
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="inbound_passthrough_ipv6",kuma_io_mesh_traffic="true"} 5
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="kuma_envoy_admin"} 6
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="localhost_3000",kuma_io_mesh_traffic="true"} 7
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="outbound_passthrough_ipv4",kuma_io_mesh_traffic="true"} 8
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="outbound_passthrough_ipv6",kuma_io_mesh_traffic="true"} 9
+# EOF

app/kuma-dp/pkg/dataplane/metrics/testdata/openmetrics_0_0_1-counter.out

@@ -0,0 +1,43 @@
+# HELP python_gc_objects_collected Objects collected during gc
+# TYPE python_gc_objects_collected counter
+python_gc_objects_collected_total{generation="0"} 309.0
+python_gc_objects_collected_total{generation="1"} 53.0
+python_gc_objects_collected_total{generation="2"} 0.0
+# HELP python_gc_objects_uncollectable Uncollectable objects found during GC
+# TYPE python_gc_objects_uncollectable counter
+python_gc_objects_uncollectable_total{generation="0"} 0.0
+python_gc_objects_uncollectable_total{generation="1"} 0.0
+python_gc_objects_uncollectable_total{generation="2"} 0.0
+# HELP python_gc_collections Number of times this generation was collected
+# TYPE python_gc_collections counter
+python_gc_collections_total{generation="0"} 77.0
+python_gc_collections_total{generation="1"} 7.0
+python_gc_collections_total{generation="2"} 0.0
+# HELP python_info Python platform information
+# TYPE python_info gauge
+python_info{implementation="CPython",major="3",minor="10",patchlevel="8",version="3.10.8"} 1.0
+# TYPE envoy_cluster_assignment_stale counter
+envoy_cluster_assignment_stale{envoy_cluster_name="access_log_sink"} 11
+envoy_cluster_assignment_stale{envoy_cluster_name="ads_cluster"} 12
+envoy_cluster_assignment_stale{envoy_cluster_name="demo-client",kuma_io_mesh_traffic="true"} 13
+envoy_cluster_assignment_stale{envoy_cluster_name="echo-server_kuma-test_svc_8080",kuma_io_mesh_traffic="true"} 306
+envoy_cluster_assignment_stale{envoy_cluster_name="inbound_passthrough_ipv4",kuma_io_mesh_traffic="true"} 14
+envoy_cluster_assignment_stale{envoy_cluster_name="inbound_passthrough_ipv6",kuma_io_mesh_traffic="true"} 15
+envoy_cluster_assignment_stale{envoy_cluster_name="kuma_envoy_admin"} 16
+envoy_cluster_assignment_stale{envoy_cluster_name="localhost_3000",kuma_io_mesh_traffic="true"} 17
+envoy_cluster_assignment_stale{envoy_cluster_name="outbound_passthrough_ipv4",kuma_io_mesh_traffic="true"} 18
+envoy_cluster_assignment_stale{envoy_cluster_name="outbound_passthrough_ipv6",kuma_io_mesh_traffic="true"} 19
+# TYPE envoy_http_downstream_rq_total counter
+envoy_http_downstream_rq_total{envoy_http_conn_manager_prefix="default_gateway",kuma_io_mesh_traffic="true"} 10
+# TYPE envoy_cluster_assignment_timeout_received counter
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="access_log_sink"} 1
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="ads_cluster"} 2
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="demo-client",kuma_io_mesh_traffic="true"} 3
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="echo-server_kuma-test_svc_8080",kuma_io_mesh_traffic="true"} 606
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="inbound_passthrough_ipv4",kuma_io_mesh_traffic="true"} 4
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="inbound_passthrough_ipv6",kuma_io_mesh_traffic="true"} 5
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="kuma_envoy_admin"} 6
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="localhost_3000",kuma_io_mesh_traffic="true"} 7
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="outbound_passthrough_ipv4",kuma_io_mesh_traffic="true"} 8
+envoy_cluster_assignment_timeout_received{envoy_cluster_name="outbound_passthrough_ipv6",kuma_io_mesh_traffic="true"} 9
+# EOF

app/kuma-dp/pkg/dataplane/metrics/testdata/openmetrics_0_1_1.in

@@ -0,0 +1,19 @@
+# HELP python_gc_objects_collected Objects collected during gc
+# TYPE python_gc_objects_collected counter
+python_gc_objects_collected_total{generation="0"} 309.0
+python_gc_objects_collected_total{generation="1"} 53.0
+python_gc_objects_collected_total{generation="2"} 0.0
+# HELP python_gc_objects_uncollectable Uncollectable objects found during GC
+# TYPE python_gc_objects_uncollectable counter
+python_gc_objects_uncollectable_total{generation="0"} 0.0
+python_gc_objects_uncollectable_total{generation="1"} 0.0
+python_gc_objects_uncollectable_total{generation="2"} 0.0
+# HELP python_gc_collections Number of times this generation was collected
+# TYPE python_gc_collections counter
+python_gc_collections_total{generation="0"} 77.0
+python_gc_collections_total{generation="1"} 7.0
+python_gc_collections_total{generation="2"} 0.0
+# HELP python_info Python platform information
+# TYPE python_info gauge
+python_info{implementation="CPython",major="3",minor="10",patchlevel="8",version="3.10.8"} 1.0
+# EOF