2023-03-03 19:31:57 +01:00
|
|
|
package metrics
|
|
|
|
|
|
|
|
import (
|
|
|
|
"github.com/pkg/errors"
|
|
|
|
"github.com/sirupsen/logrus"
|
|
|
|
)
|
|
|
|
|
|
|
|
func Run(cfg *Config) error {
|
|
|
|
logrus.Info("starting")
|
|
|
|
defer logrus.Warn("stopping")
|
|
|
|
|
|
|
|
if cfg.Source == nil {
|
|
|
|
return errors.New("no 'source' configured; exiting")
|
|
|
|
}
|
|
|
|
|
|
|
|
src, ok := cfg.Source.(Source)
|
|
|
|
if !ok {
|
|
|
|
return errors.New("invalid 'source'; exiting")
|
|
|
|
}
|
|
|
|
|
2023-03-03 20:22:22 +01:00
|
|
|
events := make(chan map[string]interface{}, 1024)
|
|
|
|
srcJoin, err := src.Start(events)
|
2023-03-03 19:31:57 +01:00
|
|
|
if err != nil {
|
|
|
|
return errors.Wrap(err, "error starting source")
|
|
|
|
}
|
|
|
|
|
2023-03-03 20:22:22 +01:00
|
|
|
go func() {
|
|
|
|
ingester := &UsageIngester{}
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case event := <-events:
|
|
|
|
if err := ingester.Ingest(event); err != nil {
|
|
|
|
logrus.Error(err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
2023-03-03 19:31:57 +01:00
|
|
|
<-srcJoin
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|