github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2025-03-11 15:34:56 +00:00
Code Issues Projects Releases Packages Wiki Activity

lib/protoparser/clusternative: 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 10:37:53 -08:00
parent 75cf5a8939
commit 086516a02b
No known key found for this signature in database
GPG key ID: A72BEC6CD3D0DED1
3 changed files with 7 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
app/vminsert/clusternative/request_handler.go
View file

@ -8,7 +8,7 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/handshake"
parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/clusternative"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/clusternative/stream"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
"github.com/VictoriaMetrics/metrics"
@ -26,7 +26,7 @@ func InsertHandler(c net.Conn) error {
if err != nil {
return fmt.Errorf("cannot perform vminsert handshake with client %q: %w", c.RemoteAddr(), err)
}
return parser.ParseStream(bc, func(rows []storage.MetricRow) error {
return stream.Parse(bc, func(rows []storage.MetricRow) error {
return insertRows(rows)
}, nil)
}

4
app/vmstorage/servers/vminsert.go
View file

@ -12,7 +12,7 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/clusternative"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/clusternative/stream"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
"github.com/VictoriaMetrics/metrics"
)
@ -110,7 +110,7 @@ func (s *VMInsertServer) run() {
}()
logger.Infof("processing vminsert conn from %s", c.RemoteAddr())
err = clusternative.ParseStream(bc, func(rows []storage.MetricRow) error {
err = stream.Parse(bc, func(rows []storage.MetricRow) error {
vminsertMetricsRead.Add(len(rows))
return s.storage.AddRows(rows, uint8(*precisionBits))
}, s.storage.IsReadOnly)

6
lib/protoparser/clusternative/streamparser.go → lib/protoparser/clusternative/stream/streamparser.go
View file

@ -1,4 +1,4 @@
package clusternative
package stream
import (
"fmt"
@ -17,14 +17,14 @@ import (
"github.com/VictoriaMetrics/metrics"
)
// ParseStream parses data sent from vminsert to bc and calls callback for parsed rows.
// Parse parses data sent from vminsert to bc and calls callback for parsed rows.
// Optional function isReadOnly must return true if the storage cannot accept new data.
// In this case the data read from bc isn't accepted and the readonly status is sent back bc.
//
// The callback can be called concurrently multiple times for streamed data from req.
//
// callback shouldn't hold block after returning.
func ParseStream(bc *handshake.BufferedConn, callback func(rows []storage.MetricRow) error, isReadOnly func() bool) error {
func Parse(bc *handshake.BufferedConn, callback func(rows []storage.MetricRow) error, isReadOnly func() bool) error {
wcr := writeconcurrencylimiter.GetReader(bc)
defer writeconcurrencylimiter.PutReader(wcr)
r := io.Reader(wcr)