From 3c9289e2f89dbf037e403c3661a10c79e2fd3764 Mon Sep 17 00:00:00 2001 From: Bram Schuur Date: Mon, 14 Sep 2026 14:47:30 +0200 Subject: [PATCH 1/2] Add bounded Receiver feature queries --- pkg/openapiclient/client.go | 157 ++--- pkg/openapiclient/client_test.go | 242 +++++++- pkg/openapiclient/features/features_client.go | 200 ++++++ .../features/features_client_test.go | 574 ++++++++++++++++++ pkg/openapiclient/features/result.go | 28 + pkg/openapiclient/options.go | 50 ++ pkg/openapiclient/transport.go | 69 +++ 7 files changed, 1187 insertions(+), 133 deletions(-) create mode 100644 pkg/openapiclient/features/features_client.go create mode 100644 pkg/openapiclient/features/features_client_test.go create mode 100644 pkg/openapiclient/features/result.go create mode 100644 pkg/openapiclient/options.go create mode 100644 pkg/openapiclient/transport.go diff --git a/pkg/openapiclient/client.go b/pkg/openapiclient/client.go index 043be99..7d551af 100644 --- a/pkg/openapiclient/client.go +++ b/pkg/openapiclient/client.go @@ -2,138 +2,69 @@ package openapiclient import ( "context" - "crypto/tls" - "net" + "errors" + "fmt" "net/http" - "net/url" "strings" - "time" - - "golang.org/x/oauth2" "github.com/StackVista/stackstate-receiver-go-client/generated/receiver_api" - log "github.com/cihub/seelog" + "golang.org/x/oauth2" ) -// OpenAPIClient provides a client for connecting to the openapi generated portion of the receiver api -type OpenAPIClient interface { - Connect() *receiver_api.APIClient -} - -// NewOpenAPIClient constructs the OpenAPIClient client -func NewOpenAPIClient(ctx context.Context, - isVerbose bool, - userAgent string, - url string, - apiToken string, - serviceAccountToken func() string, - skipSSL bool, - proxy *url.URL) (OpenAPIClient, context.Context) { - baseURL := makeBaseURL(url) - client, clientAuth := newAPIClient(isVerbose, userAgent, baseURL, apiToken, serviceAccountToken, skipSSL, proxy) - - withClient := ctx - if clientAuth != nil { - withClient = context.WithValue( - ctx, - receiver_api.ContextOAuth2, - clientAuth, - ) +// NewOpenAPIClient constructs a Receiver client and its authenticated context. +func NewOpenAPIClient(parent context.Context, opts ConnectionOptions) (*receiver_api.APIClient, context.Context, error) { + if parent == nil { + return nil, nil, errors.New("parent context is required") } - - return openAPIClientImpl{ - client: client, - Context: withClient, - receiverURL: baseURL, - }, withClient -} - -func newAPIClient( - isVerbose bool, - userAgent string, - receiverURL string, - apiKey string, - serviceAccountToken func() string, - skipSSL bool, - proxy *url.URL, -) (*receiver_api.APIClient, oauth2.TokenSource) { - configuration := receiver_api.NewConfiguration() - - transport := &http.Transport{ - DialContext: (&net.Dialer{ - Timeout: 30 * time.Second, - KeepAlive: 30 * time.Second, - }).DialContext, - MaxIdleConns: 100, - IdleConnTimeout: 90 * time.Second, - TLSHandshakeTimeout: 10 * time.Second, - ExpectContinueTimeout: 1 * time.Second, - TLSClientConfig: &tls.Config{InsecureSkipVerify: skipSSL}, + if _, err := parseEndpoint(opts.ReceiverURL, false); err != nil { + return nil, nil, fmt.Errorf("invalid receiver URL: %w", err) } - - if skipSSL { - log.Warnf("Using univerified ssl connection") + if opts.RequestTimeout <= 0 { + return nil, nil, errors.New("request timeout must be positive") } - - if proxy != nil { - log.Infof("configuring proxy through: %s", proxy.String()) - transport.Proxy = http.ProxyURL(proxy) + if opts.APIKey != "" && opts.ServiceAccountToken != nil { + return nil, nil, errors.New("exactly one receiver authentication source is required") } - - configuration.HTTPClient = &http.Client{Timeout: 30 * time.Second, Transport: transport} - configuration.UserAgent = userAgent - configuration.Servers[0] = receiver_api.ServerConfiguration{ - URL: receiverURL, - Description: "", - Variables: nil, + source := dynamicTokenSource{tokenFunc: opts.ServiceAccountToken, tokenType: "ServiceBearer"} + if opts.APIKey != "" { + source = dynamicTokenSource{tokenFunc: func() string { return opts.APIKey }, tokenType: "ApiKey"} } - configuration.Debug = isVerbose - - client := receiver_api.NewAPIClient(configuration) - - if apiKey != "" { - return client, oauth2.StaticTokenSource(&oauth2.Token{ - AccessToken: apiKey, - TokenType: "ApiKey", - }) + if _, err := source.Token(); err != nil { + return nil, nil, err } - token := serviceAccountToken() - if token != "" { - return client, dynamicTokenSource{tokenFunc: serviceAccountToken, tokenType: "ServiceBearer"} + transport, err := newTransport(opts) + if err != nil { + return nil, nil, err } - - return client, nil + cfg := receiver_api.NewConfiguration() + cfg.HTTPClient = &http.Client{ + Timeout: opts.RequestTimeout, + Transport: transport, + CheckRedirect: func(_ *http.Request, _ []*http.Request) error { return http.ErrUseLastResponse }, + } + cfg.UserAgent = opts.UserAgent + cfg.Servers[0] = receiver_api.ServerConfiguration{URL: makeBaseURL(opts.ReceiverURL)} + cfg.Debug = false + authCtx := context.WithValue(parent, receiver_api.ContextOAuth2, source) + return receiver_api.NewAPIClient(cfg), authCtx, nil } -// dynamicTokenSource calls tokenFunc on every Token() invocation so that -// refreshed credentials (e.g. rotated Kubernetes service-account tokens) are -// picked up automatically. type dynamicTokenSource struct { tokenFunc func() string tokenType string } func (d dynamicTokenSource) Token() (*oauth2.Token, error) { - return &oauth2.Token{ - AccessToken: d.tokenFunc(), - TokenType: d.tokenType, - }, nil -} - -type openAPIClientImpl struct { - client *receiver_api.APIClient - Context context.Context - receiverURL string -} - -func (c openAPIClientImpl) Connect() *receiver_api.APIClient { - // Placeholder in case we want to do something while connecting - log.Infof("Connected to receiver: %s", c.receiverURL) - - return c.client -} - -// Drop /stsAgent/ part from the url is it exists, because it is included in openapi -func makeBaseURL(url string) string { - return strings.TrimSuffix(strings.Trim(url, "/"), "/stsAgent") + if d.tokenFunc == nil { + return nil, ErrMissingCredential + } + token := d.tokenFunc() + if strings.TrimSpace(token) == "" { + return nil, ErrMissingCredential + } + // Reject header delimiters here so transport errors cannot echo credentials. + if strings.ContainsAny(token, "\r\n") { + return nil, ErrMissingCredential + } + return &oauth2.Token{AccessToken: token, TokenType: d.tokenType}, nil } diff --git a/pkg/openapiclient/client_test.go b/pkg/openapiclient/client_test.go index e49f31c..8c19291 100644 --- a/pkg/openapiclient/client_test.go +++ b/pkg/openapiclient/client_test.go @@ -1,38 +1,240 @@ package openapiclient import ( + "context" + "crypto/x509" + "encoding/pem" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" "testing" + "time" + "github.com/StackVista/stackstate-receiver-go-client/generated/receiver_api" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -func TestMakeBaseUrl(t *testing.T) { - assert.Equal(t, makeBaseURL("https://bla/"), "https://bla") - assert.Equal(t, makeBaseURL("https://bla"), "https://bla") - assert.Equal(t, makeBaseURL("https://bla/stsAgent/"), "https://bla") - assert.Equal(t, makeBaseURL("https://bla/stsAgent"), "https://bla") +func testOptions(endpoint string) ConnectionOptions { + return ConnectionOptions{ReceiverURL: endpoint, APIKey: "synthetic-key", RequestTimeout: time.Second} } -func TestServiceAccountTokenSourceReturnsRefreshedToken(t *testing.T) { - currentToken := "initial-token" - tokenFunc := func() string { return currentToken } +func TestConnectionURLAndContext(t *testing.T) { + for _, suffix := range []string{"", "/", "/stsAgent", "/stsAgent/", "/stsAgent///", "/receiver/stsAgent/"} { + t.Run(suffix, func(t *testing.T) { + expectedPath := "/stsAgent/features" + if strings.HasPrefix(suffix, "/receiver") { + expectedPath = "/receiver/stsAgent/features" + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, expectedPath, r.URL.Path) + assert.Equal(t, "ApiKey synthetic-key", r.Header.Get("Authorization")) + assert.Equal(t, "test-agent", r.UserAgent()) + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"otel-logs":true}`) + })) + defer server.Close() + type contextKey struct{} + parent, cancel := context.WithCancel(context.WithValue(context.Background(), contextKey{}, "preserved")) + defer cancel() + opts := testOptions(server.URL + suffix) + opts.UserAgent = "test-agent" + api, ctx, err := NewOpenAPIClient(parent, opts) + require.NoError(t, err) + assert.Equal(t, "preserved", ctx.Value(contextKey{})) + result, response, err := api.FeaturesAPI.GetFeatures(ctx).Execute() + require.NoError(t, err) + response.Body.Close() + assert.Equal(t, true, result["otel-logs"]) + cancel() + _, _, err = api.FeaturesAPI.GetFeatures(ctx).Execute() + require.ErrorIs(t, err, context.Canceled) + }) + } +} - _, tokenSource := newAPIClient(false, "test-agent", "https://receiver", "", tokenFunc, true, nil) - require.NotNil(t, tokenSource, "tokenSource should not be nil when a service account token is provided") +func TestConnectionValidationIsCredentialSafe(t *testing.T) { + cases := map[string]func(*ConnectionOptions){ + "relative": func(o *ConnectionOptions) { o.ReceiverURL = "/receiver" }, + "scheme": func(o *ConnectionOptions) { o.ReceiverURL = "ftp://receiver" }, + "missing host": func(o *ConnectionOptions) { o.ReceiverURL = "http:///path" }, + "userinfo": func(o *ConnectionOptions) { o.ReceiverURL = "https://user:synthetic-secret@receiver" }, + "query": func(o *ConnectionOptions) { o.ReceiverURL += "?key=synthetic-secret" }, + "empty query": func(o *ConnectionOptions) { o.ReceiverURL += "?" }, + "fragment": func(o *ConnectionOptions) { o.ReceiverURL += "#synthetic-secret" }, + "empty fragment": func(o *ConnectionOptions) { o.ReceiverURL += "#" }, + "malformed URL": func(o *ConnectionOptions) { o.ReceiverURL = "https://synthetic-secret%" }, + "timeout": func(o *ConnectionOptions) { o.RequestTimeout = 0 }, + "no auth": func(o *ConnectionOptions) { o.APIKey = "" }, + "blank auth": func(o *ConnectionOptions) { o.APIKey = " \t" }, + "invalid header": func(o *ConnectionOptions) { o.APIKey = "synthetic-secret\r\n" }, + "empty token": func(o *ConnectionOptions) { o.APIKey = ""; o.ServiceAccountToken = func() string { return "" } }, + "two auth sources": func(o *ConnectionOptions) { o.ServiceAccountToken = func() string { return "synthetic-secret" } }, + "CA": func(o *ConnectionOptions) { o.CABundlePEM = []byte("synthetic-secret") }, + "proxy": func(o *ConnectionOptions) { o.ProxyURL = "socks5://synthetic-secret" }, + } + for name, modify := range cases { + t.Run(name, func(t *testing.T) { + opts := testOptions("https://receiver/stsAgent") + modify(&opts) + client, ctx, err := NewOpenAPIClient(context.Background(), opts) + require.Error(t, err) + assert.Nil(t, client) + assert.Nil(t, ctx) + assert.NotContains(t, err.Error(), "synthetic-secret") + }) + } + _, _, err := NewOpenAPIClient(nil, testOptions("https://receiver")) + require.Error(t, err) +} - // First call should return the initial token - tok, err := tokenSource.Token() +func TestRotatingTokenUsedByFeaturesAndRBAC(t *testing.T) { + var token atomic.Value + token.Store("first") + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + assert.Equal(t, "ServiceBearer "+token.Load().(string), r.Header.Get("Authorization")) + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{}`) + })) + defer server.Close() + opts := testOptions(server.URL) + opts.APIKey = "" + opts.ServiceAccountToken = func() string { return token.Load().(string) } + api, ctx, err := NewOpenAPIClient(context.Background(), opts) require.NoError(t, err) - assert.Equal(t, "initial-token", tok.AccessToken) - assert.Equal(t, "ServiceBearer", tok.TokenType) + payload := receiver_api.RBACSnapshotRequestAsRBACRequest(&receiver_api.RBACSnapshotRequest{}) + for _, value := range []string{"first", "rotated"} { + token.Store(value) + _, response, err := api.FeaturesAPI.GetFeatures(ctx).Execute() + require.NoError(t, err) + response.Body.Close() + response, err = api.ReceiverRbacInstanceAPI.IngestInstanceRBAC(ctx).RBACRequest(payload).Execute() + require.NoError(t, err) + response.Body.Close() + } + token.Store("") + _, _, err = api.FeaturesAPI.GetFeatures(ctx).Execute() + require.ErrorIs(t, err, ErrMissingCredential) + _, err = api.ReceiverRbacInstanceAPI.IngestInstanceRBAC(ctx).RBACRequest(payload).Execute() + require.ErrorIs(t, err, ErrMissingCredential) + assert.EqualValues(t, 4, requests.Load()) +} - // Simulate token rotation (e.g. PeriodicTokenFileReader picked up a new token) - currentToken = "refreshed-token" +func TestTLSOptions(t *testing.T) { + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{}`) + })) + defer server.Close() + ca := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: server.Certificate().Raw}) + for _, tc := range []struct { + name string + ca []byte + skip bool + succeeds bool + }{ + {name: "untrusted"}, {name: "custom CA", ca: ca, succeeds: true}, {name: "explicit skip", skip: true, succeeds: true}, + } { + t.Run(tc.name, func(t *testing.T) { + opts := testOptions(server.URL) + opts.CABundlePEM = tc.ca + opts.InsecureSkipVerify = tc.skip + api, ctx, err := NewOpenAPIClient(context.Background(), opts) + require.NoError(t, err) + _, response, err := api.FeaturesAPI.GetFeatures(ctx).Execute() + if response != nil { + response.Body.Close() + } + if tc.succeeds { + require.NoError(t, err) + } else { + require.Error(t, err) + } + }) + } + opts := testOptions(server.URL) + opts.CABundlePEM = ca + transport, err := newTransport(opts) + require.NoError(t, err) + roots := transport.(boundedTransport).base.(*http.Transport).TLSClientConfig.RootCAs + system, err := x509.SystemCertPool() + require.NoError(t, err) + assert.GreaterOrEqual(t, len(roots.Subjects()), len(system.Subjects())) +} - // Subsequent call must return the refreshed token, not the stale one - tok, err = tokenSource.Token() +func TestProxyAndRedirectBoundaries(t *testing.T) { + var proxyCalls atomic.Int32 + proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + proxyCalls.Add(1) + assert.Equal(t, "http://unresolvable.invalid/stsAgent/features", r.URL.String()) + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{}`) + })) + defer proxy.Close() + opts := testOptions("http://unresolvable.invalid") + opts.ProxyURL = proxy.URL + api, ctx, err := NewOpenAPIClient(context.Background(), opts) + require.NoError(t, err) + _, response, err := api.FeaturesAPI.GetFeatures(ctx).Execute() require.NoError(t, err) - assert.Equal(t, "refreshed-token", tok.AccessToken, - "token source must return the latest token, not a cached value from initialization") + response.Body.Close() + assert.EqualValues(t, 1, proxyCalls.Load()) + t.Setenv("HTTP_PROXY", proxy.URL) + t.Setenv("HTTPS_PROXY", proxy.URL) + transport, err := newTransport(testOptions("http://receiver")) + require.NoError(t, err) + assert.Nil(t, transport.(boundedTransport).base.(*http.Transport).Proxy) + var redirected atomic.Int32 + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { redirected.Add(1); fmt.Fprint(w, `{}`) })) + defer target.Close() + redirect := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { http.Redirect(w, r, target.URL, 302) })) + defer redirect.Close() + api, ctx, err = NewOpenAPIClient(context.Background(), testOptions(redirect.URL)) + require.NoError(t, err) + _, response, err = api.FeaturesAPI.GetFeatures(ctx).Execute() + require.Error(t, err) + require.NotNil(t, response) + response.Body.Close() + assert.Equal(t, 302, response.StatusCode) + assert.Zero(t, redirected.Load()) +} + +func TestBoundedBody(t *testing.T) { + for _, size := range []int{maxFeatureResponseBytes - 1, maxFeatureResponseBytes, maxFeatureResponseBytes + 1} { + body := &boundedBody{ReadCloser: io.NopCloser(strings.NewReader(strings.Repeat("x", size))), remaining: maxFeatureResponseBytes} + content, err := io.ReadAll(body) + if size > maxFeatureResponseBytes { + require.ErrorIs(t, err, ErrResponseTooLarge) + } else { + require.NoError(t, err) + } + assert.LessOrEqual(t, len(content), maxFeatureResponseBytes) + } +} + +func TestEndpointPorts(t *testing.T) { + for _, proxy := range []bool{false, true} { + for _, port := range []string{"0", "65536", "9999999999999999999999", "", "-1", "named"} { + t.Run(fmt.Sprintf("proxy_%t_port_%s", proxy, port), func(t *testing.T) { + opts := testOptions("https://receiver/stsAgent") + invalid := "https://receiver:" + port + if proxy { + opts.ProxyURL = invalid + } else { + opts.ReceiverURL = invalid + "/stsAgent" + } + _, _, err := NewOpenAPIClient(context.Background(), opts) + require.Error(t, err) + }) + } + } + for _, endpoint := range []string{"https://receiver:443/stsAgent", "http://[::1]:8080/stsAgent", "http://[::1]/stsAgent"} { + _, _, err := NewOpenAPIClient(context.Background(), testOptions(endpoint)) + require.NoError(t, err) + } } diff --git a/pkg/openapiclient/features/features_client.go b/pkg/openapiclient/features/features_client.go new file mode 100644 index 0000000..de08f48 --- /dev/null +++ b/pkg/openapiclient/features/features_client.go @@ -0,0 +1,200 @@ +package features + +import ( + "context" + "crypto/tls" + "crypto/x509" + "errors" + "io" + "math/rand/v2" + "net" + "net/http" + "strconv" + "strings" + "time" + + "github.com/StackVista/stackstate-receiver-go-client/generated/receiver_api" + "github.com/StackVista/stackstate-receiver-go-client/pkg/openapiclient" +) + +// QueryOptions bounds a complete feature query, including attempts and retry waits. +type QueryOptions struct { + Timeout, AttemptTimeout time.Duration + MaxAttempts int + InitialBackoff, MaxBackoff time.Duration +} + +// Client queries the generated features API with bounded retries. +type Client struct { + api receiver_api.FeaturesAPI + opts QueryOptions + now func() time.Time + random func() float64 +} + +// NewClient validates query bounds and constructs a feature client. +func NewClient(api receiver_api.FeaturesAPI, opts QueryOptions) (*Client, error) { + if api == nil { + return nil, errors.New("features API is required") + } + if opts.Timeout <= 0 || opts.AttemptTimeout <= 0 || opts.AttemptTimeout > opts.Timeout || opts.MaxAttempts < 1 || opts.InitialBackoff <= 0 || opts.MaxBackoff < opts.InitialBackoff { + return nil, errors.New("invalid feature query timeout, attempt count or backoff bounds") + } + return &Client{api: api, opts: opts, now: time.Now, random: rand.Float64}, nil +} + +// FetchFeatures returns one observation after the bounded query completes. +func (c *Client) FetchFeatures(authCtx context.Context) Result { + queryCtx, cancel := context.WithTimeout(authCtx, c.opts.Timeout) + defer cancel() + result := Result{} + backoff := c.opts.InitialBackoff + for { + if queryCtx.Err() != nil { + result.Class = contextClass(authCtx, queryCtx.Err()) + break + } + attemptCtx, stop := context.WithTimeout(queryCtx, c.opts.AttemptTimeout) + values, response, err := c.api.GetFeaturesExecute(c.api.GetFeatures(attemptCtx)) + result.Attempts++ + result.StatusCode = 0 + retryAfter := time.Duration(0) + if response != nil { + result.StatusCode = response.StatusCode + if response.StatusCode == http.StatusTooManyRequests || response.StatusCode == http.StatusServiceUnavailable { + retryAfter = parseRetryAfter(response.Header.Get("Retry-After"), c.now(), c.opts.Timeout) + } + if response.Body != nil { + response.Body.Close() + } + } + result.Class = classify(authCtx, attemptCtx, values, response, err) + stop() + if result.Class == Valid { + result.Features = values + } + if (result.Class != Transient && result.Class != Timeout) || result.Attempts >= c.opts.MaxAttempts { + break + } + if queryCtx.Err() != nil { + result.Class = contextClass(authCtx, queryCtx.Err()) + break + } + delay := time.Duration(c.random() * float64(backoff)) + if retryAfter > delay { + delay = retryAfter + } + deadline, _ := queryCtx.Deadline() + if delay >= time.Until(deadline) { + break + } + timer := time.NewTimer(delay) + select { + case <-queryCtx.Done(): + timer.Stop() + result.Class = contextClass(authCtx, queryCtx.Err()) + result.FinishedAt = c.now() + return result + case <-timer.C: + } + if backoff > c.opts.MaxBackoff/2 { + backoff = c.opts.MaxBackoff + } else { + backoff *= 2 + } + } + result.FinishedAt = c.now() + return result +} + +func contextClass(parent context.Context, err error) Class { + if parent.Err() != nil { + return Canceled + } + if errors.Is(err, context.DeadlineExceeded) { + return Timeout + } + return Canceled +} + +func classify(parent, attempt context.Context, values map[string]any, response *http.Response, err error) Class { + if parent.Err() != nil { + return Canceled + } + if response != nil { + switch status := response.StatusCode; { + case status == 401 || status == 403: + return Authentication + case status == 404: + return Unsupported + case status == 408 || status == 429 || status >= 500 && status <= 599: + return Transient + case status != 200: + return Rejected + } + } + if errors.Is(err, openapiclient.ErrMissingCredential) { + return Authentication + } + var verification *tls.CertificateVerificationError + var unknown x509.UnknownAuthorityError + var hostname x509.HostnameError + var invalid x509.CertificateInvalidError + if errors.As(err, &verification) || errors.As(err, &unknown) || errors.As(err, &hostname) || errors.As(err, &invalid) { + return Configuration + } + if attempt.Err() != nil || errors.Is(err, context.DeadlineExceeded) { + return Timeout + } + if errors.Is(err, context.Canceled) { + return Canceled + } + var network net.Error + if errors.As(err, &network) && network.Timeout() { + return Timeout + } + var operation *net.OpError + if errors.As(err, &operation) || errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) || (errors.As(err, &network) && network.Temporary()) { + return Transient + } + if response == nil { + if err != nil { + return Transient + } + return Rejected + } + if err != nil || values == nil { + return Malformed + } + if capability, present := values["otel-logs"]; present { + if _, ok := capability.(bool); !ok { + return Malformed + } + } + return Valid +} + +func parseRetryAfter(value string, now time.Time, limit time.Duration) time.Duration { + value = strings.TrimSpace(value) + if value == "" { + return 0 + } + if seconds, err := strconv.ParseUint(value, 10, 64); err == nil { + if seconds > uint64(limit/time.Second) { + return limit + } + return time.Duration(seconds) * time.Second + } else if errors.Is(err, strconv.ErrRange) && strings.Trim(value, "0123456789") == "" { + return limit + } + if deadline, err := http.ParseTime(value); err == nil { + delay := deadline.Sub(now) + if delay > limit { + return limit + } + if delay > 0 { + return delay + } + } + return 0 +} diff --git a/pkg/openapiclient/features/features_client_test.go b/pkg/openapiclient/features/features_client_test.go new file mode 100644 index 0000000..4a05345 --- /dev/null +++ b/pkg/openapiclient/features/features_client_test.go @@ -0,0 +1,574 @@ +package features_test + +import ( + "context" + "crypto/x509" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/url" + "reflect" + "strings" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + "github.com/StackVista/stackstate-receiver-go-client/generated/receiver_api" + "github.com/StackVista/stackstate-receiver-go-client/pkg/openapiclient" + "github.com/StackVista/stackstate-receiver-go-client/pkg/openapiclient/features" +) + +const ( + queryTestAPIKey = "synthetic-query-api-key" + queryTestBody = "synthetic-private-response-body" + queryTestError = "synthetic-private-transport-error" +) + +func queryTestOptions() features.QueryOptions { + return features.QueryOptions{ + Timeout: 10 * time.Second, + AttemptTimeout: time.Second, + MaxAttempts: 3, + InitialBackoff: time.Nanosecond, + MaxBackoff: time.Nanosecond, + } +} + +func newQueryTestClient(t *testing.T, api receiver_api.FeaturesAPI, opts features.QueryOptions) *features.Client { + t.Helper() + client, err := features.NewClient(api, opts) + if err != nil { + t.Fatalf("NewClient: %v", err) + } + return client +} + +func newHTTPQueryTestClient(t *testing.T, handler http.HandlerFunc, opts features.QueryOptions) (*features.Client, context.Context) { + t.Helper() + server := httptest.NewServer(handler) + t.Cleanup(server.Close) + api, authCtx, err := openapiclient.NewOpenAPIClient(context.Background(), openapiclient.ConnectionOptions{ + ReceiverURL: server.URL + "/deployment/stsAgent/", + APIKey: queryTestAPIKey, + RequestTimeout: 5 * time.Second, + }) + if err != nil { + t.Fatalf("NewOpenAPIClient: %v", err) + } + return newQueryTestClient(t, api.FeaturesAPI, opts), authCtx +} + +func checkQueryResult(t *testing.T, result features.Result, class features.Class, status, attempts int, started time.Time) { + t.Helper() + if result.Class != class || result.StatusCode != status || result.Attempts != attempts { + t.Errorf("result class/status/attempts = %v/%d/%d; want %v/%d/%d", + result.Class, result.StatusCode, result.Attempts, class, status, attempts) + } + if result.FinishedAt.IsZero() || result.FinishedAt.Before(started) || result.FinishedAt.After(time.Now()) { + t.Error("FinishedAt must be the query completion time") + } + if class != features.Valid && result.Features != nil { + t.Error("failed query returned feature data") + } + encoded, err := json.Marshal(result) + if err != nil { + t.Fatalf("marshal Result: %v", err) + } + for _, rendering := range []string{string(encoded), fmt.Sprint(result), fmt.Sprintf("%+v", result), fmt.Sprintf("%#v", result)} { + for _, secret := range []string{queryTestAPIKey, queryTestBody, queryTestError} { + if strings.Contains(rendering, secret) { + t.Error("Result exposed synthetic sensitive data") + } + } + } +} + +func TestFetchFeaturesHTTPStatusAndShape(t *testing.T) { + tests := []struct { + name string + status int + contentType string + body string + class features.Class + want map[string]any + }{ + {"empty_object", 200, "application/json", `{}`, features.Valid, map[string]any{}}, + {"enabled", 200, "application/json", `{"otel-logs":true}`, features.Valid, map[string]any{"otel-logs": true}}, + {"disabled", 200, "application/json", `{"otel-logs":false}`, features.Valid, map[string]any{"otel-logs": false}}, + {"legacy_numeric_and_unknown_values", 200, "application/json", `{"rbac":true,"capacity":42,"label":"capability","nested":{"a":[1,null]}}`, features.Valid, + map[string]any{"rbac": true, "capacity": float64(42), "label": "capability", "nested": map[string]any{"a": []any{float64(1), nil}}}}, + {"json_charset", 200, "application/json; charset=utf-8", `{"otel-logs":true}`, features.Valid, map[string]any{"otel-logs": true}}, + {"empty_body", 200, "application/json", "", features.Malformed, nil}, + {"whitespace", 200, "application/json", " \n\t", features.Malformed, nil}, + {"null", 200, "application/json", `null`, features.Malformed, nil}, + {"array", 200, "application/json", `[]`, features.Malformed, nil}, + {"string", 200, "application/json", `"value"`, features.Malformed, nil}, + {"number", 200, "application/json", `42`, features.Malformed, nil}, + {"boolean", 200, "application/json", `true`, features.Malformed, nil}, + {"truncated", 200, "application/json", `{"otel-logs":`, features.Malformed, nil}, + {"trailing_json", 200, "application/json", `{} {}`, features.Malformed, nil}, + {"trailing_garbage", 200, "application/json", `{} garbage`, features.Malformed, nil}, + {"flag_string", 200, "application/json", `{"otel-logs":"true"}`, features.Malformed, nil}, + {"flag_number", 200, "application/json", `{"otel-logs":1}`, features.Malformed, nil}, + {"flag_null", 200, "application/json", `{"otel-logs":null}`, features.Malformed, nil}, + {"flag_array", 200, "application/json", `{"otel-logs":[]}`, features.Malformed, nil}, + {"flag_object", 200, "application/json", `{"otel-logs":{}}`, features.Malformed, nil}, + {"html_success", 200, "text/html", "" + queryTestBody + "", features.Malformed, nil}, + {"sensitive_decode_error", 200, "application/json", `{"` + queryTestAPIKey + `":` + queryTestBody, features.Malformed, nil}, + } + for _, status := range []int{401, 403, 404, 408, 429, 500, 502, 503, 504, 599, 400, 405, 409, 413, 418, 301, 302, 303, 304, 307, 308, 201, 202, 204, 206} { + class := features.Rejected + switch { + case status == 401 || status == 403: + class = features.Authentication + case status == 404: + class = features.Unsupported + case status == 408 || status == 429 || status >= 500: + class = features.Transient + } + tests = append(tests, struct { + name string + status int + contentType string + body string + class features.Class + want map[string]any + }{fmt.Sprintf("status_%d_html", status), status, "text/html", "" + queryTestBody + queryTestAPIKey + "", class, nil}) + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var calls atomic.Int32 + client, ctx := newHTTPQueryTestClient(t, func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + if r.Method != http.MethodGet || r.URL.Path != "/deployment/stsAgent/features" || r.URL.RawQuery != "" { + t.Error("unexpected feature request method or URL") + } + if r.Header.Get("Authorization") != "ApiKey "+queryTestAPIKey { + t.Error("attempt lost authenticated context") + } + w.Header().Set("Content-Type", tt.contentType) + w.Header().Set("Location", "/must-not-follow") + w.WriteHeader(tt.status) + _, _ = io.WriteString(w, tt.body) + }, queryTestOptions()) + started := time.Now() + result := client.FetchFeatures(ctx) + attempts := 1 + if tt.class == features.Transient { + attempts = 3 + } + checkQueryResult(t, result, tt.class, tt.status, attempts, started) + if int(calls.Load()) != attempts { + t.Errorf("HTTP calls = %d; want %d", calls.Load(), attempts) + } + if !reflect.DeepEqual(result.Features, tt.want) { + t.Error("decoded feature map differs from expected values") + } + }) + } +} + +func TestFetchFeaturesHTTPResponseLimit(t *testing.T) { + const limit = 1 << 20 + for _, chunked := range []bool{false, true} { + for _, size := range []int{limit - 1, limit, limit + 1} { + t.Run(fmt.Sprintf("chunked_%t_bytes_%d", chunked, size), func(t *testing.T) { + body := `{"padding":"` + strings.Repeat("x", size-len(`{"padding":""}`)) + `"}` + opts := queryTestOptions() + opts.MaxAttempts = 1 + client, ctx := newHTTPQueryTestClient(t, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + if chunked { + w.(http.Flusher).Flush() + } else { + w.Header().Set("Content-Length", fmt.Sprint(len(body))) + } + _, _ = io.WriteString(w, body) + }, opts) + started := time.Now() + result := client.FetchFeatures(ctx) + class := features.Valid + if size > limit { + class = features.Malformed + } + checkQueryResult(t, result, class, http.StatusOK, 1, started) + if class == features.Valid && result.Features["padding"] != strings.Repeat("x", size-len(`{"padding":""}`)) { + t.Error("in-bound response was truncated") + } + }) + } + } + for _, status := range []int{401, 404, 503} { + t.Run(fmt.Sprintf("oversized_status_%d", status), func(t *testing.T) { + opts := queryTestOptions() + opts.MaxAttempts = 1 + client, ctx := newHTTPQueryTestClient(t, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/html") + w.WriteHeader(status) + _, _ = io.WriteString(w, strings.Repeat(queryTestBody, limit/len(queryTestBody)+1)) + }, opts) + class := map[int]features.Class{401: features.Authentication, 404: features.Unsupported, 503: features.Transient}[status] + started := time.Now() + checkQueryResult(t, client.FetchFeatures(ctx), class, status, 1, started) + }) + } +} + +func TestFetchFeaturesHTTPTransientRecovery(t *testing.T) { + for _, status := range []int{408, 429, 500, 503} { + t.Run(fmt.Sprint(status), func(t *testing.T) { + var calls atomic.Int32 + client, ctx := newHTTPQueryTestClient(t, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + if calls.Add(1) < 3 { + w.WriteHeader(status) + _, _ = io.WriteString(w, queryTestBody) + return + } + _, _ = io.WriteString(w, `{"otel-logs":true}`) + }, queryTestOptions()) + started := time.Now() + result := client.FetchFeatures(ctx) + checkQueryResult(t, result, features.Valid, 200, 3, started) + if calls.Load() != 3 || result.Features["otel-logs"] != true { + t.Error("transient recovery did not return the final response") + } + }) + } +} + +type queryTestAPI struct { + ctx context.Context + calls int + execute func(context.Context, int) (map[string]any, *http.Response, error) +} + +func (api *queryTestAPI) GetFeatures(ctx context.Context) receiver_api.ApiGetFeaturesRequest { + api.ctx = ctx + return receiver_api.ApiGetFeaturesRequest{ApiService: api} +} + +func (api *queryTestAPI) GetFeaturesExecute(_ receiver_api.ApiGetFeaturesRequest) (map[string]any, *http.Response, error) { + api.calls++ + return api.execute(api.ctx, api.calls) +} + +func queryTestResponse(status int) *http.Response { + return &http.Response{ + StatusCode: status, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(queryTestBody)), + } +} + +func TestFetchFeaturesNilResponseErrors(t *testing.T) { + tests := []struct { + name string + err error + class features.Class + attempts int + }{ + {"transport", errors.New(queryTestError + queryTestAPIKey), features.Transient, 3}, + {"credential_url", &url.Error{Op: "Get", URL: "https://synthetic.invalid/?api_key=" + queryTestAPIKey, Err: errors.New(queryTestError)}, features.Transient, 3}, + {"connection_reset", &net.OpError{Op: "read", Net: "tcp", Err: errors.New(queryTestError)}, features.Transient, 3}, + {"deadline", context.DeadlineExceeded, features.Timeout, 3}, + {"canceled", context.Canceled, features.Canceled, 1}, + {"untrusted_certificate", &url.Error{Op: "Get", URL: "https://synthetic.invalid/" + queryTestAPIKey, Err: x509.UnknownAuthorityError{}}, features.Configuration, 1}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + api := &queryTestAPI{execute: func(context.Context, int) (map[string]any, *http.Response, error) { + return nil, nil, tt.err + }} + client := newQueryTestClient(t, api, queryTestOptions()) + started := time.Now() + checkQueryResult(t, client.FetchFeatures(context.Background()), tt.class, 0, tt.attempts, started) + if api.calls != tt.attempts { + t.Errorf("API calls = %d; want %d", api.calls, tt.attempts) + } + }) + }) + } + t.Run("nil_response_without_error", func(t *testing.T) { + api := &queryTestAPI{execute: func(context.Context, int) (map[string]any, *http.Response, error) { + return map[string]any{"otel-logs": true}, nil, nil + }} + opts := queryTestOptions() + opts.MaxAttempts = 1 + started := time.Now() + result := newQueryTestClient(t, api, opts).FetchFeatures(context.Background()) + if result.Class == features.Valid || result.Features != nil { + t.Error("missing HTTP response must not produce a valid observation") + } + checkQueryResult(t, result, result.Class, 0, 1, started) + }) +} + +type queryTestBodyCloser struct { + io.Reader + closed int +} + +func (body *queryTestBodyCloser) Close() error { + body.closed++ + return nil +} + +func TestFetchFeaturesClosesResponses(t *testing.T) { + for _, status := range []int{200, 401, 404, 503} { + t.Run(fmt.Sprint(status), func(t *testing.T) { + var bodies []*queryTestBodyCloser + api := &queryTestAPI{execute: func(context.Context, int) (map[string]any, *http.Response, error) { + body := &queryTestBodyCloser{Reader: strings.NewReader(queryTestBody)} + bodies = append(bodies, body) + response := queryTestResponse(status) + response.Body = body + if status == 200 { + return map[string]any{}, response, nil + } + return nil, response, errors.New(queryTestError) + }} + newQueryTestClient(t, api, queryTestOptions()).FetchFeatures(context.Background()) + for i, body := range bodies { + if body.closed != 1 { + t.Errorf("attempt %d response closed %d times; want once", i+1, body.closed) + } + } + }) + } +} + +func TestFetchFeaturesRetryAfter(t *testing.T) { + for _, status := range []int{429, 503} { + for _, format := range []string{"seconds", "http_date", "beyond_budget", "overflow", "invalid", "past_date", "negative"} { + t.Run(fmt.Sprintf("%d_%s", status, format), func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + started := time.Now() + retryAfter := "2" + switch format { + case "http_date": + retryAfter = started.Add(2 * time.Second).UTC().Format(http.TimeFormat) + case "beyond_budget": + retryAfter = "60" + case "overflow": + retryAfter = "184467440737095516160" + case "invalid": + retryAfter = "not-a-delay" + case "past_date": + retryAfter = started.Add(-time.Hour).UTC().Format(http.TimeFormat) + case "negative": + retryAfter = "-1" + } + var attemptsAt []time.Time + api := &queryTestAPI{execute: func(_ context.Context, attempt int) (map[string]any, *http.Response, error) { + attemptsAt = append(attemptsAt, time.Now()) + if attempt == 1 { + response := queryTestResponse(status) + response.Header.Set("Retry-After", retryAfter) + return nil, response, errors.New(queryTestError) + } + return map[string]any{}, queryTestResponse(200), nil + }} + result := newQueryTestClient(t, api, queryTestOptions()).FetchFeatures(context.Background()) + if format == "beyond_budget" || format == "overflow" { + if api.calls != 1 { + t.Error("retried earlier than Retry-After allowed") + } + if result.Class != features.Timeout && result.Class != features.Transient { + t.Error("unaffordable retry must end as timeout or exhausted transient") + } + checkQueryResult(t, result, result.Class, status, 1, started) + } else { + checkQueryResult(t, result, features.Valid, 200, 2, started) + if len(attemptsAt) != 2 { + t.Fatalf("attempts = %d; want 2", len(attemptsAt)) + } + if (format == "seconds" || format == "http_date") && attemptsAt[1].Sub(started) < 2*time.Second { + t.Error("retried before Retry-After") + } + if format != "seconds" && format != "http_date" && attemptsAt[1].Sub(started) > time.Nanosecond { + t.Error("invalid or expired Retry-After did not fall back to configured backoff") + } + } + if time.Since(started) > queryTestOptions().Timeout { + t.Error("Retry-After exceeded the query budget") + } + }) + }) + } + } +} + +func TestFetchFeaturesAttemptAndQueryDeadlines(t *testing.T) { + for _, wholeQuery := range []bool{false, true} { + t.Run(fmt.Sprintf("whole_query_%t", wholeQuery), func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + opts := queryTestOptions() + if wholeQuery { + opts.Timeout = 2500 * time.Millisecond + } + type contextKey struct{} + parent := context.WithValue(context.Background(), contextKey{}, "synthetic-context-value") + var attempts []context.Context + api := &queryTestAPI{execute: func(ctx context.Context, _ int) (map[string]any, *http.Response, error) { + attempts = append(attempts, ctx) + if ctx.Value(contextKey{}) != "synthetic-context-value" { + t.Error("attempt discarded parent context values") + } + deadline, ok := ctx.Deadline() + if !ok || time.Until(deadline) > opts.AttemptTimeout { + t.Error("attempt has no bounded deadline") + } + <-ctx.Done() + return nil, nil, ctx.Err() + }} + started := time.Now() + result := newQueryTestClient(t, api, opts).FetchFeatures(parent) + checkQueryResult(t, result, features.Timeout, 0, 3, started) + for _, ctx := range attempts { + if ctx.Err() == nil { + t.Error("completed attempt context was not canceled") + } + } + wantElapsed := 3 * opts.AttemptTimeout + if wholeQuery { + wantElapsed = opts.Timeout + } + if elapsed := time.Since(started); elapsed < wantElapsed || elapsed > wantElapsed+2*opts.MaxBackoff { + t.Errorf("query elapsed = %v; want %v plus bounded backoff", elapsed, wantElapsed) + } + if parent.Err() != nil { + t.Error("query canceled the parent context") + } + }) + }) + } +} + +func TestFetchFeaturesCancellation(t *testing.T) { + for _, phase := range []string{"before_query", "during_attempt", "during_backoff", "during_retry_after"} { + t.Run(phase, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + opts := queryTestOptions() + opts.InitialBackoff = time.Second + opts.MaxBackoff = time.Second + api := &queryTestAPI{execute: func(ctx context.Context, _ int) (map[string]any, *http.Response, error) { + if phase == "during_attempt" { + <-ctx.Done() + return nil, nil, ctx.Err() + } + response := queryTestResponse(503) + if phase == "during_retry_after" { + response.Header.Set("Retry-After", "5") + } + return nil, response, errors.New(queryTestError) + }} + if phase == "before_query" { + cancel() + } + client := newQueryTestClient(t, api, opts) + started := time.Now() + results := make(chan features.Result, 1) + go func() { results <- client.FetchFeatures(ctx) }() + synctest.Wait() + cancel() + synctest.Wait() + select { + case result := <-results: + attempts, status := 1, 503 + if phase == "before_query" { + attempts, status = 0, 0 + } else if phase == "during_attempt" { + status = 0 + } + checkQueryResult(t, result, features.Canceled, status, attempts, started) + if api.calls != attempts { + t.Errorf("API calls = %d; want %d", api.calls, attempts) + } + default: + t.Fatal("cancellation did not interrupt query") + } + if !time.Now().Equal(started) { + t.Error("cancellation waited for a timeout or retry delay") + } + }) + }) + } +} + +func TestFetchFeaturesCanceledHTTP(t *testing.T) { + entered := make(chan struct{}) + requestCanceled := make(chan struct{}) + client, authCtx := newHTTPQueryTestClient(t, func(_ http.ResponseWriter, r *http.Request) { + close(entered) + <-r.Context().Done() + close(requestCanceled) + }, queryTestOptions()) + ctx, cancel := context.WithCancel(authCtx) + defer cancel() + started := time.Now() + results := make(chan features.Result, 1) + go func() { results <- client.FetchFeatures(ctx) }() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("HTTP request did not reach server") + } + cancel() + select { + case result := <-results: + checkQueryResult(t, result, features.Canceled, 0, 1, started) + case <-time.After(5 * time.Second): + t.Fatal("HTTP query did not stop after cancellation") + } + select { + case <-requestCanceled: + case <-time.After(5 * time.Second): + t.Fatal("server request was not canceled") + } +} + +func TestFetchFeaturesInterruptedSuccessfulResponse(t *testing.T) { + var calls atomic.Int32 + client, ctx := newHTTPQueryTestClient(t, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + if calls.Add(1) == 1 { + w.Header().Set("Content-Length", "1000") + _, _ = io.WriteString(w, `{"otel-logs":`) + return + } + _, _ = io.WriteString(w, `{"otel-logs":true}`) + }, queryTestOptions()) + started := time.Now() + result := client.FetchFeatures(ctx) + checkQueryResult(t, result, features.Valid, 200, 2, started) +} + +func TestQueryOptionsValidation(t *testing.T) { + api := &queryTestAPI{} + for _, modify := range []func(*features.QueryOptions){ + func(o *features.QueryOptions) { o.Timeout = 0 }, + func(o *features.QueryOptions) { o.AttemptTimeout = 0 }, + func(o *features.QueryOptions) { o.AttemptTimeout = o.Timeout + time.Second }, + func(o *features.QueryOptions) { o.MaxAttempts = 0 }, + func(o *features.QueryOptions) { o.InitialBackoff = 0 }, + func(o *features.QueryOptions) { o.MaxBackoff = 0 }, + } { + opts := queryTestOptions() + modify(&opts) + if _, err := features.NewClient(api, opts); err == nil { + t.Error("accepted invalid query bounds") + } + } + if _, err := features.NewClient(nil, queryTestOptions()); err == nil { + t.Error("accepted nil features API") + } +} diff --git a/pkg/openapiclient/features/result.go b/pkg/openapiclient/features/result.go new file mode 100644 index 0000000..6525271 --- /dev/null +++ b/pkg/openapiclient/features/result.go @@ -0,0 +1,28 @@ +package features + +import "time" + +// Class is a bounded, credential-safe query outcome. +type Class string + +// Query outcome classes distinguish capability observations from failures. +const ( + Valid Class = "valid" + Unsupported Class = "unsupported" + Authentication Class = "authentication" + Transient Class = "transient" + Timeout Class = "timeout" + Malformed Class = "malformed" + Configuration Class = "configuration" + Rejected Class = "rejected" + Canceled Class = "canceled" +) + +// Result describes a completed query without retaining raw errors or response bodies. +type Result struct { + Class Class + Features map[string]any + StatusCode int + Attempts int + FinishedAt time.Time +} diff --git a/pkg/openapiclient/options.go b/pkg/openapiclient/options.go new file mode 100644 index 0000000..ce88368 --- /dev/null +++ b/pkg/openapiclient/options.go @@ -0,0 +1,50 @@ +package openapiclient + +import ( + "errors" + "net/url" + "strconv" + "strings" + "time" +) + +// ConnectionOptions configures authentication and the owned Receiver transport. +type ConnectionOptions struct { + ReceiverURL string + UserAgent string + APIKey string + ServiceAccountToken func() string + ProxyURL string + CABundlePEM []byte + InsecureSkipVerify bool + RequestTimeout time.Duration +} + +// ErrMissingCredential indicates that no usable authentication credential is available. +var ErrMissingCredential = errors.New("receiver credential is empty") + +// ErrResponseTooLarge indicates that a feature response exceeded the supported limit. +var ErrResponseTooLarge = errors.New("receiver feature response exceeds 1 MiB") + +func parseEndpoint(raw string, allowUserinfo bool) (*url.URL, error) { + u, err := url.Parse(raw) + if err != nil || u == nil || (u.Scheme != "http" && u.Scheme != "https") || u.Hostname() == "" || u.Opaque != "" { + return nil, errors.New("endpoint must be an absolute HTTP(S) URL") + } + if (!allowUserinfo && u.User != nil) || u.RawQuery != "" || u.ForceQuery || u.Fragment != "" || strings.Contains(raw, "#") { + return nil, errors.New("endpoint contains unsupported userinfo, query or fragment") + } + if port := u.Port(); port != "" { + number, err := strconv.Atoi(port) + if err != nil || number < 1 || number > 65535 { + return nil, errors.New("endpoint port must be between 1 and 65535") + } + } else if strings.HasSuffix(u.Host, ":") { + return nil, errors.New("endpoint port is empty") + } + return u, nil +} + +func makeBaseURL(raw string) string { + return strings.TrimSuffix(strings.TrimRight(raw, "/"), "/stsAgent") +} diff --git a/pkg/openapiclient/transport.go b/pkg/openapiclient/transport.go new file mode 100644 index 0000000..7297432 --- /dev/null +++ b/pkg/openapiclient/transport.go @@ -0,0 +1,69 @@ +package openapiclient + +import ( + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "io" + "net/http" + "strings" +) + +const maxFeatureResponseBytes = 1 << 20 + +func newTransport(opts ConnectionOptions) (http.RoundTripper, error) { + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.Proxy = nil + transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: opts.InsecureSkipVerify} // Explicit operator configuration. + if len(opts.CABundlePEM) != 0 { + roots, err := x509.SystemCertPool() + if err != nil { + return nil, errors.New("cannot load system certificate trust") + } + if !roots.AppendCertsFromPEM(opts.CABundlePEM) { + return nil, errors.New("CA bundle contains no valid certificates") + } + transport.TLSClientConfig.RootCAs = roots + } + if opts.ProxyURL != "" { + proxy, err := parseEndpoint(opts.ProxyURL, true) + if err != nil { + return nil, fmt.Errorf("invalid proxy URL: %w", err) + } + transport.Proxy = http.ProxyURL(proxy) + } + return boundedTransport{transport}, nil +} + +type boundedTransport struct{ base http.RoundTripper } + +func (t boundedTransport) RoundTrip(req *http.Request) (*http.Response, error) { + response, err := t.base.RoundTrip(req) + if response != nil && response.Body != nil && strings.HasSuffix(req.URL.Path, "/stsAgent/features") { + response.Body = &boundedBody{ReadCloser: response.Body, remaining: maxFeatureResponseBytes} + } + return response, err +} + +type boundedBody struct { + io.ReadCloser + remaining int +} + +func (b *boundedBody) Read(p []byte) (int, error) { + if len(p) == 0 { + return 0, nil + } + if len(p) > b.remaining+1 { + p = p[:b.remaining+1] + } + n, err := b.ReadCloser.Read(p) + if n > b.remaining { + n = b.remaining + b.remaining = 0 + return n, ErrResponseTooLarge + } + b.remaining -= n + return n, err +} From d9a80385b6125963c03dffa8f10e7a45fdc92dcd Mon Sep 17 00:00:00 2001 From: Bram Schuur Date: Mon, 14 Sep 2026 15:33:45 +0200 Subject: [PATCH 2/2] Make feature polling cancellable and preserve query outcomes --- pkg/openapiclient/features/features_poller.go | 90 ++++-- .../features/features_poller_test.go | 291 ++++++++++++++++-- 2 files changed, 322 insertions(+), 59 deletions(-) diff --git a/pkg/openapiclient/features/features_poller.go b/pkg/openapiclient/features/features_poller.go index d316763..08cc438 100644 --- a/pkg/openapiclient/features/features_poller.go +++ b/pkg/openapiclient/features/features_poller.go @@ -2,48 +2,80 @@ package features import ( "context" - "github.com/StackVista/stackstate-receiver-go-client/generated/receiver_api" - log "github.com/cihub/seelog" - "net/http" + "errors" + "math" "time" ) -// StartFeaturesPoller Polls the /features endpoint of SUSE Observabiity to observe which features are supported. -func StartFeaturesPoller(clientCtx context.Context, featuresAPI receiver_api.FeaturesAPI, interval time.Duration) (chan map[string]interface{}, func()) { +// PollOptions controls the delay between completed observations. +type PollOptions struct { + Interval time.Duration + Jitter float64 +} - outputChannel := make(chan map[string]interface{}) - stopChannel := make(chan interface{}) - ticker := time.NewTicker(interval) - // Channel that produces just a single value - init := make(chan bool, 1) - init <- true +// Poller delivers each completed query until stopped or its context is canceled. +type Poller struct { + results chan Result + done chan struct{} + cancel context.CancelFunc +} +// StartPolling waits one jittered interval before querying, including the first query. +func (c *Client) StartPolling(authCtx context.Context, opts PollOptions) (*Poller, error) { + if authCtx == nil { + return nil, errors.New("polling context is required") + } + if opts.Interval <= 0 || math.IsNaN(opts.Jitter) || opts.Jitter < 0 || opts.Jitter >= 1 || + (opts.Jitter > 0 && float64(opts.Interval)*(1+opts.Jitter) >= float64(math.MaxInt64)) { + return nil, errors.New("poll interval must be positive and representable with jitter in [0, 1)") + } + ctx, cancel := context.WithCancel(authCtx) + poller := &Poller{results: make(chan Result), done: make(chan struct{}), cancel: cancel} go func() { + defer close(poller.done) + defer close(poller.results) + defer cancel() for { + if ctx.Err() != nil { + return + } + timer := time.NewTimer(pollDelay(opts, c.random())) select { - case <-stopChannel: - ticker.Stop() - close(outputChannel) - close(init) + case <-ctx.Done(): + timer.Stop() return - case <-ticker.C: - case <-init: + case <-timer.C: } - - features, response, err := featuresAPI.GetFeaturesExecute(featuresAPI.GetFeatures(clientCtx)) - if err != nil { - log.Errorf("Error retrieving features: %v", err) - continue + if ctx.Err() != nil { + return } - - if response.StatusCode != http.StatusOK { - log.Errorf("Error retrieving features, statuscode was: %s", response.Status) - continue + result := c.FetchFeatures(ctx) + select { + case <-ctx.Done(): + return + case poller.results <- result: } - - outputChannel <- features } }() + return poller, nil +} - return outputChannel, func() { close(stopChannel) } +func pollDelay(opts PollOptions, random float64) time.Duration { + if opts.Jitter == 0 { + return opts.Interval + } + delay := time.Duration(float64(opts.Interval) * (1 + opts.Jitter*(2*random-1))) + if delay < time.Nanosecond { + return time.Nanosecond + } + return delay } + +// Results is unbuffered; slow consumers delay the next poll. +func (p *Poller) Results() <-chan Result { return p.results } + +// Stop cancels requests, waits and delivery without waiting for shutdown. +func (p *Poller) Stop() { p.cancel() } + +// Done closes after the worker has closed Results. +func (p *Poller) Done() <-chan struct{} { return p.done } diff --git a/pkg/openapiclient/features/features_poller_test.go b/pkg/openapiclient/features/features_poller_test.go index be42727..6bdef51 100644 --- a/pkg/openapiclient/features/features_poller_test.go +++ b/pkg/openapiclient/features/features_poller_test.go @@ -2,49 +2,280 @@ package features import ( "context" - "github.com/StackVista/stackstate-receiver-go-client/generated/receiver_api" - "github.com/stretchr/testify/assert" + "math" "net/http" + "sync" + "sync/atomic" "testing" + "testing/synctest" "time" + + "github.com/StackVista/stackstate-receiver-go-client/generated/receiver_api" ) -func TestFeaturePollerProducesRepeatedResults(t *testing.T) { - featuresAPI := receiver_api.NewFeaturesAPIMock() - features := make(map[string]interface{}) - featuresAPI.GetFeaturesResponse = receiver_api.GetFeaturesMockResponse{ - Result: features, - Response: &http.Response{StatusCode: http.StatusOK}, - Error: nil, - } +type pollTestAPI struct { + ctx context.Context + calls atomic.Int64 + execute func(context.Context, int) (map[string]any, *http.Response, error) +} - outputChannel, tearDown := StartFeaturesPoller(context.Background(), featuresAPI, 1*time.Second) - result := <-outputChannel - assert.Equal(t, features, result) +func (a *pollTestAPI) GetFeatures(ctx context.Context) receiver_api.ApiGetFeaturesRequest { + a.ctx = ctx + return receiver_api.ApiGetFeaturesRequest{ApiService: a} +} - result = <-outputChannel - assert.Equal(t, features, result) +func (a *pollTestAPI) GetFeaturesExecute(receiver_api.ApiGetFeaturesRequest) (map[string]any, *http.Response, error) { + call := a.calls.Add(1) + return a.execute(a.ctx, int(call)) +} - tearDown() +func pollTestClient(t *testing.T, api *pollTestAPI) *Client { + t.Helper() + client, err := NewClient(api, QueryOptions{ + Timeout: 20 * time.Second, AttemptTimeout: 5 * time.Second, MaxAttempts: 3, + InitialBackoff: time.Second, MaxBackoff: 2 * time.Second, + }) + if err != nil { + t.Fatal(err) + } + client.random = func() float64 { return 0.5 } + return client +} - _, ok := <-outputChannel - assert.False(t, ok) +func startTestPoller(ctx context.Context, t *testing.T, client *Client, opts PollOptions) *Poller { + t.Helper() + poller, err := client.StartPolling(ctx, opts) + if err != nil { + t.Fatal(err) + } + t.Cleanup(poller.Stop) + return poller } -func TestDoesNotProduceWhenBrokenAndBeAbletoTearDown(t *testing.T) { - featuresAPI := receiver_api.NewFeaturesAPIMock() - features := make(map[string]interface{}) - featuresAPI.GetFeaturesResponse = receiver_api.GetFeaturesMockResponse{ - Result: features, - Response: &http.Response{StatusCode: http.StatusBadRequest}, - Error: nil, +func assertPollerClosed(t *testing.T, poller *Poller) { + t.Helper() + <-poller.Done() + if _, ok := <-poller.Results(); ok { + t.Fatal("Results was not closed before Done") } +} - outputChannel, tearDown := StartFeaturesPoller(context.Background(), featuresAPI, 1*time.Second) - time.Sleep(1 * time.Second) +func TestPollingOptions(t *testing.T) { + client := pollTestClient(t, &pollTestAPI{}) + for _, opts := range []PollOptions{ + {}, {Interval: -time.Second}, {Interval: time.Second, Jitter: -0.1}, + {Interval: time.Second, Jitter: 1}, {Interval: time.Second, Jitter: math.NaN()}, + {Interval: time.Second, Jitter: math.Inf(1)}, {Interval: time.Second, Jitter: math.Inf(-1)}, + {Interval: time.Duration(math.MaxInt64), Jitter: 0.2}, + } { + if poller, err := client.StartPolling(context.Background(), opts); err == nil { + poller.Stop() + t.Errorf("accepted invalid options: %+v", opts) + } + } + if _, err := client.StartPolling(nil, PollOptions{Interval: time.Second}); err == nil { + t.Error("accepted nil context") + } + opts := PollOptions{Interval: time.Duration(math.MaxInt64)} + poller := startTestPoller(context.Background(), t, client, opts) + poller.Stop() + assertPollerClosed(t, poller) + if got := pollDelay(opts, 0.5); got != opts.Interval { + t.Errorf("zero jitter changed maximum interval: %v", got) + } +} + +func TestPollingDelayAndJitter(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + api := &pollTestAPI{execute: func(context.Context, int) (map[string]any, *http.Response, error) { + return map[string]any{}, &http.Response{StatusCode: 200}, nil + }} + client := pollTestClient(t, api) + draws := []float64{0, 0.5, 0.999} + draw := 0 + client.random = func() float64 { + value := draws[draw%len(draws)] + draw++ + return value + } + opts := PollOptions{Interval: 10 * time.Second, Jitter: 0.2} + poller := startTestPoller(context.Background(), t, client, opts) + for i, random := range draws { + synctest.Wait() + delay := pollDelay(opts, random) + time.Sleep(delay - time.Nanosecond) + synctest.Wait() + if api.calls.Load() != int64(i) { + t.Fatalf("query before jittered interval: calls=%d, want %d", api.calls.Load(), i) + } + time.Sleep(time.Nanosecond) + result := <-poller.Results() + if result.Class != Valid || result.Attempts != 1 || !result.FinishedAt.Equal(time.Now()) { + t.Fatalf("unexpected observation: %+v", result) + } + } + poller.Stop() + assertPollerClosed(t, poller) + }) + for _, test := range []struct { + random float64 + want time.Duration + }{{0, 8 * time.Second}, {0.5, 10 * time.Second}, {1, 12 * time.Second}} { + if got := pollDelay(PollOptions{Interval: 10 * time.Second, Jitter: 0.2}, test.random); got != test.want { + t.Errorf("delay = %v, want %v", got, test.want) + } + } + if got := pollDelay(PollOptions{Interval: time.Nanosecond, Jitter: 0.9}, 0); got != time.Nanosecond { + t.Errorf("minimum delay = %v", got) + } +} - tearDown() +func TestPollingPreservesOutcomesAndBackpressure(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + statuses := []int{200, 404, 401, 503, 200, 400, 200} + classes := []Class{Valid, Unsupported, Authentication, Transient, Malformed, Rejected, Valid} + type authKey struct{} + api := &pollTestAPI{execute: func(ctx context.Context, call int) (map[string]any, *http.Response, error) { + if ctx.Value(authKey{}) != "synthetic-auth-context" { + t.Error("query lost authenticated context") + } + if call > len(statuses) { + t.Error("unexpected overlapping or extra query") + return nil, nil, nil + } + values := map[string]any{"capacity": float64(42)} + if call == 5 { + values = nil + } + return values, &http.Response{StatusCode: statuses[call-1]}, nil + }} + client := pollTestClient(t, api) + client.opts.MaxAttempts = 1 + poller := startTestPoller(context.WithValue(context.Background(), authKey{}, "synthetic-auth-context"), t, client, + PollOptions{Interval: time.Second}) + synctest.Wait() + for i, class := range classes { + time.Sleep(time.Second) + synctest.Wait() + finishedAt := time.Now() + time.Sleep(time.Minute) + synctest.Wait() + if api.calls.Load() != int64(i+1) { + t.Fatalf("queried while output blocked: calls=%d", api.calls.Load()) + } + result := <-poller.Results() + if result.Class != class || result.StatusCode != statuses[i] || !result.FinishedAt.Equal(finishedAt) { + t.Fatalf("lost or changed outcome: %+v", result) + } + if class == Valid && result.Features["capacity"] != float64(42) { + t.Error("lost unrelated feature value") + } + synctest.Wait() + } + poller.Stop() + assertPollerClosed(t, poller) + }) +} + +func TestPollingRetriesAreOneObservation(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + api := &pollTestAPI{execute: func(_ context.Context, call int) (map[string]any, *http.Response, error) { + if call < 3 { + return nil, &http.Response{StatusCode: 503}, nil + } + return map[string]any{}, &http.Response{StatusCode: 200}, nil + }} + poller := startTestPoller(context.Background(), t, pollTestClient(t, api), PollOptions{Interval: time.Second}) + result := <-poller.Results() + if result.Class != Valid || result.Attempts != 3 || api.calls.Load() != 3 { + t.Fatalf("attempts emitted as separate observations: %+v", result) + } + poller.Stop() + assertPollerClosed(t, poller) + }) +} + +func TestPollingCancellation(t *testing.T) { + for _, phase := range []string{"interval", "http", "backoff", "output"} { + for _, parentCancel := range []bool{false, true} { + name := phase + "/stop" + if parentCancel { + name = phase + "/parent" + } + t.Run(name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + requestCanceled := false + api := &pollTestAPI{execute: func(ctx context.Context, _ int) (map[string]any, *http.Response, error) { + switch phase { + case "http": + <-ctx.Done() + requestCanceled = true + return nil, nil, ctx.Err() + case "backoff": + return nil, &http.Response{StatusCode: 503}, nil + default: + return map[string]any{}, &http.Response{StatusCode: 200}, nil + } + }} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + poller := startTestPoller(ctx, t, pollTestClient(t, api), PollOptions{Interval: time.Second}) + synctest.Wait() + if phase != "interval" { + time.Sleep(time.Second) + synctest.Wait() + } + if phase == "http" { + time.Sleep(2 * time.Second) + synctest.Wait() + } + before := time.Now() + if parentCancel { + cancel() + } else { + var stops sync.WaitGroup + for range 10 { + stops.Go(poller.Stop) + } + stops.Wait() + } + assertPollerClosed(t, poller) + poller.Stop() + if !time.Now().Equal(before) { + t.Error("shutdown waited for a timer") + } + if phase == "interval" && api.calls.Load() != 0 || phase != "interval" && api.calls.Load() != 1 { + t.Errorf("unexpected calls during cancellation: %d", api.calls.Load()) + } + if phase == "http" && !requestCanceled { + t.Error("request did not observe cancellation") + } + }) + }) + } + } +} - _, ok := <-outputChannel - assert.False(t, ok) +func TestPollingStopDoesNotWaitForRequestReturn(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + release := make(chan struct{}) + api := &pollTestAPI{execute: func(ctx context.Context, _ int) (map[string]any, *http.Response, error) { + <-ctx.Done() + <-release + return nil, nil, ctx.Err() + }} + poller := startTestPoller(context.Background(), t, pollTestClient(t, api), PollOptions{Interval: time.Second}) + synctest.Wait() + time.Sleep(time.Second) + synctest.Wait() + poller.Stop() + synctest.Wait() + select { + case <-poller.Done(): + t.Fatal("Done closed before request returned") + default: + } + close(release) + assertPollerClosed(t, poller) + }) }