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/storage: return the first error instead of the last error, since the first error usually points to the root cause

Browse source
This commit is contained in:
Aliaksandr Valialkin 2020-05-14 23:17:22 +03:00
parent 3845420a8f
commit 71d29a8fa1
1 changed files with 34 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
lib/storage/storage.go
View file

@ -1101,8 +1101,8 @@ func (s *Storage) add(rows []rawRow, mrs []MetricRow, precisionBits uint8) ([]ra
prevMetricNameRaw []byte prevMetricNameRaw []byte
) )
minTimestamp, maxTimestamp := s.tb.getMinMaxTimestamps() minTimestamp, maxTimestamp := s.tb.getMinMaxTimestamps()
// Return only the last error, since it has no sense in returning all errors. // Return only the first error, since it has no sense in returning all errors.
var lastWarn error var firstWarn error
for i := range mrs { for i := range mrs {
mr := &mrs[i] mr := &mrs[i]
if math.IsNaN(mr.Value) { if math.IsNaN(mr.Value) {
@ -1112,13 +1112,19 @@ func (s *Storage) add(rows []rawRow, mrs []MetricRow, precisionBits uint8) ([]ra
} }
if mr.Timestamp < minTimestamp { if mr.Timestamp < minTimestamp {
// Skip rows with too small timestamps outside the retention. // Skip rows with too small timestamps outside the retention.
lastWarn = fmt.Errorf("cannot insert row with too small timestamp %d outside the retention; minimum allowed timestamp is %d", mr.Timestamp, minTimestamp) if firstWarn == nil {
firstWarn = fmt.Errorf("cannot insert row with too small timestamp %d outside the retention; minimum allowed timestamp is %d",
mr.Timestamp, minTimestamp)
}
atomic.AddUint64(&s.tooSmallTimestampRows, 1) atomic.AddUint64(&s.tooSmallTimestampRows, 1)
continue continue
} }
if mr.Timestamp > maxTimestamp { if mr.Timestamp > maxTimestamp {
// Skip rows with too big timestamps significantly exceeding the current time. // Skip rows with too big timestamps significantly exceeding the current time.
lastWarn = fmt.Errorf("cannot insert row with too big timestamp %d exceeding the current time; maximum allowd timestamp is %d", mr.Timestamp, maxTimestamp) if firstWarn == nil {
firstWarn = fmt.Errorf("cannot insert row with too big timestamp %d exceeding the current time; maximum allowd timestamp is %d",
mr.Timestamp, maxTimestamp)
}
atomic.AddUint64(&s.tooBigTimestampRows, 1) atomic.AddUint64(&s.tooBigTimestampRows, 1)
continue continue
} }
@ -1152,7 +1158,9 @@ func (s *Storage) add(rows []rawRow, mrs []MetricRow, precisionBits uint8) ([]ra
// Do not stop adding rows on error - just skip invalid row. // Do not stop adding rows on error - just skip invalid row.
// This guarantees that invalid rows don't prevent // This guarantees that invalid rows don't prevent
// from adding valid rows into the storage. // from adding valid rows into the storage.
lastWarn = fmt.Errorf("cannot unmarshal MetricNameRaw %q: %s", mr.MetricNameRaw, err) if firstWarn == nil {
firstWarn = fmt.Errorf("cannot unmarshal MetricNameRaw %q: %s", mr.MetricNameRaw, err)
}
j-- j--
continue continue
} }
@ -1162,14 +1170,16 @@ func (s *Storage) add(rows []rawRow, mrs []MetricRow, precisionBits uint8) ([]ra
// Do not stop adding rows on error - just skip invalid row. // Do not stop adding rows on error - just skip invalid row.
// This guarantees that invalid rows don't prevent // This guarantees that invalid rows don't prevent
// from adding valid rows into the storage. // from adding valid rows into the storage.
lastWarn = fmt.Errorf("cannot obtain TSID for MetricName %q: %s", kb.B, err) if firstWarn == nil {
firstWarn = fmt.Errorf("cannot obtain TSID for MetricName %q: %s", kb.B, err)
}
j-- j--
continue continue
} }
s.putTSIDToCache(&r.TSID, mr.MetricNameRaw) s.putTSIDToCache(&r.TSID, mr.MetricNameRaw)
} }
if lastWarn != nil { if firstWarn != nil {
logger.Errorf("warn occurred during rows addition: %s", lastWarn) logger.Errorf("warn occurred during rows addition: %s", firstWarn)
} }
if is != nil { if is != nil {
kbPool.Put(kb) kbPool.Put(kb)
@ -1178,21 +1188,21 @@ func (s *Storage) add(rows []rawRow, mrs []MetricRow, precisionBits uint8) ([]ra
} }
rows = rows[:rowsLen+j] rows = rows[:rowsLen+j]
var lastError error var firstError error
if err := s.tb.AddRows(rows); err != nil { if err := s.tb.AddRows(rows); err != nil {
lastError = fmt.Errorf("cannot add rows to table: %s", err) firstError = fmt.Errorf("cannot add rows to table: %s", err)
} }
if err := s.updatePerDateData(rows); err != nil && lastError == nil { if err := s.updatePerDateData(rows); err != nil && firstError == nil {
lastError = fmt.Errorf("cannot update per-date data: %s", err) firstError = fmt.Errorf("cannot update per-date data: %s", err)
} }
if lastError != nil { if firstError != nil {
return rows, fmt.Errorf("error occurred during rows addition: %s", lastError) return rows, fmt.Errorf("error occurred during rows addition: %s", firstError)
} }
return rows, nil return rows, nil
} }
func (s *Storage) updatePerDateData(rows []rawRow) error { func (s *Storage) updatePerDateData(rows []rawRow) error {
var lastError error var firstError error
var date uint64 var date uint64
var hour uint64 var hour uint64
var prevTimestamp int64 var prevTimestamp int64
@ -1226,8 +1236,10 @@ func (s *Storage) updatePerDateData(rows []rawRow) error {
// when entries for all the active time series must be added to the index. // when entries for all the active time series must be added to the index.
// This should address https://github.com/VictoriaMetrics/VictoriaMetrics/issues/430 . // This should address https://github.com/VictoriaMetrics/VictoriaMetrics/issues/430 .
if todayShare16bit > (metricID&(1<<16-1)) && !nextDayMetricIDs.Has(metricID) { if todayShare16bit > (metricID&(1<<16-1)) && !nextDayMetricIDs.Has(metricID) {
if err := idb.storeDateMetricID(date+1, metricID, r.TSID.AccountID, r.TSID.ProjectID); err != nil && lastError == nil { if err := idb.storeDateMetricID(date+1, metricID, r.TSID.AccountID, r.TSID.ProjectID); err != nil {
if lastError == nil {
lastError = err lastError = err
}
continue continue
} }
s.pendingNextDayMetricIDsLock.Lock() s.pendingNextDayMetricIDsLock.Lock()
@ -1261,15 +1273,17 @@ func (s *Storage) updatePerDateData(rows []rawRow) error {
// Slow path: store the (date, metricID) entry in the indexDB. // Slow path: store the (date, metricID) entry in the indexDB.
// It is OK if the (date, metricID) entry is added multiple times to db // It is OK if the (date, metricID) entry is added multiple times to db
// by concurrent goroutines. // by concurrent goroutines.
if err := idb.storeDateMetricID(date, metricID, r.TSID.AccountID, r.TSID.ProjectID); err != nil && lastError == nil { if err := idb.storeDateMetricID(date, metricID, r.TSID.AccountID, r.TSID.ProjectID); err != nil {
if lastError == nil {
lastError = err lastError = err
}
continue continue
} }
// The metric must be added to cache only after it has been successfully added to indexDB. // The metric must be added to cache only after it has been successfully added to indexDB.
s.dateMetricIDCache.Set(date, metricID) s.dateMetricIDCache.Set(date, metricID)
} }
return lastError return firstError
} }
// dateMetricIDCache is fast cache for holding (date, metricID) entries. // dateMetricIDCache is fast cache for holding (date, metricID) entries.