From f33c2ede57caf3a5452f10dea8f3970d403fa2c6 Mon Sep 17 00:00:00 2001 From: 0xJacky Date: Wed, 12 Aug 2026 14:42:12 +0800 Subject: [PATCH] fix(nginx_log): bound index rebuilds to the container budget --- app.example.ini | 6 + internal/cgroup/limits.go | 157 +++++++++++++++ internal/cgroup/limits_test.go | 189 +++++++++++++++++ internal/nginx_log/indexer/parser.go | 137 ++++++++----- .../nginx_log/indexer/resource_limits_test.go | 116 +++++++++++ internal/nginx_log/indexer/types.go | 122 ++++++++--- internal/nginx_log/modern_services.go | 6 + internal/nginx_log/parser/parser.go | 7 +- internal/nginx_log/task_scheduler.go | 86 ++++++++ .../task_scheduler_concurrency_test.go | 190 ++++++++++++++++++ main.go | 13 +- settings/nginx_log.go | 6 + 12 files changed, 957 insertions(+), 78 deletions(-) create mode 100644 internal/cgroup/limits.go create mode 100644 internal/cgroup/limits_test.go create mode 100644 internal/nginx_log/indexer/resource_limits_test.go create mode 100644 internal/nginx_log/task_scheduler_concurrency_test.go diff --git a/app.example.ini b/app.example.ini index 0717c454..f37c8580 100644 --- a/app.example.ini +++ b/app.example.ini @@ -82,6 +82,12 @@ IndexPath = ; increase background CPU usage. Higher values reduce CPU usage at the cost ; of more stale analytics data. Values <= 0 fall back to the default 15 minutes. IncrementalIndexInterval = 15 +; Maximum number of log groups indexed at the same time. +; Each concurrent group buffers a parse batch and an index batch per rotated +; file, so this is the main lever on peak indexing memory. Values <= 0 derive +; the limit from the CPU budget the process is allowed to use, which already +; accounts for container (cgroup) limits. +MaxConcurrentIndexTasks = 0 [node] Name = Local diff --git a/internal/cgroup/limits.go b/internal/cgroup/limits.go new file mode 100644 index 00000000..32e0375a --- /dev/null +++ b/internal/cgroup/limits.go @@ -0,0 +1,157 @@ +// Package cgroup exposes the CPU and memory budget the current process is +// actually allowed to consume. +// +// Go sizes runtime.NumCPU/GOMAXPROCS from the CPU affinity mask, and +// /proc/meminfo reports whatever the kernel exposes. Inside a cgroup-limited +// container - Docker with --cpus, Kubernetes limits, and in particular an LXC +// container on Proxmox - neither reflects the real budget: the affinity mask +// still lists every host CPU while the CPU bandwidth controller throttles the +// container to a fraction of one. Sizing worker pools or memory budgets from +// the host numbers therefore oversubscribes the container by an order of +// magnitude. +// +// The helpers here read the cgroup v2 and v1 controller files directly and +// fall back to "unlimited" whenever the information is unavailable, so callers +// can clamp their own defaults without special-casing the platform. +package cgroup + +import ( + "math" + "os" + "path/filepath" + "runtime" + "strconv" + "strings" + + "github.com/shirou/gopsutil/v4/mem" +) + +// cgroupRoot is the mount point of the cgroup filesystem. It is a variable so +// tests can point the readers at a fixture directory. +var cgroupRoot = "/sys/fs/cgroup" + +// maxReasonableMemoryLimit filters the sentinel values some kernels use to mean +// "no limit" (for example math.MaxInt64 rounded down to the page size). +const maxReasonableMemoryLimit = int64(1) << 60 + +// CPUQuota returns the number of CPUs the cgroup bandwidth controller allows +// this process to use. The second return value is false when no quota is +// configured, when the platform has no cgroup filesystem, or when the values +// cannot be parsed. +func CPUQuota() (float64, bool) { + // cgroup v2: " " or "max ". + if raw, err := os.ReadFile(filepath.Join(cgroupRoot, "cpu.max")); err == nil { + fields := strings.Fields(string(raw)) + if len(fields) >= 2 && fields[0] != "max" { + quota, quotaErr := strconv.ParseInt(fields[0], 10, 64) + period, periodErr := strconv.ParseInt(fields[1], 10, 64) + if quotaErr == nil && periodErr == nil && quota > 0 && period > 0 { + return float64(quota) / float64(period), true + } + } + } + + // cgroup v1: quota and period live in separate files, quota == -1 means no limit. + quota, quotaOK := readInt64(filepath.Join(cgroupRoot, "cpu", "cpu.cfs_quota_us")) + period, periodOK := readInt64(filepath.Join(cgroupRoot, "cpu", "cpu.cfs_period_us")) + if quotaOK && periodOK && quota > 0 && period > 0 { + return float64(quota) / float64(period), true + } + + return 0, false +} + +// MemoryLimit returns the cgroup memory limit in bytes. The second return value +// is false when the cgroup does not cap memory. +func MemoryLimit() (int64, bool) { + candidates := []string{ + filepath.Join(cgroupRoot, "memory.max"), // cgroup v2 + filepath.Join(cgroupRoot, "memory", "memory.limit_in_bytes"), // cgroup v1 + } + + for _, path := range candidates { + raw, err := os.ReadFile(path) + if err != nil { + continue + } + value := strings.TrimSpace(string(raw)) + if value == "" || value == "max" { + continue + } + limit, err := strconv.ParseInt(value, 10, 64) + if err != nil || limit <= 0 || limit >= maxReasonableMemoryLimit { + continue + } + return limit, true + } + + return 0, false +} + +// readInt64 parses a cgroup file holding a single integer. +func readInt64(path string) (int64, bool) { + raw, err := os.ReadFile(path) + if err != nil { + return 0, false + } + value, err := strconv.ParseInt(strings.TrimSpace(string(raw)), 10, 64) + if err != nil { + return 0, false + } + return value, true +} + +// totalMemory is indirected so tests can simulate a host without touching the +// real /proc/meminfo. +var totalMemory = func() (uint64, error) { + stat, err := mem.VirtualMemory() + if err != nil { + return 0, err + } + return stat.Total, nil +} + +// AvailableMemory reports the memory budget this process should size itself +// against: the cgroup limit when one is set, otherwise the total system memory. +// The second return value is false when neither number is available. +func AvailableMemory() (int64, bool) { + limit, hasLimit := MemoryLimit() + + total, err := totalMemory() + if err != nil || total == 0 || total > uint64(maxReasonableMemoryLimit) { + return limit, hasLimit + } + + if hasLimit && limit < int64(total) { + return limit, true + } + return int64(total), true +} + +// AvailableCPUs reports how many CPUs may be used for sizing worker pools. +// +// It is the smaller of GOMAXPROCS and the cgroup CPU quota, and never less +// than 1. Callers should use this instead of runtime.GOMAXPROCS/NumCPU when the +// value decides how many goroutines will compete for the CPU: exceeding the +// cgroup quota does not add throughput, it only multiplies peak memory and +// causes the scheduler to thrash against CFS throttling. +func AvailableCPUs() int { + procs := runtime.GOMAXPROCS(0) + if procs < 1 { + procs = 1 + } + + quota, ok := CPUQuota() + if !ok { + return procs + } + + limited := int(math.Ceil(quota)) + if limited < 1 { + limited = 1 + } + if limited < procs { + return limited + } + return procs +} diff --git a/internal/cgroup/limits_test.go b/internal/cgroup/limits_test.go new file mode 100644 index 00000000..441102bb --- /dev/null +++ b/internal/cgroup/limits_test.go @@ -0,0 +1,189 @@ +package cgroup + +import ( + "os" + "path/filepath" + "runtime" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// useFixtureRoot points the cgroup readers at a temporary directory for the +// duration of a test. +func useFixtureRoot(t *testing.T) string { + t.Helper() + + root := t.TempDir() + previous := cgroupRoot + cgroupRoot = root + t.Cleanup(func() { cgroupRoot = previous }) + + return root +} + +func writeFixture(t *testing.T, path, content string) { + t.Helper() + + require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o755)) + require.NoError(t, os.WriteFile(path, []byte(content), 0o644)) +} + +func TestCPUQuotaCgroupV2(t *testing.T) { + root := useFixtureRoot(t) + writeFixture(t, filepath.Join(root, "cpu.max"), "150000 100000\n") + + quota, ok := CPUQuota() + require.True(t, ok) + assert.InDelta(t, 1.5, quota, 0.0001) +} + +func TestCPUQuotaCgroupV2Unlimited(t *testing.T) { + root := useFixtureRoot(t) + writeFixture(t, filepath.Join(root, "cpu.max"), "max 100000\n") + + _, ok := CPUQuota() + assert.False(t, ok) +} + +func TestCPUQuotaCgroupV1(t *testing.T) { + root := useFixtureRoot(t) + writeFixture(t, filepath.Join(root, "cpu", "cpu.cfs_quota_us"), "200000\n") + writeFixture(t, filepath.Join(root, "cpu", "cpu.cfs_period_us"), "100000\n") + + quota, ok := CPUQuota() + require.True(t, ok) + assert.InDelta(t, 2.0, quota, 0.0001) +} + +func TestCPUQuotaCgroupV1Unlimited(t *testing.T) { + root := useFixtureRoot(t) + // -1 is the kernel's "no bandwidth limit" sentinel. + writeFixture(t, filepath.Join(root, "cpu", "cpu.cfs_quota_us"), "-1\n") + writeFixture(t, filepath.Join(root, "cpu", "cpu.cfs_period_us"), "100000\n") + + _, ok := CPUQuota() + assert.False(t, ok) +} + +func TestCPUQuotaWithoutCgroupFilesystem(t *testing.T) { + useFixtureRoot(t) + + _, ok := CPUQuota() + assert.False(t, ok) +} + +// TestAvailableCPUsClampedByQuota is the core regression guard for issue #1792: +// in an LXC container the affinity mask reports every host CPU, so worker pools +// sized from GOMAXPROCS oversubscribe the container by an order of magnitude. +func TestAvailableCPUsClampedByQuota(t *testing.T) { + root := useFixtureRoot(t) + // One core, the typical Proxmox LXC allocation, on a host with many more. + writeFixture(t, filepath.Join(root, "cpu.max"), "100000 100000\n") + + assert.Equal(t, 1, AvailableCPUs()) +} + +func TestAvailableCPUsRoundsFractionalQuotaUp(t *testing.T) { + root := useFixtureRoot(t) + // 0.5 cores must still allow one worker, never zero. + writeFixture(t, filepath.Join(root, "cpu.max"), "50000 100000\n") + + assert.Equal(t, 1, AvailableCPUs()) +} + +func TestAvailableCPUsFallsBackToGOMAXPROCSWithoutQuota(t *testing.T) { + useFixtureRoot(t) + + assert.Equal(t, runtime.GOMAXPROCS(0), AvailableCPUs()) +} + +func TestAvailableCPUsNeverExceedsGOMAXPROCS(t *testing.T) { + root := useFixtureRoot(t) + // A quota far above the machine's capacity must not inflate the pool size. + writeFixture(t, filepath.Join(root, "cpu.max"), "102400000 100000\n") + + assert.Equal(t, runtime.GOMAXPROCS(0), AvailableCPUs()) +} + +func TestMemoryLimitCgroupV2(t *testing.T) { + root := useFixtureRoot(t) + writeFixture(t, filepath.Join(root, "memory.max"), "536870912\n") + + limit, ok := MemoryLimit() + require.True(t, ok) + assert.Equal(t, int64(536870912), limit) +} + +func TestMemoryLimitCgroupV2Unlimited(t *testing.T) { + root := useFixtureRoot(t) + writeFixture(t, filepath.Join(root, "memory.max"), "max\n") + + _, ok := MemoryLimit() + assert.False(t, ok) +} + +func TestMemoryLimitCgroupV1(t *testing.T) { + root := useFixtureRoot(t) + writeFixture(t, filepath.Join(root, "memory", "memory.limit_in_bytes"), "268435456\n") + + limit, ok := MemoryLimit() + require.True(t, ok) + assert.Equal(t, int64(268435456), limit) +} + +func TestMemoryLimitIgnoresSentinelValue(t *testing.T) { + root := useFixtureRoot(t) + // The classic cgroup v1 "unlimited" sentinel. + writeFixture(t, filepath.Join(root, "memory", "memory.limit_in_bytes"), "9223372036854771712\n") + + _, ok := MemoryLimit() + assert.False(t, ok) +} + +// stubTotalMemory replaces the host memory probe for the duration of a test. +func stubTotalMemory(t *testing.T, total uint64, err error) { + t.Helper() + + previous := totalMemory + totalMemory = func() (uint64, error) { return total, err } + t.Cleanup(func() { totalMemory = previous }) +} + +func TestAvailableMemoryPrefersCgroupLimit(t *testing.T) { + root := useFixtureRoot(t) + writeFixture(t, filepath.Join(root, "memory.max"), "536870912\n") // 512MB container + stubTotalMemory(t, 128<<30, nil) // 128GB host + + available, ok := AvailableMemory() + require.True(t, ok) + assert.Equal(t, int64(536870912), available) +} + +func TestAvailableMemoryFallsBackToHostTotal(t *testing.T) { + useFixtureRoot(t) + stubTotalMemory(t, 2<<30, nil) + + available, ok := AvailableMemory() + require.True(t, ok) + assert.Equal(t, int64(2<<30), available) +} + +func TestAvailableMemoryIgnoresLimitAboveHostTotal(t *testing.T) { + root := useFixtureRoot(t) + writeFixture(t, filepath.Join(root, "memory.max"), "137438953472\n") // 128GB + stubTotalMemory(t, 1<<30, nil) // 1GB host + + available, ok := AvailableMemory() + require.True(t, ok) + assert.Equal(t, int64(1<<30), available) +} + +func TestAvailableMemoryUnknown(t *testing.T) { + useFixtureRoot(t) + stubTotalMemory(t, 0, assert.AnError) + + _, ok := AvailableMemory() + assert.False(t, ok) +} diff --git a/internal/nginx_log/indexer/parser.go b/internal/nginx_log/indexer/parser.go index f7835d22..8f11a9b8 100644 --- a/internal/nginx_log/indexer/parser.go +++ b/internal/nginx_log/indexer/parser.go @@ -5,21 +5,33 @@ import ( "compress/gzip" "context" "io" - "runtime" "strings" "sync" + "sync/atomic" + "github.com/0xJacky/Nginx-UI/internal/cgroup" "github.com/0xJacky/Nginx-UI/internal/geolite" "github.com/0xJacky/Nginx-UI/internal/nginx_log/parser" "github.com/uozi-tech/cosy/logger" ) -// Global parser instances +// logParser is the process-wide parser singleton, used for both batch and +// single-line parsing. +// +// It is an atomic pointer rather than a plain global guarded by sync.Once so +// that ReleaseLogParser can drop it: the parser owns a GeoIP handle and two +// 10,000-entry caches, and after a graceful handover the retired process would +// otherwise keep them reachable - and therefore resident - forever. var ( - logParser *parser.Parser // Use the concrete type for both regular and single-line parsing - parserInitOnce sync.Once + logParser atomic.Pointer[parser.Parser] + parserInitMu sync.Mutex ) +// getLogParser returns the current parser singleton, or nil when none is installed. +func getLogParser() *parser.Parser { + return logParser.Load() +} + // geoIPOverride replaces the GeoLite-backed geo lookup when set. // // The slot defaults to nil, and only internal/demo ever fills it, so a @@ -34,58 +46,83 @@ func SetGeoIPService(service parser.GeoIPService) { geoIPOverride = service } +// maxParserWorkerCount caps the per-file parse fan-out. Parsing is only one +// stage of the pipeline, and every worker keeps a parse buffer alive. +const maxParserWorkerCount = 8 + // InitLogParser initializes the global parser once (singleton). func InitLogParser() { - parserInitOnce.Do(func() { - // Initialize the parser with production-ready configuration - config := parser.DefaultParserConfig() - config.MaxLineLength = 16 * 1024 // 16KB for large log lines - config.BatchSize = 15000 // Maximum batch size for highest frontend throughput + parserInitMu.Lock() + defer parserInitMu.Unlock() - // Derive parser worker count from available CPUs, with sane limits so that - // small machines are not overwhelmed while larger hosts can still use - // parallel parsing effectively. - maxProcs := runtime.GOMAXPROCS(0) - if maxProcs <= 0 { - maxProcs = runtime.NumCPU() - } - workerCount := maxProcs - if workerCount < 4 { - workerCount = 4 - } - if workerCount > 16 { - workerCount = 16 - } - config.WorkerCount = workerCount - // Note: Caching is handled by the CachedUserAgentParser + if logParser.Load() != nil { + return + } - // Initialize user agent parser with caching (10,000 cache size for production) - uaParser := parser.NewCachedUserAgentParser( - parser.NewSimpleUserAgentParser(), - 10000, // Large cache for production workloads - ) + // Initialize the parser with production-ready configuration + config := parser.DefaultParserConfig() + config.MaxLineLength = 16 * 1024 // 16KB for large log lines + config.BatchSize = 15000 // Maximum batch size for highest frontend throughput - // Access logs repeat the same IPs heavily; cache lookups so the - // per-line hot path avoids repeated GeoIP database queries - var geoIPService parser.GeoIPService - if geoIPOverride != nil { - geoIPService = parser.NewCachedGeoIPService(geoIPOverride, 10000) - } else if geoService, err := geolite.GetService(); err != nil { - logger.Warnf("Failed to initialize GeoIP service, geo-enrichment will be disabled: %v", err) - } else { - geoIPService = parser.NewCachedGeoIPService(parser.NewGeoLiteAdapter(geoService), 10000) - } + // Derive parser worker count from the CPUs this process may actually use, + // with sane limits so that small machines are not overwhelmed while larger + // hosts can still use parallel parsing effectively. + // + // cgroup.AvailableCPUs, not GOMAXPROCS: inside an LXC/Docker container the + // affinity mask reports every host CPU while the cgroup bandwidth + // controller throttles the process to a fraction of one, so GOMAXPROCS + // would start up to 16 parse goroutines per file on a container that is + // only allowed a single core. + workerCount := cgroup.AvailableCPUs() + if workerCount < 2 { + workerCount = 2 + } + if workerCount > maxParserWorkerCount { + workerCount = maxParserWorkerCount + } + config.WorkerCount = workerCount + // Note: Caching is handled by the CachedUserAgentParser - // Create the parser with production configuration - logParser = parser.NewParser(config, uaParser, geoIPService) + // Initialize user agent parser with caching (10,000 cache size for production) + uaParser := parser.NewCachedUserAgentParser( + parser.NewSimpleUserAgentParser(), + 10000, // Large cache for production workloads + ) - logger.Info("Nginx log processing optimization system initialized with production configuration") - }) + // Access logs repeat the same IPs heavily; cache lookups so the + // per-line hot path avoids repeated GeoIP database queries + var geoIPService parser.GeoIPService + if geoIPOverride != nil { + geoIPService = parser.NewCachedGeoIPService(geoIPOverride, 10000) + } else if geoService, err := geolite.GetService(); err != nil { + logger.Warnf("Failed to initialize GeoIP service, geo-enrichment will be disabled: %v", err) + } else { + geoIPService = parser.NewCachedGeoIPService(parser.NewGeoLiteAdapter(geoService), 10000) + } + + // Create the parser with production configuration + logParser.Store(parser.NewParser(config, uaParser, geoIPService)) + + logger.Info("Nginx log processing optimization system initialized with production configuration") +} + +// ReleaseLogParser drops the parser singleton and the GeoIP handle and caches +// it owns. +// +// After a graceful handover the retired process stays alive as a connection +// proxy for the new binary, so anything left reachable from a package global +// can never be collected. Releasing the parser lets that memory go back to the +// OS instead of doubling the resident set of the container for the lifetime of +// the process. +func ReleaseLogParser() { + parserInitMu.Lock() + defer parserInitMu.Unlock() + logParser.Store(nil) } // IsLogParserInitialized returns true if the global parser singleton has been created. func IsLogParserInitialized() bool { - return logParser != nil + return getLogParser() != nil } // ParseLogLine parses a raw log line into a structured LogDocument using optimized parsing @@ -94,12 +131,13 @@ func ParseLogLine(line string) (*LogDocument, error) { return nil, nil } - if logParser == nil { + activeParser := getLogParser() + if activeParser == nil { return nil, ErrLogParserNotInitialized } // Use parser for single line processing - entry, err := logParser.ParseLine(line) + entry, err := activeParser.ParseLine(line) if err != nil { return nil, err } @@ -113,7 +151,8 @@ func ParseLogLine(line string) (*LogDocument, error) { // bounded regardless of file size. Returns the number of processed and // failed lines. func ParseLogStreamBatches(ctx context.Context, reader io.Reader, filePath string, fn func(docs []*LogDocument) error) (processed, failed int, err error) { - if logParser == nil { + activeParser := getLogParser() + if activeParser == nil { return 0, 0, ErrLogParserNotInitialized } @@ -130,7 +169,7 @@ func ParseLogStreamBatches(ctx context.Context, reader io.Reader, filePath strin // The main log path is constant for the whole file; compute it once mainLogPath := getMainLogPathFromFile(filePath) - parseResult, err := logParser.StreamParseBatches(ctx, actualReader, func(entries []*parser.AccessLogEntry) error { + parseResult, err := activeParser.StreamParseBatches(ctx, actualReader, func(entries []*parser.AccessLogEntry) error { docs := make([]*LogDocument, 0, len(entries)) for _, entry := range entries { docs = append(docs, convertToLogDocument(entry, filePath, mainLogPath)) diff --git a/internal/nginx_log/indexer/resource_limits_test.go b/internal/nginx_log/indexer/resource_limits_test.go new file mode 100644 index 00000000..1cc354a3 --- /dev/null +++ b/internal/nginx_log/indexer/resource_limits_test.go @@ -0,0 +1,116 @@ +package indexer + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// stubResourceBudget simulates a container that is allowed the given number of +// CPUs and bytes of memory, regardless of what the host actually has. +func stubResourceBudget(t *testing.T, cpus int, memoryBytes int64, memoryKnown bool) { + t.Helper() + + previousCPUs := availableCPUs + previousMemory := availableMemory + availableCPUs = func() int { return cpus } + availableMemory = func() (int64, bool) { return memoryBytes, memoryKnown } + t.Cleanup(func() { + availableCPUs = previousCPUs + availableMemory = previousMemory + }) +} + +// TestDefaultIndexerConfigRespectsContainerCPUBudget is the regression guard for +// issue #1792. A Proxmox LXC container gets one core while the affinity mask +// advertises the whole host, and sizing the pools from the host number is what +// turned the post-upgrade index rebuild into an OOM. +func TestDefaultIndexerConfigRespectsContainerCPUBudget(t *testing.T) { + stubResourceBudget(t, 1, 512*1024*1024, true) + + config := DefaultIndexerConfig() + + assert.Equal(t, 2, config.WorkerCount, "worker count must not exceed the small-container floor") + assert.Equal(t, 1, config.FileGroupConcurrency, "a single-core container must index one file at a time") + assert.Equal(t, 15000, config.BatchSize, "batch size must use the smallest tier on a single core") + assert.Equal(t, 1, config.ShardCount) +} + +func TestDefaultIndexerConfigCapsLargeHosts(t *testing.T) { + stubResourceBudget(t, 64, 256*1024*1024*1024, true) + + config := DefaultIndexerConfig() + + assert.LessOrEqual(t, config.WorkerCount, maxDefaultWorkerCount) + assert.LessOrEqual(t, config.FileGroupConcurrency, maxDefaultFileGroupConcurrency) + assert.LessOrEqual(t, config.MemoryQuota, maxIndexMemoryQuota) +} + +func TestDefaultIndexerConfigScalesBetweenExtremes(t *testing.T) { + stubResourceBudget(t, 4, 8*1024*1024*1024, true) + + config := DefaultIndexerConfig() + + assert.Equal(t, 4, config.WorkerCount) + assert.Equal(t, 2, config.FileGroupConcurrency) + assert.Equal(t, 18000, config.BatchSize) + assert.Equal(t, max(4, config.WorkerCount*2), config.MaxQueueSize) +} + +// TestDefaultMemoryQuotaFollowsContainerLimit checks that the indexer +// backpressure valve is sized against the container, not the host. Before the +// fix the quota was a hardcoded 1GB, which never engaged inside a 512MB LXC. +func TestDefaultMemoryQuotaFollowsContainerLimit(t *testing.T) { + tests := []struct { + name string + available int64 + known bool + expected int64 + }{ + { + name: "512MB container gets a quarter of its budget", + available: 512 * 1024 * 1024, + known: true, + expected: 128 * 1024 * 1024, + }, + { + name: "tiny container is floored, not zeroed", + available: 64 * 1024 * 1024, + known: true, + expected: minIndexMemoryQuota, + }, + { + name: "large host is capped at the historical budget", + available: 128 * 1024 * 1024 * 1024, + known: true, + expected: maxIndexMemoryQuota, + }, + { + name: "unknown budget keeps the historical default", + available: 0, + known: false, + expected: maxIndexMemoryQuota, + }, + } + + for _, testCase := range tests { + t.Run(testCase.name, func(t *testing.T) { + stubResourceBudget(t, 4, testCase.available, testCase.known) + + assert.Equal(t, testCase.expected, DefaultMemoryQuota()) + }) + } +} + +// TestGetConfigNeverExceedsContainerMemoryBudget makes sure an aggressive +// scenario profile cannot re-introduce a quota larger than the container. +func TestGetConfigNeverExceedsContainerMemoryBudget(t *testing.T) { + stubResourceBudget(t, 16, 1024*1024*1024, true) + + config := GetConfig("max_performance") + + require.NotNil(t, config) + assert.LessOrEqual(t, config.MemoryQuota, DefaultMemoryQuota()) + assert.Equal(t, max(4, config.WorkerCount*2), config.MaxQueueSize) +} diff --git a/internal/nginx_log/indexer/types.go b/internal/nginx_log/indexer/types.go index 9d083549..c2cae022 100644 --- a/internal/nginx_log/indexer/types.go +++ b/internal/nginx_log/indexer/types.go @@ -2,9 +2,9 @@ package indexer import ( "context" - "runtime" "time" + "github.com/0xJacky/Nginx-UI/internal/cgroup" "github.com/blevesearch/bleve/v2" "github.com/blevesearch/bleve/v2/mapping" ) @@ -59,58 +59,120 @@ type Config struct { FileGroupConcurrency int `json:"file_group_concurrency"` // Max concurrent files within a log group (0 = use WorkerCount) } -// DefaultIndexerConfig returns default indexer configuration with processor optimization -func DefaultIndexerConfig() *Config { - maxProcs := runtime.GOMAXPROCS(0) +// Absolute ceilings for the derived defaults. +// +// Indexing throughput is bounded by Bleve/Scorch segment building and disk I/O +// long before it is bounded by parse parallelism, so scaling these linearly +// with the CPU count only multiplies peak memory. The caps keep a 64-core host +// from opening dozens of buffered batches at once. +const ( + maxDefaultWorkerCount = 8 + maxDefaultFileGroupConcurrency = 4 - // Dynamically scale batch size based on CPU cores - // Significantly increased batch sizes to maximize frontend indexing throughput + // minIndexMemoryQuota / maxIndexMemoryQuota bound the derived memory quota. + minIndexMemoryQuota = int64(64 * 1024 * 1024) + maxIndexMemoryQuota = int64(1024 * 1024 * 1024) + + // indexMemoryQuotaFraction is the share of the container memory budget the + // indexer may retain for queued and in-flight batches. + indexMemoryQuotaFraction = 4 +) + +// availableCPUs reports the CPU budget used to size the indexer, and +// availableMemory the memory budget. +// +// They deliberately do not use runtime.GOMAXPROCS / total RAM directly: inside +// an LXC or Docker container the affinity mask still lists every host CPU while +// the cgroup bandwidth controller throttles the process to a fraction of one, +// so the host numbers overestimate the real budget by an order of magnitude. +// +// Both are variables so tests can simulate a constrained container. +var ( + availableCPUs = cgroup.AvailableCPUs + availableMemory = cgroup.AvailableMemory +) + +// DefaultIndexerConfig returns default indexer configuration sized from the +// CPU and memory budget this process is actually allowed to use. +func DefaultIndexerConfig() *Config { + cpus := availableCPUs() + + // Dynamically scale batch size based on usable CPU cores baseBatchSize := 15000 - if maxProcs >= 16 { + if cpus >= 16 { baseBatchSize = 25000 // High-core systems (16+ cores) - maximum throughput - } else if maxProcs >= 8 { + } else if cpus >= 8 { baseBatchSize = 20000 // Mid-range systems (8-15 cores) - high throughput - } else if maxProcs >= 4 { + } else if cpus >= 4 { baseBatchSize = 18000 // Standard systems (4-7 cores) - good throughput } // Derive conservative, CPU-aware defaults to avoid oversubscribing small machines. - // Treat GOMAXPROCS as the upper bound for CPU-bound worker concurrency. - workerCount := maxProcs - if workerCount < 2 { - workerCount = 2 - } + workerCount := clampInt(cpus, 2, maxDefaultWorkerCount) + + // Limit file-level concurrency to at most half of the usable CPUs. Every + // concurrent file holds its own parse batch plus a buffered index batch, so + // this factor multiplies peak memory directly. + fileGroupConcurrency := clampInt(cpus/2, 1, maxDefaultFileGroupConcurrency) - // Limit file-level concurrency to at most half of the logical CPUs by default. - fileGroupConcurrency := maxProcs / 2 - if fileGroupConcurrency < 2 { - fileGroupConcurrency = 2 - } shardCount := 1 - if maxProcs >= 8 { + if cpus >= 8 { shardCount = 2 } return &Config{ IndexPath: "./log-index", ShardCount: shardCount, - WorkerCount: workerCount, // One worker per logical CPU by default (min 2) - BatchSize: baseBatchSize, // Dynamically scaled based on CPU cores + WorkerCount: workerCount, // One worker per usable CPU, capped (min 2) + BatchSize: baseBatchSize, // Dynamically scaled based on usable CPU cores FlushInterval: 5 * time.Second, MaxQueueSize: max(4, workerCount*2), EnableCompression: true, - MemoryQuota: 1024 * 1024 * 1024, // 1GB - MaxSegmentSize: 64 * 1024 * 1024, // 64MB + MemoryQuota: DefaultMemoryQuota(), + MaxSegmentSize: 64 * 1024 * 1024, // 64MB OptimizeInterval: 30 * time.Minute, EnableMetrics: true, - FileGroupConcurrency: fileGroupConcurrency, // Default: up to 50% of logical CPUs for file-level parallelism + FileGroupConcurrency: fileGroupConcurrency, // Default: up to 50% of usable CPUs, capped } } +// DefaultMemoryQuota derives the indexer memory quota from the cgroup memory +// limit, falling back to the historical 1GB budget when no limit is visible. +// +// The quota is the backpressure valve for indexing: IndexDocuments blocks until +// the estimated size of a batch fits, so a quota that ignores a small container +// lets every concurrent parse pipeline queue its batch and pushes the process +// straight into the OOM killer. +func DefaultMemoryQuota() int64 { + available, ok := availableMemory() + if !ok || available <= 0 { + return maxIndexMemoryQuota + } + + quota := available / indexMemoryQuotaFraction + if quota < minIndexMemoryQuota { + quota = minIndexMemoryQuota + } + if quota > maxIndexMemoryQuota { + quota = maxIndexMemoryQuota + } + return quota +} + +func clampInt(value, minValue, maxValue int) int { + if value < minValue { + return minValue + } + if value > maxValue { + return maxValue + } + return value +} + // GetConfig returns configuration optimized for specific scenarios func GetConfig(scenario string) *Config { base := DefaultIndexerConfig() - maxProcs := runtime.GOMAXPROCS(0) + maxProcs := availableCPUs() switch scenario { case "high_throughput": @@ -172,6 +234,14 @@ func GetConfig(scenario string) *Config { base.MaxSegmentSize = 128 * 1024 * 1024 // 128MB segments } + // A scenario must never raise the quota above what the container is allowed + // to use: when a cgroup memory limit is visible it wins over the profile. + if available, ok := availableMemory(); ok && available > 0 { + if budget := DefaultMemoryQuota(); base.MemoryQuota > budget { + base.MemoryQuota = budget + } + } + // IndexDocuments waits for completion, so retaining thousands of whole // batches cannot improve worker throughput and only expands peak memory. base.MaxQueueSize = max(4, base.WorkerCount*2) diff --git a/internal/nginx_log/modern_services.go b/internal/nginx_log/modern_services.go index 043b395a..3193c0d5 100644 --- a/internal/nginx_log/modern_services.go +++ b/internal/nginx_log/modern_services.go @@ -611,6 +611,12 @@ func StopServices() { globalSearcher = nil } + // Release the parser singleton along with the GeoIP handle and the two + // 10,000-entry caches it owns. It lives in a package global, so without this + // it stays reachable - and resident - for the whole life of a process that + // has already handed its listeners over to a new binary. + indexer.ReleaseLogParser() + // Reset state globalLogFileManager = nil servicesInitialized = false diff --git a/internal/nginx_log/parser/parser.go b/internal/nginx_log/parser/parser.go index feaa101c..4939df00 100644 --- a/internal/nginx_log/parser/parser.go +++ b/internal/nginx_log/parser/parser.go @@ -5,11 +5,12 @@ import ( "bytes" "context" "io" - "runtime" "strconv" "sync" "time" "unsafe" + + "github.com/0xJacky/Nginx-UI/internal/cgroup" ) // Parser provides high-performance log parsing with zero-copy optimizations @@ -228,7 +229,9 @@ func (p *Parser) parseLinesSingleThreaded(ctx context.Context, lines []string, s func (p *Parser) parseLinesParallel(ctx context.Context, lines []string, startTime time.Time) *ParseResult { numWorkers := p.config.WorkerCount if numWorkers <= 0 { - numWorkers = runtime.NumCPU() + // cgroup.AvailableCPUs rather than runtime.NumCPU: the affinity mask + // reports every host CPU inside a cgroup-limited container. + numWorkers = cgroup.AvailableCPUs() } if numWorkers > len(lines)/10 { diff --git a/internal/nginx_log/task_scheduler.go b/internal/nginx_log/task_scheduler.go index a8a76b85..6c5a64f2 100644 --- a/internal/nginx_log/task_scheduler.go +++ b/internal/nginx_log/task_scheduler.go @@ -7,8 +7,10 @@ import ( "sync/atomic" "time" + "github.com/0xJacky/Nginx-UI/internal/cgroup" "github.com/0xJacky/Nginx-UI/internal/event" "github.com/0xJacky/Nginx-UI/internal/nginx_log/indexer" + "github.com/0xJacky/Nginx-UI/settings" "github.com/uozi-tech/cosy/logger" ) @@ -23,6 +25,41 @@ type TaskScheduler struct { wg sync.WaitGroup taskLocks map[string]*sync.Mutex // Per-log-group locks locksMutex sync.RWMutex // Protects taskLocks map + + // runSlots bounds how many log groups may be indexed at the same time. + // + // The per-log-group locks only stop the same group from running twice; without + // this semaphore a rebuild schedules one goroutine per log group and every one + // of them runs concurrently. Each of those in turn fans out to + // FileGroupConcurrency files, and each file holds a parse batch plus a + // buffered index batch, so peak memory is + // groups x files x batches - unbounded in the number of configured sites. + // That is what makes a post-upgrade full rebuild exhaust RAM and saturate the + // CPU on a small container (issue #1792). + runSlots chan struct{} +} + +// defaultMaxConcurrentIndexTasks caps the auto-derived log group concurrency. +// Indexing is dominated by Bleve segment building and disk I/O, so more +// concurrent groups mostly buys extra resident memory. +const defaultMaxConcurrentIndexTasks = 2 + +// maxConcurrentIndexTasks returns how many log groups may be indexed +// concurrently. The value is derived from the CPU budget the process is +// actually allowed to use, and can be overridden through settings. +func maxConcurrentIndexTasks() int { + if configured := settings.NginxLogSettings.MaxConcurrentIndexTasks; configured > 0 { + return configured + } + + slots := cgroup.AvailableCPUs() / 2 + if slots < 1 { + slots = 1 + } + if slots > defaultMaxConcurrentIndexTasks { + slots = defaultMaxConcurrentIndexTasks + } + return slots } // Global task scheduler instance @@ -80,12 +117,43 @@ func InitTaskScheduler(ctx context.Context) { // NewTaskScheduler creates a new task scheduler func NewTaskScheduler(parentCtx context.Context) *TaskScheduler { ctx, cancel := context.WithCancel(parentCtx) + slots := maxConcurrentIndexTasks() + logger.Debugf("Task scheduler limiting concurrent log group indexing to %d", slots) return &TaskScheduler{ logFileManager: GetLogFileManager(), modernIndexer: GetIndexer(), ctx: ctx, cancel: cancel, taskLocks: make(map[string]*sync.Mutex), + runSlots: make(chan struct{}, slots), + } +} + +// acquireRunSlot blocks until a global indexing slot is free. It returns false +// when the scheduler or the caller's context is cancelled while waiting. +func (ts *TaskScheduler) acquireRunSlot(ctx context.Context) bool { + if ts.runSlots == nil { + return true + } + + select { + case ts.runSlots <- struct{}{}: + return true + case <-ctx.Done(): + return false + case <-ts.ctx.Done(): + return false + } +} + +// releaseRunSlot returns a slot acquired by acquireRunSlot. +func (ts *TaskScheduler) releaseRunSlot() { + if ts.runSlots == nil { + return + } + select { + case <-ts.runSlots: + default: } } @@ -176,6 +244,24 @@ func (ts *TaskScheduler) executeIndexTask(ctx context.Context, logPath string, p default: } + // Wait for a global slot before touching the indexer. Indexing a log group + // fans out over its rotated files and buffers a batch per file, so the number + // of groups running at once has to be bounded or a full rebuild across many + // sites allocates without limit. + if !ts.acquireRunSlot(ctx) { + logger.Debugf("Context cancelled while waiting for an indexing slot: %s", logPath) + return + } + defer ts.releaseRunSlot() + + // The wait can be long; re-check cancellation before doing the work. + select { + case <-ctx.Done(): + logger.Debugf("Context cancelled, skipping task for %s", logPath) + return + default: + } + logger.Debugf("Executing indexing task: %s", logPath) // Get processing manager for global state updates diff --git a/internal/nginx_log/task_scheduler_concurrency_test.go b/internal/nginx_log/task_scheduler_concurrency_test.go new file mode 100644 index 00000000..69ce80ea --- /dev/null +++ b/internal/nginx_log/task_scheduler_concurrency_test.go @@ -0,0 +1,190 @@ +package nginx_log + +import ( + "context" + "runtime" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/0xJacky/Nginx-UI/settings" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func withMaxConcurrentIndexTasks(t *testing.T, value int) { + t.Helper() + + previous := settings.NginxLogSettings.MaxConcurrentIndexTasks + settings.NginxLogSettings.MaxConcurrentIndexTasks = value + t.Cleanup(func() { + settings.NginxLogSettings.MaxConcurrentIndexTasks = previous + }) +} + +func TestMaxConcurrentIndexTasksHonoursSetting(t *testing.T) { + withMaxConcurrentIndexTasks(t, 7) + + assert.Equal(t, 7, maxConcurrentIndexTasks()) +} + +func TestMaxConcurrentIndexTasksAutoDerivedIsBounded(t *testing.T) { + withMaxConcurrentIndexTasks(t, 0) + + slots := maxConcurrentIndexTasks() + + assert.GreaterOrEqual(t, slots, 1, "at least one group must be indexable") + assert.LessOrEqual(t, slots, defaultMaxConcurrentIndexTasks, + "auto-derived concurrency must stay capped regardless of the host CPU count") +} + +// TestRunSlotsBoundConcurrency is the regression guard for issue #1792: a +// post-upgrade rebuild schedules one task per log group, and before the fix all +// of them ran at once, each fanning out over its rotated files. Peak memory then +// scaled with the number of configured sites and the container ran out of RAM. +func TestRunSlotsBoundConcurrency(t *testing.T) { + withMaxConcurrentIndexTasks(t, 2) + + scheduler := &TaskScheduler{ + taskLocks: make(map[string]*sync.Mutex), + runSlots: make(chan struct{}, maxConcurrentIndexTasks()), + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + scheduler.ctx = ctx + scheduler.cancel = cancel + + const groups = 32 + + var inFlight int32 + var peak int32 + var waitGroup sync.WaitGroup + + release := make(chan struct{}) + + for i := 0; i < groups; i++ { + waitGroup.Add(1) + go func() { + defer waitGroup.Done() + + if !scheduler.acquireRunSlot(ctx) { + return + } + defer scheduler.releaseRunSlot() + + current := atomic.AddInt32(&inFlight, 1) + for { + observed := atomic.LoadInt32(&peak) + if current <= observed || atomic.CompareAndSwapInt32(&peak, observed, current) { + break + } + } + + <-release + atomic.AddInt32(&inFlight, -1) + }() + } + + // Give every goroutine a chance to pile up on the semaphore. + require.Eventually(t, func() bool { + return atomic.LoadInt32(&inFlight) == 2 + }, 2*time.Second, 5*time.Millisecond, "the first two tasks should acquire slots immediately") + + assert.LessOrEqual(t, atomic.LoadInt32(&peak), int32(2), + "no more than the configured number of log groups may index concurrently") + + close(release) + waitGroup.Wait() + + assert.Equal(t, int32(2), atomic.LoadInt32(&peak), + "all %d groups must have run through exactly 2 slots", groups) + assert.Equal(t, int32(0), atomic.LoadInt32(&inFlight)) + assert.Len(t, scheduler.runSlots, 0, "every acquired slot must be released") +} + +// TestAcquireRunSlotUnblocksOnCancellation makes sure queued tasks cannot pin +// goroutines for the life of the process when the scheduler shuts down. +func TestAcquireRunSlotUnblocksOnCancellation(t *testing.T) { + withMaxConcurrentIndexTasks(t, 1) + + scheduler := &TaskScheduler{ + taskLocks: make(map[string]*sync.Mutex), + runSlots: make(chan struct{}, 1), + } + ctx, cancel := context.WithCancel(context.Background()) + scheduler.ctx = ctx + scheduler.cancel = cancel + + require.True(t, scheduler.acquireRunSlot(ctx), "the first task takes the only slot") + + // Queue many waiters so the goroutine signal dominates unrelated background + // activity in this package's other tests. + const waiters = 50 + + baseline := runtime.NumGoroutine() + + blocked := make(chan bool, waiters) + for i := 0; i < waiters; i++ { + go func() { + blocked <- scheduler.acquireRunSlot(ctx) + }() + } + + require.Eventually(t, func() bool { + return runtime.NumGoroutine() >= baseline+waiters + }, 2*time.Second, 10*time.Millisecond, "every waiter should be parked on the semaphore") + + select { + case <-blocked: + t.Fatal("no waiter may proceed while the only slot is taken") + case <-time.After(100 * time.Millisecond): + } + + cancel() + + for i := 0; i < waiters; i++ { + select { + case acquired := <-blocked: + assert.False(t, acquired, "a cancelled waiter must not report a slot") + case <-time.After(2 * time.Second): + t.Fatal("cancelling the scheduler must release blocked waiters") + } + } + + require.Eventually(t, func() bool { + return runtime.NumGoroutine() < baseline+waiters/2 + }, 2*time.Second, 10*time.Millisecond, "blocked waiters must not leak goroutines") +} + +// TestRunSlotsSurviveRepeatedScheduling checks that repeatedly acquiring and +// releasing slots neither leaks capacity nor accumulates goroutines, which is +// what a long-lived instance does across many incremental indexing cycles. +func TestRunSlotsSurviveRepeatedScheduling(t *testing.T) { + withMaxConcurrentIndexTasks(t, 2) + + scheduler := &TaskScheduler{ + taskLocks: make(map[string]*sync.Mutex), + runSlots: make(chan struct{}, maxConcurrentIndexTasks()), + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + scheduler.ctx = ctx + scheduler.cancel = cancel + + // Settle the runtime before sampling, so background goroutines started by + // earlier tests do not count against the baseline. + runtime.GC() + before := runtime.NumGoroutine() + + for cycle := 0; cycle < 200; cycle++ { + require.True(t, scheduler.acquireRunSlot(ctx)) + scheduler.releaseRunSlot() + } + + assert.Len(t, scheduler.runSlots, 0, "slots must be fully returned after each cycle") + + require.Eventually(t, func() bool { + return runtime.NumGoroutine() <= before+2 + }, 2*time.Second, 10*time.Millisecond, "repeated scheduling must not grow the goroutine count") +} diff --git a/main.go b/main.go index c602b36a..d23b7669 100644 --- a/main.go +++ b/main.go @@ -7,6 +7,7 @@ import ( "net" "os" "os/signal" + "runtime/debug" "syscall" "github.com/0xJacky/Nginx-UI/internal/cert" @@ -172,7 +173,17 @@ func main() { programCtx, cancel := context.WithCancel(mainCtx) // Store the cancel function so the Shutdown callback can use it. programCancel = cancel - return Program(programCtx, confPath)(l) + err := Program(programCtx, confPath)(l) + + // After a graceful handover this process does not exit: risefront + // keeps it alive as a connection proxy in front of the newly spawned + // binary. Its heap is dead but not yet returned to the OS, and inside + // a memory-limited container the retired resident set is charged + // against the same limit as the new process. Hand it back eagerly + // instead of waiting for the background scavenger. + debug.FreeOSMemory() + + return err }, Shutdown: func() { // This is called by risefront.Restart() to shut down the old program. diff --git a/settings/nginx_log.go b/settings/nginx_log.go index fa118197..1f951616 100644 --- a/settings/nginx_log.go +++ b/settings/nginx_log.go @@ -8,6 +8,12 @@ type NginxLog struct { // IncrementalIndexInterval controls how often the incremental indexing job runs, in minutes. // When set to 0 or a negative value, a conservative default will be used. IncrementalIndexInterval int `json:"incremental_index_interval"` + // MaxConcurrentIndexTasks caps how many log groups are indexed at the same + // time. Each concurrent group buffers a parse batch and an index batch per + // rotated file, so this is the main lever on peak indexing memory. + // When set to 0 or a negative value, the value is derived from the CPU + // budget the process is allowed to use. + MaxConcurrentIndexTasks int `json:"max_concurrent_index_tasks"` } var NginxLogSettings = &NginxLog{}