From 91f1061de398613bc7e1dcacd94c347b4d7675e6 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Thu, 30 Jul 2026 22:39:15 +0200 Subject: [PATCH] feat: populate the full DRIVER_CONFIG report (stage 2) Fills in the full DRIVER_CONFIG JSON report in the approved v2 cross-driver schema shape, replacing the stage-1 {"version":1} placeholder. All groups are populated from Configuration and Policies when the report is built, i.e. once per Cluster as it initializes. connection-pool.desired-connections-count reports the core connections per host rather than the maximum: HostConnectionPool opens core connections when it initializes and only grows towards the maximum under load, which is what the schema's "connections to open per host" means. Both protocol-v3 defaults are 1, so this only shows on a tuned pool -- where reporting the maximum was wrong in both directions, overstating a configured core/max pair and falling back to the v3 default when only the core count was set. Where a configured value falls outside what the schema can express, an optional key or group is omitted rather than emitted as a number the schema rejects: a disabled read timeout (connection.read and control-plane.system-queries.timeout.client-side-ms), a negative SO_LINGER, a non-positive socket buffer size. Omission is not always available -- connection.connect.timeout-ms and query-defaults.request.timeout-ms are required and positive-only, and query-defaults.consistency is an enum without the serial levels, all three reachable through unvalidated 3.x setters -- so those are reported as-is and the limitation is documented on the class: the reporter neither fabricates an in-range value, which would misreport a setting an operator may have chosen on purpose, nor drops the whole report over one field. Recorded as a cross-driver schema gap, to be fixed the way control-plane.schema-agreement.timeout-ms already admits 0. Adds public getters to DCAwareRoundRobinPolicy and RackAwareRoundRobinPolicy needed to report node-location-preference and dc-failover, and makes PagingOptimizingLoadBalancingPolicy implement ChainableLoadBalancingPolicy so the reporter can unwrap the LB policy Cluster.Manager wraps at runtime. Caps the report at 32KiB of UTF-8 (MAX_DRIVER_CONFIG_LENGTH), matching the 4.x sibling PR #968, gocql #964 and csharp-driver #262. Beyond cross-driver parity this is a correctness fix: CBUtil.writeString writes each STARTUP value with a 16-bit length prefix and no bounds check, so a value over 65535 bytes truncates the prefix modulo 65536 while still appending the whole body -- a corrupt frame and a failed handshake, and not something the fail-safe try/catch can contain since nothing throws. Parts of the report are user-supplied and unbounded (DC/rack names, consistency levels, custom policy class names). Over the limit means WARN and no DRIVER_CONFIG. Hardens the other two ways reporting could break a connection rather than merely fail to report: - The fail-safe catch also covers InternalError, since customPolicy() calls getClass().getSimpleName() on arbitrary user policy objects (documented JDK edge case for certain synthetic classes). Not a bare Error, so OutOfMemoryError/StackOverflowError still surface. - The load balancing policy chain walk is bounded at 16 policies and shared by both callers. It follows getChildPolicy() on arbitrary user policies, so a cyclic chain used to spin forever on the Cluster initialization path -- the one failure mode the try/catch cannot contain, because it hangs rather than throws. A custom load balancing policy is now named after the policy the user configured rather than PagingOptimizingLoadBalancingPolicy. Cluster .Manager wraps every session's policy in that internal class, and it is the outermost element of the chain, so every custom policy was reported as {"type":"custom","name":"PagingOptimizingLoadBalancingPolicy"}. Adds a JSON-Schema conformance test suite (mirroring the 4.x sibling PR #968): the normative schema is shipped as a test resource and validated via com.networknt:json-schema-validator (pinned to 1.5.x, the last line still targeting Java 8), covering every discriminated- union branch and optional group the 3.x reporter can emit, plus a negative test proving additionalProperties=false is enforced. Co-Authored-By: Claude Sonnet 5 Co-Authored-By: Claude Opus 5 (1M context) --- driver-core/pom.xml | 5 + .../core/DefaultDriverConfigReporter.java | 510 ++++++++++- .../driver/core/DriverConfigReporter.java | 5 +- .../policies/DCAwareRoundRobinPolicy.java | 35 + .../PagingOptimizingLoadBalancingPolicy.java | 7 +- .../policies/RackAwareRoundRobinPolicy.java | 58 ++ .../core/DefaultDriverConfigReporterTest.java | 759 +++++++++++++++- .../core/DriverConfigReportingCcmTest.java | 30 +- .../driver-config-report-v1.schema.json | 846 ++++++++++++++++++ pom.xml | 7 + 10 files changed, 2236 insertions(+), 26 deletions(-) create mode 100644 driver-core/src/test/resources/config/driver-config-report-v1.schema.json diff --git a/driver-core/pom.xml b/driver-core/pom.xml index 8af93c038de..998f0a05b5f 100644 --- a/driver-core/pom.xml +++ b/driver-core/pom.xml @@ -187,6 +187,11 @@ test + + com.networknt + json-schema-validator + test + diff --git a/driver-core/src/main/java/com/datastax/driver/core/DefaultDriverConfigReporter.java b/driver-core/src/main/java/com/datastax/driver/core/DefaultDriverConfigReporter.java index 30c01848608..f35ea7560b7 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/DefaultDriverConfigReporter.java +++ b/driver-core/src/main/java/com/datastax/driver/core/DefaultDriverConfigReporter.java @@ -15,9 +15,32 @@ */ package com.datastax.driver.core; +import com.datastax.driver.core.policies.ChainableLoadBalancingPolicy; +import com.datastax.driver.core.policies.ConstantReconnectionPolicy; +import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy; +import com.datastax.driver.core.policies.DefaultRetryPolicy; +import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy; +import com.datastax.driver.core.policies.ExponentialReconnectionPolicy; +import com.datastax.driver.core.policies.FallthroughRetryPolicy; +import com.datastax.driver.core.policies.HostFilterPolicy; +import com.datastax.driver.core.policies.LatencyAwarePolicy; +import com.datastax.driver.core.policies.LoadBalancingPolicy; +import com.datastax.driver.core.policies.NoSpeculativeExecutionPolicy; +import com.datastax.driver.core.policies.PagingOptimizingLoadBalancingPolicy; +import com.datastax.driver.core.policies.Policies; +import com.datastax.driver.core.policies.RackAwareRoundRobinPolicy; +import com.datastax.driver.core.policies.ReconnectionPolicy; +import com.datastax.driver.core.policies.RetryPolicy; +import com.datastax.driver.core.policies.RoundRobinPolicy; +import com.datastax.driver.core.policies.SpeculativeExecutionPolicy; +import com.datastax.driver.core.policies.TokenAwarePolicy; +import com.datastax.driver.core.policies.WhiteListPolicy; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -30,6 +53,31 @@ * reused for the lifetime of that {@code Cluster} — it is never rebuilt while the session is in * flight, so a control-connection reconnect costs nothing and always reports the same * configuration. + * + *

The report follows the approved v2 cross-driver schema: kebab-case keys, nested objects, and + * omission of any key or group that has no value (nothing is ever emitted as {@code null}). + * The same applies where a configured value falls outside what the schema can express but the key + * is optional: a disabled read timeout, a disabled {@code SO_LINGER} and an unbounded page + * size are omitted rather than emitted as a number the schema rejects. + * + *

Known limitation: that omission is not always available. A few schema fields are + * required and constrained to a positive integer, yet the 3.x option behind them accepts a + * value outside that range — and none of those setters validate their argument. {@link + * SocketOptions#setConnectTimeoutMillis(int)} and {@link SocketOptions#setReadTimeoutMillis(int)} + * both document a non-positive value as "no timeout", which lands in {@code + * connection.connect.timeout-ms} and {@code query-defaults.request.timeout-ms}; and {@code + * query-defaults.consistency} is an enum without the serial levels, which {@link + * QueryOptions#setConsistencyLevel(ConsistencyLevel)} nevertheless accepts. Such a value is + * reported as-is: the reporter deliberately neither fabricates an in-range value — which + * would misreport a setting an operator may have chosen on purpose — nor drops the whole report + * over one field, so the document is accurate but fails schema validation. Tracked as a + * cross-driver schema gap: the fix is to let those fields express the value, the way {@code + * control-plane.schema-agreement.timeout-ms} already admits 0. + * + *

One consequence of the two rules together: the read timeout feeds three fields, two of them + * optional, so disabling it omits {@code connection.read} and {@code + * control-plane.system-queries.timeout.client-side-ms} while {@code + * query-defaults.request.timeout-ms} still reports 0. */ public class DefaultDriverConfigReporter implements DriverConfigReporter { @@ -44,6 +92,38 @@ public class DefaultDriverConfigReporter implements DriverConfigReporter { */ static final int SCHEMA_VERSION = 1; + /** + * Upper bound on the UTF-8 size of the {@code DRIVER_CONFIG} value; a longer report is dropped + * rather than sent. + * + *

{@code STARTUP} options are serialized with {@code CBUtil.writeStringMap}, which writes each + * value with a 16-bit length prefix and no bounds check: a value longer than 65535 bytes would + * silently truncate that prefix modulo 65536 while still appending the whole body, corrupting the + * frame and failing the handshake. Note that nothing throws on that path, so it is not a failure + * the {@code try/catch} in {@link #buildReport()} could contain. + * + *

Most of this report is fixed-shape, but some of it is user-supplied and unbounded — + * datacenter and rack names, consistency levels, and the class names of custom policy objects — + * so enforcing a limit here keeps "reporting must never prevent a connection from being + * established" a property of this class rather than of the user's configuration. 32KiB is + * generous for a configuration report, and is the same limit the other ScyllaDB drivers apply. + */ + static final int MAX_DRIVER_CONFIG_LENGTH = 32 * 1024; + + /** + * Upper bound on the number of policies visited while walking a load balancing policy chain. + * + *

The walk follows {@code ChainableLoadBalancingPolicy.getChildPolicy()} on arbitrary + * user-supplied policy objects, so a policy that returns itself — or any cycle — would otherwise + * spin forever on the {@link Cluster} initialization path. That is the one failure mode the + * {@code try/catch} in {@link #buildReport()} cannot contain, because it hangs rather than + * throws. + * + *

The built-in chains are a handful of policies deep at most, so hitting this bound means a + * malformed chain rather than a legitimately deep one. + */ + private static final int MAX_POLICY_CHAIN_LENGTH = 16; + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); protected final Configuration configuration; @@ -57,10 +137,28 @@ public String buildReport() { // Configuration reporting is a best-effort diagnostic aid, so any failure here (a bad config // read, a misbehaving policy while introspecting, a serialization error) must be swallowed // rather than allowed to propagate: it is built on the Cluster-initialization path, which must - // not fail because of a diagnostic. + // not fail because of a diagnostic. Also catches InternalError specifically: customPolicy() + // calls getClass().getSimpleName() on arbitrary user-supplied policy objects, which has a + // documented JDK edge case throwing InternalError for certain synthetic classes. Deliberately + // not a bare `Error` — that would also swallow OutOfMemoryError/StackOverflowError, masking a + // real JVM-level failure instead of this one narrow, documented case. try { - return buildJson(); - } catch (RuntimeException e) { + String json = buildJson(); + if (json == null) { + return null; + } + // Measured on the encoded bytes, since that is what the length prefix on the wire counts. + int length = json.getBytes(StandardCharsets.UTF_8).length; + if (length > MAX_DRIVER_CONFIG_LENGTH) { + LOGGER.warn( + "The driver configuration report is {} bytes long, which exceeds the {} byte limit; " + + "skipping DRIVER_CONFIG", + length, + MAX_DRIVER_CONFIG_LENGTH); + return null; + } + return json; + } catch (InternalError | RuntimeException e) { LOGGER.warn( "Error while building the driver configuration report; skipping driver config reporting", e); @@ -68,12 +166,7 @@ public String buildReport() { } } - /** - * Builds the compact, single-line JSON configuration report. - * - *

Stage 1 emits only the schema {@code version}; the individual configuration groups are - * populated in {@link #populateConfig(ObjectNode)} in a later stage. - */ + /** Builds the compact, single-line JSON configuration report. */ protected String buildJson() { ObjectNode root = OBJECT_MAPPER.createObjectNode(); root.put("version", SCHEMA_VERSION); @@ -88,11 +181,402 @@ protected String buildJson() { } /** - * Populates the configuration groups onto the report root. Placeholder in Stage 1; Stage 2 fills - * in {@code connection}, {@code socket}, the policy groups, {@code query-defaults}, {@code tls}, - * etc. from {@link #configuration}. + * Populates the configuration groups onto the report root from {@link #configuration} and its + * policies, following the v2 schema. Keys the driver has no equivalent for (or cannot introspect + * in 3.x) are omitted rather than emitted as {@code null}. */ protected void populateConfig(ObjectNode root) { - // Stage 2: populate configuration groups from `configuration`. + Policies policies = configuration.getPolicies(); + // The load balancing policy chain feeds both load-balancing-policy and + // node-location-preference, so it is walked once here and handed to both. + LoadBalancingPolicy lbPolicy = policies.getLoadBalancingPolicy(); + List lbChain = policyChain(lbPolicy); + root.set("connection", connection()); + root.set("socket", socket()); + root.set("control-plane", controlPlane()); + root.set("reconnection-policy", reconnectionPolicy(policies)); + root.set("retry-policy", retryPolicy(policies)); + // speculative-execution-policy is optional: omitted when there is no speculative execution. + ObjectNode specEx = speculativeExecutionPolicy(policies); + if (specEx != null) { + root.set("speculative-execution-policy", specEx); + } + root.set("load-balancing-policy", loadBalancingPolicy(lbChain, lbPolicy)); + // node-location-preference is optional: omitted when the LB policy carries no DC/rack notion. + ObjectNode nodeLocation = nodeLocationPreference(lbChain); + if (nodeLocation != null) { + root.set("node-location-preference", nodeLocation); + } + root.set("connection-pool", connectionPool()); + root.set("query-defaults", queryDefaults(policies)); + root.set("tls", tls()); + } + + private ObjectNode connection() { + SocketOptions socketOptions = configuration.getSocketOptions(); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + // Required and positive-only in the schema, so a non-positive connect timeout (which disables + // it) is reported as-is for want of a valid representation — see the class javadoc. + n.set( + "connect", + OBJECT_MAPPER + .createObjectNode() + .put("timeout-ms", socketOptions.getConnectTimeoutMillis())); + // Optional group, positive-only: a non-positive read timeout disables read timeouts, so omit + // the group rather than report a number the schema rejects. + int readTimeoutMillis = socketOptions.getReadTimeoutMillis(); + if (readTimeoutMillis > 0) { + n.set("read", OBJECT_MAPPER.createObjectNode().put("timeout-ms", readTimeoutMillis)); + } + // No socket-level write timeout in 3.x -> omit "write". "heartbeat" is a reserved-empty + // placeholder in v2 (the heartbeat interval has no home this schema version) -> omit. + return n; + } + + private ObjectNode socket() { + SocketOptions o = configuration.getSocketOptions(); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + // Boolean options report the effective on/off state; when the driver leaves them unset the + // OS/platform default applies, approximated here (tcp-no-delay defaults on, the others off). + n.put("tcp-no-delay", boolOrDefault(o.getTcpNoDelay(), true)); + n.put("keep-alive", boolOrDefault(o.getKeepAlive(), false)); + n.put("reuse-address", boolOrDefault(o.getReuseAddress(), false)); + // All three groups below are optional, so a value the schema cannot express is omitted rather + // than emitted: a negative SO_LINGER means lingering close is disabled (the schema takes a + // non-negative interval, so 0 is still reported), and a non-positive buffer size leaves the + // JDK/OS default in place (the schema takes a positive size). + Integer soLinger = o.getSoLinger(); + if (soLinger != null && soLinger >= 0) { + n.set("linger", OBJECT_MAPPER.createObjectNode().put("interval-s", soLinger)); + } + Integer receiveBufferSize = o.getReceiveBufferSize(); + if (receiveBufferSize != null && receiveBufferSize > 0) { + n.set( + "receive-buffer", OBJECT_MAPPER.createObjectNode().put("size-bytes", receiveBufferSize)); + } + Integer sendBufferSize = o.getSendBufferSize(); + if (sendBufferSize != null && sendBufferSize > 0) { + n.set("send-buffer", OBJECT_MAPPER.createObjectNode().put("size-bytes", sendBufferSize)); + } + return n; + } + + private ObjectNode controlPlane() { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + // 3.x has no dedicated control-connection timeout; internal/system queries use the read + // timeout. + // There is no client-configurable server-side ("USING TIMEOUT") timeout -> omit server-side-ms. + ObjectNode timeout = OBJECT_MAPPER.createObjectNode(); + // Optional and positive-only, and a non-positive read timeout disables read timeouts: omit + // rather than report a number the schema rejects. The enclosing "timeout" object is required, + // so it stays (empty). + int clientSideMs = configuration.getSocketOptions().getReadTimeoutMillis(); + if (clientSideMs > 0) { + timeout.put("client-side-ms", clientSideMs); + } + n.set("system-queries", OBJECT_MAPPER.createObjectNode().set("timeout", timeout)); + // Required and non-negative in the schema, and 0 is meaningful (do not wait for agreement). + // Cluster.Builder rejects a non-positive wait, but ProtocolOptions can be constructed with one + // directly, and a negative wait behaves exactly like 0 — so normalizing it is exact rather than + // invented, and keeps the required field in range. + long schemaAgreementMs = + Math.max(0L, configuration.getProtocolOptions().getMaxSchemaAgreementWaitSeconds() * 1000L); + n.set( + "schema-agreement", OBJECT_MAPPER.createObjectNode().put("timeout-ms", schemaAgreementMs)); + return n; + } + + private ObjectNode reconnectionPolicy(Policies policies) { + ReconnectionPolicy policy = policies.getReconnectionPolicy(); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + if (policy instanceof ExponentialReconnectionPolicy) { + ExponentialReconnectionPolicy p = (ExponentialReconnectionPolicy) policy; + n.put("type", "exponential"); + n.put("base-ms", p.getBaseDelayMs()); + n.put("max-ms", p.getMaxDelayMs()); + // 3.x built-in reconnection policies are unbounded -> omit max-attempts. + } else if (policy instanceof ConstantReconnectionPolicy) { + n.put("type", "constant"); + n.put("delay-ms", ((ConstantReconnectionPolicy) policy).getConstantDelayMs()); + } else { + customPolicy(n, policy); + } + return n; + } + + private ObjectNode retryPolicy(Policies policies) { + RetryPolicy policy = policies.getRetryPolicy(); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + if (policy instanceof DefaultRetryPolicy) { + n.put("type", "standard-error-aware"); + } else if (policy instanceof DowngradingConsistencyRetryPolicy) { + n.put("type", "downgrading-consistency"); + } else if (policy instanceof FallthroughRetryPolicy) { + n.put("type", "fallthrough"); + } else { + // LoggingRetryPolicy / IdempotenceAwareRetryPolicy wrap a child but expose no getter, so only + // the outer type can be reported. + customPolicy(n, policy); + } + return n; + } + + private ObjectNode speculativeExecutionPolicy(Policies policies) { + SpeculativeExecutionPolicy policy = policies.getSpeculativeExecutionPolicy(); + if (policy instanceof NoSpeculativeExecutionPolicy) { + return null; + } + // Constant/Percentile parameters are not introspectable in 3.x (no getters), so a valid + // built-in object cannot be produced; report as custom, which at least surfaces the policy. + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + customPolicy(n, policy); + return n; + } + + private ObjectNode loadBalancingPolicy( + List chain, LoadBalancingPolicy policy) { + boolean tokenAware = false; + boolean latencyAware = false; + boolean whiteList = false; + boolean hostFilter = false; + DCAwareRoundRobinPolicy dcAware = null; + RackAwareRoundRobinPolicy rackAware = null; + boolean roundRobin = false; + // Policies with no normalized flag or type of their own in v1 leave no trace here: an + // ErrorAwarePolicy wrapping a DC-aware policy, say, is reported as plain "dc-aware". + for (LoadBalancingPolicy current : chain) { + if (current instanceof TokenAwarePolicy) { + tokenAware = true; + } else if (current instanceof LatencyAwarePolicy) { + latencyAware = true; + } else if (current instanceof WhiteListPolicy) { + whiteList = true; + } else if (current instanceof HostFilterPolicy) { + hostFilter = true; + } else if (current instanceof DCAwareRoundRobinPolicy) { + dcAware = (DCAwareRoundRobinPolicy) current; + } else if (current instanceof RackAwareRoundRobinPolicy) { + rackAware = (RackAwareRoundRobinPolicy) current; + } else if (current instanceof RoundRobinPolicy) { + roundRobin = true; + } + } + + String type; + if (tokenAware) { + type = "token-aware"; + } else if (dcAware != null) { + type = "dc-aware"; + } else if (rackAware != null) { + type = "rack-aware"; + } else if (whiteList) { + type = "white-list"; + } else if (hostFilter) { + type = "host-filter"; + } else if (roundRobin) { + type = "round-robin"; + } else { + type = null; + } + + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + if (type == null) { + customPolicy(n, configuredPolicy(chain, policy)); + return n; + } + n.put("type", type); + n.put("token-aware", tokenAware); + // 3.x has no replica-shuffle option / no getter for the token-aware replica ordering. + n.put("shuffle", false); + boolean dcFailover = + (dcAware != null && dcAware.getUsedHostsPerRemoteDc() > 0) + || (rackAware != null && rackAware.getUsedHostsPerRemoteDc() > 0); + n.put("dc-failover", dcFailover); + n.put("latency-awareness", latencyAware); + return n; + } + + private ObjectNode nodeLocationPreference(List chain) { + DCAwareRoundRobinPolicy dcAware = null; + RackAwareRoundRobinPolicy rackAware = null; + for (LoadBalancingPolicy current : chain) { + if (current instanceof DCAwareRoundRobinPolicy) { + dcAware = (DCAwareRoundRobinPolicy) current; + } else if (current instanceof RackAwareRoundRobinPolicy) { + rackAware = (RackAwareRoundRobinPolicy) current; + } + } + + if (rackAware != null) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + boolean explicit = rackAware.isLocalDcExplicit() && rackAware.isLocalRackExplicit(); + n.put("type", explicit ? "rack" : "rack-auto"); + putIfNotNull(n, "local-dc", rackAware.getLocalDc()); + putIfNotNull(n, "local-rack", rackAware.getLocalRack()); + return n; + } + if (dcAware != null) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + n.put("type", dcAware.isLocalDcExplicit() ? "dc" : "dc-auto"); + putIfNotNull(n, "local-dc", dcAware.getLocalDc()); + return n; + } + return null; + } + + private ObjectNode connectionPool() { + PoolingOptions pooling = configuration.getPoolingOptions(); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + // 3.x has a single per-host pool type; shard distribution is handled inside it. A per-shard + // count (what the schema's "shard" type carries) could not be reported anyway: this report is + // built before the control connection is up, so no node's shard count is known yet. + n.put("type", "host"); + // The core, not the maximum: HostConnectionPool opens core connections when it initializes and + // only grows towards the maximum under load, which is what "connections to open per host" + // means. Both v3 defaults are 1, so this is only distinguishable on a tuned pool. + n.put( + "desired-connections-count", + effective( + pooling.getCoreConnectionsPerHost(HostDistance.LOCAL), + v3PoolDefault(PoolingOptions.CORE_POOL_LOCAL_KEY))); + n.set( + "connection", + OBJECT_MAPPER + .createObjectNode() + .put( + "max-requests", + effective( + pooling.getMaxRequestsPerConnection(HostDistance.LOCAL), + v3PoolDefault(PoolingOptions.MAX_REQUESTS_PER_CONNECTION_LOCAL_KEY)))); + n.set( + "shard-aware", + OBJECT_MAPPER + .createObjectNode() + .put("enabled", configuration.getProtocolOptions().isUseAdvancedShardAwareness())); + return n; + } + + private ObjectNode queryDefaults(Policies policies) { + QueryOptions q = configuration.getQueryOptions(); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + // A non-positive fetch size means paging is unbounded, which the schema has no sentinel for, so + // this optional group is omitted entirely. + int fetchSize = q.getFetchSize(); + if (fetchSize > 0) { + n.set("page", OBJECT_MAPPER.createObjectNode().put("size", fetchSize)); + } + // Required, so a serial level — which the schema's enum does not admit but QueryOptions accepts + // — is reported as-is; see the class javadoc. + n.put("consistency", q.getConsistencyLevel().name()); + if (q.getSerialConsistencyLevel() != null) { + n.put("serial-consistency", q.getSerialConsistencyLevel().name()); + } + n.put("idempotence", q.getDefaultIdempotence()); + // Client-side timestamps are assigned unless the server-side generator is configured. + n.put( + "client-timestamps", + !(policies.getTimestampGenerator() instanceof ServerSideTimestampGenerator)); + // 3.x has no per-request timeout of its own: the read timeout bounds every request, so it is + // also what connection.read and control-plane.system-queries report. Required and positive-only + // here, though, so unlike those two a disabled read timeout is reported as-is rather than + // omitted — see the class javadoc. + n.set( + "request", + OBJECT_MAPPER + .createObjectNode() + .put("timeout-ms", configuration.getSocketOptions().getReadTimeoutMillis())); + return n; + } + + private ObjectNode tls() { + SSLOptions sslOptions = configuration.getProtocolOptions().getSSLOptions(); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + n.put("enabled", sslOptions != null); + // 3.x exposes no hostname-verification flag; SniSSLOptions hard-codes it on, otherwise it is + // not + // introspectable and reported as false (best-effort; the schema requires a boolean). + n.put("hostname-verification", sslOptions instanceof SniSSLOptions); + return n; + } + + /** + * Returns the load balancing policy chain, outermost policy first, by following {@code + * ChainableLoadBalancingPolicy.getChildPolicy()} for at most {@link #MAX_POLICY_CHAIN_LENGTH} + * policies. Both the {@code load-balancing-policy} and the {@code node-location-preference} + * groups are derived from this chain, since {@code Cluster.Manager} wraps the configured policy + * at runtime. + */ + private static List policyChain(LoadBalancingPolicy policy) { + List chain = new ArrayList(); + LoadBalancingPolicy current = policy; + while (current != null && chain.size() < MAX_POLICY_CHAIN_LENGTH) { + chain.add(current); + current = + current instanceof ChainableLoadBalancingPolicy + ? ((ChainableLoadBalancingPolicy) current).getChildPolicy() + : null; + } + if (current != null) { + // Only reachable from a user policy whose getChildPolicy() chain is cyclic or absurdly deep; + // report what was seen rather than walking forever or dropping the whole report. + LOGGER.warn( + "Stopped walking the load balancing policy chain after {} policies; reporting only those. " + + "Does a ChainableLoadBalancingPolicy in the chain return a cyclic child policy?", + MAX_POLICY_CHAIN_LENGTH); + } + return chain; + } + + /** + * The outermost policy of {@code chain} that the user actually configured, i.e. the first one + * that is not the internal {@code PagingOptimizingLoadBalancingPolicy} wrapper {@code + * Cluster.Manager} puts around every session's policy. Without this, every custom policy would be + * reported under that wrapper's name, which tells an operator nothing about what the client runs. + * + *

The outermost is preferred over the innermost because it is what the user handed to the + * builder; a custom policy chaining to further custom policies is described by its outer one. + */ + private static LoadBalancingPolicy configuredPolicy( + List chain, LoadBalancingPolicy fallback) { + for (LoadBalancingPolicy policy : chain) { + if (!(policy instanceof PagingOptimizingLoadBalancingPolicy)) { + return policy; + } + } + return fallback; + } + + private static void customPolicy(ObjectNode node, Object policy) { + node.put("type", "custom"); + node.put("name", policy.getClass().getSimpleName()); + } + + private static void putIfNotNull(ObjectNode node, String key, String value) { + if (value != null) { + node.put(key, value); + } + } + + private static boolean boolOrDefault(Boolean value, boolean defaultValue) { + return value == null ? defaultValue : value; + } + + /** + * The effective per-host pool default for protocol v3+ (which ScyllaDB always negotiates). Needed + * as a fallback because {@link PoolingOptions} returns {@link PoolingOptions#UNSET} until the + * protocol version is known, which only happens once the control connection is up — after this + * report is built. A value the user configured explicitly takes precedence. + * + *

Looked up here rather than in a static field so that a failure stays inside {@link + * #buildReport()}'s fail-safe handling instead of breaking class initialization on the {@link + * Connection.Factory} path. + */ + private static int v3PoolDefault(String key) { + return PoolingOptions.DEFAULTS.get(ProtocolVersion.V3).get(key); + } + + /** {@code value} unless it is unset ({@link PoolingOptions#UNSET} is negative) or nonsensical. */ + private static int effective(int value, int defaultValue) { + return value <= 0 ? defaultValue : value; } } diff --git a/driver-core/src/main/java/com/datastax/driver/core/DriverConfigReporter.java b/driver-core/src/main/java/com/datastax/driver/core/DriverConfigReporter.java index 7ada1c34a2b..c50ba315484 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/DriverConfigReporter.java +++ b/driver-core/src/main/java/com/datastax/driver/core/DriverConfigReporter.java @@ -39,7 +39,10 @@ public interface DriverConfigReporter { * *

Implementations must not throw: a failure to build the report must be swallowed (and * logged) rather than propagated, so that a diagnostic aid can never break cluster - * initialization. + * initialization. For the same reason they must return {@code null} rather than a report that + * exceeds {@code DefaultDriverConfigReporter.MAX_DRIVER_CONFIG_LENGTH}: {@code STARTUP} option + * values carry an unchecked 16-bit length prefix, so an oversized one corrupts the frame instead + * of merely being useless. * * @return the report to send under the {@code DRIVER_CONFIG} startup option, or {@code null} to * send nothing. diff --git a/driver-core/src/main/java/com/datastax/driver/core/policies/DCAwareRoundRobinPolicy.java b/driver-core/src/main/java/com/datastax/driver/core/policies/DCAwareRoundRobinPolicy.java index a1274f6b458..5882bf587e5 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/policies/DCAwareRoundRobinPolicy.java +++ b/driver-core/src/main/java/com/datastax/driver/core/policies/DCAwareRoundRobinPolicy.java @@ -77,6 +77,7 @@ public static Builder builder() { private final int usedHostsPerRemoteDc; private final boolean dontHopForLocalCL; + private final boolean localDcExplicit; private volatile Configuration configuration; @@ -90,6 +91,40 @@ private DCAwareRoundRobinPolicy( this.localDc = localDc == null ? UNSET : localDc; this.usedHostsPerRemoteDc = usedHostsPerRemoteDc; this.dontHopForLocalCL = !allowRemoteDCsForLocalConsistencyLevel; + this.localDcExplicit = !Strings.isNullOrEmpty(localDc); + } + + /** + * The datacenter this policy considers local, or {@code null} if it has neither been configured + * explicitly nor inferred yet. When {@link #isLocalDcExplicit()} is {@code false}, this is the + * datacenter inferred from the first contacted node, which is only available once the policy has + * been initialized. + * + * @return the local datacenter name, or {@code null}. + */ + public String getLocalDc() { + String dc = localDc; + return Strings.isNullOrEmpty(dc) ? null : dc; + } + + /** + * Whether the local datacenter was configured explicitly (as opposed to being inferred from the + * first contacted node). + * + * @return {@code true} if the local datacenter was set explicitly. + */ + public boolean isLocalDcExplicit() { + return localDcExplicit; + } + + /** + * The number of hosts per remote datacenter that this policy considers for failover (0 means no + * remote failover). + * + * @return the number of used hosts per remote datacenter. + */ + public int getUsedHostsPerRemoteDc() { + return usedHostsPerRemoteDc; } @Override diff --git a/driver-core/src/main/java/com/datastax/driver/core/policies/PagingOptimizingLoadBalancingPolicy.java b/driver-core/src/main/java/com/datastax/driver/core/policies/PagingOptimizingLoadBalancingPolicy.java index edfe690900e..2096e3ad914 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/policies/PagingOptimizingLoadBalancingPolicy.java +++ b/driver-core/src/main/java/com/datastax/driver/core/policies/PagingOptimizingLoadBalancingPolicy.java @@ -12,7 +12,7 @@ import java.util.Iterator; import java.util.concurrent.CopyOnWriteArrayList; -public class PagingOptimizingLoadBalancingPolicy implements LoadBalancingPolicy { +public class PagingOptimizingLoadBalancingPolicy implements ChainableLoadBalancingPolicy { private final LoadBalancingPolicy wrapped; private volatile CopyOnWriteArrayList hosts; @@ -20,6 +20,11 @@ public PagingOptimizingLoadBalancingPolicy(LoadBalancingPolicy loadBalancingPoli wrapped = loadBalancingPolicy; } + @Override + public LoadBalancingPolicy getChildPolicy() { + return wrapped; + } + @Override public void init(Cluster cluster, Collection hosts) { this.hosts = new CopyOnWriteArrayList(hosts); diff --git a/driver-core/src/main/java/com/datastax/driver/core/policies/RackAwareRoundRobinPolicy.java b/driver-core/src/main/java/com/datastax/driver/core/policies/RackAwareRoundRobinPolicy.java index ab3d019cbb8..7f3b5860f8b 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/policies/RackAwareRoundRobinPolicy.java +++ b/driver-core/src/main/java/com/datastax/driver/core/policies/RackAwareRoundRobinPolicy.java @@ -90,6 +90,8 @@ public static Builder builder() { private final int usedHostsPerRemoteDc; private final boolean dontHopForLocalCL; + private final boolean localDcExplicit; + private final boolean localRackExplicit; private volatile Configuration configuration; @@ -109,6 +111,62 @@ public RackAwareRoundRobinPolicy( this.localRack = localRack == null ? UNSET : localRack; this.usedHostsPerRemoteDc = usedHostsPerRemoteDc; this.dontHopForLocalCL = !allowRemoteDCsForLocalConsistencyLevel; + this.localDcExplicit = !Strings.isNullOrEmpty(localDc); + this.localRackExplicit = !Strings.isNullOrEmpty(localRack); + } + + /** + * The datacenter this policy considers local, or {@code null} if it has neither been configured + * explicitly nor inferred yet. When {@link #isLocalDcExplicit()} is {@code false}, this is the + * datacenter inferred from the first contacted node, which is only available once the policy has + * been initialized. + * + * @return the local datacenter name, or {@code null}. + */ + public String getLocalDc() { + String dc = localDc; + return Strings.isNullOrEmpty(dc) ? null : dc; + } + + /** + * The rack this policy considers local, or {@code null} if it has neither been configured + * explicitly nor inferred yet. + * + * @return the local rack name, or {@code null}. + */ + public String getLocalRack() { + String rack = localRack; + return Strings.isNullOrEmpty(rack) ? null : rack; + } + + /** + * Whether the local datacenter was configured explicitly (as opposed to being inferred from the + * first contacted node). + * + * @return {@code true} if the local datacenter was set explicitly. + */ + public boolean isLocalDcExplicit() { + return localDcExplicit; + } + + /** + * Whether the local rack was configured explicitly (as opposed to being inferred from the first + * contacted node). + * + * @return {@code true} if the local rack was set explicitly. + */ + public boolean isLocalRackExplicit() { + return localRackExplicit; + } + + /** + * The number of hosts per remote datacenter that this policy considers for failover (0 means no + * remote failover). + * + * @return the number of used hosts per remote datacenter. + */ + public int getUsedHostsPerRemoteDc() { + return usedHostsPerRemoteDc; } @Override diff --git a/driver-core/src/test/java/com/datastax/driver/core/DefaultDriverConfigReporterTest.java b/driver-core/src/test/java/com/datastax/driver/core/DefaultDriverConfigReporterTest.java index cabc6a591a5..6b791daba99 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/DefaultDriverConfigReporterTest.java +++ b/driver-core/src/test/java/com/datastax/driver/core/DefaultDriverConfigReporterTest.java @@ -17,10 +17,65 @@ import static org.assertj.core.api.Assertions.assertThat; +import com.datastax.driver.core.policies.ConstantReconnectionPolicy; +import com.datastax.driver.core.policies.ConstantSpeculativeExecutionPolicy; +import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy; +import com.datastax.driver.core.policies.DelegatingLoadBalancingPolicy; +import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy; +import com.datastax.driver.core.policies.FallthroughRetryPolicy; +import com.datastax.driver.core.policies.HostFilterPolicy; +import com.datastax.driver.core.policies.LatencyAwarePolicy; +import com.datastax.driver.core.policies.LoadBalancingPolicy; +import com.datastax.driver.core.policies.LoggingRetryPolicy; +import com.datastax.driver.core.policies.PagingOptimizingLoadBalancingPolicy; +import com.datastax.driver.core.policies.RackAwareRoundRobinPolicy; +import com.datastax.driver.core.policies.ReconnectionPolicy; +import com.datastax.driver.core.policies.RoundRobinPolicy; +import com.datastax.driver.core.policies.TokenAwarePolicy; +import com.datastax.driver.core.policies.WhiteListPolicy; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.networknt.schema.JsonSchema; +import com.networknt.schema.JsonSchemaFactory; +import com.networknt.schema.SpecVersion; +import com.networknt.schema.ValidationMessage; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.Collection; +import java.util.Collections; +import java.util.Iterator; +import java.util.Set; import org.testng.annotations.Test; public class DefaultDriverConfigReporterTest { + private static final ObjectMapper MAPPER = new ObjectMapper(); + + // The normative v1 JSON Schema from the design doc, shipped verbatim as a test resource. Loaded + // once and pinned to draft 2020-12 (its declared $schema); its internal "#/$defs/..." refs + // resolve locally, so validation needs no network access. + // + // "v1" and the "v2 schema" the reporter's javadoc mentions are the same artifact: the report's + // "version" field (and this document) is at 1, while v2 is the revision of the design doc that + // defines that shape. + private static final JsonSchema SCHEMA = loadSchema(); + + private static JsonSchema loadSchema() { + try (InputStream in = + DefaultDriverConfigReporterTest.class.getResourceAsStream( + "/config/driver-config-report-v1.schema.json")) { + return JsonSchemaFactory.getInstance(SpecVersion.VersionFlag.V202012) + .getSchema(MAPPER.readTree(in)); + } catch (Exception e) { + throw new AssertionError("Cannot load the DRIVER_CONFIG v1 JSON Schema resource", e); + } + } + + // ---- Stage 1: default / fail-safe ----------------------------------------------------------- + private static Configuration config() { return Configuration.builder().build(); } @@ -37,10 +92,11 @@ public void should_enable_driver_config_reporting_by_default() { } @Test(groups = "unit") - public void should_report_schema_version() { - // Stage 1 emits only the schema version. - assertThat(new DefaultDriverConfigReporter(config()).buildReport()) - .isEqualTo("{\"version\":1}"); + public void should_report_schema_version_and_config_groups() throws Exception { + JsonNode report = MAPPER.readTree(new DefaultDriverConfigReporter(config()).buildReport()); + + assertThat(report.path("version").asInt()).isEqualTo(1); + assertThat(report.has("connection")).isTrue(); } @Test(groups = "unit") @@ -57,4 +113,699 @@ protected String buildJson() { // no DRIVER_CONFIG option is sent, and nothing else about the connection is affected. assertThat(reporter.buildReport()).isNull(); } + + @Test(groups = "unit") + public void should_be_fail_safe_when_report_build_throws_internal_error() { + DefaultDriverConfigReporter reporter = + new DefaultDriverConfigReporter(config()) { + @Override + protected String buildJson() { + // customPolicy() calls getClass().getSimpleName() on arbitrary user-supplied policy + // objects, which has a documented JDK edge case throwing InternalError for certain + // synthetic classes. + throw new InternalError("simulated getSimpleName() JDK edge case"); + } + }; + + assertThat(reporter.buildReport()).isNull(); + } + + @Test(groups = "unit") + public void should_skip_driver_config_when_it_exceeds_the_size_limit() { + // STARTUP option values are written with an unchecked 16-bit length prefix, so an oversized + // report would corrupt the frame and fail the handshake rather than merely be useless. Parts of + // the report come from unbounded user-supplied values (DC/rack names, consistency levels, + // custom policy class names), so the limit has to be enforced here. + assertThat(reporting(oversizedReport()).buildReport()).isNull(); + } + + @Test(groups = "unit") + public void should_return_driver_config_that_is_just_within_the_size_limit() { + String atLimit = padTo(DefaultDriverConfigReporter.MAX_DRIVER_CONFIG_LENGTH); + + assertThat(reporting(atLimit).buildReport()).isEqualTo(atLimit); + } + + @Test(groups = "unit") + public void should_report_default_configuration_within_the_size_limit() { + // Tripwire: the real report is nowhere near the limit today. If it ever grows past it, this + // fails loudly instead of DRIVER_CONFIG silently disappearing from the wire. + String report = + new DefaultDriverConfigReporter(Cluster.builder().getConfiguration()).buildReport(); + + assertThat(report).isNotNull(); + assertThat(report.getBytes(StandardCharsets.UTF_8).length) + .isLessThanOrEqualTo(DefaultDriverConfigReporter.MAX_DRIVER_CONFIG_LENGTH); + } + + @Test(groups = "unit", timeOut = 30000) + public void should_not_follow_a_cyclic_load_balancing_policy_chain_forever() throws Exception { + // getChildPolicy() is walked on arbitrary user policies, so a cyclic chain would spin forever + // on the cluster-initialization path. That is the one failure mode the reporter's try/catch + // cannot contain, since it hangs rather than throws. The walk is bounded, so the report is + // still produced, describing the outermost policy. + JsonNode report = + report(Cluster.builder().withLoadBalancingPolicy(new CyclicLoadBalancingPolicy())); + + assertThat(report.path("load-balancing-policy").path("type").asText()).isEqualTo("custom"); + assertConformsToSchema(report); + } + + @Test(groups = "unit") + public void should_name_the_configured_policy_not_the_internal_wrapper() throws Exception { + // Cluster.Manager wraps every configured policy in PagingOptimizingLoadBalancingPolicy, so + // naming the outermost policy of the chain would report that internal wrapper for every custom + // policy, telling an operator nothing about what the client actually runs. + JsonNode report = + report( + Cluster.builder() + .withLoadBalancingPolicy( + new PagingOptimizingLoadBalancingPolicy(new CustomLoadBalancingPolicy()))); + + JsonNode policy = report.path("load-balancing-policy"); + assertThat(policy.path("type").asText()).isEqualTo("custom"); + assertThat(policy.path("name").asText()).isEqualTo("CustomLoadBalancingPolicy"); + assertConformsToSchema(report); + } + + // ---- Stage 2: full report (v2 shape) -------------------------------------------------------- + + @Test(groups = "unit") + public void should_report_default_configuration_shape() throws Exception { + JsonNode report = report(Cluster.builder()); + + assertThat(report.path("version").asInt()).isEqualTo(1); + + // connection: connect + read only (no write timeout, no heartbeat in v2). + JsonNode connection = report.path("connection"); + assertThat(connection.path("connect").path("timeout-ms").asInt()).isEqualTo(5000); + assertThat(connection.path("read").path("timeout-ms").asInt()).isEqualTo(12000); + assertThat(connection.has("write")).isFalse(); + assertThat(connection.has("heartbeat")).isFalse(); + + // socket: booleans present; buffers/linger omitted when unset. + JsonNode socket = report.path("socket"); + assertThat(socket.path("tcp-no-delay").asBoolean()).isTrue(); + assertThat(socket.path("keep-alive").asBoolean()).isFalse(); + assertThat(socket.path("reuse-address").asBoolean()).isFalse(); + assertThat(socket.has("linger")).isFalse(); + assertThat(socket.has("receive-buffer")).isFalse(); + assertThat(socket.has("send-buffer")).isFalse(); + + // control-plane. + JsonNode controlPlane = report.path("control-plane"); + assertThat(controlPlane.path("system-queries").path("timeout").path("client-side-ms").asInt()) + .isEqualTo(12000); + assertThat(controlPlane.path("system-queries").path("timeout").has("server-side-ms")).isFalse(); + assertThat(controlPlane.path("schema-agreement").path("timeout-ms").asInt()).isEqualTo(10000); + + // reconnection: exponential, unbounded (no max-attempts). + JsonNode reconnection = report.path("reconnection-policy"); + assertThat(reconnection.path("type").asText()).isEqualTo("exponential"); + assertThat(reconnection.path("base-ms").asInt()).isEqualTo(1000); + assertThat(reconnection.path("max-ms").asInt()).isEqualTo(600000); + assertThat(reconnection.has("max-attempts")).isFalse(); + + // retry / speculative execution. + assertThat(report.path("retry-policy").path("type").asText()).isEqualTo("standard-error-aware"); + assertThat(report.has("speculative-execution-policy")).isFalse(); + + // load balancing: default is token-aware over DC-aware (auto DC). + JsonNode lb = report.path("load-balancing-policy"); + assertThat(lb.path("type").asText()).isEqualTo("token-aware"); + assertThat(lb.path("token-aware").asBoolean()).isTrue(); + assertThat(lb.path("shuffle").asBoolean()).isFalse(); + assertThat(lb.path("dc-failover").asBoolean()).isFalse(); + assertThat(lb.path("latency-awareness").asBoolean()).isFalse(); + + // node-location-preference: inferred DC, not yet resolved. + JsonNode nodeLocation = report.path("node-location-preference"); + assertThat(nodeLocation.path("type").asText()).isEqualTo("dc-auto"); + assertThat(nodeLocation.has("local-dc")).isFalse(); + + // connection pool (effective v3+ defaults; shard-aware on by default via Cluster.Builder). + JsonNode pool = report.path("connection-pool"); + assertThat(pool.path("type").asText()).isEqualTo("host"); + assertThat(pool.path("desired-connections-count").asInt()).isEqualTo(1); + assertThat(pool.path("connection").path("max-requests").asInt()).isEqualTo(1024); + assertThat(pool.path("shard-aware").path("enabled").asBoolean()).isTrue(); + + // query defaults. + JsonNode query = report.path("query-defaults"); + assertThat(query.path("page").path("size").asInt()).isEqualTo(5000); + assertThat(query.path("consistency").asText()).isEqualTo("LOCAL_ONE"); + assertThat(query.path("serial-consistency").asText()).isEqualTo("SERIAL"); + assertThat(query.path("idempotence").asBoolean()).isFalse(); + assertThat(query.path("client-timestamps").asBoolean()).isTrue(); + assertThat(query.path("request").path("timeout-ms").asInt()).isEqualTo(12000); + + // tls off. + assertThat(report.path("tls").path("enabled").asBoolean()).isFalse(); + assertThat(report.path("tls").path("hostname-verification").asBoolean()).isFalse(); + } + + @Test(groups = "unit") + public void should_report_constant_reconnection_policy() throws Exception { + JsonNode report = + report(Cluster.builder().withReconnectionPolicy(new ConstantReconnectionPolicy(2500))); + + JsonNode reconnection = report.path("reconnection-policy"); + assertThat(reconnection.path("type").asText()).isEqualTo("constant"); + assertThat(reconnection.path("delay-ms").asInt()).isEqualTo(2500); + assertThat(reconnection.has("max-attempts")).isFalse(); + } + + @Test(groups = "unit") + public void should_discriminate_retry_policies() throws Exception { + assertThat( + report(Cluster.builder().withRetryPolicy(FallthroughRetryPolicy.INSTANCE)) + .path("retry-policy") + .path("type") + .asText()) + .isEqualTo("fallthrough"); + assertThat( + report(Cluster.builder().withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE)) + .path("retry-policy") + .path("type") + .asText()) + .isEqualTo("downgrading-consistency"); + } + + @Test(groups = "unit") + public void should_report_configured_speculative_execution_as_custom() throws Exception { + JsonNode report = + report( + Cluster.builder() + .withSpeculativeExecutionPolicy(new ConstantSpeculativeExecutionPolicy(100L, 2))); + + JsonNode specEx = report.path("speculative-execution-policy"); + assertThat(specEx.isMissingNode()).isFalse(); + // Parameters are not introspectable in 3.x, so it is surfaced as custom with the class name. + assertThat(specEx.path("type").asText()).isEqualTo("custom"); + assertThat(specEx.path("name").asText()).isEqualTo("ConstantSpeculativeExecutionPolicy"); + } + + @Test(groups = "unit") + public void should_report_explicit_datacenter_node_location_preference() throws Exception { + JsonNode report = + report( + Cluster.builder() + .withLoadBalancingPolicy( + new TokenAwarePolicy( + DCAwareRoundRobinPolicy.builder().withLocalDc("dc1").build()))); + + JsonNode nodeLocation = report.path("node-location-preference"); + assertThat(nodeLocation.path("type").asText()).isEqualTo("dc"); + assertThat(nodeLocation.path("local-dc").asText()).isEqualTo("dc1"); + assertThat(nodeLocation.has("local-rack")).isFalse(); + // token-aware wrapper is still reflected in the LB group. + assertThat(report.path("load-balancing-policy").path("type").asText()).isEqualTo("token-aware"); + } + + @Test(groups = "unit") + public void should_unwrap_paging_optimizing_load_balancing_policy() throws Exception { + // At runtime Cluster.Manager wraps the configured LB policy in a + // PagingOptimizingLoadBalancingPolicy, so the reporter must unwrap it to recover the real + // policy's flags and location preference (rather than reporting it as a custom policy). + JsonNode report = + report( + Cluster.builder() + .withLoadBalancingPolicy( + new PagingOptimizingLoadBalancingPolicy( + new TokenAwarePolicy( + DCAwareRoundRobinPolicy.builder().withLocalDc("dc1").build())))); + + assertThat(report.path("load-balancing-policy").path("type").asText()).isEqualTo("token-aware"); + assertThat(report.path("load-balancing-policy").path("token-aware").asBoolean()).isTrue(); + JsonNode nodeLocation = report.path("node-location-preference"); + assertThat(nodeLocation.path("type").asText()).isEqualTo("dc"); + assertThat(nodeLocation.path("local-dc").asText()).isEqualTo("dc1"); + } + + @Test(groups = "unit") + public void should_report_rack_node_location_preference() throws Exception { + JsonNode report = + report( + Cluster.builder() + .withLoadBalancingPolicy( + new RackAwareRoundRobinPolicy("dc1", "rack1", 0, false, false, false))); + + JsonNode nodeLocation = report.path("node-location-preference"); + assertThat(nodeLocation.path("type").asText()).isEqualTo("rack"); + assertThat(nodeLocation.path("local-dc").asText()).isEqualTo("dc1"); + assertThat(nodeLocation.path("local-rack").asText()).isEqualTo("rack1"); + assertThat(report.path("load-balancing-policy").path("type").asText()).isEqualTo("rack-aware"); + } + + @Test(groups = "unit") + public void should_report_server_side_timestamps_as_disabled_client_timestamps() + throws Exception { + JsonNode report = + report(Cluster.builder().withTimestampGenerator(ServerSideTimestampGenerator.INSTANCE)); + + assertThat(report.path("query-defaults").path("client-timestamps").asBoolean()).isFalse(); + } + + @Test(groups = "unit") + public void should_report_tls_enabled() throws Exception { + JsonNode report = report(Cluster.builder().withSSL()); + + assertThat(report.path("tls").path("enabled").asBoolean()).isTrue(); + // Default JDK SSL options expose no hostname-verification flag -> reported false. + assertThat(report.path("tls").path("hostname-verification").asBoolean()).isFalse(); + } + + @Test(groups = "unit") + public void should_report_socket_overrides() throws Exception { + SocketOptions socketOptions = + new SocketOptions() + .setKeepAlive(true) + .setReuseAddress(true) + .setSoLinger(15) + .setReceiveBufferSize(4096) + .setSendBufferSize(8192); + + JsonNode socket = report(Cluster.builder().withSocketOptions(socketOptions)).path("socket"); + + assertThat(socket.path("keep-alive").asBoolean()).isTrue(); + assertThat(socket.path("reuse-address").asBoolean()).isTrue(); + assertThat(socket.path("linger").path("interval-s").asInt()).isEqualTo(15); + assertThat(socket.path("receive-buffer").path("size-bytes").asInt()).isEqualTo(4096); + assertThat(socket.path("send-buffer").path("size-bytes").asInt()).isEqualTo(8192); + } + + @Test(groups = "unit") + public void should_report_dc_failover_when_remote_hosts_are_used() throws Exception { + JsonNode dcAware = + report( + Cluster.builder() + .withLoadBalancingPolicy( + new TokenAwarePolicy( + DCAwareRoundRobinPolicy.builder() + .withLocalDc("dc1") + .withUsedHostsPerRemoteDc(2) + .build()))); + + assertThat(dcAware.path("load-balancing-policy").path("dc-failover").asBoolean()).isTrue(); + assertConformsToSchema(dcAware); + + // The same flag is derived from either locality-aware policy, so both getters are exercised. + JsonNode rackAware = + report( + Cluster.builder() + .withLoadBalancingPolicy( + new RackAwareRoundRobinPolicy("dc1", "rack1", 2, false, false, false))); + + assertThat(rackAware.path("load-balancing-policy").path("dc-failover").asBoolean()).isTrue(); + assertConformsToSchema(rackAware); + } + + @Test(groups = "unit") + public void should_report_the_core_pool_size_rather_than_the_maximum() throws Exception { + // HostConnectionPool opens core connections when it initializes and only grows towards the + // maximum under load, so the core count is what "connections to open per host" means. + JsonNode pool = + report( + Cluster.builder() + .withPoolingOptions( + new PoolingOptions() + .setConnectionsPerHost(HostDistance.LOCAL, 2, 8) + .setMaxRequestsPerConnection(HostDistance.LOCAL, 512))) + .path("connection-pool"); + + assertThat(pool.path("desired-connections-count").asInt()).isEqualTo(2); + assertThat(pool.path("connection").path("max-requests").asInt()).isEqualTo(512); + } + + @Test(groups = "unit") + public void should_report_the_core_pool_size_when_the_maximum_is_left_unset() throws Exception { + // Configuring only the core count is legal (the max stays UNSET until protocol negotiation), + // and + // the pool will open that many connections — so the v3 fallback must not take over here. + JsonNode pool = + report( + Cluster.builder() + .withPoolingOptions( + new PoolingOptions().setCoreConnectionsPerHost(HostDistance.LOCAL, 4))) + .path("connection-pool"); + + assertThat(pool.path("desired-connections-count").asInt()).isEqualTo(4); + } + + @Test(groups = "unit") + public void should_omit_a_disabled_read_timeout_but_report_the_request_timeout_as_is() + throws Exception { + // A non-positive read timeout disables read timeouts. It feeds three schema fields: the two + // optional ones are omitted rather than emitted as a value the schema rejects, while + // query-defaults.request.timeout-ms is required and positive-only, so it is reported as-is. + JsonNode report = + report(Cluster.builder().withSocketOptions(new SocketOptions().setReadTimeoutMillis(0))); + + assertThat(report.path("connection").has("read")).isFalse(); + JsonNode timeout = report.path("control-plane").path("system-queries").path("timeout"); + assertThat(timeout.isObject()).isTrue(); + assertThat(timeout.has("client-side-ms")).isFalse(); + assertThat(report.path("query-defaults").path("request").path("timeout-ms").asInt()) + .isEqualTo(0); + + // Pins the known limitation: that one required field is the only thing keeping this report from + // validating. If the schema gains a way to express "disabled", this test says where to look. + Set errors = SCHEMA.validate(report); + assertThat(errors).hasSize(1); + assertThat(errors.iterator().next().getMessage()).contains("query-defaults.request.timeout-ms"); + } + + @Test(groups = "unit") + public void should_omit_socket_values_the_schema_cannot_express() throws Exception { + // A negative SO_LINGER disables lingering close and a non-positive buffer size leaves the + // JDK/OS default in place; all three groups are optional, so they are omitted. + JsonNode socket = + report( + Cluster.builder() + .withSocketOptions( + new SocketOptions() + .setSoLinger(-1) + .setReceiveBufferSize(0) + .setSendBufferSize(-1))) + .path("socket"); + + assertThat(socket.has("linger")).isFalse(); + assertThat(socket.has("receive-buffer")).isFalse(); + assertThat(socket.has("send-buffer")).isFalse(); + } + + @Test(groups = "unit") + public void should_report_a_zero_linger_interval() throws Exception { + // 0 means "close immediately, discarding unsent data", which the schema does admit. + JsonNode socket = + report(Cluster.builder().withSocketOptions(new SocketOptions().setSoLinger(0))) + .path("socket"); + + assertThat(socket.path("linger").path("interval-s").asInt()).isEqualTo(0); + } + + // ==================== Schema conformance ==================== + // + // These build a config, serialize it via the reporter, and validate the produced JSON against the + // normative v1 JSON Schema (the same document ScyllaDB uses to interpret DRIVER_CONFIG). They + // cover every discriminated-union branch and optional-group case the 3.x reporter can emit, + // turning the "every emitted document is schema-valid" invariant into an enforced test. + // + // The one documented exception is a required field whose 3.x option accepts a value the schema + // cannot express; see the reporter's class javadoc and + // should_omit_a_disabled_read_timeout_but_report_the_request_timeout_as_is, which pins it. + + @Test(groups = "unit") + public void should_conform_to_schema_for_default_report() throws Exception { + assertConformsToSchema(report(Cluster.builder())); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_constant_reconnection_policy() throws Exception { + assertConformsToSchema( + report(Cluster.builder().withReconnectionPolicy(new ConstantReconnectionPolicy(2500)))); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_custom_reconnection_policy() throws Exception { + assertConformsToSchema( + report(Cluster.builder().withReconnectionPolicy(new CustomReconnectionPolicy()))); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_custom_retry_policy() throws Exception { + assertConformsToSchema( + report( + Cluster.builder() + .withRetryPolicy(new LoggingRetryPolicy(FallthroughRetryPolicy.INSTANCE)))); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_downgrading_consistency_retry_policy() throws Exception { + assertConformsToSchema( + report(Cluster.builder().withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE))); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_custom_speculative_execution_policy() throws Exception { + assertConformsToSchema( + report( + Cluster.builder() + .withSpeculativeExecutionPolicy(new ConstantSpeculativeExecutionPolicy(100L, 2)))); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_explicit_datacenter_node_location() throws Exception { + assertConformsToSchema( + report( + Cluster.builder() + .withLoadBalancingPolicy( + new TokenAwarePolicy( + DCAwareRoundRobinPolicy.builder().withLocalDc("dc1").build())))); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_rack_node_location() throws Exception { + assertConformsToSchema( + report( + Cluster.builder() + .withLoadBalancingPolicy( + new RackAwareRoundRobinPolicy("dc1", "rack1", 0, false, false, false)))); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_inferred_rack_node_location() throws Exception { + // Neither DC nor rack configured: reported as rack-auto, with both names absent because the + // policy has not been initialized (and so has inferred nothing) at report time. + JsonNode report = + report( + Cluster.builder() + .withLoadBalancingPolicy( + new RackAwareRoundRobinPolicy(null, null, 0, false, true, true))); + + assertThat(report.path("node-location-preference").path("type").asText()) + .isEqualTo("rack-auto"); + assertConformsToSchema(report); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_round_robin_load_balancing_policy() throws Exception { + JsonNode report = report(Cluster.builder().withLoadBalancingPolicy(new RoundRobinPolicy())); + + assertThat(report.path("load-balancing-policy").path("type").asText()).isEqualTo("round-robin"); + // No DC/rack notion at all, so the whole group is omitted. + assertThat(report.has("node-location-preference")).isFalse(); + assertConformsToSchema(report); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_datacenter_aware_load_balancing_policy() + throws Exception { + JsonNode report = + report( + Cluster.builder() + .withLoadBalancingPolicy( + DCAwareRoundRobinPolicy.builder().withLocalDc("dc1").build())); + + assertThat(report.path("load-balancing-policy").path("type").asText()).isEqualTo("dc-aware"); + assertThat(report.path("load-balancing-policy").path("token-aware").asBoolean()).isFalse(); + assertConformsToSchema(report); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_white_list_load_balancing_policy() throws Exception { + JsonNode report = + report( + Cluster.builder() + .withLoadBalancingPolicy( + new WhiteListPolicy( + new RoundRobinPolicy(), + Collections.singletonList(new InetSocketAddress("127.0.0.1", 9042))))); + + assertThat(report.path("load-balancing-policy").path("type").asText()).isEqualTo("white-list"); + assertConformsToSchema(report); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_host_filter_load_balancing_policy() throws Exception { + JsonNode report = + report( + Cluster.builder() + .withLoadBalancingPolicy( + HostFilterPolicy.fromDCWhiteList( + new RoundRobinPolicy(), Collections.singletonList("dc1")))); + + assertThat(report.path("load-balancing-policy").path("type").asText()).isEqualTo("host-filter"); + assertConformsToSchema(report); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_latency_aware_load_balancing_policy() throws Exception { + JsonNode report = + report( + Cluster.builder() + .withLoadBalancingPolicy( + LatencyAwarePolicy.builder(new RoundRobinPolicy()).build())); + + JsonNode lb = report.path("load-balancing-policy"); + // The latency-aware wrapper contributes a flag, not a type: the type comes from its child. + assertThat(lb.path("type").asText()).isEqualTo("round-robin"); + assertThat(lb.path("latency-awareness").asBoolean()).isTrue(); + assertConformsToSchema(report); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_unwrapped_paging_optimizing_load_balancing_policy() + throws Exception { + assertConformsToSchema( + report( + Cluster.builder() + .withLoadBalancingPolicy( + new PagingOptimizingLoadBalancingPolicy( + new TokenAwarePolicy( + DCAwareRoundRobinPolicy.builder().withLocalDc("dc1").build()))))); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_tls_enabled() throws Exception { + assertConformsToSchema(report(Cluster.builder().withSSL())); + } + + @Test(groups = "unit") + public void should_conform_to_schema_for_socket_overrides() throws Exception { + SocketOptions socketOptions = + new SocketOptions() + .setKeepAlive(true) + .setReuseAddress(true) + .setSoLinger(15) + .setReceiveBufferSize(4096) + .setSendBufferSize(8192); + assertConformsToSchema(report(Cluster.builder().withSocketOptions(socketOptions))); + } + + @Test(groups = "unit") + public void should_reject_a_report_that_violates_the_schema() throws Exception { + // Sanity check that the validator actually enforces the schema (rather than accepting + // anything): an unknown top-level key must be rejected, since the schema sets + // additionalProperties=false. + ObjectNode report = (ObjectNode) report(Cluster.builder()); + report.put("bogus-unknown-key", "x"); + assertThat(SCHEMA.validate(report)).as("unknown top-level key must be rejected").isNotEmpty(); + } + + // ---- Helpers -------------------------------------------------------------------------------- + + /** Builds the full report from a cluster builder and parses it. */ + private static JsonNode report(Cluster.Builder builder) throws IOException { + String json = new DefaultDriverConfigReporter(builder.getConfiguration()).buildJson(); + return MAPPER.readTree(json); + } + + private static void assertConformsToSchema(JsonNode report) { + Set errors = SCHEMA.validate(report); + assertThat(errors).as("schema violations in %s", report).isEmpty(); + } + + /** A reporter that reports {@code json} verbatim, bypassing the configuration read. */ + private static DefaultDriverConfigReporter reporting(final String json) { + return new DefaultDriverConfigReporter(config()) { + @Override + protected String buildJson() { + return json; + } + }; + } + + /** + * A report that is within the limit by {@link String#length()} but over it once encoded, so that + * the check is pinned to UTF-8 bytes rather than characters. + */ + private static String oversizedReport() { + StringBuilder sb = new StringBuilder("{\"version\":1,\"pad\":\""); + // 3 bytes each in UTF-8, so two thirds of the limit in characters is over it in bytes. + for (int i = 0; i < (DefaultDriverConfigReporter.MAX_DRIVER_CONFIG_LENGTH * 2) / 3; i++) { + sb.append('€'); + } + String report = sb.append("\"}").toString(); + assertThat(report.length()).isLessThan(DefaultDriverConfigReporter.MAX_DRIVER_CONFIG_LENGTH); + assertThat(report.getBytes(StandardCharsets.UTF_8).length) + .isGreaterThan(DefaultDriverConfigReporter.MAX_DRIVER_CONFIG_LENGTH); + return report; + } + + /** A single-byte-per-character report of exactly {@code length} bytes. */ + private static String padTo(int length) { + String prefix = "{\"version\":1,\"pad\":\""; + String suffix = "\"}"; + StringBuilder sb = new StringBuilder(prefix); + for (int i = prefix.length() + suffix.length(); i < length; i++) { + sb.append('x'); + } + String report = sb.append(suffix).toString(); + assertThat(report.getBytes(StandardCharsets.UTF_8).length).isEqualTo(length); + return report; + } + + /** + * A user policy that matches none of the built-in types and is not chainable, so it is reported + * as {@code custom}. Only its class name is ever read by the reporter. + */ + private static class CustomLoadBalancingPolicy implements LoadBalancingPolicy { + @Override + public void init(Cluster cluster, Collection hosts) {} + + @Override + public HostDistance distance(Host host) { + return HostDistance.LOCAL; + } + + @Override + public Iterator newQueryPlan(String loggedKeyspace, Statement statement) { + return Collections.emptyList().iterator(); + } + + @Override + public void onAdd(Host host) {} + + @Override + public void onUp(Host host) {} + + @Override + public void onDown(Host host) {} + + @Override + public void onRemove(Host host) {} + + @Override + public void close() {} + } + + /** + * A user reconnection policy that is neither of the built-ins, so it is reported as {@code + * custom}. Only its class name is ever read by the reporter. + */ + private static class CustomReconnectionPolicy implements ReconnectionPolicy { + @Override + public ReconnectionSchedule newSchedule() { + throw new UnsupportedOperationException("never scheduled in this test"); + } + + @Override + public void init(Cluster cluster) {} + + @Override + public void close() {} + } + + /** A chainable policy whose child is itself, i.e. a chain the reporter must not walk forever. */ + private static class CyclicLoadBalancingPolicy extends DelegatingLoadBalancingPolicy { + CyclicLoadBalancingPolicy() { + super(new RoundRobinPolicy()); + } + + @Override + public LoadBalancingPolicy getChildPolicy() { + return this; + } + } } diff --git a/driver-core/src/test/java/com/datastax/driver/core/DriverConfigReportingCcmTest.java b/driver-core/src/test/java/com/datastax/driver/core/DriverConfigReportingCcmTest.java index a51ed8568d3..000a03e2a34 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/DriverConfigReportingCcmTest.java +++ b/driver-core/src/test/java/com/datastax/driver/core/DriverConfigReportingCcmTest.java @@ -18,6 +18,9 @@ import static org.assertj.core.api.Assertions.assertThat; import com.datastax.driver.core.utils.ScyllaVersion; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; @@ -35,7 +38,8 @@ * shared value; *

  • {@code SESSION_ID} is shared across every {@code Session} obtained from the same {@code * Cluster} (it is Cluster-scoped, not Session-scoped — see {@link Connection.Factory}); - *
  • {@code DRIVER_CONFIG} is stored for exactly one connection (the control connection); + *
  • {@code DRIVER_CONFIG} is stored for exactly one connection (the control connection), as the + * full versioned v2 JSON report; *
  • with reporting disabled, {@code SESSION_ID} is still stored but {@code DRIVER_CONFIG} is * not. * @@ -43,9 +47,8 @@ *

    The cluster under test uses the default configuration — no {@code withDriverConfigReporting} * call — so these also assert that reporting is enabled by default. * - *

    Stage 1 emits only the schema version, so {@code DRIVER_CONFIG} is asserted to be {@code - * {"version":1}}. ScyllaDB-only: {@code system.clients.client_options} is a Scylla feature (added - * in ScyllaDB 2026.1). + *

    ScyllaDB-only: {@code system.clients.client_options} is a Scylla feature (added in ScyllaDB + * 2026.1). */ @ScyllaVersion( minOSS = "2026.1", @@ -77,10 +80,23 @@ public void should_store_session_id_on_all_connections_and_driver_config_on_cont } } - // DRIVER_CONFIG is stored for exactly one connection (the control connection). Stage 1 reports - // only the schema version. + // DRIVER_CONFIG is stored for exactly one connection (the control connection), as the versioned + // JSON report the driver builds. assertThat(driverConfigs).hasSize(1); - assertThat(driverConfigs.get(0)).isEqualTo("{\"version\":1}"); + JsonNode report = parse(driverConfigs.get(0)); + assertThat(report.path("version").asInt()).isEqualTo(1); + // Spot-check a few required v2 groups made it through intact (no truncation). + assertThat(report.has("connection")).isTrue(); + assertThat(report.has("load-balancing-policy")).isTrue(); + assertThat(report.path("connection-pool").path("shard-aware").has("enabled")).isTrue(); + } + + private JsonNode parse(String json) { + try { + return new ObjectMapper().readTree(json); + } catch (IOException e) { + throw new AssertionError("DRIVER_CONFIG is not valid JSON: " + json, e); + } } @Test(groups = "short") diff --git a/driver-core/src/test/resources/config/driver-config-report-v1.schema.json b/driver-core/src/test/resources/config/driver-config-report-v1.schema.json new file mode 100644 index 00000000000..b2b5886bb7c --- /dev/null +++ b/driver-core/src/test/resources/config/driver-config-report-v1.schema.json @@ -0,0 +1,846 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://scylladb.com/schemas/driver-client-options/v1.json", + "title": "ScyllaDB driver DRIVER_CONFIG configuration", + "description": "Schema for the JSON value sent under the STARTUP option key DRIVER_CONFIG, describing the effective client configuration. The top-level object must include `version` and the required configuration groups listed by this schema. Unknown top-level keys are rejected. Built-in groups reject unknown keys and require the keys listed in each group; custom policy objects may include additional implementation-specific public attributes where explicitly allowed.", + "type": "object", + "additionalProperties": false, + "required": [ + "version", + "connection", + "socket", + "control-plane", + "reconnection-policy", + "retry-policy", + "load-balancing-policy", + "connection-pool", + "query-defaults", + "tls" + ], + "properties": { + "version": { + "description": "Major schema version. Adding keys is backward-compatible and does not bump this; only changing/removing the meaning of an existing key does.", + "type": "integer", + "const": 1 + }, + "connection": { + "$ref": "#/$defs/connection" + }, + "socket": { + "$ref": "#/$defs/socket" + }, + "control-plane": { + "$ref": "#/$defs/control-plane" + }, + "reconnection-policy": { + "$ref": "#/$defs/reconnection-policy" + }, + "retry-policy": { + "$ref": "#/$defs/retry-policy" + }, + "speculative-execution-policy": { + "$ref": "#/$defs/speculative-execution-policy" + }, + "load-balancing-policy": { + "$ref": "#/$defs/load-balancing-policy" + }, + "node-location-preference": { + "$ref": "#/$defs/node-location-preference" + }, + "connection-pool": { + "$ref": "#/$defs/connection-pool" + }, + "query-defaults": { + "$ref": "#/$defs/query-defaults" + }, + "tls": { + "$ref": "#/$defs/tls" + } + }, + "$defs": { + "positiveInteger": { + "type": "integer", + "minimum": 1 + }, + "nonNegativeInteger": { + "type": "integer", + "minimum": 0 + }, + "retryPolicyBackoff": { + "description": "Delay inserted between retry attempts of a retry policy. Discriminated union: when present, `type` selects the backoff algorithm and each algorithm carries only its own parameters. Absent when there is no delay between attempts.", + "oneOf": [ + { + "type": "object", + "description": "Exponential backoff: the delay starts at base-ms and doubles after each attempt (capped at max-ms), with a small random jitter to de-synchronize concurrent retries.", + "additionalProperties": false, + "required": [ + "type", + "base-ms" + ], + "properties": { + "type": { + "const": "exponential", + "description": "Exponential backoff algorithm." + }, + "base-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Initial delay between retries in milliseconds; the starting delay that doubles each attempt." + }, + "max-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum delay between retries in milliseconds; the exponentially growing delay is capped here. Absent when no maximum delay is configured." + } + } + }, + { + "type": "object", + "description": "Constant backoff: a fixed delay is inserted between every retry attempt.", + "additionalProperties": false, + "required": [ + "type", + "delay-ms" + ], + "properties": { + "type": { + "const": "constant", + "description": "Constant (fixed-delay) backoff algorithm." + }, + "delay-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Fixed delay between retries in milliseconds." + } + } + } + ] + }, + "connection": { + "description": "Connection-level settings: socket read/write/connect timeouts plus the CQL-level idle heartbeat. Durations are in milliseconds. Optional duration fields are absent when unset or not applicable.", + "type": "object", + "required": [ + "connect" + ], + "additionalProperties": false, + "properties": { + "connect": { + "type": "object", + "description": "Settings for establishing a TCP/CQL connection to a node.", + "required": [ + "timeout-ms" + ], + "additionalProperties": false, + "properties": { + "timeout-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Timeout for establishing a TCP/CQL connection to a node." + } + } + }, + "read": { + "type": "object", + "description": "Settings for reading from a connection.", + "additionalProperties": false, + "properties": { + "timeout-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Read operation timeout." + } + } + }, + "write": { + "type": "object", + "description": "Settings for writing to a connection. Direction-specific options such as write coalescing are expected to be added here in a future schema version.", + "additionalProperties": false, + "properties": { + "coalescing": { + "type": "object", + "description": "Settings for write coalescing. It is a placeholder for v2", + "additionalProperties": false, + "properties": {} + }, + "timeout-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Write operation timeout." + } + } + }, + "heartbeat": { + "type": "object", + "description": "Reserved for CQL-level idle heartbeat settings. Optional and intentionally empty in this schema version. It is a placeholder for v2", + "additionalProperties": false, + "properties": {} + } + } + }, + "control-plane": { + "description": "Control-plane timeout settings for internal/system queries run over the control connection and for schema agreement. Each value is in milliseconds. Optional values are absent when unset or not applicable.", + "type": "object", + "required": [ + "system-queries", + "schema-agreement" + ], + "additionalProperties": false, + "properties": { + "system-queries": { + "type": "object", + "description": "Settings for internal/system queries run over the control connection.", + "additionalProperties": false, + "required": [ + "timeout" + ], + "properties": { + "timeout": { + "type": "object", + "description": "Timeouts applied to internal/system queries. Each value is in milliseconds. Optional values are absent when unset or not applicable.", + "additionalProperties": false, + "properties": { + "client-side-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "A client-side timeout for internal queries." + }, + "server-side-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "A server-side timeout for internal queries." + } + } + } + } + }, + "schema-agreement": { + "type": "object", + "description": "Settings for schema agreement across nodes.", + "additionalProperties": false, + "required": [ + "timeout-ms" + ], + "properties": { + "timeout-ms": { + "$ref": "#/$defs/nonNegativeInteger", + "description": "Maximum time to wait for schema agreement across nodes. Always a concrete value; 0 means do not wait for agreement." + } + } + } + } + }, + "socket": { + "description": "Low-level TCP socket options applied to client connections. Boolean options (tcp-no-delay, keep-alive, reuse-address) report the effective on/off state: when no explicit value is configured, the OS/platform default is reported. Buffer sizes are in bytes and linger is in seconds; these fields are absent when unset (kernel auto-tuned buffer / linger disabled).", + "type": "object", + "required": [ + "tcp-no-delay", + "keep-alive", + "reuse-address" + ], + "additionalProperties": false, + "properties": { + "tcp-no-delay": { + "type": "boolean", + "description": "TCP_NODELAY: disable Nagle's algorithm. Reports the effective value; when no explicit value is configured, the OS/platform default is reported." + }, + "keep-alive": { + "type": "boolean", + "description": "SO_KEEPALIVE: OS-level TCP keep-alive probes on idle connections. Reports the effective on/off state; when no explicit value is configured, the OS/platform default is reported." + }, + "reuse-address": { + "type": "boolean", + "description": "SO_REUSEADDR: allow reuse of a local address. Reports the effective on/off state; when no explicit value is configured, the OS/platform default is reported." + }, + "linger": { + "type": "object", + "required": [ + "interval-s" + ], + "additionalProperties": false, + "properties": { + "interval-s": { + "$ref": "#/$defs/nonNegativeInteger", + "description": "SO_LINGER lingering-close interval in seconds." + } + } + }, + "receive-buffer": { + "type": "object", + "required": [ + "size-bytes" + ], + "additionalProperties": false, + "properties": { + "size-bytes": { + "$ref": "#/$defs/positiveInteger", + "description": "SO_RCVBUF socket receive buffer size hint in bytes." + } + } + }, + "send-buffer": { + "type": "object", + "required": [ + "size-bytes" + ], + "additionalProperties": false, + "properties": { + "size-bytes": { + "$ref": "#/$defs/positiveInteger", + "description": "SO_SNDBUF socket send buffer size hint in bytes." + } + } + } + } + }, + "reconnection-policy": { + "description": "Defines how connection attempts to a node are retried after a connection failure.", + "oneOf": [ + { + "type": "object", + "description": "Exponential backoff reconnection policy.", + "additionalProperties": false, + "required": [ + "type", + "base-ms", + "max-ms" + ], + "properties": { + "type": { + "const": "exponential", + "description": "Reconnection policy type." + }, + "base-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Initial delay before the first reconnection attempt in milliseconds. Always a concrete value when this policy is reported." + }, + "max-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum delay between reconnection attempts (in milliseconds). Always a concrete value when this policy is reported." + }, + "max-attempts": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum number of reconnection attempts before giving up. Absent when attempts are unlimited." + } + } + }, + { + "type": "object", + "description": "Constant delay reconnection policy.", + "additionalProperties": false, + "required": [ + "type", + "delay-ms" + ], + "properties": { + "type": { + "const": "constant", + "description": "Reconnection policy type." + }, + "delay-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Fixed delay between reconnection attempts (in milliseconds). Always a concrete value when this policy is reported." + }, + "max-attempts": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum number of reconnection attempts before giving up. Absent when attempts are unlimited." + } + } + }, + { + "type": "object", + "description": "A user-supplied reconnection policy that is not one of the built-ins. Identified by `name` only. Implementations that can introspect a policy instance MAY also serialize its public attributes as additional properties on this object.", + "additionalProperties": true, + "required": [ + "type", + "name" + ], + "properties": { + "type": { + "const": "custom", + "description": "Reconnection policy type: a user-supplied policy." + }, + "name": { + "type": "string", + "description": "Name of the custom policy (e.g. the public type name of the user-provided policy)." + } + } + }, + { + "type": "null", + "description": "No reconnection attempts will be made." + } + ] + }, + "retry-policy": { + "description": "Controls whether and how a failed query is retried. Discriminated on `type`; each policy only permits its own parameters.", + "oneOf": [ + { + "type": "object", + "description": "Error-type-aware retry policy with fixed, non-configurable rules: retry at most once on Unavailable (on the next node), retry on a read timeout only when enough replicas responded but the data was not retrieved, retry on a write timeout only for batch-log writes, retry the next node on Overloaded/ServerError/Bootstrapping/broken-connection when the request is idempotent, and never retry serial (LWT) reads. It can have backoff enabled.", + "additionalProperties": false, + "required": [ + "type" + ], + "properties": { + "type": { + "const": "standard-error-aware", + "description": "Retry policy type." + }, + "backoff": { + "$ref": "#/$defs/retryPolicyBackoff", + "description": "Delay inserted between retries." + } + } + }, + { + "type": "object", + "description": "Simple retry policy with a fixed number of retries.", + "additionalProperties": false, + "required": [ + "type", + "max-retries" + ], + "properties": { + "type": { + "const": "simple", + "description": "Retry policy type." + }, + "backoff": { + "$ref": "#/$defs/retryPolicyBackoff", + "description": "Delay inserted between retries." + }, + "max-retries": { + "$ref": "#/$defs/nonNegativeInteger", + "description": "Maximum number of retries before giving up. Always a concrete value when this policy is reported; 0 means no retries." + } + } + }, + { + "type": "object", + "description": "Fall-through retry policy: never retries anything and always rethrows the original error to the caller. Every error type — read timeout, write timeout, unavailable, and unexpected request errors (connection errors, Overloaded, ServerError, Bootstrapping) — is propagated unchanged. This is a true no-op and is stricter than the 'never' policy, which still retries the next host on connection/server errors.", + "additionalProperties": false, + "required": [ + "type" + ], + "properties": { + "type": { + "const": "fallthrough", + "description": "Retry policy type." + } + } + }, + { + "type": "object", + "description": "Downgrading-consistency retry policy: retries at a lower consistency level on failure.", + "additionalProperties": false, + "required": [ + "type" + ], + "properties": { + "type": { + "const": "downgrading-consistency", + "description": "Retry policy type." + }, + "backoff": { + "$ref": "#/$defs/retryPolicyBackoff", + "description": "Delay inserted between retries." + } + } + }, + { + "type": "object", + "description": "A user-supplied retry policy that is not one of the built-ins. Identified by `name` only. Implementations that can introspect a policy instance MAY also serialize its public attributes as additional properties on this object.", + "additionalProperties": true, + "required": [ + "type", + "name" + ], + "properties": { + "type": { + "const": "custom", + "description": "Retry policy type: a user-supplied policy." + }, + "name": { + "type": "string", + "description": "Name of the custom policy (e.g. the public type name of the user-provided policy)." + }, + "description": { + "type": "string", + "description": "Textual description of what this policy does." + } + } + } + ] + }, + "speculative-execution-policy": { + "description": "Controls pre-emptive duplicate requests to other replicas. Discriminated on `type`; each policy only permits its own parameters.", + "oneOf": [ + { + "type": "object", + "description": "Constant-delay speculative execution: launch extra executions after a fixed delay.", + "additionalProperties": false, + "required": [ + "type", + "max-executions", + "delay-ms" + ], + "properties": { + "type": { + "const": "constant", + "description": "Speculative execution policy type." + }, + "max-executions": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum number of speculative executions per request." + }, + "delay-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Delay before launching each additional execution (in milliseconds)." + } + } + }, + { + "type": "object", + "description": "Percentile-based speculative execution: launch extra executions once latency exceeds a percentile threshold.", + "additionalProperties": false, + "required": [ + "type", + "max-executions", + "percentile" + ], + "properties": { + "type": { + "const": "percentile", + "description": "Speculative execution policy type." + }, + "max-executions": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum number of speculative executions per request." + }, + "percentile": { + "type": "number", + "exclusiveMinimum": 0, + "exclusiveMaximum": 100, + "description": "Latency percentile (0–100, exclusive; e.g. 99.0) that triggers an additional execution." + } + } + }, + { + "type": "object", + "description": "A user-supplied speculative execution policy that is not one of the built-ins. Identified by `name` only. Implementations that can introspect a policy instance MAY also serialize its public attributes as additional properties on this object.", + "additionalProperties": true, + "required": [ + "type", + "name" + ], + "properties": { + "type": { + "const": "custom", + "description": "Speculative execution policy type: a user-supplied policy." + }, + "name": { + "type": "string", + "description": "Name of the custom policy (e.g. the public type name of the user-provided policy)." + }, + "description": { + "type": "string", + "description": "Textual description of what this policy does." + } + } + } + ] + }, + "load-balancing-policy": { + "description": "Load balancing / host selection policy. Discriminated on `type`: a built-in policy reports the normalized flags below, while a user-supplied policy is reported as type 'custom' with a `name` and, optionally, serialized public attributes.", + "oneOf": [ + { + "type": "object", + "description": "A built-in load balancing policy, reported with normalized location/awareness flags.", + "additionalProperties": false, + "required": [ + "type", + "shuffle", + "token-aware", + "dc-failover", + "latency-awareness" + ], + "properties": { + "type": { + "enum": [ + "token-aware", + "round-robin", + "dc-aware", + "rack-aware", + "dc-inferring", + "basic", + "white-list", + "host-filter", + "default" + ], + "description": "Policy type one of e.g. token-aware, round-robin, dc-aware, rack-aware." + }, + "token-aware": { + "type": "boolean", + "description": "Whether queries are routed to token replicas." + }, + "shuffle": { + "type": "boolean", + "description": "Whether candidate order in a query plan is shuffled for regular queries. LWT and strongly consistent queries are not shuffled." + }, + "dc-failover": { + "type": "boolean", + "description": "Whether requests may fail over to remote datacenters." + }, + "latency-awareness": { + "type": "boolean", + "description": "Whether latency-aware host ordering is enabled." + } + } + }, + { + "type": "object", + "description": "A user-supplied load balancing policy that is not one of the built-ins. Identified by `name` only. Implementations that can introspect a policy instance MAY also serialize its public attributes as additional properties on this object.", + "additionalProperties": true, + "required": [ + "type", + "name" + ], + "properties": { + "type": { + "const": "custom", + "description": "Load balancing policy type: a user-supplied policy." + }, + "name": { + "type": "string", + "description": "Name of the custom policy (e.g. the public type name of the user-provided policy)." + }, + "description": { + "type": "string", + "description": "Textual description of what this policy does." + } + } + } + ] + }, + "node-location-preference": { + "description": "Session-level datacenter/rack preference, set independently of the load balancing policy. Some implementations let users set a preferred DC/rack directly on the session configuration; the load balancing policy and other components read this preference unless a policy overrides it. May be sourced from different places; if DC/rack preferences are specified in the load balancing policy, they should be reported here.", + "oneOf": [ + { + "type": "object", + "description": "Explicitly configured datacenter preference.", + "additionalProperties": false, + "required": [ + "type", + "local-dc" + ], + "properties": { + "type": { + "const": "dc", + "description": "Session-level location preference: explicit datacenter." + }, + "local-dc": { + "type": "string", + "description": "Explicitly configured preferred datacenter." + } + } + }, + { + "type": "object", + "description": "Explicitly configured datacenter and rack preference.", + "additionalProperties": false, + "required": [ + "type", + "local-dc", + "local-rack" + ], + "properties": { + "type": { + "const": "rack", + "description": "Session-level location preference: explicit datacenter and rack." + }, + "local-dc": { + "type": "string", + "description": "Explicitly configured preferred datacenter." + }, + "local-rack": { + "type": "string", + "description": "Explicitly configured preferred rack." + } + } + }, + { + "type": "object", + "description": "Datacenter preference inferred from the first node the client connects to.", + "additionalProperties": false, + "required": [ + "type" + ], + "properties": { + "type": { + "const": "dc-auto", + "description": "Session-level location preference: inferred datacenter." + }, + "local-dc": { + "type": "string", + "description": "Inferred preferred datacenter. Absent when not yet known at report time." + } + } + }, + { + "type": "object", + "description": "Datacenter and rack preference inferred from the first node the client connects to.", + "additionalProperties": false, + "required": [ + "type" + ], + "properties": { + "type": { + "const": "rack-auto", + "description": "Session-level location preference: inferred datacenter and rack." + }, + "local-dc": { + "type": "string", + "description": "Inferred preferred datacenter. Absent when not yet known at report time." + }, + "local-rack": { + "type": "string", + "description": "Inferred preferred rack. Absent when not yet known at report time." + } + } + } + ] + }, + "connection-pool": { + "description": "Connection pooling configuration.", + "type": "object", + "required": [ + "type", + "desired-connections-count", + "shard-aware" + ], + "additionalProperties": false, + "properties": { + "type": { + "enum": [ + "host", + "shard" + ], + "description": "What each pool is keyed by: per host or per shard." + }, + "desired-connections-count": { + "$ref": "#/$defs/positiveInteger", + "description": "Number of connections to open per host or per shard." + }, + "connection": { + "type": "object", + "description": "Per-connection pool settings.", + "required": [ + "max-requests" + ], + "additionalProperties": false, + "properties": { + "max-requests": { + "$ref": "#/$defs/positiveInteger", + "description": "Maximum number of in-flight requests per connection." + } + } + }, + "shard-aware": { + "type": "object", + "required": [ + "enabled" + ], + "additionalProperties": false, + "properties": { + "enabled": { + "type": "boolean", + "description": "Whether the client is configured to use ScyllaDB's dedicated shard-aware port (default 19042, TLS 19043) to reach a chosen shard in a single connect, versus the fallback of opening connections on the normal port and reading the server-assigned shard. Reports configuration intent; at runtime the port must also be advertised by the server and reachable, otherwise the client falls back transparently." + } + } + } + } + }, + "query-defaults": { + "description": "Default per-request settings applied to statements that do not override them.", + "type": "object", + "required": [ + "consistency", + "idempotence", + "client-timestamps", + "request" + ], + "additionalProperties": false, + "properties": { + "page": { + "type": "object", + "required": [ + "size" + ], + "additionalProperties": false, + "properties": { + "size": { + "$ref": "#/$defs/positiveInteger", + "description": "Default page (fetch) size for result sets. Absent when page is not limited." + } + } + }, + "consistency": { + "description": "Default consistency level applied to requests that do not override it. Always present when this group is reported.", + "type": "string", + "enum": [ + "ANY", + "ONE", + "TWO", + "THREE", + "QUORUM", + "ALL", + "LOCAL_QUORUM", + "EACH_QUORUM", + "LOCAL_ONE" + ] + }, + "serial-consistency": { + "description": "Default serial consistency for LWT/conditional statements. Absent when unset; the server default applies.", + "type": "string", + "enum": [ + "SERIAL", + "LOCAL_SERIAL" + ] + }, + "idempotence": { + "description": "Default idempotence flag applied to statements that do not set their own.", + "type": "boolean" + }, + "client-timestamps": { + "description": "True when the client assigns the write timestamp client-side (protocol-level/USING TIMESTAMP) instead of letting the coordinator assign it.", + "type": "boolean" + }, + "request": { + "type": "object", + "description": "Default request-level settings.", + "required": [ + "timeout-ms" + ], + "additionalProperties": false, + "properties": { + "timeout-ms": { + "$ref": "#/$defs/positiveInteger", + "description": "Client-side timeout for a single request/query in milliseconds. Always a concrete value when query defaults are reported." + } + } + } + } + }, + "tls": { + "description": "TLS/SSL transport settings. Reports only booleans; never credentials, keys, or host lists.", + "type": "object", + "required": [ + "enabled", + "hostname-verification" + ], + "additionalProperties": false, + "properties": { + "enabled": { + "type": "boolean", + "description": "Whether TLS is enabled for connections." + }, + "hostname-verification": { + "type": "boolean", + "description": "Whether the server hostname is verified against its certificate." + } + } + } + } +} diff --git a/pom.xml b/pom.xml index d6c97121022..fbc9ab2909b 100644 --- a/pom.xml +++ b/pom.xml @@ -88,6 +88,7 @@ 1.2.13 3.0.8 0.27.2 + 1.5.9 3.5.4 127.0.1. @@ -402,6 +403,12 @@ 4.3.0 + + com.networknt + json-schema-validator + ${json-schema-validator.version} + +