github/sub2api
4
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-28 08:08:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(monitor-v2): resolve composite group error facts to concrete account platform

Browse Source 
Composite groups are a routing layer. The usage-side aggregation already
maps their traffic to the concrete account platform via
usageLogEffectivePlatformExpr, but the error aggregation read
ops_error_logs.platform verbatim. Pre-routing failures on composite groups
fall back to the group platform 'composite', which is never an enabled
monitor v2 config platform, so those error facts were filtered out of
every channel monitor v2 query and composite groups looked permanently
healthy.

Join groups/accounts in the error dedup CTE and resolve the effective
platform the same way as the usage side: account platform first, then a
real recorded platform, else 'unknown'. The window delete+rewrite
recompute rewrites historical facts with the corrected key.
This commit is contained in:
jjaw
2026-08-23 02:33:09 +08:00
parent d45135d87d
commit 49752060fc
2 changed files with 34 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/repository/channel_monitor_v2_aggregation.go
+22 -12
View File
@@ -247,18 +247,28 @@ WITH dedup AS (
FROM ops_error_logs
WHERE created_at >= $1 AND created_at < $2 AND NULLIF(request_id, '') IS NOT NULL
)
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
SELECT DISTINCT ON (COALESCE(NULLIF(current_error.request_id, ''), 'error:' || current_error.id::text))
date_trunc('minute', current_error.created_at) AS bucket_start,
-- Composite groups are a routing layer: resolve the concrete account
-- platform (mirrors usageLogEffectivePlatformExpr on the usage side) so
-- error facts share the usage facts' platform key. Without this, composite
-- group errors aggregate under platform 'composite', which is never an
-- enabled config platform, and are filtered out of every monitor v2 query.
lower(CASE
WHEN g.platform = 'composite' THEN COALESCE(NULLIF(TRIM(a.platform)), NULLIF(NULLIF(lower(TRIM(current_error.platform)), ''), 'composite'), 'unknown')
ELSE COALESCE(NULLIF(TRIM(current_error.platform), ''), 'unknown')
END) AS platform,
COALESCE(current_error.group_id, 0) AS group_id,
COALESCE(NULLIF(TRIM(current_error.requested_model), ''), NULLIF(TRIM(current_error.model), ''), 'unknown') AS model,
current_error.user_id, current_error.error_type, current_error.error_owner, COALESCE(current_error.status_code, 0) AS status_code,
COALESCE(current_error.upstream_status_code, 0) AS upstream_status_code,
lower(CONCAT_WS(' ', current_error.error_type, current_error.error_source, current_error.error_message, current_error.upstream_error_message, current_error.upstream_error_detail, current_error.error_body)) AS text,
(CASE WHEN jsonb_typeof(current_error.upstream_errors) = 'array' THEN jsonb_array_length(current_error.upstream_errors) > 0 ELSE FALSE END
OR current_error.error_owner = 'provider' OR current_error.upstream_status_code IS NOT NULL) AS upstream_affected,
CASE WHEN jsonb_typeof(current_error.upstream_errors) = 'array' THEN jsonb_array_length(current_error.upstream_errors) ELSE 0 END AS upstream_attempts
FROM ops_error_logs current_error
LEFT JOIN groups g ON g.id = current_error.group_id
LEFT JOIN accounts a ON a.id = current_error.account_id
WHERE (
(NULLIF(current_error.request_id, '') IS NULL AND current_error.created_at >= $1 AND current_error.created_at < $2)
OR (
@@ -269,7 +279,7 @@ WITH dedup AS (
)
AND NOT current_error.is_count_tokens
AND (COALESCE(current_error.status_code, 0) >= 400 OR current_error.error_type = 'cyber_policy')
ORDER BY COALESCE(NULLIF(request_id, ''), 'error:' || id::text), created_at DESC, id DESC
ORDER BY COALESCE(NULLIF(current_error.request_id, ''), 'error:' || current_error.id::text), current_error.created_at DESC, current_error.id DESC
), classified AS (
SELECT *, CASE
-- Keep in lockstep with service.ClassifyChannelMonitorV2Error needles.
backend/internal/repository/channel_monitor_v2_repo_test.go
+12 -1
View File
@@ -101,12 +101,23 @@ func TestChannelMonitorV2ErrorAggregationCountsFinalUserErrorsOnly(t *testing.T)
require.Contains(t, query, "candidate_ids")
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")
require.Contains(t, query, "jsonb_array_length(current_error.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 TestChannelMonitorV2ErrorAggregationResolvesCompositePlatform(t *testing.T) {
query := strings.ToLower(channelMonitorV2ErrorAggregationSQL)
// Composite groups are a routing layer: error facts must resolve the concrete
// account platform (joining groups/accounts) so they aggregate under the same
// platform key as usage facts instead of the never-enabled 'composite' platform.
require.Contains(t, query, "g.platform = 'composite'")
require.Contains(t, query, "left join groups g on g.id = current_error.group_id")
require.Contains(t, query, "left join accounts a on a.id = current_error.account_id")
require.Contains(t, query, "a.platform")
}
func TestChannelMonitorV2UsageSuccessExcludesCyberBillingRows(t *testing.T) {
for _, query := range []string{channelMonitorV2UsageMetricsSQL, channelMonitorV2UserMetricsSQL} {
require.Contains(t, query, "COALESCE(ul.request_type, 0) NOT IN (4, 6)")