github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2025-01-30 15:22:07 +00:00
Code Issues Projects Releases Packages Wiki Activity

lib/promscrape/discovery/kubernetes: reload objects on object parse error

Browse source 
Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1240
This commit is contained in:
Aliaksandr Valialkin 2021-05-18 23:25:42 +03:00
parent e8a6c6927d
commit eb8093ca6b
1 changed files with 10 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
lib/promscrape/discovery/kubernetes/api_watcher.go
View file

@ -548,6 +548,7 @@ func (uw *urlWatcher) watchForUpdates() {
if err != nil {
if !errors.Is(err, io.EOF) {
logger.Errorf("error when reading WatchEvent stream from %q: %s", requestURL, err)
uw.resourceVersion = ""
}
backoffSleep()
continue
@ -567,19 +568,19 @@ func (uw *urlWatcher) readObjectUpdateStream(r io.Reader) error {
case "ADDED", "MODIFIED":
o, err := uw.parseObject(we.Object)
if err != nil {
return err
return fmt.Errorf("cannot parse %s object: %w", we.Type, err)
}
key := o.key()
uw.gw.mu.Lock()
if _, ok := uw.objectsByKey[key]; !ok {
if we.Type == "MODIFIED" {
return fmt.Errorf("cannot update object %q, since it is missing in local cache", key)
// This is expected condition after recovering from the bookmarked resourceVersion.
}
uw.objectsCount.Inc()
uw.objectsAdded.Inc()
} else {
if we.Type == "ADDED" {
return fmt.Errorf("cannot add object %q, since it already exists in local cache", key)
// This is expected condition after recovering from the bookmarked resourceVersion.
}
uw.objectsUpdated.Inc()
}
@ -594,16 +595,17 @@ func (uw *urlWatcher) readObjectUpdateStream(r io.Reader) error {
case "DELETED":
o, err := uw.parseObject(we.Object)
if err != nil {
return err
return fmt.Errorf("cannot parse %s object: %w", we.Type, err)
}
key := o.key()
uw.gw.mu.Lock()
if _, ok := uw.objectsByKey[key]; !ok {
return fmt.Errorf("cannot delete object %q, since it is missing in local cache", key)
// This is expected condition after recovering from the bookmarked resourceVersion.
} else {
uw.objectsCount.Dec()
uw.objectsRemoved.Inc()
delete(uw.objectsByKey, key)
}
uw.objectsCount.Dec()
uw.objectsRemoved.Inc()
delete(uw.objectsByKey, key)
for aw := range uw.aws {
aw.removeScrapeWorks(uw, key)
}