109 lines
3.8 KiB
Go
109 lines
3.8 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
"github.com/QuantumNous/new-api/common"
|
|
"github.com/QuantumNous/new-api/model"
|
|
fingerprint "github.com/QuantumNous/new-api/pkg/model_fingerprint"
|
|
)
|
|
|
|
// Routing evidence has no statistical window. Each detector recovers separately.
|
|
func readGroupQuality(ctx context.Context, now time.Time) (map[string]AstraSamplingHistory, map[string]DetectionHistory, error) {
|
|
settings, err := model.GetStatusPageGroups(ctx)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
channels, err := samplingChannels(ctx, false)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
channelGroups := map[int][]string{}
|
|
for _, channel := range channels {
|
|
channelGroups[channel.Id] = channel.GetGroups()
|
|
}
|
|
latest, benchmarks := map[string]AstraSamplingHistory{}, map[string]DetectionHistory{}
|
|
observe := func(channelID int, h AstraSamplingHistory) {
|
|
if h.At <= 0 || h.At > now.Unix() || h.ValidSamples <= 0 || h.Closest == "" || (h.Status != "match" && h.Status != "mismatch") {
|
|
return
|
|
}
|
|
for _, group := range channelGroups[channelID] {
|
|
if h.At > settings[group].ResetBefore && h.At > latest[group].At {
|
|
latest[group] = h
|
|
}
|
|
}
|
|
}
|
|
var tasks []model.SystemTask
|
|
if err = model.DB.WithContext(ctx).Select("state").Where("type = ?", AstraSamplingTaskType).Find(&tasks).Error; err != nil {
|
|
return nil, nil, err
|
|
}
|
|
for _, task := range tasks {
|
|
var state AstraSamplingState
|
|
if task.DecodeState(&state) != nil {
|
|
continue
|
|
}
|
|
for _, round := range state.Rounds {
|
|
if round.FinishedAt <= 0 || round.Result.Engine != fingerprint.Version {
|
|
continue
|
|
}
|
|
observe(round.ChannelID, AstraSamplingHistory{ID: round.ID, At: round.StartedAt, Status: AstraSamplingResultStatus(round.Result), ValidSamples: round.Result.ValidSamples, Closest: round.Result.Closest})
|
|
}
|
|
}
|
|
var runs []model.DetectionRun
|
|
if err = model.DB.WithContext(ctx).Select("id", "channel_id", "kind", "created_at", "quality", "result").Where("status = ? AND settled = ? AND created_at <= ?", "succeeded", true, now.Unix()).Find(&runs).Error; err != nil {
|
|
return nil, nil, err
|
|
}
|
|
for _, run := range runs {
|
|
if run.Kind == "fingerprint" {
|
|
var result fingerprint.Result
|
|
if common.UnmarshalJsonStr(run.Result, &result) != nil || result.Engine != fingerprint.Version {
|
|
continue
|
|
}
|
|
observe(run.ChannelID, AstraSamplingHistory{ID: run.ID, At: run.CreatedAt, Status: AstraSamplingResultStatus(result), ValidSamples: result.ValidSamples, Closest: result.Closest})
|
|
}
|
|
|
|
}
|
|
return latest, benchmarks, nil
|
|
}
|
|
|
|
func cachedGroupQuality(ctx context.Context) (map[string]AstraSamplingHistory, map[string]DetectionHistory, error) {
|
|
groupDegradationSnapshot.Lock()
|
|
defer groupDegradationSnapshot.Unlock()
|
|
now := time.Now()
|
|
if now.Sub(groupDegradationSnapshot.updated) >= 15*time.Second {
|
|
ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
|
|
latest, benchmarks, err := readGroupQuality(ctx, now)
|
|
cancel()
|
|
if err == nil {
|
|
groupDegradationSnapshot.latest, groupDegradationSnapshot.benchmark = latest, benchmarks
|
|
} else if groupDegradationSnapshot.latest == nil {
|
|
return nil, nil, errors.New("group quality temporarily unavailable")
|
|
}
|
|
groupDegradationSnapshot.updated = now
|
|
}
|
|
return groupDegradationSnapshot.latest, groupDegradationSnapshot.benchmark, nil
|
|
}
|
|
|
|
func RecentDegradedGroups(ctx context.Context) (map[string]bool, error) {
|
|
latest, _, err := cachedGroupQuality(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result := map[string]bool{}
|
|
cutoff := time.Now().Add(-AstraSamplingWindow).Unix()
|
|
for group, h := range latest {
|
|
if h.At >= cutoff && h.Closest != fingerprint.Model {
|
|
result[group] = true
|
|
}
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func InvalidateGroupQualityCache() {
|
|
groupDegradationSnapshot.Lock()
|
|
defer groupDegradationSnapshot.Unlock()
|
|
groupDegradationSnapshot.updated = time.Time{}
|
|
}
|