Skip to content

remotewrite/sender: Migrate Remote Write 2.0 sender tests to new framework #261

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

Closed
bwplotka wants to merge 1 commit into from
Closed

remotewrite/sender: Migrate Remote Write 2.0 sender tests to new framework #261

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

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions remotewrite/receiver/exemplar_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,8 +114,8 @@ func TestExemplarWithCreatedTimestamp(t *testing.T) {
createdTime := now.Add(-2 * time.Hour)

sample := SampleWithLabels{
Labels: map[string]string{"__name__": "http_requests_total", "job": "api"},
Value: 250.0,
Labels: map[string]string{"__name__": "http_requests_total", "job": "api"},
Value: 250.0,
StartTimestamp: &createdTime,
}
exemplar := ExemplarWithLabels{
Expand Down
6 changes: 3 additions & 3 deletions remotewrite/receiver/histograms_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,9 +118,9 @@ func TestHistogramWithCreatedTimestamp(t *testing.T) {
createdTime := now.Add(-30 * time.Minute)

hist := HistogramWithLabels{
Labels: map[string]string{"__name__": "request_duration_seconds", "job": "api"},
Histogram: histogram(1.5, true, true, true, false, false),
StartTimestamp: &createdTime,
Labels: map[string]string{"__name__": "request_duration_seconds", "job": "api"},
Histogram: histogram(1.5, true, true, true, false, false),
StartTimestamp: &createdTime,
}

runComplianceTest(t, "", "Histogram with created timestamp",
Expand Down
6 changes: 3 additions & 3 deletions remotewrite/receiver/metadata_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,9 +159,9 @@ func TestCounterMetadataWithCreatedTimestamp(t *testing.T) {
createdTime := now.Add(-1 * time.Hour)

sample := SampleWithLabels{
Labels: map[string]string{"__name__": "http_requests_total", "job": "api"},
Value: 150.0,
StartTimestamp: &createdTime,
Labels: map[string]string{"__name__": "http_requests_total", "job": "api"},
Value: 150.0,
StartTimestamp: &createdTime,
}
metadata := MetadataWithLabels{
Labels: basicMetric("http_requests_total"),
Expand Down
6 changes: 3 additions & 3 deletions remotewrite/receiver/metric_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,9 +95,9 @@ func TestCounterWithCreatedTimestamp(t *testing.T) {
createdTime := now.Add(-1 * time.Hour)

sample := SampleWithLabels{
Labels: map[string]string{"__name__": "http_requests_total", "job": "api"},
Value: 100.0,
StartTimestamp: &createdTime,
Labels: map[string]string{"__name__": "http_requests_total", "job": "api"},
Value: 100.0,
StartTimestamp: &createdTime,
}

runComplianceTest(t, "", "Counter with created timestamp",
Expand Down
219 changes: 23 additions & 196 deletions remotewrite/sender/backoff_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,219 +14,46 @@
package sender

import (
"fmt"
"net/http"
"sync"
"testing"
"time"
)

// TimestampTrackingReceiver wraps MockReceiver to track request timestamps.
type TimestampTrackingReceiver struct {
*MockReceiver
mu sync.Mutex
timestamps []time.Time
}

// NewTimestampTrackingReceiver creates a receiver that tracks request timestamps.
func NewTimestampTrackingReceiver(baseReceiver *MockReceiver) *TimestampTrackingReceiver {
return &TimestampTrackingReceiver{
MockReceiver: baseReceiver,
timestamps: make([]time.Time, 0),
}
}

// RecordTimestamp records the current time for a request.
func (ttr *TimestampTrackingReceiver) RecordTimestamp() {
ttr.mu.Lock()
defer ttr.mu.Unlock()
ttr.timestamps = append(ttr.timestamps, time.Now())
}

// GetTimestamps returns all recorded timestamps.
func (ttr *TimestampTrackingReceiver) GetTimestamps() []time.Time {
ttr.mu.Lock()
defer ttr.mu.Unlock()
return append([]time.Time{}, ttr.timestamps...)
}

// TestBackoffBehavior validates backoff and exponential backoff implementation.
func TestBackoffBehavior_Old(t *testing.T) {
t.Skip("TODO: Revise and move to a new framework")
"github.com/prometheus/client_golang/exp/api/remote"
"github.com/stretchr/testify/require"
)

tests := []struct {
name string
description string
rfcLevel string
scrapeData string
setup func(*MockReceiver)
validator func(*testing.T, *TimestampTrackingReceiver)
}{
func backoffTests() []Test {
return []Test{
{
name: "exponential_backoff_on_retries",
description: "Sender SHOULD use exponential backoff when retrying",
rfcLevel: "SHOULD",
scrapeData: "test_metric 42\n",
setup: func(mr *MockReceiver) {
mr.SetResponse(MockReceiverResponse{
Name: "backoff_required",
Description: "Sender MUST use a backoff algorithm to prevent overwhelming the server",
RFCLevel: MustLevel,
ScrapeData: "test_metric 42\n",
Version: remote.WriteV2MessageType,
TestResponses: []ReceiverResponse{
{
StatusCode: http.StatusServiceUnavailable,
Body: "Service unavailable",
})
},
validator: func(t *testing.T, ttr *TimestampTrackingReceiver) {
timestamps := ttr.GetTimestamps()
if len(timestamps) < 3 {
should(t, len(timestamps) >= 3, "Need at least 3 requests to validate backoff pattern")
t.Logf("Only %d requests observed, cannot validate backoff", len(timestamps))
return
}

// Calculate intervals between requests.
intervals := make([]time.Duration, 0)
for i := 1; i < len(timestamps); i++ {
interval := timestamps[i].Sub(timestamps[i-1])
intervals = append(intervals, interval)
t.Logf("Interval %d: %v", i, interval)
}

// Check that intervals are increasing (exponential backoff).
if len(intervals) >= 2 {
for i := 1; i < len(intervals); i++ {
// Allow some tolerance for timing jitter.
// Second interval should be >= first interval (or close).
ratio := float64(intervals[i]) / float64(intervals[i-1])
should(t, ratio >= 0.8, fmt.Sprintf("Backoff intervals should increase or stay similar (exponential), got ratio %.2f", ratio))
}
}
},
},
{
name: "backoff_required",
description: "Sender MUST use a backoff algorithm to prevent overwhelming the server",
rfcLevel: "MUST",
scrapeData: "test_metric 42\n",
setup: func(mr *MockReceiver) {
mr.SetResponse(MockReceiverResponse{
},
{
StatusCode: http.StatusServiceUnavailable,
Body: "Service unavailable",
})
},
{
StatusCode: http.StatusNoContent,
},
},
validator: func(t *testing.T, ttr *TimestampTrackingReceiver) {
timestamps := ttr.GetTimestamps()
must(t).True(len(timestamps) >= 2, "Sender must retry on 5xx errors")

if len(timestamps) < 2 {
t.Fatalf("Expected at least 2 requests (initial + retry), got %d", len(timestamps))
}
Validate: func(t *testing.T, res ReceiverResult) {
require.GreaterOrEqual(t, len(res.Requests), 3, "Expected at least 3 requests (initial + 2 retries)")

// MUST: Verify that backoff exists (delay between all requests).
minBackoffDelay := 1 * time.Millisecond
for i := 1; i < len(timestamps); i++ {
interval := timestamps[i].Sub(timestamps[i-1])
must(t).True(interval >= minBackoffDelay,
for i := 1; i < len(res.Requests); i++ {
interval := res.Requests[i].Received.Sub(res.Requests[i-1].Received)
require.GreaterOrEqual(t, interval, minBackoffDelay,
"Sender MUST implement backoff between all retry attempts, interval %d was: %v", i, interval)
t.Logf("Interval %d: %v", i, interval)
}

// Additional validation: Check that delays increase over time (if 3+ requests).
if len(timestamps) >= 3 {
firstDelay := timestamps[1].Sub(timestamps[0])
secondDelay := timestamps[2].Sub(timestamps[1])
t.Logf("First delay: %v, Second delay: %v", firstDelay, secondDelay)

// This validates increasing backoff pattern
if secondDelay < firstDelay {
t.Logf("Note: Second delay (%v) is shorter than first delay (%v), backoff may not be increasing", secondDelay, firstDelay)
}
}
},
},
{
name: "backoff_max_delay",
description: "Backoff SHOULD have a reasonable maximum delay",
rfcLevel: "SHOULD",
scrapeData: "test_metric 42\n",
setup: func(mr *MockReceiver) {
mr.SetResponse(MockReceiverResponse{
StatusCode: http.StatusBadGateway,
Body: "Bad gateway",
})
},
validator: func(t *testing.T, ttr *TimestampTrackingReceiver) {
timestamps := ttr.GetTimestamps()
if len(timestamps) < 2 {
t.Logf("Only %d requests, cannot validate max delay", len(timestamps))
return
}

// Check that no interval exceeds a reasonable maximum (e.g., 60 seconds).
maxReasonableDelay := 60 * time.Second

for i := 1; i < len(timestamps); i++ {
interval := timestamps[i].Sub(timestamps[i-1])
should(t, interval <= maxReasonableDelay, fmt.Sprintf("Backoff interval too large: %v > %v", interval, maxReasonableDelay))
}

t.Logf("Observed %d retry attempts over %v",
len(timestamps), timestamps[len(timestamps)-1].Sub(timestamps[0]))
},
},
{
name: "backoff_with_jitter",
description: "Sender MAY add jitter to backoff delays",
rfcLevel: "RECOMMENDED",
scrapeData: "test_metric 42\n",
setup: func(mr *MockReceiver) {
mr.SetResponse(MockReceiverResponse{
StatusCode: http.StatusServiceUnavailable,
Body: "Service unavailable",
})
},
validator: func(t *testing.T, ttr *TimestampTrackingReceiver) {
timestamps := ttr.GetTimestamps()
if len(timestamps) < 3 {
may(t, len(timestamps) >= 3, "Jitter validation requires multiple retries")
return
}

intervals := make([]time.Duration, 0)
for i := 1; i < len(timestamps); i++ {
intervals = append(intervals, timestamps[i].Sub(timestamps[i-1]))
}

// With jitter, intervals shouldn't be exactly doubling.
may(t, len(intervals) > 0, "Intervals should exist for jitter analysis")
t.Logf("Intervals: %v (jitter may cause variance)", intervals)
},
},
}

for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
t.Attr("rfcLevel", tt.rfcLevel)
t.Attr("description", tt.description)

forEachSender(t, func(t *testing.T, targetName string, target Sender) {
receiver := NewMockReceiver()
defer receiver.Close()

tracker := NewTimestampTrackingReceiver(receiver)

tt.setup(receiver)

originalHandler := receiver.server.Config.Handler
receiver.server.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
tracker.RecordTimestamp()
originalHandler.ServeHTTP(w, r)
})

scrapeTarget := NewMockScrapeTarget(tt.scrapeData)
defer scrapeTarget.Close()

runAutoTargetWithCustomReceiver(t, targetName, target, receiver.URL(), scrapeTarget, 20*time.Second)
tt.validator(t, tracker)
})
})
}
}
Loading
Loading