diff --git a/sdk/go.mod b/sdk/go.mod index a1fb29dd5d..752c17093e 100644 --- a/sdk/go.mod +++ b/sdk/go.mod @@ -6,6 +6,7 @@ toolchain go1.25.9 require ( connectrpc.com/connect v1.19.1 + connectrpc.com/grpchealth v1.4.0 github.com/Masterminds/semver/v3 v3.4.0 github.com/google/uuid v1.6.0 github.com/gowebpki/jcs v1.0.1 diff --git a/sdk/go.sum b/sdk/go.sum index e9722785c7..2a908588b0 100644 --- a/sdk/go.sum +++ b/sdk/go.sum @@ -2,6 +2,8 @@ buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.6-2025060316535 buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.6-20250603165357-b52ab10f4468.1/go.mod h1:avRlCjnFzl98VPaeCtJ24RrV/wwHFzB8sWXhj26+n/U= connectrpc.com/connect v1.19.1 h1:R5M57z05+90EfEvCY1b7hBxDVOUl45PrtXtAV2fOC14= connectrpc.com/connect v1.19.1/go.mod h1:tN20fjdGlewnSFeZxLKb0xwIZ6ozc3OQs2hTXy4du9w= +connectrpc.com/grpchealth v1.4.0 h1:MJC96JLelARPgZTiRF9KRfY/2N9OcoQvF2EWX07v2IE= +connectrpc.com/grpchealth v1.4.0/go.mod h1:WhW6m1EzTmq3Ky1FE8EfkIpSDc6TfUx2M2KqZO3ts/Q= github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0= github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= diff --git a/sdk/options.go b/sdk/options.go index b3f6d3223a..ba63bb092a 100644 --- a/sdk/options.go +++ b/sdk/options.go @@ -175,7 +175,10 @@ func WithPlatformConfiguration(platformConfiguration PlatformConfiguration) Opti } } -// WithConnectionValidation will validate connection to a healthy, running platform +// WithConnectionValidation will validate connection to a healthy, running platform at +// SDK construction time. For runtime readiness probes (e.g. Kubernetes liveness/readiness +// endpoints), use SDK.IsHealthy(ctx) instead — it honors ctx for deadlines and tracing +// and does not fail-fast at construction. func WithConnectionValidation() Option { return func(c *config) { c.shouldValidatePlatformConnectivity = true diff --git a/sdk/sdk.go b/sdk/sdk.go index fcf9021108..fa809e6e12 100644 --- a/sdk/sdk.go +++ b/sdk/sdk.go @@ -32,9 +32,12 @@ import ( const ( // Failure while connecting to a service. // Check your configuration and/or retry. - ErrGrpcDialFailed = Error("failed to dial grpc endpoint") - ErrShutdownFailed = Error("failed to shutdown sdk") - ErrPlatformUnreachable = Error("platform unreachable or not responding") + ErrGrpcDialFailed = Error("failed to dial grpc endpoint") + ErrShutdownFailed = Error("failed to shutdown sdk") + ErrPlatformUnreachable = Error("platform unreachable or not responding") + // ErrHealthCheckUnsupported is returned by SDK.IsHealthy when the SDK is configured + // in IPC mode, which does not support the gRPC Health protocol. + ErrHealthCheckUnsupported = Error("health check not supported in IPC mode") ErrPlatformConfigFailed = Error("failed to retrieve platform configuration") ErrPlatformEndpointMalformed = Error("platform endpoint is malformed") ErrPlatformIssuerNotFound = Error("issuer not found in well-known idp configuration") @@ -310,6 +313,26 @@ func (s SDK) Conn() *ConnectRPCConnection { return s.conn } +// IsHealthy reports whether the platform's gRPC Health v1 endpoint is reachable and SERVING. +// The check honors ctx for deadline and cancellation; OTEL tracing works automatically when +// otelconnect.NewInterceptor is registered via WithExtraClientOptions at SDK construction. +// +// Returns: +// - (true, nil) when the platform reports SERVING. +// - (false, nil) when the platform is reachable but reports NOT_SERVING or UNKNOWN. +// - (false, ErrHealthCheckUnsupported) when the SDK is configured in IPC mode. +// - (false, error) wrapping ErrPlatformUnreachable on transport failure or ctx errors. +func (s SDK) IsHealthy(ctx context.Context) (bool, error) { + if s.ipc || s.conn == nil { + return false, ErrHealthCheckUnsupported + } + healthy, err := checkPlatformHealth(ctx, s.conn.Endpoint, s.conn.Client, s.conn.Options) + if err != nil { + return false, errors.Join(ErrPlatformUnreachable, err) + } + return healthy, nil +} + type TdfType string const ( @@ -413,21 +436,42 @@ func isValidManifest(manifest string, intensity SchemaValidationIntensity) (bool return true, nil } -// Test connectability to the platform and validate a healthy status -func validateHealthyPlatformConnection(platformEndpoint string, httpClient *http.Client, options []connect.ClientOption) error { +// checkPlatformHealth issues a single gRPC Health v1 Check against the platform endpoint and +// reports whether the response is SERVING. ctx controls deadline, cancellation, and trace-context. +func checkPlatformHealth( + ctx context.Context, + endpoint string, + httpClient *http.Client, + options []connect.ClientOption, +) (bool, error) { + checkURL, err := url.JoinPath(endpoint, "grpc.health.v1.Health", "Check") + if err != nil { + return false, err + } healthClient := connect.NewClient[healthpb.HealthCheckRequest, healthpb.HealthCheckResponse]( httpClient, - platformEndpoint+"/grpc.health.v1.Health/Check", + checkURL, options..., ) - res, err := healthClient.CallUnary( - context.Background(), - connect.NewRequest(&healthpb.HealthCheckRequest{}), - ) - if err != nil || res.Msg.GetStatus() != healthpb.HealthCheckResponse_SERVING { - return errors.Join(ErrPlatformUnreachable, err) + res, err := healthClient.CallUnary(ctx, connect.NewRequest(&healthpb.HealthCheckRequest{})) + if err != nil { + return false, err } + return res.Msg.GetStatus() == healthpb.HealthCheckResponse_SERVING, nil +} +// validateHealthyPlatformConnection is the construction-time reachability gate used by New when +// WithConnectionValidation is set. Callers pass cfg.extraClientOptions (pre-auth) because the +// auth and audit interceptors are assembled after this gate fires; the runtime SDK.IsHealthy +// method uses the post-interceptor s.conn.Options instead. +func validateHealthyPlatformConnection(platformEndpoint string, httpClient *http.Client, options []connect.ClientOption) error { + healthy, err := checkPlatformHealth(context.Background(), platformEndpoint, httpClient, options) + if err != nil { + return errors.Join(ErrPlatformUnreachable, err) + } + if !healthy { + return ErrPlatformUnreachable + } return nil } diff --git a/sdk/sdk_test.go b/sdk/sdk_test.go index 8137ebe76e..810dede16c 100644 --- a/sdk/sdk_test.go +++ b/sdk/sdk_test.go @@ -2,10 +2,15 @@ package sdk_test import ( "bytes" + "context" "encoding/base64" + "net/http" + "net/http/httptest" "reflect" "testing" + "time" + "connectrpc.com/grpchealth" "github.com/opentdf/platform/protocol/go/policy/attributes/attributesconnect" "github.com/opentdf/platform/protocol/go/policy/kasregistry/kasregistryconnect" "github.com/opentdf/platform/protocol/go/policy/resourcemapping/resourcemappingconnect" @@ -313,3 +318,198 @@ func Test_GetType_Invalid2Bytes(t *testing.T) { assert.Equal(t, sdk.Invalid, tdfType) } + +func TestErrHealthCheckUnsupported_Distinct(t *testing.T) { + assert.NotEqual(t, sdk.ErrHealthCheckUnsupported, sdk.ErrPlatformUnreachable) + assert.Equal(t, "health check not supported in IPC mode", sdk.ErrHealthCheckUnsupported.Error()) +} + +// newHealthTestServer starts an httptest.Server serving grpc.health.v1.Health with the +// configured status for the empty service name (the SDK's reachability probe). +func newHealthTestServer(t *testing.T, status grpchealth.Status) *httptest.Server { + t.Helper() + checker := grpchealth.NewStaticChecker() + checker.SetStatus("", status) + mux := http.NewServeMux() + path, handler := grpchealth.NewHandler(checker) + mux.Handle(path, handler) + return httptest.NewServer(mux) +} + +func TestSDK_IsHealthy_IPCMode_ReturnsErrHealthCheckUnsupported(t *testing.T) { + // IPC mode requires a coreConn; provide a dummy one to satisfy sdk.New. + dummyConn := &sdk.ConnectRPCConnection{ + Endpoint: "http://localhost:0", + Client: http.DefaultClient, + } + s, err := sdk.New("", + sdk.WithIPC(), + sdk.WithCustomCoreConnection(dummyConn), + sdk.WithPlatformConfiguration(sdk.PlatformConfiguration{ + "idp": map[string]interface{}{ + "issuer": "https://example.org", + "authorization_endpoint": "https://example.org/auth", + "token_endpoint": "https://example.org/token", + }, + }), + ) + require.NoError(t, err) + require.NotNil(t, s) + + healthy, err := s.IsHealthy(context.Background()) + assert.False(t, healthy) + require.ErrorIs(t, err, sdk.ErrHealthCheckUnsupported) +} + +func TestSDK_IsHealthy_Unreachable_ReturnsErrPlatformUnreachable(t *testing.T) { + s, err := sdk.New(badPlatformEndpoint, + sdk.WithPlatformConfiguration(sdk.PlatformConfiguration{ + "idp": map[string]interface{}{ + "issuer": "https://example.org", + "authorization_endpoint": "https://example.org/auth", + "token_endpoint": "https://example.org/token", + }, + }), + ) + require.NoError(t, err) + require.NotNil(t, s) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + start := time.Now() + healthy, err := s.IsHealthy(ctx) + elapsed := time.Since(start) + + assert.False(t, healthy) + require.ErrorIs(t, err, sdk.ErrPlatformUnreachable) + assert.Less(t, elapsed, 2*time.Second, "health check should return promptly against a closed port, not wait for the ctx deadline") +} + +func TestSDK_IsHealthy_ContextCanceled_ReturnsQuickly(t *testing.T) { + s, err := sdk.New(badPlatformEndpoint, + sdk.WithPlatformConfiguration(sdk.PlatformConfiguration{ + "idp": map[string]interface{}{ + "issuer": "https://example.org", + "authorization_endpoint": "https://example.org/auth", + "token_endpoint": "https://example.org/token", + }, + }), + ) + require.NoError(t, err) + require.NotNil(t, s) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // canceled before the call + + start := time.Now() + healthy, err := s.IsHealthy(ctx) + elapsed := time.Since(start) + + assert.False(t, healthy) + require.Error(t, err) + require.ErrorIs(t, err, sdk.ErrPlatformUnreachable) + require.ErrorIs(t, err, context.Canceled) + assert.Less(t, elapsed, 500*time.Millisecond, "pre-canceled ctx should short-circuit") +} + +func TestSDK_IsHealthy_Serving(t *testing.T) { + ts := newHealthTestServer(t, grpchealth.StatusServing) + defer ts.Close() + + s, err := sdk.New(ts.URL, + sdk.WithPlatformConfiguration(sdk.PlatformConfiguration{ + "idp": map[string]interface{}{ + "issuer": "https://example.org", + "authorization_endpoint": "https://example.org/auth", + "token_endpoint": "https://example.org/token", + }, + }), + ) + require.NoError(t, err) + require.NotNil(t, s) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + healthy, err := s.IsHealthy(ctx) + require.NoError(t, err) + assert.True(t, healthy) +} + +func TestSDK_IsHealthy_NotServing(t *testing.T) { + ts := newHealthTestServer(t, grpchealth.StatusNotServing) + defer ts.Close() + + s, err := sdk.New(ts.URL, + sdk.WithPlatformConfiguration(sdk.PlatformConfiguration{ + "idp": map[string]interface{}{ + "issuer": "https://example.org", + "authorization_endpoint": "https://example.org/auth", + "token_endpoint": "https://example.org/token", + }, + }), + ) + require.NoError(t, err) + require.NotNil(t, s) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + healthy, err := s.IsHealthy(ctx) + require.NoError(t, err) + assert.False(t, healthy) +} + +// TestSDK_IsHealthy_Unknown locks the contract that an UNKNOWN status from a reachable +// platform returns (false, nil) — distinct from transport errors which wrap ErrPlatformUnreachable. +func TestSDK_IsHealthy_Unknown(t *testing.T) { + ts := newHealthTestServer(t, grpchealth.StatusUnknown) + defer ts.Close() + + s, err := sdk.New(ts.URL, + sdk.WithPlatformConfiguration(sdk.PlatformConfiguration{ + "idp": map[string]interface{}{ + "issuer": "https://example.org", + "authorization_endpoint": "https://example.org/auth", + "token_endpoint": "https://example.org/token", + }, + }), + ) + require.NoError(t, err) + require.NotNil(t, s) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + healthy, err := s.IsHealthy(ctx) + require.NoError(t, err) + assert.False(t, healthy) +} + +// TestSDK_IsHealthy_TrailingSlashEndpoint verifies that a platform endpoint +// with a trailing slash does not produce a double-slash in the request URL, +// which strict HTTP routers can reject. +func TestSDK_IsHealthy_TrailingSlashEndpoint(t *testing.T) { + ts := newHealthTestServer(t, grpchealth.StatusServing) + defer ts.Close() + + s, err := sdk.New(ts.URL+"/", + sdk.WithPlatformConfiguration(sdk.PlatformConfiguration{ + "idp": map[string]interface{}{ + "issuer": "https://example.org", + "authorization_endpoint": "https://example.org/auth", + "token_endpoint": "https://example.org/token", + }, + }), + ) + require.NoError(t, err) + require.NotNil(t, s) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + healthy, err := s.IsHealthy(ctx) + require.NoError(t, err) + assert.True(t, healthy) +}