chunker.go #1

// Package chunker writes compressed log chunk files to LogChunksDir.
//
// Each chunk is written as a gzipped file named:
//
//	log.<YYYY-MM-DD>.<startOffset>.<endOffset>.gz
//
// The date component prevents naming collisions after a P4LOG rotation
// and makes files easy to sort and identify by date.
//
// External automation is expected to pick up files matching log.*.gz,
// ship them, and delete them.
//
// When MaxLogChunks > 0 and the file count reaches the limit, the oldest
// files (by modification time) are deleted to make room, and a warning is
// logged that log data has been lost. This keeps the service running even
// when the downstream consumer is not keeping up.
package chunker

import (
	"bytes"
	"compress/gzip"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sort"
	"syscall"
	"time"

	"workshop.perforce.com/p4lf/internal/config"
	"workshop.perforce.com/p4lf/internal/tailer"
)

// Logger is the minimal interface Chunker needs.
type Logger interface {
	Infof(format string, args ...interface{})
	Warnf(format string, args ...interface{})
	Debugf(format string, args ...interface{})
	Errorf(format string, args ...interface{})
}

// Chunker writes log chunks to disk.
type Chunker struct {
	cfg config.Config
	log Logger
}

// New creates a Chunker.
func New(cfg config.Config, log Logger) (*Chunker, error) {
	if err := os.MkdirAll(cfg.LogChunksDir, 0755); err != nil {
		return nil, fmt.Errorf("creating LogChunksDir %q: %w", cfg.LogChunksDir, err)
	}
	return &Chunker{cfg: cfg, log: log}, nil
}

// Write compresses a Chunk and writes it to LogChunksDir.
// When MaxLogChunks > 0 and the current file count is at or above the limit,
// the oldest files are deleted to bring the count below the limit before
// writing the new chunk. A warning is logged for each deleted file because
// that log data will not be consumed by the downstream system.
// Returns (false, nil) if the write was skipped due to MinLogSpace, (true, nil) on success.
func (c *Chunker) Write(chunk *tailer.Chunk) (bool, error) {
	if err := c.enforceMaxLogChunks(); err != nil {
		return false, err
	}

	if ok, reason := c.checkSpaceGuard(); !ok {
		c.log.Warnf("Skipping chunk write: %s", reason)
		return false, nil
	}

	date := time.Now().Format("2006-01-02")
	finalName := fmt.Sprintf("log.%s.%d.%d.gz", date, chunk.StartOffset, chunk.EndOffset)
	finalPath := filepath.Join(c.cfg.LogChunksDir, finalName)

	gzTmp := filepath.Join(c.cfg.LogChunksDir,
		fmt.Sprintf("P4LOG_chunk.%d.%d.gz.tmp", chunk.StartOffset, chunk.EndOffset))

	// Write gzip directly to a temp file, then rename.
	if err := c.writeGzip(gzTmp, chunk.Data); err != nil {
		os.Remove(gzTmp)
		return false, fmt.Errorf("writing gzip chunk: %w", err)
	}

	if err := os.Rename(gzTmp, finalPath); err != nil {
		os.Remove(gzTmp)
		return false, fmt.Errorf("renaming chunk to %q: %w", finalPath, err)
	}

	c.log.Infof("Wrote chunk %s (%d bytes raw, offsets %d..%d)",
		finalName, len(chunk.Data), chunk.StartOffset, chunk.EndOffset)
	return true, nil
}

func (c *Chunker) writeGzip(path string, data []byte) error {
	f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_EXCL, 0644)
	if err != nil {
		return err
	}
	gz := gzip.NewWriter(f)
	if _, err := io.Copy(gz, bytes.NewReader(data)); err != nil {
		gz.Close()
		f.Close()
		return err
	}
	if err := gz.Close(); err != nil {
		f.Close()
		return err
	}
	return f.Close()
}

// enforceMaxLogChunks deletes the oldest chunk files when the count is at or
// above MaxLogChunks, making room for the next write. A warning is emitted for
// each file deleted because that ingested log data will be lost. If MaxLogChunks
// is 0, no enforcement is done.
func (c *Chunker) enforceMaxLogChunks() error {
	if c.cfg.MaxLogChunks == 0 {
		return nil
	}

	files, err := chunkFilesSortedByAge(c.cfg.LogChunksDir)
	if err != nil {
		c.log.Warnf("Could not list chunk files for MaxLogChunks enforcement: %v", err)
		return nil
	}

	excess := len(files) - c.cfg.MaxLogChunks + 1 // +1 to make room for the new file
	if excess <= 0 {
		return nil
	}

	c.log.Warnf("MaxLogChunks=%d reached (%d files present); deleting %d oldest file(s) — log data will be lost",
		c.cfg.MaxLogChunks, len(files), excess)

	for i := 0; i < excess; i++ {
		path := filepath.Join(c.cfg.LogChunksDir, files[i].name)
		if err := os.Remove(path); err != nil {
			c.log.Errorf("Failed to delete old chunk file %q: %v", path, err)
		} else {
			c.log.Warnf("Deleted oldest chunk file %q to stay within MaxLogChunks limit", files[i].name)
		}
	}
	return nil
}

// checkSpaceGuard returns (false, reason) if MinLogSpace is violated,
// otherwise (true, "").
func (c *Chunker) checkSpaceGuard() (bool, string) {
	sp := c.cfg.MinLogSpace
	if sp.IsPercent && sp.Percent > 0 {
		pct, err := freeSpacePercent(c.cfg.LogChunksDir)
		if err != nil {
			c.log.Warnf("Could not check disk space: %v", err)
		} else if pct < sp.Percent {
			return false, fmt.Sprintf("MinLogSpace %.1f%% not met (%.1f%% free)", sp.Percent, pct)
		}
	} else if !sp.IsPercent && sp.Bytes > 0 {
		free, err := freeSpaceBytes(c.cfg.LogChunksDir)
		if err != nil {
			c.log.Warnf("Could not check disk space: %v", err)
		} else if free < sp.Bytes {
			return false, fmt.Sprintf("MinLogSpace %d bytes not met (%d bytes free)", sp.Bytes, free)
		}
	}
	return true, ""
}

type chunkFileInfo struct {
	name    string
	modTime time.Time
}

// chunkFilesSortedByAge returns chunk files matching log.*.gz sorted oldest-first.
func chunkFilesSortedByAge(dir string) ([]chunkFileInfo, error) {
	entries, err := os.ReadDir(dir)
	if err != nil {
		return nil, err
	}
	var files []chunkFileInfo
	for _, e := range entries {
		if e.IsDir() {
			continue
		}
		matched, _ := filepath.Match("log.*.gz", e.Name())
		if !matched {
			continue
		}
		info, err := e.Info()
		if err != nil {
			continue
		}
		files = append(files, chunkFileInfo{name: e.Name(), modTime: info.ModTime()})
	}
	sort.Slice(files, func(i, j int) bool {
		return files[i].modTime.Before(files[j].modTime)
	})
	return files, nil
}

func freeSpaceBytes(path string) (int64, error) {
	var stat syscall.Statfs_t
	if err := syscall.Statfs(path, &stat); err != nil {
		return 0, err
	}
	return int64(stat.Bavail) * int64(stat.Bsize), nil
}

func freeSpacePercent(path string) (float64, error) {
	var stat syscall.Statfs_t
	if err := syscall.Statfs(path, &stat); err != nil {
		return 0, err
	}
	if stat.Blocks == 0 {
		return 0, fmt.Errorf("filesystem reports 0 total blocks")
	}
	return float64(stat.Bavail) / float64(stat.Blocks) * 100.0, nil
}

// WaitUntilGuardsPassed blocks until the MinLogSpace guard is satisfied,
// checking every checkInterval. MaxLogChunks is handled by deleting old files
// rather than waiting, so it is not checked here. Returns when clear or when
// done is closed.
func (c *Chunker) WaitUntilGuardsPassed(checkInterval time.Duration, done <-chan struct{}) {
	for {
		ok, reason := c.checkSpaceGuard()
		if ok {
			return
		}
		c.log.Warnf("Waiting: %s (checking again in %s)", reason, checkInterval)
		select {
		case <-done:
			return
		case <-time.After(checkInterval):
		}
	}
}