okooooooo/gpt-load
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(requestlog): source retention from runtime snapshot

Browse Source
This commit is contained in:
tbphp committed 2026-07-25 00:42:32 +08:00
1 parent cdaa0b929d
commit fffcbe2263
12 files changed
+188 -266
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/container/container.go
+3
View File
@@ -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
internal/container/container_test.go
+39
View File
@@ -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())
internal/container/runtime_settings.go
+15
View File
@@ -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
}
internal/control/health.go
+28 -30
View File
@@ -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),
}
}
internal/control/health_test.go
+1 -1
View File
@@ -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)
internal/requestlog/query_test.go
+4 -5
View File
@@ -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)
}
internal/requestlog/retention.go
+5 -48
View File
@@ -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)
internal/requestlog/retention_test.go
+24 -137
View File
@@ -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)
internal/requestlog/service.go
+30 -24
View File
@@ -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 +
internal/requestlog/service_test.go
+12 -6
View File
@@ -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)
}
})
}
}
internal/requestlog/test_helpers_test.go
+13
View File
@@ -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 {
internal/requestlog/types.go
+14 -15
View File
@@ -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
}