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

app/vminsert: do not drop data in reroutedBuf if all the storage nodes are unhealthy

Browse source
This commit is contained in:
Aliaksandr Valialkin 2019-08-23 10:38:19 +03:00
parent a5dc54efc3
commit 697de90893
1 changed files with 6 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
app/vminsert/netstorage/netstorage.go
View file

@ -343,6 +343,12 @@ func addToReroutedBuf(buf []byte, rows int) error {
}
func spreadReroutedBufToStorageNodes(swapBuf []byte) ([]byte, error) {
healthyStorageNodes := getHealthyStorageNodes()
if len(healthyStorageNodes) == 0 {
// No more vmstorage nodes to write data to.
return swapBuf, fmt.Errorf("all the storage nodes are unhealthy")
}
reroutedLock.Lock()
reroutedBuf, swapBuf = swapBuf[:0], reroutedBuf
rows := reroutedRows
@ -354,26 +360,6 @@ func spreadReroutedBufToStorageNodes(swapBuf []byte) ([]byte, error) {
return swapBuf, nil
}
healthyStorageNodes := getHealthyStorageNodes()
if len(healthyStorageNodes) == 0 {
// No more vmstorage nodes to write data to.
// Try returning the the data to reroutedBuf if it has enough free space.
recovered := false
reroutedLock.Lock()
if len(swapBuf)+len(reroutedBuf) <= reroutedBufMaxSize {
swapBuf = append(swapBuf, reroutedBuf...)
reroutedBuf, swapBuf = swapBuf, reroutedBuf[:0]
reroutedRows += rows
recovered = true
}
reroutedLock.Unlock()
if recovered {
return swapBuf, nil
}
rowsLostTotal.Add(rows)
return swapBuf, fmt.Errorf("all the %d vmstorage nodes are unealthy; lost %d rows", len(storageNodes), rows)
}
var mr storage.MetricRow
src := swapBuf
rowsProcessed := 0