mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-14 03:52:26 +08:00
fix(channel-monitor-v2): default mode v1 and gentle adaptive backfill
Default channel_monitor_mode to v1 (opt-in V2) so upgrades keep active probes; existing explicit v2 rows are left alone via ON CONFLICT DO NOTHING plus migration checksum compatibility for already-applied 195. V2 first-enable backfill no longer compresses ticks to 5s or uses 24h chunks. Each tick does recent overlap plus at most one historical chunk with depth-based ceilings (2h/4h/6h), adaptive grow/shrink, and failure backoff. Error request_id dedup is bounded by a 90-minute lookback so ops_error_logs is not scanned for full history. Also align hide_throughput parse default with privacy-preserving public runtime (missing key → true).
This commit is contained in:
@@ -238,6 +238,8 @@ WHEN ` + column + ` <= 300000 THEN 300000 WHEN ` + column + ` <= 600000 THEN 600
|
||||
ELSE 2147483647 END`
|
||||
}
|
||||
|
||||
// Error dedup lookback: request_id branch is bounded by chunk start minus 90
|
||||
// minutes so candidate_ids never forces a full-history scan of ops_error_logs.
|
||||
const channelMonitorV2ErrorAggregationSQL = `
|
||||
WITH dedup AS (
|
||||
WITH candidate_ids AS MATERIALIZED (
|
||||
@@ -259,7 +261,11 @@ WITH dedup AS (
|
||||
FROM ops_error_logs current_error
|
||||
WHERE (
|
||||
(NULLIF(current_error.request_id, '') IS NULL AND current_error.created_at >= $1 AND current_error.created_at < $2)
|
||||
OR current_error.request_id IN (SELECT request_id FROM candidate_ids)
|
||||
OR (
|
||||
current_error.request_id IN (SELECT request_id FROM candidate_ids)
|
||||
AND current_error.created_at >= $1 - INTERVAL '90 minutes'
|
||||
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')
|
||||
|
||||
@@ -102,6 +102,9 @@ func TestChannelMonitorV2ErrorAggregationCountsFinalUserErrorsOnly(t *testing.T)
|
||||
require.Contains(t, query, "where bucket_start >= $1 and bucket_start < $2")
|
||||
require.Contains(t, query, "upstream_affected_requests")
|
||||
require.Contains(t, query, "jsonb_array_length(upstream_errors) > 0")
|
||||
// request_id dedup must be time-bounded (no full-history scan).
|
||||
require.Contains(t, query, "interval '90 minutes'")
|
||||
require.Contains(t, query, "current_error.created_at >= $1 - interval '90 minutes'")
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2UsageSuccessExcludesCyberBillingRows(t *testing.T) {
|
||||
|
||||
@@ -81,6 +81,9 @@ var migrationChecksumCompatibilityRules = map[string]migrationChecksumCompatibil
|
||||
"123_fix_legacy_auth_source_grant_on_signup_defaults.sql": newMigrationChecksumCompatibilityRule("2ce43c2cd89e9f9e1febd34a407ed9e84d177386c5544b6f02c1f58a21129f57", "6cd33422f215dcd1f486ab6f35c0ea5805d9ca69bb25906d94bc649156657145"),
|
||||
"159_batch_image_foundation.sql": newMigrationChecksumCompatibilityRule("d902b70982025ec519749faf058aab7631e82c3f48167b9a4ae4db718eb72cce", "82da85b5d98e67a0507647b873a40373e84538e4adafdeed6767c0ac8b6570b2"),
|
||||
"161_batch_image_pricing_snapshot.sql": newMigrationChecksumCompatibilityRule("4012af3e43636cb6af22e0176d59d1fcc70615c0f310194329461ae462c4fbd6", "96d915c9b7a6941ae99039e0ff3f1a61481eb9bddd933d11c6fadb2274554e87"),
|
||||
// 195 originally seeded mode=v2; flipped to v1 (safe default / opt-in v2). Existing DBs
|
||||
// that already applied the v2 seed keep their row and the historical checksum.
|
||||
"195_channel_monitor_mode.sql": newMigrationChecksumCompatibilityRule("13f3792f3e3e53ee96e26415c884cf8062c77172824b54fcc9a8c0c2b1f185ec", "4c74fe33ef2274cc72e1bb49671e651274532c034b29f5b2982c2a4c88d101a6"),
|
||||
}
|
||||
|
||||
// ApplyMigrations 将嵌入的 SQL 迁移文件应用到指定的数据库。
|
||||
|
||||
@@ -976,7 +976,7 @@ func TestAPIContracts(t *testing.T) {
|
||||
"balance_low_notify_recharge_url": "",
|
||||
"account_quota_notify_emails": [],
|
||||
"channel_monitor_enabled": true,
|
||||
"channel_monitor_mode": "v2",
|
||||
"channel_monitor_mode": "v1",
|
||||
"channel_monitor_hide_throughput": true,
|
||||
"channel_monitor_default_interval_seconds": 60,
|
||||
"available_channels_enabled": false,
|
||||
@@ -1281,7 +1281,7 @@ func TestAPIContracts(t *testing.T) {
|
||||
"balance_low_notify_recharge_url": "",
|
||||
"account_quota_notify_emails": [],
|
||||
"channel_monitor_enabled": true,
|
||||
"channel_monitor_mode": "v2",
|
||||
"channel_monitor_mode": "v1",
|
||||
"channel_monitor_hide_throughput": true,
|
||||
"channel_monitor_default_interval_seconds": 60,
|
||||
"available_channels_enabled": false,
|
||||
|
||||
@@ -48,10 +48,10 @@ func TestRunCheck_NilRuntimeReaderFailsClosedAsV2(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestNormalizeChannelMonitorMode(t *testing.T) {
|
||||
require.Equal(t, ChannelMonitorModeV2, normalizeChannelMonitorMode(""))
|
||||
require.Equal(t, ChannelMonitorModeV1, normalizeChannelMonitorMode(""))
|
||||
require.Equal(t, ChannelMonitorModeV1, normalizeChannelMonitorMode("v1"))
|
||||
require.Equal(t, ChannelMonitorModeV2, normalizeChannelMonitorMode("v2"))
|
||||
require.Equal(t, ChannelMonitorModeV2, normalizeChannelMonitorMode("invalid"))
|
||||
require.Equal(t, ChannelMonitorModeV1, normalizeChannelMonitorMode("invalid"))
|
||||
require.Equal(t, ChannelMonitorModeV1, normalizeChannelMonitorMode(" V1 "))
|
||||
}
|
||||
|
||||
|
||||
@@ -20,11 +20,22 @@ const (
|
||||
channelMonitorV2BootstrapWindow = ChannelMonitorV2BootstrapProductWindow
|
||||
// First tick after upgrade prioritizes the default 90m view (with small padding).
|
||||
channelMonitorV2BootstrapFirst = 2 * time.Hour
|
||||
// Subsequent bootstrap chunks grow so 24h/7d/30d fill without waiting full 90d pace.
|
||||
// Order after first tick: 22h → full 1d chunks until 30d, then 1d toward 90d.
|
||||
channelMonitorV2RecentOverlap = 10 * time.Minute
|
||||
channelMonitorV2BackfillChunk = 24 * time.Hour
|
||||
channelMonitorV2MinBackfillChunk = time.Hour
|
||||
// Always refresh a small trailing window so late writes land without
|
||||
// re-aggregating large history every tick.
|
||||
channelMonitorV2RecentOverlap = 10 * time.Minute
|
||||
|
||||
// Gentle backfill: small adaptive chunks, never default 24h hammering.
|
||||
// Initial historical chunk after the 2h seed.
|
||||
channelMonitorV2BackfillChunkInit = time.Hour
|
||||
channelMonitorV2MinBackfillChunk = 15 * time.Minute
|
||||
// Depth-based ceilings (product phases 90m → 1d → 7d → 30d → 90d).
|
||||
channelMonitorV2MaxChunkNear1d = 2 * time.Hour
|
||||
channelMonitorV2MaxChunkNear7d = 4 * time.Hour
|
||||
channelMonitorV2MaxChunkFar = 6 * time.Hour
|
||||
|
||||
// Soft adaptive timing: grow only when a recompute is clearly cheap.
|
||||
channelMonitorV2GrowChunkUnder = 15 * time.Second
|
||||
channelMonitorV2MaxBackoff = 10 * time.Minute
|
||||
)
|
||||
|
||||
// channelMonitorRuntimeSubscriber is the optional settings hook that lets the
|
||||
@@ -49,6 +60,8 @@ type ChannelMonitorV2Aggregator struct {
|
||||
backfillAt time.Time
|
||||
backfillChunk time.Duration
|
||||
backfillFailures int
|
||||
// nextWaitFloor is applied after runOnce when failures require backoff.
|
||||
nextWaitFloor time.Duration
|
||||
// cursorLoaded is true after the first successful watermark read (or init).
|
||||
cursorLoaded bool
|
||||
// hasAggregated is true once any recompute in this process (or durable data) exists.
|
||||
@@ -66,7 +79,7 @@ func NewChannelMonitorV2Aggregator(repo ChannelMonitorV2Repository, db *sql.DB,
|
||||
instanceID: uuid.NewString(),
|
||||
stopCh: make(chan struct{}),
|
||||
kickCh: make(chan struct{}, 1),
|
||||
backfillChunk: channelMonitorV2BackfillChunk,
|
||||
backfillChunk: channelMonitorV2BackfillChunkInit,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -161,12 +174,13 @@ func (s *ChannelMonitorV2Aggregator) loop() {
|
||||
}
|
||||
cancel()
|
||||
s.runOnce()
|
||||
// While product bootstrap is incomplete, tick faster so 90m/24h/7d/30d
|
||||
// fill without waiting a full refresh_interval between historical chunks.
|
||||
if s.bootstrapActive() {
|
||||
if interval > 5*time.Second {
|
||||
interval = 5 * time.Second
|
||||
}
|
||||
// Hard gate: never compress bootstrap to multi-Hz ticks. Soft gate: on
|
||||
// repeated failures raise the wait floor (exponential backoff).
|
||||
s.mu.Lock()
|
||||
floor := s.nextWaitFloor
|
||||
s.mu.Unlock()
|
||||
if floor > interval {
|
||||
interval = floor
|
||||
}
|
||||
if !s.wait(interval) {
|
||||
return
|
||||
@@ -250,14 +264,13 @@ func (s *ChannelMonitorV2Aggregator) runOnce() {
|
||||
// Phase 1 (first upgrade / empty): seed the default 90m UI window quickly.
|
||||
if !hasData || cursor.IsZero() {
|
||||
start := now.Add(-channelMonitorV2BootstrapFirst)
|
||||
started := time.Now()
|
||||
if err := s.repo.RecomputeRange(ctx, start, now); err != nil {
|
||||
logger.LegacyPrintf("service.channel_monitor_v2", "[ChannelMonitorV2] bootstrap recent aggregation failed: %v", err)
|
||||
s.recordBackfillFailure(now, cursor)
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
s.backfillAt = start
|
||||
s.hasAggregated = true
|
||||
s.mu.Unlock()
|
||||
s.recordBackfillSuccess(start, time.Since(started), now)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -267,27 +280,34 @@ func (s *ChannelMonitorV2Aggregator) runOnce() {
|
||||
return
|
||||
}
|
||||
|
||||
// Phase 2: walk history backward in chunks until retention max (90d).
|
||||
// Phase 2: walk history backward at most one chunk per tick until retention max (90d).
|
||||
// Product UI (30d) fills first; remaining 30–90d continues silently.
|
||||
retentionCutoff := now.Add(-channelMonitorV2RetentionMax)
|
||||
if !cursor.After(retentionCutoff) {
|
||||
return
|
||||
}
|
||||
// Walk backward in 1d chunks. After the 2h seed, the next chunk reaches
|
||||
// past 24h so the default + daily UI ranges fill within a few ticks; 7d/30d
|
||||
// follow as more chunks complete (banner tracks progress against 30d).
|
||||
end := cursor
|
||||
s.mu.Lock()
|
||||
chunk := s.backfillChunk
|
||||
s.mu.Unlock()
|
||||
if chunk <= 0 {
|
||||
chunk = channelMonitorV2BackfillChunk
|
||||
chunk = channelMonitorV2BackfillChunkInit
|
||||
}
|
||||
maxChunk := channelMonitorV2MaxChunkForDepth(now, end)
|
||||
if chunk > maxChunk {
|
||||
chunk = maxChunk
|
||||
}
|
||||
if chunk < channelMonitorV2MinBackfillChunk {
|
||||
chunk = channelMonitorV2MinBackfillChunk
|
||||
}
|
||||
start := end.Add(-chunk)
|
||||
// Once bootstrap reaches historical data, keep chunks on day boundaries so
|
||||
// daily rollups never depend on 1m rows from two independently pruned chunks.
|
||||
if end.Before(now.Add(-7 * 24 * time.Hour)) {
|
||||
start = end.Add(-chunk).Truncate(24 * time.Hour)
|
||||
aligned := end.Add(-chunk).Truncate(24 * time.Hour)
|
||||
if aligned.Before(end) {
|
||||
start = aligned
|
||||
}
|
||||
}
|
||||
if start.Before(retentionCutoff) {
|
||||
start = retentionCutoff
|
||||
@@ -295,25 +315,87 @@ func (s *ChannelMonitorV2Aggregator) runOnce() {
|
||||
if !start.Before(end) {
|
||||
return
|
||||
}
|
||||
started := time.Now()
|
||||
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)
|
||||
s.mu.Lock()
|
||||
s.backfillFailures++
|
||||
if s.backfillChunk > channelMonitorV2MinBackfillChunk {
|
||||
s.backfillChunk /= 2
|
||||
if s.backfillChunk < channelMonitorV2MinBackfillChunk {
|
||||
s.backfillChunk = channelMonitorV2MinBackfillChunk
|
||||
}
|
||||
}
|
||||
s.mu.Unlock()
|
||||
s.recordBackfillFailure(now, end)
|
||||
return
|
||||
}
|
||||
s.recordBackfillSuccess(start, time.Since(started), now)
|
||||
}
|
||||
|
||||
// channelMonitorV2MaxChunkForDepth returns the hard ceiling for a historical
|
||||
// chunk ending at `end` (earliest already covered / next walk end).
|
||||
func channelMonitorV2MaxChunkForDepth(now, end time.Time) time.Duration {
|
||||
age := now.Sub(end)
|
||||
switch {
|
||||
case age < 24*time.Hour:
|
||||
return channelMonitorV2MaxChunkNear1d
|
||||
case age < 7*24*time.Hour:
|
||||
return channelMonitorV2MaxChunkNear7d
|
||||
default:
|
||||
return channelMonitorV2MaxChunkFar
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Aggregator) recordBackfillSuccess(coveredFrom time.Time, elapsed time.Duration, now time.Time) {
|
||||
s.mu.Lock()
|
||||
s.backfillChunk = channelMonitorV2BackfillChunk
|
||||
s.backfillFailures = 0
|
||||
s.backfillAt = start
|
||||
defer s.mu.Unlock()
|
||||
s.backfillAt = coveredFrom
|
||||
s.hasAggregated = true
|
||||
s.mu.Unlock()
|
||||
s.backfillFailures = 0
|
||||
s.nextWaitFloor = 0
|
||||
maxChunk := channelMonitorV2MaxChunkForDepth(now, coveredFrom)
|
||||
// Grow slowly only when the recompute was clearly cheap.
|
||||
if elapsed > 0 && elapsed < channelMonitorV2GrowChunkUnder {
|
||||
next := s.backfillChunk
|
||||
if next <= 0 {
|
||||
next = channelMonitorV2BackfillChunkInit
|
||||
}
|
||||
next = time.Duration(float64(next) * 1.5)
|
||||
if next > maxChunk {
|
||||
next = maxChunk
|
||||
}
|
||||
if next < channelMonitorV2MinBackfillChunk {
|
||||
next = channelMonitorV2MinBackfillChunk
|
||||
}
|
||||
s.backfillChunk = next
|
||||
return
|
||||
}
|
||||
// Keep a healthy chunk within the depth ceiling after success.
|
||||
if s.backfillChunk <= 0 || s.backfillChunk > maxChunk {
|
||||
s.backfillChunk = channelMonitorV2BackfillChunkInit
|
||||
if s.backfillChunk > maxChunk {
|
||||
s.backfillChunk = maxChunk
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ChannelMonitorV2Aggregator) recordBackfillFailure(now, end time.Time) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.backfillFailures++
|
||||
// Shrink chunk toward the minimum so large DBs self-throttle.
|
||||
if s.backfillChunk <= 0 {
|
||||
s.backfillChunk = channelMonitorV2BackfillChunkInit
|
||||
}
|
||||
s.backfillChunk /= 2
|
||||
if s.backfillChunk < channelMonitorV2MinBackfillChunk {
|
||||
s.backfillChunk = channelMonitorV2MinBackfillChunk
|
||||
}
|
||||
maxChunk := channelMonitorV2MaxChunkForDepth(now, end)
|
||||
if s.backfillChunk > maxChunk {
|
||||
s.backfillChunk = maxChunk
|
||||
}
|
||||
// Exponential backoff on wait floor: 1m, 2m, 4m… capped at 10m.
|
||||
floor := time.Minute << uint(s.backfillFailures-1)
|
||||
if floor > channelMonitorV2MaxBackoff {
|
||||
floor = channelMonitorV2MaxBackoff
|
||||
}
|
||||
if floor < time.Minute {
|
||||
floor = time.Minute
|
||||
}
|
||||
s.nextWaitFloor = floor
|
||||
}
|
||||
|
||||
// ensureCursor restores durable backfill_cursor after process restart so progress
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
//go:build unit
|
||||
|
||||
package service
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestChannelMonitorV2MaxChunkForDepth(t *testing.T) {
|
||||
now := time.Date(2026, 8, 8, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
// Within last day → tightest ceiling (2h).
|
||||
require.Equal(t, channelMonitorV2MaxChunkNear1d, channelMonitorV2MaxChunkForDepth(now, now.Add(-2*time.Hour)))
|
||||
// Between 1d and 7d → 4h.
|
||||
require.Equal(t, channelMonitorV2MaxChunkNear7d, channelMonitorV2MaxChunkForDepth(now, now.Add(-2*24*time.Hour)))
|
||||
// Older than 7d → 6h (never 24h default).
|
||||
require.Equal(t, channelMonitorV2MaxChunkFar, channelMonitorV2MaxChunkForDepth(now, now.Add(-10*24*time.Hour)))
|
||||
require.Less(t, channelMonitorV2MaxChunkFar, 24*time.Hour)
|
||||
require.Equal(t, time.Hour, channelMonitorV2BackfillChunkInit)
|
||||
require.Equal(t, 15*time.Minute, channelMonitorV2MinBackfillChunk)
|
||||
}
|
||||
|
||||
func TestChannelMonitorV2AggregatorAdaptiveChunk(t *testing.T) {
|
||||
s := NewChannelMonitorV2Aggregator(nil, nil, nil)
|
||||
now := time.Date(2026, 8, 8, 12, 0, 0, 0, time.UTC)
|
||||
cursor := now.Add(-3 * time.Hour)
|
||||
|
||||
// Failure shrinks chunk and sets backoff floor.
|
||||
s.backfillChunk = 2 * time.Hour
|
||||
s.recordBackfillFailure(now, cursor)
|
||||
require.Equal(t, time.Hour, s.backfillChunk)
|
||||
require.Equal(t, time.Minute, s.nextWaitFloor)
|
||||
require.Equal(t, 1, s.backfillFailures)
|
||||
|
||||
// Repeated failure halves again and raises floor.
|
||||
s.recordBackfillFailure(now, cursor)
|
||||
require.Equal(t, 30*time.Minute, s.backfillChunk)
|
||||
require.Equal(t, 2*time.Minute, s.nextWaitFloor)
|
||||
|
||||
// Fast success grows within depth ceiling and clears backoff.
|
||||
s.recordBackfillSuccess(cursor.Add(-30*time.Minute), 5*time.Second, now)
|
||||
require.Equal(t, 0, s.backfillFailures)
|
||||
require.Equal(t, time.Duration(0), s.nextWaitFloor)
|
||||
require.Greater(t, s.backfillChunk, 30*time.Minute)
|
||||
require.LessOrEqual(t, s.backfillChunk, channelMonitorV2MaxChunkForDepth(now, cursor.Add(-30*time.Minute)))
|
||||
}
|
||||
@@ -394,7 +394,7 @@ const (
|
||||
SettingKeyChannelMonitorEnabled = "channel_monitor_enabled"
|
||||
|
||||
// SettingKeyChannelMonitorMode selects exclusive implementation:
|
||||
// "v1" active probes, "v2" passive aggregation. Default "v2".
|
||||
// "v1" active probes, "v2" passive aggregation. Default "v1" (opt-in to v2).
|
||||
SettingKeyChannelMonitorMode = "channel_monitor_mode"
|
||||
|
||||
// ChannelMonitorModeV1/V2 are the only accepted mode values.
|
||||
|
||||
@@ -185,7 +185,7 @@ func (s *SettingService) InitializeDefaultSettings(ctx context.Context) error {
|
||||
|
||||
// Channel monitor defaults (enabled, 60s)
|
||||
SettingKeyChannelMonitorEnabled: "true",
|
||||
SettingKeyChannelMonitorMode: ChannelMonitorModeV2,
|
||||
SettingKeyChannelMonitorMode: ChannelMonitorModeV1,
|
||||
SettingKeyChannelMonitorDefaultIntervalSeconds: "60",
|
||||
SettingKeyChannelMonitorHideThroughput: "true",
|
||||
|
||||
@@ -787,9 +787,9 @@ func (s *SettingService) parseSettings(settings map[string]string) *SystemSettin
|
||||
result.ChannelMonitorDefaultIntervalSeconds = parseChannelMonitorInterval(
|
||||
settings[SettingKeyChannelMonitorDefaultIntervalSeconds],
|
||||
)
|
||||
// 默认隐藏吞吐(迁移 206 的隐私默认):未配置时必须与 setting_public.go 的
|
||||
// 公开读取路径给出同一个值,否则管理端看到“未隐藏”而用户端实际已隐藏。
|
||||
result.ChannelMonitorHideThroughput = !isFalseSettingValue(settings[SettingKeyChannelMonitorHideThroughput])
|
||||
// 默认隐藏吞吐(迁移 206 的隐私默认):未配置时必须与 setting_public.go 的
|
||||
// 公开读取路径给出同一个值,否则管理端看到“未隐藏”而用户端实际已隐藏。
|
||||
result.ChannelMonitorHideThroughput = !isFalseSettingValue(settings[SettingKeyChannelMonitorHideThroughput])
|
||||
|
||||
// Available channels feature (default: disabled; strict true)
|
||||
result.AvailableChannelsEnabled = settings[SettingKeyAvailableChannelsEnabled] == "true"
|
||||
|
||||
@@ -373,15 +373,15 @@ const (
|
||||
channelMonitorIntervalMin = 15
|
||||
channelMonitorIntervalMax = 3600
|
||||
channelMonitorIntervalFallback = 60
|
||||
defaultChannelMonitorMode = ChannelMonitorModeV2
|
||||
defaultChannelMonitorMode = ChannelMonitorModeV1
|
||||
)
|
||||
|
||||
// normalizeChannelMonitorMode accepts only v1/v2; empty/invalid → v2.
|
||||
// normalizeChannelMonitorMode accepts only v1/v2; empty/invalid → v1 (safe default).
|
||||
func normalizeChannelMonitorMode(raw string) string {
|
||||
switch strings.ToLower(strings.TrimSpace(raw)) {
|
||||
case ChannelMonitorModeV1:
|
||||
case ChannelMonitorModeV1, "":
|
||||
return ChannelMonitorModeV1
|
||||
case ChannelMonitorModeV2, "":
|
||||
case ChannelMonitorModeV2:
|
||||
return ChannelMonitorModeV2
|
||||
default:
|
||||
return defaultChannelMonitorMode
|
||||
@@ -433,7 +433,7 @@ func (r ChannelMonitorRuntime) PassiveAggregationAllowed() bool {
|
||||
}
|
||||
|
||||
// GetChannelMonitorRuntime reads the channel monitor feature flags directly from
|
||||
// the settings store. Fail-open: on error returns Enabled=true, Mode=v2, default interval.
|
||||
// the settings store. Fail-open: on error returns Enabled=true, Mode=v1, default interval.
|
||||
func (s *SettingService) GetChannelMonitorRuntime(ctx context.Context) ChannelMonitorRuntime {
|
||||
if s == nil || s.settingRepo == nil {
|
||||
return ChannelMonitorRuntime{
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
-- Channel monitor exclusive mode: v1 (active probes) or v2 (passive aggregation).
|
||||
-- Default v2 keeps existing V2-first deployments unchanged.
|
||||
-- Default v1 preserves existing active-probe behavior; operators opt in to v2.
|
||||
INSERT INTO settings (key, value)
|
||||
VALUES ('channel_monitor_mode', 'v2')
|
||||
VALUES ('channel_monitor_mode', 'v1')
|
||||
ON CONFLICT (key) DO NOTHING;
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
# Channel Monitor V2 Safe Defaults & Gentle Backfill
|
||||
|
||||
**Date:** 2026-08-08
|
||||
**Status:** Approved for implementation
|
||||
**Branch:** `fix/channel-monitor-v2-ops-ui-blockers` (onto channel-monitor-v2)
|
||||
|
||||
## Problem
|
||||
|
||||
1. **H2** — Migration 195 and code defaults set `channel_monitor_mode=v2`, so every upgrade without an existing key auto-opts into passive V2, stops V1 probes, and may leave users on an empty V2 view.
|
||||
2. **M1** — First-enable bootstrap forces 5s ticks and ~24h `RecomputeRange` chunks until `now-30d`, hammering the primary DB.
|
||||
3. Error dedup SQL matches `request_id` without a `created_at` bound, scanning `ops_error_logs` history via `candidate_ids`.
|
||||
|
||||
## Decisions
|
||||
|
||||
| Topic | Decision |
|
||||
|-------|----------|
|
||||
| Default mode | **v1** (keep active probes). V2 is explicit opt-in. |
|
||||
| Existing `channel_monitor_mode=v2` rows | **Unchanged** (`ON CONFLICT DO NOTHING` / no force rewrite). |
|
||||
| Switch to v2 | V1 runner `fire()` and `RunCheck` require `ActiveProbesAllowed()` → probes stop. |
|
||||
| Backfill | Unified hard gates + adaptive soft gates (same rules, different observed load). |
|
||||
| Optional profile knob | **Not in this change** (YAGNI). |
|
||||
|
||||
## Mode defaults
|
||||
|
||||
- Migration `195_channel_monitor_mode.sql`: insert `'v1'` when key missing.
|
||||
- If 195 already applied with old checksum, add migration checksum compatibility (do not force rewrite applied DBs to v1).
|
||||
- Code: `defaultChannelMonitorMode = v1`; empty/invalid normalize → v1.
|
||||
- Frontend Settings form default and public feature flag fallback → v1 when missing/invalid.
|
||||
- Nil settings on V1 `RunCheck` path remains **fail-closed** (no probes) for test safety — independent of product default.
|
||||
|
||||
## Gentle backfill (low-resource phases)
|
||||
|
||||
### Hard gates (all servers)
|
||||
|
||||
- Tick interval = `refresh_interval` (60 or 300s). **No 5s bootstrap override.**
|
||||
- Each tick: (1) recent overlap (~10m), then (2) **at most one** historical chunk while product bootstrap incomplete / retention walk ongoing.
|
||||
- Single leader lock; single short transaction budget (~55s).
|
||||
- Chunk ceiling by depth (product phases 90m → 1d → 7d → 30d → silent 90d).
|
||||
|
||||
### Soft gates (adaptive)
|
||||
|
||||
- Initial historical chunk: **1h** (first seed still **2h** for 90m UI).
|
||||
- Min chunk: **15m**; max by depth: ≤1d → **2h**, ≤7d → **4h**, older → **6h**.
|
||||
- Success + fast: grow slowly (×1.5) up to depth max.
|
||||
- Failure/timeout: halve chunk; exponential backoff on wait (cap 10m).
|
||||
- Progressive UI: show whatever is covered; bootstrap banner vs 30d product window.
|
||||
|
||||
### Error dedup
|
||||
|
||||
- Bound the `request_id IN candidate_ids` branch with
|
||||
`created_at >= $1 - INTERVAL '90 minutes' AND created_at < $2`.
|
||||
|
||||
## Out of scope
|
||||
|
||||
- Force existing v2 → v1.
|
||||
- Separate backfill workers / read replicas.
|
||||
- Admin `backfill_profile` setting (may add later).
|
||||
@@ -21,11 +21,11 @@ export default {
|
||||
enabled: 'Enable Channel Monitor',
|
||||
enabledHint: 'Disabling stops both the V1 scheduler and V2 aggregation; existing config and history are kept.',
|
||||
mode: 'Monitor mode',
|
||||
modeHint: 'Only one implementation can be active: V2 never sends upstream probes; V1 probes configured monitors on a schedule.',
|
||||
modeHint: 'Default is V1 (active probes). Switch to V2 only when you want passive aggregation; only one implementation can be active at a time.',
|
||||
modeV2: 'V2 passive monitoring',
|
||||
modeV1: 'V1 active probes',
|
||||
modeV2Hint: 'Aggregates health metrics from real gateway traffic without upstream probe traffic.',
|
||||
modeV1Hint: 'Runs scheduled upstream health checks for configured channel monitors (probe traffic).',
|
||||
modeV2Hint: 'Opt-in: aggregates health metrics from real gateway traffic without upstream probe traffic. V1 probes stop while V2 is active.',
|
||||
modeV1Hint: 'Default: runs scheduled upstream health checks for configured channel monitors (probe traffic).',
|
||||
defaultInterval: 'Default check interval (seconds)',
|
||||
defaultIntervalHint: 'V1 only: default interval for new monitors (overridable per monitor). Range 15 – 3600 seconds.',
|
||||
hideThroughput: 'Hide throughput rates from users (RPM / TPM)',
|
||||
|
||||
@@ -21,11 +21,11 @@ export default {
|
||||
enabled: '启用渠道监控',
|
||||
enabledHint: '关闭后 V1 调度器与 V2 聚合均停止;已有配置与历史保留。',
|
||||
mode: '监控模式',
|
||||
modeHint: '同一时间只能启用一种实现:V2 不产生上游探测流量;V1 会按配置的监控项主动探活。',
|
||||
modeHint: '默认 V1(主动探测)。仅在需要被动聚合时切换到 V2;同一时间只能启用一种实现。',
|
||||
modeV2: 'V2 被动监控',
|
||||
modeV1: 'V1 主动探测',
|
||||
modeV2Hint: '基于真实网关流量聚合健康指标,不向上游发送探活请求。',
|
||||
modeV1Hint: '按配置的渠道监控项定时发起上游健康检查(产生探测流量)。',
|
||||
modeV2Hint: '需主动选择:基于真实网关流量聚合健康指标,不向上游发送探活请求;启用期间 V1 探测停止。',
|
||||
modeV1Hint: '默认模式:按配置的渠道监控项定时发起上游健康检查(产生探测流量)。',
|
||||
defaultInterval: '默认检测间隔(秒)',
|
||||
defaultIntervalHint: '仅 V1 模式使用:新建渠道监控时表单的默认值,可被单个渠道覆盖。范围 15 – 3600 秒。',
|
||||
hideThroughput: '对用户隐藏吞吐速率(RPM / TPM)',
|
||||
|
||||
@@ -160,11 +160,11 @@ export function isChannelMonitorRouteEnabled(): boolean {
|
||||
|
||||
export type ChannelMonitorMode = 'v1' | 'v2'
|
||||
|
||||
/** Exclusive channel-monitor implementation. Invalid/missing → v2. */
|
||||
/** Exclusive channel-monitor implementation. Invalid/missing → v1 (opt-in to v2). */
|
||||
export function getChannelMonitorMode(): ChannelMonitorMode {
|
||||
const appStore = useAppStore()
|
||||
const mode = appStore.cachedPublicSettings?.channel_monitor_mode
|
||||
return mode === 'v1' ? 'v1' : 'v2'
|
||||
return mode === 'v2' ? 'v2' : 'v1'
|
||||
}
|
||||
|
||||
export function isChannelMonitorV1Mode(): boolean {
|
||||
|
||||
@@ -9516,7 +9516,7 @@ const form = reactive<SettingsForm>({
|
||||
account_quota_notify_emails: [] as NotifyEmailEntry[],
|
||||
// Channel Monitor feature switch
|
||||
channel_monitor_enabled: true,
|
||||
channel_monitor_mode: 'v2' as 'v1' | 'v2',
|
||||
channel_monitor_mode: 'v1' as 'v1' | 'v2',
|
||||
channel_monitor_default_interval_seconds: 60,
|
||||
channel_monitor_hide_throughput: false,
|
||||
// Available Channels feature switch
|
||||
@@ -10516,7 +10516,7 @@ async function loadSettings() {
|
||||
form.login_agreement_mode =
|
||||
settings.login_agreement_mode === "checkbox" ? "checkbox" : "modal";
|
||||
form.channel_monitor_mode =
|
||||
settings.channel_monitor_mode === "v1" ? "v1" : "v2";
|
||||
settings.channel_monitor_mode === "v2" ? "v2" : "v1";
|
||||
form.channel_monitor_hide_throughput = Boolean(
|
||||
settings.channel_monitor_hide_throughput
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user