From f987fb9c8b18c5ff1657b899c7c85db34d0effd3 Mon Sep 17 00:00:00 2001 From: Aliaksandr Valialkin Date: Mon, 13 Feb 2023 10:46:50 -0800 Subject: [PATCH] lib/protoparser/promremotewrite: extract stream parsing code into a separate stream package This is a follow-up for 057698f7fbfd1d1322a888f7b26bebead83fe115 --- app/vmagent/promremotewrite/request_handler.go | 4 ++-- app/vminsert/promremotewrite/request_handler.go | 4 ++-- .../promremotewrite/{ => stream}/streamparser.go | 6 +++--- 3 files changed, 7 insertions(+), 7 deletions(-) rename lib/protoparser/promremotewrite/{ => stream}/streamparser.go (94%) diff --git a/app/vmagent/promremotewrite/request_handler.go b/app/vmagent/promremotewrite/request_handler.go index e3e08abe6b..acba222a99 100644 --- a/app/vmagent/promremotewrite/request_handler.go +++ b/app/vmagent/promremotewrite/request_handler.go @@ -10,7 +10,7 @@ import ( "github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb" "github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal" parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common" - parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/promremotewrite" + "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/promremotewrite/stream" "github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics" "github.com/VictoriaMetrics/metrics" ) @@ -27,7 +27,7 @@ func InsertHandler(at *auth.Token, req *http.Request) error { if err != nil { return err } - return parser.ParseStream(req.Body, func(tss []prompb.TimeSeries) error { + return stream.Parse(req.Body, func(tss []prompb.TimeSeries) error { return insertRows(at, tss, extraLabels) }) } diff --git a/app/vminsert/promremotewrite/request_handler.go b/app/vminsert/promremotewrite/request_handler.go index 3c51828c92..4fc5912279 100644 --- a/app/vminsert/promremotewrite/request_handler.go +++ b/app/vminsert/promremotewrite/request_handler.go @@ -9,7 +9,7 @@ import ( "github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb" "github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal" parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common" - parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/promremotewrite" + "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/promremotewrite/stream" "github.com/VictoriaMetrics/VictoriaMetrics/lib/storage" "github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics" "github.com/VictoriaMetrics/metrics" @@ -27,7 +27,7 @@ func InsertHandler(at *auth.Token, req *http.Request) error { if err != nil { return err } - return parser.ParseStream(req.Body, func(tss []prompb.TimeSeries) error { + return stream.Parse(req.Body, func(tss []prompb.TimeSeries) error { return insertRows(at, tss, extraLabels) }) } diff --git a/lib/protoparser/promremotewrite/streamparser.go b/lib/protoparser/promremotewrite/stream/streamparser.go similarity index 94% rename from lib/protoparser/promremotewrite/streamparser.go rename to lib/protoparser/promremotewrite/stream/streamparser.go index 44a5072605..23d7693c74 100644 --- a/lib/protoparser/promremotewrite/streamparser.go +++ b/lib/protoparser/promremotewrite/stream/streamparser.go @@ -1,4 +1,4 @@ -package promremotewrite +package stream import ( "bufio" @@ -18,10 +18,10 @@ import ( var maxInsertRequestSize = flagutil.NewBytes("maxInsertRequestSize", 32*1024*1024, "The maximum size in bytes of a single Prometheus remote_write API request") -// ParseStream parses Prometheus remote_write message from reader and calls callback for the parsed timeseries. +// Parse parses Prometheus remote_write message from reader and calls callback for the parsed timeseries. // // callback shouldn't hold tss after returning. -func ParseStream(r io.Reader, callback func(tss []prompb.TimeSeries) error) error { +func Parse(r io.Reader, callback func(tss []prompb.TimeSeries) error) error { wcr := writeconcurrencylimiter.GetReader(r) defer writeconcurrencylimiter.PutReader(wcr) r = wcr