Repository: VictoriaMetrics/VictoriaMetrics
Pull request #5893 (2024-02-28 21:13:26)

{vmagent,vminsert}: added firehose http destination opentelemetry data ingestion support
support Cloudwatch Streams opentelemetry message parsing received from Firehose HTTP destination

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/5893?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) Report
Attention: Patch coverage is `4.47761%` with `64 lines` in your changes are missing coverage. Please review.
> Project coverage is 56.65%. Comparing base [(`8aaa828`)](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/8aaa828ba3f49a43cb8de89a0bcae160f7988c14?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) to head [(`cb2aa92`)](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/5893?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).
> Report is 232 commits behind head on master.

> :exclamation: Current head cb2aa92 differs from pull request most recent head d558726. Consider uploading reports for the commit d558726 to get more accurate results

| [Files](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/5893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) | Patch % | Lines |
|---|---|---|
| [lib/protoparser/firehose/parser.go](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/5893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL3Byb3RvcGFyc2VyL2ZpcmVob3NlL3BhcnNlci5nbw==) | 0.00% | [49 Missing :warning: ](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/5893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) |
| [app/vmagent/opentelemetry/request\_handler.go](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/5893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWdlbnQvb3BlbnRlbGVtZXRyeS9yZXF1ZXN0X2hhbmRsZXIuZ28=) | 0.00% | [6 Missing :warning: ](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/5893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) |
| [app/vminsert/opentelemetry/request\_handler.go](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/5893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtaW5zZXJ0L29wZW50ZWxlbWV0cnkvcmVxdWVzdF9oYW5kbGVyLmdv) | 0.00% | [6 Missing :warning: ](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/5893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) |
| [...b/protoparser/opentelemetry/stream/streamparser.go](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/5893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL3Byb3RvcGFyc2VyL29wZW50ZWxlbWV0cnkvc3RyZWFtL3N0cmVhbXBhcnNlci5nbw==) | 50.00% | [2 Missing and 1 partial :warning: ](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/5893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #5893      +/-   ##
==========================================
- Coverage   60.37%   56.65%   -3.72%     
==========================================
  Files         411      520     +109     
  Lines       76609    71564    -5045     
==========================================
- Hits        46253    40547    -5706     
- Misses      27794    28155     +361     
- Partials     2562     2862     +300     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/5893?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).


Patch:
diff --git a/app/vmagent/opentelemetry/request_handler.go b/app/vmagent/opentelemetry/request_handler.go
index e923f750c..5809b8c13 100644
--- a/app/vmagent/opentelemetry/request_handler.go
+++ b/app/vmagent/opentelemetry/request_handler.go
@@ -9,6 +9,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/firehose"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 	"github.com/VictoriaMetrics/metrics"
@@ -27,10 +28,15 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 		return err
 	}
 	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+	var processBody func(*[]byte) error
 	if req.Header.Get("Content-Type") == "application/json" {
-		return fmt.Errorf("json encoding isn't supported for opentelemetry format. Use protobuf encoding")
+		if req.Header.Get("X-Amz-Firehouse-Protocol-Version") != "" {
+			processBody = firehose.ProcessRequestBody
+		} else {
+			return fmt.Errorf("json encoding isn't supported for opentelemetry format. Use protobuf encoding")
+		}
 	}
-	return stream.ParseStream(req.Body, isGzipped, func(tss []prompbmarshal.TimeSeries) error {
+	return stream.ParseStream(req.Body, isGzipped, processBody, func(tss []prompbmarshal.TimeSeries) error {
 		return insertRows(at, tss, extraLabels)
 	})
 }
diff --git a/app/vminsert/opentelemetry/request_handler.go b/app/vminsert/opentelemetry/request_handler.go
index 165fa275c..d9c1786b2 100644
--- a/app/vminsert/opentelemetry/request_handler.go
+++ b/app/vminsert/opentelemetry/request_handler.go
@@ -8,6 +8,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/firehose"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/stream"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -24,10 +25,15 @@ func InsertHandler(req *http.Request) error {
 		return err
 	}
 	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+	var processBody func(*[]byte) error
 	if req.Header.Get("Content-Type") == "application/json" {
-		return fmt.Errorf("json encoding isn't supported for opentelemetry format. Use protobuf encoding")
+		if req.Header.Get("X-Amz-Firehose-Protocol-Version") != "" {
+			processBody = firehose.ProcessRequestBody
+		} else {
+			return fmt.Errorf("json encoding isn't supported for opentelemetry format. Use protobuf encoding")
+		}
 	}
-	return stream.ParseStream(req.Body, isGzipped, func(tss []prompbmarshal.TimeSeries) error {
+	return stream.ParseStream(req.Body, isGzipped, processBody, func(tss []prompbmarshal.TimeSeries) error {
 		return insertRows(tss, extraLabels)
 	})
 }
diff --git a/lib/protoparser/firehose/parser.go b/lib/protoparser/firehose/parser.go
new file mode 100644
index 000000000..48c7d6f4c
--- /dev/null
+++ b/lib/protoparser/firehose/parser.go
@@ -0,0 +1,91 @@
+package firehose
+
+import (
+	"encoding/base64"
+	"fmt"
+	"github.com/valyala/fastjson"
+)
+
+// ProcessRequestBody converts Cloudwatch Stream protobuf metrics HTTP body delivered via Firehose into protobuf message.
+func ProcessRequestBody(b *[]byte) error {
+	return unmarshalRequest(b)
+}
+
+var jsonParserPool fastjson.ParserPool
+
+// unmarshalRequest extracts and decodes b64 data from Firehose HTTP destination request
+//
+//	{
+//	  "requestId": "<uuid-string>",
+//	  "timestamp": <int64-value>,
+//	  "records": [
+//	    {
+//	      "data": "<base64-encoded-payload>"
+//	    }
+//	  ]
+//	}
+func unmarshalRequest(b *[]byte) error {
+	p := jsonParserPool.Get()
+	defer jsonParserPool.Put(p)
+
+	v, err := p.ParseBytes(*b)
+	if err != nil {
+		return err
+	}
+	o, err := v.Object()
+	if err != nil {
+		return fmt.Errorf("cannot find Firehose Request objects: %w", err)
+	}
+	index := 0
+	o.Visit(func(k []byte, v *fastjson.Value) {
+		if err != nil {
+			return
+		}
+		switch string(k) {
+		case "records":
+			recordObjects, errLocal := v.Array()
+			if errLocal != nil {
+				err = fmt.Errorf("cannot find Records array in Firehose Request object: %w", errLocal)
+				return
+			}
+			for _, fr := range recordObjects {
+				recordObject, errLocal := fr.Object()
+				if errLocal != nil {
+					err = fmt.Errorf("cannot find Record object: %w", errLocal)
+					return
+				}
+				if errLocal := unmarshalRecord(b, &index, recordObject); errLocal != nil {
+					err = fmt.Errorf("cannot unmarshal Record object: %w", errLocal)
+					return
+				}
+			}
+		}
+	})
+	*b = (*b)[:index]
+	if err != nil {
+		return fmt.Errorf("cannot parse Firehose Request object: %w", err)
+	}
+	return nil
+}
+
+func unmarshalRecord(b *[]byte, index *int, o *fastjson.Object) error {
+	var err error
+	var inc int
+	o.Visit(func(k []byte, v *fastjson.Value) {
+		if v.Type() != fastjson.TypeString {
+			err = fmt.Errorf("invalid record data type, %q", v.Type())
+			return
+		}
+		valueBytes := v.GetStringBytes()
+		if len(valueBytes) == 0 {
+			return
+		}
+		inc, err = base64.StdEncoding.Decode((*b)[*index:], valueBytes)
+		if err != nil {
+			err = fmt.Errorf("failed to decode and append Firehose Record data: %w", err)
+			return
+		}
+		*index = *index + inc
+	})
+	return err
+}
diff --git a/lib/protoparser/opentelemetry/stream/streamparser.go b/lib/protoparser/opentelemetry/stream/streamparser.go
index 9d4dab2cc..30bd0e1dd 100644
--- a/lib/protoparser/opentelemetry/stream/streamparser.go
+++ b/lib/protoparser/opentelemetry/stream/streamparser.go
@@ -21,7 +21,7 @@ import (
 // ParseStream parses OpenTelemetry protobuf or json data from r and calls callback for the parsed rows.
 //
 // callback shouldn't hold tss items after returning.
-func ParseStream(r io.Reader, isGzipped bool, callback func(tss []prompbmarshal.TimeSeries) error) error {
+func ParseStream(r io.Reader, isGzipped bool, processBody func(*[]byte) error, callback func(tss []prompbmarshal.TimeSeries) error) error {
 	wcr := writeconcurrencylimiter.GetReader(r)
 	defer writeconcurrencylimiter.PutReader(wcr)
 	r = wcr
@@ -37,7 +37,7 @@ func ParseStream(r io.Reader, isGzipped bool, callback func(tss []prompbmarshal.
 
 	wr := getWriteContext()
 	defer putWriteContext(wr)
-	req, err := wr.readAndUnpackRequest(r)
+	req, err := wr.readAndUnpackRequest(r, processBody)
 	if err != nil {
 		return fmt.Errorf("cannot unpack OpenTelemetry metrics: %w", err)
 	}
@@ -257,11 +257,16 @@ func resetLabels(labels []prompbmarshal.Label) []prompbmarshal.Label {
 	return labels[:0]
 }
 
-func (wr *writeContext) readAndUnpackRequest(r io.Reader) (*pb.ExportMetricsServiceRequest, error) {
+func (wr *writeContext) readAndUnpackRequest(r io.Reader, processBody func(*[]byte) error) (*pb.ExportMetricsServiceRequest, error) {
 	if _, err := wr.bb.ReadFrom(r); err != nil {
 		return nil, fmt.Errorf("cannot read request: %w", err)
 	}
 	var req pb.ExportMetricsServiceRequest
+	if processBody != nil {
+		if err := processBody(&wr.bb.B); err != nil {
+			return nil, fmt.Errorf("cannot process request body: %w", err)
+		}
+	}
 	if err := req.UnmarshalProtobuf(wr.bb.B); err != nil {
 		return nil, fmt.Errorf("cannot unmarshal request from %d bytes: %w", len(wr.bb.B), err)
 	}


Test patch:
diff --git a/lib/protoparser/firehose/parser_test.go b/lib/protoparser/firehose/parser_test.go
new file mode 100644
index 000000000..647f927df
--- /dev/null
+++ b/lib/protoparser/firehose/parser_test.go
@@ -0,0 +1,16 @@
+package firehose
+
+import (
+	"testing"
+)
+
+func TestProcessRequestBody(t *testing.T) {
+	data := []byte(`{"records": [{"data": "SGVsbG8sIA=="}, {"data": "d29ybGQh"}]}`)
+	err := ProcessRequestBody(&data)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if string(data) != "Hello, world!" {
+		t.Fatalf("unexpected string: %q", string(data))
+	}
+}
diff --git a/lib/protoparser/opentelemetry/stream/streamparser_test.go b/lib/protoparser/opentelemetry/stream/streamparser_test.go
index c5de4c38a..1d5281d4a 100644
--- a/lib/protoparser/opentelemetry/stream/streamparser_test.go
+++ b/lib/protoparser/opentelemetry/stream/streamparser_test.go
@@ -121,7 +121,7 @@ func TestParseStream(t *testing.T) {
 
 func checkParseStream(data []byte, checkSeries func(tss []prompbmarshal.TimeSeries) error) error {
 	// Verify parsing without compression
-	if err := ParseStream(bytes.NewBuffer(data), false, checkSeries); err != nil {
+	if err := ParseStream(bytes.NewBuffer(data), false, nil, checkSeries); err != nil {
 		return fmt.Errorf("error when parsing data: %w", err)
 	}
 
@@ -134,7 +134,7 @@ func checkParseStream(data []byte, checkSeries func(tss []prompbmarshal.TimeSeri
 	if err := zw.Close(); err != nil {
 		return fmt.Errorf("cannot close gzip writer: %w", err)
 	}
-	if err := ParseStream(&bb, true, checkSeries); err != nil {
+	if err := ParseStream(&bb, true, nil, checkSeries); err != nil {
 		return fmt.Errorf("error when parsing compressed data: %w", err)
 	}
 
diff --git a/lib/protoparser/opentelemetry/stream/streamparser_timing_test.go b/lib/protoparser/opentelemetry/stream/streamparser_timing_test.go
index c2186cb95..e1f73aaf1 100644
--- a/lib/protoparser/opentelemetry/stream/streamparser_timing_test.go
+++ b/lib/protoparser/opentelemetry/stream/streamparser_timing_test.go
@@ -24,7 +24,7 @@ func BenchmarkParseStream(b *testing.B) {
 		data := pbRequest.MarshalProtobuf(nil)
 
 		for p.Next() {
-			err := ParseStream(bytes.NewBuffer(data), false, func(tss []prompbmarshal.TimeSeries) error {
+			err := ParseStream(bytes.NewBuffer(data), false, nil, func(tss []prompbmarshal.TimeSeries) error {
 				return nil
 			})
 			if err != nil {