diff --git a/CONFIGURATION.md b/CONFIGURATION.md index b622e15f60..76b507e757 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -54,6 +54,48 @@ Other Quarkus properties that are specifically relevant for the service: | `stargate.jsonapi.operations.database-config.ddl-delay-millis` | `int` | `2000` | Delay between create table and create index to get the schema sync. | | `stargate.jsonapi.operations.vectorize-enabled` | `boolean` | `false` | Flag to enable server side vectorization. | +### Database readiness + +`GET /v1/health/ready` is an authenticated database readiness endpoint used for both Astra and +Cassandra deployments. It uses the request's tenant, `Token` header, and `User-Agent` to obtain a +session through the normal session cache. The Data API does not store separate readiness +credentials. + +The endpoint executes `SELECT * FROM datastax_sla.check LIMIT 1` at `LOCAL_QUORUM`, using the +`table-read` driver profile for the remaining read settings. An `UP` response therefore confirms +that the coordinator can complete a read from a replicated table at local quorum. It does not +validate every tenant's credentials, write availability, or cross-region availability. + +The deployment must provide a dedicated canary tenant and credentials for this request and must +provision a `datastax_sla.check` table that the canary principal can read. Its replication factor +must be appropriate for the deployment (greater than one in a multi-node local data center) so +`LOCAL_QUORUM` requires responses from multiple replicas. Astra callers must use the canary database +hostname so the tenant and region are resolved from `Host`; Cassandra ignores the tenant portion of +`Host`. The caller must also send the full User-Agent configured by +`stargate.jsonapi.operations.sla-user-agent`. The comparison is case-insensitive. Requests with a +missing or different User-Agent are rejected before accessing the session cache, and the endpoint +fails closed when the SLA User-Agent is not configured. This ensures the canary session uses the +shorter SLA session TTL instead of being treated like normal client traffic. Do not reuse the canary +credentials for normal traffic, because using the same cached session with a non-SLA User-Agent can +extend its lifetime. + +The check is fully asynchronous and has a five-second timeout. It returns HTTP 200 with +`{"status":"UP"}` after a successful read and HTTP 503 with `{"status":"DOWN"}` after a database +failure, timeout, or missing SLA User-Agent configuration. It returns the standard Data API error +response with HTTP 401 when the `Token` header is missing or authentication fails, and HTTP 403 when +the request User-Agent does not match the configured SLA User-Agent. Probe integrations must use the +HTTP status as the readiness contract rather than parsing the response body's `status` field alone. + +Kubernetes or an SLA checker must call each pod directly for this endpoint to control per-pod +readiness. An external request sent through a load balancer does not establish which pod is ready. +Restrict the endpoint to trusted probe traffic with deployment controls such as a NetworkPolicy, +mTLS, or an ingress ACL and rate limit. The User-Agent check is an operational guard, not an +authentication boundary. Kubernetes `httpGet` headers cannot reference a Secret, so +delivery of the canary token is intentionally outside the Data API configuration. Prefer an +external checker or a Secret-mounted file read by an `exec` probe; do not put the token literally in +the probe command or shell trace. The unauthenticated Quarkus health endpoints under the +non-application path do not include this database check. + ## Jsonapi metering configuration *Configuration for jsonapi metering, defined by [JsonApiMetricsConfig.java](io/stargate/sgv2/jsonapi/api/v1/metrics/JsonApiMetricsConfig.java).* diff --git a/src/main/java/io/stargate/sgv2/jsonapi/api/health/DatabaseReadinessCheck.java b/src/main/java/io/stargate/sgv2/jsonapi/api/health/DatabaseReadinessCheck.java new file mode 100644 index 0000000000..6f4c3e0d50 --- /dev/null +++ b/src/main/java/io/stargate/sgv2/jsonapi/api/health/DatabaseReadinessCheck.java @@ -0,0 +1,73 @@ +package io.stargate.sgv2.jsonapi.api.health; + +import com.datastax.oss.driver.api.core.DefaultConsistencyLevel; +import com.datastax.oss.driver.api.core.cql.SimpleStatement; +import com.google.common.annotations.VisibleForTesting; +import io.smallrye.mutiny.Uni; +import io.stargate.sgv2.jsonapi.api.request.RequestContext; +import io.stargate.sgv2.jsonapi.service.cqldriver.CQLSessionCache; +import io.stargate.sgv2.jsonapi.service.cqldriver.executor.CommandQueryExecutor; +import java.time.Duration; +import java.util.Objects; +import java.util.function.Supplier; + +/** + * Runs the database probe exposed at {@code GET /v1/health/ready}. + * + *

This class is constructed by the JAX-RS resource and is not a CDI bean or a MicroProfile + * health check. The caller's request context supplies the tenant, token, and User-Agent for both + * Astra and Cassandra connections. + */ +public final class DatabaseReadinessCheck { + + private static final String READINESS_QUERY = "SELECT * FROM datastax_sla.check LIMIT 1"; + private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(5); + + private final Supplier sessionCacheSupplier; + private final SimpleStatement statement; + private final Duration timeout; + + public DatabaseReadinessCheck(Supplier sessionCacheSupplier) { + this(sessionCacheSupplier, DEFAULT_TIMEOUT); + } + + @VisibleForTesting + DatabaseReadinessCheck(CQLSessionCache sessionCache, Duration timeout) { + this(() -> sessionCache, timeout); + } + + private DatabaseReadinessCheck(Supplier sessionCacheSupplier, Duration timeout) { + this.sessionCacheSupplier = + Objects.requireNonNull(sessionCacheSupplier, "sessionCacheSupplier must not be null"); + this.timeout = Objects.requireNonNull(timeout, "timeout must not be null"); + this.statement = + SimpleStatement.builder(READINESS_QUERY) + .setConsistencyLevel(DefaultConsistencyLevel.LOCAL_QUORUM) + .setTimeout(timeout) + .build(); + } + + /** + * Executes a replicated table read at {@code LOCAL_QUORUM}, using the {@code table-read} driver + * profile for the remaining read settings. + */ + public Uni check(RequestContext requestContext) { + Objects.requireNonNull(requestContext, "requestContext must not be null"); + + return Uni.createFrom() + .deferred( + () -> { + var sessionCache = + Objects.requireNonNull( + sessionCacheSupplier.get(), "sessionCacheSupplier returned null"); + return new CommandQueryExecutor( + sessionCache, requestContext, CommandQueryExecutor.QueryTarget.TABLE) + .executeRead(statement) + .replaceWithVoid(); + }) + // The statement timeout bounds driver I/O; this also bounds asynchronous session lookup. + .ifNoItem() + .after(timeout) + .fail(); + } +} diff --git a/src/main/java/io/stargate/sgv2/jsonapi/api/v1/DatabaseReadinessResource.java b/src/main/java/io/stargate/sgv2/jsonapi/api/v1/DatabaseReadinessResource.java new file mode 100644 index 0000000000..7e39b952c2 --- /dev/null +++ b/src/main/java/io/stargate/sgv2/jsonapi/api/v1/DatabaseReadinessResource.java @@ -0,0 +1,139 @@ +package io.stargate.sgv2.jsonapi.api.v1; + +import io.quarkus.security.UnauthorizedException; +import io.smallrye.mutiny.Uni; +import io.stargate.sgv2.jsonapi.api.health.DatabaseReadinessCheck; +import io.stargate.sgv2.jsonapi.api.model.command.CommandResult; +import io.stargate.sgv2.jsonapi.api.model.command.tracing.RequestTracing; +import io.stargate.sgv2.jsonapi.api.request.RequestContext; +import io.stargate.sgv2.jsonapi.config.constants.OpenApiConstants; +import io.stargate.sgv2.jsonapi.exception.APISecurityException; +import io.stargate.sgv2.jsonapi.service.cqldriver.CqlSessionCacheSupplier; +import jakarta.inject.Inject; +import jakarta.ws.rs.GET; +import jakarta.ws.rs.Path; +import jakarta.ws.rs.Produces; +import jakarta.ws.rs.core.MediaType; +import jakarta.ws.rs.core.Response; +import java.util.Collections; +import java.util.IdentityHashMap; +import org.eclipse.microprofile.openapi.annotations.Operation; +import org.eclipse.microprofile.openapi.annotations.media.Content; +import org.eclipse.microprofile.openapi.annotations.media.Schema; +import org.eclipse.microprofile.openapi.annotations.responses.APIResponse; +import org.eclipse.microprofile.openapi.annotations.responses.APIResponses; +import org.eclipse.microprofile.openapi.annotations.security.SecurityRequirement; +import org.jboss.resteasy.reactive.RestResponse; + +/** + * Authenticated database readiness endpoint registered through Quarkus JAX-RS resource discovery. + * + *

{@code GET /v1/health/ready} runs the same request-scoped probe for Astra and Cassandra. A + * request must use the configured SLA User-Agent. A successful probe returns HTTP 200; invalid + * credentials return HTTP 401; a missing or different SLA User-Agent returns HTTP 403; and a + * database failure, timeout, or missing SLA configuration returns HTTP 503. The existing {@code + * /v1/*} security policy rejects requests without a token before this resource is called. + */ +@Path(DatabaseReadinessResource.BASE_PATH) +@Produces(MediaType.APPLICATION_JSON) +@SecurityRequirement(name = OpenApiConstants.SecuritySchemes.TOKEN) +public class DatabaseReadinessResource { + + public static final String BASE_PATH = GeneralResource.BASE_PATH + "/health/ready"; + + private static final ReadinessResponse UP = new ReadinessResponse("UP"); + private static final ReadinessResponse DOWN = new ReadinessResponse("DOWN"); + + private final DatabaseReadinessCheck readinessCheck; + private final RequestContext requestContext; + private final CqlSessionCacheSupplier sessionCacheSupplier; + + @Inject + public DatabaseReadinessResource( + CqlSessionCacheSupplier sessionCacheSupplier, RequestContext requestContext) { + this.readinessCheck = new DatabaseReadinessCheck(sessionCacheSupplier); + this.requestContext = requestContext; + this.sessionCacheSupplier = sessionCacheSupplier; + } + + @GET + @Operation( + summary = "Check database readiness", + description = + "Uses the authenticated request tenant and token to perform a LOCAL_QUORUM read.") + @APIResponses({ + @APIResponse( + responseCode = "200", + description = "The database completed the readiness read.", + content = + @Content( + mediaType = MediaType.APPLICATION_JSON, + schema = @Schema(implementation = ReadinessResponse.class))), + @APIResponse( + responseCode = "401", + description = "The token is missing or invalid.", + content = + @Content( + mediaType = MediaType.APPLICATION_JSON, + schema = @Schema(implementation = CommandResult.class))), + @APIResponse( + responseCode = "403", + description = "The request does not use the configured SLA User-Agent."), + @APIResponse( + responseCode = "503", + description = "The SLA User-Agent is not configured, or the database check failed.", + content = + @Content( + mediaType = MediaType.APPLICATION_JSON, + schema = @Schema(implementation = ReadinessResponse.class))) + }) + public Uni> ready() { + var configuredSlaUserAgent = sessionCacheSupplier.slaUserAgent(); + if (configuredSlaUserAgent.isEmpty()) { + return Uni.createFrom().item(response(Response.Status.SERVICE_UNAVAILABLE, DOWN)); + } + if (!configuredSlaUserAgent.get().equals(requestContext.userAgent())) { + return Uni.createFrom().item(response(Response.Status.FORBIDDEN)); + } + + return readinessCheck + .check(requestContext) + .map(ignored -> response(Response.Status.OK, UP)) + .onFailure(DatabaseReadinessResource::isUnauthorized) + .recoverWithItem(failure -> unauthorizedResponse()) + .onFailure() + .recoverWithItem(failure -> response(Response.Status.SERVICE_UNAVAILABLE, DOWN)); + } + + private static boolean isUnauthorized(Throwable failure) { + var current = failure; + var seen = Collections.newSetFromMap(new IdentityHashMap()); + while (current != null && seen.add(current)) { + if (current instanceof UnauthorizedException + || current instanceof APISecurityException apiException + && apiException.httpStatus == Response.Status.UNAUTHORIZED.getStatusCode()) { + return true; + } + current = current.getCause(); + } + return false; + } + + private static RestResponse unauthorizedResponse() { + var commandResult = + CommandResult.statusOnlyBuilder(RequestTracing.NO_OP) + .addThrowable(APISecurityException.Code.UNAUTHENTICATED_REQUEST.get()) + .build(); + return response(Response.Status.UNAUTHORIZED, commandResult); + } + + private static RestResponse response(Response.Status status) { + return RestResponse.ResponseBuilder.create(status).build(); + } + + private static RestResponse response(Response.Status status, Object entity) { + return RestResponse.ResponseBuilder.create(status, entity).build(); + } + + public record ReadinessResponse(String status) {} +} diff --git a/src/main/java/io/stargate/sgv2/jsonapi/metrics/TenantRequestMetricsFilter.java b/src/main/java/io/stargate/sgv2/jsonapi/metrics/TenantRequestMetricsFilter.java index e365b4891e..b277403ff3 100644 --- a/src/main/java/io/stargate/sgv2/jsonapi/metrics/TenantRequestMetricsFilter.java +++ b/src/main/java/io/stargate/sgv2/jsonapi/metrics/TenantRequestMetricsFilter.java @@ -21,6 +21,7 @@ import io.micrometer.core.instrument.Tag; import io.micrometer.core.instrument.Tags; import io.stargate.sgv2.jsonapi.api.request.RequestContext; +import io.stargate.sgv2.jsonapi.api.v1.DatabaseReadinessResource; import io.stargate.sgv2.jsonapi.api.v1.metrics.MetricsConfig; import jakarta.enterprise.context.ApplicationScoped; import jakarta.inject.Inject; @@ -71,31 +72,37 @@ public TenantRequestMetricsFilter( @ServerResponseFilter public void record( ContainerRequestContext requestContext, ContainerResponseContext responseContext) { - // only if enabled - if (config.enabled()) { - - // resolve tenant - Tag tenantTag = Tag.of(config.tenantTag(), this.requestContext.tenant().toString()); + if (!config.enabled() || isDatabaseReadinessRequest(requestContext)) { + return; + } - // resolve error - boolean error = responseContext.getStatus() >= 500; - Tag errorTag = error ? ExceptionMetrics.TAG_ERROR_TRUE : ExceptionMetrics.TAG_ERROR_FALSE; + // resolve tenant + Tag tenantTag = Tag.of(config.tenantTag(), this.requestContext.tenant().toString()); - // check if we need user agent as well - Tags tags = Tags.of(tenantTag, errorTag); - if (config.userAgentTagEnabled()) { - String userAgentValue = getUserAgentValue(requestContext); - tags = tags.and(Tag.of(config.userAgentTag(), userAgentValue)); - } + // resolve error + boolean error = responseContext.getStatus() >= 500; + Tag errorTag = error ? ExceptionMetrics.TAG_ERROR_TRUE : ExceptionMetrics.TAG_ERROR_FALSE; - // add http status code - if (config.statusTagEnabled()) { - tags = tags.and(Tag.of(config.statusTag(), String.valueOf(responseContext.getStatus()))); - } + // check if we need user agent as well + Tags tags = Tags.of(tenantTag, errorTag); + if (config.userAgentTagEnabled()) { + String userAgentValue = getUserAgentValue(requestContext); + tags = tags.and(Tag.of(config.userAgentTag(), userAgentValue)); + } - // record - meterRegistry.counter(config.metricName(), tags).increment(); + // add http status code + if (config.statusTagEnabled()) { + tags = tags.and(Tag.of(config.statusTag(), String.valueOf(responseContext.getStatus()))); } + + // record + meterRegistry.counter(config.metricName(), tags).increment(); + } + + private static boolean isDatabaseReadinessRequest(ContainerRequestContext requestContext) { + var requestPath = requestContext.getUriInfo().getRequestUri().getPath(); + return DatabaseReadinessResource.BASE_PATH.equals(requestPath) + || (DatabaseReadinessResource.BASE_PATH + "/").equals(requestPath); } private String getUserAgentValue(ContainerRequestContext requestContext) { diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/cqldriver/CqlSessionCacheSupplier.java b/src/main/java/io/stargate/sgv2/jsonapi/service/cqldriver/CqlSessionCacheSupplier.java index 766c36899f..7e569d60e4 100644 --- a/src/main/java/io/stargate/sgv2/jsonapi/service/cqldriver/CqlSessionCacheSupplier.java +++ b/src/main/java/io/stargate/sgv2/jsonapi/service/cqldriver/CqlSessionCacheSupplier.java @@ -10,6 +10,7 @@ import java.time.Duration; import java.util.List; import java.util.Objects; +import java.util.Optional; import java.util.function.Supplier; import org.eclipse.microprofile.config.inject.ConfigProperty; @@ -23,6 +24,7 @@ public class CqlSessionCacheSupplier implements Supplier { private final CQLSessionCache singleton; + private final Optional slaUserAgent; @Inject public CqlSessionCacheSupplier( @@ -53,11 +55,14 @@ public CqlSessionCacheSupplier( dbConfig.cassandraPort(), () -> schemaObjectCacheSupplier.get().getSchemaChangeListener()); + slaUserAgent = + operationsConfig.slaUserAgent().filter(value -> !value.isBlank()).map(UserAgent::new); + singleton = new CQLSessionCache( dbConfig.sessionCacheMaxSize(), Duration.ofSeconds(dbConfig.sessionCacheTtlSeconds()), - operationsConfig.slaUserAgent().map(UserAgent::new).orElse(null), + slaUserAgent.orElse(null), Duration.ofSeconds(dbConfig.slaSessionCacheTtlSeconds()), credentialsFactory, sessionFactory, @@ -70,4 +75,9 @@ public CqlSessionCacheSupplier( public CQLSessionCache get() { return singleton; } + + /** Gets the configured User-Agent that selects the shorter SLA session-cache TTL. */ + public Optional slaUserAgent() { + return slaUserAgent; + } } diff --git a/src/main/resources/application.yaml b/src/main/resources/application.yaml index 354232535c..a978343bbc 100644 --- a/src/main/resources/application.yaml +++ b/src/main/resources/application.yaml @@ -159,7 +159,7 @@ quarkus: http-server: # ignore all non-application uris, as well as the custom set suppress-non-application-uris: true - ignore-patterns: /,/metrics,/swagger-ui.*,.*\.html + ignore-patterns: /,/metrics,/swagger-ui.*,.*\.html,/v1/health/ready/? # due to the https://github.com/quarkusio/quarkus/issues/24938 # we need to define uri templating on our own for now diff --git a/src/test/java/io/stargate/sgv2/jsonapi/api/health/DatabaseReadinessCheckTest.java b/src/test/java/io/stargate/sgv2/jsonapi/api/health/DatabaseReadinessCheckTest.java new file mode 100644 index 0000000000..6893797446 --- /dev/null +++ b/src/test/java/io/stargate/sgv2/jsonapi/api/health/DatabaseReadinessCheckTest.java @@ -0,0 +1,147 @@ +package io.stargate.sgv2.jsonapi.api.health; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.same; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.DefaultConsistencyLevel; +import com.datastax.oss.driver.api.core.cql.AsyncResultSet; +import com.datastax.oss.driver.api.core.cql.SimpleStatement; +import io.smallrye.mutiny.TimeoutException; +import io.smallrye.mutiny.Uni; +import io.smallrye.mutiny.helpers.test.UniAssertSubscriber; +import io.stargate.sgv2.jsonapi.api.request.RequestContext; +import io.stargate.sgv2.jsonapi.api.request.UserAgent; +import io.stargate.sgv2.jsonapi.api.request.tenant.Tenant; +import io.stargate.sgv2.jsonapi.config.DatabaseType; +import io.stargate.sgv2.jsonapi.service.cqldriver.CQLSessionCache; +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +public class DatabaseReadinessCheckTest { + + private static final Duration TIMEOUT = Duration.ofMillis(100); + private static final String ASTRA_TENANT_ID = "60b5dccb-e91d-4a60-987b-7588cd8aa1e3"; + + private CQLSessionCache sessionCache; + private CqlSession session; + private RequestContext requestContext; + private DatabaseReadinessCheck readinessCheck; + + @BeforeEach + public void setup() { + sessionCache = mock(CQLSessionCache.class); + session = mock(CqlSession.class); + requestContext = + new RequestContext( + Tenant.create(DatabaseType.ASTRA, ASTRA_TENANT_ID, "us-west-2"), + "astra-token", + new UserAgent("Datastax-SLA-Checker")); + readinessCheck = new DatabaseReadinessCheck(sessionCache, TIMEOUT); + } + + @Test + public void successfulCheckUsesRequestContextAndAsyncDistributedQuery() { + var resultSet = mock(AsyncResultSet.class); + var resultFuture = new CompletableFuture(); + when(sessionCache.getSession(requestContext)).thenReturn(Uni.createFrom().item(session)); + when(session.executeAsync(any(SimpleStatement.class))).thenReturn(resultFuture); + + var subscriber = + readinessCheck + .check(requestContext) + .subscribe() + .withSubscriber(UniAssertSubscriber.create()); + + subscriber.assertSubscribed().assertNotTerminated(); + resultFuture.complete(resultSet); + subscriber.awaitItem().assertItem(null).assertCompleted(); + + verify(sessionCache).getSession(same(requestContext)); + var statementCaptor = ArgumentCaptor.forClass(SimpleStatement.class); + verify(session).executeAsync(statementCaptor.capture()); + assertThat(statementCaptor.getValue().getQuery()) + .isEqualTo("SELECT * FROM datastax_sla.check LIMIT 1"); + assertThat(statementCaptor.getValue().getTimeout()).isEqualTo(TIMEOUT); + assertThat(statementCaptor.getValue().getConsistencyLevel()) + .isEqualTo(DefaultConsistencyLevel.LOCAL_QUORUM); + assertThat(statementCaptor.getValue().getExecutionProfileName()).isEqualTo("table-read"); + verify(session, never()).execute(any(SimpleStatement.class)); + verify(session, never()).isClosed(); + verify(sessionCache, never()).evictSession(any(RequestContext.class)); + } + + @Test + public void sessionAcquisitionFailureIsPropagated() { + when(sessionCache.getSession(requestContext)) + .thenReturn(Uni.createFrom().failure(new IllegalStateException("Cannot connect"))); + + readinessCheck + .check(requestContext) + .subscribe() + .withSubscriber(UniAssertSubscriber.create()) + .awaitFailure() + .assertFailedWith(IllegalStateException.class, "Cannot connect"); + + verify(session, never()).executeAsync(any(SimpleStatement.class)); + verify(sessionCache, never()).evictSession(any(RequestContext.class)); + } + + @Test + public void asynchronousQueryFailureIsPropagated() { + var resultFuture = new CompletableFuture(); + when(sessionCache.getSession(requestContext)).thenReturn(Uni.createFrom().item(session)); + when(session.executeAsync(any(SimpleStatement.class))).thenReturn(resultFuture); + + var subscriber = + readinessCheck + .check(requestContext) + .subscribe() + .withSubscriber(UniAssertSubscriber.create()); + resultFuture.completeExceptionally(new IllegalStateException("Query failed")); + + subscriber.awaitFailure().assertFailedWith(IllegalStateException.class, "Query failed"); + verify(sessionCache, never()).evictSession(any(RequestContext.class)); + } + + @Test + public void reactiveTimeoutBoundsSessionAcquisition() { + readinessCheck = new DatabaseReadinessCheck(sessionCache, Duration.ofMillis(20)); + when(sessionCache.getSession(requestContext)).thenReturn(Uni.createFrom().nothing()); + + readinessCheck + .check(requestContext) + .subscribe() + .withSubscriber(UniAssertSubscriber.create()) + .awaitFailure() + .assertFailedWith(TimeoutException.class); + + verify(session, never()).executeAsync(any(SimpleStatement.class)); + verify(sessionCache, never()).evictSession(any(RequestContext.class)); + } + + @Test + public void reactiveTimeoutBoundsAsynchronousQuery() { + readinessCheck = new DatabaseReadinessCheck(sessionCache, Duration.ofMillis(20)); + when(sessionCache.getSession(requestContext)).thenReturn(Uni.createFrom().item(session)); + when(session.executeAsync(any(SimpleStatement.class))).thenReturn(new CompletableFuture<>()); + + readinessCheck + .check(requestContext) + .subscribe() + .withSubscriber(UniAssertSubscriber.create()) + .awaitFailure() + .assertFailedWith(TimeoutException.class); + + verify(session).executeAsync(any(SimpleStatement.class)); + verify(sessionCache, never()).evictSession(any(RequestContext.class)); + } +} diff --git a/src/test/java/io/stargate/sgv2/jsonapi/api/v1/DatabaseReadinessResourceTest.java b/src/test/java/io/stargate/sgv2/jsonapi/api/v1/DatabaseReadinessResourceTest.java new file mode 100644 index 0000000000..c3db7999f8 --- /dev/null +++ b/src/test/java/io/stargate/sgv2/jsonapi/api/v1/DatabaseReadinessResourceTest.java @@ -0,0 +1,239 @@ +package io.stargate.sgv2.jsonapi.api.v1; + +import static io.restassured.RestAssured.given; +import static org.assertj.core.api.Assertions.assertThat; +import static org.hamcrest.Matchers.equalTo; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.cql.AsyncResultSet; +import com.datastax.oss.driver.api.core.cql.SimpleStatement; +import io.quarkus.security.UnauthorizedException; +import io.quarkus.test.InjectMock; +import io.quarkus.test.junit.QuarkusTest; +import io.quarkus.test.junit.TestProfile; +import io.smallrye.mutiny.Uni; +import io.stargate.sgv2.jsonapi.api.request.RequestContext; +import io.stargate.sgv2.jsonapi.api.request.UserAgent; +import io.stargate.sgv2.jsonapi.config.constants.HttpConstants; +import io.stargate.sgv2.jsonapi.exception.APISecurityException; +import io.stargate.sgv2.jsonapi.service.cqldriver.CQLSessionCache; +import io.stargate.sgv2.jsonapi.service.cqldriver.CqlSessionCacheSupplier; +import io.stargate.sgv2.jsonapi.testresource.NoGlobalResourcesTestProfile; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +@QuarkusTest +@TestProfile(DatabaseReadinessResourceTest.AstraProfile.class) +public class DatabaseReadinessResourceTest { + + private static final String TENANT_ID = "60b5dccb-e91d-4a60-987b-7588cd8aa1e3"; + private static final String REGION = "us-west-2"; + private static final String ASTRA_HOST = TENANT_ID + "-" + REGION + ".apps.astra.datastax.com"; + private static final String TOKEN = "astra-canary-token"; + private static final String SLA_USER_AGENT = "Datastax-SLA-Checker"; + + @InjectMock CqlSessionCacheSupplier sessionCacheSupplier; + + private CQLSessionCache sessionCache; + private CqlSession session; + + @BeforeEach + public void setup() { + sessionCache = mock(CQLSessionCache.class); + session = mock(CqlSession.class); + when(sessionCacheSupplier.get()).thenReturn(sessionCache); + when(sessionCacheSupplier.slaUserAgent()) + .thenReturn(Optional.of(new UserAgent(SLA_USER_AGENT))); + } + + @Test + public void missingTokenIsRejectedBeforeDatabaseAccess() { + given() + .header("Host", ASTRA_HOST) + .header("User-Agent", SLA_USER_AGENT) + .when() + .get(DatabaseReadinessResource.BASE_PATH) + .then() + .statusCode(401) + .body("errors[0].errorCode", equalTo("MISSING_AUTHENTICATION_TOKEN")); + + verifyNoInteractions(sessionCache); + } + + @ParameterizedTest + @ValueSource(strings = {"", "/"}) + public void astraRequestUsesResolvedTenantTokenAndSlaUserAgent(String pathSuffix) { + var resultSet = mock(AsyncResultSet.class); + var capturedContext = new AtomicReference(); + when(sessionCache.getSession(any(RequestContext.class))) + .thenAnswer( + invocation -> { + RequestContext context = invocation.getArgument(0); + capturedContext.set( + new RequestContextSnapshot( + context.tenant().toString(), + context.tenant().region(), + context.authToken(), + context.userAgent().toString())); + return Uni.createFrom().item(session); + }); + when(session.executeAsync(any(SimpleStatement.class))) + .thenReturn(CompletableFuture.completedFuture(resultSet)); + + authenticatedRequest() + .when() + .get(DatabaseReadinessResource.BASE_PATH + pathSuffix) + .then() + .statusCode(200) + .body("status", equalTo("UP")); + + assertThat(capturedContext.get()) + .isEqualTo(new RequestContextSnapshot(TENANT_ID, REGION, TOKEN, SLA_USER_AGENT)); + } + + @Test + public void wrongSlaUserAgentIsRejectedBeforeDatabaseAccess() { + given() + .header("Host", ASTRA_HOST) + .header(HttpConstants.AUTHENTICATION_TOKEN_HEADER_NAME, TOKEN) + .header("User-Agent", "ordinary-client") + .when() + .get(DatabaseReadinessResource.BASE_PATH) + .then() + .statusCode(403); + + verifyNoInteractions(sessionCache); + } + + @Test + public void missingSlaUserAgentConfigurationReturnsServiceUnavailable() { + when(sessionCacheSupplier.slaUserAgent()).thenReturn(Optional.empty()); + + authenticatedRequest() + .when() + .get(DatabaseReadinessResource.BASE_PATH) + .then() + .statusCode(503) + .body("status", equalTo("DOWN")); + + verifyNoInteractions(sessionCache); + } + + @Test + public void databaseFailureReturnsServiceUnavailableWithoutDetails() { + when(sessionCache.getSession(any(RequestContext.class))) + .thenReturn(Uni.createFrom().failure(new IllegalStateException("sensitive failure"))); + + var response = + authenticatedRequest() + .when() + .get(DatabaseReadinessResource.BASE_PATH) + .then() + .statusCode(503) + .body("status", equalTo("DOWN")) + .extract() + .asString(); + + assertThat(response).doesNotContain("sensitive failure", TOKEN, TENANT_ID); + } + + @Test + public void invalidTokenFailureReturnsUnauthorizedWithoutDetails() { + when(sessionCache.getSession(any(RequestContext.class))) + .thenThrow(new UnauthorizedException("sensitive credential failure")); + + var response = + authenticatedRequest() + .when() + .get(DatabaseReadinessResource.BASE_PATH) + .then() + .statusCode(401) + .body("errors[0].errorCode", equalTo("UNAUTHENTICATED_REQUEST")) + .extract() + .asString(); + + assertThat(response).doesNotContain("sensitive credential failure", TOKEN, TENANT_ID); + } + + @Test + public void databaseAuthenticationFailureReturnsUnauthorizedWithoutDetails() { + var authenticationFailure = + APISecurityException.Code.UNAUTHENTICATED_REQUEST.withPreformattedMessage( + "sensitive database authentication failure"); + when(sessionCache.getSession(any(RequestContext.class))) + .thenReturn(Uni.createFrom().failure(authenticationFailure)); + + var response = + authenticatedRequest() + .when() + .get(DatabaseReadinessResource.BASE_PATH) + .then() + .statusCode(401) + .body("errors[0].errorCode", equalTo("UNAUTHENTICATED_REQUEST")) + .extract() + .asString(); + + assertThat(response) + .doesNotContain("sensitive database authentication failure", TOKEN, TENANT_ID); + } + + @Test + public void cyclicFailureCauseReturnsServiceUnavailable() { + var firstFailure = new IllegalStateException("first sensitive failure"); + var secondFailure = new IllegalArgumentException("second sensitive failure"); + firstFailure.initCause(secondFailure); + secondFailure.initCause(firstFailure); + when(sessionCache.getSession(any(RequestContext.class))) + .thenReturn(Uni.createFrom().failure(firstFailure)); + + var response = + authenticatedRequest() + .when() + .get(DatabaseReadinessResource.BASE_PATH) + .then() + .statusCode(503) + .body("status", equalTo("DOWN")) + .extract() + .asString(); + + assertThat(response).doesNotContain("first sensitive failure", "second sensitive failure"); + } + + private io.restassured.specification.RequestSpecification authenticatedRequest() { + return given() + .header("Host", ASTRA_HOST) + .header(HttpConstants.AUTHENTICATION_TOKEN_HEADER_NAME, TOKEN) + .header("User-Agent", SLA_USER_AGENT); + } + + private record RequestContextSnapshot( + String tenantId, String region, String authToken, String userAgent) {} + + public static class AstraProfile implements NoGlobalResourcesTestProfile { + + @Override + public Map getConfigOverrides() { + return Map.of( + "stargate.jsonapi.operations.database-config.type", + "ASTRA", + "stargate.multi-tenancy.enabled", + "true", + "stargate.multi-tenancy.tenant-resolver.type", + "subdomain", + "stargate.multi-tenancy.tenant-resolver.subdomain.max-chars", + "36", + "stargate.jsonapi.operations.sla-user-agent", + SLA_USER_AGENT); + } + } +} diff --git a/src/test/java/io/stargate/sgv2/jsonapi/api/v1/SessionEvictionIntegrationTest.java b/src/test/java/io/stargate/sgv2/jsonapi/api/v1/SessionEvictionIntegrationTest.java index 322dbd3711..5afb578c28 100644 --- a/src/test/java/io/stargate/sgv2/jsonapi/api/v1/SessionEvictionIntegrationTest.java +++ b/src/test/java/io/stargate/sgv2/jsonapi/api/v1/SessionEvictionIntegrationTest.java @@ -16,6 +16,7 @@ import java.io.IOException; import java.net.ServerSocket; import java.util.Collections; +import java.util.HashMap; import java.util.Map; import org.junit.jupiter.api.Test; import org.slf4j.Logger; @@ -30,6 +31,7 @@ public class SessionEvictionIntegrationTest extends AbstractCollectionIntegratio private static final Logger LOGGER = LoggerFactory.getLogger(SessionEvictionIntegrationTest.class); + private static final String READINESS_USER_AGENT = "DataAPI-Readiness-Test/1.0"; /** * Overridden to ensure we connect to the isolated container created for this test. @@ -50,6 +52,13 @@ protected int getCassandraCqlPort() { @Test public void testSessionEvictionOnAllNodesFailed() { + if (!executeCqlStatement( + "CREATE KEYSPACE IF NOT EXISTS datastax_sla " + + "WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}", + "CREATE TABLE IF NOT EXISTS datastax_sla.check (id text PRIMARY KEY)")) { + throw new AssertionError("Failed to provision the distributed readiness table"); + } + // 1. Insert and find initial data to ensure the database is healthy before the test insertDoc( """ @@ -69,6 +78,8 @@ public void testSessionEvictionOnAllNodesFailed() { .body("$", responseIsFindSuccess()) .body("data.document._id", is("before_crash")); + waitForReadinessStatus("UP", 30_000); + // 2. Stop the container to simulate DB failure // Use low-level dockerClient to stop the container without triggering Testcontainers' // cleanup/termination logic (which dbContainer.stop() would do). @@ -86,11 +97,14 @@ public void testSessionEvictionOnAllNodesFailed() { .body( "errors[0].errorCode", is(DatabaseException.Code.FAILED_TO_CONNECT_TO_DATABASE.name())); + waitForReadinessStatus("DOWN", 60_000); + // 4. Restart the container to simulate recovery getDockerClient().startContainerCmd(getContainerId()).exec(); // 5. Wait for the database to become responsive again waitForDbRecovery(); + waitForReadinessStatus("UP", 120_000); // 6. Verify Session Recovery: check the data before crashing // Not to check that cassandra works, but to check that we are running the same container as @@ -225,6 +239,52 @@ private boolean isApiReady() { } } + /** Polls the authenticated database readiness endpoint until it has the expected status. */ + private void waitForReadinessStatus(String expectedStatus, long timeoutMillis) { + var start = System.currentTimeMillis(); + var expectedStatusCode = "UP".equals(expectedStatus) ? 200 : 503; + Response lastResponse = null; + + while (System.currentTimeMillis() - start < timeoutMillis) { + try { + lastResponse = + given() + .headers(getHeaders()) + .header("User-Agent", READINESS_USER_AGENT) + .when() + .get(DatabaseReadinessResource.BASE_PATH); + var jsonPath = lastResponse.jsonPath(); + var overallStatus = jsonPath.getString("status"); + + if (lastResponse.statusCode() == expectedStatusCode + && expectedStatus.equals(overallStatus)) { + return; + } + } catch (Exception e) { + LOGGER.debug("Readiness endpoint not in expected state yet", e); + } + + try { + Thread.sleep(1000); + } catch (InterruptedException ignored) { + Thread.currentThread().interrupt(); + break; + } + } + + var lastResponseDescription = + lastResponse == null + ? "no response" + : "HTTP " + lastResponse.statusCode() + ": " + lastResponse.asString(); + throw new AssertionError( + "Readiness did not become " + + expectedStatus + + " within " + + timeoutMillis + + " ms. Last response: " + + lastResponseDescription); + } + /** Checks if Cassandra is up and normal by running "nodetool status" inside the container. */ private boolean isCassandraUp(DockerClient dockerClient, String containerId) { try { @@ -317,9 +377,10 @@ protected GenericContainer baseCassandraContainer(boolean reuse) { */ @Override public Map start() { - var props = super.start(); + var props = new HashMap<>(super.start()); + props.put("stargate.jsonapi.operations.sla-user-agent", READINESS_USER_AGENT); sessionEvictionCassandraContainer = super.getCassandraContainer(); - return props; + return Map.copyOf(props); } /** diff --git a/src/test/java/io/stargate/sgv2/jsonapi/metrics/TenantRequestMetricsFilterTest.java b/src/test/java/io/stargate/sgv2/jsonapi/metrics/TenantRequestMetricsFilterTest.java new file mode 100644 index 0000000000..648e2bcaf2 --- /dev/null +++ b/src/test/java/io/stargate/sgv2/jsonapi/metrics/TenantRequestMetricsFilterTest.java @@ -0,0 +1,43 @@ +package io.stargate.sgv2.jsonapi.metrics; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import io.micrometer.core.instrument.MeterRegistry; +import io.stargate.sgv2.jsonapi.api.request.RequestContext; +import io.stargate.sgv2.jsonapi.api.v1.DatabaseReadinessResource; +import io.stargate.sgv2.jsonapi.api.v1.metrics.MetricsConfig; +import jakarta.ws.rs.container.ContainerRequestContext; +import jakarta.ws.rs.container.ContainerResponseContext; +import jakarta.ws.rs.core.UriInfo; +import java.net.URI; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +public class TenantRequestMetricsFilterTest { + + @ParameterizedTest + @ValueSource(strings = {"", "/"}) + public void databaseReadinessIsNotCountedAsTenantTraffic(String pathSuffix) { + var meterRegistry = mock(MeterRegistry.class); + var dataApiRequestContext = mock(RequestContext.class); + var metricsConfig = mock(MetricsConfig.class); + var tenantRequestConfig = mock(MetricsConfig.TenantRequestCounterConfig.class); + when(metricsConfig.tenantRequestCounter()).thenReturn(tenantRequestConfig); + when(tenantRequestConfig.enabled()).thenReturn(true); + + var requestContext = mock(ContainerRequestContext.class); + var uriInfo = mock(UriInfo.class); + when(requestContext.getUriInfo()).thenReturn(uriInfo); + when(uriInfo.getRequestUri()) + .thenReturn( + URI.create("http://localhost" + DatabaseReadinessResource.BASE_PATH + pathSuffix)); + + var filter = + new TenantRequestMetricsFilter(meterRegistry, dataApiRequestContext, metricsConfig); + filter.record(requestContext, mock(ContainerResponseContext.class)); + + verifyNoInteractions(meterRegistry, dataApiRequestContext); + } +} diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/cqldriver/CqlSessionCacheSupplierTests.java b/src/test/java/io/stargate/sgv2/jsonapi/service/cqldriver/CqlSessionCacheSupplierTests.java index de0cfca852..976344d441 100644 --- a/src/test/java/io/stargate/sgv2/jsonapi/service/cqldriver/CqlSessionCacheSupplierTests.java +++ b/src/test/java/io/stargate/sgv2/jsonapi/service/cqldriver/CqlSessionCacheSupplierTests.java @@ -7,6 +7,7 @@ import com.datastax.oss.driver.api.core.metadata.schema.SchemaChangeListener; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import io.stargate.sgv2.jsonapi.TestConstants; +import io.stargate.sgv2.jsonapi.api.request.UserAgent; import io.stargate.sgv2.jsonapi.config.DatabaseType; import io.stargate.sgv2.jsonapi.config.OperationsConfig; import io.stargate.sgv2.jsonapi.service.schema.SchemaObjectCache; @@ -24,6 +25,24 @@ public class CqlSessionCacheSupplierTests { public void testSingleton() { // Not a lot to test, just checking it always returns the same instance. + var factory = newFactory(Optional.of(TEST_CONSTANTS.SLA_USER_AGENT_NAME)); + + var sessionCache1 = factory.get(); + var sessionCache2 = factory.get(); + + assertThat(sessionCache1) + .as("Session cache should be the same instance") + .isSameAs(sessionCache2); + assertThat(factory.slaUserAgent()).contains(new UserAgent(TEST_CONSTANTS.SLA_USER_AGENT_NAME)); + } + + @Test + public void blankSlaUserAgentIsTreatedAsUnconfigured() { + assertThat(newFactory(Optional.of(" ")).slaUserAgent()).isEmpty(); + } + + private CqlSessionCacheSupplier newFactory(Optional slaUserAgent) { + var dbConfig = mock(OperationsConfig.DatabaseConfig.class); when(dbConfig.type()).thenReturn(DatabaseType.ASTRA); when(dbConfig.localDatacenter()).thenReturn("datacenter1"); @@ -35,8 +54,7 @@ public void testSingleton() { var operationsConfig = mock(OperationsConfig.class); when(operationsConfig.databaseConfig()).thenReturn(dbConfig); - when(operationsConfig.slaUserAgent()) - .thenReturn(Optional.of(TEST_CONSTANTS.SLA_USER_AGENT_NAME)); + when(operationsConfig.slaUserAgent()).thenReturn(slaUserAgent); var mockSchemaObjectCacheSupplier = mock(SchemaObjectCacheSupplier.class); var mockSchemaObjectCache = mock(SchemaObjectCache.class); @@ -44,15 +62,7 @@ public void testSingleton() { when(mockSchemaObjectCache.getSchemaChangeListener()) .thenReturn(mock(SchemaChangeListener.class)); - var factory = - new CqlSessionCacheSupplier( - "testApp", operationsConfig, new SimpleMeterRegistry(), mockSchemaObjectCacheSupplier); - - var sessionCache1 = factory.get(); - var sessionCache2 = factory.get(); - - assertThat(sessionCache1) - .as("Session cache should be the same instance") - .isSameAs(sessionCache2); + return new CqlSessionCacheSupplier( + "testApp", operationsConfig, new SimpleMeterRegistry(), mockSchemaObjectCacheSupplier); } }