diff --git a/.gitignore b/.gitignore index bbe5f099..ee00ae9b 100644 --- a/.gitignore +++ b/.gitignore @@ -28,3 +28,12 @@ logs/ # Dashboard UI internal/dashboard/ui/node_modules/cp.log + +# SQLite data. The provider's database normally lives in $CP_PATH, outside the +# repo — but CP_PATH can be pointed here during development, and the WAL and +# shared-memory sidecars appear beside it. +*.db +*.db-wal +*.db-shm +*.sqlite +*.sqlite3 diff --git a/conf/config.go b/conf/config.go index bed9fb03..43dc59ca 100644 --- a/conf/config.go +++ b/conf/config.go @@ -86,6 +86,16 @@ type ComputeNode struct { HealthCheck HealthCheck `toml:"HealthCheck,omitempty"` RequestLimits RequestLimits `toml:"RequestLimits,omitempty"` Dashboard Dashboard `toml:"Dashboard,omitempty"` + RequestLog RequestLog `toml:"RequestLog,omitempty"` +} + +// RequestLog controls how long the per-request history kept for the +// Transactions view is retained. It is one row per served request, so both +// limits exist: age for how far back it is useful, and a row cap so a burst +// cannot fill the disk before the age limit bites. +type RequestLog struct { + RetentionDays int `toml:"RetentionDays"` // Default: 7 + MaxRows int64 `toml:"MaxRows"` // Default: 200000 } // Dashboard configures the web UI served by `computing-provider dashboard`. diff --git a/docs/configuration.md b/docs/configuration.md index ff1744d4..b1289d7f 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -298,6 +298,27 @@ overloaded GPU. Both can also be changed at runtime, globally or per model, through the REST API — see the endpoint table in the [main README](../README.md#rest-api). +### Request history + +```toml +[RequestLog] +RetentionDays = 7 # how far back the Transactions view reaches (default 7) +MaxRows = 200000 # hard cap on stored rows (default 200000) +``` + +Every served request is stored, so the Transactions view — and its model and +source filters — survive a restart. Before this existed the list lived only in +a 1000-entry in-memory ring, so each restart emptied it, and a node that +restarts often kept almost no history at all. + +Two limits rather than one: `RetentionDays` is how far back the data stays +useful, and `MaxRows` stops a traffic burst filling the disk before the age +limit ever applies. Whichever binds first wins, and pruning runs hourly. + +Writes are batched and happen off the request path, so recording never delays +serving; under a burst large enough to fill the queue, records are dropped and +the count is logged rather than blocking inference. + ### models.json field reference Each key in `models.json` is the marketplace model ID (must match a value in `Models` above). diff --git a/internal/computing/inference_metrics.go b/internal/computing/inference_metrics.go index 0e962612..ece65939 100644 --- a/internal/computing/inference_metrics.go +++ b/internal/computing/inference_metrics.go @@ -57,6 +57,8 @@ type InferenceMetrics struct { // Request history (circular buffer) requestHistory []RequestMetric + // requestSink persists each request, when a store is attached. + requestSink func(RequestMetric) historyMu sync.RWMutex maxHistorySize int } @@ -407,11 +409,25 @@ func (m *InferenceMetrics) GetPrometheusMetrics() string { return sb.String() } +// SetRequestSink installs a persistent store fed alongside the in-memory ring. +// The ring stays: it answers without touching the database, and it is what the +// model detail view reads. +func (m *InferenceMetrics) SetRequestSink(sink func(RequestMetric)) { + m.historyMu.Lock() + defer m.historyMu.Unlock() + m.requestSink = sink +} + // RecordRequest adds a request to the history circular buffer func (m *InferenceMetrics) RecordRequest(req RequestMetric) { m.historyMu.Lock() defer m.historyMu.Unlock() + if m.requestSink != nil { + // Never blocks — the sink queues and returns. + m.requestSink(req) + } + if len(m.requestHistory) >= m.maxHistorySize { // Remove oldest entry (circular buffer) m.requestHistory = m.requestHistory[1:] diff --git a/internal/computing/inference_service.go b/internal/computing/inference_service.go index 78c29034..967969d8 100644 --- a/internal/computing/inference_service.go +++ b/internal/computing/inference_service.go @@ -70,6 +70,7 @@ type InferenceService struct { retryPolicy *RetryPolicy gpuCollector *GPUMetricsCollector metricsHistory *MetricsHistory + requestStore *RequestStore alertMonitor *alertMonitor selfCheck *selfCheckRunner noticeLimiter *noticeLimiter @@ -116,6 +117,17 @@ func NewInferenceService(nodeID, cpPath string) *InferenceService { // Create metrics history for persistence metricsHistory := NewMetricsHistory() + // Per-request history, so the Transactions view survives a restart. + // GetConfig is nil until InitConfig has run, which is not guaranteed for + // callers that construct the service directly; the store's own defaults + // cover that. + var retentionDays int + var maxRows int64 + if cfg := conf.GetConfig(); cfg != nil { + retentionDays, maxRows = cfg.RequestLog.RetentionDays, cfg.RequestLog.MaxRows + } + requestStore := NewRequestStore(retentionDays, maxRows) + s := &InferenceService{ nodeID: nodeID, cpPath: cpPath, @@ -127,6 +139,7 @@ func NewInferenceService(nodeID, cpPath string) *InferenceService { retryPolicy: retryPolicy, gpuCollector: gpuCollector, metricsHistory: metricsHistory, + requestStore: requestStore, } // Set up registry callbacks to update modelMappings for backward compatibility @@ -345,6 +358,15 @@ func (s *InferenceService) Start() error { s.alertMonitor.Start() s.selfCheck.Start() + // Start persistent request history, and feed it from the metrics recorder. + if s.requestStore != nil { + if err := s.requestStore.Start(); err != nil { + logs.GetLogger().Warnf("Failed to start request history store: %v", err) + } else if s.client != nil && s.client.metrics != nil { + s.client.metrics.SetRequestSink(s.requestStore.Record) + } + } + // Start metrics history recorder if s.metricsHistory != nil { if err := s.metricsHistory.Start(func() *InferenceMetricsData { @@ -384,6 +406,9 @@ func (s *InferenceService) Stop() { if s.metricsHistory != nil { s.metricsHistory.Stop() } + if s.requestStore != nil { + s.requestStore.Stop() + } if s.alertMonitor != nil { s.alertMonitor.Stop() } @@ -1271,7 +1296,18 @@ func (s *InferenceService) GetRequestHistory(limit int, modelFilter string) []Re // QueryRequestHistory returns one page of request history along with the total // matching the filters. +// +// Served from the persistent store when there is one, so the list reaches back +// past the last restart. The in-memory ring is the fallback: it is all that +// exists before the store has started, or if the database is unavailable. func (s *InferenceService) QueryRequestHistory(q RequestHistoryQuery) RequestHistoryPage { + if s.requestStore != nil { + if page, err := s.requestStore.Query(q); err == nil { + return page + } else { + logs.GetLogger().Warnf("Request history query failed, falling back to memory: %v", err) + } + } if s.client == nil { return RequestHistoryPage{Requests: []RequestMetric{}, Limit: q.Limit, Offset: q.Offset} } diff --git a/internal/computing/request_history_store.go b/internal/computing/request_history_store.go new file mode 100644 index 00000000..245795f9 --- /dev/null +++ b/internal/computing/request_history_store.go @@ -0,0 +1,338 @@ +package computing + +import ( + "sync" + "sync/atomic" + "time" + + "github.com/filswan/go-mcs-sdk/mcs/api/common/logs" + "github.com/swanchain/computing-provider-v2/internal/db" + "gorm.io/gorm" +) + +// RequestHistoryEntity is one served request, kept so the Transactions view +// survives a restart. +// +// Until this existed the request list lived only in a 1000-entry in-memory +// ring, so every restart emptied it — and this agent restarts often enough +// that the earnings panel routinely reports counter resets. A filter and a +// pager over a list that only reaches back to the last restart are not much +// use. +type RequestHistoryEntity struct { + ID uint `gorm:"primaryKey;autoIncrement"` + RequestID string `gorm:"index"` + Model string `gorm:"index"` + Source string `gorm:"index"` + StartTime time.Time `gorm:"index;not null"` + EndTime time.Time + LatencyMs float64 + TokensIn int + TokensOut int + Streaming bool + Success bool + // ErrorReason can be an upstream body, so it is truncated before storage: + // a row per request means an unbounded field is an unbounded table. + ErrorReason string +} + +func (RequestHistoryEntity) TableName() string { + return "request_history" +} + +const ( + // maxErrorReasonBytes caps what one failure can cost on disk. + maxErrorReasonBytes = 512 + // writeBatchSize is how many rows one insert carries. The database allows a + // single open connection, so writes are batched rather than made per + // request — otherwise a busy node serialises inference behind SQLite. + writeBatchSize = 200 + // flushInterval bounds how stale the stored history can be. The dashboard + // polls every 10s, so a second or two of lag is invisible. + flushInterval = 2 * time.Second + // queueSize is the burst absorbed before records are dropped. Dropping is + // deliberate: recording history must never slow down or block serving. + queueSize = 4096 +) + +// RequestStore persists request history and serves queries over it. +type RequestStore struct { + queue chan RequestMetric + stop chan struct{} + done chan struct{} + mu sync.Mutex + running bool + + retentionDays int + maxRows int64 + + // dropped counts records discarded because the queue was full, so the + // condition is visible rather than silent. + dropped atomic.Int64 +} + +// NewRequestStore builds a store with the given retention. A non-positive +// value falls back to the default. +func NewRequestStore(retentionDays int, maxRows int64) *RequestStore { + if retentionDays <= 0 { + retentionDays = 7 + } + if maxRows <= 0 { + maxRows = 200_000 + } + return &RequestStore{ + queue: make(chan RequestMetric, queueSize), + stop: make(chan struct{}), + done: make(chan struct{}), + retentionDays: retentionDays, + maxRows: maxRows, + } +} + +// Start migrates the table and begins the writer and pruner. +func (s *RequestStore) Start() error { + s.mu.Lock() + if s.running { + s.mu.Unlock() + return nil + } + s.running = true + s.mu.Unlock() + + database := db.NewDbService() + if database == nil { + return nil + } + if err := database.AutoMigrate(&RequestHistoryEntity{}); err != nil { + return err + } + + go s.writeLoop() + go s.pruneLoop() + logs.GetLogger().Info("Request history store started") + return nil +} + +// Stop drains what is queued and shuts the writer down. +func (s *RequestStore) Stop() { + s.mu.Lock() + if !s.running { + s.mu.Unlock() + return + } + s.running = false + s.mu.Unlock() + + close(s.stop) + <-s.done + if dropped := s.dropped.Load(); dropped > 0 { + logs.GetLogger().Warnf("Request history dropped %d record(s) under load", dropped) + } +} + +// Record queues one request. It never blocks: a full queue drops the record +// rather than making the inference path wait on storage. +func (s *RequestStore) Record(req RequestMetric) { + if s == nil { + return + } + if len(req.ErrorReason) > maxErrorReasonBytes { + req.ErrorReason = req.ErrorReason[:maxErrorReasonBytes] + } + select { + case s.queue <- req: + default: + s.dropped.Add(1) + } +} + +func (s *RequestStore) writeLoop() { + defer close(s.done) + ticker := time.NewTicker(flushInterval) + defer ticker.Stop() + + batch := make([]RequestHistoryEntity, 0, writeBatchSize) + flush := func() { + if len(batch) == 0 { + return + } + if database := db.NewDbService(); database != nil { + if err := database.CreateInBatches(batch, writeBatchSize).Error; err != nil { + logs.GetLogger().Warnf("Failed to persist request history: %v", err) + } + } + batch = batch[:0] + } + + for { + select { + case req := <-s.queue: + batch = append(batch, entityFor(req)) + if len(batch) >= writeBatchSize { + flush() + } + case <-ticker.C: + flush() + case <-s.stop: + // Drain whatever is queued so a clean shutdown does not discard + // the requests served just before it. + for { + select { + case req := <-s.queue: + batch = append(batch, entityFor(req)) + if len(batch) >= writeBatchSize { + flush() + } + continue + default: + } + break + } + flush() + return + } + } +} + +func entityFor(req RequestMetric) RequestHistoryEntity { + return RequestHistoryEntity{ + RequestID: req.RequestID, + Model: req.Model, + Source: string(req.Source), + StartTime: req.StartTime, + EndTime: req.EndTime, + LatencyMs: req.LatencyMs, + TokensIn: req.TokensIn, + TokensOut: req.TokensOut, + Streaming: req.Streaming, + Success: req.Success, + ErrorReason: req.ErrorReason, + } +} + +func (s *RequestStore) pruneLoop() { + // Prune on start as well as on the timer, so a node restarted more often + // than the interval still enforces its retention. + s.prune() + ticker := time.NewTicker(time.Hour) + defer ticker.Stop() + for { + select { + case <-ticker.C: + s.prune() + case <-s.stop: + return + } + } +} + +// prune enforces both limits: age, and a row cap so a burst cannot fill the +// disk before the age limit ever bites. +func (s *RequestStore) prune() { + database := db.NewDbService() + if database == nil { + return + } + + cutoff := time.Now().AddDate(0, 0, -s.retentionDays) + if result := database.Where("start_time < ?", cutoff).Delete(&RequestHistoryEntity{}); result.Error != nil { + logs.GetLogger().Warnf("Failed to prune request history by age: %v", result.Error) + } else if result.RowsAffected > 0 { + logs.GetLogger().Debugf("Pruned %d request history row(s) older than %d days", result.RowsAffected, s.retentionDays) + } + + var count int64 + if err := database.Model(&RequestHistoryEntity{}).Count(&count).Error; err != nil { + return + } + if count <= s.maxRows { + return + } + // Delete the oldest rows above the cap, identified by a threshold id rather + // than by offset: ids are monotonic, so this is one indexed delete instead + // of a scan. + var threshold uint + if err := database.Model(&RequestHistoryEntity{}). + Order("id desc").Offset(int(s.maxRows)).Limit(1). + Pluck("id", &threshold).Error; err != nil || threshold == 0 { + return + } + if result := database.Where("id <= ?", threshold).Delete(&RequestHistoryEntity{}); result.Error != nil { + logs.GetLogger().Warnf("Failed to prune request history by row cap: %v", result.Error) + } else if result.RowsAffected > 0 { + logs.GetLogger().Debugf("Pruned %d request history row(s) over the %d row cap", result.RowsAffected, s.maxRows) + } +} + +// Query returns one page of stored history, newest first, with the total +// matching the filters. +func (s *RequestStore) Query(q RequestHistoryQuery) (RequestHistoryPage, error) { + page := RequestHistoryPage{Requests: []RequestMetric{}, Limit: q.Limit, Offset: q.Offset} + if page.Limit <= 0 { + page.Limit = 100 + } + if page.Offset < 0 { + page.Offset = 0 + } + + database := db.NewDbService() + if database == nil { + return page, nil + } + + build := func() *gorm.DB { + tx := database.Model(&RequestHistoryEntity{}) + if q.Model != "" { + tx = tx.Where("model = ?", q.Model) + } + if q.Source != "" { + if q.Source == string(SourceHub) { + // Rows written before the source field existed carry none, and + // every one of them arrived over the WebSocket. Excluding them + // would make an operator's older history vanish the moment + // they filter to Hub. + tx = tx.Where("source = ? OR source = ''", q.Source) + } else { + tx = tx.Where("source = ?", q.Source) + } + } + return tx + } + + var total int64 + if err := build().Count(&total).Error; err != nil { + return page, err + } + page.Total = int(total) + + var rows []RequestHistoryEntity + if err := build(). + Order("start_time desc, id desc"). + Offset(page.Offset).Limit(page.Limit). + Find(&rows).Error; err != nil { + return page, err + } + for _, row := range rows { + page.Requests = append(page.Requests, RequestMetric{ + RequestID: row.RequestID, + Model: row.Model, + StartTime: row.StartTime, + EndTime: row.EndTime, + LatencyMs: row.LatencyMs, + TokensIn: row.TokensIn, + TokensOut: row.TokensOut, + Streaming: row.Streaming, + Success: row.Success, + ErrorReason: row.ErrorReason, + Source: RequestSource(row.Source), + }) + } + return page, nil +} + +// Dropped reports how many records were discarded because the queue was full. +func (s *RequestStore) Dropped() int64 { + if s == nil { + return 0 + } + return s.dropped.Load() +} diff --git a/internal/computing/request_history_store_test.go b/internal/computing/request_history_store_test.go new file mode 100644 index 00000000..147c194e --- /dev/null +++ b/internal/computing/request_history_store_test.go @@ -0,0 +1,234 @@ +package computing + +import ( + "testing" + "time" + + "github.com/swanchain/computing-provider-v2/internal/db" +) + +// withStore gives each test its own database file, so these exercise the real +// SQLite path rather than a stand-in. +func withStore(t *testing.T, retentionDays int, maxRows int64) *RequestStore { + t.Helper() + db.InitDb(t.TempDir()) + if db.NewDbService() == nil { + t.Skip("database unavailable") + } + store := NewRequestStore(retentionDays, maxRows) + if err := store.Start(); err != nil { + t.Fatalf("start store: %v", err) + } + t.Cleanup(store.Stop) + return store +} + +// flush pushes queued records to disk by stopping the writer, which drains. +func flush(t *testing.T, store *RequestStore) { + t.Helper() + store.Stop() +} + +func record(store *RequestStore, id, model string, source RequestSource, at time.Time) { + store.Record(RequestMetric{ + RequestID: id, Model: model, Source: source, + StartTime: at, EndTime: at.Add(time.Second), + LatencyMs: 120, TokensIn: 10, TokensOut: 5, Success: true, + }) +} + +func TestRequestStorePersistsAndPages(t *testing.T) { + store := withStore(t, 7, 1000) + base := time.Now().UTC().Add(-time.Hour) + for i := 0; i < 5; i++ { + record(store, string(rune('a'+i)), "m", SourceHub, base.Add(time.Duration(i)*time.Minute)) + } + flush(t, store) + + page, err := store.Query(RequestHistoryQuery{Limit: 2}) + if err != nil { + t.Fatalf("query: %v", err) + } + if page.Total != 5 { + t.Errorf("total = %d, want 5", page.Total) + } + if len(page.Requests) != 2 { + t.Fatalf("page held %d rows, want 2", len(page.Requests)) + } + // Newest first. + if page.Requests[0].RequestID != "e" || page.Requests[1].RequestID != "d" { + t.Errorf("page 1 = %s,%s — want the two newest (e,d)", page.Requests[0].RequestID, page.Requests[1].RequestID) + } + + second, err := store.Query(RequestHistoryQuery{Limit: 2, Offset: 2}) + if err != nil { + t.Fatalf("query: %v", err) + } + if second.Requests[0].RequestID != "c" { + t.Errorf("page 2 started at %s, want c", second.Requests[0].RequestID) + } +} + +func TestRequestStoreFiltersByModelAndSource(t *testing.T) { + store := withStore(t, 7, 1000) + now := time.Now().UTC() + record(store, "h1", "alpha", SourceHub, now) + record(store, "p1", "alpha", SourceHealth, now) + record(store, "h2", "beta", SourceHub, now) + flush(t, store) + + hub, err := store.Query(RequestHistoryQuery{Source: string(SourceHub)}) + if err != nil { + t.Fatalf("query: %v", err) + } + if hub.Total != 2 { + t.Errorf("hub total = %d, want 2", hub.Total) + } + + combined, err := store.Query(RequestHistoryQuery{Model: "alpha", Source: string(SourceHub)}) + if err != nil { + t.Fatalf("query: %v", err) + } + if combined.Total != 1 || combined.Requests[0].RequestID != "h1" { + t.Errorf("model+source = %+v, want just h1", combined.Requests) + } +} + +// Rows written before the source column carry none, and all of them arrived +// over the WebSocket. Filtering to Hub must include them, matching what the +// in-memory path does. +func TestRequestStoreTreatsSourcelessRowsAsHub(t *testing.T) { + store := withStore(t, 7, 1000) + now := time.Now().UTC() + store.Record(RequestMetric{RequestID: "old", Model: "m", StartTime: now, Success: true}) // no Source + record(store, "new", "m", SourceHub, now.Add(time.Minute)) + record(store, "probe", "m", SourceHealth, now.Add(2*time.Minute)) + flush(t, store) + + hub, err := store.Query(RequestHistoryQuery{Source: string(SourceHub)}) + if err != nil { + t.Fatalf("query: %v", err) + } + if hub.Total != 2 { + t.Fatalf("hub total = %d, want 2 — the unlabelled row must count as hub", hub.Total) + } +} + +// History outliving a restart is the whole point: a second store over the same +// database must see what the first one wrote. +func TestRequestStoreSurvivesRestart(t *testing.T) { + dir := t.TempDir() + db.InitDb(dir) + if db.NewDbService() == nil { + t.Skip("database unavailable") + } + + first := NewRequestStore(7, 1000) + if err := first.Start(); err != nil { + t.Fatalf("start: %v", err) + } + record(first, "before-restart", "m", SourceHub, time.Now().UTC()) + first.Stop() // drains + + second := NewRequestStore(7, 1000) + if err := second.Start(); err != nil { + t.Fatalf("restart: %v", err) + } + defer second.Stop() + + page, err := second.Query(RequestHistoryQuery{Limit: 10}) + if err != nil { + t.Fatalf("query: %v", err) + } + if page.Total != 1 || page.Requests[0].RequestID != "before-restart" { + t.Errorf("after restart the store held %+v, want the record written before it", page.Requests) + } +} + +func TestRequestStorePrunesByAge(t *testing.T) { + store := withStore(t, 1, 1000) + now := time.Now().UTC() + record(store, "old", "m", SourceHub, now.AddDate(0, 0, -3)) + record(store, "fresh", "m", SourceHub, now) + flush(t, store) + + store.prune() + + page, err := store.Query(RequestHistoryQuery{Limit: 10}) + if err != nil { + t.Fatalf("query: %v", err) + } + if page.Total != 1 || page.Requests[0].RequestID != "fresh" { + t.Errorf("after pruning: %+v, want only the row inside the retention window", page.Requests) + } +} + +func TestRequestStorePrunesByRowCap(t *testing.T) { + store := withStore(t, 30, 3) + now := time.Now().UTC() + for i := 0; i < 10; i++ { + record(store, string(rune('a'+i)), "m", SourceHub, now.Add(time.Duration(i)*time.Second)) + } + flush(t, store) + + store.prune() + + page, err := store.Query(RequestHistoryQuery{Limit: 50}) + if err != nil { + t.Fatalf("query: %v", err) + } + if page.Total != 3 { + t.Errorf("total = %d, want the 3-row cap enforced", page.Total) + } + // The cap must keep the newest, not an arbitrary three. + if page.Requests[0].RequestID != "j" { + t.Errorf("newest kept = %s, want j", page.Requests[0].RequestID) + } +} + +// Recording must never block serving, even when the writer cannot keep up. +func TestRequestStoreRecordNeverBlocks(t *testing.T) { + store := NewRequestStore(7, 1000) // not started: nothing drains the queue + now := time.Now().UTC() + done := make(chan struct{}) + go func() { + for i := 0; i < queueSize+500; i++ { + store.Record(RequestMetric{RequestID: "x", Model: "m", StartTime: now}) + } + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Record blocked when the queue was full — it must drop instead") + } + if store.Dropped() == 0 { + t.Error("overflow should be counted, not silent") + } +} + +// An upstream error body can be arbitrarily long, and there is one row per +// request. +func TestRequestStoreTruncatesLongErrors(t *testing.T) { + store := withStore(t, 7, 1000) + long := make([]byte, 4000) + for i := range long { + long[i] = 'x' + } + store.Record(RequestMetric{ + RequestID: "e", Model: "m", Source: SourceHub, + StartTime: time.Now().UTC(), ErrorReason: string(long), + }) + flush(t, store) + + page, err := store.Query(RequestHistoryQuery{Limit: 1}) + if err != nil { + t.Fatalf("query: %v", err) + } + if len(page.Requests) != 1 { + t.Fatalf("expected the row back") + } + if got := len(page.Requests[0].ErrorReason); got != maxErrorReasonBytes { + t.Errorf("stored error reason is %d bytes, want it capped at %d", got, maxErrorReasonBytes) + } +}