package logmower import ( "context" "errors" "fmt" "io" "os" "path/filepath" "strings" "sync" "time" "github.com/jtagcat/util" prom "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/mongo" mongoOpt "go.mongodb.org/mongo-driver/mongo/options" "go.uber.org/zap" "k8s.io/apimachinery/pkg/util/wait" ) var ( promCatchupDone = promauto.NewGaugeVec(prom.GaugeOpts{ Subsystem: "file", Name: "catchupped", Help: "Files where initial backlog has been sent; (total <= watcher_file_count)", }, []string{"filename"}) // TODO: rm filename? promFileErr = promauto.NewCounterVec(prom.CounterOpts{ Subsystem: "file", Name: "errors_count", Help: "Error count for reading files", }, []string{"filename"}) ) type ( submitter struct { l *zap.Logger hostInfo HostInfo db *mongo.Collection sync.WaitGroup } ) const SendQueueLimit = 1024 // TODO: caller may call duplicate shipFile of same name on file replace; sends might not work properly func (s *submitter) shipFile(ctx context.Context, name string, deleteAfterRead bool) { baseName := filepath.Base(name) sendChan := make(chan mLog, SendQueueLimit) defer close(sendChan) go s.sender(name, sendChan) // TODO: better way to kill or wait for mongo sendQueue before retrying (or duplicates?) wait.ManagedExponentialBackoffWithContext(ctx, defaultBackoff(), func() (done bool, _ error) { // err := s.shipFileRoutine(ctx, name, sendChan) if err == nil { return true, nil } promFileErr.WithLabelValues(baseName).Add(1) s.l.Error("shipping file", zap.String("filename", name), zap.Error(err)) return false, nil // nil since we want to loop and keep retrying indefinitely }) } func (s *submitter) shipFileRoutine(ctx context.Context, name string, sendQueue chan<- mLog) error { baseName := filepath.Base(name) // TODO: better way for respecting ?killing sender for retry for { if len(sendQueue) == 0 { break } time.Sleep(time.Second) } // get files with offset offsetResult, err := mongoWithErr(s.db.FindOne(mongoTimeoutCtx(ctx), bson.D{{Key: mongoKeyHostInfoId, Value: s.hostInfo.id}, {Key: mongoKeyFileBasename, Value: baseName}}, &mongoOpt.FindOneOptions{Sort: bson.D{{Key: mongoKeyOffset, Value: -1}}}, // sort descending (get largest) )) if err != nil && !errors.Is(err, mongo.ErrNoDocuments) { return fmt.Errorf("retrieving mongo offset: %w", err) } var log mLog if err := offsetResult.Decode(&log); err != nil && !errors.Is(err, mongo.ErrNoDocuments) { return fmt.Errorf("decoding mongo offset: %w", err) } fi, err := os.Stat(name) if err != nil { return fmt.Errorf("getting original file size: %w", err) } startSize := fi.Size() sctx, cancel := context.WithCancel(ctx) defer cancel() lineChan, errChan, err := util.TailFile(sctx, name, log.Offset, io.SeekStart) if err != nil { return fmt.Errorf("tailing file: %w", err) } var catchUpped bool // cache promCatchupDone.WithLabelValues(baseName).Set(0) for { select { case err := <-errChan: return fmt.Errorf("tailing file: %w", err) case line, ok := <-lineChan: if !ok { return nil } if !catchUpped { catchUpped = line.EndOffset >= startSize if catchUpped { promCatchupDone.WithLabelValues(baseName).Set(1) } } if line.String == "" { continue } var collectTime time.Time var stdErr, format, log string split := strings.SplitN(line.String, " ", 4) if len(split) != 4 { log = line.String promLineParsingErr.WithLabelValues(baseName).Add(1) s.l.Error("parsing line", zap.Error(fmt.Errorf("expected at least 3 spaces in container log")), zap.Int("got", len(split)-1), zap.String("file", name)) } else { stdErr, format, log = split[1], split[2], split[3] collectTime, err = time.Parse(time.RFC3339Nano, split[0]) if err != nil { promLineParsingErr.WithLabelValues(baseName).Add(1) s.l.Error("parsing line time", zap.Error(err), zap.String("file", name)) } } sendQueue <- mLog{ HostInfo: s.hostInfo, File: baseName, Offset: line.EndOffset, ShipTime: time.Now(), CollectTime: collectTime, StdErr: stdErr == "stderr", // or stdout Format: format, Content: log, } } } } func mongoWithErr[t interface{ Err() error }](mongoWrap t) (t, error) { return mongoWrap, mongoWrap.Err() } // func JitterUntilCancelWithContext(pctx context.Context, f func(context.Context, context.CancelFunc), period time.Duration, jitterFactor float64, sliding bool) { // ctx, cancel := context.WithCancel(pctx) // wait.JitterUntil(func() { f(ctx, cancel) }, period, jitterFactor, sliding, ctx.Done()) // }