Skip to content

fix(querytee): Fix TestProxyEndpoint_QuerySplitting clock-drift bug #21779

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jnewbigin merged 1 commit into from
May 8, 2026
Merged

fix(querytee): Fix TestProxyEndpoint_QuerySplitting clock-drift bug #21779

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 21 additions & 38 deletions pkg/querytee/proxy_endpoint_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -770,10 +770,13 @@ func TestProxyEndpoint_QuerySplitting(t *testing.T) {
receivedQueries := []string{}

step := "60s"
stepMs := int64(60000)

// Calculate the split boundary: step-aligned threshold + step
splitBoundary := stepAlignUp(threshold, stepMs).Add(time.Duration(stepMs) * time.Millisecond)
// The split boundary sits midway between threshold and the recent portion
// of the spans-threshold subtests (threshold + 1h). A generous margin keeps
// the boundary stable even when v2End drifts: the engine router computes
// v2End from time.Now() at request time, while threshold here is captured
// at test setup, and logs queries don't step-align v2End.
splitBoundary := threshold.Add(30 * time.Minute)

server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
Expand All @@ -784,10 +787,10 @@ func TestProxyEndpoint_QuerySplitting(t *testing.T) {
require.NoError(t, err)
endTime := rangeQuery.End

// If end time is before or at split boundary, return old response
// Otherwise return recent response
// If end time is before the split boundary, return the old response.
// Otherwise return the recent response.
w.WriteHeader(200)
if endTime.Before(splitBoundary) || endTime.Equal(splitBoundary) {
if endTime.Before(splitBoundary) {
_, _ = w.Write([]byte(oldResponseBody))
} else {
_, _ = w.Write([]byte(recentResponseBody))
Expand Down Expand Up @@ -884,29 +887,21 @@ func TestProxyEndpoint_QuerySplitting(t *testing.T) {

assert.Equal(t, 3, len(receivedQueries), "expected 3 queries, 1 for the recent portion, and 1 to both v1 and v2 for comparison, got %d", len(receivedQueries))

// Verify that old queries go to both backends
// Step is 60s = 60000ms from the test query
stepMs := int64(60000)

// Align threshold UP (as it's treated as an end boundary in alignStartEnd)
// This is v2End in the engine_router
alignedThreshold := stepAlignUp(threshold, stepMs)

// Based on observed behavior, old queries end at alignedThreshold + gap
oldQueryBoundary := alignedThreshold.Add(time.Duration(stepMs) * time.Millisecond)

// Classify queries by whether their end time matches the request's end:
// the post-v2 (recent) split runs to the request end, while the v2 (old)
// split ends at v2End, which is computed dynamically from time.Now() and
// can drift relative to `threshold` captured at test setup.
oldQueries := 0
recentQueries := 0
for _, q := range receivedQueries {
if strings.Contains(q, "end=") {
endStr := extractQueryParam(q, "end")
endTime, _ := parseTimestamp(endStr)

// Queries ending at or before oldQueryBoundary are "old"
if endTime.Before(oldQueryBoundary) || endTime.Equal(oldQueryBoundary) {
oldQueries++
} else {
if endTime.Equal(end) {
recentQueries++
} else {
oldQueries++
}
}
}
Expand Down Expand Up @@ -962,22 +957,19 @@ func TestProxyEndpoint_QuerySplitting(t *testing.T) {
// - 3 queries total: 1 for recent portion (v1 only), 2 for old portion (v1 and v2 for comparison)
assert.Equal(t, 3, len(receivedQueries), "expected 3 queries for v2 metric query, 1 for recent portion and 2 for old portion comparison, got %d", len(receivedQueries))

// Verify that old queries go to both backends
stepMs := int64(60000)
alignedThreshold := stepAlignUp(threshold, stepMs)
oldQueryBoundary := alignedThreshold.Add(time.Duration(stepMs) * time.Millisecond)

// Classify queries by whether their end time matches the request's end
// (the recent portion) or not (the old portion sent to both backends).
oldQueries := 0
recentQueries := 0
for _, q := range receivedQueries {
if strings.Contains(q, "end=") {
endStr := extractQueryParam(q, "end")
endTime, _ := parseTimestamp(endStr)

if endTime.Before(oldQueryBoundary) || endTime.Equal(oldQueryBoundary) {
oldQueries++
} else {
if endTime.Equal(end) {
recentQueries++
} else {
oldQueries++
}
}
}
Expand Down Expand Up @@ -1018,15 +1010,6 @@ func parseTimestamp(value string) (time.Time, error) {
return time.Unix(0, nanos), nil
}

// stepAlignUp aligns a timestamp up to the nearest step boundary
func stepAlignUp(t time.Time, stepMs int64) time.Time {
timestampMs := t.UnixMilli()
if mod := timestampMs % stepMs; mod != 0 {
timestampMs += stepMs - mod
}
return time.Unix(0, timestampMs*1e6)
}

// Helper function to extract query parameters from a query string
func extractQueryParam(queryString, param string) string {
parts := strings.SplitSeq(queryString, "&")
Expand Down
Loading