// Package chunker writes compressed log chunk files to LogChunksDir. // // Each chunk is written as a gzipped file named: // // log....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): } } }