beenull/crowdsec
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

Fix: don't break the loop, retry.

Browse source
This commit is contained in:
Mathieu Lecarme 2022-06-20 18:14:56 +02:00 committed by lperdereau
parent e76d58c77a
commit be95db0c02
1 changed files with 10 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
pkg/acquisition/modules/loki/loki.go
View file

@ -337,6 +337,7 @@ func (l *LokiSource) StreamingAcquisition(out chan types.Event, t *tomb.Tomb) er
if err != nil { if err != nil {
return errors.Wrap(err, "error while getting StreamingAcquisition") return errors.Wrap(err, "error while getting StreamingAcquisition")
} }
ll := l.logger.WithField("websocket url", l.lokiWebsocket)
t.Go(func() error { t.Go(func() error {
for { for {
ctx, cancel := context.WithTimeout(context.TODO(), readyTimeout) ctx, cancel := context.WithTimeout(context.TODO(), readyTimeout)
@ -352,19 +353,22 @@ func (l *LokiSource) StreamingAcquisition(out chan types.Event, t *tomb.Tomb) er
return nil return nil
} }
if res == nil { // no body, it's a network error, not a HTTP error if res == nil { // no body, it's a network error, not a HTTP error
return errors.Wrap(err, "loki StreamingAcquisition error before HTTP stack") ll.WithError(err).Error("loki StreamingAcquisition error before HTTP stack")
break
} }
buf, err2 := ioutil.ReadAll(res.Body) buf, err2 := ioutil.ReadAll(res.Body)
if err2 == nil { if err2 == nil {
return fmt.Errorf("loki websocket (%s) error %v : %s", l.lokiWebsocket, err, string(buf)) ll.WithField("body", string(buf)).WithField("status", res.StatusCode).Error("loki http error")
break
} }
ll.WithError(err2).Error("can't read loki http body")
return err2 break
} }
defer c.Close() defer c.Close()
_, reader, err := c.NextReader() _, reader, err := c.NextReader()
if err != nil { if err != nil {
return errors.Wrap(err, "loki StreamingAcquisition error while reading JSON websocket") ll.WithError(err).Error("loki StreamingAcquisition error while reading JSON websocket")
break
} }
var resp Tail var resp Tail
decoder := json.NewDecoder(reader) decoder := json.NewDecoder(reader)
@ -377,7 +381,7 @@ func (l *LokiSource) StreamingAcquisition(out chan types.Event, t *tomb.Tomb) er
if err == io.EOF { // the websocket is closed if err == io.EOF { // the websocket is closed
break break
} }
return errors.Wrap(err, "loki StreamingAcquisition error while parsing JSON websocket") ll.WithError(err).Error("loki StreamingAcquisition error while parsing JSON websocket")
} }
l.logger.WithField("type", t).WithField("message", resp).Debug("Message receveid") l.logger.WithField("type", t).WithField("message", resp).Debug("Message receveid")
l.readOneTail(resp, out) l.readOneTail(resp, out)