// Package chunker writes compressed log chunk files to LogChunksDir. // // Each chunk is written as: // 1. P4LOG_chunk.. (raw bytes) // 2. P4LOG_chunk...gz (gzipped) // 3. log...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): } } }