From 73d18fbc7a39a87c04456aac7ff0303e0784cd19 Mon Sep 17 00:00:00 2001 From: hagen1778 Date: Tue, 28 Nov 2023 15:04:15 +0100 Subject: [PATCH] lib/protoparser/datadog: follow-up after 543f218fe96574b9b2189c8350bb09afa349e3bb * prevent /api/v1 from panic on parsing rows * add tests for Extract function for v1 and v2 api's * separate request types in different pools to prevent different objects mixing * add changelog line https://github.com/VictoriaMetrics/VictoriaMetrics/commit/543f218fe96574b9b2189c8350bb09afa349e3bb Signed-off-by: hagen1778 (cherry picked from commit 98d0f81f21342296b20e13e2855dec9156c8b73c) Signed-off-by: hagen1778 --- docs/CHANGELOG.md | 1 + lib/protoparser/datadog/api/series/v1/api.go | 4 +- .../datadog/api/series/v1/api_test.go | 49 ++++++++++----- .../datadog/api/series/v2/api_test.go | 55 ++++++++++++----- .../datadog/api/sketches/beta/api.go | 4 +- .../datadog/stream/streamparser.go | 61 ++++++++++++++----- 6 files changed, 122 insertions(+), 52 deletions(-) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 0fd86aa715..b32eabd4d6 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -30,6 +30,7 @@ The sandbox cluster installation is running under the constant load generated by * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): add `-remoteWrite.disableOnDiskQueue` command-line flag, which can be used for disabling data queueing to disk when the remote storage cannot keep up with the data ingestion rate. See [these docs](https://docs.victoriametrics.com/vmagent.html#disabling-on-disk-persistence) and [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/2110). * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): add support for reading and writing samples via [Google PubSub](https://cloud.google.com/pubsub). See [these docs](https://docs.victoriametrics.com/vmagent.html#google-pubsub-integration). +* FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): add support for Datadog `/api/v2/series` and `/api/beta/sketches` ingestion protocols to vmagent/vminsert components. See this [doc](https://docs.victoriametrics.com/#how-to-send-data-from-datadog-agent) for examples. Thanks to @AndrewChubatiuk for the [pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/5094). * FEATURE: reduce the default value for `-import.maxLineLen` command-line flag from 100MB to 10MB in order to prevent excessive memory usage during data import via [/api/v1/import](https://docs.victoriametrics.com/#how-to-import-data-in-json-line-format). * FEATURE: [MetricsQL](https://docs.victoriametrics.com/MetricsQL.html): add [day_of_year()](https://docs.victoriametrics.com/MetricsQL.html#day_of_year) function, which returns the day of the year for each of the given unix timestamps. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5345) for details. Thanks to @luckyxiaoqiang for the [pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/5368/). diff --git a/lib/protoparser/datadog/api/series/v1/api.go b/lib/protoparser/datadog/api/series/v1/api.go index 135b911934..cecf98f0a8 100644 --- a/lib/protoparser/datadog/api/series/v1/api.go +++ b/lib/protoparser/datadog/api/series/v1/api.go @@ -30,10 +30,10 @@ func (r *Request) Extract(fn func(prompbmarshal.TimeSeries) error, sanitizeFn fu if ts <= 0 { ts = float64(currentTimestamp) } - samples[j] = prompbmarshal.Sample{ + samples = append(samples, prompbmarshal.Sample{ Timestamp: int64(ts * 1000), Value: val, - } + }) } ts := prompbmarshal.TimeSeries{ Samples: samples, diff --git a/lib/protoparser/datadog/api/series/v1/api_test.go b/lib/protoparser/datadog/api/series/v1/api_test.go index 7573ccd428..b218e0665a 100644 --- a/lib/protoparser/datadog/api/series/v1/api_test.go +++ b/lib/protoparser/datadog/api/series/v1/api_test.go @@ -3,6 +3,8 @@ package datadog import ( "reflect" "testing" + + "github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal" ) func TestRequestUnmarshalFailure(t *testing.T) { @@ -20,22 +22,37 @@ func TestRequestUnmarshalFailure(t *testing.T) { f(`[]`) } -func unmarshalRequestValidator(t *testing.T, s []byte, reqExpected *Request) { - t.Helper() - req := new(Request) - if err := req.Unmarshal(s); err != nil { - t.Fatalf("unexpected error in Unmarshal(%q): %s", s, err) - } - if !reflect.DeepEqual(req, reqExpected) { - t.Fatalf("unexpected row;\ngot\n%+v\nwant\n%+v", req, reqExpected) - } -} +func TestRequestExtract(t *testing.T) { + fn := func(s []byte, reqExpected *Request, samplesExp int) { + t.Helper() + req := new(Request) + if err := req.Unmarshal(s); err != nil { + t.Fatalf("unexpected error in Unmarshal(%q): %s", s, err) + } + if !reflect.DeepEqual(req, reqExpected) { + t.Fatalf("unexpected row;\ngot\n%+v\nwant\n%+v", req, reqExpected) + } -func TestRequestUnmarshalSuccess(t *testing.T) { - unmarshalRequestValidator( - t, []byte("{}"), new(Request), - ) - unmarshalRequestValidator(t, []byte(` + var samplesTotal int + cb := func(ts prompbmarshal.TimeSeries) error { + samplesTotal += len(ts.Samples) + return nil + } + sanitizeFn := func(name string) string { + return name + } + if err := req.Extract(cb, sanitizeFn); err != nil { + t.Fatalf("error when extracting data: %s", err) + } + + if samplesTotal != samplesExp { + t.Fatalf("expected to extract %d samples; got %d", samplesExp, samplesTotal) + } + + } + + fn([]byte("{}"), new(Request), 0) + fn([]byte(` { "series": [ { @@ -67,5 +84,5 @@ func TestRequestUnmarshalSuccess(t *testing.T) { "environment:test", }, }}, - }) + }, 1) } diff --git a/lib/protoparser/datadog/api/series/v2/api_test.go b/lib/protoparser/datadog/api/series/v2/api_test.go index a40a4776ad..4655d2e834 100644 --- a/lib/protoparser/datadog/api/series/v2/api_test.go +++ b/lib/protoparser/datadog/api/series/v2/api_test.go @@ -3,6 +3,8 @@ package datadog import ( "reflect" "testing" + + "github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal" ) func TestRequestUnmarshalFailure(t *testing.T) { @@ -20,22 +22,37 @@ func TestRequestUnmarshalFailure(t *testing.T) { f(`[]`) } -func unmarshalRequestValidator(t *testing.T, s []byte, reqExpected *Request) { - t.Helper() - req := new(Request) - if err := req.Unmarshal(s); err != nil { - t.Fatalf("unexpected error in Unmarshal(%q): %s", s, err) - } - if !reflect.DeepEqual(req, reqExpected) { - t.Fatalf("unexpected row;\ngot\n%+v\nwant\n%+v", req, reqExpected) - } -} +func TestRequestExtract(t *testing.T) { + fn := func(s []byte, reqExpected *Request, samplesExp int) { + t.Helper() + req := new(Request) + if err := req.Unmarshal(s); err != nil { + t.Fatalf("unexpected error in Unmarshal(%q): %s", s, err) + } + if !reflect.DeepEqual(req, reqExpected) { + t.Fatalf("unexpected row;\ngot\n%+v\nwant\n%+v", req, reqExpected) + } -func TestRequestUnmarshalSuccess(t *testing.T) { - unmarshalRequestValidator( - t, []byte("{}"), new(Request), - ) - unmarshalRequestValidator(t, []byte(` + var samplesTotal int + cb := func(ts prompbmarshal.TimeSeries) error { + samplesTotal += len(ts.Samples) + return nil + } + sanitizeFn := func(name string) string { + return name + } + if err := req.Extract(cb, sanitizeFn); err != nil { + t.Fatalf("error when extracting data: %s", err) + } + + if samplesTotal != samplesExp { + t.Fatalf("expected to extract %d samples; got %d", samplesExp, samplesTotal) + } + + } + fn([]byte("{}"), new(Request), 0) + + fn([]byte(` { "series": [ { @@ -53,6 +70,9 @@ func TestRequestUnmarshalSuccess(t *testing.T) { "points": [{ "timestamp": 1575317847, "value": 0.5 + },{ + "timestamp": 1575317848, + "value": 0.6 }], "tags": [ "environment:test" @@ -74,10 +94,13 @@ func TestRequestUnmarshalSuccess(t *testing.T) { Points: []point{{ Timestamp: 1575317847, Value: 0.5, + }, { + Timestamp: 1575317848, + Value: 0.6, }}, Tags: []string{ "environment:test", }, }}, - }) + }, 2) } diff --git a/lib/protoparser/datadog/api/sketches/beta/api.go b/lib/protoparser/datadog/api/sketches/beta/api.go index 5577a6c9c2..4c857871d0 100644 --- a/lib/protoparser/datadog/api/sketches/beta/api.go +++ b/lib/protoparser/datadog/api/sketches/beta/api.go @@ -25,7 +25,7 @@ func (r *Request) Extract(fn func(prompbmarshal.TimeSeries) error, sanitizeFn fu for _, sketch := range r.SketchPayload.Sketches { sketchSeries := make([]prompbmarshal.TimeSeries, 5) for _, point := range sketch.Dogsketches { - timestamp := int64(point.Ts * 1000) + timestamp := point.Ts * 1000 updateSeries(sketchSeries, sanitizeFn(sketch.Metric), timestamp, map[string]float64{ "max": point.Max, "min": point.Min, @@ -35,7 +35,7 @@ func (r *Request) Extract(fn func(prompbmarshal.TimeSeries) error, sanitizeFn fu }) } for _, point := range sketch.Distributions { - timestamp := int64(point.Ts * 1000) + timestamp := point.Ts * 1000 updateSeries(sketchSeries, sanitizeFn(sketch.Metric), timestamp, map[string]float64{ "max": point.Max, "min": point.Min, diff --git a/lib/protoparser/datadog/stream/streamparser.go b/lib/protoparser/datadog/stream/streamparser.go index 1279829b75..24beb06814 100644 --- a/lib/protoparser/datadog/stream/streamparser.go +++ b/lib/protoparser/datadog/stream/streamparser.go @@ -70,35 +70,36 @@ func Parse(req *http.Request, callback func(prompbmarshal.TimeSeries) error) err apiVersion := insertApisVersionRegex.ReplaceAllString(req.URL.Path, "${version}") apiKind := insertApisVersionRegex.ReplaceAllString(req.URL.Path, "${kind}") - ddReq := getRequest() - defer putRequest(ddReq) - + var ddReq datadog.Request switch apiKind { case "series": switch apiVersion { case "v1": - ddReq = new(apiSeriesV1.Request) + ddReq = getSeriesV1Request() + defer putSeriesV1Request(ddReq) case "v2": - ddReq = new(apiSeriesV2.Request) + ddReq = getSeriesV2Request() + defer putSeriesV2Request(ddReq) default: return fmt.Errorf( - "API version %q of Datadog series endpoint is not supported", + "API version %q of DataDog series endpoint is not supported", apiVersion, ) } case "sketches": switch apiVersion { case "beta": - ddReq = new(apiSketchesBeta.Request) + ddReq = getSketchesBetaRequest() + defer putSketchesBetaRequest(ddReq) default: return fmt.Errorf( - "API version %q of Datadog sketches endpoint is not supported", + "API version %q of DataDog sketches endpoint is not supported", apiVersion, ) } default: return fmt.Errorf( - "API kind %q of Datadog API is not supported", + "API kind %q of DataDog API is not supported", apiKind, ) } @@ -182,19 +183,47 @@ func putPushCtx(ctx *pushCtx) { var pushCtxPool sync.Pool var pushCtxPoolCh = make(chan *pushCtx, cgroup.AvailableCPUs()) -func getRequest() datadog.Request { - v := requestPool.Get() +func getSeriesV1Request() *apiSeriesV1.Request { + v := seriesV1RequestPool.Get() if v == nil { - return nil + return &apiSeriesV1.Request{} } - return v.(datadog.Request) + return v.(*apiSeriesV1.Request) } -func putRequest(req datadog.Request) { - requestPool.Put(req) +func putSeriesV1Request(req datadog.Request) { + seriesV1RequestPool.Put(req) } -var requestPool sync.Pool +var seriesV1RequestPool sync.Pool + +func getSeriesV2Request() *apiSeriesV2.Request { + v := seriesV2RequestPool.Get() + if v == nil { + return &apiSeriesV2.Request{} + } + return v.(*apiSeriesV2.Request) +} + +func putSeriesV2Request(req datadog.Request) { + seriesV2RequestPool.Put(req) +} + +var seriesV2RequestPool sync.Pool + +func getSketchesBetaRequest() *apiSketchesBeta.Request { + v := sketchesBetaRequestPool.Get() + if v == nil { + return &apiSketchesBeta.Request{} + } + return v.(*apiSketchesBeta.Request) +} + +func putSketchesBetaRequest(req datadog.Request) { + sketchesBetaRequestPool.Put(req) +} + +var sketchesBetaRequestPool sync.Pool // sanitizeName performs DataDog-compatible sanitizing for metric names //