package logmower import ( "context" "path/filepath" "time" "github.com/jtagcat/util" prom "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" "go.uber.org/zap" ) var ( promShipperQueued = promauto.NewGaugeVec(prom.GaugeOpts{ Namespace: PrometheusPrefix, // Subsystem: "shipper", Name: "shipper_record", // "queued", Help: "Log records in queue to be batched and sent to database", }, []string{"filename"}) promShipperDbSent = promauto.NewCounterVec(prom.CounterOpts{ Namespace: PrometheusPrefix, // Subsystem: "shipper", Name: "record", // "sent", Help: "Log records successfully committed to database", }, []string{"filename"}) promShipperBatchSizeResult = promauto.NewHistogram(prom.HistogramOpts{ Namespace: PrometheusPrefix, // Subsystem: "shipper", Name: "bulk_submission_message", // "items_in_batch" Help: "Batch size for database submissions", Buckets: []float64{1, 5, 10, 50, 100, 500, 1000, 5000, 10000}, }) promShipperMongoSentError = promauto.NewCounterVec(prom.CounterOpts{ Namespace: PrometheusPrefix, // Subsystem: "shipper", Name: "insertion_error", // "errors", Help: "Errors while submitting to database", // TODO: }, []string{"filename"}) promShipperSynced = promauto.NewGaugeVec(prom.GaugeOpts{ Namespace: PrometheusPrefix, Subsystem: "shipper", Name: "batches_synced", Help: "All batches available have been committed database (0 or 1)", }, []string{"filename"}) ) const ( MaxBatchItems = 10000 MaxBatchTime = 5 * time.Second ) func (s *submitter) sender(name string, sendQueue <-chan mLog) { baseName := filepath.Base(name) batched := make(chan []mLog) go func() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() go func() { baseName := filepath.Base(name) for { promShipperQueued.WithLabelValues(baseName).Set(float64( len(sendQueue))) timer := time.NewTimer(time.Second) select { case <-ctx.Done(): return case <-timer.C: } } }() util.Batch(MaxBatchItems, MaxBatchTime, sendQueue, batched) // returns when sendQueue is closed }() s.Add(1) defer s.Done() for { promShipperSynced.WithLabelValues(baseName).Set(1) batch, ok := <-batched if !ok { return } promShipperSynced.WithLabelValues(baseName).Set(0) var batchBson []interface{} // mongo does not like typing for _, b := range batch { batchBson = append(batchBson, b.toBson()) } result, err := s.db.InsertMany(mongoTimeoutCtx(context.Background()), batchBson, nil) if err != nil { s.l.Error("submission to database", zap.Error(err)) // TODO: add some selective retry here or something continue } promShipperDbSent.WithLabelValues(baseName).Add(float64( len(result.InsertedIDs))) } }