mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-09-01 15:02:58 +08:00
feat(channel-monitor-v2): 实现被动聚合、分层保留与只读 API
基于 usage_logs/ops_error_logs 做分钟聚合与 5m/1h/12h/1d rollup，提供 snapshot/matrix/errors/users 等接口；V1 在 mode=v2 时停用主动探测， 并对用户端按需脱敏绝对量与 RPM/TPM。
This commit is contained in:
IanShaw027
@@ -0,0 +1,204 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/response"
|
||||
"github.com/Wei-Shaw/sub2api/internal/server/middleware"
|
||||
"github.com/Wei-Shaw/sub2api/internal/service"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
type ChannelMonitorV2Handler struct {
|
||||
service *service.ChannelMonitorV2Service
|
||||
}
|
||||
|
||||
func NewChannelMonitorV2Handler(svc *service.ChannelMonitorV2Service) *ChannelMonitorV2Handler {
|
||||
return &ChannelMonitorV2Handler{service: svc}
|
||||
}
|
||||
|
||||
// channelMonitorV2IsAdmin is true when the request already passed admin auth
|
||||
// (shared Dimensions/Errors handlers serve both user and admin route groups).
|
||||
func channelMonitorV2IsAdmin(c *gin.Context) bool {
|
||||
role, ok := middleware.GetUserRoleFromContext(c)
|
||||
return ok && role == service.RoleAdmin
|
||||
}
|
||||
|
||||
func (h *ChannelMonitorV2Handler) GetConfig(c *gin.Context) {
|
||||
cfg, err := h.service.GetConfig(c.Request.Context())
|
||||
if err != nil {
|
||||
response.ErrorFrom(c, err)
|
||||
return
|
||||
}
|
||||
response.Success(c, cfg)
|
||||
}
|
||||
|
||||
func (h *ChannelMonitorV2Handler) UpdateConfig(c *gin.Context) {
|
||||
var input service.ChannelMonitorV2Config
|
||||
if err := c.ShouldBindJSON(&input); err != nil {
|
||||
response.BadRequest(c, "invalid channel monitor v2 config")
|
||||
return
|
||||
}
|
||||
subject, ok := middleware.GetAuthSubjectFromContext(c)
|
||||
if !ok || subject.UserID <= 0 {
|
||||
response.Unauthorized(c, "user not found in context")
|
||||
return
|
||||
}
|
||||
updated, err := h.service.UpdateConfig(c.Request.Context(), input, input.Version, subject.UserID)
|
||||
if err != nil {
|
||||
if errors.Is(err, service.ErrChannelMonitorV2ConfigConflict) {
|
||||
response.Error(c, http.StatusConflict, err.Error())
|
||||
return
|
||||
}
|
||||
if errors.Is(err, service.ErrChannelMonitorV2InvalidConfig) {
|
||||
response.BadRequest(c, err.Error())
|
||||
return
|
||||
}
|
||||
response.ErrorFrom(c, err)
|
||||
return
|
||||
}
|
||||
response.Success(c, updated)
|
||||
}
|
||||
|
||||
func (h *ChannelMonitorV2Handler) Dimensions(c *gin.Context) {
|
||||
filter, ok := h.parseFilter(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
result, err := h.service.Dimensions(c.Request.Context(), filter)
|
||||
if err != nil {
|
||||
response.ErrorFrom(c, err)
|
||||
return
|
||||
}
|
||||
// Admin and user share this handler; only non-admin responses strip volume.
|
||||
if !channelMonitorV2IsAdmin(c) {
|
||||
service.RedactChannelMonitorV2Dimensions(result)
|
||||
}
|
||||
response.Success(c, result)
|
||||
}
|
||||
|
||||
func (h *ChannelMonitorV2Handler) Snapshot(c *gin.Context) { h.snapshot(c, false) }
|
||||
func (h *ChannelMonitorV2Handler) AdminSnapshot(c *gin.Context) { h.snapshot(c, true) }
|
||||
func (h *ChannelMonitorV2Handler) Models(c *gin.Context) { h.models(c, false) }
|
||||
func (h *ChannelMonitorV2Handler) AdminModels(c *gin.Context) { h.models(c, true) }
|
||||
func (h *ChannelMonitorV2Handler) Matrix(c *gin.Context) { h.matrix(c, false) }
|
||||
func (h *ChannelMonitorV2Handler) AdminMatrix(c *gin.Context) { h.matrix(c, true) }
|
||||
func (h *ChannelMonitorV2Handler) Users(c *gin.Context) { h.users(c, false) }
|
||||
func (h *ChannelMonitorV2Handler) AdminUsers(c *gin.Context) { h.users(c, true) }
|
||||
|
||||
func (h *ChannelMonitorV2Handler) snapshot(c *gin.Context, admin bool) {
|
||||
filter, ok := h.parseFilter(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
result, err := h.service.Snapshot(c.Request.Context(), filter, admin)
|
||||
if err != nil {
|
||||
response.ErrorFrom(c, err)
|
||||
return
|
||||
}
|
||||
response.Success(c, result)
|
||||
}
|
||||
|
||||
func (h *ChannelMonitorV2Handler) models(c *gin.Context, admin bool) {
|
||||
filter, ok := h.parseFilter(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
result, err := h.service.Models(c.Request.Context(), filter, admin)
|
||||
if err != nil {
|
||||
response.ErrorFrom(c, err)
|
||||
return
|
||||
}
|
||||
response.Success(c, result)
|
||||
}
|
||||
|
||||
func (h *ChannelMonitorV2Handler) matrix(c *gin.Context, admin bool) {
|
||||
filter, ok := h.parseFilter(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
groupBy, err := service.ParseChannelMonitorV2GroupBy(c.Query("group_by"))
|
||||
if err != nil {
|
||||
response.BadRequest(c, err.Error())
|
||||
return
|
||||
}
|
||||
result, err := h.service.Matrix(c.Request.Context(), filter, groupBy, admin)
|
||||
if err != nil {
|
||||
response.ErrorFrom(c, err)
|
||||
return
|
||||
}
|
||||
response.Success(c, result)
|
||||
}
|
||||
|
||||
func (h *ChannelMonitorV2Handler) Errors(c *gin.Context) {
|
||||
filter, ok := h.parseFilter(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
result, err := h.service.ErrorsForViewer(c.Request.Context(), filter, channelMonitorV2IsAdmin(c))
|
||||
if err != nil {
|
||||
response.ErrorFrom(c, err)
|
||||
return
|
||||
}
|
||||
response.Success(c, result)
|
||||
}
|
||||
|
||||
func (h *ChannelMonitorV2Handler) users(c *gin.Context, admin bool) {
|
||||
filter, ok := h.parseFilter(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
subject, exists := middleware.GetAuthSubjectFromContext(c)
|
||||
if !exists {
|
||||
response.Error(c, http.StatusUnauthorized, "user not found in context")
|
||||
return
|
||||
}
|
||||
result, err := h.service.Users(c.Request.Context(), filter, subject.UserID, admin)
|
||||
if err != nil {
|
||||
response.ErrorFrom(c, err)
|
||||
return
|
||||
}
|
||||
response.Success(c, result)
|
||||
}
|
||||
|
||||
func (h *ChannelMonitorV2Handler) parseFilter(c *gin.Context) (service.ChannelMonitorV2Filter, bool) {
|
||||
groups, err := parseChannelMonitorV2GroupIDs(queryList(c, "group_id"))
|
||||
if err != nil {
|
||||
response.BadRequest(c, "invalid group_id")
|
||||
return service.ChannelMonitorV2Filter{}, false
|
||||
}
|
||||
filter, err := h.service.ParseFilter(c.Query("range"), queryList(c, "platform"), queryList(c, "model"), groups)
|
||||
if err != nil {
|
||||
response.BadRequest(c, err.Error())
|
||||
return service.ChannelMonitorV2Filter{}, false
|
||||
}
|
||||
return filter, true
|
||||
}
|
||||
|
||||
func queryList(c *gin.Context, key string) []string {
|
||||
values := c.QueryArray(key)
|
||||
result := make([]string, 0, len(values))
|
||||
for _, value := range values {
|
||||
for _, part := range strings.Split(value, ",") {
|
||||
if part = strings.TrimSpace(part); part != "" {
|
||||
result = append(result, part)
|
||||
}
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func parseChannelMonitorV2GroupIDs(values []string) ([]int64, error) {
|
||||
result := make([]int64, 0, len(values))
|
||||
for _, value := range values {
|
||||
id, err := strconv.ParseInt(value, 10, 64)
|
||||
if err != nil || id <= 0 {
|
||||
return nil, errors.New("invalid group id")
|
||||
}
|
||||
result = append(result, id)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/service"
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestChannelMonitorV2QueryListSupportsRepeatedAndCommaValues(t *testing.T) {
|
||||
c, _ := gin.CreateTestContext(nil)
|
||||
c.Request = httptest.NewRequest("GET", "/?platform=openai,grok&platform=anthropic", nil)
|
||||
require.Equal(t, []string{"openai", "grok", "anthropic"}, queryList(c, "platform"))
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2GroupByQueryDefaultsAndRejectsInvalid(t *testing.T) {
|
||||
groupBy, err := service.ParseChannelMonitorV2GroupBy("")
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, service.ChannelMonitorV2GroupByPlatformGroup, groupBy)
|
||||
_, err = service.ParseChannelMonitorV2GroupBy("invalid")
|
||||
require.Error(t, err)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2MatrixHandlerRejectsInvalidGroupBy(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
recorder := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(recorder)
|
||||
c.Request = httptest.NewRequest(http.MethodGet, "/channel-monitor-v2/matrix?group_by=invalid", nil)
|
||||
h := NewChannelMonitorV2Handler(service.NewChannelMonitorV2Service(nil))
|
||||
h.Matrix(c)
|
||||
require.Equal(t, http.StatusBadRequest, recorder.Code)
|
||||
}
|
||||
@@ -0,0 +1,410 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Platform is derived from group/account (usage_logs has no provider column on upstream schema).
|
||||
const channelMonitorV2PlatformSQL = `lower(COALESCE(NULLIF(TRIM(g.platform), ''), NULLIF(TRIM(a.platform), ''), 'unknown'))`
|
||||
const channelMonitorV2ModelSQL = `COALESCE(NULLIF(TRIM(ul.requested_model), ''), NULLIF(TRIM(ul.model), ''), 'unknown')`
|
||||
|
||||
// Tiered retention balances UI windows against storage:
|
||||
//
|
||||
// 1m facts → short (late writes + rebuild rollups)
|
||||
// 5m/1h/12h/1d rollups → longer, aligned to 90m / 24h / 7d / 30d(+audit)
|
||||
//
|
||||
// Backfill may still write short-lived 1m rows for old windows so rollups can be
|
||||
// built; prune at end of each recompute drops them past their TTL while rollups remain.
|
||||
const (
|
||||
channelMonitorV2RetentionUser1m = 3 * 24 * time.Hour
|
||||
channelMonitorV2RetentionMetrics1m = 7 * 24 * time.Hour
|
||||
channelMonitorV2RetentionError1m = 7 * 24 * time.Hour
|
||||
channelMonitorV2RetentionHistogram1m = 7 * 24 * time.Hour
|
||||
channelMonitorV2RetentionRollup5m = 7 * 24 * time.Hour // bucket_seconds=300
|
||||
channelMonitorV2RetentionRollup1h = 30 * 24 * time.Hour // 3600
|
||||
channelMonitorV2RetentionRollup12h = 45 * 24 * time.Hour // 43200
|
||||
channelMonitorV2RetentionRollup1d = 90 * 24 * time.Hour // 86400
|
||||
channelMonitorV2RetentionMax = channelMonitorV2RetentionRollup1d
|
||||
)
|
||||
|
||||
// channelMonitorV2MaxRetention is the longest stored window (1d rollup). Used to
|
||||
// clamp recompute/backfill so we never scan older than product history needs.
|
||||
func channelMonitorV2MaxRetention() time.Duration {
|
||||
return channelMonitorV2RetentionMax
|
||||
}
|
||||
|
||||
func channelMonitorV2RetentionCutoff(now time.Time, retention time.Duration) time.Time {
|
||||
return now.UTC().Truncate(time.Minute).Add(-retention)
|
||||
}
|
||||
|
||||
type channelMonitorV2RetentionRule struct {
|
||||
table string
|
||||
retention time.Duration
|
||||
bucketSeconds int // 0 = fact table (no bucket_seconds column)
|
||||
}
|
||||
|
||||
// channelMonitorV2RetentionRules is ordered coarse→fine for predictable prune plans.
|
||||
var channelMonitorV2RetentionRules = []channelMonitorV2RetentionRule{
|
||||
{table: "channel_monitor_v2_user_metrics_1m", retention: channelMonitorV2RetentionUser1m},
|
||||
{table: "channel_monitor_v2_metrics_1m", retention: channelMonitorV2RetentionMetrics1m},
|
||||
{table: "channel_monitor_v2_error_metrics_1m", retention: channelMonitorV2RetentionError1m},
|
||||
{table: "channel_monitor_v2_latency_histograms_1m", retention: channelMonitorV2RetentionHistogram1m},
|
||||
{table: "channel_monitor_v2_metrics_rollup", retention: channelMonitorV2RetentionRollup5m, bucketSeconds: 300},
|
||||
{table: "channel_monitor_v2_user_metrics_rollup", retention: channelMonitorV2RetentionRollup5m, bucketSeconds: 300},
|
||||
{table: "channel_monitor_v2_error_metrics_rollup", retention: channelMonitorV2RetentionRollup5m, bucketSeconds: 300},
|
||||
{table: "channel_monitor_v2_latency_histograms_rollup", retention: channelMonitorV2RetentionRollup5m, bucketSeconds: 300},
|
||||
{table: "channel_monitor_v2_metrics_rollup", retention: channelMonitorV2RetentionRollup1h, bucketSeconds: 3600},
|
||||
{table: "channel_monitor_v2_user_metrics_rollup", retention: channelMonitorV2RetentionRollup1h, bucketSeconds: 3600},
|
||||
{table: "channel_monitor_v2_error_metrics_rollup", retention: channelMonitorV2RetentionRollup1h, bucketSeconds: 3600},
|
||||
{table: "channel_monitor_v2_latency_histograms_rollup", retention: channelMonitorV2RetentionRollup1h, bucketSeconds: 3600},
|
||||
{table: "channel_monitor_v2_metrics_rollup", retention: channelMonitorV2RetentionRollup12h, bucketSeconds: 43200},
|
||||
{table: "channel_monitor_v2_user_metrics_rollup", retention: channelMonitorV2RetentionRollup12h, bucketSeconds: 43200},
|
||||
{table: "channel_monitor_v2_error_metrics_rollup", retention: channelMonitorV2RetentionRollup12h, bucketSeconds: 43200},
|
||||
{table: "channel_monitor_v2_latency_histograms_rollup", retention: channelMonitorV2RetentionRollup12h, bucketSeconds: 43200},
|
||||
{table: "channel_monitor_v2_metrics_rollup", retention: channelMonitorV2RetentionRollup1d, bucketSeconds: 86400},
|
||||
{table: "channel_monitor_v2_user_metrics_rollup", retention: channelMonitorV2RetentionRollup1d, bucketSeconds: 86400},
|
||||
{table: "channel_monitor_v2_error_metrics_rollup", retention: channelMonitorV2RetentionRollup1d, bucketSeconds: 86400},
|
||||
{table: "channel_monitor_v2_latency_histograms_rollup", retention: channelMonitorV2RetentionRollup1d, bucketSeconds: 86400},
|
||||
}
|
||||
|
||||
func (r *channelMonitorV2Repository) pruneChannelMonitorV2Retention(ctx context.Context, tx *sql.Tx, now time.Time) error {
|
||||
for _, rule := range channelMonitorV2RetentionRules {
|
||||
cutoff := channelMonitorV2RetentionCutoff(now, rule.retention)
|
||||
var err error
|
||||
if rule.bucketSeconds == 0 {
|
||||
_, err = tx.ExecContext(ctx, fmt.Sprintf(`DELETE FROM %s WHERE bucket_start < $1`, rule.table), cutoff)
|
||||
} else {
|
||||
_, err = tx.ExecContext(ctx,
|
||||
fmt.Sprintf(`DELETE FROM %s WHERE bucket_seconds = $1 AND bucket_start < $2`, rule.table),
|
||||
rule.bucketSeconds, cutoff,
|
||||
)
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("prune %s (bucket_seconds=%d): %w", rule.table, rule.bucketSeconds, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *channelMonitorV2Repository) RecomputeRange(ctx context.Context, start, end time.Time) (err error) {
|
||||
start = start.UTC().Truncate(time.Minute)
|
||||
end = end.UTC().Truncate(time.Minute)
|
||||
now := time.Now().UTC().Truncate(time.Minute)
|
||||
// Clamp to longest rollup TTL so backfill does not scan beyond product history.
|
||||
maxCutoff := channelMonitorV2RetentionCutoff(now, channelMonitorV2MaxRetention())
|
||||
if start.Before(maxCutoff) {
|
||||
start = maxCutoff
|
||||
}
|
||||
if !start.Before(end) {
|
||||
return nil
|
||||
}
|
||||
tx, err := r.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() {
|
||||
if err != nil {
|
||||
_ = tx.Rollback()
|
||||
}
|
||||
}()
|
||||
|
||||
// Idempotent window rewrite: drop existing facts/rollups in [start,end) then re-insert.
|
||||
for _, table := range []string{
|
||||
"channel_monitor_v2_latency_histograms_rollup",
|
||||
"channel_monitor_v2_error_metrics_rollup",
|
||||
"channel_monitor_v2_user_metrics_rollup",
|
||||
"channel_monitor_v2_metrics_rollup",
|
||||
"channel_monitor_v2_latency_histograms_1m",
|
||||
"channel_monitor_v2_error_metrics_1m",
|
||||
"channel_monitor_v2_user_metrics_1m",
|
||||
"channel_monitor_v2_metrics_1m",
|
||||
} {
|
||||
if _, err = tx.ExecContext(ctx, fmt.Sprintf("DELETE FROM %s WHERE bucket_start >= $1 AND bucket_start < $2", table), start, end); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if _, err = tx.ExecContext(ctx, fmt.Sprintf(channelMonitorV2UsageMetricsSQL, channelMonitorV2PlatformSQL, channelMonitorV2ModelSQL), start, end); err != nil {
|
||||
return fmt.Errorf("aggregate channel monitor v2 usage: %w", err)
|
||||
}
|
||||
if _, err = tx.ExecContext(ctx, fmt.Sprintf(channelMonitorV2UserMetricsSQL, channelMonitorV2PlatformSQL, channelMonitorV2ModelSQL), start, end); err != nil {
|
||||
return fmt.Errorf("aggregate channel monitor v2 users: %w", err)
|
||||
}
|
||||
if _, err = tx.ExecContext(ctx, fmt.Sprintf(channelMonitorV2HistogramSQL, channelMonitorV2PlatformSQL, channelMonitorV2ModelSQL, channelMonitorV2HistogramBoundSQL("latency.value_ms")), start, end); err != nil {
|
||||
return fmt.Errorf("aggregate channel monitor v2 histograms: %w", err)
|
||||
}
|
||||
if _, err = tx.ExecContext(ctx, channelMonitorV2ErrorAggregationSQL, start, end); err != nil {
|
||||
return fmt.Errorf("aggregate channel monitor v2 errors: %w", err)
|
||||
}
|
||||
if err = r.recomputeFixedRollups(ctx, tx, start, end); err != nil {
|
||||
return err
|
||||
}
|
||||
// Drop rows past per-tier TTL (1m short, coarse rollups long). Safe after rollup
|
||||
// so a backfill chunk can build 1d rollups from temporary 1m rows then discard 1m.
|
||||
if err = r.pruneChannelMonitorV2Retention(ctx, tx, now); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = tx.ExecContext(ctx, channelMonitorV2WatermarkSQL, start, end); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = tx.Commit(); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
const channelMonitorV2UsageMetricsSQL = `
|
||||
INSERT INTO channel_monitor_v2_metrics_1m (
|
||||
bucket_start, platform, group_id, model, success_requests,
|
||||
input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens,
|
||||
ttft_sum_ms, ttft_count, duration_sum_ms, duration_count, computed_at
|
||||
)
|
||||
SELECT date_trunc('minute', ul.created_at), %s, COALESCE(ul.group_id, 0), %s,
|
||||
COUNT(DISTINCT COALESCE(NULLIF(ul.request_id, ''), 'usage:' || ul.id::text))
|
||||
FILTER (WHERE COALESCE(ul.request_type, 0) NOT IN (4, 6)),
|
||||
SUM(COALESCE(ul.input_tokens, 0)), SUM(COALESCE(ul.output_tokens, 0)),
|
||||
SUM(COALESCE(ul.cache_creation_tokens, 0)), SUM(COALESCE(ul.cache_read_tokens, 0)),
|
||||
COALESCE(SUM(ul.first_token_ms) FILTER (WHERE ul.first_token_ms IS NOT NULL), 0), COUNT(ul.first_token_ms),
|
||||
COALESCE(SUM(ul.duration_ms) FILTER (WHERE ul.duration_ms IS NOT NULL), 0), COUNT(ul.duration_ms), NOW()
|
||||
FROM usage_logs ul
|
||||
LEFT JOIN groups g ON g.id = ul.group_id
|
||||
LEFT JOIN accounts a ON a.id = ul.account_id
|
||||
WHERE ul.created_at >= $1 AND ul.created_at < $2
|
||||
GROUP BY 1, 2, 3, 4`
|
||||
|
||||
const channelMonitorV2UserMetricsSQL = `
|
||||
INSERT INTO channel_monitor_v2_user_metrics_1m (
|
||||
bucket_start, platform, group_id, model, user_id, success_requests,
|
||||
input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens,
|
||||
ttft_sum_ms, ttft_count, duration_sum_ms, duration_count, computed_at
|
||||
)
|
||||
SELECT date_trunc('minute', ul.created_at), %s, COALESCE(ul.group_id, 0), %s, ul.user_id,
|
||||
COUNT(DISTINCT COALESCE(NULLIF(ul.request_id, ''), 'usage:' || ul.id::text))
|
||||
FILTER (WHERE COALESCE(ul.request_type, 0) NOT IN (4, 6)),
|
||||
SUM(COALESCE(ul.input_tokens, 0)), SUM(COALESCE(ul.output_tokens, 0)),
|
||||
SUM(COALESCE(ul.cache_creation_tokens, 0)), SUM(COALESCE(ul.cache_read_tokens, 0)),
|
||||
COALESCE(SUM(ul.first_token_ms) FILTER (WHERE ul.first_token_ms IS NOT NULL), 0), COUNT(ul.first_token_ms),
|
||||
COALESCE(SUM(ul.duration_ms) FILTER (WHERE ul.duration_ms IS NOT NULL), 0), COUNT(ul.duration_ms), NOW()
|
||||
FROM usage_logs ul
|
||||
LEFT JOIN groups g ON g.id = ul.group_id
|
||||
LEFT JOIN accounts a ON a.id = ul.account_id
|
||||
WHERE ul.created_at >= $1 AND ul.created_at < $2 AND ul.user_id IS NOT NULL
|
||||
GROUP BY 1, 2, 3, 4, 5`
|
||||
|
||||
const channelMonitorV2HistogramSQL = `
|
||||
INSERT INTO channel_monitor_v2_latency_histograms_1m (
|
||||
bucket_start, platform, group_id, model, user_id, metric, upper_bound_ms, sample_count
|
||||
)
|
||||
SELECT date_trunc('minute', ul.created_at), %s, COALESCE(ul.group_id, 0), %s,
|
||||
audience.user_id, latency.metric, %s, COUNT(*)
|
||||
FROM usage_logs ul
|
||||
LEFT JOIN groups g ON g.id = ul.group_id
|
||||
LEFT JOIN accounts a ON a.id = ul.account_id
|
||||
CROSS JOIN LATERAL (VALUES (0::bigint), (ul.user_id)) audience(user_id)
|
||||
CROSS JOIN LATERAL (VALUES ('ttft'::text, ul.first_token_ms), ('duration'::text, ul.duration_ms)) latency(metric, value_ms)
|
||||
WHERE ul.created_at >= $1 AND ul.created_at < $2
|
||||
AND audience.user_id IS NOT NULL AND latency.value_ms IS NOT NULL AND latency.value_ms >= 0
|
||||
GROUP BY 1, 2, 3, 4, 5, 6, 7`
|
||||
|
||||
func channelMonitorV2HistogramBoundSQL(column string) string {
|
||||
return `CASE
|
||||
WHEN ` + column + ` <= 50 THEN 50 WHEN ` + column + ` <= 100 THEN 100
|
||||
WHEN ` + column + ` <= 250 THEN 250 WHEN ` + column + ` <= 500 THEN 500
|
||||
WHEN ` + column + ` <= 1000 THEN 1000 WHEN ` + column + ` <= 2000 THEN 2000
|
||||
WHEN ` + column + ` <= 3000 THEN 3000 WHEN ` + column + ` <= 5000 THEN 5000
|
||||
WHEN ` + column + ` <= 8000 THEN 8000 WHEN ` + column + ` <= 10000 THEN 10000
|
||||
WHEN ` + column + ` <= 15000 THEN 15000 WHEN ` + column + ` <= 30000 THEN 30000
|
||||
WHEN ` + column + ` <= 60000 THEN 60000 WHEN ` + column + ` <= 120000 THEN 120000
|
||||
WHEN ` + column + ` <= 300000 THEN 300000 WHEN ` + column + ` <= 600000 THEN 600000
|
||||
ELSE 2147483647 END`
|
||||
}
|
||||
|
||||
const channelMonitorV2ErrorAggregationSQL = `
|
||||
WITH dedup AS (
|
||||
SELECT DISTINCT ON (COALESCE(NULLIF(request_id, ''), 'error:' || id::text))
|
||||
date_trunc('minute', created_at) AS bucket_start,
|
||||
lower(COALESCE(NULLIF(TRIM(platform), ''), 'unknown')) AS platform,
|
||||
COALESCE(group_id, 0) AS group_id,
|
||||
COALESCE(NULLIF(TRIM(requested_model), ''), NULLIF(TRIM(model), ''), 'unknown') AS model,
|
||||
user_id, error_type, error_owner, COALESCE(status_code, 0) AS status_code,
|
||||
COALESCE(upstream_status_code, 0) AS upstream_status_code,
|
||||
lower(CONCAT_WS(' ', error_type, error_source, error_message, upstream_error_message, upstream_error_detail, error_body)) AS text,
|
||||
(CASE WHEN jsonb_typeof(upstream_errors) = 'array' THEN jsonb_array_length(upstream_errors) > 0 ELSE FALSE END
|
||||
OR error_owner = 'provider' OR upstream_status_code IS NOT NULL) AS upstream_affected,
|
||||
CASE WHEN jsonb_typeof(upstream_errors) = 'array' THEN jsonb_array_length(upstream_errors) ELSE 0 END AS upstream_attempts
|
||||
FROM ops_error_logs current_error
|
||||
WHERE current_error.created_at >= $1 AND current_error.created_at < $2 AND NOT current_error.is_count_tokens
|
||||
AND (COALESCE(current_error.status_code, 0) >= 400 OR current_error.error_type = 'cyber_policy')
|
||||
AND (NULLIF(current_error.request_id, '') IS NULL OR NOT EXISTS (
|
||||
SELECT 1 FROM ops_error_logs newer
|
||||
WHERE newer.request_id = current_error.request_id
|
||||
AND NOT newer.is_count_tokens
|
||||
AND (COALESCE(newer.status_code, 0) >= 400 OR newer.error_type = 'cyber_policy')
|
||||
AND newer.created_at < $2
|
||||
AND (newer.created_at, newer.id) > (current_error.created_at, current_error.id)
|
||||
))
|
||||
ORDER BY COALESCE(NULLIF(request_id, ''), 'error:' || id::text), created_at DESC, id DESC
|
||||
), classified AS (
|
||||
SELECT *, CASE
|
||||
-- Keep in lockstep with service.ClassifyChannelMonitorV2Error needles.
|
||||
WHEN error_type = 'cyber_policy' OR text LIKE ANY(ARRAY['%content policy%','%content_policy%','%safety policy%','%moderation%','%blocked keyword%']) THEN 'content_policy'
|
||||
WHEN status_code = 401 OR upstream_status_code = 401 OR text LIKE ANY(ARRAY['%unauthorized%','%invalid api key%','%invalid_api_key%','%authentication%','%api_key_disabled%']) THEN 'authentication'
|
||||
WHEN text LIKE ANY(ARRAY['%context window%','%context length%','%maximum prompt length%','%too many tokens%','%max_tokens%']) THEN 'context_limit'
|
||||
WHEN text LIKE ANY(ARRAY['%failed to deserialize%','%missing required parameter%','%invalid request%','%invalid_request%','%tool_choice%']) THEN 'invalid_request'
|
||||
WHEN text LIKE ANY(ARRAY['%does not support the requested model%','%not supported by any configured account%','%model not supported%','%unsupported model%']) THEN 'model_unsupported'
|
||||
WHEN text LIKE ANY(ARRAY['%group not allowed%','%group_not_allowed%','%group access%']) THEN 'group_access'
|
||||
WHEN text LIKE ANY(ARRAY['%run out of credits%','%insufficient balance%','%insufficient quota%','%subscription%','%quota exceeded%','%billing hard limit%']) THEN 'quota_or_balance'
|
||||
WHEN text LIKE ANY(ARRAY['%no available accounts%','%no healthy account%','%no healthy upstream account%','%failover budget exhausted%','%account pool%']) THEN 'account_pool_unavailable'
|
||||
WHEN status_code = 429 OR upstream_status_code = 429 OR text LIKE ANY(ARRAY['%rate limit%','%rate_limit%','%high demand%','%overloaded%','%concurrency limit%','%capacity%']) THEN 'rate_or_capacity'
|
||||
WHEN status_code IN (408,504) OR text LIKE ANY(ARRAY['%timeout%','%deadline exceeded%','%error code: 524%','%gateway time-out%','%gateway timeout%']) THEN 'timeout'
|
||||
WHEN text LIKE ANY(ARRAY['%transport%','%stream_read_error%','%connection reset%','%connection refused%','%tls%','%http2%','%missing terminal event%','%unexpected eof%']) THEN 'transport_or_stream'
|
||||
WHEN status_code = 403 OR upstream_status_code = 403 THEN 'upstream_forbidden'
|
||||
WHEN status_code = 404 OR upstream_status_code = 404 THEN 'not_found'
|
||||
WHEN status_code = 499 OR text LIKE ANY(ARRAY['%client cancelled%','%client canceled%','%context canceled%']) THEN 'client_cancelled'
|
||||
WHEN upstream_status_code >= 500 OR (error_owner = 'provider' AND status_code >= 500) THEN 'upstream_5xx'
|
||||
WHEN status_code >= 500 OR error_type = 'internal' OR error_owner = 'system' THEN 'internal'
|
||||
ELSE 'other' END AS category
|
||||
FROM dedup
|
||||
), metric_rows AS (
|
||||
INSERT INTO channel_monitor_v2_metrics_1m (bucket_start, platform, group_id, model, error_requests, upstream_affected_requests, upstream_attempt_count, computed_at)
|
||||
SELECT bucket_start, platform, group_id, model, COUNT(*), COUNT(*) FILTER (WHERE upstream_affected), SUM(upstream_attempts), NOW()
|
||||
FROM classified GROUP BY 1,2,3,4
|
||||
ON CONFLICT (bucket_start, platform, group_id, model) DO UPDATE SET
|
||||
error_requests = EXCLUDED.error_requests, upstream_affected_requests = EXCLUDED.upstream_affected_requests,
|
||||
upstream_attempt_count = EXCLUDED.upstream_attempt_count, computed_at = NOW()
|
||||
), user_rows AS (
|
||||
INSERT INTO channel_monitor_v2_user_metrics_1m (bucket_start, platform, group_id, model, user_id, error_requests, computed_at)
|
||||
SELECT bucket_start, platform, group_id, model, user_id, COUNT(*), NOW()
|
||||
FROM classified WHERE user_id IS NOT NULL GROUP BY 1,2,3,4,5
|
||||
ON CONFLICT (bucket_start, platform, group_id, model, user_id) DO UPDATE SET error_requests = EXCLUDED.error_requests, computed_at = NOW()
|
||||
)
|
||||
INSERT INTO channel_monitor_v2_error_metrics_1m (bucket_start, platform, group_id, model, error_category, taxonomy_version, error_requests)
|
||||
SELECT bucket_start, platform, group_id, model, category, 1, COUNT(*) FROM classified GROUP BY 1,2,3,4,5
|
||||
ON CONFLICT (bucket_start, platform, group_id, model, error_category, taxonomy_version)
|
||||
DO UPDATE SET error_requests = EXCLUDED.error_requests`
|
||||
|
||||
// Floor matches channelMonitorV2RetentionMax (90d). Keep the INTERVAL literal in
|
||||
// sync when changing channelMonitorV2RetentionRollup1d.
|
||||
const channelMonitorV2WatermarkSQL = `
|
||||
INSERT INTO channel_monitor_v2_watermarks (id, usage_coverage_start, error_coverage_start, data_through, last_successful_at, backfill_cursor, updated_at)
|
||||
VALUES (
|
||||
1,
|
||||
GREATEST($1, COALESCE((SELECT created_at FROM usage_logs ORDER BY created_at ASC LIMIT 1), $1)),
|
||||
GREATEST($1, COALESCE((SELECT created_at FROM ops_error_logs ORDER BY created_at ASC LIMIT 1), $1)),
|
||||
$2, NOW(), $1, NOW()
|
||||
)
|
||||
ON CONFLICT (id) DO UPDATE SET
|
||||
usage_coverage_start = GREATEST(
|
||||
date_trunc('minute', NOW()) - INTERVAL '90 days',
|
||||
LEAST(COALESCE(channel_monitor_v2_watermarks.usage_coverage_start, EXCLUDED.usage_coverage_start), EXCLUDED.usage_coverage_start)
|
||||
),
|
||||
error_coverage_start = GREATEST(
|
||||
date_trunc('minute', NOW()) - INTERVAL '90 days',
|
||||
LEAST(COALESCE(channel_monitor_v2_watermarks.error_coverage_start, EXCLUDED.error_coverage_start), EXCLUDED.error_coverage_start)
|
||||
),
|
||||
data_through = GREATEST(COALESCE(channel_monitor_v2_watermarks.data_through, EXCLUDED.data_through), EXCLUDED.data_through),
|
||||
last_successful_at = NOW(),
|
||||
backfill_cursor = LEAST(COALESCE(channel_monitor_v2_watermarks.backfill_cursor, EXCLUDED.backfill_cursor), EXCLUDED.backfill_cursor),
|
||||
updated_at = NOW()`
|
||||
|
||||
var channelMonitorV2FixedRollupSeconds = []int{300, 3600, 43200, 86400}
|
||||
|
||||
func (r *channelMonitorV2Repository) recomputeFixedRollups(ctx context.Context, tx *sql.Tx, start, end time.Time) error {
|
||||
for _, seconds := range channelMonitorV2FixedRollupSeconds {
|
||||
interval := fmt.Sprintf("%d seconds", seconds)
|
||||
for _, table := range []string{
|
||||
"channel_monitor_v2_latency_histograms_rollup",
|
||||
"channel_monitor_v2_error_metrics_rollup",
|
||||
"channel_monitor_v2_user_metrics_rollup",
|
||||
"channel_monitor_v2_metrics_rollup",
|
||||
} {
|
||||
if _, err := tx.ExecContext(ctx, fmt.Sprintf(channelMonitorV2FixedRollupDeleteSQL, table), interval, seconds, start, end); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, channelMonitorV2MetricsRollupSQL, interval, seconds, start, end); err != nil {
|
||||
return fmt.Errorf("roll up channel monitor v2 metrics %ds: %w", seconds, err)
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, channelMonitorV2UserMetricsRollupSQL, interval, seconds, start, end); err != nil {
|
||||
return fmt.Errorf("roll up channel monitor v2 user metrics %ds: %w", seconds, err)
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, channelMonitorV2HistogramRollupSQL, interval, seconds, start, end); err != nil {
|
||||
return fmt.Errorf("roll up channel monitor v2 histograms %ds: %w", seconds, err)
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, channelMonitorV2ErrorRollupSQL, interval, seconds, start, end); err != nil {
|
||||
return fmt.Errorf("roll up channel monitor v2 errors %ds: %w", seconds, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
const channelMonitorV2FixedRollupBoundsSQL = `
|
||||
WITH bounds AS (
|
||||
SELECT
|
||||
date_bin($1::interval, $3::timestamptz, TIMESTAMPTZ '1970-01-01') AS start_at,
|
||||
date_bin($1::interval, $4::timestamptz - INTERVAL '1 microsecond', TIMESTAMPTZ '1970-01-01') + $1::interval AS end_at
|
||||
)`
|
||||
|
||||
const channelMonitorV2FixedRollupDeleteSQL = channelMonitorV2FixedRollupBoundsSQL + `
|
||||
DELETE FROM %s
|
||||
USING bounds
|
||||
WHERE bucket_seconds = $2::integer
|
||||
AND bucket_start >= bounds.start_at
|
||||
AND bucket_start < bounds.end_at`
|
||||
|
||||
const channelMonitorV2MetricsRollupSQL = `
|
||||
INSERT INTO channel_monitor_v2_metrics_rollup (
|
||||
bucket_start, bucket_seconds, platform, group_id, model, success_requests, error_requests,
|
||||
upstream_affected_requests, upstream_attempt_count, input_tokens, output_tokens,
|
||||
cache_creation_tokens, cache_read_tokens, ttft_sum_ms, ttft_count, duration_sum_ms,
|
||||
duration_count, computed_at
|
||||
)
|
||||
` + channelMonitorV2FixedRollupBoundsSQL + `
|
||||
SELECT date_bin($1::interval, m.bucket_start, TIMESTAMPTZ '1970-01-01'), $2::integer,
|
||||
platform, group_id, model, SUM(success_requests), SUM(error_requests),
|
||||
SUM(upstream_affected_requests), SUM(upstream_attempt_count), SUM(input_tokens),
|
||||
SUM(output_tokens), SUM(cache_creation_tokens), SUM(cache_read_tokens),
|
||||
SUM(ttft_sum_ms), SUM(ttft_count), SUM(duration_sum_ms), SUM(duration_count), NOW()
|
||||
FROM channel_monitor_v2_metrics_1m m, bounds
|
||||
WHERE m.bucket_start >= bounds.start_at AND m.bucket_start < bounds.end_at
|
||||
GROUP BY 1, 2, 3, 4, 5`
|
||||
|
||||
const channelMonitorV2UserMetricsRollupSQL = `
|
||||
INSERT INTO channel_monitor_v2_user_metrics_rollup (
|
||||
bucket_start, bucket_seconds, platform, group_id, model, user_id, success_requests,
|
||||
error_requests, input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens,
|
||||
ttft_sum_ms, ttft_count, duration_sum_ms, duration_count, computed_at
|
||||
)
|
||||
` + channelMonitorV2FixedRollupBoundsSQL + `
|
||||
SELECT date_bin($1::interval, m.bucket_start, TIMESTAMPTZ '1970-01-01'), $2::integer,
|
||||
platform, group_id, model, user_id, SUM(success_requests), SUM(error_requests),
|
||||
SUM(input_tokens), SUM(output_tokens), SUM(cache_creation_tokens), SUM(cache_read_tokens),
|
||||
SUM(ttft_sum_ms), SUM(ttft_count), SUM(duration_sum_ms), SUM(duration_count), NOW()
|
||||
FROM channel_monitor_v2_user_metrics_1m m, bounds
|
||||
WHERE m.bucket_start >= bounds.start_at AND m.bucket_start < bounds.end_at
|
||||
GROUP BY 1, 2, 3, 4, 5, 6`
|
||||
|
||||
const channelMonitorV2HistogramRollupSQL = `
|
||||
INSERT INTO channel_monitor_v2_latency_histograms_rollup (
|
||||
bucket_start, bucket_seconds, platform, group_id, model, user_id, metric, upper_bound_ms, sample_count
|
||||
)
|
||||
` + channelMonitorV2FixedRollupBoundsSQL + `
|
||||
SELECT date_bin($1::interval, h.bucket_start, TIMESTAMPTZ '1970-01-01'), $2::integer,
|
||||
platform, group_id, model, user_id, metric, upper_bound_ms, SUM(sample_count)
|
||||
FROM channel_monitor_v2_latency_histograms_1m h, bounds
|
||||
WHERE h.bucket_start >= bounds.start_at AND h.bucket_start < bounds.end_at
|
||||
GROUP BY 1, 2, 3, 4, 5, 6, 7, 8`
|
||||
|
||||
const channelMonitorV2ErrorRollupSQL = `
|
||||
INSERT INTO channel_monitor_v2_error_metrics_rollup (
|
||||
bucket_start, bucket_seconds, platform, group_id, model, error_category, taxonomy_version, error_requests
|
||||
)
|
||||
` + channelMonitorV2FixedRollupBoundsSQL + `
|
||||
SELECT date_bin($1::interval, e.bucket_start, TIMESTAMPTZ '1970-01-01'), $2::integer,
|
||||
platform, group_id, model, error_category, taxonomy_version, SUM(error_requests)
|
||||
FROM channel_monitor_v2_error_metrics_1m e, bounds
|
||||
WHERE e.bucket_start >= bounds.start_at AND e.bucket_start < bounds.end_at
|
||||
GROUP BY 1, 2, 3, 4, 5, 6, 7`
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,265 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/service"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestChannelMonitorV2DisplayModelIsPlatformScoped(t *testing.T) {
|
||||
cfg := service.ChannelMonitorV2Config{Platforms: []service.ChannelMonitorV2PlatformConfig{
|
||||
{Platform: "openai", Enabled: true, Models: []string{"shared", "gpt-5"}},
|
||||
{Platform: "grok", Enabled: true, Models: []string{"grok-4"}},
|
||||
// Empty models list must NOT collapse everything into __other__.
|
||||
{Platform: "anthropic", Enabled: true, Models: []string{}},
|
||||
}}
|
||||
require.Equal(t, "shared", channelMonitorV2DisplayModel(cfg, "openai", "shared"))
|
||||
require.Equal(t, service.ChannelMonitorV2OtherModel, channelMonitorV2DisplayModel(cfg, "grok", "shared"))
|
||||
require.Equal(t, "claude-sonnet-4", channelMonitorV2DisplayModel(cfg, "anthropic", "claude-sonnet-4"))
|
||||
// Unconfigured platform still surfaces the real model name.
|
||||
require.Equal(t, "gemini-2.5-pro", channelMonitorV2DisplayModel(cfg, "gemini", "gemini-2.5-pro"))
|
||||
require.True(t, channelMonitorV2ModelSelected(service.ChannelMonitorV2Filter{Models: []string{service.ChannelMonitorV2OtherModel}}, cfg, "grok", "shared"))
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2MatrixDimensionKey(t *testing.T) {
|
||||
cfg := service.ChannelMonitorV2Config{Platforms: []service.ChannelMonitorV2PlatformConfig{{Platform: "openai", Enabled: true, Models: []string{"gpt-5"}}}}
|
||||
key := channelMonitorV2MatrixDimensionKey(service.ChannelMonitorV2GroupByPlatformGroupModel, cfg, "openai", 7, "gpt-5")
|
||||
require.Equal(t, channelMonitorV2MatrixKey{platform: "openai", groupID: 7, model: "gpt-5"}, key)
|
||||
key = channelMonitorV2MatrixDimensionKey(service.ChannelMonitorV2GroupByPlatformModel, cfg, "openai", 7, "unlisted")
|
||||
require.Equal(t, channelMonitorV2MatrixKey{platform: "openai", model: service.ChannelMonitorV2OtherModel}, key)
|
||||
key = channelMonitorV2MatrixDimensionKey(service.ChannelMonitorV2GroupByPlatform, cfg, "openai", 7, "gpt-5")
|
||||
require.Equal(t, channelMonitorV2MatrixKey{platform: "openai"}, key)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2HistogramPercentilesAreMergedFromCounts(t *testing.T) {
|
||||
// 100 samples: 50@100, 40@500, 10@1000
|
||||
// target = int64(total*p + 0.999999) truncates: p50→50, p90→90, p95→95
|
||||
// cumulative hits: p50@100, p90@500 (50+40), p95@1000
|
||||
histogram := map[int64]int64{100: 50, 500: 40, 1000: 10}
|
||||
require.Equal(t, int64(100), *histPercentile(histogram, .5))
|
||||
require.Equal(t, int64(500), *histPercentile(histogram, .9))
|
||||
require.Equal(t, int64(1000), *histPercentile(histogram, .95))
|
||||
require.Nil(t, histPercentile(nil, .95))
|
||||
// latencyMetric exposes avg + p50 + p90 + p95
|
||||
lat := latencyMetric(1000, 10, histogram)
|
||||
require.NotNil(t, lat.AvgMs)
|
||||
require.NotNil(t, lat.P50Ms)
|
||||
require.NotNil(t, lat.P90Ms)
|
||||
require.NotNil(t, lat.P95Ms)
|
||||
require.Equal(t, int64(100), *lat.P50Ms)
|
||||
require.Equal(t, int64(500), *lat.P90Ms)
|
||||
require.Equal(t, int64(1000), *lat.P95Ms)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2MetricIncludesSuccessRate(t *testing.T) {
|
||||
acc := newMetricAccumulator()
|
||||
acc.success, acc.errors = 80, 20
|
||||
metric := acc.metric(1, false)
|
||||
require.Equal(t, int64(100), metric.RequestCount)
|
||||
require.InDelta(t, 0.8, metric.SuccessRate, 0.0001)
|
||||
require.InDelta(t, 0.2, metric.ErrorRate, 0.0001)
|
||||
require.Nil(t, metric.UpstreamAffectedRequests)
|
||||
|
||||
adminMetric := acc.metric(1, true)
|
||||
require.NotNil(t, adminMetric.UpstreamAffectedRequests)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2WhereUsesConfiguredScopeAndEmptyFilterMeansAllConfigured(t *testing.T) {
|
||||
filter := service.ChannelMonitorV2Filter{Start: time.Unix(1, 0), End: time.Unix(2, 0)}
|
||||
cfg := service.ChannelMonitorV2Config{
|
||||
Platforms: []service.ChannelMonitorV2PlatformConfig{{Platform: "openai", Enabled: true}, {Platform: "grok", Enabled: false}},
|
||||
GroupIDs: []int64{3, 4},
|
||||
}
|
||||
where, args := channelMonitorV2Where(filter, cfg, "m")
|
||||
require.Contains(t, where, "m.platform = ANY($3)")
|
||||
require.Contains(t, where, "m.group_id = ANY($4)")
|
||||
require.Len(t, args, 4)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2WhereRejectsGroupFilterOutsideConfiguredScope(t *testing.T) {
|
||||
filter := service.ChannelMonitorV2Filter{
|
||||
Start: time.Unix(1, 0), End: time.Unix(2, 0), GroupIDs: []int64{9},
|
||||
}
|
||||
cfg := service.ChannelMonitorV2Config{
|
||||
Platforms: []service.ChannelMonitorV2PlatformConfig{{Platform: "openai", Enabled: true}},
|
||||
GroupIDs: []int64{3, 4},
|
||||
}
|
||||
where, args := channelMonitorV2Where(filter, cfg, "m")
|
||||
require.Contains(t, where, "FALSE")
|
||||
require.NotContains(t, where, "m.group_id = ANY")
|
||||
require.Len(t, args, 3)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2ErrorAggregationCountsFinalUserErrorsOnly(t *testing.T) {
|
||||
query := strings.ToLower(channelMonitorV2ErrorAggregationSQL)
|
||||
require.Contains(t, query, "not current_error.is_count_tokens")
|
||||
require.Contains(t, query, "error_type = 'cyber_policy'")
|
||||
require.Contains(t, query, "distinct on")
|
||||
require.Contains(t, query, "not exists")
|
||||
require.Contains(t, query, "upstream_affected_requests")
|
||||
require.Contains(t, query, "jsonb_array_length(upstream_errors) > 0")
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2UsageSuccessExcludesCyberBillingRows(t *testing.T) {
|
||||
for _, query := range []string{channelMonitorV2UsageMetricsSQL, channelMonitorV2UserMetricsSQL} {
|
||||
require.Contains(t, query, "COALESCE(ul.request_type, 0) NOT IN (4, 6)")
|
||||
}
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2RatesUseCoveredWindow(t *testing.T) {
|
||||
start := time.Date(2026, 8, 1, 0, 0, 0, 0, time.UTC)
|
||||
filter := service.ChannelMonitorV2Filter{Start: start, End: start.Add(24 * time.Hour)}
|
||||
coverage := service.ChannelMonitorV2Coverage{CoverageStart: start.Add(6 * time.Hour), DataThrough: start.Add(18 * time.Hour)}
|
||||
require.Equal(t, 12*60.0, channelMonitorV2CoveredMinutes(filter, coverage))
|
||||
effective := channelMonitorV2CommonCoverageFilter(filter, coverage)
|
||||
require.Equal(t, coverage.CoverageStart, effective.Start)
|
||||
require.Equal(t, coverage.DataThrough, effective.End)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2TierRetentionPolicy(t *testing.T) {
|
||||
require.Equal(t, 3*24*time.Hour, channelMonitorV2RetentionUser1m)
|
||||
require.Equal(t, 7*24*time.Hour, channelMonitorV2RetentionMetrics1m)
|
||||
require.Equal(t, 7*24*time.Hour, channelMonitorV2RetentionError1m)
|
||||
require.Equal(t, 7*24*time.Hour, channelMonitorV2RetentionHistogram1m)
|
||||
require.Equal(t, 7*24*time.Hour, channelMonitorV2RetentionRollup5m)
|
||||
require.Equal(t, 30*24*time.Hour, channelMonitorV2RetentionRollup1h)
|
||||
require.Equal(t, 45*24*time.Hour, channelMonitorV2RetentionRollup12h)
|
||||
require.Equal(t, 90*24*time.Hour, channelMonitorV2RetentionRollup1d)
|
||||
require.Equal(t, channelMonitorV2RetentionRollup1d, channelMonitorV2MaxRetention())
|
||||
require.Contains(t, channelMonitorV2WatermarkSQL, "INTERVAL '90 days'")
|
||||
|
||||
// Every fixed rollup second must appear with a retention rule.
|
||||
wantSeconds := map[int]time.Duration{
|
||||
300: channelMonitorV2RetentionRollup5m,
|
||||
3600: channelMonitorV2RetentionRollup1h,
|
||||
43200: channelMonitorV2RetentionRollup12h,
|
||||
86400: channelMonitorV2RetentionRollup1d,
|
||||
}
|
||||
seen := map[int]time.Duration{}
|
||||
for _, rule := range channelMonitorV2RetentionRules {
|
||||
if rule.bucketSeconds == 0 {
|
||||
require.True(t, rule.retention > 0)
|
||||
continue
|
||||
}
|
||||
if prev, ok := seen[rule.bucketSeconds]; ok {
|
||||
require.Equal(t, prev, rule.retention)
|
||||
}
|
||||
seen[rule.bucketSeconds] = rule.retention
|
||||
}
|
||||
for seconds, want := range wantSeconds {
|
||||
got, ok := seen[seconds]
|
||||
require.Truef(t, ok, "missing retention rule for bucket_seconds=%d", seconds)
|
||||
require.Equal(t, want, got)
|
||||
}
|
||||
|
||||
now := time.Date(2026, 8, 7, 12, 0, 0, 0, time.UTC)
|
||||
require.Equal(t, now.Add(-7*24*time.Hour), channelMonitorV2RetentionCutoff(now, channelMonitorV2RetentionMetrics1m))
|
||||
require.Equal(t, now.Add(-90*24*time.Hour), channelMonitorV2RetentionCutoff(now, channelMonitorV2MaxRetention()))
|
||||
}
|
||||
|
||||
// Needles present in service.ClassifyChannelMonitorV2Error must appear in the
|
||||
// aggregation SQL CASE so rollup categories match drilldown classification.
|
||||
func TestChannelMonitorV2SQLTaxonomyContainsGoNeedles(t *testing.T) {
|
||||
sql := channelMonitorV2ErrorAggregationSQL
|
||||
needles := []string{
|
||||
"blocked keyword",
|
||||
"invalid_api_key",
|
||||
"max_tokens",
|
||||
"invalid_request",
|
||||
"model not supported",
|
||||
"billing hard limit",
|
||||
"no healthy upstream account",
|
||||
"rate_limit",
|
||||
"gateway timeout",
|
||||
"connection refused",
|
||||
"unexpected eof",
|
||||
}
|
||||
for _, needle := range needles {
|
||||
require.Containsf(t, strings.ToLower(sql), strings.ToLower(needle), "SQL taxonomy missing Go needle %q", needle)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyIgnoredErrorsAdjustsRatesKeepsAbsoluteVolume(t *testing.T) {
|
||||
m := service.ChannelMonitorV2Metric{
|
||||
RequestCount: 100,
|
||||
ErrorRequests: 20,
|
||||
ErrorRate: 0.20,
|
||||
SuccessRate: 0.80,
|
||||
}
|
||||
// Success absolute still 80 → success rate stays 0.80 even after ignoring 5 errors.
|
||||
m.SuccessRequests = 80
|
||||
applyIgnoredErrors(&m, 5)
|
||||
require.Equal(t, int64(100), m.RequestCount)
|
||||
require.Equal(t, int64(20), m.ErrorRequests)
|
||||
require.InDelta(t, 0.15, m.ErrorRate, 0.0001)
|
||||
require.InDelta(t, 0.80, m.SuccessRate, 0.0001)
|
||||
|
||||
// Clamp ignored > errors: scored error_rate → 0; success stays true ratio.
|
||||
m2 := service.ChannelMonitorV2Metric{RequestCount: 10, ErrorRequests: 2, SuccessRequests: 8, ErrorRate: 0.2, SuccessRate: 0.8}
|
||||
applyIgnoredErrors(&m2, 99)
|
||||
require.InDelta(t, 0.0, m2.ErrorRate, 0.0001)
|
||||
require.InDelta(t, 0.8, m2.SuccessRate, 0.0001)
|
||||
|
||||
// No-op when ignored is zero
|
||||
m3 := service.ChannelMonitorV2Metric{RequestCount: 10, ErrorRequests: 2, SuccessRequests: 8, ErrorRate: 0.2, SuccessRate: 0.8}
|
||||
applyIgnoredErrors(&m3, 0)
|
||||
require.InDelta(t, 0.2, m3.ErrorRate, 0.0001)
|
||||
require.InDelta(t, 0.8, m3.SuccessRate, 0.0001)
|
||||
}
|
||||
|
||||
func TestRedactChannelMonitorV2MetricZerosVolume(t *testing.T) {
|
||||
// Service helper is in service package; covered there. Keep a smoke note that
|
||||
// rates survive a manual zeroing of volume fields used by the UI contract.
|
||||
m := service.ChannelMonitorV2Metric{
|
||||
RequestCount: 100, ErrorRequests: 10, SuccessRequests: 90,
|
||||
TokenCount: 1000, RPM: 5, TPM: 50, ErrorRate: 0.1, SuccessRate: 0.9, CacheRate: 0.4,
|
||||
}
|
||||
// Mimic redact: zero volume only
|
||||
m.RequestCount, m.ErrorRequests, m.SuccessRequests, m.TokenCount = 0, 0, 0, 0
|
||||
require.Equal(t, 0.1, m.ErrorRate)
|
||||
require.Equal(t, 5.0, m.RPM)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2CatalogFilterClearsMultiSelectDimensions(t *testing.T) {
|
||||
start := time.Unix(1, 0)
|
||||
end := time.Unix(2, 0)
|
||||
filter := service.ChannelMonitorV2Filter{
|
||||
Start: start, End: end, Bucket: time.Minute,
|
||||
Platforms: []string{"openai"}, GroupIDs: []int64{3}, Models: []string{"gpt-5"},
|
||||
}
|
||||
catalog := channelMonitorV2CatalogFilter(filter)
|
||||
require.Nil(t, catalog.Platforms)
|
||||
require.Nil(t, catalog.GroupIDs)
|
||||
require.Nil(t, catalog.Models)
|
||||
// Time window / coverage-related fields remain.
|
||||
require.Equal(t, start, catalog.Start)
|
||||
require.Equal(t, end, catalog.End)
|
||||
require.Equal(t, time.Minute, catalog.Bucket)
|
||||
|
||||
cfg := service.ChannelMonitorV2Config{
|
||||
Platforms: []service.ChannelMonitorV2PlatformConfig{
|
||||
{Platform: "openai", Enabled: true},
|
||||
{Platform: "grok", Enabled: true},
|
||||
},
|
||||
GroupIDs: []int64{3, 4},
|
||||
}
|
||||
catalogWhere, catalogArgs := channelMonitorV2Where(catalog, cfg, "m")
|
||||
_, metricArgs := channelMonitorV2Where(filter, cfg, "m")
|
||||
|
||||
// Catalog WHERE still applies config scope (enabled platforms + group allow-list).
|
||||
require.Contains(t, catalogWhere, "m.platform = ANY")
|
||||
require.Contains(t, catalogWhere, "m.group_id = ANY")
|
||||
require.Len(t, catalogArgs, 4) // start, end, platforms, groups
|
||||
require.Len(t, metricArgs, 4)
|
||||
|
||||
// Metrics WHERE is narrower once multi-select platforms/groups are applied.
|
||||
require.NotEqual(t, catalogArgs, metricArgs)
|
||||
// Group seeding without multi-select uses full config allow-list.
|
||||
require.Equal(t, []int64{3, 4}, configuredChannelMonitorV2GroupIDs(catalog, cfg))
|
||||
require.Equal(t, []int64{3}, configuredChannelMonitorV2GroupIDs(filter, cfg))
|
||||
}
|
||||
|
||||
|
||||
@@ -157,3 +157,18 @@ var (
|
||||
"CHANNEL_MONITOR_KEY_DECRYPT_FAILED", "api key decryption failed; please re-edit the monitor with a fresh key",
|
||||
)
|
||||
)
|
||||
|
||||
var (
|
||||
ErrChannelMonitorDisabled = infraerrors.Forbidden(
|
||||
"CHANNEL_MONITOR_DISABLED",
|
||||
"channel monitor feature is disabled",
|
||||
)
|
||||
ErrChannelMonitorActiveProbesRetired = infraerrors.Forbidden(
|
||||
"CHANNEL_MONITOR_ACTIVE_PROBES_RETIRED",
|
||||
"channel monitor active probes are retired in v2 mode",
|
||||
)
|
||||
ErrChannelMonitorModeMismatch = infraerrors.Forbidden(
|
||||
"CHANNEL_MONITOR_MODE_MISMATCH",
|
||||
"channel monitor mode does not allow this operation",
|
||||
)
|
||||
)
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
//go:build unit
|
||||
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
type channelMonitorRuntimeStub struct {
|
||||
rt ChannelMonitorRuntime
|
||||
}
|
||||
|
||||
func (s channelMonitorRuntimeStub) GetChannelMonitorRuntime(context.Context) ChannelMonitorRuntime {
|
||||
return s.rt
|
||||
}
|
||||
|
||||
func TestRunCheck_ModeV2NeverProbes(t *testing.T) {
|
||||
svc := NewChannelMonitorService(nil, nil)
|
||||
svc.SetRuntimeReader(channelMonitorRuntimeStub{rt: ChannelMonitorRuntime{
|
||||
Enabled: true,
|
||||
Mode: ChannelMonitorModeV2,
|
||||
}})
|
||||
|
||||
results, err := svc.RunCheck(context.Background(), 1)
|
||||
require.ErrorIs(t, err, ErrChannelMonitorActiveProbesRetired)
|
||||
require.Nil(t, results)
|
||||
}
|
||||
|
||||
func TestRunCheck_DisabledReturnsDisabled(t *testing.T) {
|
||||
svc := NewChannelMonitorService(nil, nil)
|
||||
svc.SetRuntimeReader(channelMonitorRuntimeStub{rt: ChannelMonitorRuntime{
|
||||
Enabled: false,
|
||||
Mode: ChannelMonitorModeV1,
|
||||
}})
|
||||
|
||||
_, err := svc.RunCheck(context.Background(), 1)
|
||||
require.ErrorIs(t, err, ErrChannelMonitorDisabled)
|
||||
}
|
||||
|
||||
func TestRunCheck_NilRuntimeReaderFailsClosedAsV2(t *testing.T) {
|
||||
svc := NewChannelMonitorService(nil, nil)
|
||||
// No SetRuntimeReader → probeRuntime defaults to mode=v2 (retired).
|
||||
_, err := svc.RunCheck(context.Background(), 1)
|
||||
require.ErrorIs(t, err, ErrChannelMonitorActiveProbesRetired)
|
||||
}
|
||||
|
||||
func TestNormalizeChannelMonitorMode(t *testing.T) {
|
||||
require.Equal(t, ChannelMonitorModeV2, normalizeChannelMonitorMode(""))
|
||||
require.Equal(t, ChannelMonitorModeV1, normalizeChannelMonitorMode("v1"))
|
||||
require.Equal(t, ChannelMonitorModeV2, normalizeChannelMonitorMode("v2"))
|
||||
require.Equal(t, ChannelMonitorModeV2, normalizeChannelMonitorMode("invalid"))
|
||||
require.Equal(t, ChannelMonitorModeV1, normalizeChannelMonitorMode(" V1 "))
|
||||
}
|
||||
|
||||
func TestChannelMonitorRuntimeActiveProbesAllowed(t *testing.T) {
|
||||
require.False(t, (ChannelMonitorRuntime{Enabled: false, Mode: ChannelMonitorModeV1}).ActiveProbesAllowed())
|
||||
require.True(t, (ChannelMonitorRuntime{Enabled: true, Mode: ChannelMonitorModeV1}).ActiveProbesAllowed())
|
||||
require.False(t, (ChannelMonitorRuntime{Enabled: true, Mode: ChannelMonitorModeV2}).ActiveProbesAllowed())
|
||||
require.True(t, (ChannelMonitorRuntime{Enabled: true, Mode: ChannelMonitorModeV2}).PassiveAggregationAllowed())
|
||||
}
|
||||
@@ -257,8 +257,11 @@ func (r *ChannelMonitorRunner) runScheduled(ctx context.Context, task *scheduled
|
||||
// fire 提交一次检测到 worker 池。功能开关关闭时跳过本次（不取消任务，
|
||||
// 重新启用时立即恢复）；池满或重复在飞时也跳过。
|
||||
func (r *ChannelMonitorRunner) fire(ctx context.Context, task *scheduledMonitor) {
|
||||
if r.settingService != nil && !r.settingService.GetChannelMonitorRuntime(ctx).Enabled {
|
||||
return
|
||||
if r.settingService != nil {
|
||||
rt := r.settingService.GetChannelMonitorRuntime(ctx)
|
||||
if !rt.ActiveProbesAllowed() {
|
||||
return
|
||||
}
|
||||
}
|
||||
if !r.tryAcquireInFlight(task.id) {
|
||||
slog.Debug("channel_monitor: skip already in-flight",
|
||||
|
||||
@@ -61,10 +61,19 @@ type ChannelMonitorRepository interface {
|
||||
UpdateAggregationWatermark(ctx context.Context, date time.Time) error
|
||||
}
|
||||
|
||||
// channelMonitorRuntimeReader is the optional settings view used to gate V1
|
||||
// active probes by channel_monitor_enabled + channel_monitor_mode.
|
||||
type channelMonitorRuntimeReader interface {
|
||||
GetChannelMonitorRuntime(ctx context.Context) ChannelMonitorRuntime
|
||||
}
|
||||
|
||||
// ChannelMonitorService 渠道监控管理服务。
|
||||
type ChannelMonitorService struct {
|
||||
repo ChannelMonitorRepository
|
||||
encryptor SecretEncryptor
|
||||
// settings is optional; when nil, RunCheck fails closed for active probes
|
||||
// (mode defaults to v2 / retired) so tests without settings never hit upstream.
|
||||
settings channelMonitorRuntimeReader
|
||||
// scheduler 由 wire 通过 SetScheduler 注入；CRUD 后调用对应钩子即时同步任务。
|
||||
// 测试或未注入场景下保持 nil，所有钩子调用变为 no-op。
|
||||
scheduler MonitorScheduler
|
||||
@@ -83,6 +92,22 @@ func NewChannelMonitorService(repo ChannelMonitorRepository, encryptor SecretEnc
|
||||
return &ChannelMonitorService{repo: repo, encryptor: encryptor}
|
||||
}
|
||||
|
||||
// SetRuntimeReader injects the settings reader used to gate active probes.
|
||||
// Optional: when unset, active probes are treated as mode=v2 (retired).
|
||||
func (s *ChannelMonitorService) SetRuntimeReader(r channelMonitorRuntimeReader) {
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
s.settings = r
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorService) probeRuntime(ctx context.Context) ChannelMonitorRuntime {
|
||||
if s == nil || s.settings == nil {
|
||||
return ChannelMonitorRuntime{Enabled: true, Mode: ChannelMonitorModeV2}
|
||||
}
|
||||
return s.settings.GetChannelMonitorRuntime(ctx)
|
||||
}
|
||||
|
||||
// ---------- CRUD ----------
|
||||
|
||||
// List 列表查询（支持 provider/enabled/search 过滤 + 分页）。
|
||||
@@ -427,7 +452,16 @@ func (s *ChannelMonitorService) ListHistory(ctx context.Context, id int64, model
|
||||
|
||||
// RunCheck 同步触发对一个监控的检测：并发跑 primary + extra 模型，
|
||||
// 写历史记录并更新 last_checked_at。返回每个模型的检测结果。
|
||||
// 仅当 channel_monitor_enabled=true 且 channel_monitor_mode=v1 时真正探测；
|
||||
// mode=v2 时返回 ErrChannelMonitorActiveProbesRetired，不产生上游流量。
|
||||
func (s *ChannelMonitorService) RunCheck(ctx context.Context, id int64) ([]*CheckResult, error) {
|
||||
rt := s.probeRuntime(ctx)
|
||||
if !rt.Enabled {
|
||||
return nil, ErrChannelMonitorDisabled
|
||||
}
|
||||
if !rt.ActiveProbesAllowed() {
|
||||
return nil, ErrChannelMonitorActiveProbesRetired
|
||||
}
|
||||
m, err := s.Get(ctx, id) // 已解密 APIKey
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -0,0 +1,969 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
ChannelMonitorV2OtherModel = "__other__"
|
||||
ChannelMonitorV2TaxonomyVersion = 1
|
||||
)
|
||||
|
||||
var (
|
||||
ErrChannelMonitorV2InvalidRange = errors.New("invalid channel monitor v2 range")
|
||||
ErrChannelMonitorV2InvalidGroupBy = errors.New("invalid channel monitor v2 group_by")
|
||||
ErrChannelMonitorV2InvalidConfig = errors.New("invalid channel monitor v2 config")
|
||||
ErrChannelMonitorV2ConfigConflict = errors.New("channel monitor v2 config was modified")
|
||||
)
|
||||
|
||||
type ChannelMonitorV2GroupBy string
|
||||
|
||||
const (
|
||||
ChannelMonitorV2GroupByPlatform ChannelMonitorV2GroupBy = "platform"
|
||||
ChannelMonitorV2GroupByPlatformGroup ChannelMonitorV2GroupBy = "platform_group"
|
||||
ChannelMonitorV2GroupByPlatformModel ChannelMonitorV2GroupBy = "platform_model"
|
||||
ChannelMonitorV2GroupByPlatformGroupModel ChannelMonitorV2GroupBy = "platform_group_model"
|
||||
)
|
||||
|
||||
type ChannelMonitorV2PlatformConfig struct {
|
||||
Platform string `json:"platform"`
|
||||
Enabled bool `json:"enabled"`
|
||||
Models []string `json:"models"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2Config struct {
|
||||
Version int `json:"version"`
|
||||
Enabled bool `json:"enabled"`
|
||||
RefreshIntervalSeconds int `json:"refresh_interval_seconds"`
|
||||
Platforms []ChannelMonitorV2PlatformConfig `json:"platforms"`
|
||||
GroupIDs []int64 `json:"group_ids"`
|
||||
HealthThresholds ChannelMonitorV2HealthThresholds `json:"health_thresholds"`
|
||||
// IgnoredErrorCategories are excluded from error_rate / health scoring.
|
||||
// They still appear in the error breakdown with ignored=true (greyed in UI).
|
||||
// Unknown categories always roll into "other" via the taxonomy classifier.
|
||||
IgnoredErrorCategories []string `json:"ignored_error_categories"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
UpdatedBy *int64 `json:"updated_by,omitempty"`
|
||||
}
|
||||
|
||||
// ChannelMonitorV2ErrorCategories is the ordered, versioned taxonomy used by the
|
||||
// classifier and admin settings UI. Unmatched errors become "other".
|
||||
var ChannelMonitorV2ErrorCategories = []string{
|
||||
"content_policy",
|
||||
"authentication",
|
||||
"context_limit",
|
||||
"invalid_request",
|
||||
"model_unsupported",
|
||||
"group_access",
|
||||
"quota_or_balance",
|
||||
"account_pool_unavailable",
|
||||
"rate_or_capacity",
|
||||
"timeout",
|
||||
"transport_or_stream",
|
||||
"upstream_forbidden",
|
||||
"not_found",
|
||||
"client_cancelled",
|
||||
"upstream_5xx",
|
||||
"internal",
|
||||
"other",
|
||||
}
|
||||
|
||||
type ChannelMonitorV2Filter struct {
|
||||
Range string
|
||||
Platforms []string
|
||||
GroupIDs []int64
|
||||
Models []string
|
||||
Start time.Time
|
||||
End time.Time
|
||||
Bucket time.Duration
|
||||
}
|
||||
|
||||
type ChannelMonitorV2Metric struct {
|
||||
SuccessRequests int64 `json:"success_requests"`
|
||||
ErrorRequests int64 `json:"error_requests"`
|
||||
RequestCount int64 `json:"request_count"`
|
||||
InputTokens int64 `json:"input_tokens"`
|
||||
OutputTokens int64 `json:"output_tokens"`
|
||||
CacheCreationTokens int64 `json:"cache_creation_tokens"`
|
||||
CacheReadTokens int64 `json:"cache_read_tokens"`
|
||||
TokenCount int64 `json:"token_count"`
|
||||
RPM float64 `json:"rpm"`
|
||||
TPM float64 `json:"tpm"`
|
||||
ErrorRate float64 `json:"error_rate"`
|
||||
SuccessRate float64 `json:"success_rate"`
|
||||
CacheRate float64 `json:"cache_rate"`
|
||||
CacheRateNumerator int64 `json:"cache_rate_numerator"`
|
||||
CacheRateDenominator int64 `json:"cache_rate_denominator"`
|
||||
TTFT ChannelMonitorV2Latency `json:"ttft"`
|
||||
Duration ChannelMonitorV2Latency `json:"duration"`
|
||||
UpstreamAffectedRequests *int64 `json:"upstream_affected_requests,omitempty"`
|
||||
UpstreamAttemptCount *int64 `json:"upstream_attempt_count,omitempty"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2Latency struct {
|
||||
SampleCount int64 `json:"sample_count"`
|
||||
P50Ms *int64 `json:"p50_ms"`
|
||||
P90Ms *int64 `json:"p90_ms"`
|
||||
P95Ms *int64 `json:"p95_ms"`
|
||||
AvgMs *float64 `json:"avg_ms"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2Health struct {
|
||||
Overall string `json:"overall"`
|
||||
ErrorRate string `json:"error_rate"`
|
||||
TTFT string `json:"ttft"`
|
||||
Cache string `json:"cache"`
|
||||
// Score is 0–100 when samples are sufficient; omitted/null when unknown.
|
||||
// Overall blends error-rate, TTFT p50, and cache rate (weights in Thresholds).
|
||||
Score *float64 `json:"score,omitempty"`
|
||||
ErrorRateScore *float64 `json:"error_rate_score,omitempty"`
|
||||
TTFTScore *float64 `json:"ttft_score,omitempty"`
|
||||
CacheScore *float64 `json:"cache_score,omitempty"`
|
||||
MinimumSample int64 `json:"minimum_sample"`
|
||||
Thresholds ChannelMonitorV2HealthThresholds `json:"thresholds"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2HealthThresholds struct {
|
||||
// MinimumSample is required before scoring request/latency/cache signals.
|
||||
MinimumSample int64 `json:"minimum_sample"`
|
||||
// WarningErrorRate / CriticalErrorRate map to discrete bands for legacy UI.
|
||||
WarningErrorRate float64 `json:"warning_error_rate"`
|
||||
CriticalErrorRate float64 `json:"critical_error_rate"`
|
||||
// TargetTTFTMs is the primary TTFT p50 budget (ms). At or below this is full TTFT score.
|
||||
TargetTTFTMs int64 `json:"target_ttft_ms"`
|
||||
WarningTTFTMs int64 `json:"warning_ttft_ms"`
|
||||
CriticalTTFTMs int64 `json:"critical_ttft_ms"`
|
||||
// WarningCacheRate / CriticalCacheRate: cache rate below these → warning/critical bands.
|
||||
// Higher cache rate is better; defaults 20% warning / 5% critical.
|
||||
WarningCacheRate float64 `json:"warning_cache_rate"`
|
||||
CriticalCacheRate float64 `json:"critical_cache_rate"`
|
||||
// ErrorWeight + TTFTWeight + CacheWeight should sum to 1.0.
|
||||
ErrorWeight float64 `json:"error_weight"`
|
||||
TTFTWeight float64 `json:"ttft_weight"`
|
||||
CacheWeight float64 `json:"cache_weight"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2Coverage struct {
|
||||
RequestedStart time.Time `json:"requested_start"`
|
||||
CoverageStart time.Time `json:"coverage_start"`
|
||||
DataThrough time.Time `json:"data_through"`
|
||||
ComputedAt time.Time `json:"computed_at"`
|
||||
AggregationLagSeconds int64 `json:"aggregation_lag_seconds"`
|
||||
CoverageComplete bool `json:"coverage_complete"`
|
||||
BucketSeconds int `json:"bucket_seconds"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2TrendPoint struct {
|
||||
BucketStart time.Time `json:"bucket_start"`
|
||||
Metrics ChannelMonitorV2Metric `json:"metrics"`
|
||||
Health ChannelMonitorV2Health `json:"health"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2Snapshot struct {
|
||||
Config ChannelMonitorV2Config `json:"config"`
|
||||
Coverage ChannelMonitorV2Coverage `json:"coverage"`
|
||||
Metrics ChannelMonitorV2Metric `json:"metrics"`
|
||||
Health ChannelMonitorV2Health `json:"health"`
|
||||
Trend []ChannelMonitorV2TrendPoint `json:"trend"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2Dimension struct {
|
||||
Value string `json:"value"`
|
||||
Label string `json:"label"`
|
||||
Platform string `json:"platform,omitempty"`
|
||||
RequestCount int64 `json:"request_count"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2GroupDimension struct {
|
||||
ID int64 `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Platform string `json:"platform,omitempty"`
|
||||
RequestCount int64 `json:"request_count"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2Dimensions struct {
|
||||
Platforms []ChannelMonitorV2Dimension `json:"platforms"`
|
||||
Groups []ChannelMonitorV2GroupDimension `json:"groups"`
|
||||
Models []ChannelMonitorV2Dimension `json:"models"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2ModelRow struct {
|
||||
Platform string `json:"platform"`
|
||||
Model string `json:"model"`
|
||||
Metrics ChannelMonitorV2Metric `json:"metrics"`
|
||||
Health ChannelMonitorV2Health `json:"health"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2MatrixRow struct {
|
||||
Platform string `json:"platform"`
|
||||
GroupID *int64 `json:"group_id,omitempty"`
|
||||
GroupName string `json:"group_name,omitempty"`
|
||||
Model string `json:"model,omitempty"`
|
||||
Metrics ChannelMonitorV2Metric `json:"metrics"`
|
||||
Health ChannelMonitorV2Health `json:"health"`
|
||||
Buckets []ChannelMonitorV2TrendPoint `json:"buckets"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2Matrix struct {
|
||||
GroupBy ChannelMonitorV2GroupBy `json:"group_by"`
|
||||
Coverage ChannelMonitorV2Coverage `json:"coverage"`
|
||||
Items []ChannelMonitorV2MatrixRow `json:"items"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2ErrorRow struct {
|
||||
Category string `json:"category"`
|
||||
Count int64 `json:"count"`
|
||||
Rate float64 `json:"rate"`
|
||||
Details []ChannelMonitorV2ErrorDetail `json:"details,omitempty"`
|
||||
// Ignored is true when this category is excluded from error_rate scoring
|
||||
// via config.ignored_error_categories. Still shown in breakdown (UI greys it).
|
||||
Ignored bool `json:"ignored"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2ErrorDetail struct {
|
||||
Platform string `json:"platform,omitempty"`
|
||||
Model string `json:"model,omitempty"`
|
||||
ErrorType string `json:"error_type,omitempty"`
|
||||
StatusCode int `json:"status_code,omitempty"`
|
||||
UpstreamStatusCode int `json:"upstream_status_code,omitempty"`
|
||||
Message string `json:"message,omitempty"`
|
||||
Count int64 `json:"count"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2UserRow struct {
|
||||
UserID *int64 `json:"user_id,omitempty"`
|
||||
Rank int `json:"rank"`
|
||||
Email string `json:"email,omitempty"`
|
||||
Username string `json:"username,omitempty"`
|
||||
DisplayLabel string `json:"display_label"`
|
||||
IsSelf bool `json:"is_self"`
|
||||
CanDrilldown bool `json:"can_drilldown"`
|
||||
Metrics ChannelMonitorV2Metric `json:"metrics"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2List[T any] struct {
|
||||
Coverage ChannelMonitorV2Coverage `json:"coverage"`
|
||||
Items []T `json:"items"`
|
||||
}
|
||||
|
||||
type ChannelMonitorV2Repository interface {
|
||||
GetConfig(ctx context.Context) (*ChannelMonitorV2Config, error)
|
||||
UpdateConfig(ctx context.Context, config ChannelMonitorV2Config, expectedVersion int) (*ChannelMonitorV2Config, error)
|
||||
GetDimensions(ctx context.Context, filter ChannelMonitorV2Filter, config ChannelMonitorV2Config) (*ChannelMonitorV2Dimensions, error)
|
||||
GetSnapshot(ctx context.Context, filter ChannelMonitorV2Filter, config ChannelMonitorV2Config, includeAdmin bool) (*ChannelMonitorV2Snapshot, error)
|
||||
GetModels(ctx context.Context, filter ChannelMonitorV2Filter, config ChannelMonitorV2Config, includeAdmin bool) (*ChannelMonitorV2List[ChannelMonitorV2ModelRow], error)
|
||||
GetMatrix(ctx context.Context, filter ChannelMonitorV2Filter, config ChannelMonitorV2Config, groupBy ChannelMonitorV2GroupBy, includeAdmin bool) (*ChannelMonitorV2Matrix, error)
|
||||
// GetErrors loads category rates. When includeAdmin is false, implementations
|
||||
// must omit error Details (no ops_error_logs sample scan) for privacy.
|
||||
GetErrors(ctx context.Context, filter ChannelMonitorV2Filter, config ChannelMonitorV2Config, includeAdmin bool) (*ChannelMonitorV2List[ChannelMonitorV2ErrorRow], error)
|
||||
GetUsers(ctx context.Context, filter ChannelMonitorV2Filter, config ChannelMonitorV2Config, includeAdmin bool) (*ChannelMonitorV2List[ChannelMonitorV2UserRow], error)
|
||||
RecomputeRange(ctx context.Context, start, end time.Time) error
|
||||
}
|
||||
|
||||
type ChannelMonitorV2Service struct {
|
||||
repo ChannelMonitorV2Repository
|
||||
settings channelMonitorRuntimeReader
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
func NewChannelMonitorV2Service(repo ChannelMonitorV2Repository) *ChannelMonitorV2Service {
|
||||
return &ChannelMonitorV2Service{repo: repo, now: func() time.Time { return time.Now().UTC() }}
|
||||
}
|
||||
|
||||
// SetRuntimeReader wires optional settings for privacy flags (hide throughput).
|
||||
func (s *ChannelMonitorV2Service) SetRuntimeReader(r channelMonitorRuntimeReader) {
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
s.settings = r
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Service) hideThroughputForViewer(ctx context.Context, admin bool) bool {
|
||||
if admin || s == nil || s.settings == nil {
|
||||
return false
|
||||
}
|
||||
return s.settings.GetChannelMonitorRuntime(ctx).HideThroughput
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Service) GetConfig(ctx context.Context) (*ChannelMonitorV2Config, error) {
|
||||
return s.repo.GetConfig(ctx)
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Service) getEnabledConfig(ctx context.Context) (*ChannelMonitorV2Config, error) {
|
||||
cfg, err := s.repo.GetConfig(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if cfg == nil || !cfg.Enabled {
|
||||
return nil, ErrChannelMonitorDisabled
|
||||
}
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Service) UpdateConfig(ctx context.Context, cfg ChannelMonitorV2Config, expectedVersion int, actorID int64) (*ChannelMonitorV2Config, error) {
|
||||
if err := normalizeChannelMonitorV2Config(&cfg); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cfg.UpdatedBy = &actorID
|
||||
return s.repo.UpdateConfig(ctx, cfg, expectedVersion)
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Service) ParseFilter(rangeValue string, platforms, models []string, groupIDs []int64) (ChannelMonitorV2Filter, error) {
|
||||
now := s.now().UTC()
|
||||
var window, bucket time.Duration
|
||||
switch strings.TrimSpace(rangeValue) {
|
||||
case "", "90m":
|
||||
rangeValue, window, bucket = "90m", 90*time.Minute, 5*time.Minute
|
||||
case "24h":
|
||||
window, bucket = 24*time.Hour, time.Hour
|
||||
case "7d":
|
||||
window, bucket = 7*24*time.Hour, 12*time.Hour
|
||||
case "30d":
|
||||
window, bucket = 30*24*time.Hour, 24*time.Hour
|
||||
default:
|
||||
return ChannelMonitorV2Filter{}, fmt.Errorf("%w: %s", ErrChannelMonitorV2InvalidRange, rangeValue)
|
||||
}
|
||||
start, end := now.Add(-window), now
|
||||
if bucket > time.Minute {
|
||||
// Align display windows to a fixed number of whole buckets. The trailing
|
||||
// bucket may point slightly into the future; SQL simply has no future rows,
|
||||
// while the current partial bucket can still be shown and recomputed often.
|
||||
end = now.Truncate(bucket).Add(bucket)
|
||||
start = end.Add(-window)
|
||||
}
|
||||
return ChannelMonitorV2Filter{
|
||||
Range: rangeValue, Platforms: normalizeStringSet(platforms), Models: normalizeStringSet(models), GroupIDs: normalizeInt64Set(groupIDs),
|
||||
Start: start, End: end, Bucket: bucket,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Service) Dimensions(ctx context.Context, filter ChannelMonitorV2Filter) (*ChannelMonitorV2Dimensions, error) {
|
||||
cfg, err := s.getEnabledConfig(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dims, err := s.repo.GetDimensions(ctx, filter, *cfg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Dimension request_count is operational volume; strip for non-admin callers
|
||||
// at the API edge. Dimensions is shared by user/admin routes — redaction is
|
||||
// applied in the handler for user routes only, so keep raw here.
|
||||
return dims, nil
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Service) Snapshot(ctx context.Context, filter ChannelMonitorV2Filter, admin bool) (*ChannelMonitorV2Snapshot, error) {
|
||||
cfg, err := s.getEnabledConfig(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
snap, err := s.repo.GetSnapshot(ctx, filter, *cfg, admin)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !admin && snap != nil {
|
||||
redactChannelMonitorV2Snapshot(snap, s.hideThroughputForViewer(ctx, admin))
|
||||
}
|
||||
return snap, nil
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Service) Models(ctx context.Context, filter ChannelMonitorV2Filter, admin bool) (*ChannelMonitorV2List[ChannelMonitorV2ModelRow], error) {
|
||||
cfg, err := s.getEnabledConfig(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
list, err := s.repo.GetModels(ctx, filter, *cfg, admin)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !admin && list != nil {
|
||||
hideTP := s.hideThroughputForViewer(ctx, admin)
|
||||
for i := range list.Items {
|
||||
redactChannelMonitorV2Metric(&list.Items[i].Metrics, hideTP)
|
||||
}
|
||||
}
|
||||
return list, nil
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Service) Matrix(ctx context.Context, filter ChannelMonitorV2Filter, groupBy ChannelMonitorV2GroupBy, admin bool) (*ChannelMonitorV2Matrix, error) {
|
||||
if !groupBy.Valid() {
|
||||
return nil, fmt.Errorf("%w: %s", ErrChannelMonitorV2InvalidGroupBy, groupBy)
|
||||
}
|
||||
cfg, err := s.getEnabledConfig(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
matrix, err := s.repo.GetMatrix(ctx, filter, *cfg, groupBy, admin)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !admin && matrix != nil {
|
||||
hideTP := s.hideThroughputForViewer(ctx, admin)
|
||||
for i := range matrix.Items {
|
||||
redactChannelMonitorV2Metric(&matrix.Items[i].Metrics, hideTP)
|
||||
for j := range matrix.Items[i].Buckets {
|
||||
redactChannelMonitorV2Metric(&matrix.Items[i].Buckets[j].Metrics, hideTP)
|
||||
}
|
||||
}
|
||||
}
|
||||
return matrix, nil
|
||||
}
|
||||
|
||||
func ParseChannelMonitorV2GroupBy(value string) (ChannelMonitorV2GroupBy, error) {
|
||||
groupBy := ChannelMonitorV2GroupBy(strings.TrimSpace(value))
|
||||
if groupBy == "" {
|
||||
// Default presentation: platform / group (matches operator mental model).
|
||||
groupBy = ChannelMonitorV2GroupByPlatformGroup
|
||||
}
|
||||
if !groupBy.Valid() {
|
||||
return "", fmt.Errorf("%w: %s", ErrChannelMonitorV2InvalidGroupBy, value)
|
||||
}
|
||||
return groupBy, nil
|
||||
}
|
||||
|
||||
func (g ChannelMonitorV2GroupBy) Valid() bool {
|
||||
switch g {
|
||||
case ChannelMonitorV2GroupByPlatform, ChannelMonitorV2GroupByPlatformGroup, ChannelMonitorV2GroupByPlatformModel, ChannelMonitorV2GroupByPlatformGroupModel:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Service) Errors(ctx context.Context, filter ChannelMonitorV2Filter) (*ChannelMonitorV2List[ChannelMonitorV2ErrorRow], error) {
|
||||
return s.ErrorsForViewer(ctx, filter, false)
|
||||
}
|
||||
|
||||
// ErrorsForViewer returns the error breakdown.
|
||||
// Non-admin callers receive category rates + ignored flags only: absolute Count
|
||||
// is zeroed and Details (upstream messages / status codes / volume) are omitted.
|
||||
func (s *ChannelMonitorV2Service) ErrorsForViewer(ctx context.Context, filter ChannelMonitorV2Filter, admin bool) (*ChannelMonitorV2List[ChannelMonitorV2ErrorRow], error) {
|
||||
cfg, err := s.getEnabledConfig(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
list, err := s.repo.GetErrors(ctx, filter, *cfg, admin)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !admin && list != nil {
|
||||
for i := range list.Items {
|
||||
list.Items[i].Count = 0
|
||||
list.Items[i].Details = nil
|
||||
}
|
||||
}
|
||||
return list, nil
|
||||
}
|
||||
|
||||
// RedactChannelMonitorV2Dimensions clears absolute request counts on filter chips.
|
||||
func RedactChannelMonitorV2Dimensions(dims *ChannelMonitorV2Dimensions) {
|
||||
if dims == nil {
|
||||
return
|
||||
}
|
||||
for i := range dims.Platforms {
|
||||
dims.Platforms[i].RequestCount = 0
|
||||
}
|
||||
for i := range dims.Models {
|
||||
dims.Models[i].RequestCount = 0
|
||||
}
|
||||
for i := range dims.Groups {
|
||||
dims.Groups[i].RequestCount = 0
|
||||
}
|
||||
}
|
||||
|
||||
func redactChannelMonitorV2Snapshot(snap *ChannelMonitorV2Snapshot, hideThroughput bool) {
|
||||
if snap == nil {
|
||||
return
|
||||
}
|
||||
redactChannelMonitorV2Metric(&snap.Metrics, hideThroughput)
|
||||
for i := range snap.Trend {
|
||||
redactChannelMonitorV2Metric(&snap.Trend[i].Metrics, hideThroughput)
|
||||
}
|
||||
// Public snapshot only needs display thresholds + refresh cadence, not
|
||||
// operational allow-lists (group_ids, model inventories, ignored categories).
|
||||
redactChannelMonitorV2PublicConfig(&snap.Config)
|
||||
}
|
||||
|
||||
// redactChannelMonitorV2PublicConfig strips operator-policy fields from config
|
||||
// embedded in user-facing snapshots. Full config remains on admin /config.
|
||||
func redactChannelMonitorV2PublicConfig(cfg *ChannelMonitorV2Config) {
|
||||
if cfg == nil {
|
||||
return
|
||||
}
|
||||
cfg.GroupIDs = nil
|
||||
cfg.IgnoredErrorCategories = nil
|
||||
cfg.UpdatedBy = nil
|
||||
for i := range cfg.Platforms {
|
||||
cfg.Platforms[i].Models = nil
|
||||
}
|
||||
}
|
||||
|
||||
// redactChannelMonitorV2Metric zeros absolute volume counters while keeping rates
|
||||
// (error_rate, success_rate, cache_rate) and latency percentiles.
|
||||
// When hideThroughput is true, also zeros RPM/TPM so users cannot reverse-estimate
|
||||
// fleet scale from rate × window length.
|
||||
func redactChannelMonitorV2Metric(m *ChannelMonitorV2Metric, hideThroughput bool) {
|
||||
if m == nil {
|
||||
return
|
||||
}
|
||||
m.SuccessRequests = 0
|
||||
m.ErrorRequests = 0
|
||||
m.RequestCount = 0
|
||||
m.InputTokens = 0
|
||||
m.OutputTokens = 0
|
||||
m.CacheCreationTokens = 0
|
||||
m.CacheReadTokens = 0
|
||||
m.TokenCount = 0
|
||||
m.CacheRateNumerator = 0
|
||||
m.CacheRateDenominator = 0
|
||||
// Latency sample_count is also a volume signal.
|
||||
m.TTFT.SampleCount = 0
|
||||
m.Duration.SampleCount = 0
|
||||
m.UpstreamAffectedRequests = nil
|
||||
m.UpstreamAttemptCount = nil
|
||||
if hideThroughput {
|
||||
m.RPM = 0
|
||||
m.TPM = 0
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Service) Users(ctx context.Context, filter ChannelMonitorV2Filter, viewerID int64, admin bool) (*ChannelMonitorV2List[ChannelMonitorV2UserRow], error) {
|
||||
cfg, err := s.getEnabledConfig(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result, err := s.repo.GetUsers(ctx, filter, *cfg, admin)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if result == nil {
|
||||
result = &ChannelMonitorV2List[ChannelMonitorV2UserRow]{}
|
||||
}
|
||||
selfIndex := -1
|
||||
for i := range result.Items {
|
||||
result.Items[i].Rank = i + 1
|
||||
if result.Items[i].UserID != nil && *result.Items[i].UserID == viewerID {
|
||||
selfIndex = i
|
||||
result.Items[i].IsSelf = true
|
||||
}
|
||||
}
|
||||
// Viewer with no traffic in this window is still shown (and highlighted) so
|
||||
// ranking always answers "where am I?" — not only when already in the top list.
|
||||
if selfIndex < 0 && viewerID > 0 {
|
||||
id := viewerID
|
||||
selfRow := ChannelMonitorV2UserRow{
|
||||
UserID: &id,
|
||||
Rank: 0, // unranked / no traffic in window
|
||||
IsSelf: true,
|
||||
CanDrilldown: true,
|
||||
DisplayLabel: "Me",
|
||||
Metrics: ChannelMonitorV2Metric{},
|
||||
}
|
||||
result.Items = append(result.Items, selfRow)
|
||||
selfIndex = len(result.Items) - 1
|
||||
}
|
||||
result.Items = channelMonitorV2TopUsersWithSelf(result.Items, selfIndex, 10)
|
||||
hideTP := s.hideThroughputForViewer(ctx, admin)
|
||||
if admin {
|
||||
// Keep identity for admin; still mark self for UI highlight.
|
||||
for i := range result.Items {
|
||||
if result.Items[i].UserID != nil && *result.Items[i].UserID == viewerID {
|
||||
result.Items[i].IsSelf = true
|
||||
if result.Items[i].DisplayLabel == "" || result.Items[i].DisplayLabel == "Me" {
|
||||
// Prefer real label when available from repo.
|
||||
if result.Items[i].Username != "" {
|
||||
result.Items[i].DisplayLabel = result.Items[i].Username
|
||||
} else if result.Items[i].Email != "" {
|
||||
result.Items[i].DisplayLabel = result.Items[i].Email
|
||||
} else {
|
||||
result.Items[i].DisplayLabel = "Me"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
for i := range result.Items {
|
||||
redactChannelMonitorV2Metric(&result.Items[i].Metrics, hideTP)
|
||||
}
|
||||
for i := range result.Items {
|
||||
row := &result.Items[i]
|
||||
if row.UserID != nil && *row.UserID == viewerID {
|
||||
row.IsSelf, row.CanDrilldown, row.DisplayLabel = true, true, "Me"
|
||||
continue
|
||||
}
|
||||
row.UserID, row.Email, row.Username, row.CanDrilldown = nil, "", "", false
|
||||
row.DisplayLabel = fmt.Sprintf("Other user #%d", i+1)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func channelMonitorV2TopUsersWithSelf(items []ChannelMonitorV2UserRow, selfIndex int, limit int) []ChannelMonitorV2UserRow {
|
||||
if limit <= 0 {
|
||||
return items
|
||||
}
|
||||
if len(items) <= limit {
|
||||
return items
|
||||
}
|
||||
out := append([]ChannelMonitorV2UserRow(nil), items[:limit]...)
|
||||
if selfIndex >= limit && selfIndex < len(items) {
|
||||
out = append(out, items[selfIndex])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func normalizeChannelMonitorV2Config(cfg *ChannelMonitorV2Config) error {
|
||||
if cfg.RefreshIntervalSeconds == 0 {
|
||||
cfg.RefreshIntervalSeconds = 300
|
||||
}
|
||||
if cfg.RefreshIntervalSeconds != 60 && cfg.RefreshIntervalSeconds != 300 {
|
||||
return fmt.Errorf("%w: refresh_interval_seconds must be 60 or 300", ErrChannelMonitorV2InvalidConfig)
|
||||
}
|
||||
var err error
|
||||
cfg.GroupIDs, err = normalizeChannelMonitorV2GroupIDs(cfg.GroupIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cfg.IgnoredErrorCategories = normalizeChannelMonitorV2IgnoredCategories(cfg.IgnoredErrorCategories)
|
||||
cfg.HealthThresholds = NormalizeChannelMonitorV2HealthThresholds(cfg.HealthThresholds)
|
||||
seen := make(map[string]struct{}, len(cfg.Platforms))
|
||||
for i := range cfg.Platforms {
|
||||
p := &cfg.Platforms[i]
|
||||
p.Platform = strings.ToLower(strings.TrimSpace(p.Platform))
|
||||
if p.Platform == "" {
|
||||
return fmt.Errorf("%w: empty platform", ErrChannelMonitorV2InvalidConfig)
|
||||
}
|
||||
if _, ok := seen[p.Platform]; ok {
|
||||
return fmt.Errorf("%w: duplicate platform %s", ErrChannelMonitorV2InvalidConfig, p.Platform)
|
||||
}
|
||||
seen[p.Platform] = struct{}{}
|
||||
p.Models = normalizeStringSet(p.Models)
|
||||
}
|
||||
sort.Slice(cfg.Platforms, func(i, j int) bool { return cfg.Platforms[i].Platform < cfg.Platforms[j].Platform })
|
||||
return nil
|
||||
}
|
||||
|
||||
// DefaultChannelMonitorV2IgnoredErrorCategories are factory defaults for
|
||||
// ignored_error_categories: excluded from error_rate / health scoring only.
|
||||
// Operators can clear or extend via admin config.
|
||||
var DefaultChannelMonitorV2IgnoredErrorCategories = []string{
|
||||
"authentication",
|
||||
"client_cancelled",
|
||||
"content_policy",
|
||||
"context_limit",
|
||||
"group_access",
|
||||
"model_unsupported",
|
||||
"not_found",
|
||||
"quota_or_balance",
|
||||
}
|
||||
|
||||
func DefaultChannelMonitorV2HealthThresholds() ChannelMonitorV2HealthThresholds {
|
||||
return ChannelMonitorV2HealthThresholds{
|
||||
MinimumSample: 50,
|
||||
WarningErrorRate: 0.05,
|
||||
CriticalErrorRate: 0.20,
|
||||
TargetTTFTMs: 3000,
|
||||
WarningTTFTMs: 3000,
|
||||
CriticalTTFTMs: 10000,
|
||||
// A zero/zero cache threshold means cache misses do not affect health
|
||||
// until an operator explicitly configures cache scoring.
|
||||
WarningCacheRate: 0,
|
||||
CriticalCacheRate: 0,
|
||||
ErrorWeight: 0.60,
|
||||
TTFTWeight: 0.20,
|
||||
CacheWeight: 0.20,
|
||||
}
|
||||
}
|
||||
|
||||
func NormalizeChannelMonitorV2HealthThresholds(in ChannelMonitorV2HealthThresholds) ChannelMonitorV2HealthThresholds {
|
||||
def := DefaultChannelMonitorV2HealthThresholds()
|
||||
if in.MinimumSample <= 0 {
|
||||
in.MinimumSample = def.MinimumSample
|
||||
}
|
||||
if in.MinimumSample < 1 {
|
||||
in.MinimumSample = 1
|
||||
}
|
||||
if in.MinimumSample > 10000 {
|
||||
in.MinimumSample = 10000
|
||||
}
|
||||
if in.WarningErrorRate <= 0 {
|
||||
in.WarningErrorRate = def.WarningErrorRate
|
||||
}
|
||||
if in.CriticalErrorRate <= 0 {
|
||||
in.CriticalErrorRate = def.CriticalErrorRate
|
||||
}
|
||||
if in.CriticalErrorRate < in.WarningErrorRate {
|
||||
in.CriticalErrorRate = in.WarningErrorRate
|
||||
}
|
||||
if in.TargetTTFTMs <= 0 {
|
||||
in.TargetTTFTMs = def.TargetTTFTMs
|
||||
}
|
||||
if in.WarningTTFTMs <= 0 {
|
||||
in.WarningTTFTMs = def.WarningTTFTMs
|
||||
}
|
||||
if in.WarningTTFTMs < in.TargetTTFTMs {
|
||||
in.WarningTTFTMs = in.TargetTTFTMs + 1
|
||||
}
|
||||
if in.CriticalTTFTMs <= 0 {
|
||||
in.CriticalTTFTMs = def.CriticalTTFTMs
|
||||
}
|
||||
if in.CriticalTTFTMs < in.WarningTTFTMs {
|
||||
in.CriticalTTFTMs = in.WarningTTFTMs
|
||||
}
|
||||
if in.WarningCacheRate < 0 {
|
||||
in.WarningCacheRate = 0
|
||||
}
|
||||
if in.CriticalCacheRate < 0 {
|
||||
in.CriticalCacheRate = 0
|
||||
}
|
||||
if in.WarningCacheRate > 1 {
|
||||
in.WarningCacheRate = 1
|
||||
}
|
||||
if in.CriticalCacheRate > 1 {
|
||||
in.CriticalCacheRate = 1
|
||||
}
|
||||
if in.CriticalCacheRate > in.WarningCacheRate {
|
||||
in.CriticalCacheRate = in.WarningCacheRate
|
||||
}
|
||||
if in.ErrorWeight <= 0 && in.TTFTWeight <= 0 && in.CacheWeight <= 0 {
|
||||
in.ErrorWeight, in.TTFTWeight, in.CacheWeight = def.ErrorWeight, def.TTFTWeight, def.CacheWeight
|
||||
}
|
||||
return in
|
||||
}
|
||||
|
||||
func normalizeChannelMonitorV2IgnoredCategories(values []string) []string {
|
||||
allowed := make(map[string]struct{}, len(ChannelMonitorV2ErrorCategories))
|
||||
for _, c := range ChannelMonitorV2ErrorCategories {
|
||||
allowed[c] = struct{}{}
|
||||
}
|
||||
seen := make(map[string]struct{}, len(values))
|
||||
out := make([]string, 0, len(values))
|
||||
for _, value := range values {
|
||||
value = strings.ToLower(strings.TrimSpace(value))
|
||||
if value == "" {
|
||||
continue
|
||||
}
|
||||
if _, ok := allowed[value]; !ok {
|
||||
// Unknown labels are dropped so a typo cannot silently create a new category.
|
||||
continue
|
||||
}
|
||||
if _, ok := seen[value]; ok {
|
||||
continue
|
||||
}
|
||||
seen[value] = struct{}{}
|
||||
out = append(out, value)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// ChannelMonitorV2IgnoredCategorySet returns a set for O(1) membership checks.
|
||||
func ChannelMonitorV2IgnoredCategorySet(cfg ChannelMonitorV2Config) map[string]struct{} {
|
||||
set := make(map[string]struct{}, len(cfg.IgnoredErrorCategories))
|
||||
for _, c := range cfg.IgnoredErrorCategories {
|
||||
set[c] = struct{}{}
|
||||
}
|
||||
return set
|
||||
}
|
||||
|
||||
func normalizeStringSet(values []string) []string {
|
||||
seen := make(map[string]struct{}, len(values))
|
||||
out := make([]string, 0, len(values))
|
||||
for _, value := range values {
|
||||
value = strings.TrimSpace(value)
|
||||
if value == "" {
|
||||
continue
|
||||
}
|
||||
if _, ok := seen[value]; ok {
|
||||
continue
|
||||
}
|
||||
seen[value] = struct{}{}
|
||||
out = append(out, value)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
func normalizeInt64Set(values []int64) []int64 {
|
||||
seen := make(map[int64]struct{}, len(values))
|
||||
out := make([]int64, 0, len(values))
|
||||
for _, value := range values {
|
||||
if value <= 0 {
|
||||
continue
|
||||
}
|
||||
if _, ok := seen[value]; ok {
|
||||
continue
|
||||
}
|
||||
seen[value] = struct{}{}
|
||||
out = append(out, value)
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
|
||||
return out
|
||||
}
|
||||
|
||||
func normalizeChannelMonitorV2GroupIDs(values []int64) ([]int64, error) {
|
||||
seen := make(map[int64]struct{}, len(values))
|
||||
out := make([]int64, 0, len(values))
|
||||
for _, value := range values {
|
||||
if value <= 0 {
|
||||
return nil, fmt.Errorf("%w: group_ids must be positive", ErrChannelMonitorV2InvalidConfig)
|
||||
}
|
||||
if _, ok := seen[value]; ok {
|
||||
continue
|
||||
}
|
||||
seen[value] = struct{}{}
|
||||
out = append(out, value)
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func ChannelMonitorV2HealthFor(metrics ChannelMonitorV2Metric) ChannelMonitorV2Health {
|
||||
return ChannelMonitorV2HealthForWithThresholds(metrics, DefaultChannelMonitorV2HealthThresholds())
|
||||
}
|
||||
|
||||
func ChannelMonitorV2HealthForWithThresholds(metrics ChannelMonitorV2Metric, thresholds ChannelMonitorV2HealthThresholds) ChannelMonitorV2Health {
|
||||
thresholds = NormalizeChannelMonitorV2HealthThresholds(thresholds)
|
||||
result := ChannelMonitorV2Health{
|
||||
Overall: "unknown", ErrorRate: "unknown", TTFT: "unknown", Cache: "unknown",
|
||||
MinimumSample: thresholds.MinimumSample, Thresholds: thresholds,
|
||||
}
|
||||
|
||||
type scored struct {
|
||||
score float64
|
||||
weight float64
|
||||
band string
|
||||
}
|
||||
parts := make([]scored, 0, 3)
|
||||
|
||||
if metrics.RequestCount >= result.MinimumSample {
|
||||
s := errorRateScore(metrics.ErrorRate, thresholds.CriticalErrorRate)
|
||||
result.ErrorRateScore = &s
|
||||
result.ErrorRate = healthBand(metrics.ErrorRate, thresholds.WarningErrorRate, thresholds.CriticalErrorRate)
|
||||
parts = append(parts, scored{score: s, weight: thresholds.ErrorWeight, band: result.ErrorRate})
|
||||
}
|
||||
// Prefer p50 for TTFT scoring; fall back to p95 only if p50 is missing.
|
||||
if metrics.TTFT.SampleCount >= result.MinimumSample {
|
||||
var ttftMs *int64
|
||||
if metrics.TTFT.P50Ms != nil {
|
||||
ttftMs = metrics.TTFT.P50Ms
|
||||
} else if metrics.TTFT.P95Ms != nil {
|
||||
ttftMs = metrics.TTFT.P95Ms
|
||||
}
|
||||
if ttftMs != nil {
|
||||
s := ttftP50Score(float64(*ttftMs), float64(thresholds.TargetTTFTMs), float64(thresholds.CriticalTTFTMs))
|
||||
result.TTFTScore = &s
|
||||
result.TTFT = healthBand(float64(*ttftMs), float64(thresholds.WarningTTFTMs), float64(thresholds.CriticalTTFTMs))
|
||||
parts = append(parts, scored{score: s, weight: thresholds.TTFTWeight, band: result.TTFT})
|
||||
}
|
||||
}
|
||||
// Cache: need a meaningful denominator; higher rate is better.
|
||||
if metrics.CacheRateDenominator >= result.MinimumSample {
|
||||
s := cacheRateScore(metrics.CacheRate)
|
||||
if thresholds.WarningCacheRate <= 0 && thresholds.CriticalCacheRate <= 0 {
|
||||
// A zero/zero cache threshold means "do not penalize cache misses".
|
||||
s = 100
|
||||
}
|
||||
result.CacheScore = &s
|
||||
// Invert for healthBand (lower is worse): use (1 - rate) against warning/critical floors.
|
||||
result.Cache = cacheRateBand(metrics.CacheRate, thresholds.WarningCacheRate, thresholds.CriticalCacheRate)
|
||||
parts = append(parts, scored{score: s, weight: thresholds.CacheWeight, band: result.Cache})
|
||||
}
|
||||
|
||||
if len(parts) == 0 {
|
||||
return result
|
||||
}
|
||||
var weightSum, scoreSum float64
|
||||
for _, p := range parts {
|
||||
weightSum += p.weight
|
||||
scoreSum += p.weight * p.score
|
||||
}
|
||||
if weightSum <= 0 {
|
||||
return result
|
||||
}
|
||||
overall := scoreSum / weightSum
|
||||
result.Score = &overall
|
||||
result.Overall = scoreBand(overall)
|
||||
return result
|
||||
}
|
||||
|
||||
// errorRateScore maps error rate to 0–100. 0% → 100; at/above critical → 0 (linear).
|
||||
func errorRateScore(errorRate, critical float64) float64 {
|
||||
if critical <= 0 {
|
||||
critical = 0.05
|
||||
}
|
||||
if errorRate <= 0 {
|
||||
return 100
|
||||
}
|
||||
if errorRate >= critical {
|
||||
return 0
|
||||
}
|
||||
return 100 * (1 - errorRate/critical)
|
||||
}
|
||||
|
||||
// ttftP50Score maps TTFT p50 ms to 0–100.
|
||||
// At/below target → 100; at/above critical → 0; linear in between.
|
||||
func ttftP50Score(p50Ms, targetMs, criticalMs float64) float64 {
|
||||
if targetMs <= 0 {
|
||||
targetMs = 2500
|
||||
}
|
||||
if criticalMs <= targetMs {
|
||||
criticalMs = targetMs * 2.4
|
||||
}
|
||||
if p50Ms <= targetMs {
|
||||
return 100
|
||||
}
|
||||
if p50Ms >= criticalMs {
|
||||
return 0
|
||||
}
|
||||
return 100 * (1 - (p50Ms-targetMs)/(criticalMs-targetMs))
|
||||
}
|
||||
|
||||
// cacheRateScore maps cache hit rate to 0–100 (higher is better, linear).
|
||||
func cacheRateScore(cacheRate float64) float64 {
|
||||
if cacheRate <= 0 {
|
||||
return 0
|
||||
}
|
||||
if cacheRate >= 1 {
|
||||
return 100
|
||||
}
|
||||
return 100 * cacheRate
|
||||
}
|
||||
|
||||
// cacheRateBand: below critical → critical; below warning → warning; else healthy.
|
||||
func cacheRateBand(cacheRate, warning, critical float64) string {
|
||||
if cacheRate < critical {
|
||||
return "critical"
|
||||
}
|
||||
if cacheRate < warning {
|
||||
return "warning"
|
||||
}
|
||||
return "healthy"
|
||||
}
|
||||
|
||||
// scoreBand maps continuous 0–100 scores to coarse labels for legacy consumers.
|
||||
func scoreBand(score float64) string {
|
||||
switch {
|
||||
case score >= 80:
|
||||
return "healthy"
|
||||
case score >= 50:
|
||||
return "warning"
|
||||
default:
|
||||
return "critical"
|
||||
}
|
||||
}
|
||||
|
||||
func healthBand(value, warning, critical float64) string {
|
||||
if value >= critical {
|
||||
return "critical"
|
||||
}
|
||||
if value >= warning {
|
||||
return "warning"
|
||||
}
|
||||
return "healthy"
|
||||
}
|
||||
@@ -0,0 +1,227 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
const (
|
||||
channelMonitorV2AggregatorLockKey = "channel-monitor-v2-aggregator"
|
||||
// Backfill walks back to the longest stored tier (1d rollup = 90d). Per-tier
|
||||
// prune in the repository drops short-lived 1m/user/hist facts earlier.
|
||||
channelMonitorV2RetentionMax = 90 * 24 * time.Hour
|
||||
channelMonitorV2RecentOverlap = 10 * time.Minute
|
||||
channelMonitorV2InitialWindow = 2 * time.Hour
|
||||
channelMonitorV2BackfillChunk = 24 * time.Hour
|
||||
)
|
||||
|
||||
// channelMonitorRuntimeSubscriber is the optional settings hook that lets the
|
||||
// aggregator wake immediately when channel_monitor_enabled / mode flips.
|
||||
type channelMonitorRuntimeSubscriber interface {
|
||||
SubscribeChannelMonitorRuntime(listener func()) (unsubscribe func())
|
||||
}
|
||||
|
||||
type ChannelMonitorV2Aggregator struct {
|
||||
repo ChannelMonitorV2Repository
|
||||
db *sql.DB
|
||||
settings channelMonitorRuntimeReader
|
||||
instanceID string
|
||||
stopCh chan struct{}
|
||||
// kickCh wakes the loop early after a settings change (buffered 1).
|
||||
kickCh chan struct{}
|
||||
startOnce sync.Once
|
||||
stopOnce sync.Once
|
||||
mu sync.Mutex
|
||||
backfillAt time.Time
|
||||
unsub func()
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
func NewChannelMonitorV2Aggregator(repo ChannelMonitorV2Repository, db *sql.DB, settings channelMonitorRuntimeReader) *ChannelMonitorV2Aggregator {
|
||||
return &ChannelMonitorV2Aggregator{
|
||||
repo: repo,
|
||||
db: db,
|
||||
settings: settings,
|
||||
instanceID: uuid.NewString(),
|
||||
stopCh: make(chan struct{}),
|
||||
kickCh: make(chan struct{}, 1),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Aggregator) Start() {
|
||||
if s == nil || s.repo == nil {
|
||||
return
|
||||
}
|
||||
s.startOnce.Do(func() {
|
||||
s.mu.Lock()
|
||||
s.ctx, s.cancel = context.WithCancel(context.Background())
|
||||
s.mu.Unlock()
|
||||
if sub, ok := s.settings.(channelMonitorRuntimeSubscriber); ok && sub != nil {
|
||||
unsub := sub.SubscribeChannelMonitorRuntime(func() {
|
||||
s.kick()
|
||||
})
|
||||
s.mu.Lock()
|
||||
stopped := s.ctx == nil
|
||||
if !stopped {
|
||||
select {
|
||||
case <-s.ctx.Done():
|
||||
stopped = true
|
||||
default:
|
||||
}
|
||||
}
|
||||
if !stopped {
|
||||
s.unsub = unsub
|
||||
}
|
||||
s.mu.Unlock()
|
||||
if stopped && unsub != nil {
|
||||
unsub()
|
||||
}
|
||||
}
|
||||
go s.loop()
|
||||
})
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Aggregator) Stop() {
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
s.stopOnce.Do(func() {
|
||||
s.mu.Lock()
|
||||
cancel := s.cancel
|
||||
unsub := s.unsub
|
||||
s.cancel = nil
|
||||
s.unsub = nil
|
||||
s.mu.Unlock()
|
||||
if cancel != nil {
|
||||
cancel()
|
||||
}
|
||||
if unsub != nil {
|
||||
unsub()
|
||||
}
|
||||
close(s.stopCh)
|
||||
})
|
||||
}
|
||||
|
||||
// kick wakes the aggregation loop so mode flips take effect without waiting
|
||||
// for the next refresh interval.
|
||||
func (s *ChannelMonitorV2Aggregator) kick() {
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case s.kickCh <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Aggregator) loop() {
|
||||
for {
|
||||
interval := time.Minute
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
if !s.passiveAggregationAllowed(ctx) {
|
||||
cancel()
|
||||
if !s.wait(interval) {
|
||||
return
|
||||
}
|
||||
continue
|
||||
}
|
||||
if cfg, err := s.repo.GetConfig(ctx); err == nil {
|
||||
if !cfg.Enabled {
|
||||
cancel()
|
||||
if !s.wait(interval) {
|
||||
return
|
||||
}
|
||||
continue
|
||||
}
|
||||
if cfg.RefreshIntervalSeconds > 0 {
|
||||
interval = time.Duration(cfg.RefreshIntervalSeconds) * time.Second
|
||||
}
|
||||
}
|
||||
cancel()
|
||||
s.runOnce()
|
||||
if !s.wait(interval) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Aggregator) passiveAggregationAllowed(ctx context.Context) bool {
|
||||
if s == nil || s.settings == nil {
|
||||
// Fail closed without settings: do not aggregate under ambiguous mode.
|
||||
return false
|
||||
}
|
||||
return s.settings.GetChannelMonitorRuntime(ctx).PassiveAggregationAllowed()
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Aggregator) wait(interval time.Duration) bool {
|
||||
timer := time.NewTimer(interval)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-timer.C:
|
||||
return true
|
||||
case <-s.kickCh:
|
||||
// Drain any coalesced kicks so a burst of settings writes only wakes once.
|
||||
for {
|
||||
select {
|
||||
case <-s.kickCh:
|
||||
default:
|
||||
return true
|
||||
}
|
||||
}
|
||||
case <-s.stopCh:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Aggregator) runOnce() {
|
||||
s.mu.Lock()
|
||||
parent := s.ctx
|
||||
s.mu.Unlock()
|
||||
if parent == nil {
|
||||
parent = context.Background()
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(parent, 55*time.Second)
|
||||
defer cancel()
|
||||
release, acquired := tryAcquireSingletonLeaderLock(ctx, nil, s.db, channelMonitorV2AggregatorLockKey, s.instanceID, 2*time.Minute)
|
||||
if !acquired {
|
||||
return
|
||||
}
|
||||
if release != nil {
|
||||
defer release()
|
||||
}
|
||||
|
||||
now := time.Now().UTC().Truncate(time.Minute)
|
||||
if s.backfillAt.IsZero() {
|
||||
start := now.Add(-channelMonitorV2InitialWindow)
|
||||
if err := s.repo.RecomputeRange(ctx, start, now); err != nil {
|
||||
logger.LegacyPrintf("service.channel_monitor_v2", "[ChannelMonitorV2] recent aggregation failed: %v", err)
|
||||
return
|
||||
}
|
||||
s.backfillAt = start
|
||||
return
|
||||
}
|
||||
|
||||
if err := s.repo.RecomputeRange(ctx, now.Add(-channelMonitorV2RecentOverlap), now); err != nil {
|
||||
logger.LegacyPrintf("service.channel_monitor_v2", "[ChannelMonitorV2] overlap aggregation failed: %v", err)
|
||||
return
|
||||
}
|
||||
cutoff := now.Add(-channelMonitorV2RetentionMax)
|
||||
if s.backfillAt.After(cutoff) {
|
||||
end := s.backfillAt
|
||||
start := end.Add(-channelMonitorV2BackfillChunk)
|
||||
if start.Before(cutoff) {
|
||||
start = cutoff
|
||||
}
|
||||
if err := s.repo.RecomputeRange(ctx, start, end); err != nil {
|
||||
logger.LegacyPrintf("service.channel_monitor_v2", "[ChannelMonitorV2] backfill failed %s..%s: %v", start, end, err)
|
||||
return
|
||||
}
|
||||
s.backfillAt = start
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
package service
|
||||
|
||||
import "strings"
|
||||
|
||||
type ChannelMonitorV2ErrorInput struct {
|
||||
ErrorType string
|
||||
ErrorOwner string
|
||||
ErrorSource string
|
||||
StatusCode int
|
||||
UpstreamStatusCode int
|
||||
Message string
|
||||
}
|
||||
|
||||
// ClassifyChannelMonitorV2Error is deliberately ordered. Once rows are stored
|
||||
// with a taxonomy version, changing this order requires a new version.
|
||||
func ClassifyChannelMonitorV2Error(input ChannelMonitorV2ErrorInput) string {
|
||||
errorType := strings.ToLower(strings.TrimSpace(input.ErrorType))
|
||||
owner := strings.ToLower(strings.TrimSpace(input.ErrorOwner))
|
||||
text := strings.ToLower(strings.Join([]string{input.ErrorType, input.ErrorSource, input.Message}, " "))
|
||||
|
||||
if errorType == "cyber_policy" || channelMonitorV2ContainsAny(text, "content policy", "content_policy", "safety policy", "moderation", "blocked keyword") {
|
||||
return "content_policy"
|
||||
}
|
||||
if input.StatusCode == 401 || input.UpstreamStatusCode == 401 || channelMonitorV2ContainsAny(text, "unauthorized", "invalid api key", "invalid_api_key", "authentication", "api_key_disabled") {
|
||||
return "authentication"
|
||||
}
|
||||
if channelMonitorV2ContainsAny(text, "context window", "context length", "maximum prompt length", "too many tokens", "max_tokens") {
|
||||
return "context_limit"
|
||||
}
|
||||
if channelMonitorV2ContainsAny(text, "failed to deserialize", "missing required parameter", "invalid request", "invalid_request", "tool_choice") {
|
||||
return "invalid_request"
|
||||
}
|
||||
if channelMonitorV2ContainsAny(text, "does not support the requested model", "not supported by any configured account", "model not supported", "unsupported model") {
|
||||
return "model_unsupported"
|
||||
}
|
||||
if channelMonitorV2ContainsAny(text, "group not allowed", "group_not_allowed", "group access") {
|
||||
return "group_access"
|
||||
}
|
||||
if channelMonitorV2ContainsAny(text, "run out of credits", "insufficient balance", "insufficient quota", "subscription", "quota exceeded", "billing hard limit") {
|
||||
return "quota_or_balance"
|
||||
}
|
||||
if channelMonitorV2ContainsAny(text, "no available accounts", "no healthy account", "no healthy upstream account", "failover budget exhausted", "account pool") {
|
||||
return "account_pool_unavailable"
|
||||
}
|
||||
if input.StatusCode == 429 || input.UpstreamStatusCode == 429 || channelMonitorV2ContainsAny(text, "rate limit", "rate_limit", "high demand", "overloaded", "concurrency limit", "capacity") {
|
||||
return "rate_or_capacity"
|
||||
}
|
||||
if input.StatusCode == 408 || input.StatusCode == 504 || channelMonitorV2ContainsAny(text, "timeout", "deadline exceeded", "error code: 524", "gateway time-out", "gateway timeout") {
|
||||
return "timeout"
|
||||
}
|
||||
if channelMonitorV2ContainsAny(text, "transport", "stream_read_error", "connection reset", "connection refused", "tls", "http2", "missing terminal event", "unexpected eof") {
|
||||
return "transport_or_stream"
|
||||
}
|
||||
if input.StatusCode == 403 || input.UpstreamStatusCode == 403 {
|
||||
return "upstream_forbidden"
|
||||
}
|
||||
if input.StatusCode == 404 || input.UpstreamStatusCode == 404 {
|
||||
return "not_found"
|
||||
}
|
||||
if input.StatusCode == 499 || channelMonitorV2ContainsAny(text, "client cancelled", "client canceled", "context canceled") {
|
||||
return "client_cancelled"
|
||||
}
|
||||
if input.UpstreamStatusCode >= 500 || (owner == "provider" && input.StatusCode >= 500) {
|
||||
return "upstream_5xx"
|
||||
}
|
||||
if input.StatusCode >= 500 || errorType == "internal" || owner == "system" {
|
||||
return "internal"
|
||||
}
|
||||
return "other"
|
||||
}
|
||||
|
||||
func channelMonitorV2ContainsAny(value string, needles ...string) bool {
|
||||
for _, needle := range needles {
|
||||
if strings.Contains(value, needle) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,433 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
type channelMonitorV2RepoStub struct {
|
||||
config serviceChannelMonitorV2ConfigAlias
|
||||
users *ChannelMonitorV2List[ChannelMonitorV2UserRow]
|
||||
matrix *ChannelMonitorV2Matrix
|
||||
errors *ChannelMonitorV2List[ChannelMonitorV2ErrorRow]
|
||||
snap *ChannelMonitorV2Snapshot
|
||||
group ChannelMonitorV2GroupBy
|
||||
admin bool
|
||||
}
|
||||
|
||||
// Alias keeps composite literals readable without introducing another package.
|
||||
type serviceChannelMonitorV2ConfigAlias = ChannelMonitorV2Config
|
||||
|
||||
func (s *channelMonitorV2RepoStub) GetConfig(context.Context) (*ChannelMonitorV2Config, error) {
|
||||
cfg := ChannelMonitorV2Config(s.config)
|
||||
return &cfg, nil
|
||||
}
|
||||
func (s *channelMonitorV2RepoStub) UpdateConfig(context.Context, ChannelMonitorV2Config, int) (*ChannelMonitorV2Config, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (s *channelMonitorV2RepoStub) GetDimensions(context.Context, ChannelMonitorV2Filter, ChannelMonitorV2Config) (*ChannelMonitorV2Dimensions, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (s *channelMonitorV2RepoStub) GetSnapshot(_ context.Context, _ ChannelMonitorV2Filter, _ ChannelMonitorV2Config, admin bool) (*ChannelMonitorV2Snapshot, error) {
|
||||
s.admin = admin
|
||||
if s.snap == nil {
|
||||
return nil, nil
|
||||
}
|
||||
// Shallow copy so public redaction does not mutate fixture.
|
||||
cfg := s.snap.Config
|
||||
cfg.Platforms = append([]ChannelMonitorV2PlatformConfig(nil), s.snap.Config.Platforms...)
|
||||
for i := range cfg.Platforms {
|
||||
cfg.Platforms[i].Models = append([]string(nil), s.snap.Config.Platforms[i].Models...)
|
||||
}
|
||||
cfg.GroupIDs = append([]int64(nil), s.snap.Config.GroupIDs...)
|
||||
cfg.IgnoredErrorCategories = append([]string(nil), s.snap.Config.IgnoredErrorCategories...)
|
||||
out := *s.snap
|
||||
out.Config = cfg
|
||||
return &out, nil
|
||||
}
|
||||
func (s *channelMonitorV2RepoStub) GetModels(context.Context, ChannelMonitorV2Filter, ChannelMonitorV2Config, bool) (*ChannelMonitorV2List[ChannelMonitorV2ModelRow], error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (s *channelMonitorV2RepoStub) GetMatrix(_ context.Context, _ ChannelMonitorV2Filter, _ ChannelMonitorV2Config, groupBy ChannelMonitorV2GroupBy, admin bool) (*ChannelMonitorV2Matrix, error) {
|
||||
s.group, s.admin = groupBy, admin
|
||||
return s.matrix, nil
|
||||
}
|
||||
func (s *channelMonitorV2RepoStub) GetErrors(_ context.Context, _ ChannelMonitorV2Filter, _ ChannelMonitorV2Config, admin bool) (*ChannelMonitorV2List[ChannelMonitorV2ErrorRow], error) {
|
||||
s.admin = admin
|
||||
if s.errors != nil {
|
||||
// Return a shallow copy so service redaction does not mutate the fixture.
|
||||
out := &ChannelMonitorV2List[ChannelMonitorV2ErrorRow]{
|
||||
Coverage: s.errors.Coverage,
|
||||
Items: append([]ChannelMonitorV2ErrorRow(nil), s.errors.Items...),
|
||||
}
|
||||
for i := range out.Items {
|
||||
if len(out.Items[i].Details) > 0 {
|
||||
out.Items[i].Details = append([]ChannelMonitorV2ErrorDetail(nil), out.Items[i].Details...)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
func (s *channelMonitorV2RepoStub) GetUsers(context.Context, ChannelMonitorV2Filter, ChannelMonitorV2Config, bool) (*ChannelMonitorV2List[ChannelMonitorV2UserRow], error) {
|
||||
return s.users, nil
|
||||
}
|
||||
func (s *channelMonitorV2RepoStub) RecomputeRange(context.Context, time.Time, time.Time) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2ParseFilterDefaultsAndBuckets(t *testing.T) {
|
||||
svc := &ChannelMonitorV2Service{now: func() time.Time { return time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC) }}
|
||||
|
||||
filter, err := svc.ParseFilter("", []string{"openai", "openai", ""}, []string{"gpt-5"}, []int64{2, 1, 2, 0})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "90m", filter.Range)
|
||||
require.Equal(t, 5*time.Minute, filter.Bucket)
|
||||
require.Equal(t, []string{"openai"}, filter.Platforms)
|
||||
require.Equal(t, []int64{1, 2}, filter.GroupIDs)
|
||||
require.Equal(t, 90*time.Minute, filter.End.Sub(filter.Start))
|
||||
|
||||
filter, err = svc.ParseFilter("30d", nil, nil, nil)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 24*time.Hour, filter.Bucket)
|
||||
_, err = svc.ParseFilter("15d", nil, nil, nil)
|
||||
require.ErrorIs(t, err, ErrChannelMonitorV2InvalidRange)
|
||||
}
|
||||
|
||||
func TestParseChannelMonitorV2GroupBy(t *testing.T) {
|
||||
groupBy, err := ParseChannelMonitorV2GroupBy("")
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, ChannelMonitorV2GroupByPlatformGroup, groupBy)
|
||||
for _, value := range []ChannelMonitorV2GroupBy{ChannelMonitorV2GroupByPlatform, ChannelMonitorV2GroupByPlatformGroup, ChannelMonitorV2GroupByPlatformModel, ChannelMonitorV2GroupByPlatformGroupModel} {
|
||||
parsed, parseErr := ParseChannelMonitorV2GroupBy(string(value))
|
||||
require.NoError(t, parseErr)
|
||||
require.Equal(t, value, parsed)
|
||||
}
|
||||
_, err = ParseChannelMonitorV2GroupBy("group")
|
||||
require.ErrorIs(t, err, ErrChannelMonitorV2InvalidGroupBy)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2MatrixForwardsGroupingAndAdminScope(t *testing.T) {
|
||||
want := &ChannelMonitorV2Matrix{GroupBy: ChannelMonitorV2GroupByPlatformModel}
|
||||
repo := &channelMonitorV2RepoStub{config: ChannelMonitorV2Config{Enabled: true}, matrix: want}
|
||||
result, err := NewChannelMonitorV2Service(repo).Matrix(context.Background(), ChannelMonitorV2Filter{}, ChannelMonitorV2GroupByPlatformModel, true)
|
||||
require.NoError(t, err)
|
||||
require.Same(t, want, result)
|
||||
require.Equal(t, ChannelMonitorV2GroupByPlatformModel, repo.group)
|
||||
require.True(t, repo.admin)
|
||||
|
||||
_, err = NewChannelMonitorV2Service(repo).Matrix(context.Background(), ChannelMonitorV2Filter{}, "bad", false)
|
||||
require.ErrorIs(t, err, ErrChannelMonitorV2InvalidGroupBy)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2ConfigValidation(t *testing.T) {
|
||||
cfg := ChannelMonitorV2Config{
|
||||
Platforms: []ChannelMonitorV2PlatformConfig{
|
||||
{Platform: " OpenAI ", Enabled: true, Models: []string{"gpt-5", "gpt-5", ""}},
|
||||
{Platform: "anthropic", Enabled: true},
|
||||
},
|
||||
GroupIDs: []int64{3, 1, 3},
|
||||
}
|
||||
require.NoError(t, normalizeChannelMonitorV2Config(&cfg))
|
||||
require.Equal(t, 300, cfg.RefreshIntervalSeconds)
|
||||
require.Equal(t, "anthropic", cfg.Platforms[0].Platform)
|
||||
require.Equal(t, []int64{1, 3}, cfg.GroupIDs)
|
||||
|
||||
cfg.RefreshIntervalSeconds = 120
|
||||
require.ErrorIs(t, normalizeChannelMonitorV2Config(&cfg), ErrChannelMonitorV2InvalidConfig)
|
||||
|
||||
cfg.RefreshIntervalSeconds = 60
|
||||
cfg.GroupIDs = []int64{0}
|
||||
require.ErrorIs(t, normalizeChannelMonitorV2Config(&cfg), ErrChannelMonitorV2InvalidConfig)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2ErrorTaxonomyPriority(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
in ChannelMonitorV2ErrorInput
|
||||
want string
|
||||
}{
|
||||
{"cyber", ChannelMonitorV2ErrorInput{ErrorType: "cyber_policy", StatusCode: 200}, "content_policy"},
|
||||
{"auth before forbidden", ChannelMonitorV2ErrorInput{StatusCode: 403, Message: "invalid API key"}, "authentication"},
|
||||
{"context", ChannelMonitorV2ErrorInput{Message: "maximum prompt length exceeded"}, "context_limit"},
|
||||
{"unsupported", ChannelMonitorV2ErrorInput{Message: "not supported by any configured account"}, "model_unsupported"},
|
||||
{"pool", ChannelMonitorV2ErrorInput{Message: "No available accounts"}, "account_pool_unavailable"},
|
||||
{"timeout", ChannelMonitorV2ErrorInput{Message: "error code: 524"}, "timeout"},
|
||||
{"upstream", ChannelMonitorV2ErrorInput{ErrorOwner: "provider", StatusCode: 502, UpstreamStatusCode: 502}, "upstream_5xx"},
|
||||
{"other", ChannelMonitorV2ErrorInput{StatusCode: 400, Message: "unknown"}, "other"},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) { require.Equal(t, tt.want, ClassifyChannelMonitorV2Error(tt.in)) })
|
||||
}
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2HealthBlendsErrorTTFTAndCache(t *testing.T) {
|
||||
// error 3%/5% → 40; ttft p50 2s → 100; cache 50% → 50
|
||||
// overall = (0.6*40 + 0.2*100 + 0.2*50) / 1.0 = 54 → warning
|
||||
p50 := int64(2000)
|
||||
p95 := int64(9000)
|
||||
thresholds := ChannelMonitorV2HealthThresholds{
|
||||
MinimumSample: 20,
|
||||
WarningErrorRate: 0.02,
|
||||
CriticalErrorRate: 0.05,
|
||||
TargetTTFTMs: 2500,
|
||||
WarningTTFTMs: 2501,
|
||||
CriticalTTFTMs: 6000,
|
||||
WarningCacheRate: 0.20,
|
||||
CriticalCacheRate: 0.05,
|
||||
ErrorWeight: 0.60,
|
||||
TTFTWeight: 0.20,
|
||||
CacheWeight: 0.20,
|
||||
}
|
||||
metrics := ChannelMonitorV2Metric{
|
||||
RequestCount: 100,
|
||||
ErrorRate: 0.03,
|
||||
CacheRate: 0.50,
|
||||
CacheRateDenominator: 100,
|
||||
TTFT: ChannelMonitorV2Latency{SampleCount: 100, P50Ms: &p50, P95Ms: &p95},
|
||||
}
|
||||
health := ChannelMonitorV2HealthForWithThresholds(metrics, thresholds)
|
||||
require.Equal(t, "warning", health.ErrorRate)
|
||||
require.Equal(t, "healthy", health.TTFT)
|
||||
require.Equal(t, "healthy", health.Cache)
|
||||
require.NotNil(t, health.Score)
|
||||
require.NotNil(t, health.CacheScore)
|
||||
require.InDelta(t, 50.0, *health.CacheScore, 0.01)
|
||||
require.InDelta(t, 54.0, *health.Score, 0.01)
|
||||
require.Equal(t, "warning", health.Overall)
|
||||
|
||||
// Perfect signals → 100
|
||||
p50OK := int64(1000)
|
||||
metrics.ErrorRate = 0
|
||||
metrics.CacheRate = 1
|
||||
metrics.TTFT.P50Ms = &p50OK
|
||||
health = ChannelMonitorV2HealthForWithThresholds(metrics, thresholds)
|
||||
require.Equal(t, "healthy", health.Overall)
|
||||
require.NotNil(t, health.Score)
|
||||
require.InDelta(t, 100.0, *health.Score, 0.01)
|
||||
|
||||
// Small samples stay unknown
|
||||
health = ChannelMonitorV2HealthFor(ChannelMonitorV2Metric{RequestCount: 2})
|
||||
require.Equal(t, "unknown", health.Overall)
|
||||
require.Nil(t, health.Score)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2DefaultHealthThresholdsAreTolerant(t *testing.T) {
|
||||
p50 := int64(2500)
|
||||
health := ChannelMonitorV2HealthFor(ChannelMonitorV2Metric{
|
||||
RequestCount: 100,
|
||||
ErrorRate: 0.03,
|
||||
CacheRate: 0,
|
||||
CacheRateDenominator: 100,
|
||||
TTFT: ChannelMonitorV2Latency{SampleCount: 100, P50Ms: &p50},
|
||||
})
|
||||
require.Equal(t, "healthy", health.ErrorRate)
|
||||
require.Equal(t, "healthy", health.TTFT)
|
||||
require.Equal(t, "healthy", health.Cache)
|
||||
require.Equal(t, "healthy", health.Overall)
|
||||
require.NotNil(t, health.CacheScore)
|
||||
require.InDelta(t, 100.0, *health.CacheScore, 0.01)
|
||||
}
|
||||
|
||||
func TestErrorRateTTFTAndCacheScoreHelpers(t *testing.T) {
|
||||
require.InDelta(t, 100.0, errorRateScore(0, 0.05), 0.001)
|
||||
require.InDelta(t, 0.0, errorRateScore(0.05, 0.05), 0.001)
|
||||
require.InDelta(t, 40.0, errorRateScore(0.03, 0.05), 0.001)
|
||||
|
||||
require.InDelta(t, 100.0, ttftP50Score(2500, 2500, 6000), 0.001)
|
||||
require.InDelta(t, 100.0, ttftP50Score(1000, 2500, 6000), 0.001)
|
||||
require.InDelta(t, 0.0, ttftP50Score(6000, 2500, 6000), 0.001)
|
||||
require.InDelta(t, 50.0, ttftP50Score(4250, 2500, 6000), 0.001)
|
||||
|
||||
require.InDelta(t, 0.0, cacheRateScore(0), 0.001)
|
||||
require.InDelta(t, 100.0, cacheRateScore(1), 0.001)
|
||||
require.InDelta(t, 50.0, cacheRateScore(0.5), 0.001)
|
||||
require.Equal(t, "critical", cacheRateBand(0.02, 0.20, 0.05))
|
||||
require.Equal(t, "warning", cacheRateBand(0.10, 0.20, 0.05))
|
||||
require.Equal(t, "healthy", cacheRateBand(0.50, 0.20, 0.05))
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2UsersRemovesOtherUserIdentity(t *testing.T) {
|
||||
selfID, otherID := int64(7), int64(9)
|
||||
repo := &channelMonitorV2RepoStub{config: ChannelMonitorV2Config{Enabled: true}, users: &ChannelMonitorV2List[ChannelMonitorV2UserRow]{Items: []ChannelMonitorV2UserRow{
|
||||
{UserID: &otherID, Email: "other@example.com", Username: "other"},
|
||||
{UserID: &selfID, Email: "self@example.com", Username: "self"},
|
||||
}}}
|
||||
result, err := NewChannelMonitorV2Service(repo).Users(context.Background(), ChannelMonitorV2Filter{}, selfID, false)
|
||||
require.NoError(t, err)
|
||||
require.Nil(t, result.Items[0].UserID)
|
||||
require.Empty(t, result.Items[0].Email)
|
||||
require.Equal(t, "Other user #1", result.Items[0].DisplayLabel)
|
||||
require.Equal(t, selfID, *result.Items[1].UserID)
|
||||
require.True(t, result.Items[1].IsSelf)
|
||||
require.Equal(t, "Me", result.Items[1].DisplayLabel)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2UsersAppendsSelfWhenMissingFromRanking(t *testing.T) {
|
||||
selfID, otherID := int64(7), int64(9)
|
||||
repo := &channelMonitorV2RepoStub{config: ChannelMonitorV2Config{Enabled: true}, users: &ChannelMonitorV2List[ChannelMonitorV2UserRow]{Items: []ChannelMonitorV2UserRow{
|
||||
{UserID: &otherID, Email: "other@example.com", Username: "other", Metrics: ChannelMonitorV2Metric{RequestCount: 10}},
|
||||
}}}
|
||||
result, err := NewChannelMonitorV2Service(repo).Users(context.Background(), ChannelMonitorV2Filter{}, selfID, false)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, result.Items, 2)
|
||||
self := result.Items[1]
|
||||
require.True(t, self.IsSelf)
|
||||
require.Equal(t, "Me", self.DisplayLabel)
|
||||
require.Equal(t, selfID, *self.UserID)
|
||||
require.Equal(t, 0, self.Rank) // unranked / no traffic in window
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2TopUsersKeepsSelfOutsideLimit(t *testing.T) {
|
||||
items := make([]ChannelMonitorV2UserRow, 0, 12)
|
||||
for i := 1; i <= 12; i++ {
|
||||
id := int64(i)
|
||||
items = append(items, ChannelMonitorV2UserRow{UserID: &id, Rank: i, DisplayLabel: fmt.Sprintf("u%d", i)})
|
||||
}
|
||||
// self is rank 12 (index 11)
|
||||
out := channelMonitorV2TopUsersWithSelf(items, 11, 10)
|
||||
require.Len(t, out, 11)
|
||||
require.Equal(t, int64(12), *out[10].UserID)
|
||||
require.Equal(t, 12, out[10].Rank)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2ReadAPIsRejectDisabledConfig(t *testing.T) {
|
||||
repo := &channelMonitorV2RepoStub{config: ChannelMonitorV2Config{Enabled: false}}
|
||||
_, err := NewChannelMonitorV2Service(repo).Matrix(context.Background(), ChannelMonitorV2Filter{}, ChannelMonitorV2GroupByPlatform, false)
|
||||
require.ErrorIs(t, err, ErrChannelMonitorDisabled)
|
||||
}
|
||||
|
||||
func TestNormalizeChannelMonitorV2IgnoredCategories(t *testing.T) {
|
||||
got := normalizeChannelMonitorV2IgnoredCategories([]string{"timeout", "TIMEOUT", "not-a-real", "", "timeout", "other"})
|
||||
require.Equal(t, []string{"other", "timeout"}, got)
|
||||
}
|
||||
|
||||
func TestApplyIgnoredErrorsAdjustsRateOnly(t *testing.T) {
|
||||
// applyIgnoredErrors lives in repository package; mirror the scored-rate formula.
|
||||
// scored errors = total_errors - ignored; error_rate = scored / request_count
|
||||
// success_rate stays SuccessRequests / RequestCount (ignored ≠ success).
|
||||
m := ChannelMonitorV2Metric{RequestCount: 100, ErrorRequests: 20, SuccessRequests: 80, ErrorRate: 0.2, SuccessRate: 0.8}
|
||||
ignored := int64(5)
|
||||
counted := m.ErrorRequests - ignored
|
||||
errorRate := float64(counted) / float64(m.RequestCount)
|
||||
successRate := float64(m.SuccessRequests) / float64(m.RequestCount)
|
||||
require.InDelta(t, 0.15, errorRate, 0.0001)
|
||||
require.InDelta(t, 0.80, successRate, 0.0001)
|
||||
require.Equal(t, int64(20), m.ErrorRequests)
|
||||
require.Equal(t, int64(100), m.RequestCount)
|
||||
}
|
||||
|
||||
func TestErrorsForViewerStripsDetailsAndCountsForNonAdmin(t *testing.T) {
|
||||
fixture := &ChannelMonitorV2List[ChannelMonitorV2ErrorRow]{
|
||||
Items: []ChannelMonitorV2ErrorRow{{
|
||||
Category: "timeout",
|
||||
Count: 42,
|
||||
Rate: 0.4,
|
||||
Details: []ChannelMonitorV2ErrorDetail{{
|
||||
Platform: "anthropic",
|
||||
Model: "claude",
|
||||
ErrorType: "upstream",
|
||||
StatusCode: 504,
|
||||
UpstreamStatusCode: 504,
|
||||
Message: "gateway timeout sk-secret",
|
||||
Count: 12,
|
||||
}},
|
||||
}},
|
||||
}
|
||||
repo := &channelMonitorV2RepoStub{config: ChannelMonitorV2Config{Enabled: true}, errors: fixture}
|
||||
svc := NewChannelMonitorV2Service(repo)
|
||||
|
||||
userList, err := svc.ErrorsForViewer(context.Background(), ChannelMonitorV2Filter{}, false)
|
||||
require.NoError(t, err)
|
||||
require.False(t, repo.admin)
|
||||
require.Len(t, userList.Items, 1)
|
||||
require.Zero(t, userList.Items[0].Count)
|
||||
require.Empty(t, userList.Items[0].Details)
|
||||
require.InDelta(t, 0.4, userList.Items[0].Rate, 0.0001)
|
||||
|
||||
adminList, err := svc.ErrorsForViewer(context.Background(), ChannelMonitorV2Filter{}, true)
|
||||
require.NoError(t, err)
|
||||
require.True(t, repo.admin)
|
||||
require.Equal(t, int64(42), adminList.Items[0].Count)
|
||||
require.Len(t, adminList.Items[0].Details, 1)
|
||||
require.Contains(t, adminList.Items[0].Details[0].Message, "gateway timeout")
|
||||
}
|
||||
|
||||
func TestSnapshotRedactsPublicConfigPolicyFields(t *testing.T) {
|
||||
updatedBy := int64(9)
|
||||
repo := &channelMonitorV2RepoStub{
|
||||
config: ChannelMonitorV2Config{Enabled: true},
|
||||
snap: &ChannelMonitorV2Snapshot{
|
||||
Config: ChannelMonitorV2Config{
|
||||
Version: 3,
|
||||
Enabled: true,
|
||||
RefreshIntervalSeconds: 300,
|
||||
Platforms: []ChannelMonitorV2PlatformConfig{
|
||||
{Platform: "openai", Enabled: true, Models: []string{"gpt-5"}},
|
||||
},
|
||||
GroupIDs: []int64{1, 2},
|
||||
IgnoredErrorCategories: []string{"timeout"},
|
||||
UpdatedBy: &updatedBy,
|
||||
},
|
||||
Metrics: ChannelMonitorV2Metric{RequestCount: 100, ErrorRate: 0.1, SuccessRate: 0.9, RPM: 5},
|
||||
},
|
||||
}
|
||||
svc := NewChannelMonitorV2Service(repo)
|
||||
snap, err := svc.Snapshot(context.Background(), ChannelMonitorV2Filter{}, false)
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, snap.Config.GroupIDs)
|
||||
require.Empty(t, snap.Config.IgnoredErrorCategories)
|
||||
require.Nil(t, snap.Config.UpdatedBy)
|
||||
require.Empty(t, snap.Config.Platforms[0].Models)
|
||||
require.Equal(t, 300, snap.Config.RefreshIntervalSeconds)
|
||||
require.Zero(t, snap.Metrics.RequestCount)
|
||||
require.InDelta(t, 0.1, snap.Metrics.ErrorRate, 0.0001)
|
||||
require.InDelta(t, 5.0, snap.Metrics.RPM, 0.0001)
|
||||
}
|
||||
|
||||
func TestRedactChannelMonitorV2MetricKeepsRates(t *testing.T) {
|
||||
p50 := int64(100)
|
||||
m := ChannelMonitorV2Metric{
|
||||
SuccessRequests: 90, ErrorRequests: 10, RequestCount: 100,
|
||||
InputTokens: 1, OutputTokens: 2, CacheCreationTokens: 3, CacheReadTokens: 4,
|
||||
TokenCount: 10, RPM: 12, TPM: 34, ErrorRate: 0.1, SuccessRate: 0.9,
|
||||
CacheRate: 0.5, CacheRateNumerator: 4, CacheRateDenominator: 8,
|
||||
TTFT: ChannelMonitorV2Latency{SampleCount: 50, P50Ms: &p50},
|
||||
}
|
||||
upstream := int64(3)
|
||||
m.UpstreamAffectedRequests = &upstream
|
||||
redactChannelMonitorV2Metric(&m, false)
|
||||
require.Zero(t, m.RequestCount)
|
||||
require.Zero(t, m.ErrorRequests)
|
||||
require.Zero(t, m.TokenCount)
|
||||
require.Zero(t, m.CacheRateNumerator)
|
||||
require.Zero(t, m.TTFT.SampleCount)
|
||||
require.Nil(t, m.UpstreamAffectedRequests)
|
||||
require.InDelta(t, 0.1, m.ErrorRate, 0.0001)
|
||||
require.InDelta(t, 12.0, m.RPM, 0.0001)
|
||||
require.InDelta(t, 34.0, m.TPM, 0.0001)
|
||||
require.NotNil(t, m.TTFT.P50Ms)
|
||||
require.Equal(t, int64(100), *m.TTFT.P50Ms)
|
||||
|
||||
redactChannelMonitorV2Metric(&m, true)
|
||||
require.Zero(t, m.RPM)
|
||||
require.Zero(t, m.TPM)
|
||||
require.InDelta(t, 0.1, m.ErrorRate, 0.0001)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2HealthTTFTAtTargetIsHealthy(t *testing.T) {
|
||||
p50 := int64(2500)
|
||||
m := ChannelMonitorV2Metric{
|
||||
RequestCount: 100, ErrorRate: 0,
|
||||
CacheRate: 1, CacheRateDenominator: 100,
|
||||
TTFT: ChannelMonitorV2Latency{SampleCount: 100, P50Ms: &p50},
|
||||
}
|
||||
h := ChannelMonitorV2HealthFor(m)
|
||||
require.Equal(t, "healthy", h.TTFT)
|
||||
require.NotNil(t, h.TTFTScore)
|
||||
require.InDelta(t, 100.0, *h.TTFTScore, 0.01)
|
||||
}
|
||||
Reference in New Issue
Block a user