seewe/sub2api
1
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-14 12:02:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: anchor channel monitor buckets to UTC

Browse Source
This commit is contained in:
reallyfeishi
2026-09-08 23:02:06 +08:00
parent 270eac6973
commit f180fbb884
3 changed files with 36 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/repository/channel_monitor_v2_aggregation.go
+16 -6
View File
@@ -395,11 +395,21 @@ func sameFixedRollupBucket(start, end time.Time, seconds int) bool {
return start.Truncate(interval).Equal(end.Add(-time.Nanosecond).Truncate(interval))
}
// PostgreSQL interprets a TIMESTAMPTZ literal without an explicit offset in
// the current session timezone. Keep date_bin's origin fixed in UTC so bucket
// boundaries do not shift when the database session runs in Asia/Shanghai (or
// any other non-UTC timezone).
const channelMonitorV2DateBinOrigin = "TIMESTAMPTZ '1970-01-01 00:00:00+00'"
func channelMonitorV2DateBinExpr(column string) string {
return "date_bin($1::interval," + column + "," + channelMonitorV2DateBinOrigin + ")"
}
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
date_bin($1::interval, $3::timestamptz, ` + channelMonitorV2DateBinOrigin + `) AS start_at,
date_bin($1::interval, $4::timestamptz - INTERVAL '1 microsecond', ` + channelMonitorV2DateBinOrigin + `) + $1::interval AS end_at
)`
const channelMonitorV2FixedRollupDeleteSQL = channelMonitorV2FixedRollupBoundsSQL + `
@@ -417,7 +427,7 @@ INSERT INTO channel_monitor_v2_metrics_rollup (
duration_count, computed_at
)
` + channelMonitorV2FixedRollupBoundsSQL + `
SELECT date_bin($1::interval, m.bucket_start, TIMESTAMPTZ '1970-01-01'), $2::integer,
SELECT date_bin($1::interval, m.bucket_start, ` + channelMonitorV2DateBinOrigin + `), $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),
@@ -433,7 +443,7 @@ INSERT INTO channel_monitor_v2_user_metrics_rollup (
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,
SELECT date_bin($1::interval, m.bucket_start, ` + channelMonitorV2DateBinOrigin + `), $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()
@@ -446,7 +456,7 @@ 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,
SELECT date_bin($1::interval, h.bucket_start, ` + channelMonitorV2DateBinOrigin + `), $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
@@ -457,7 +467,7 @@ 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,
SELECT date_bin($1::interval, e.bucket_start, ` + channelMonitorV2DateBinOrigin + `), $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
backend/internal/repository/channel_monitor_v2_repo.go
+4 -4
View File
@@ -949,7 +949,7 @@ func (r *channelMonitorV2Repository) loadFacts(ctx context.Context, filter servi
} else {
args = append([]any{fmt.Sprintf("%d seconds", int(filter.Bucket.Seconds()))}, args...)
where = shiftSQLPlaceholders(where, 1)
bucketExpr = "date_bin($1::interval,m.bucket_start,TIMESTAMPTZ '1970-01-01')"
bucketExpr = channelMonitorV2DateBinExpr("m.bucket_start")
group = bucketExpr + "," + group
}
}
@@ -991,7 +991,7 @@ func (r *channelMonitorV2Repository) loadHistograms(ctx context.Context, filter
args = []any{fmt.Sprintf("%d seconds", int(filter.Bucket.Seconds()))}
args = append(args, oldArgs...)
where = shiftSQLPlaceholders(where, 1)
bucketExpr = "date_bin($1::interval,h.bucket_start,TIMESTAMPTZ '1970-01-01')"
bucketExpr = channelMonitorV2DateBinExpr("h.bucket_start")
group = bucketExpr + "," + group
}
}
@@ -1444,7 +1444,7 @@ func (r *channelMonitorV2Repository) loadIgnoredErrorCounts(
if filter.Bucket > 0 && bucketSeconds == 0 {
args = append([]any{fmt.Sprintf("%d seconds", int(filter.Bucket.Seconds()))}, args...)
where = shiftSQLPlaceholders(where, 1)
bucketExpr = "date_bin($1::interval,e.bucket_start,TIMESTAMPTZ '1970-01-01')"
bucketExpr = channelMonitorV2DateBinExpr("e.bucket_start")
groupBy = bucketExpr + ", e.platform, e.model"
}
args = append(args, pq.Array(cfg.IgnoredErrorCategories), service.ChannelMonitorV2TaxonomyVersion)
@@ -1540,7 +1540,7 @@ func (r *channelMonitorV2Repository) loadIgnoredErrorCountsByMatrixKey(
if filter.Bucket > 0 && bucketSeconds == 0 {
args = append([]any{fmt.Sprintf("%d seconds", int(filter.Bucket.Seconds()))}, args...)
where = shiftSQLPlaceholders(where, 1)
bucketExpr = "date_bin($1::interval,e.bucket_start,TIMESTAMPTZ '1970-01-01')"
bucketExpr = channelMonitorV2DateBinExpr("e.bucket_start")
groupSQL = bucketExpr + ", e.platform, e.group_id, e.model"
}
args = append(args, pq.Array(cfg.IgnoredErrorCategories), service.ChannelMonitorV2TaxonomyVersion)
backend/internal/repository/channel_monitor_v2_repo_test.go
+16
View File
@@ -12,6 +12,22 @@ import (
"github.com/stretchr/testify/require"
)
func TestChannelMonitorV2DateBinOriginIsUTC(t *testing.T) {
require.Equal(t, "TIMESTAMPTZ '1970-01-01 00:00:00+00'", channelMonitorV2DateBinOrigin)
require.Equal(t, "date_bin($1::interval,m.bucket_start,TIMESTAMPTZ '1970-01-01 00:00:00+00')", channelMonitorV2DateBinExpr("m.bucket_start"))
for _, query := range []string{
channelMonitorV2FixedRollupBoundsSQL,
channelMonitorV2MetricsRollupSQL,
channelMonitorV2UserMetricsRollupSQL,
channelMonitorV2HistogramRollupSQL,
channelMonitorV2ErrorRollupSQL,
} {
require.Contains(t, query, channelMonitorV2DateBinOrigin)
require.NotContains(t, query, "TIMESTAMPTZ '1970-01-01'")
}
}
func TestChannelMonitorV2DisplayModelIsPlatformScoped(t *testing.T) {
cfg := service.ChannelMonitorV2Config{Platforms: []service.ChannelMonitorV2PlatformConfig{
{Platform: "openai", Enabled: true, Models: []string{"shared", "gpt-5"}},