github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2024-11-21 14:44:00 +00:00
Code Issues Projects Releases Packages Wiki Activity

lib/protoparser/datadog: extract stream parsing code into a separate stream package

Browse source 
This is a follow-up for 057698f7fb
This commit is contained in:
Aliaksandr Valialkin 2023-02-13 09:51:35 -08:00
parent f3be9483f4
commit f83d6d69b2
No known key found for this signature in database
GPG key ID: A72BEC6CD3D0DED1
4 changed files with 12 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
app/vmagent/datadog/request_handler.go
View file

@ -9,6 +9,7 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadog"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadog/stream"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
"github.com/VictoriaMetrics/metrics"
)
@ -28,7 +29,7 @@ func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
return err
}
ce := req.Header.Get("Content-Encoding")
return parser.ParseStream(req.Body, ce, func(series []parser.Series) error {
return stream.Parse(req.Body, ce, func(series []parser.Series) error {
return insertRows(at, series, extraLabels)
})
}

3
app/vminsert/datadog/request_handler.go
View file

@ -8,6 +8,7 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadog"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadog/stream"
"github.com/VictoriaMetrics/metrics"
)
@ -25,7 +26,7 @@ func InsertHandlerForHTTP(req *http.Request) error {
return err
}
ce := req.Header.Get("Content-Encoding")
return parser.ParseStream(req.Body, ce, func(series []parser.Series) error {
return stream.Parse(req.Body, ce, func(series []parser.Series) error {
return insertRows(series, extraLabels)
})
}

13
lib/protoparser/datadog/streamparser.go → lib/protoparser/datadog/stream/streamparser.go
View file

@ -1,4 +1,4 @@
package datadog
package stream
import (
"bufio"
@ -13,6 +13,7 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadog"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
"github.com/VictoriaMetrics/metrics"
)
@ -33,7 +34,7 @@ var (
// ParseStream parses DataDog POST request for /api/v1/series from reader and calls callback for the parsed request.
//
// callback shouldn't hold series after returning.
func ParseStream(r io.Reader, contentEncoding string, callback func(series []Series) error) error {
func Parse(r io.Reader, contentEncoding string, callback func(series []datadog.Series) error) error {
wcr := writeconcurrencylimiter.GetReader(r)
defer writeconcurrencylimiter.PutReader(wcr)
r = wcr
@ -143,15 +144,15 @@ func putPushCtx(ctx *pushCtx) {
var pushCtxPool sync.Pool
var pushCtxPoolCh = make(chan *pushCtx, cgroup.AvailableCPUs())
func getRequest() *Request {
func getRequest() *datadog.Request {
v := requestPool.Get()
if v == nil {
return &Request{}
return &datadog.Request{}
}
return v.(*Request)
return v.(*datadog.Request)
}
func putRequest(req *Request) {
func putRequest(req *datadog.Request) {
requestPool.Put(req)
}

2
lib/protoparser/datadog/streamparser_test.go → lib/protoparser/datadog/stream/streamparser_test.go
View file

@ -1,4 +1,4 @@
package datadog
package stream
import (
"testing"