logmower-shipper/cmd/submitter.go

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 (
	promFileInitialSeekSkipped = promauto.NewGaugeVec(prom.GaugeOpts{
		Namespace: PrometheusPrefix,
		// Subsystem: "file",
		Name: "skipped_bytes",
		Help: "Bytes skipped in file after discovering",
	}, []string{"filename"})
	promFileCatchupDone = promauto.NewGaugeVec(prom.GaugeOpts{
		Namespace: PrometheusPrefix,
		Subsystem: "file",
		Name:      "catchupped",
		Help:      "(0 or) 1 if initial backlog has been sent; (total <= watcher_file_count)",
	}, []string{"filename"}) // TODO: rm filename?
	promFileErr = promauto.NewCounterVec(prom.CounterOpts{
		Namespace: PrometheusPrefix,
		Subsystem: "file",
		Name:      "errors_count",
		Help:      "Errors while reading file",
	}, []string{"filename"})
	promFileLineSize = promauto.NewHistogramVec(prom.HistogramOpts{
		Namespace: PrometheusPrefix,
		// Subsystem: "file",
		Name:    "line_size_bytes",
		Help:    "Log line size in bytes",
		Buckets: []float64{80, 160, 320, 640, 1280},
	}, []string{"filename"})
	promRecordPrefixParsingErr = promauto.NewCounterVec(prom.CounterOpts{
		Namespace: PrometheusPrefix,
		Subsystem: "record",
		Name:      "parsing_errors",
		Help:      "Errors while parsing log line prefixes",
	}, []string{"filename"})
	promRecordDroppedTooLarge = promauto.NewCounterVec(prom.CounterOpts{
		Namespace: PrometheusPrefix,
		// Subsystem: "record",
		Name: "dropped_lines", // "dropped",
		Help: "Records dropped due to being too large",
	}, []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 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 offset from database: %w", err)
	}

	var log mLog
	if err := offsetResult.Decode(&log); err != nil && !errors.Is(err, mongo.ErrNoDocuments) {
		return fmt.Errorf("decoding offset from database: %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()

	promFileInitialSeekSkipped.WithLabelValues(baseName).Set(float64(log.Offset))

	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
	promFileCatchupDone.WithLabelValues(baseName).Set(0)

	// TODO: partial line combining
	// TODO: promRecordDroppedTooLarge

	for {
		select {
		case err := <-errChan:
			return fmt.Errorf("tailing file: %w", err)

		case line, ok := <-lineChan:
			if !ok {
				return nil
			}

			promFileLineSize.WithLabelValues(baseName).Observe(float64(len(line.String)))

			if !catchUpped {
				catchUpped = line.EndOffset >= startSize

				if catchUpped {
					promFileCatchupDone.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
				promRecordPrefixParsingErr.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 {
					promRecordPrefixParsingErr.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())
// }