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

app/vmselect/netstorage: properly read trace from vmstorage when it returns error message to vmselect

Browse source
This commit is contained in:
Aliaksandr Valialkin 2022-06-01 14:35:00 +03:00
parent fedfc9e686
commit c92bc5394f
No known key found for this signature in database
GPG key ID: A72BEC6CD3D0DED1
1 changed files with 17 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
app/vmselect/netstorage/netstorage.go
View file

@ -2005,6 +2005,7 @@ func (sn *storageNode) execOnConn(qt *querytracer.Tracer, funcName string, f fun
var er *errRemote var er *errRemote
if errors.As(err, &er) { if errors.As(err, &er) {
// Remote error. The connection may be re-used. Return it to the pool. // Remote error. The connection may be re-used. Return it to the pool.
_ = readTrace(qt, bc)
sn.connPool.Put(bc) sn.connPool.Put(bc)
} else { } else {
// Local error. // Local error.
@ -2019,26 +2020,31 @@ func (sn *storageNode) execOnConn(qt *querytracer.Tracer, funcName string, f fun
} }
// Read trace from the response // Read trace from the response
bb := traceJSONBufPool.Get() if err := readTrace(qt, bc); err != nil {
bb.B, err = readBytes(bb.B[:0], bc, maxTraceJSONSize)
if err != nil {
// Close the connection instead of returning it to the pool, // Close the connection instead of returning it to the pool,
// since it may be broken. // since it may be broken.
_ = bc.Close() _ = bc.Close()
return fmt.Errorf("cannot read trace for funcName=%q from the server: %w", funcName, err) return err
} }
if err := qt.AddJSON(bb.B); err != nil {
// Close the connection instead of returning it to the pool,
// since it may be broken.
_ = bc.Close()
return fmt.Errorf("cannot read trace for funcName=%q from the server: %w", funcName, err)
}
traceJSONBufPool.Put(bb)
// Return the connection back to the pool, assuming it is healthy. // Return the connection back to the pool, assuming it is healthy.
sn.connPool.Put(bc) sn.connPool.Put(bc)
return nil return nil
} }
func readTrace(qt *querytracer.Tracer, bc *handshake.BufferedConn) error {
bb := traceJSONBufPool.Get()
var err error
bb.B, err = readBytes(bb.B[:0], bc, maxTraceJSONSize)
if err != nil {
return fmt.Errorf("cannot read trace from the server: %w", err)
}
if err := qt.AddJSON(bb.B); err != nil {
return fmt.Errorf("cannot parse trace read from the server: %w", err)
}
traceJSONBufPool.Put(bb)
return nil
}
var traceJSONBufPool bytesutil.ByteBufferPool var traceJSONBufPool bytesutil.ByteBufferPool
const maxTraceJSONSize = 1024 * 1024 const maxTraceJSONSize = 1024 * 1024