package services import ( "fmt" "net/http" "net/http/httptest" "sync/atomic" "testing" "time" "lokistack/models" "lokistack/repository" ) // 条数(index/stats)只给"已结束且还没有条数"的天补,并且每天最多 3 次。 // 这条护栏的直接目的:别让 30 分钟一次的定时任务把整天 index/stats 反复重跑。 func TestCaptureDailyEntriesOnlyForPastDaysAndWrites(t *testing.T) { setupDailyVolumeDB(t) var calls int32 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { atomic.AddInt32(&calls, 1) if r.URL.Path != "/loki/api/v1/index/stats" { t.Errorf("意外请求路径: %s", r.URL.Path) } w.Header().Set("Content-Type", "application/json") fmt.Fprint(w, `{"streams":3,"chunks":9,"bytes":123456,"entries":456}`) })) defer srv.Close() src := &models.LogSource{ID: 1, Name: "test", TargetType: "url", URL: srv.URL} yesterday := dateKey(time.Now().AddDate(0, 0, -1)) today := dateKey(time.Now()) db := repository.GetDB() db.Lock() db.DailyVolumeSnapshots = []models.DailyVolumeSnapshot{ {ID: 1, Date: yesterday, SourceID: 1, Bytes: 100}, {ID: 2, Date: today, SourceID: 1, Bytes: 10, Partial: true}, } db.Unlock() captureDailyEntries(src) got := snapshotsByDate(t) if snap := got[yesterday]; snap.Entries != 456 { t.Fatalf("昨天的条数应写入 456,得到 %d", snap.Entries) } if snap := got[yesterday]; snap.EntriesAttempts != 1 { t.Errorf("昨天应记录 1 次尝试,得到 %d", snap.EntriesAttempts) } if snap := got[today]; snap.EntriesAttempts != 0 || snap.Entries != 0 { t.Errorf("今天没结束,不应抓条数(attempts=%d entries=%d)", snap.EntriesAttempts, snap.Entries) } if n := atomic.LoadInt32(&calls); n != 1 { t.Errorf("应只打 1 次 index/stats,实际 %d 次", n) } // 1 小时内重复触发不应再打 Loki captureDailyEntries(src) if n := atomic.LoadInt32(&calls); n != 1 { t.Errorf("重试间隔内不应再查询,实际 %d 次", n) } } // 失败(Loki 报错)时必须吃得下,并且最多尝试 entriesMaxAttempts 次就放弃。 func TestCaptureDailyEntriesRetryBudget(t *testing.T) { setupDailyVolumeDB(t) var calls int32 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { atomic.AddInt32(&calls, 1) http.Error(w, "boom", http.StatusInternalServerError) })) defer srv.Close() src := &models.LogSource{ID: 1, Name: "test", TargetType: "url", URL: srv.URL} yesterday := dateKey(time.Now().AddDate(0, 0, -1)) db := repository.GetDB() db.Lock() db.DailyVolumeSnapshots = []models.DailyVolumeSnapshot{{ID: 1, Date: yesterday, SourceID: 1, Bytes: 100}} db.Unlock() // 模拟"每轮定时任务都到点":把上次尝试时间清空,绕过 1 小时间隔 for i := 0; i < entriesMaxAttempts+3; i++ { captureDailyEntries(src) db.Lock() for j := range db.DailyVolumeSnapshots { db.DailyVolumeSnapshots[j].LastEntriesAttempt = time.Time{} } db.Unlock() } if n := atomic.LoadInt32(&calls); n != entriesMaxAttempts { t.Errorf("应恰好尝试 %d 次后放弃,实际 %d 次", entriesMaxAttempts, n) } if snap := snapshotsByDate(t)[yesterday]; snap.Entries != 0 || snap.EntriesAttempts != entriesMaxAttempts { t.Errorf("失败后应留 entries=0 / attempts=%d,得到 entries=%d attempts=%d", entriesMaxAttempts, snap.Entries, snap.EntriesAttempts) } }