From 849951be1d1a7ee9f9302006ccb187bf5b4e36f3 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Wed, 22 Jul 2026 23:51:18 +0300 Subject: =?UTF-8?q?feat:=20DTail=20fork=20=E2=80=94=20server/client=20feat?= =?UTF-8?q?ure=20development?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Squashed development of the snonux/dtail fork's product code (internal/, cmd/) since diverging from mimecast/dtail. Major areas: - Read/output path: the former "turbo" channel-less path is now the single, default server-side read/output path for cat/grep/tail and MapReduce; the old channel-based path and its config/env toggles were removed. - MapReduce: single aggregate implementation (server + serverless) fed directly by a processor pipeline, with input-exhausted finalization via the shutdown coordinator; high-concurrency and data-race fixes. - Journal source reads (journal:unit.service) via journalctl, Linux-gated behind a journal-v1 capability. - Auth-key fast reconnect: in-memory per-user public-key cache with TTL/max-keys, registered over an authenticated session (AUTHKEY), checked before authorized_keys. - Interactive query reload (--interactive-query) with SESSION START/UPDATE generation boundaries and capability negotiation. - Client-side deadlines: --timeout / --shutdownAfter as context deadlines; follow shutdown handling. - Client logging: diagnostics-only daily log by default, opt-in payload tee via --log-payload. - Numerous correctness fixes (buffer-pool double-recycle races, EOF-sentinel leaks, glob-expansion cap, TOCTOU in CSV parsing) with accompanying unit tests. Co-Authored-By: Claude Opus 4.8 --- internal/profiling/profiler.go | 227 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 227 insertions(+) create mode 100644 internal/profiling/profiler.go (limited to 'internal/profiling/profiler.go') diff --git a/internal/profiling/profiler.go b/internal/profiling/profiler.go new file mode 100644 index 0000000..8af2567 --- /dev/null +++ b/internal/profiling/profiler.go @@ -0,0 +1,227 @@ +package profiling + +import ( + "fmt" + "os" + "path/filepath" + "runtime" + "runtime/pprof" + "time" + + "log" +) + +// Profiler manages CPU and memory profiling for dtail commands +type Profiler struct { + cpuProfile *os.File + memProfile string + profileDir string + commandName string + enabled bool +} + +// Config holds profiling configuration +type Config struct { + // Enable CPU profiling + CPUProfile bool + // Enable memory profiling + MemProfile bool + // Directory to store profiles + ProfileDir string + // Command name for profile naming + CommandName string +} + +// NewProfiler creates a new profiler instance +func NewProfiler(cfg Config) *Profiler { + if !cfg.CPUProfile && !cfg.MemProfile { + return &Profiler{enabled: false} + } + + p := &Profiler{ + profileDir: cfg.ProfileDir, + commandName: cfg.CommandName, + enabled: true, + } + + // Create profile directory if it doesn't exist + if p.profileDir == "" { + p.profileDir = "profiles" + } + if err := os.MkdirAll(p.profileDir, 0755); err != nil { + log.Printf("Failed to create profile directory: %v", err) + p.enabled = false + return p + } + + // Start CPU profiling if enabled + if cfg.CPUProfile { + p.startCPUProfile() + } + + // Set memory profile path if enabled + if cfg.MemProfile { + timestamp := time.Now().Format("20060102_150405") + p.memProfile = filepath.Join(p.profileDir, fmt.Sprintf("%s_mem_%s.prof", p.commandName, timestamp)) + } + + return p +} + +// startCPUProfile starts CPU profiling +func (p *Profiler) startCPUProfile() { + timestamp := time.Now().Format("20060102_150405") + cpuProfilePath := filepath.Join(p.profileDir, fmt.Sprintf("%s_cpu_%s.prof", p.commandName, timestamp)) + + f, err := os.Create(cpuProfilePath) + if err != nil { + log.Printf("Failed to create CPU profile file: %v", err) + return + } + + if err := pprof.StartCPUProfile(f); err != nil { + log.Printf("Failed to start CPU profile: %v", err) + f.Close() + return + } + + p.cpuProfile = f + log.Printf("Started CPU profiling: %s", cpuProfilePath) +} + +// Stop stops all profiling and writes profiles to disk +func (p *Profiler) Stop() { + if !p.enabled { + return + } + + // Stop CPU profiling + if p.cpuProfile != nil { + pprof.StopCPUProfile() + p.cpuProfile.Close() + log.Printf("Stopped CPU profiling") + } + + // Write memory profile + if p.memProfile != "" { + p.writeMemProfile() + } +} + +// writeMemProfile writes memory allocation profile to disk +func (p *Profiler) writeMemProfile() { + f, err := os.Create(p.memProfile) + if err != nil { + log.Printf("Failed to create memory profile file: %v", err) + return + } + defer f.Close() + + // Force GC before capturing memory profile for more accurate results + runtime.GC() + + if err := pprof.WriteHeapProfile(f); err != nil { + log.Printf("Failed to write memory profile: %v", err) + return + } + + log.Printf("Wrote memory profile: %s", p.memProfile) + + // Also write allocation profile for detailed allocation tracking + allocProfilePath := filepath.Join(p.profileDir, + fmt.Sprintf("%s_alloc_%s.prof", p.commandName, time.Now().Format("20060102_150405"))) + + allocFile, err := os.Create(allocProfilePath) + if err != nil { + log.Printf("Failed to create allocation profile file: %v", err) + return + } + defer allocFile.Close() + + // Set allocation profiling rate to capture more samples + runtime.MemProfileRate = 1 + + if err := pprof.Lookup("allocs").WriteTo(allocFile, 0); err != nil { + log.Printf("Failed to write allocation profile: %v", err) + return + } + + log.Printf("Wrote allocation profile: %s", allocProfilePath) +} + +// Snapshot takes a memory snapshot at any point during execution +func (p *Profiler) Snapshot(label string) { + if !p.enabled || p.memProfile == "" { + return + } + + timestamp := time.Now().Format("20060102_150405") + snapshotPath := filepath.Join(p.profileDir, + fmt.Sprintf("%s_snapshot_%s_%s.prof", p.commandName, label, timestamp)) + + f, err := os.Create(snapshotPath) + if err != nil { + log.Printf("Failed to create snapshot file: %v", err) + return + } + defer f.Close() + + runtime.GC() + if err := pprof.WriteHeapProfile(f); err != nil { + log.Printf("Failed to write snapshot: %v", err) + return + } + + log.Printf("Wrote memory snapshot: %s (label: %s)", snapshotPath, label) +} + +// ProfileMetrics captures and returns current runtime metrics +type ProfileMetrics struct { + // Memory statistics + Alloc uint64 // Bytes allocated and still in use + TotalAlloc uint64 // Bytes allocated (even if freed) + Sys uint64 // Bytes obtained from system + NumGC uint32 // Number of completed GC cycles + LastGC time.Time // Time of last GC + PauseTotalNs uint64 // Total GC pause time in nanoseconds + + // Goroutine count + NumGoroutine int + + // CPU count + NumCPU int +} + +// GetMetrics returns current runtime metrics +func GetMetrics() ProfileMetrics { + var m runtime.MemStats + runtime.ReadMemStats(&m) + + return ProfileMetrics{ + Alloc: m.Alloc, + TotalAlloc: m.TotalAlloc, + Sys: m.Sys, + NumGC: m.NumGC, + LastGC: time.Unix(0, int64(m.LastGC)), + PauseTotalNs: m.PauseTotalNs, + NumGoroutine: runtime.NumGoroutine(), + NumCPU: runtime.NumCPU(), + } +} + +// LogMetrics logs current runtime metrics +func (p *Profiler) LogMetrics(label string) { + if !p.enabled { + return + } + + metrics := GetMetrics() + log.Printf("Profile metrics [%s]: alloc=%.2fMB total_alloc=%.2fMB sys=%.2fMB num_gc=%d gc_pause=%.2fms goroutines=%d", + label, + float64(metrics.Alloc)/1024/1024, + float64(metrics.TotalAlloc)/1024/1024, + float64(metrics.Sys)/1024/1024, + metrics.NumGC, + float64(metrics.PauseTotalNs)/1e6, + metrics.NumGoroutine) +} \ No newline at end of file -- cgit v1.2.3