274 lines
8.5 KiB
Go
274 lines
8.5 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"sort"
|
|
"time"
|
|
|
|
"github.com/QuantumNous/new-api/common"
|
|
"github.com/QuantumNous/new-api/model"
|
|
fingerprint "github.com/QuantumNous/new-api/pkg/model_fingerprint"
|
|
)
|
|
|
|
type DetectionHistory struct {
|
|
Closest string `json:"closest,omitempty"`
|
|
Weight float64 `json:"relative_weight,omitempty"`
|
|
ID string `json:"id"`
|
|
At int64 `json:"at"`
|
|
Status string `json:"status"`
|
|
Quality string `json:"quality"`
|
|
UserInitiated bool `json:"user_initiated"`
|
|
}
|
|
type BenchmarkSummary struct {
|
|
LatestValid *DetectionHistory `json:"-"`
|
|
Channels int `json:"channels"`
|
|
IntervalMinutes int `json:"interval_minutes"`
|
|
IntervalMaxMinutes int `json:"interval_max_minutes"`
|
|
WindowHours int `json:"window_hours"`
|
|
Normal int `json:"normal"`
|
|
Degraded int `json:"degraded"`
|
|
Unknown int `json:"unknown"`
|
|
Failed int `json:"failed"`
|
|
Running int `json:"running"`
|
|
History []DetectionHistory `json:"history"`
|
|
}
|
|
type ChannelDetectionEvidence struct {
|
|
At int64
|
|
Success bool
|
|
}
|
|
type GroupDetectionSummary struct {
|
|
ProbeHistory []ProbeHistoryPoint
|
|
Candy *BenchmarkSummary
|
|
Benchmark *BenchmarkSummary
|
|
Status int
|
|
At int64
|
|
}
|
|
|
|
func DetectionFallbackStatus(ids []int, evidence map[int]ChannelDetectionEvidence, now int64) (int, int64) {
|
|
good, bad, at := 0, 0, int64(0)
|
|
for _, id := range ids {
|
|
e, ok := evidence[id]
|
|
if !ok || e.At < now-7200 {
|
|
continue
|
|
}
|
|
at = max(at, e.At)
|
|
if e.Success {
|
|
good++
|
|
} else {
|
|
bad++
|
|
}
|
|
}
|
|
if len(ids) > 0 && good == len(ids) {
|
|
return 1, at
|
|
}
|
|
if good > 0 {
|
|
return 2, at
|
|
}
|
|
if len(ids) > 0 && bad == len(ids) {
|
|
return 0, at
|
|
}
|
|
return -1, at
|
|
}
|
|
|
|
func GroupDetectionSummaries(ctx context.Context, now time.Time, fingerprints map[string]*AstraSamplingSummary, settings map[string]model.StatusPageGroup) (map[string]GroupDetectionSummary, error) {
|
|
channels, err := samplingChannels(ctx, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
groups := map[string][]int{}
|
|
byChannel := map[int][]string{}
|
|
benchmarks := map[string]*BenchmarkSummary{}
|
|
candies := map[string]*BenchmarkSummary{}
|
|
evidence := map[int]ChannelDetectionEvidence{}
|
|
observe := func(id int, at int64, success bool) {
|
|
if at > evidence[id].At {
|
|
evidence[id] = ChannelDetectionEvidence{At: at, Success: success}
|
|
}
|
|
}
|
|
for _, channel := range channels {
|
|
byChannel[channel.Id] = channel.GetGroups()
|
|
for _, group := range channel.GetGroups() {
|
|
groups[group] = append(groups[group], channel.Id)
|
|
if ChannelSupportsAstra(&channel) {
|
|
if benchmarks[group] == nil {
|
|
benchmarks[group] = &BenchmarkSummary{IntervalMinutes: 30, WindowHours: 6, History: []DetectionHistory{}}
|
|
}
|
|
benchmarks[group].Channels++
|
|
}
|
|
}
|
|
}
|
|
candyRanges, err := DetectionScheduleRanges(ctx, channels, "candy")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for group, summary := range benchmarks {
|
|
r := candyRanges[group]
|
|
candies[group] = &BenchmarkSummary{Channels: summary.Channels, IntervalMinutes: r[0], IntervalMaxMinutes: r[1], WindowHours: 6, History: []DetectionHistory{}}
|
|
}
|
|
var tasks []model.SystemTask
|
|
ranges, err := DetectionScheduleRanges(ctx, channels, "benchmark")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for group, summary := range benchmarks {
|
|
r := ranges[group]
|
|
summary.IntervalMinutes, summary.IntervalMaxMinutes = r[0], r[1]
|
|
}
|
|
if err = model.DB.WithContext(ctx).Where("type IN ? AND created_at >= ?", []string{AstraSamplingTaskType, model.SystemTaskTypeChannelAvailability}, now.Add(-6*time.Hour).Unix()).Find(&tasks).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
for _, task := range tasks {
|
|
if task.Type == AstraSamplingTaskType {
|
|
var state AstraSamplingState
|
|
if task.DecodeState(&state) != nil {
|
|
continue
|
|
}
|
|
for _, round := range state.Rounds {
|
|
if round.FinishedAt <= 0 {
|
|
continue
|
|
}
|
|
if round.Result.ValidSamples > 0 {
|
|
observe(round.ChannelID, round.FinishedAt, true)
|
|
} else if round.Reason == "transport_error" || round.Reason == "channel_unavailable" {
|
|
observe(round.ChannelID, round.FinishedAt, false)
|
|
}
|
|
}
|
|
} else {
|
|
var state ChannelAvailabilityState
|
|
if task.DecodeState(&state) != nil {
|
|
continue
|
|
}
|
|
for _, round := range state.Rounds {
|
|
observe(round.ChannelID, round.At, round.Success)
|
|
}
|
|
}
|
|
}
|
|
var runs []model.DetectionRun
|
|
if err = model.DB.WithContext(ctx).Omit("html", "price", "remote_id").Where("created_at >= ?", now.Add(-6*time.Hour).Unix()).Find(&runs).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
ids := []string{}
|
|
for _, run := range runs {
|
|
ids = append(ids, run.ID)
|
|
}
|
|
var calls []model.DetectionCall
|
|
if len(ids) > 0 {
|
|
if err = model.DB.WithContext(ctx).Select("run_id", "finished_at", "available").Where("run_id IN ? AND state = ?", ids, "finished").Find(&calls).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
runChannels := map[string]int{}
|
|
for _, run := range runs {
|
|
runChannels[run.ID] = run.ChannelID
|
|
}
|
|
for _, call := range calls {
|
|
if call.Available != nil {
|
|
observe(runChannels[call.RunID], call.FinishedAt, *call.Available)
|
|
}
|
|
}
|
|
for _, run := range runs {
|
|
if run.Kind != "fingerprint" && run.Kind != "benchmark" && run.Kind != "candy" {
|
|
continue
|
|
}
|
|
for _, group := range byChannel[run.ChannelID] {
|
|
if cutoff := settings[group].ResetBefore; cutoff > 0 && run.CreatedAt <= cutoff {
|
|
continue
|
|
}
|
|
if run.Kind == "fingerprint" {
|
|
summary := fingerprints[group]
|
|
if summary == nil || !run.Settled {
|
|
continue
|
|
}
|
|
var result fingerprint.Result
|
|
if common.UnmarshalJsonStr(run.Result, &result) != nil {
|
|
result.Status = "error"
|
|
}
|
|
if result.Status == "" {
|
|
result.Status = "error"
|
|
}
|
|
if result.Engine != fingerprint.Version {
|
|
continue
|
|
}
|
|
result.Status = AstraSamplingResultStatus(result)
|
|
switch result.Status {
|
|
case "match":
|
|
summary.Matches++
|
|
case "mismatch":
|
|
summary.Mismatches++
|
|
case "inconclusive":
|
|
summary.Inconclusive++
|
|
default:
|
|
summary.Errors++
|
|
}
|
|
summary.History = append(summary.History, AstraSamplingHistory{ID: run.ID, At: run.CreatedAt, Status: result.Status, ValidSamples: result.ValidSamples, Closest: result.Closest, Gap: result.Gap, UserInitiated: true})
|
|
if result.ValidSamples > 0 && result.Closest != "" && (result.Status == "match" || result.Status == "mismatch") && (summary.LatestValid == nil || run.CreatedAt > summary.LatestValid.At) {
|
|
round := summary.History[len(summary.History)-1]
|
|
summary.LatestValid = &round
|
|
}
|
|
continue
|
|
}
|
|
s := benchmarks[group]
|
|
if run.Kind == "candy" {
|
|
s = candies[group]
|
|
}
|
|
if s == nil {
|
|
continue
|
|
}
|
|
history := DetectionHistory{ID: run.ID, At: run.CreatedAt, Status: run.Status, Quality: run.Quality, UserInitiated: run.UserID > 0}
|
|
var metadata fingerprint.Result
|
|
if common.UnmarshalJsonStr(run.Result, &metadata) == nil {
|
|
history.Closest, history.Weight = metadata.Closest, metadata.Weight
|
|
}
|
|
s.History = append(s.History, history)
|
|
if run.Status == "succeeded" && (run.Quality == "normal" || run.Quality == "degraded") && (s.LatestValid == nil || history.At > s.LatestValid.At) {
|
|
s.LatestValid = &history
|
|
}
|
|
switch run.Status {
|
|
case "pending", "running":
|
|
s.Running++
|
|
case "failed":
|
|
s.Failed++
|
|
default:
|
|
switch run.Quality {
|
|
case "normal":
|
|
s.Normal++
|
|
case "degraded":
|
|
s.Degraded++
|
|
default:
|
|
s.Unknown++
|
|
}
|
|
}
|
|
}
|
|
}
|
|
for _, s := range fingerprints {
|
|
if n := s.Matches + s.Mismatches; n > 0 {
|
|
rate := float64(s.Mismatches) / float64(n)
|
|
s.Rate = &rate
|
|
}
|
|
sort.SliceStable(s.History, func(i, j int) bool { return s.History[i].At > s.History[j].At })
|
|
if len(s.History) > 12 {
|
|
s.History = s.History[:12]
|
|
}
|
|
}
|
|
results := map[string]GroupDetectionSummary{}
|
|
for group, ids := range groups {
|
|
status, at := DetectionFallbackStatus(ids, evidence, now.Unix())
|
|
s := benchmarks[group]
|
|
if s != nil {
|
|
sort.SliceStable(s.History, func(i, j int) bool { return s.History[i].At > s.History[j].At })
|
|
if len(s.History) > 12 {
|
|
s.History = s.History[:12]
|
|
}
|
|
}
|
|
candy := candies[group]
|
|
if candy != nil {
|
|
sort.SliceStable(candy.History, func(i, j int) bool { return candy.History[i].At > candy.History[j].At })
|
|
if len(candy.History) > 12 {
|
|
candy.History = candy.History[:12]
|
|
}
|
|
}
|
|
results[group] = GroupDetectionSummary{Candy: candy, Benchmark: s, Status: status, At: at, ProbeHistory: GroupProbeHistory(ids, tasks)}
|
|
}
|
|
return results, nil
|
|
}
|