Implement Cassandra-Based Readiness Probe for Data API

[clun]

Background

Currently, Data API pods may remain in a Ready state even when they are unable to communicate with Cassandra. This can lead to scenarios where Kubernetes continues routing traffic to unhealthy pods, resulting in request failures and inconsistent behavior across replicas.

In Kubernetes, readiness probes should reflect whether a pod can successfully perform its core function. Since the Data API depends on Cassandra for all data operations, a pod that cannot connect to Cassandra should fail its readiness check and be removed from service until connectivity is restored.

This approach would simplify the consumer experience by making the Data API either fully available or unavailable, rather than partially functional depending on which pod receives the request.

---

Problem Statement

Several failure scenarios can occur today:

• Network partition where only some Data API pods can reach Cassandra.
• Cassandra outage or degraded connectivity.
• Invalid or expired Cassandra sessions.
• Infrastructure issues affecting pod-to-Cassandra communication.

In these situations, affected Data API pods may still be considered healthy by Kubernetes and continue receiving traffic, leading to failed API requests.

---

Proposed Solution

Implement a Quarkus MicroProfile Readiness Health Check that validates Cassandra connectivity.

The readiness check should:

1. Obtain a Cassandra CqlSession from the existing session cache.
2. Verify that the session is open and usable.
3. Execute a lightweight query against Cassandra (e.g. SELECT release_version FROM system.local).
4. Mark the readiness check as:
   • UP when the query succeeds.
   • DOWN when session acquisition or query execution fails.

When the readiness check reports DOWN, Kubernetes will automatically remove the pod from service endpoints until connectivity is restored.

---

Expected Benefits

• Prevent traffic from being routed to pods that cannot access Cassandra.
• Improve reliability during partial network failures.
• Align Data API behavior with Kubernetes readiness best practices.
• Provide a simpler service contract for consumers where Data API availability accurately reflects Cassandra availability.
• Reduce intermittent and difficult-to-diagnose request failures.

Proposal

package io.stargate.sgv2.jsonapi.api.health;

import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.cql.SimpleStatement;
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.api.request.tenant.TenantFactory;
import io.stargate.sgv2.jsonapi.config.OperationsConfig;
import io.stargate.sgv2.jsonapi.service.cqldriver.CQLSessionCache;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import java.time.Duration;
import org.eclipse.microprofile.health.HealthCheck;
import org.eclipse.microprofile.health.HealthCheckResponse;
import org.eclipse.microprofile.health.HealthCheckResponseBuilder;
import org.eclipse.microprofile.health.Readiness;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Health check that verifies Cassandra connectivity for the Data API readiness probe.
 *
 * <p>This check verifies that the Data API can establish and maintain connections to Cassandra by:
 * <ul>
 *   <li>Obtaining a CQL session from the session cache</li>
 *   <li>Verifying the session is not closed</li>
 *   <li>Executing a simple query against system.local to confirm connectivity</li>
 * </ul>
 *
 * <p>In Kubernetes, this readiness check ensures that pods which cannot reach Cassandra
 * are marked as not ready and removed from the service load balancer, preventing
 * traffic from being routed to unhealthy pods.
 *
 * <p><b>Multi-tenancy Note:</b> This health check uses a default tenant configuration.
 * In multi-tenant deployments, this verifies base Cassandra connectivity but not
 * tenant-specific credentials or permissions. The check will pass if the Data API
 * can connect to Cassandra with default/system credentials.
 */
@Readiness
@ApplicationScoped
public class CassandraConnectionHealthCheck implements HealthCheck {

  private static final Logger LOGGER = LoggerFactory.getLogger(CassandraConnectionHealthCheck.class);

  // Simple query to verify connectivity - reads from system.local which always exists
  private static final String HEALTH_CHECK_QUERY = "SELECT release_version FROM system.local";
  private static final Duration HEALTH_CHECK_TIMEOUT = Duration.ofSeconds(5);
  
  // Health check uses a synthetic tenant ID for verification
  private static final String HEALTH_CHECK_TENANT_ID = "health-check";
  private static final String HEALTH_CHECK_AUTH_TOKEN = "";

  @Inject
  CQLSessionCache sessionCache;

  @Inject
  OperationsConfig operationsConfig;

  @Override
  public HealthCheckResponse call() {

    HealthCheckResponseBuilder responseBuilder = HealthCheckResponse.named("cassandra-connection");

    try {
      // Create a tenant for health check purposes
      Tenant healthCheckTenant = TenantFactory.instance().create(HEALTH_CHECK_TENANT_ID);
      
      // Create a minimal RequestContext for the health check
      RequestContext requestContext = new RequestContext(
          healthCheckTenant,
          HEALTH_CHECK_AUTH_TOKEN,
          new UserAgent("DataAPI-HealthCheck/1.0")
      );

      // Get or create a session - this will verify we can connect to Cassandra
      CqlSession session = sessionCache.getSession(requestContext)
          .await()
          .atMost(HEALTH_CHECK_TIMEOUT);

      // Verify the session is not closed
      if (session.isClosed()) {
        LOGGER.warn("Cassandra session is closed during health check");
        return responseBuilder
            .down()
            .withData("reason", "Session is closed")
            .build();
      }

      // Execute a simple query to verify connectivity and query execution
      SimpleStatement statement = SimpleStatement.builder(HEALTH_CHECK_QUERY)
          .setTimeout(HEALTH_CHECK_TIMEOUT)
          .setConsistencyLevel(operationsConfig.queriesConfig().consistency().reads())
          .build();

      var resultSet = session.execute(statement);
      var row = resultSet.one();
      
      // If we got here, the query succeeded
      String version = row != null ? row.getString("release_version") : "unknown";

      LOGGER.trace("Cassandra health check passed, version: {}", version);
      
      return responseBuilder
          .up()
          .withData("cassandra_version", version)
          .withData("session_name", session.getName())
          .build();

    } catch (Exception e) {
      LOGGER.error("Cassandra health check failed", e);
      return responseBuilder
          .down()
          .withData("error", e.getClass().getSimpleName())
          .withData("message", e.getMessage() != null ? e.getMessage() : "Unknown error")
          .build();
    }
  }
}

[amorton]

Do we have any artefacts of what logs look like what we get into this situation ? Checking system.local does not check we can get data from the cluster, a C* node could still be partitioned from other nodes and return this query because it is not replicated.

the health check has to happen in a security context from a request, we cannot but DB credentials into config or the code.

Also, above is using blocking IO, we cannot do that.