diff --git a/internal/container/container.go b/internal/container/container.go index 8a74bf4a..8ee45ec8 100644 --- a/internal/container/container.go +++ b/internal/container/container.go @@ -49,6 +49,9 @@ func BuildContainer() (*dig.Container, error) { func(limiter *ratelimit.AccessKeyRPM) gateway.AccessKeyRPMLimiter { return limiter }, + func(manager *state.Manager) requestlog.RetentionPolicyProvider { + return retentionSnapshotProvider{manager: manager} + }, requestlog.NewService, func(service *requestlog.Service) telemetry.RequestLogSink { return service diff --git a/internal/container/container_test.go b/internal/container/container_test.go index ada565de..c3a96d20 100644 --- a/internal/container/container_test.go +++ b/internal/container/container_test.go @@ -57,6 +57,45 @@ func TestBuildContainerDoesNotInitializeUnusedRuntimeStore(t *testing.T) { } } +func TestBuildContainerWiresRequestLogRetentionSnapshotProvider(t *testing.T) { + t.Setenv("AUTH_KEY", "test-auth-key") + t.Setenv("DATA_DIR", t.TempDir()) + t.Setenv("DATABASE_DSN", ":memory:") + t.Setenv("ENCRYPTION_KEY", "test-master-key-long") + + dependencyContainer, err := BuildContainer() + if err != nil { + t.Fatalf("BuildContainer() error = %v", err) + } + err = dependencyContainer.Invoke(func( + _ *requestlog.Service, + policy requestlog.RetentionPolicyProvider, + manager *state.Manager, + db *gorm.DB, + ) { + t.Cleanup(func() { + sqlDB, dbErr := db.DB() + if dbErr == nil { + _ = sqlDB.Close() + } + }) + if got := policy.RequestLogRetentionDays(); got != 7 { + t.Fatalf("uninitialized retention days = %d, want 7", got) + } + if _, publishErr := manager.Publish(state.CompileInput{ + SystemSettings: config.Settings{state.SettingRequestLogRetentionDays: 30}, + }); publishErr != nil { + t.Fatalf("Publish() error = %v", publishErr) + } + if got := policy.RequestLogRetentionDays(); got != 30 { + t.Fatalf("published retention days = %d, want 30", got) + } + }) + if err != nil { + t.Fatalf("resolve request-log retention graph: %v", err) + } +} + func TestBuildContainerResolvesAllDialects(t *testing.T) { t.Setenv("AUTH_KEY", "test-auth-key") t.Setenv("DATA_DIR", t.TempDir()) diff --git a/internal/container/runtime_settings.go b/internal/container/runtime_settings.go new file mode 100644 index 00000000..3c5d808a --- /dev/null +++ b/internal/container/runtime_settings.go @@ -0,0 +1,15 @@ +package container + +import "gpt-load/internal/state" + +type retentionSnapshotProvider struct { + manager *state.Manager +} + +func (provider retentionSnapshotProvider) RequestLogRetentionDays() int { + snapshot := provider.manager.Current() + if snapshot == nil { + return state.DefaultRuntimeSettings().RequestLogRetentionDays + } + return snapshot.Settings.RequestLogRetentionDays +} diff --git a/internal/control/health.go b/internal/control/health.go index 19ad13b8..9978f7f2 100644 --- a/internal/control/health.go +++ b/internal/control/health.go @@ -54,21 +54,20 @@ type healthProblemKeyResponse struct { } type requestLogHealthResponse struct { - EnqueuedTotal uint64 `json:"enqueued_total"` - PersistedTotal uint64 `json:"persisted_total"` - DroppedNotRunningTotal uint64 `json:"dropped_not_running_total"` - DroppedQueueFullTotal uint64 `json:"dropped_queue_full_total"` - DroppedStoppingTotal uint64 `json:"dropped_stopping_total"` - DroppedPersistFailedTotal uint64 `json:"dropped_persist_failed_total"` - DroppedShutdownTotal uint64 `json:"dropped_shutdown_total"` - DroppedTotal uint64 `json:"dropped_total"` - WriteFailureTotal uint64 `json:"write_failure_total"` - RetentionInvalidSettingTotal uint64 `json:"retention_invalid_setting_total"` - RetentionDeleteFailureTotal uint64 `json:"retention_delete_failure_total"` - QueueDepth int `json:"queue_depth"` - QueueCapacity int `json:"queue_capacity"` - LastWriteFailureAt *time.Time `json:"last_write_failure_at"` - LastRetentionFailureAt *time.Time `json:"last_retention_failure_at"` + EnqueuedTotal uint64 `json:"enqueued_total"` + PersistedTotal uint64 `json:"persisted_total"` + DroppedNotRunningTotal uint64 `json:"dropped_not_running_total"` + DroppedQueueFullTotal uint64 `json:"dropped_queue_full_total"` + DroppedStoppingTotal uint64 `json:"dropped_stopping_total"` + DroppedPersistFailedTotal uint64 `json:"dropped_persist_failed_total"` + DroppedShutdownTotal uint64 `json:"dropped_shutdown_total"` + DroppedTotal uint64 `json:"dropped_total"` + WriteFailureTotal uint64 `json:"write_failure_total"` + RetentionDeleteFailureTotal uint64 `json:"retention_delete_failure_total"` + QueueDepth int `json:"queue_depth"` + QueueCapacity int `json:"queue_capacity"` + LastWriteFailureAt *time.Time `json:"last_write_failure_at"` + LastRetentionFailureAt *time.Time `json:"last_retention_failure_at"` } type runtimeHealthResponse struct { @@ -217,21 +216,20 @@ func (service *Service) RuntimeHealth() (runtimeHealthResponse, error) { func mapRequestLogHealth(stats requestlog.Stats) requestLogHealthResponse { return requestLogHealthResponse{ - EnqueuedTotal: stats.EnqueuedTotal, - PersistedTotal: stats.PersistedTotal, - DroppedNotRunningTotal: stats.DroppedNotRunningTotal, - DroppedQueueFullTotal: stats.DroppedQueueFullTotal, - DroppedStoppingTotal: stats.DroppedStoppingTotal, - DroppedPersistFailedTotal: stats.DroppedPersistFailedTotal, - DroppedShutdownTotal: stats.DroppedShutdownTotal, - DroppedTotal: stats.DroppedTotal, - WriteFailureTotal: stats.WriteFailureTotal, - RetentionInvalidSettingTotal: stats.RetentionInvalidSettingTotal, - RetentionDeleteFailureTotal: stats.RetentionDeleteFailureTotal, - QueueDepth: stats.QueueDepth, - QueueCapacity: stats.QueueCapacity, - LastWriteFailureAt: optionalUTC(stats.LastWriteFailureAt), - LastRetentionFailureAt: optionalUTC(stats.LastRetentionFailureAt), + EnqueuedTotal: stats.EnqueuedTotal, + PersistedTotal: stats.PersistedTotal, + DroppedNotRunningTotal: stats.DroppedNotRunningTotal, + DroppedQueueFullTotal: stats.DroppedQueueFullTotal, + DroppedStoppingTotal: stats.DroppedStoppingTotal, + DroppedPersistFailedTotal: stats.DroppedPersistFailedTotal, + DroppedShutdownTotal: stats.DroppedShutdownTotal, + DroppedTotal: stats.DroppedTotal, + WriteFailureTotal: stats.WriteFailureTotal, + RetentionDeleteFailureTotal: stats.RetentionDeleteFailureTotal, + QueueDepth: stats.QueueDepth, + QueueCapacity: stats.QueueCapacity, + LastWriteFailureAt: optionalUTC(stats.LastWriteFailureAt), + LastRetentionFailureAt: optionalUTC(stats.LastRetentionFailureAt), } } diff --git a/internal/control/health_test.go b/internal/control/health_test.go index fa395df3..26ac3ef7 100644 --- a/internal/control/health_test.go +++ b/internal/control/health_test.go @@ -230,7 +230,7 @@ func TestRuntimeHealthJSONOmitsScoresCredentialsAndZeroTimes(t *testing.T) { for _, forbidden := range []string{ "cipher-must-not-appear", "encrypted", "hash", "header_rules", "percentage", "success_rate", "score", "average_latency", - "0001-01-01", + "retention_invalid_setting_total", "0001-01-01", } { if strings.Contains(lower, forbidden) { t.Fatalf("health JSON exposes %q: %s", forbidden, body) diff --git a/internal/requestlog/query_test.go b/internal/requestlog/query_test.go index 70f9864f..0fb78152 100644 --- a/internal/requestlog/query_test.go +++ b/internal/requestlog/query_test.go @@ -9,7 +9,6 @@ import ( "gorm.io/gorm" - "gpt-load/internal/platform/redact" "gpt-load/internal/protocol" "gpt-load/internal/storage" "gpt-load/internal/storage/models" @@ -18,7 +17,7 @@ import ( func TestServiceListUsesStableKeysetCursor(t *testing.T) { db := openRequestLogQueryDB(t) - service := NewService(db, redact.New()) + service := newRequestLogTestService(db) completedAt := time.Date(2026, time.July, 24, 12, 0, 0, 123, time.UTC) older := completedAt.Add(-time.Nanosecond) for _, row := range []models.RequestLog{ @@ -69,7 +68,7 @@ func TestServiceListUsesStableKeysetCursor(t *testing.T) { func TestServiceListAppliesAllFiltersAndGroupJSON(t *testing.T) { db := openRequestLogQueryDB(t) - service := NewService(db, redact.New()) + service := newRequestLogTestService(db) from := time.Date(2026, time.July, 24, 11, 0, 0, 0, time.UTC) to := from.Add(time.Hour) targetID := "00000000-0000-4000-8000-000000000201" @@ -252,7 +251,7 @@ func TestServiceListBatchLoadsCurrentAccessKeyNames(t *testing.T) { t.Fatalf("register query callback: %v", err) } - page, err := NewService(db, redact.New()).List(context.Background(), ListQuery{Limit: 50}) + page, err := newRequestLogTestService(db).List(context.Background(), ListQuery{Limit: 50}) if err != nil { t.Fatalf("List() error = %v", err) } @@ -294,7 +293,7 @@ func TestServiceListNormalizesNullAttemptsToEmptyArray(t *testing.T) { row.Attempts = nil createRequestLogQueryRow(t, db, row) - page, err := NewService(db, redact.New()).List(context.Background(), ListQuery{Limit: 50}) + page, err := newRequestLogTestService(db).List(context.Background(), ListQuery{Limit: 50}) if err != nil { t.Fatalf("List() error = %v", err) } diff --git a/internal/requestlog/retention.go b/internal/requestlog/retention.go index 39fe4f21..558f90af 100644 --- a/internal/requestlog/retention.go +++ b/internal/requestlog/retention.go @@ -2,20 +2,16 @@ package requestlog import ( "context" - "strconv" "time" "gpt-load/internal/storage/models" ) -const RetentionSettingKey = "request_log_retention_days" +type RetentionPolicyProvider interface { + RequestLogRetentionDays() int +} -const ( - defaultRetentionDays = int64(7) - minRetentionDays = int64(1) - maxRetentionDays = int64(365) - retentionBatchSize = 1000 -) +const retentionBatchSize = 1000 // Sweep removes request logs strictly older than the configured retention // boundary. Configuration and database failures are intentionally isolated @@ -28,18 +24,7 @@ func (service *Service) Sweep(ctx context.Context, now time.Time) { return } - days, valid, err := service.loadRetentionDays(ctx) - if err != nil { - if ctx.Err() == nil { - service.recordRetentionDeleteFailure(now) - } - return - } - if !valid { - service.recordRetentionInvalidSetting(now) - return - } - + days := service.retentionPolicy.RequestLogRetentionDays() cutoff := now.UTC().Add(-time.Duration(days) * 24 * time.Hour) for { if ctx.Err() != nil { @@ -79,34 +64,6 @@ func (service *Service) Sweep(ctx context.Context, now time.Time) { } } -func (service *Service) loadRetentionDays(ctx context.Context) (int64, bool, error) { - var setting models.SystemSetting - result := service.db.WithContext(ctx). - Where("key = ?", RetentionSettingKey). - Find(&setting) - if result.Error != nil { - return 0, false, result.Error - } - if result.RowsAffected == 0 { - return defaultRetentionDays, true, nil - } - - days, err := strconv.ParseInt(setting.Value, 10, 64) - if err != nil || - strconv.FormatInt(days, 10) != setting.Value || - days < minRetentionDays || - days > maxRetentionDays { - return 0, false, nil - } - return days, true, nil -} - -func (service *Service) recordRetentionInvalidSetting(now time.Time) { - service.retentionInvalidTotal.Add(1) - service.recordRetentionFailureAt(now) - service.warn("retention_invalid_setting", 0) -} - func (service *Service) recordRetentionDeleteFailure(now time.Time) { service.retentionDeleteTotal.Add(1) service.recordRetentionFailureAt(now) diff --git a/internal/requestlog/retention_test.go b/internal/requestlog/retention_test.go index f389f43e..0df7b72e 100644 --- a/internal/requestlog/retention_test.go +++ b/internal/requestlog/retention_test.go @@ -1,11 +1,11 @@ package requestlog import ( - "bytes" "context" "errors" "fmt" "io" + "strconv" "strings" "sync/atomic" "testing" @@ -18,74 +18,30 @@ import ( "gpt-load/internal/storage/models" ) -func TestRetentionSettingDefaultsAndStrictValidation(t *testing.T) { +func TestServiceSweepUsesSnapshotRetentionPolicy(t *testing.T) { now := time.Date(2026, time.July, 24, 12, 0, 0, 0, time.UTC) - - t.Run("missing uses seven days without persisting a default", func(t *testing.T) { - db := openRequestLogQueryDB(t) - service := NewService(db, redact.New()) - createRetentionRow(t, db, 1, now.Add(-8*24*time.Hour)) - createRetentionRow(t, db, 2, now.Add(-7*24*time.Hour)) - createRetentionRow(t, db, 3, now.Add(-6*24*time.Hour)) - - service.Sweep(context.Background(), now) - - assertRetentionRows(t, db, 2, 3) - var settingCount int64 - if err := db.Model(&models.SystemSetting{}). - Where("key = ?", RetentionSettingKey). - Count(&settingCount).Error; err != nil { - t.Fatalf("count retention settings: %v", err) - } - if settingCount != 0 { - t.Fatalf("retention setting rows = %d, want 0", settingCount) - } - if stats := service.Stats(); stats.RetentionInvalidSettingTotal != 0 || - stats.RetentionDeleteFailureTotal != 0 || !stats.LastRetentionFailureAt.IsZero() { - t.Fatalf("Stats() = %#v, want no retention failures", stats) - } - }) - - for _, days := range []int64{1, 7, 365} { - t.Run(fmt.Sprintf("accepts_%d", days), func(t *testing.T) { + for _, days := range []int{1, 7, 365} { + t.Run(strconv.Itoa(days), func(t *testing.T) { db := openRequestLogQueryDB(t) - service := NewService(db, redact.New()) - storeRetentionSetting(t, db, fmt.Sprintf("%d", days)) + service := NewService(db, redact.New(), staticRetentionPolicy{days: days}) cutoff := now.Add(-time.Duration(days) * 24 * time.Hour) createRetentionRow(t, db, 1, cutoff.Add(-time.Nanosecond)) createRetentionRow(t, db, 2, cutoff) - createRetentionRow(t, db, 3, cutoff.Add(time.Nanosecond)) - service.Sweep(context.Background(), now) - - assertRetentionRows(t, db, 2, 3) - if stats := service.Stats(); stats.RetentionInvalidSettingTotal != 0 || - stats.RetentionDeleteFailureTotal != 0 { - t.Fatalf("Stats() = %#v, want no retention failures", stats) + var settingQueries atomic.Int64 + const callbackName = "test:no_runtime_setting_query" + if err := db.Callback().Query().After("gorm:query").Register(callbackName, func(tx *gorm.DB) { + if tx.Statement.Table == "system_settings" { + settingQueries.Add(1) + } + }); err != nil { + t.Fatal(err) } - }) - } - - for _, value := range []string{ - "0", "-1", "366", "1.5", "1e2", `"7"`, "true", "null", - " 7", "7 ", "07", "+7", - } { - t.Run("rejects_"+strings.NewReplacer(`"`, "quote", " ", "space", "+", "plus", "-", "minus", ".", "dot").Replace(value), func(t *testing.T) { - db := openRequestLogQueryDB(t) - service := NewService(db, redact.New()) - service.now = func() time.Time { return now } - discardRetentionWarnings(service) - storeRetentionSetting(t, db, value) - createRetentionRow(t, db, 1, now.Add(-400*24*time.Hour)) service.Sweep(context.Background(), now) - - assertRetentionRows(t, db, 1) - stats := service.Stats() - if stats.RetentionInvalidSettingTotal != 1 || - stats.RetentionDeleteFailureTotal != 0 || - !stats.LastRetentionFailureAt.Equal(now) { - t.Fatalf("Stats() = %#v, want one invalid-setting failure at %v", stats, now) + assertRetentionRows(t, db, 2) + if settingQueries.Load() != 0 { + t.Fatalf("system_settings queries = %d", settingQueries.Load()) } }) } @@ -93,7 +49,7 @@ func TestRetentionSettingDefaultsAndStrictValidation(t *testing.T) { func TestServiceSweepDeletesStrictlyOlderRowsInBatches(t *testing.T) { db := openRequestLogQueryDB(t) - service := NewService(db, redact.New()) + service := newRequestLogTestService(db) now := time.Date(2026, time.July, 24, 12, 0, 0, 0, time.UTC) cutoff := now.Add(-7 * 24 * time.Hour) @@ -123,51 +79,18 @@ func TestServiceSweepDeletesStrictlyOlderRowsInBatches(t *testing.T) { if got := deleteBatches.Load(); got != 2 { t.Fatalf("delete batches = %d, want 2", got) } - if stats := service.Stats(); stats.RetentionInvalidSettingTotal != 0 || - stats.RetentionDeleteFailureTotal != 0 || !stats.LastRetentionFailureAt.IsZero() { + if stats := service.Stats(); stats.RetentionDeleteFailureTotal != 0 || + !stats.LastRetentionFailureAt.IsZero() { t.Fatalf("Stats() = %#v, want no retention failures", stats) } } -func TestServiceSweepSkipsInvalidSettingAndTracksFailure(t *testing.T) { - db := openRequestLogQueryDB(t) - service := NewService(db, redact.New()) - now := time.Date(2026, time.July, 24, 12, 0, 0, 0, time.UTC) - service.now = func() time.Time { return now } - logger := logrus.New() - var output bytes.Buffer - logger.SetOutput(&output) - service.logger = logger - - const unsafeSetting = `"retention-secret-value"` - storeRetentionSetting(t, db, unsafeSetting) - createRetentionRow(t, db, 1, now.Add(-400*24*time.Hour)) - - service.Sweep(context.Background(), now) - service.Sweep(context.Background(), now.Add(30*time.Second)) - - assertRetentionRows(t, db, 1) - stats := service.Stats() - if stats.RetentionInvalidSettingTotal != 2 || - stats.RetentionDeleteFailureTotal != 0 || - !stats.LastRetentionFailureAt.Equal(now.Add(30*time.Second)) { - t.Fatalf("Stats() = %#v, want two invalid-setting failures", stats) - } - logged := output.String() - if strings.Contains(logged, unsafeSetting) || strings.Contains(logged, "retention-secret-value") { - t.Fatalf("warning leaked setting content: %q", logged) - } - if got := strings.Count(logged, "level=warning"); got != 1 { - t.Fatalf("warning count = %d, want throttled single warning; output=%q", got, logged) - } -} - func TestServiceSweepStopsOnContextAndDeleteFailure(t *testing.T) { now := time.Date(2026, time.July, 24, 12, 0, 0, 0, time.UTC) t.Run("empty result is successful completion", func(t *testing.T) { db := openRequestLogQueryDB(t) - service := NewService(db, redact.New()) + service := newRequestLogTestService(db) service.Sweep(context.Background(), now) @@ -179,7 +102,7 @@ func TestServiceSweepStopsOnContextAndDeleteFailure(t *testing.T) { t.Run("already canceled context is not a failure", func(t *testing.T) { db := openRequestLogQueryDB(t) - service := NewService(db, redact.New()) + service := newRequestLogTestService(db) createRetentionRow(t, db, 1, now.Add(-8*24*time.Hour)) ctx, cancel := context.WithCancel(context.Background()) cancel() @@ -195,7 +118,7 @@ func TestServiceSweepStopsOnContextAndDeleteFailure(t *testing.T) { t.Run("cancellation between batches stops without failure", func(t *testing.T) { db := openRequestLogQueryDB(t) - service := NewService(db, redact.New()) + service := newRequestLogTestService(db) cutoff := now.Add(-7 * 24 * time.Hour) rows := make([]models.RequestLog, 0, retentionBatchSize+1) for index := 1; index <= retentionBatchSize+1; index++ { @@ -231,33 +154,9 @@ func TestServiceSweepStopsOnContextAndDeleteFailure(t *testing.T) { } }) - t.Run("setting select failure is tracked", func(t *testing.T) { - db := openRequestLogQueryDB(t) - service := NewService(db, redact.New()) - service.now = func() time.Time { return now } - discardRetentionWarnings(service) - const callbackName = "test:retention_setting_query_failure" - if err := db.Callback().Query().Before("gorm:query").Register(callbackName, func(tx *gorm.DB) { - if tx.Statement.Table == "system_settings" { - tx.AddError(errors.New("forced setting query failure")) - } - }); err != nil { - t.Fatalf("register query callback: %v", err) - } - - service.Sweep(context.Background(), now) - - stats := service.Stats() - if stats.RetentionDeleteFailureTotal != 1 || - stats.RetentionInvalidSettingTotal != 0 || - !stats.LastRetentionFailureAt.Equal(now) { - t.Fatalf("Stats() = %#v, want one select failure", stats) - } - }) - t.Run("expired ID selection failure is tracked without deleting", func(t *testing.T) { db := openRequestLogQueryDB(t) - service := NewService(db, redact.New()) + service := newRequestLogTestService(db) service.now = func() time.Time { return now } discardRetentionWarnings(service) createRetentionRow(t, db, 1, now.Add(-8*24*time.Hour)) @@ -281,7 +180,6 @@ func TestServiceSweepStopsOnContextAndDeleteFailure(t *testing.T) { } stats := service.Stats() if stats.RetentionDeleteFailureTotal != 1 || - stats.RetentionInvalidSettingTotal != 0 || !stats.LastRetentionFailureAt.Equal(now) { t.Fatalf("Stats() = %#v, want one expired ID select failure", stats) } @@ -289,7 +187,7 @@ func TestServiceSweepStopsOnContextAndDeleteFailure(t *testing.T) { t.Run("delete failure is tracked and stops the sweep", func(t *testing.T) { db := openRequestLogQueryDB(t) - service := NewService(db, redact.New()) + service := newRequestLogTestService(db) service.now = func() time.Time { return now } discardRetentionWarnings(service) createRetentionRow(t, db, 1, now.Add(-8*24*time.Hour)) @@ -307,23 +205,12 @@ func TestServiceSweepStopsOnContextAndDeleteFailure(t *testing.T) { assertRetentionRows(t, db, 1) stats := service.Stats() if stats.RetentionDeleteFailureTotal != 1 || - stats.RetentionInvalidSettingTotal != 0 || !stats.LastRetentionFailureAt.Equal(now) { t.Fatalf("Stats() = %#v, want one delete failure", stats) } }) } -func storeRetentionSetting(t *testing.T, db *gorm.DB, value string) { - t.Helper() - if err := db.Create(&models.SystemSetting{ - Key: RetentionSettingKey, - Value: value, - }).Error; err != nil { - t.Fatalf("create retention setting %q: %v", value, err) - } -} - func createRetentionRow(t *testing.T, db *gorm.DB, index int, completedAt time.Time) { t.Helper() row := retentionRow(index, completedAt) diff --git a/internal/requestlog/service.go b/internal/requestlog/service.go index a91a4106..0449a641 100644 --- a/internal/requestlog/service.go +++ b/internal/requestlog/service.go @@ -37,14 +37,15 @@ type workerTimerFactory func(time.Duration) workerTimer var _ telemetry.RequestLogSink = (*Service)(nil) type Service struct { - db *gorm.DB - queue chan telemetry.RequestEvent - writer batchWriter - redactor *redact.Redactor - timerFactory workerTimerFactory - logger *logrus.Logger - now func() time.Time - startErr error + db *gorm.DB + queue chan telemetry.RequestEvent + writer batchWriter + redactor *redact.Redactor + retentionPolicy RetentionPolicyProvider + timerFactory workerTimerFactory + logger *logrus.Logger + now func() time.Time + startErr error stateMu sync.Mutex state lifecycleState @@ -62,7 +63,6 @@ type Service struct { droppedPersistFailedTotal atomic.Uint64 droppedShutdownTotal atomic.Uint64 writeFailureTotal atomic.Uint64 - retentionInvalidTotal atomic.Uint64 retentionDeleteTotal atomic.Uint64 statsMu sync.Mutex @@ -73,7 +73,11 @@ type Service struct { lastWarningAt time.Time } -func NewService(db *gorm.DB, redactor *redact.Redactor) *Service { +func NewService( + db *gorm.DB, + redactor *redact.Redactor, + retentionPolicy RetentionPolicyProvider, +) *Service { service := newService( &gormBatchWriter{db: db}, redactor, @@ -82,8 +86,11 @@ func NewService(db *gorm.DB, redactor *redact.Redactor) *Service { }, ) service.db = db + service.retentionPolicy = retentionPolicy if db == nil { service.startErr = fmt.Errorf("request log database is nil") + } else if retentionPolicy == nil { + service.startErr = fmt.Errorf("request log retention policy provider is nil") } return service } @@ -200,20 +207,19 @@ func (service *Service) Stats() Stats { service.statsMu.Unlock() stats := Stats{ - EnqueuedTotal: service.enqueuedTotal.Load(), - PersistedTotal: service.persistedTotal.Load(), - DroppedNotRunningTotal: service.droppedNotRunningTotal.Load(), - DroppedQueueFullTotal: service.droppedQueueFullTotal.Load(), - DroppedStoppingTotal: service.droppedStoppingTotal.Load(), - DroppedPersistFailedTotal: service.droppedPersistFailedTotal.Load(), - DroppedShutdownTotal: service.droppedShutdownTotal.Load(), - WriteFailureTotal: service.writeFailureTotal.Load(), - RetentionInvalidSettingTotal: service.retentionInvalidTotal.Load(), - RetentionDeleteFailureTotal: service.retentionDeleteTotal.Load(), - QueueDepth: len(service.queue), - QueueCapacity: cap(service.queue), - LastWriteFailureAt: lastWriteFailureAt, - LastRetentionFailureAt: lastRetentionFailureAt, + EnqueuedTotal: service.enqueuedTotal.Load(), + PersistedTotal: service.persistedTotal.Load(), + DroppedNotRunningTotal: service.droppedNotRunningTotal.Load(), + DroppedQueueFullTotal: service.droppedQueueFullTotal.Load(), + DroppedStoppingTotal: service.droppedStoppingTotal.Load(), + DroppedPersistFailedTotal: service.droppedPersistFailedTotal.Load(), + DroppedShutdownTotal: service.droppedShutdownTotal.Load(), + WriteFailureTotal: service.writeFailureTotal.Load(), + RetentionDeleteFailureTotal: service.retentionDeleteTotal.Load(), + QueueDepth: len(service.queue), + QueueCapacity: cap(service.queue), + LastWriteFailureAt: lastWriteFailureAt, + LastRetentionFailureAt: lastRetentionFailureAt, } stats.DroppedTotal = stats.DroppedNotRunningTotal + stats.DroppedQueueFullTotal + diff --git a/internal/requestlog/service_test.go b/internal/requestlog/service_test.go index 923478ca..202baf80 100644 --- a/internal/requestlog/service_test.go +++ b/internal/requestlog/service_test.go @@ -82,12 +82,18 @@ func TestServiceEmitLifecycleAndDeepCopy(t *testing.T) { t.Fatalf("Start() after new.Stop error = %v, want ErrNotRestartable", err) } - misconfigured := NewService(nil, redact.New()) - if err := misconfigured.Start(); err == nil { - t.Fatal("Start() with nil database error = nil, want initialization failure") - } - if err := misconfigured.Start(); !errors.Is(err, ErrNotRestartable) { - t.Fatalf("Start() after initialization failure error = %v, want ErrNotRestartable", err) + for name, misconfigured := range map[string]*Service{ + "nil database": NewService(nil, redact.New(), staticRetentionPolicy{days: 7}), + "nil provider": NewService(openRequestLogQueryDB(t), redact.New(), nil), + } { + t.Run(name, func(t *testing.T) { + if err := misconfigured.Start(); err == nil { + t.Fatal("Start() error = nil, want initialization failure") + } + if err := misconfigured.Start(); !errors.Is(err, ErrNotRestartable) { + t.Fatalf("Start() after initialization failure error = %v, want ErrNotRestartable", err) + } + }) } } diff --git a/internal/requestlog/test_helpers_test.go b/internal/requestlog/test_helpers_test.go index 703dc6cb..5d4407b9 100644 --- a/internal/requestlog/test_helpers_test.go +++ b/internal/requestlog/test_helpers_test.go @@ -6,10 +6,23 @@ import ( "testing" "time" + "gorm.io/gorm" + + "gpt-load/internal/platform/redact" "gpt-load/internal/storage/models" "gpt-load/internal/telemetry" ) +type staticRetentionPolicy struct{ days int } + +func (policy staticRetentionPolicy) RequestLogRetentionDays() int { + return policy.days +} + +func newRequestLogTestService(db *gorm.DB) *Service { + return NewService(db, redact.New(), staticRetentionPolicy{days: 7}) +} + type batchWriterFunc func(context.Context, []models.RequestLog) error func (fn batchWriterFunc) WriteBatch(ctx context.Context, rows []models.RequestLog) error { diff --git a/internal/requestlog/types.go b/internal/requestlog/types.go index e56ad6df..6674b173 100644 --- a/internal/requestlog/types.go +++ b/internal/requestlog/types.go @@ -84,19 +84,18 @@ type Page struct { } type Stats struct { - EnqueuedTotal uint64 - PersistedTotal uint64 - DroppedNotRunningTotal uint64 - DroppedQueueFullTotal uint64 - DroppedStoppingTotal uint64 - DroppedPersistFailedTotal uint64 - DroppedShutdownTotal uint64 - DroppedTotal uint64 - WriteFailureTotal uint64 - RetentionInvalidSettingTotal uint64 - RetentionDeleteFailureTotal uint64 - QueueDepth int - QueueCapacity int - LastWriteFailureAt time.Time - LastRetentionFailureAt time.Time + EnqueuedTotal uint64 + PersistedTotal uint64 + DroppedNotRunningTotal uint64 + DroppedQueueFullTotal uint64 + DroppedStoppingTotal uint64 + DroppedPersistFailedTotal uint64 + DroppedShutdownTotal uint64 + DroppedTotal uint64 + WriteFailureTotal uint64 + RetentionDeleteFailureTotal uint64 + QueueDepth int + QueueCapacity int + LastWriteFailureAt time.Time + LastRetentionFailureAt time.Time }