ITWorx Pulse release export
bd774932d5
Public source validation / validate (push) Failing after 3m8s
339 lines
12 KiB
Go
339 lines
12 KiB
Go
package livesampler_test
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/itworx/pulse/internal/live"
|
|
"github.com/itworx/pulse/internal/livesampler"
|
|
"github.com/itworx/pulse/internal/metriccatalog"
|
|
"github.com/itworx/pulse/internal/prometheus"
|
|
"github.com/itworx/pulse/internal/queryplan"
|
|
)
|
|
|
|
func fixedNow() time.Time { return time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC) }
|
|
|
|
func liveRequest() queryplan.Request {
|
|
now := fixedNow()
|
|
return queryplan.Request{
|
|
Metric: "container.cpu.utilization",
|
|
Scope: map[string]string{"container": "media_server"},
|
|
Range: queryplan.Range{From: now.Add(-time.Minute), To: now, StepSeconds: 15},
|
|
Aggregation: "avg",
|
|
MaxSeries: 2,
|
|
MaxPoints: 60,
|
|
}
|
|
}
|
|
|
|
func epoch(at time.Time) string { return fmt.Sprintf("%.3f", float64(at.UnixMilli())/1000) }
|
|
|
|
type recorder struct {
|
|
mu sync.Mutex
|
|
queries []string
|
|
}
|
|
|
|
func (r *recorder) record(query string) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
r.queries = append(r.queries, query)
|
|
}
|
|
|
|
func (r *recorder) last() string {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
if len(r.queries) == 0 {
|
|
return ""
|
|
}
|
|
return r.queries[len(r.queries)-1]
|
|
}
|
|
|
|
func prometheusServer(t *testing.T, recorded *recorder, handler http.HandlerFunc) *httptest.Server {
|
|
t.Helper()
|
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.URL.Path != "/api/v1/query" {
|
|
t.Errorf("unexpected path: %s", r.URL.Path)
|
|
}
|
|
if recorded != nil {
|
|
recorded.record(r.URL.Query().Get("query"))
|
|
}
|
|
handler(w, r)
|
|
}))
|
|
t.Cleanup(server.Close)
|
|
return server
|
|
}
|
|
|
|
func jsonServer(t *testing.T, recorded *recorder, body string) *httptest.Server {
|
|
t.Helper()
|
|
return prometheusServer(t, recorded, func(w http.ResponseWriter, _ *http.Request) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_, _ = w.Write([]byte(body))
|
|
})
|
|
}
|
|
|
|
func testSampler(t *testing.T, server *httptest.Server, options livesampler.Options, limits prometheus.Limits) *livesampler.Sampler {
|
|
t.Helper()
|
|
if limits.Timeout == 0 {
|
|
limits.Timeout = time.Second
|
|
}
|
|
client, err := prometheus.New(server.URL, nil, limits)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
registry, err := metriccatalog.DefaultRegistry()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if options.Now == nil {
|
|
options.Now = fixedNow
|
|
}
|
|
sampler, err := livesampler.New(registry, client, options)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return sampler
|
|
}
|
|
|
|
func TestSamplerReturnsFreshSamplesFromApprovedTemplate(t *testing.T) {
|
|
recorded := &recorder{}
|
|
body := fmt.Sprintf(`{"status":"success","data":{"resultType":"vector","result":[
|
|
{"metric":{"__name__":"container_cpu","container":"media_server","instance":"server-2"},"value":[%s,"11"]},
|
|
{"metric":{"__name__":"container_cpu","container":"media_server","instance":"server-1"},"value":[%s,"42.5"]}]}}`,
|
|
epoch(fixedNow().Add(-5*time.Second)), epoch(fixedNow().Add(-10*time.Second)))
|
|
sampler := testSampler(t, jsonServer(t, recorded, body), livesampler.Options{}, prometheus.Limits{})
|
|
samples, err := sampler.Sample(context.Background(), liveRequest())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(samples) != 2 {
|
|
t.Fatalf("samples=%+v", samples)
|
|
}
|
|
if samples[0].Series >= samples[1].Series {
|
|
t.Fatalf("samples are not deterministically ordered: %+v", samples)
|
|
}
|
|
first := samples[0]
|
|
if first.Freshness != "fresh" || first.Value == nil || *first.Value != 42.5 {
|
|
t.Fatalf("first=%+v", first)
|
|
}
|
|
if first.Timestamp != fixedNow().Add(-10*time.Second) || first.Labels["instance"] != "server-1" {
|
|
t.Fatalf("first=%+v", first)
|
|
}
|
|
if _, ok := first.Labels["__name__"]; ok {
|
|
t.Fatalf("internal label leaked: %+v", first.Labels)
|
|
}
|
|
if query := recorded.last(); !strings.Contains(query, `name="media_server"`) || strings.Contains(query, `container="media_server"`) || !strings.Contains(query, "[1m]") {
|
|
t.Fatalf("query=%s", query)
|
|
}
|
|
}
|
|
|
|
func TestSamplerReportsEmptyResultAsUnknown(t *testing.T) {
|
|
sampler := testSampler(t, jsonServer(t, nil, `{"status":"success","data":{"resultType":"vector","result":[]}}`), livesampler.Options{}, prometheus.Limits{})
|
|
samples, err := sampler.Sample(context.Background(), liveRequest())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(samples) != 1 {
|
|
t.Fatalf("empty result was reported as %d samples", len(samples))
|
|
}
|
|
if samples[0].Freshness != "unknown" || samples[0].Value != nil || samples[0].Labels["reason"] != livesampler.ReasonNoData {
|
|
t.Fatalf("samples=%+v", samples)
|
|
}
|
|
if samples[0].Series != "container.cpu.utilization" || samples[0].Timestamp != fixedNow() {
|
|
t.Fatalf("samples=%+v", samples)
|
|
}
|
|
}
|
|
|
|
func TestSamplerReportsUpstreamFailureAsUnknown(t *testing.T) {
|
|
server := prometheusServer(t, nil, func(w http.ResponseWriter, _ *http.Request) {
|
|
http.Error(w, "unavailable", http.StatusServiceUnavailable)
|
|
})
|
|
sampler := testSampler(t, server, livesampler.Options{}, prometheus.Limits{})
|
|
samples, err := sampler.Sample(context.Background(), liveRequest())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(samples) != 1 || samples[0].Freshness != "unknown" || samples[0].Labels["reason"] != livesampler.ReasonSourceError {
|
|
t.Fatalf("samples=%+v", samples)
|
|
}
|
|
}
|
|
|
|
func TestSamplerReportsTimeoutAsUnknownAndBoundsItsOwnDeadline(t *testing.T) {
|
|
release := make(chan struct{})
|
|
t.Cleanup(func() { close(release) })
|
|
server := prometheusServer(t, nil, func(w http.ResponseWriter, r *http.Request) {
|
|
select {
|
|
case <-release:
|
|
case <-r.Context().Done():
|
|
case <-time.After(5 * time.Second):
|
|
}
|
|
})
|
|
sampler := testSampler(t, server, livesampler.Options{Timeout: 25 * time.Millisecond}, prometheus.Limits{Timeout: 10 * time.Second})
|
|
started := time.Now()
|
|
samples, err := sampler.Sample(context.Background(), liveRequest())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if elapsed := time.Since(started); elapsed > 2*time.Second {
|
|
t.Fatalf("sampler did not bound its own deadline: %s", elapsed)
|
|
}
|
|
if len(samples) != 1 || samples[0].Freshness != "unknown" || samples[0].Labels["reason"] != livesampler.ReasonTimeout {
|
|
t.Fatalf("samples=%+v", samples)
|
|
}
|
|
}
|
|
|
|
func TestSamplerReportsStaleScrapeAsStaleWithLastKnownValue(t *testing.T) {
|
|
body := fmt.Sprintf(`{"status":"success","data":{"resultType":"vector","result":[
|
|
{"metric":{"container":"media_server"},"value":[%s,"7"]}]}}`, epoch(fixedNow().Add(-5*time.Minute)))
|
|
sampler := testSampler(t, jsonServer(t, nil, body), livesampler.Options{}, prometheus.Limits{})
|
|
samples, err := sampler.Sample(context.Background(), liveRequest())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(samples) != 1 {
|
|
t.Fatalf("samples=%+v", samples)
|
|
}
|
|
if samples[0].Freshness != "stale" || samples[0].Labels["reason"] != livesampler.ReasonStale {
|
|
t.Fatalf("stale scrape was not marked: %+v", samples[0])
|
|
}
|
|
if samples[0].Value == nil || *samples[0].Value != 7 || samples[0].Timestamp != fixedNow().Add(-5*time.Minute) {
|
|
t.Fatalf("last known value and age were lost: %+v", samples[0])
|
|
}
|
|
}
|
|
|
|
func TestSamplerRefusesResultsAboveTheSeriesBudget(t *testing.T) {
|
|
at := epoch(fixedNow())
|
|
body := fmt.Sprintf(`{"status":"success","data":{"resultType":"vector","result":[
|
|
{"metric":{"container":"a"},"value":[%s,"1"]},
|
|
{"metric":{"container":"b"},"value":[%s,"2"]},
|
|
{"metric":{"container":"c"},"value":[%s,"3"]}]}}`, at, at, at)
|
|
sampler := testSampler(t, jsonServer(t, nil, body), livesampler.Options{}, prometheus.Limits{})
|
|
if _, err := sampler.Sample(context.Background(), liveRequest()); err == nil {
|
|
t.Fatal("expected the request series budget to be enforced")
|
|
} else {
|
|
var sampleError livesampler.Error
|
|
if !errors.As(err, &sampleError) || sampleError.Code != "LIVE_SAMPLE_SERIES_LIMIT" {
|
|
t.Fatalf("err=%v", err)
|
|
}
|
|
}
|
|
request := liveRequest()
|
|
request.MaxSeries = 0
|
|
samples, err := sampler.Sample(context.Background(), request)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(samples) != 3 {
|
|
t.Fatalf("samples=%+v", samples)
|
|
}
|
|
unbounded := testSampler(t, jsonServer(t, nil, body), livesampler.Options{MaxSeries: 2}, prometheus.Limits{})
|
|
if _, err := unbounded.Sample(context.Background(), request); err == nil {
|
|
t.Fatal("expected the sampler series budget to bound an unbounded request")
|
|
}
|
|
}
|
|
|
|
func TestSamplerKeepsOnlyTheNewestPointPerSeries(t *testing.T) {
|
|
body := fmt.Sprintf(`{"status":"success","data":{"resultType":"matrix","result":[
|
|
{"metric":{"container":"media_server"},"values":[[%s,"1"],[%s,"2"],[%s,"3"]]}]}}`,
|
|
epoch(fixedNow().Add(-30*time.Second)), epoch(fixedNow().Add(-20*time.Second)), epoch(fixedNow().Add(-10*time.Second)))
|
|
sampler := testSampler(t, jsonServer(t, nil, body), livesampler.Options{}, prometheus.Limits{})
|
|
samples, err := sampler.Sample(context.Background(), liveRequest())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(samples) != 1 || samples[0].Value == nil || *samples[0].Value != 3 {
|
|
t.Fatalf("samples=%+v", samples)
|
|
}
|
|
if samples[0].Timestamp != fixedNow().Add(-10*time.Second) || samples[0].Freshness != "fresh" {
|
|
t.Fatalf("samples=%+v", samples)
|
|
}
|
|
}
|
|
|
|
func TestSamplerRejectsUnknownMetricAndUnusableConstruction(t *testing.T) {
|
|
sampler := testSampler(t, jsonServer(t, nil, `{"status":"success","data":{"resultType":"vector","result":[]}}`), livesampler.Options{}, prometheus.Limits{})
|
|
request := liveRequest()
|
|
request.Metric = "container.cpu.unknown"
|
|
var sampleError livesampler.Error
|
|
if _, err := sampler.Sample(context.Background(), request); !errors.As(err, &sampleError) || sampleError.Code != "LIVE_SAMPLE_METRIC_UNKNOWN" {
|
|
t.Fatalf("err=%v", err)
|
|
}
|
|
registry, err := metriccatalog.DefaultRegistry()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := livesampler.New(registry, nil, livesampler.Options{}); !errors.Is(err, livesampler.ErrSourceRequired) {
|
|
t.Fatalf("err=%v", err)
|
|
}
|
|
if _, err := livesampler.New(metriccatalog.Registry{}, stubSource{}, livesampler.Options{}); !errors.Is(err, livesampler.ErrCatalogRequired) {
|
|
t.Fatalf("err=%v", err)
|
|
}
|
|
if _, err := livesampler.New(registry, stubSource{}, livesampler.Options{MaxSeries: 100000}); !errors.Is(err, livesampler.ErrOptionsInvalid) {
|
|
t.Fatalf("err=%v", err)
|
|
}
|
|
if _, err := livesampler.New(registry, stubSource{}, livesampler.Options{Timeout: time.Hour}); !errors.Is(err, livesampler.ErrOptionsInvalid) {
|
|
t.Fatalf("err=%v", err)
|
|
}
|
|
var missing *livesampler.Sampler
|
|
if _, err := missing.Sample(context.Background(), liveRequest()); !errors.Is(err, livesampler.ErrSourceRequired) {
|
|
t.Fatalf("err=%v", err)
|
|
}
|
|
}
|
|
|
|
type stubSource struct{}
|
|
|
|
func (stubSource) Query(context.Context, string, *time.Time) (prometheus.QueryResult, error) {
|
|
return prometheus.QueryResult{}, nil
|
|
}
|
|
|
|
func TestSamplerHonorsCallerCancellation(t *testing.T) {
|
|
sampler := testSampler(t, jsonServer(t, nil, `{"status":"success","data":{"resultType":"vector","result":[]}}`), livesampler.Options{}, prometheus.Limits{})
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
if _, err := sampler.Sample(ctx, liveRequest()); !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("err=%v", err)
|
|
}
|
|
}
|
|
|
|
func TestSamplerIsSafeForConcurrentSubscriptions(t *testing.T) {
|
|
body := fmt.Sprintf(`{"status":"success","data":{"resultType":"vector","result":[
|
|
{"metric":{"container":"media_server"},"value":[%s,"5"]}]}}`, epoch(fixedNow()))
|
|
sampler := testSampler(t, jsonServer(t, &recorder{}, body), livesampler.Options{}, prometheus.Limits{MaxConcurrency: 4})
|
|
var group sync.WaitGroup
|
|
failures := make(chan error, 32)
|
|
for worker := 0; worker < 8; worker++ {
|
|
group.Add(1)
|
|
go func() {
|
|
defer group.Done()
|
|
for round := 0; round < 8; round++ {
|
|
samples, err := sampler.Sample(context.Background(), liveRequest())
|
|
if err != nil {
|
|
failures <- err
|
|
return
|
|
}
|
|
if len(samples) != 1 || samples[0].Freshness != "fresh" || samples[0].Value == nil || *samples[0].Value != 5 {
|
|
failures <- fmt.Errorf("unexpected samples: %+v", samples)
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
group.Wait()
|
|
close(failures)
|
|
for err := range failures {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func TestSamplerSatisfiesLiveSampler(t *testing.T) {
|
|
sampler := testSampler(t, jsonServer(t, nil, `{"status":"success","data":{"resultType":"vector","result":[]}}`), livesampler.Options{}, prometheus.Limits{})
|
|
var contract live.Sampler = sampler
|
|
samples, err := contract.Sample(context.Background(), liveRequest())
|
|
if err != nil || len(samples) == 0 {
|
|
t.Fatalf("samples=%+v err=%v", samples, err)
|
|
}
|
|
}
|