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

wip

Browse source
This commit is contained in:
Aliaksandr Valialkin 2023-09-22 13:20:28 +02:00
parent 0be8960875
commit baaec4c3e1
No known key found for this signature in database
GPG key ID: A72BEC6CD3D0DED1
1 changed files with 35 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
lib/storage/storage.go
View file

@ -1989,6 +1989,7 @@ func (s *Storage) add(rows []rawRow, dstMrs []*MetricRow, mrs []MetricRow, preci
if err != nil { if err != nil {
err = fmt.Errorf("cannot update per-date data: %w", err) err = fmt.Errorf("cannot update per-date data: %w", err)
} else { } else {
s.addMissingMetricIDToTSIDEntries(rows)
s.tb.MustAddRows(rows) s.tb.MustAddRows(rows)
} }
if err != nil { if err != nil {
@ -1997,6 +1998,40 @@ func (s *Storage) add(rows []rawRow, dstMrs []*MetricRow, mrs []MetricRow, preci
return nil return nil
} }
func (s *Storage) addMissingMetricIDToTSIDEntries(rows []rawRow) {
// this function is needed for recovering missing metricID -> TSID entries
// in the issue https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4972
idb := s.idb()
is := idb.getIndexSearch(0, 0, noDeadline)
defer idb.putIndexSearch(is)
ii := getIndexItems()
defer putIndexItems(ii)
var tsidBuf TSID
for i := range rows {
tsid := &rows[i].TSID
err := idb.getFromMetricIDCache(&tsidBuf, tsid.MetricID)
if err == nil && *tsid == tsidBuf {
continue
}
if err != io.EOF {
logger.Panicf("FATAL: unexpected error when obtaining TSID for the given MetricID from cache: %s", err)
}
if !is.getTSIDByMetricID(&tsidBuf, tsid.MetricID) || *tsid != tsidBuf {
logger.Infof("Creating missing MetricID->TSID entry for MetricID=%d", tsid.MetricID)
ii.B = marshalCommonPrefix(ii.B, nsPrefixMetricIDToMetricName, tsid.AccountID, tsid.ProjectID)
ii.B = encoding.MarshalUint64(ii.B, tsid.MetricID)
ii.B = tsid.Marshal(ii.B)
ii.Next()
}
idb.putToMetricIDCache(tsid.MetricID, tsid)
}
if len(ii.Items) > 0 {
is.db.tb.AddItems(ii.Items)
}
}
var storageAddRowsLogger = logger.WithThrottler("storageAddRows", 5*time.Second) var storageAddRowsLogger = logger.WithThrottler("storageAddRows", 5*time.Second)
// SetLogNewSeries updates new series logging. // SetLogNewSeries updates new series logging.