commit
a909aabcd6
|
@ -289,19 +289,34 @@ func main() {
|
||||||
//start go-routines for parsing, buckets pour and ouputs.
|
//start go-routines for parsing, buckets pour and ouputs.
|
||||||
for i := 0; i < nbParser; i++ {
|
for i := 0; i < nbParser; i++ {
|
||||||
parsersTomb.Go(func() error {
|
parsersTomb.Go(func() error {
|
||||||
return runParse(inputLineChan, inputEventChan, *parserCTX, parserNodes)
|
err := runParse(inputLineChan, inputEventChan, *parserCTX, parserNodes)
|
||||||
|
if err != nil {
|
||||||
|
log.Errorf("runParse error : %s", err)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return nil
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
for i := 0; i < nbParser; i++ {
|
for i := 0; i < nbParser; i++ {
|
||||||
bucketsTomb.Go(func() error {
|
bucketsTomb.Go(func() error {
|
||||||
return runPour(inputEventChan, holders, buckets)
|
err := runPour(inputEventChan, holders, buckets)
|
||||||
|
if err != nil {
|
||||||
|
log.Errorf("runPour error : %s", err)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return nil
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
for i := 0; i < nbParser; i++ {
|
for i := 0; i < nbParser; i++ {
|
||||||
outputsTomb.Go(func() error {
|
outputsTomb.Go(func() error {
|
||||||
return runOutput(inputEventChan, outputEventChan, holders, buckets, *postOverflowCTX, postOverflowNodes, outputProfiles, outputRunner)
|
err := runOutput(inputEventChan, outputEventChan, holders, buckets, *postOverflowCTX, postOverflowNodes, outputProfiles, outputRunner)
|
||||||
|
if err != nil {
|
||||||
|
log.Errorf("runPour error : %s", err)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return nil
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -59,9 +59,12 @@ LOOP:
|
||||||
if cConfig.Profiling {
|
if cConfig.Profiling {
|
||||||
bucketStat.AddTime(time.Since(start))
|
bucketStat.AddTime(time.Since(start))
|
||||||
}
|
}
|
||||||
if err := lastProcessedItem.UnmarshalText([]byte(parsed.MarshaledTime)); err != nil {
|
if len(parsed.MarshaledTime) != 0 {
|
||||||
return fmt.Errorf("failed to unmarshal item : %s", err)
|
if err := lastProcessedItem.UnmarshalText([]byte(parsed.MarshaledTime)); err != nil {
|
||||||
|
log.Debugf("failed to unmarshal time from event : %s", err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
log.Infof("Sending signal Bucketify")
|
log.Infof("Sending signal Bucketify")
|
||||||
|
|
Loading…
Reference in a new issue