chunker.go #2

// Package chunker writes compressed log chunk files to LogChunksDir.
//
// Each chunk is written as:
//  1. P4LOG_chunk.<startOffset>.<endOffset>      (raw bytes)
//  2. P4LOG_chunk.<startOffset>.<endOffset>.gz   (gzipped)
//  3. log.<startOffset>.<endOffset>.gz            (final name, renamed from step 2)
//
// External automation is expected to pick up files matching log.*.gz,
// ship them, and delete them.
package chunker

import (
	"bytes"
	"compress/gzip"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"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, subject to
// MaxLogChunks and MinLogSpace guards.  Returns (false, nil) if the write
// was skipped due to a guard condition, (true, nil) on success.
func (c *Chunker) Write(chunk *tailer.Chunk) (bool, error) {
	if ok, reason := c.checkGuards(); !ok {
		c.log.Warnf("Skipping chunk write: %s", reason)
		return false, nil
	}

	finalName := fmt.Sprintf("log.%d.%d.gz", 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, %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()
}

// checkGuards returns (false, reason) if MaxLogChunks or MinLogSpace is
// violated, otherwise (true, "").
func (c *Chunker) checkGuards() (bool, string) {
	if c.cfg.MaxLogChunks > 0 {
		count, err := countChunkFiles(c.cfg.LogChunksDir)
		if err != nil {
			c.log.Warnf("Could not count chunk files: %v", err)
		} else if count >= c.cfg.MaxLogChunks {
			return false, fmt.Sprintf("MaxLogChunks %d reached (%d files present)",
				c.cfg.MaxLogChunks, count)
		}
	}

	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, ""
}

// countChunkFiles counts files matching log.*.gz in LogChunksDir.
func countChunkFiles(dir string) (int, error) {
	entries, err := os.ReadDir(dir)
	if err != nil {
		return 0, err
	}
	count := 0
	for _, e := range entries {
		if !e.IsDir() {
			m, _ := filepath.Match("log.*.gz", e.Name())
			if m {
				count++
			}
		}
	}
	return count, 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 MaxLogChunks and MinLogSpace guards are
// satisfied, checking every checkInterval. Returns when clear or when done is closed.
func (c *Chunker) WaitUntilGuardsPassed(checkInterval time.Duration, done <-chan struct{}) {
	for {
		ok, reason := c.checkGuards()
		if ok {
			return
		}
		c.log.Warnf("Waiting: %s (checking again in %s)", reason, checkInterval)
		select {
		case <-done:
			return
		case <-time.After(checkInterval):
		}
	}
}