Documentation
¶
Index ¶
- Constants
- func LogAutoDetectResult(r AutoDetectResult, source string)
- func LogScanDelta(start, end ScanSnapshot)
- func LogStartupResources()
- func ReadMemory() (total, available uint64)
- type ActiveWorkersFn
- type AutoDetectResult
- type MetricSnapshot
- type MetricsHookFn
- type Option
- type Profiler
- type ResizableSemaphore
- type Scaler
- type ScanSnapshot
Constants ¶
const ( // Per-chunk resource budgets (calibrated from benchmarks + safety margin). // Benchmarked: ~40 MB RSS, ~6 FDs per concurrent chunk. // Budgeted with safety margin for heavy templates (code, headless). MemPerChunkBytes = 200 * 1024 * 1024 // 200 MB FDPerChunk = 20 FDReserved = 1024 // FDs reserved for agent itself (NATS, logs, runtime) // CPU oversubscribe ratio. Most chunks on large subnets are lightweight // (naabu port filter finds nothing → chunk done in 1-2s). Starting higher // lets the agent burn through dead IPs fast. The scaler shrinks if a batch // of chunks hits live hosts and causes real CPU/memory pressure. CPUOversubscribeRatio = 4.0 // Hard limits. MinParallelism = 1 MaxParallelism = 64 )
Variables ¶
This section is empty.
Functions ¶
func LogAutoDetectResult ¶
func LogAutoDetectResult(r AutoDetectResult, source string)
LogAutoDetectResult logs the full computation for operator visibility.
func LogScanDelta ¶
func LogScanDelta(start, end ScanSnapshot)
LogScanDelta logs the resource delta between start and end snapshots.
func LogStartupResources ¶
func LogStartupResources()
LogStartupResources logs a one-time snapshot of the machine/container resources the agent can see. Call this at startup before workers begin.
func ReadMemory ¶
func ReadMemory() (total, available uint64)
readMemory returns (total, available) memory in bytes. Uses cgroup memory limits first, falls back to /proc/meminfo.
Types ¶
type ActiveWorkersFn ¶
type ActiveWorkersFn func() int32
ActiveWorkersFn returns the current number of active workers.
func ActiveWorkersFromAtomic ¶
func ActiveWorkersFromAtomic(a *atomic.Int32) ActiveWorkersFn
Ensure atomic.Int32 satisfies the interface expectation for WorkerPool. This helper creates an ActiveWorkersFn from an *atomic.Int32.
type AutoDetectResult ¶
type AutoDetectResult struct {
ChunkParallelism int
CPUWorkers int
MemWorkers int
FDWorkers int
Bottleneck string // which dimension was the tightest
// Detected raw values
EffectiveCPUs int
AvailableMemMB uint64
FDLimit int
FDUsed int
}
AutoDetectResult holds the computed parallelism and per-dimension breakdown.
func ComputeChunkParallelism ¶
func ComputeChunkParallelism(scanParallelism int) AutoDetectResult
ComputeChunkParallelism detects machine resources and computes the optimal chunk parallelism using: min(W_cpu, W_mem, W_fd), clamped to [1, 64].
The returned value is the TOTAL number of concurrent chunks the machine can handle, regardless of how many scans are running. Global backpressure is enforced by the shared ResizableSemaphore in the caller — no per-scan division is applied here.
type MetricSnapshot ¶
type MetricSnapshot struct {
CPUPercent float64
RSSMB uint64
HeapAllocMB uint64
HeapSysMB uint64
FDUsed int
FDLimit int
Goroutines int
ActiveWorkers int32
MemTotalMB uint64
MemAvailMB uint64
}
MetricSnapshot is a point-in-time resource measurement passed to the metrics hook.
type MetricsHookFn ¶
type MetricsHookFn func(snap MetricSnapshot)
MetricsHookFn is called after each profiler sample with the collected metrics.
type Option ¶
type Option func(*Profiler)
Option configures the Profiler.
func WithMetricsHook ¶
func WithMetricsHook(fn MetricsHookFn) Option
WithMetricsHook sets a callback invoked after each sample.
type Profiler ¶
type Profiler struct {
// contains filtered or unexported fields
}
Profiler samples system resources at a fixed interval and logs them as structured slog entries. The output is designed for offline analysis to calibrate per-worker resource budgets (memory, FDs, CPU).
type ResizableSemaphore ¶
type ResizableSemaphore struct {
// contains filtered or unexported fields
}
ResizableSemaphore controls concurrency with a dynamically adjustable limit. Workers call Acquire before starting work and Release when done. Resize changes the concurrency limit without disrupting in-flight work.
Implementation: a buffered channel acts as the token pool. Acquire takes a token, Release returns one. Resize grows or shrinks the channel by adding or draining tokens.
func NewResizableSemaphore ¶
func NewResizableSemaphore(initial, maxSize int) *ResizableSemaphore
NewResizableSemaphore creates a semaphore with the given initial capacity. maxSize is the upper bound for Resize calls.
func (*ResizableSemaphore) Acquire ¶
func (s *ResizableSemaphore) Acquire(ctx context.Context) error
Acquire blocks until a token is available or ctx is cancelled. Returns nil on success, ctx.Err() on cancellation.
func (*ResizableSemaphore) Available ¶
func (s *ResizableSemaphore) Available() int
Available returns the number of tokens currently in the pool (not held by workers).
func (*ResizableSemaphore) InUse ¶
func (s *ResizableSemaphore) InUse() int
InUse returns the approximate number of tokens held by workers.
func (*ResizableSemaphore) Release ¶
func (s *ResizableSemaphore) Release()
Release returns a token to the pool. Must be called once per successful Acquire.
func (*ResizableSemaphore) Resize ¶
func (s *ResizableSemaphore) Resize(newSize int)
Resize changes the concurrency limit. Growing adds tokens immediately (waking blocked Acquire calls). Shrinking removes tokens from the pool; if not enough free tokens, in-flight workers will naturally drain as they Release — their tokens get dropped instead of returned.
func (*ResizableSemaphore) Size ¶
func (s *ResizableSemaphore) Size() int
Size returns the current concurrency limit.
type Scaler ¶
type Scaler struct {
// contains filtered or unexported fields
}
Scaler resizes a ResizableSemaphore based on resource pressure and chunk duration trends.
func NewScaler ¶
func NewScaler(sem *ResizableSemaphore) *Scaler
NewScaler creates a Scaler that adjusts sem based on resource pressure.
func (*Scaler) RecordChunkDuration ¶
RecordChunkDuration appends the duration of a completed chunk to the rolling window.
type ScanSnapshot ¶
type ScanSnapshot struct {
ScanID string
ChunkID string
Event string // "start" or "end"
Time time.Time
RSS uint64
HeapAlloc uint64
HeapSys uint64
FDUsed int
Goroutines int
Workers int32
}
ScanSnapshot captures a resource snapshot tied to a scan lifecycle event. Call at scan start and end to compute per-scan deltas.
func TakeScanSnapshot ¶
func TakeScanSnapshot(scanID, chunkID, event string, activeWorkers int32) ScanSnapshot
TakeScanSnapshot captures current resource state for a scan event.
Source Files