diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java b/core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java index 56a6fe9c2be..24bf038a7fd 100644 --- a/core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java +++ b/core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java @@ -1173,7 +1173,22 @@ public enum DefaultDriverOption implements DriverOption { * *

Value-Type: boolean */ - ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES("advanced.address-translator.resolve-addresses"); + ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES("advanced.address-translator.resolve-addresses"), + /** + * Whether the driver reports its effective client configuration to ScyllaDB at connection time. + * + *

When {@code true}, the driver adds two entries to the CQL {@code STARTUP} options, which + * ScyllaDB stores in {@code system.clients.client_options} so operators can inspect driver + * settings while investigating incidents: a {@code SESSION_ID} on every connection (so the server + * can group a session's connections) and a compact JSON payload under the {@code DRIVER_CONFIG} + * key on the control connection only, describing the effective configuration of the driver's + * default execution profile (connection/socket settings, timeouts, retry/reconnection/ + * speculative-execution/load-balancing policies, connection pooling, query defaults, and TLS). + * When {@code false}, neither entry is sent and there is no change on the wire. + * + *

Value type: boolean + */ + CLIENT_CONFIG_REPORTING_ENABLED("advanced.client-config-reporting.enabled"); private final String path; diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java b/core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java index 60190fc1cce..86b985437cc 100644 --- a/core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java +++ b/core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java @@ -400,6 +400,7 @@ protected static void fillWithDriverDefaults(OptionsMap map) { // values) with no sensible scalar default, analogous to how CONFIG_RELOAD_INTERVAL is omitted. map.put(TypedDriverOption.CLIENT_ROUTES_NATIVE_TRANSPORT_PORT, 9042); map.put(TypedDriverOption.CLIENT_ROUTES_SHARD_AWARENESS_ENABLED, false); + map.put(TypedDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, false); } @Immutable diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java b/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java index e7c606cade8..e7865d449ef 100644 --- a/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java +++ b/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java @@ -976,6 +976,11 @@ public String toString() { new TypedDriverOption<>( DefaultDriverOption.CLIENT_ROUTES_SHARD_AWARENESS_ENABLED, GenericType.BOOLEAN); + /** Whether the driver reports its client configuration to ScyllaDB at connection time. */ + public static final TypedDriverOption CLIENT_CONFIG_REPORTING_ENABLED = + new TypedDriverOption<>( + DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, GenericType.BOOLEAN); + private static Iterable> introspectBuiltInValues() { try { ImmutableList.Builder> result = ImmutableList.builder(); diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannelOptions.java b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannelOptions.java index 208cf52ac22..378fd2dc0b8 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannelOptions.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannelOptions.java @@ -47,15 +47,24 @@ public static Builder builder() { public final String ownerLogPrefix; + /** + * Whether the channel should report the driver configuration ({@code DRIVER_CONFIG}) in its + * {@code STARTUP} options. Set only for the control connection, so the (potentially large) config + * blob is sent once per session rather than on every pooled connection. + */ + public final boolean reportConfig; + private DriverChannelOptions( CqlIdentifier keyspace, List eventTypes, EventCallback eventCallback, - String ownerLogPrefix) { + String ownerLogPrefix, + boolean reportConfig) { this.keyspace = keyspace; this.eventTypes = eventTypes; this.eventCallback = eventCallback; this.ownerLogPrefix = ownerLogPrefix; + this.reportConfig = reportConfig; } public static class Builder { @@ -63,6 +72,7 @@ public static class Builder { private List eventTypes = Collections.emptyList(); private EventCallback eventCallback = null; private String ownerLogPrefix = null; + private boolean reportConfig = false; public Builder withKeyspace(CqlIdentifier keyspace) { this.keyspace = keyspace; @@ -82,8 +92,17 @@ public Builder withOwnerLogPrefix(String ownerLogPrefix) { return this; } + /** + * Marks this channel as the one that reports {@code DRIVER_CONFIG} (the control connection). + */ + public Builder reportConfig(boolean reportConfig) { + this.reportConfig = reportConfig; + return this; + } + public DriverChannelOptions build() { - return new DriverChannelOptions(keyspace, eventTypes, eventCallback, ownerLogPrefix); + return new DriverChannelOptions( + keyspace, eventTypes, eventCallback, ownerLogPrefix, reportConfig); } } } diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandler.java b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandler.java index 85bffdb016e..46959cd9746 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandler.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandler.java @@ -194,6 +194,11 @@ Message getRequest() { if (featureStore != null) { featureStore.populateStartupOptions(startupOptions); } + // Adds SESSION_ID on every connection and DRIVER_CONFIG on the control connection + // (options.reportConfig); no-op when client config reporting is disabled. + context + .getDriverConfigReporter() + .populateStartupOptions(startupOptions, options.reportConfig); return request = new Startup(startupOptions); case GET_CLUSTER_NAME: return request = CLUSTER_NAME_QUERY; diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporter.java b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporter.java new file mode 100644 index 00000000000..9e66601123d --- /dev/null +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporter.java @@ -0,0 +1,431 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.internal.core.context; + +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverExecutionProfile; +import com.datastax.oss.driver.api.core.connection.ReconnectionPolicy; +import com.datastax.oss.driver.api.core.loadbalancing.LoadBalancingPolicy; +import com.datastax.oss.driver.api.core.retry.RetryPolicy; +import com.datastax.oss.driver.api.core.specex.SpeculativeExecutionPolicy; +import com.datastax.oss.driver.api.core.uuid.Uuids; +import com.datastax.oss.driver.internal.core.connection.ConstantReconnectionPolicy; +import com.datastax.oss.driver.internal.core.connection.ExponentialReconnectionPolicy; +import com.datastax.oss.driver.internal.core.loadbalancing.DcInferringLoadBalancingPolicy; +import com.datastax.oss.driver.internal.core.loadbalancing.DefaultLoadBalancingPolicy; +import com.datastax.oss.driver.internal.core.retry.ConsistencyDowngradingRetryPolicy; +import com.datastax.oss.driver.internal.core.retry.DefaultRetryPolicy; +import com.datastax.oss.driver.internal.core.specex.ConstantSpeculativeExecutionPolicy; +import com.datastax.oss.driver.internal.core.specex.NoSpeculativeExecutionPolicy; +import com.datastax.oss.driver.internal.core.time.ServerSideTimestampGenerator; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import java.util.Map; +import java.util.UUID; +import net.jcip.annotations.ThreadSafe; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Default {@link DriverConfigReporter}: serializes the driver configuration to the cross-driver + * {@code DRIVER_CONFIG} JSON shape and adds it to the control connection's {@code STARTUP} options. + * + *

The blob is (re)built on demand every time the control connection initializes, so it always + * reflects the current (possibly reloaded) configuration without any caching. + * + *

Follows the schema's omission principle throughout: a key the Java driver has no equivalent + * for is left out of the JSON entirely rather than reported as {@code null}. + * + *

Known limitation: the report always describes {@link + * com.datastax.oss.driver.api.core.config.DriverExecutionProfile#DEFAULT_NAME the default execution + * profile}, not whichever profile a given request actually runs with. A session that relies on + * named execution profiles for some of its traffic will have that traffic's real settings + * (consistency level, timeouts, retry policy, ...) differ from what {@code DRIVER_CONFIG} reports. + * Reporting per-profile configuration would need a schema shape for multiple profiles, which the + * cross-driver schema doesn't define; this is a known gap, not an oversight. + * + *

Thread safety: this class is safe to use as shipped, but {@link #buildJson()} and + * {@link #populateConfig(ObjectNode, DriverExecutionProfile)} are {@code protected} purely to let + * tests override them (see {@code DefaultDriverConfigReporterTest}); a production subclass that + * overrides either to add caching or other mutable state is responsible for its own synchronization + * — {@code buildJson()} runs on every control-connection (re)initialization and may be called + * concurrently with a reconnect racing a fresh session start. + */ +@ThreadSafe +public class DefaultDriverConfigReporter implements DriverConfigReporter { + + private static final Logger LOG = LoggerFactory.getLogger(DefaultDriverConfigReporter.class); + + /** STARTUP option key under which the config JSON is sent. */ + public static final String DRIVER_CONFIG_KEY = "DRIVER_CONFIG"; + + /** STARTUP option key under which the per-session identifier is sent. */ + public static final String SESSION_ID_KEY = "SESSION_ID"; + + /** + * Major schema version. Adding keys is backward-compatible and does not bump this; only + * changing/removing the meaning of an existing key does. + */ + static final int SCHEMA_VERSION = 1; + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + protected final InternalDriverContext context; + + // Dedicated, driver-generated identifier for this session. Not derived from the (user-settable, + // Insights-oriented) CLIENT_ID, so that it is guaranteed unique per session as the grouping key + // requires. The reporter is a per-session singleton (built once via LazyReference), so this value + // is stable and shared across all of the session's connections. + private final UUID sessionId = Uuids.random(); + + public DefaultDriverConfigReporter(InternalDriverContext context) { + this.context = context; + } + + @Override + public void populateStartupOptions( + Map startupOptions, boolean reportDriverConfig) { + // Configuration reporting is a best-effort diagnostic aid: it runs on the connection + // initialization path, so any failure here (a bad config read, a misbehaving policy while + // introspecting, a serialization error) must be swallowed rather than allowed to break the + // connection — which would prevent the session from establishing or reconnecting. + try { + if (!isEnabled()) { + return; + } + // SESSION_ID on every connection so the server can group a session's connections. + startupOptions.put(SESSION_ID_KEY, sessionId.toString()); + // DRIVER_CONFIG blob only on the control connection. + if (reportDriverConfig) { + String json = buildJson(); + if (json != null) { + startupOptions.put(DRIVER_CONFIG_KEY, json); + } + } + } catch (RuntimeException e) { + LOG.warn( + "Error while building the driver configuration report; skipping client config reporting", + e); + } + } + + private boolean isEnabled() { + return context + .getConfig() + .getDefaultProfile() + .getBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, false); + } + + /** + * Builds the compact, single-line JSON configuration report. + * + *

Relies on the policy/generator {@code LazyReference}s (reconnection, retry, speculative + * execution, load balancing, timestamp generator, SSL engine factory) already being resolved by + * the time this runs, which holds today because session bootstrap eagerly forces them before + * {@code ProtocolInitHandler} sends the first {@code STARTUP}. That ordering isn't enforced by + * this class; a future change to session bootstrap that defers one of those references could make + * this the first caller to resolve it, from a Netty event-loop thread mid-{@code STARTUP} build. + */ + protected String buildJson() { + ObjectNode root = OBJECT_MAPPER.createObjectNode(); + root.put("version", SCHEMA_VERSION); + populateConfig(root, context.getConfig().getDefaultProfile()); + try { + return OBJECT_MAPPER.writeValueAsString(root); + } catch (JsonProcessingException e) { + // An in-memory node tree should never fail to serialize; never let it break connection setup. + LOG.warn("Failed to serialize driver configuration report; skipping DRIVER_CONFIG", e); + return null; + } + } + + /** + * Populates the configuration groups onto the report root, from the default execution profile + * plus the context's policies. Each group follows the cross-driver schema; a key the Java driver + * has no equivalent for is omitted rather than reported as {@code null}. + */ + protected void populateConfig(ObjectNode root, DriverExecutionProfile config) { + root.set("connection", connection(config)); + root.set("socket", socket(config)); + root.set("control-plane", controlPlane(config)); + root.set("reconnection-policy", reconnectionPolicy(config)); + root.set("retry-policy", retryPolicy()); + // No null variant in the schema for this group: omitted entirely when there is none. + ObjectNode specExec = speculativeExecutionPolicy(config); + if (specExec != null) { + root.set("speculative-execution-policy", specExec); + } + root.set("load-balancing-policy", loadBalancingPolicy(config)); + root.set("node-location-preference", nodeLocationPreference(config)); + root.set("connection-pool", connectionPool(config)); + root.set("query-defaults", queryDefaults(config)); + root.set("tls", tls(config)); + } + + private ObjectNode connection(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + ObjectNode connect = OBJECT_MAPPER.createObjectNode(); + connect.put( + "timeout-ms", + config.getDuration(DefaultDriverOption.CONNECTION_CONNECT_TIMEOUT).toMillis()); + n.set("connect", connect); + // The Java driver has no socket-level read/write timeouts, and connection.heartbeat is a + // reserved empty placeholder in this schema version (no slot for HEARTBEAT_INTERVAL/TIMEOUT + // yet) — all three are omitted entirely rather than reported as empty/null. + return n; + } + + private ObjectNode socket(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + n.put("tcp-no-delay", config.getBoolean(DefaultDriverOption.SOCKET_TCP_NODELAY, true)); + // keep-alive / reuse-address are not set by default; report the effective driver value (the OS + // default, which the driver leaves untouched, is approximated as false when unset). + n.put("keep-alive", config.getBoolean(DefaultDriverOption.SOCKET_KEEP_ALIVE, false)); + n.put("reuse-address", config.getBoolean(DefaultDriverOption.SOCKET_REUSE_ADDRESS, false)); + if (config.isDefined(DefaultDriverOption.SOCKET_LINGER_INTERVAL)) { + ObjectNode linger = OBJECT_MAPPER.createObjectNode(); + linger.put("interval-s", config.getInt(DefaultDriverOption.SOCKET_LINGER_INTERVAL)); + n.set("linger", linger); + } + if (config.isDefined(DefaultDriverOption.SOCKET_RECEIVE_BUFFER_SIZE)) { + ObjectNode receiveBuffer = OBJECT_MAPPER.createObjectNode(); + receiveBuffer.put( + "size-bytes", config.getInt(DefaultDriverOption.SOCKET_RECEIVE_BUFFER_SIZE)); + n.set("receive-buffer", receiveBuffer); + } + if (config.isDefined(DefaultDriverOption.SOCKET_SEND_BUFFER_SIZE)) { + ObjectNode sendBuffer = OBJECT_MAPPER.createObjectNode(); + sendBuffer.put("size-bytes", config.getInt(DefaultDriverOption.SOCKET_SEND_BUFFER_SIZE)); + n.set("send-buffer", sendBuffer); + } + return n; + } + + private ObjectNode controlPlane(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + + ObjectNode timeout = OBJECT_MAPPER.createObjectNode(); + timeout.put( + "client-side-ms", + config.getDuration(DefaultDriverOption.CONTROL_CONNECTION_TIMEOUT).toMillis()); + // The Java metadata/control timeout is client-side; the driver does not add USING TIMEOUT, so + // the server-side field is omitted (not applicable). + ObjectNode systemQueries = OBJECT_MAPPER.createObjectNode(); + systemQueries.set("timeout", timeout); + n.set("system-queries", systemQueries); + + ObjectNode schemaAgreement = OBJECT_MAPPER.createObjectNode(); + schemaAgreement.put( + "timeout-ms", + config.getDuration(DefaultDriverOption.CONTROL_CONNECTION_AGREEMENT_TIMEOUT).toMillis()); + n.set("schema-agreement", schemaAgreement); + + return n; + } + + private ObjectNode reconnectionPolicy(DriverExecutionProfile config) { + ReconnectionPolicy policy = context.getReconnectionPolicy(); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + // Exact-class checks, not instanceof: none of these built-ins are final, so a user subclass + // (e.g. to tweak one method) must fall through to the "custom" branch below rather than be + // misreported as the unmodified built-in. + if (policy.getClass() == ExponentialReconnectionPolicy.class) { + n.put("type", "exponential"); + n.put("base-ms", config.getDuration(DefaultDriverOption.RECONNECTION_BASE_DELAY).toMillis()); + n.put("max-ms", config.getDuration(DefaultDriverOption.RECONNECTION_MAX_DELAY).toMillis()); + // Java's built-in reconnection policies are unbounded: max-attempts is omitted. + } else if (policy.getClass() == ConstantReconnectionPolicy.class) { + n.put("type", "constant"); + n.put("delay-ms", config.getDuration(DefaultDriverOption.RECONNECTION_BASE_DELAY).toMillis()); + } else { + customPolicy(n, policy); + } + return n; + } + + private ObjectNode retryPolicy() { + RetryPolicy policy = context.getRetryPolicy(DriverExecutionProfile.DEFAULT_NAME); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + // Exact-class check, not instanceof: DefaultRetryPolicy/ConsistencyDowngradingRetryPolicy are + // not final, so a user subclass must fall through to "custom" rather than be misreported. + if (policy.getClass() == DefaultRetryPolicy.class) { + n.put("type", "standard-error-aware"); + // No configurable backoff: omitted. + } else if (policy.getClass() == ConsistencyDowngradingRetryPolicy.class) { + n.put("type", "downgrading-consistency"); + // No configurable backoff: omitted. + } else { + customPolicy(n, policy); + } + return n; + } + + /** + * Returns {@code null} when there is no speculative execution policy to report, in which case the + * whole group is omitted from the report (the schema has no null variant for it). + */ + private ObjectNode speculativeExecutionPolicy(DriverExecutionProfile config) { + SpeculativeExecutionPolicy policy = + context.getSpeculativeExecutionPolicy(DriverExecutionProfile.DEFAULT_NAME); + // Exact-class checks, not instanceof: neither built-in is final. + if (policy.getClass() == NoSpeculativeExecutionPolicy.class) { + return null; + } + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + if (policy.getClass() == ConstantSpeculativeExecutionPolicy.class) { + n.put("type", "constant"); + n.put("max-executions", config.getInt(DefaultDriverOption.SPECULATIVE_EXECUTION_MAX)); + n.put( + "delay-ms", + config.getDuration(DefaultDriverOption.SPECULATIVE_EXECUTION_DELAY).toMillis()); + } else { + customPolicy(n, policy); + } + return n; + } + + private ObjectNode loadBalancingPolicy(DriverExecutionProfile config) { + LoadBalancingPolicy policy = + context.getLoadBalancingPolicy(DriverExecutionProfile.DEFAULT_NAME); + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + // DcInferringLoadBalancingPolicy extends DefaultLoadBalancingPolicy, overriding only how the + // local DC is discovered; every flag below is unaffected, so the two share a branch and differ + // only in the reported "type". Exact-class checks (not instanceof) so an actual user subclass + // of either still falls through to "custom" below. + if (policy.getClass() == DefaultLoadBalancingPolicy.class + || policy.getClass() == DcInferringLoadBalancingPolicy.class) { + n.put( + "type", + policy.getClass() == DcInferringLoadBalancingPolicy.class ? "dc-inferring" : "default"); + // The Java default policy is always token-aware and has no replica-shuffle option. + n.put("token-aware", true); + n.put("shuffle", false); + n.put( + "dc-failover", + config.getInt(DefaultDriverOption.LOAD_BALANCING_DC_FAILOVER_MAX_NODES_PER_REMOTE_DC, 0) + > 0); + n.put("latency-awareness", false); + } else { + customPolicy(n, policy); + } + return n; + } + + /** + * Session-level datacenter/rack preference. Java has no session-level locality API separate from + * the load balancing policy, so this is sourced from the same places the (default) load balancing + * policy itself reads locality from: the local DC can be set either programmatically via {@link + * com.datastax.oss.driver.api.core.session.SessionBuilder#withLocalDatacenter} (which takes + * precedence, mirroring {@code OptionalLocalDcHelper}) or via config; the local rack has no + * programmatic override and is config-only. + */ + private ObjectNode nodeLocationPreference(DriverExecutionProfile config) { + String localDc = context.getLocalDatacenter(DriverExecutionProfile.DEFAULT_NAME); + boolean dcExplicit = localDc != null; + if (!dcExplicit && config.isDefined(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER)) { + localDc = config.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER); + dcExplicit = true; + } + boolean rackExplicit = config.isDefined(DefaultDriverOption.LOAD_BALANCING_LOCAL_RACK); + String localRack = + rackExplicit ? config.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_RACK) : null; + + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + if (dcExplicit && rackExplicit) { + n.put("type", "rack"); + n.put("local-dc", localDc); + n.put("local-rack", localRack); + } else if (dcExplicit) { + n.put("type", "dc"); + n.put("local-dc", localDc); + } else if (rackExplicit) { + // DC isn't explicit (DefaultLoadBalancingPolicy will infer one from the first contacted + // node), but rack was configured explicitly on its own: report rack-auto so the known rack + // isn't silently dropped. local-dc is omitted since it isn't known yet at report time. + n.put("type", "rack-auto"); + n.put("local-rack", localRack); + } else { + // DC is inferred from the first contacted node; not known yet at control-connection-init + // report time, so local-dc is omitted. + n.put("type", "dc-auto"); + } + return n; + } + + private ObjectNode connectionPool(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + n.put("type", "host"); + n.put( + "desired-connections-count", config.getInt(DefaultDriverOption.CONNECTION_POOL_LOCAL_SIZE)); + if (config.isDefined(DefaultDriverOption.CONNECTION_MAX_REQUESTS)) { + ObjectNode connection = OBJECT_MAPPER.createObjectNode(); + connection.put("max-requests", config.getInt(DefaultDriverOption.CONNECTION_MAX_REQUESTS)); + n.set("connection", connection); + } + ObjectNode shardAware = OBJECT_MAPPER.createObjectNode(); + shardAware.put( + "enabled", + config.getBoolean(DefaultDriverOption.CONNECTION_ADVANCED_SHARD_AWARENESS_ENABLED, true)); + n.set("shard-aware", shardAware); + return n; + } + + private ObjectNode queryDefaults(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + int pageSize = config.getInt(DefaultDriverOption.REQUEST_PAGE_SIZE); + if (pageSize > 0) { + ObjectNode page = OBJECT_MAPPER.createObjectNode(); + page.put("size", pageSize); + n.set("page", page); + } + // pageSize <= 0 means paging is unbounded: the "page" group is omitted entirely (the schema + // has no "unbounded" sentinel). + n.put("consistency", config.getString(DefaultDriverOption.REQUEST_CONSISTENCY)); + if (config.isDefined(DefaultDriverOption.REQUEST_SERIAL_CONSISTENCY)) { + n.put("serial-consistency", config.getString(DefaultDriverOption.REQUEST_SERIAL_CONSISTENCY)); + } + n.put("idempotence", config.getBoolean(DefaultDriverOption.REQUEST_DEFAULT_IDEMPOTENCE)); + // Client-side timestamps are assigned unless the server-side generator is configured. + n.put( + "client-timestamps", + !(context.getTimestampGenerator() instanceof ServerSideTimestampGenerator)); + ObjectNode request = OBJECT_MAPPER.createObjectNode(); + request.put("timeout-ms", config.getDuration(DefaultDriverOption.REQUEST_TIMEOUT).toMillis()); + n.set("request", request); + return n; + } + + private ObjectNode tls(DriverExecutionProfile config) { + ObjectNode n = OBJECT_MAPPER.createObjectNode(); + boolean enabled = context.getSslEngineFactory().isPresent(); + n.put("enabled", enabled); + n.put( + "hostname-verification", + enabled && config.getBoolean(DefaultDriverOption.SSL_HOSTNAME_VALIDATION, true)); + return n; + } + + private void customPolicy(ObjectNode node, Object policy) { + node.put("type", "custom"); + // getSimpleName() is empty for an anonymous class (a common way to supply a one-off policy); + // fall back to the full (binary) name so the policy is still identifiable. + String name = policy.getClass().getSimpleName(); + node.put("name", name.isEmpty() ? policy.getClass().getName() : name); + } +} diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContext.java b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContext.java index 52d9cdb264c..3d1d5b82b87 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContext.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContext.java @@ -232,6 +232,8 @@ public class DefaultDriverContext implements InternalDriverContext { new LazyReference<>("requestThrottler", this::buildRequestThrottler, cycleDetector); private final LazyReference> startupOptionsRef = new LazyReference<>("startupOptions", this::buildStartupOptions, cycleDetector); + private final LazyReference driverConfigReporterRef = + new LazyReference<>("driverConfigReporter", this::buildDriverConfigReporter, cycleDetector); private final LazyReference nodeStateListenerRef; private final LazyReference schemaChangeListenerRef; private final LazyReference requestTrackerRef; @@ -369,6 +371,10 @@ protected Map buildStartupOptions() { .build(); } + protected DriverConfigReporter buildDriverConfigReporter() { + return new DefaultDriverConfigReporter(this); + } + protected Map buildLoadBalancingPolicies() { return Reflection.buildFromConfigProfiles( this, @@ -1226,6 +1232,12 @@ public Map getStartupOptions() { return startupOptionsRef.get(); } + @NonNull + @Override + public DriverConfigReporter getDriverConfigReporter() { + return driverConfigReporterRef.get(); + } + protected RequestLogFormatter buildRequestLogFormatter() { return new RequestLogFormatter(this); } diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/context/DriverConfigReporter.java b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DriverConfigReporter.java new file mode 100644 index 00000000000..d46bd3c6b71 --- /dev/null +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/context/DriverConfigReporter.java @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.internal.core.context; + +import java.util.Map; + +/** + * Adds the client-configuration-reporting entries to a connection's CQL {@code STARTUP} options, so + * ScyllaDB can store them in {@code system.clients.client_options} and operators can inspect a + * client's effective driver settings while investigating incidents. + * + *

Two entries are produced, both governed by {@code advanced.client-config-reporting.enabled}: + * + *

+ */ +public interface DriverConfigReporter { + + /** + * Adds the reporting entries to the given startup options: {@code SESSION_ID} on every + * connection, plus {@code DRIVER_CONFIG} when {@code reportDriverConfig} is true (the control + * connection). Does nothing when configuration reporting is disabled. + * + *

Called from the protocol-initialization handler for every connection. + * + *

Implementations must not throw: this runs on the connection initialization path, so a + * failure to build the report must be swallowed (and logged) rather than propagated, otherwise it + * would prevent the session from establishing or reconnecting. + * + * @param reportDriverConfig whether this connection should also carry the full {@code + * DRIVER_CONFIG} blob; true only for the control connection. + */ + void populateStartupOptions(Map startupOptions, boolean reportDriverConfig); +} diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/context/InternalDriverContext.java b/core/src/main/java/com/datastax/oss/driver/internal/core/context/InternalDriverContext.java index 81349b0c665..87933b7385e 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/context/InternalDriverContext.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/context/InternalDriverContext.java @@ -176,6 +176,13 @@ public interface InternalDriverContext extends DriverContext { @NonNull Map getStartupOptions(); + /** + * The component that builds the driver-configuration report ({@code DRIVER_CONFIG}) sent on the + * control connection's Startup message. + */ + @NonNull + DriverConfigReporter getDriverConfigReporter(); + /** * A list of additional components to notify of session lifecycle events. * diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/control/ControlConnection.java b/core/src/main/java/com/datastax/oss/driver/internal/core/control/ControlConnection.java index 52b32255c4b..9da3a2a8aa2 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/control/ControlConnection.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/control/ControlConnection.java @@ -327,6 +327,7 @@ private void init( DriverChannelOptions.builder() .withEvents(eventTypes, ControlConnection.this) .withOwnerLogPrefix(logPrefix + "|control") + .reportConfig(true) .build(); Queue nodes = diff --git a/core/src/main/resources/reference.conf b/core/src/main/resources/reference.conf index 721afa1cd76..a93fe42772c 100644 --- a/core/src/main/resources/reference.conf +++ b/core/src/main/resources/reference.conf @@ -1204,6 +1204,27 @@ datastax-java-driver { } + advanced.client-config-reporting { + + # Whether the driver reports its effective client configuration to ScyllaDB at connection time. + # + # When true, the driver adds two entries to the CQL STARTUP options, which ScyllaDB stores in + # system.clients.client_options so operators can inspect driver settings while investigating + # incidents: a SESSION_ID on every connection (so the server can group a session's connections) + # and a compact JSON payload under the DRIVER_CONFIG key on the control connection only, + # describing the effective configuration of the driver's default execution profile (connection/ + # socket settings, timeouts, retry/reconnection/speculative-execution/load-balancing policies, + # connection pooling, query defaults, and TLS). When false, neither entry is sent and there is + # no change on the wire. + # + # Required: no + # Modifiable at runtime: yes, the new value will be used for connections initialized after the change. + # Overridable in a profile: no + # Default: false + enabled = false + + } + # Whether to resolve the addresses passed to `basic.contact-points`. # # If this is true, addresses are created with `InetSocketAddress(String, int)`: the host name will diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryTestBase.java b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryTestBase.java index b25a1e9ad71..2780d5bdec9 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryTestBase.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryTestBase.java @@ -141,6 +141,8 @@ public void setup() throws InterruptedException { when(context.getEventBus()).thenReturn(eventBus); when(context.getWriteCoalescer()).thenReturn(new PassThroughWriteCoalescer(null)); when(context.getCompressor()).thenReturn(compressor); + // The init handler consults the config reporter for every connection; default to a no-op. + when(context.getDriverConfigReporter()).thenReturn((startupOptions, reportDriverConfig) -> {}); // Start local server ServerBootstrap serverBootstrap = diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandlerTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandlerTest.java index 8057ccdc48e..a7051ac466d 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandlerTest.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandlerTest.java @@ -41,6 +41,7 @@ import com.datastax.oss.driver.internal.core.DefaultProtocolVersionRegistry; import com.datastax.oss.driver.internal.core.ProtocolVersionRegistry; import com.datastax.oss.driver.internal.core.TestResponses; +import com.datastax.oss.driver.internal.core.context.DefaultDriverConfigReporter; import com.datastax.oss.driver.internal.core.context.InternalDriverContext; import com.datastax.oss.driver.internal.core.metadata.TestNodeFactory; import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList; @@ -102,6 +103,9 @@ public void setup() { when(defaultProfile.getDuration(DefaultDriverOption.HEARTBEAT_INTERVAL)) .thenReturn(Duration.ofSeconds(30)); when(internalDriverContext.getProtocolVersionRegistry()).thenReturn(protocolVersionRegistry); + // The init handler consults the config reporter for every connection; default to a no-op. + when(internalDriverContext.getDriverConfigReporter()) + .thenReturn((startupOptions, reportDriverConfig) -> {}); channel .pipeline() @@ -153,6 +157,70 @@ public void should_initialize() { assertThat(connectFuture).isSuccess(); } + // Mirrors the real reporter: SESSION_ID on every connection, DRIVER_CONFIG only when asked. + private void stubConfigReporter() { + when(internalDriverContext.getDriverConfigReporter()) + .thenReturn( + (startupOptions, reportDriverConfig) -> { + startupOptions.put(DefaultDriverConfigReporter.SESSION_ID_KEY, "test-session-id"); + if (reportDriverConfig) { + startupOptions.put( + DefaultDriverConfigReporter.DRIVER_CONFIG_KEY, "{\"version\":1}"); + } + }); + } + + @Test + public void should_report_session_id_and_driver_config_on_control_connection() { + stubConfigReporter(); + channel + .pipeline() + .addLast( + ChannelFactory.INIT_HANDLER_NAME, + new ProtocolInitHandler( + internalDriverContext, + DefaultProtocolVersion.V4, + null, + END_POINT, + DriverChannelOptions.builder().reportConfig(true).build(), + heartbeatHandler, + false)); + + channel.connect(new InetSocketAddress("localhost", 9042)); + + Frame requestFrame = readOutboundFrame(); + assertThat(requestFrame.message).isInstanceOf(Startup.class); + Startup startup = (Startup) requestFrame.message; + assertThat(startup.options).containsKey(DefaultDriverConfigReporter.SESSION_ID_KEY); + assertThat(startup.options).containsKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); + } + + @Test + public void should_report_session_id_but_not_driver_config_on_pool_connection() { + stubConfigReporter(); + channel + .pipeline() + .addLast( + ChannelFactory.INIT_HANDLER_NAME, + new ProtocolInitHandler( + internalDriverContext, + DefaultProtocolVersion.V4, + null, + END_POINT, + // DriverChannelOptions.DEFAULT has reportConfig = false (a pool connection). + DriverChannelOptions.DEFAULT, + heartbeatHandler, + false)); + + channel.connect(new InetSocketAddress("localhost", 9042)); + + Frame requestFrame = readOutboundFrame(); + assertThat(requestFrame.message).isInstanceOf(Startup.class); + Startup startup = (Startup) requestFrame.message; + assertThat(startup.options).containsKey(DefaultDriverConfigReporter.SESSION_ID_KEY); + assertThat(startup.options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); + } + @Test public void should_query_supported_options() { channel diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporterTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporterTest.java new file mode 100644 index 00000000000..79a85a19f1a --- /dev/null +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverConfigReporterTest.java @@ -0,0 +1,620 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.internal.core.context; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfig; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +import com.datastax.oss.driver.api.core.config.DriverExecutionProfile; +import com.datastax.oss.driver.api.core.config.OptionsMap; +import com.datastax.oss.driver.api.core.config.TypedDriverOption; +import com.datastax.oss.driver.api.core.connection.ReconnectionPolicy; +import com.datastax.oss.driver.api.core.context.DriverContext; +import com.datastax.oss.driver.api.core.loadbalancing.LoadBalancingPolicy; +import com.datastax.oss.driver.api.core.retry.RetryPolicy; +import com.datastax.oss.driver.api.core.specex.SpeculativeExecutionPolicy; +import com.datastax.oss.driver.api.core.ssl.SslEngineFactory; +import com.datastax.oss.driver.api.core.time.TimestampGenerator; +import com.datastax.oss.driver.internal.core.connection.ConstantReconnectionPolicy; +import com.datastax.oss.driver.internal.core.connection.ExponentialReconnectionPolicy; +import com.datastax.oss.driver.internal.core.loadbalancing.DcInferringLoadBalancingPolicy; +import com.datastax.oss.driver.internal.core.loadbalancing.DefaultLoadBalancingPolicy; +import com.datastax.oss.driver.internal.core.retry.ConsistencyDowngradingRetryPolicy; +import com.datastax.oss.driver.internal.core.retry.DefaultRetryPolicy; +import com.datastax.oss.driver.internal.core.specex.ConstantSpeculativeExecutionPolicy; +import com.datastax.oss.driver.internal.core.specex.NoSpeculativeExecutionPolicy; +import com.datastax.oss.driver.internal.core.time.ServerSideTimestampGenerator; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.function.Consumer; +import org.junit.Before; +import org.junit.Test; + +public class DefaultDriverConfigReporterTest { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + // ---- Fixtures for the gating / fail-safe tests (bare mock profile) ---- + private InternalDriverContext mockContext; + private DriverExecutionProfile mockProfile; + private DefaultDriverConfigReporter reporter; + + @Before + public void setup() { + mockContext = mock(InternalDriverContext.class); + DriverConfig config = mock(DriverConfig.class); + mockProfile = mock(DriverExecutionProfile.class); + when(mockContext.getConfig()).thenReturn(config); + when(config.getDefaultProfile()).thenReturn(mockProfile); + reporter = new DefaultDriverConfigReporter(mockContext); + } + + private void enableReporting(boolean enabled) { + when(mockProfile.getBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, false)) + .thenReturn(enabled); + } + + // ==================== Gating ==================== + + @Test + public void should_add_nothing_when_disabled() { + enableReporting(false); + Map options = new HashMap<>(); + reporter.populateStartupOptions(options, /* reportDriverConfig= */ true); + assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.SESSION_ID_KEY); + assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); + } + + @Test + public void should_add_session_id_only_on_pool_connection() { + enableReporting(true); + Map options = new HashMap<>(); + reporter.populateStartupOptions(options, /* reportDriverConfig= */ false); + String sessionId = options.get(DefaultDriverConfigReporter.SESSION_ID_KEY); + assertThat(sessionId).isNotNull(); + assertThat(UUID.fromString(sessionId)).isNotNull(); // valid UUID + assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); + } + + @Test + public void should_use_a_stable_session_id_across_connections() { + enableReporting(true); + Map control = new HashMap<>(); + Map pool = new HashMap<>(); + reporter.populateStartupOptions(control, false); + reporter.populateStartupOptions(pool, false); + assertThat(pool.get(DefaultDriverConfigReporter.SESSION_ID_KEY)) + .isEqualTo(control.get(DefaultDriverConfigReporter.SESSION_ID_KEY)); + } + + @Test + public void should_use_a_distinct_session_id_per_reporter() { + enableReporting(true); + Map first = new HashMap<>(); + reporter.populateStartupOptions(first, false); + Map second = new HashMap<>(); + new DefaultDriverConfigReporter(mockContext).populateStartupOptions(second, false); + assertThat(second.get(DefaultDriverConfigReporter.SESSION_ID_KEY)) + .isNotEqualTo(first.get(DefaultDriverConfigReporter.SESSION_ID_KEY)); + } + + // ==================== Fail-safe ==================== + + @Test + public void should_not_throw_when_reading_the_flag_fails() { + when(mockProfile.getBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, false)) + .thenThrow(new IllegalStateException("config blew up")); + Map options = new HashMap<>(); + reporter.populateStartupOptions(options, true); // must not throw + assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.SESSION_ID_KEY); + assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); + } + + @Test + public void should_keep_session_id_but_skip_driver_config_when_building_fails() { + enableReporting(true); + DefaultDriverConfigReporter throwing = + new DefaultDriverConfigReporter(mockContext) { + @Override + protected String buildJson() { + throw new IllegalStateException("introspection blew up"); + } + }; + Map options = new HashMap<>(); + throwing.populateStartupOptions(options, true); // must not throw + assertThat(options).containsKey(DefaultDriverConfigReporter.SESSION_ID_KEY); + assertThat(options).doesNotContainKey(DefaultDriverConfigReporter.DRIVER_CONFIG_KEY); + } + + // ==================== Report content ==================== + + @Test + public void should_report_default_configuration() throws Exception { + JsonNode report = report(defaultsReporter(map -> {})); + + assertThat(report.get("version").asInt()).isEqualTo(DefaultDriverConfigReporter.SCHEMA_VERSION); + + // Groups always present for the default profile. + for (String group : + new String[] { + "connection", + "socket", + "control-plane", + "reconnection-policy", + "retry-policy", + "load-balancing-policy", + "node-location-preference", + "connection-pool", + "query-defaults", + "tls" + }) { + assertThat(report.has(group)).as("group %s present", group).isTrue(); + } + // No speculative execution policy configured by default: the group has no null variant in + // the schema, so it is omitted entirely rather than reported as null. + assertThat(report.has("speculative-execution-policy")).isFalse(); + + JsonNode connection = report.get("connection"); + assertThat(connection.get("connect").get("timeout-ms").asLong()).isPositive(); + // No socket-level read/write timeout, and connection.heartbeat has no schema slot yet: all + // three are omitted rather than present-with-null/empty. + assertThat(connection.has("read")).isFalse(); + assertThat(connection.has("write")).isFalse(); + assertThat(connection.has("heartbeat")).isFalse(); + + JsonNode socket = report.get("socket"); + assertThat(socket.get("tcp-no-delay").asBoolean()).isTrue(); + assertThat(socket.get("keep-alive").asBoolean()).isFalse(); + assertThat(socket.has("linger")).isFalse(); + assertThat(socket.has("receive-buffer")).isFalse(); + assertThat(socket.has("send-buffer")).isFalse(); + + // Java-specific reconciliation: no server-side (USING TIMEOUT) internal-query timeout. + JsonNode controlPlane = report.get("control-plane"); + assertThat(controlPlane.get("system-queries").get("timeout").get("client-side-ms").asLong()) + .isPositive(); + assertThat(controlPlane.get("system-queries").get("timeout").has("server-side-ms")).isFalse(); + assertThat(controlPlane.get("schema-agreement").get("timeout-ms").asLong()).isPositive(); + + JsonNode reconnection = report.get("reconnection-policy"); + assertThat(reconnection.get("type").asText()).isEqualTo("exponential"); + assertThat(reconnection.get("base-ms").asLong()).isPositive(); + assertThat(reconnection.get("max-ms").asLong()).isPositive(); + // Java's built-in reconnection policies are unbounded: max-attempts is omitted. + assertThat(reconnection.has("max-attempts")).isFalse(); + + JsonNode retry = report.get("retry-policy"); + assertThat(retry.get("type").asText()).isEqualTo("standard-error-aware"); + assertThat(retry.has("backoff")).isFalse(); + + JsonNode lb = report.get("load-balancing-policy"); + assertThat(lb.get("type").asText()).isEqualTo("default"); + assertThat(lb.get("token-aware").asBoolean()).isTrue(); + assertThat(lb.get("shuffle").asBoolean()).isFalse(); + assertThat(lb.get("latency-awareness").asBoolean()).isFalse(); + assertThat(lb.get("dc-failover").asBoolean()).isFalse(); + // local-dc/local-rack are no longer reported here; see node-location-preference below. + assertThat(lb.has("local-dc")).isFalse(); + assertThat(lb.has("local-rack")).isFalse(); + + // local-datacenter not configured in the defaults => DC is inferred (dc-auto), and the value + // isn't known yet at report time. + JsonNode nodeLocation = report.get("node-location-preference"); + assertThat(nodeLocation.get("type").asText()).isEqualTo("dc-auto"); + assertThat(nodeLocation.has("local-dc")).isFalse(); + + JsonNode pool = report.get("connection-pool"); + assertThat(pool.get("type").asText()).isEqualTo("host"); + assertThat(pool.get("desired-connections-count").asInt()).isPositive(); + assertThat(pool.get("shard-aware").get("enabled").asBoolean()).isTrue(); + + JsonNode query = report.get("query-defaults"); + assertThat(query.get("consistency").asText()).isEqualTo("LOCAL_ONE"); + assertThat(query.get("idempotence").asBoolean()).isFalse(); + assertThat(query.get("client-timestamps").asBoolean()).isTrue(); + assertThat(query.get("request").get("timeout-ms").asLong()).isPositive(); + assertThat(query.get("page").get("size").asInt()).isPositive(); + + JsonNode tls = report.get("tls"); + assertThat(tls.get("enabled").asBoolean()).isFalse(); + assertThat(tls.get("hostname-verification").asBoolean()).isFalse(); + } + + @Test + public void should_report_constant_reconnection_policy() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ConstantReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode reconnection = report(r).get("reconnection-policy"); + assertThat(reconnection.get("type").asText()).isEqualTo("constant"); + assertThat(reconnection.get("delay-ms").asLong()).isPositive(); + assertThat(reconnection.has("max-attempts")).isFalse(); + } + + @Test + public void should_report_custom_reconnection_policy() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ReconnectionPolicy.class), // neither exponential nor constant + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode reconnection = report(r).get("reconnection-policy"); + assertThat(reconnection.get("type").asText()).isEqualTo("custom"); + assertThat(reconnection.get("name").asText()).isNotEmpty(); + } + + @Test + public void should_report_a_reconnection_policy_subclass_as_custom() throws Exception { + // A real (anonymous) subclass of a built-in, not a mock: proves the exact-class check doesn't + // misclassify user customizations of a built-in as the plain built-in + // (ConstantReconnectionPolicy + // is not final, and a real subclass of it already exists elsewhere in this repo's test code). + ReconnectionPolicy subclass = new ConstantReconnectionPolicy(policyConstructionContext()) {}; + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + subclass, + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode reconnection = report(r).get("reconnection-policy"); + assertThat(reconnection.get("type").asText()).isEqualTo("custom"); + // Also exercises the anonymous-class name fallback: getSimpleName() is empty for an anonymous + // class, so the reported name must fall back to the (non-empty) binary class name. + assertThat(reconnection.get("name").asText()).isNotEmpty(); + } + + @Test + public void should_report_downgrading_consistency_retry_policy() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(ConsistencyDowngradingRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + assertThat(report(r).get("retry-policy").get("type").asText()) + .isEqualTo("downgrading-consistency"); + } + + @Test + public void should_report_a_retry_policy_subclass_as_custom() throws Exception { + // Real (anonymous) subclass, not a mock: DefaultRetryPolicy is not final, and a real subclass + // of it already exists elsewhere in this repo's test code (osgi-tests' CustomRetryPolicy). + RetryPolicy subclass = new DefaultRetryPolicy(policyConstructionContext(), "default") {}; + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + subclass, + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode retry = report(r).get("retry-policy"); + assertThat(retry.get("type").asText()).isEqualTo("custom"); + assertThat(retry.get("name").asText()).isNotEmpty(); + } + + @Test + public void should_report_constant_speculative_execution_policy() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults( + map -> { + map.put(TypedDriverOption.SPECULATIVE_EXECUTION_MAX, 3); + map.put(TypedDriverOption.SPECULATIVE_EXECUTION_DELAY, Duration.ofMillis(100)); + }), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(ConstantSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode spec = report(r).get("speculative-execution-policy"); + assertThat(spec.get("type").asText()).isEqualTo("constant"); + assertThat(spec.get("max-executions").asInt()).isEqualTo(3); + assertThat(spec.get("delay-ms").asLong()).isEqualTo(100); + } + + @Test + public void should_report_a_speculative_execution_policy_subclass_as_custom() throws Exception { + // Real (anonymous) subclass, not a mock: NoSpeculativeExecutionPolicy is not final. A subclass + // must be reported as "custom", not silently treated the same as "no policy" (which would drop + // the whole group). + SpeculativeExecutionPolicy subclass = + new NoSpeculativeExecutionPolicy(policyConstructionContext(), "default") {}; + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + subclass, + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode spec = report(r).get("speculative-execution-policy"); + assertThat(spec.get("type").asText()).isEqualTo("custom"); + assertThat(spec.get("name").asText()).isNotEmpty(); + } + + @Test + public void should_report_dc_inferring_load_balancing_policy() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DcInferringLoadBalancingPolicy.class), // extends DefaultLoadBalancingPolicy + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode lb = report(r).get("load-balancing-policy"); + // Must be its own "dc-inferring" type, not misclassified as "default" via instanceof. + assertThat(lb.get("type").asText()).isEqualTo("dc-inferring"); + assertThat(lb.get("token-aware").asBoolean()).isTrue(); + } + + @Test + public void should_report_custom_load_balancing_policy() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(LoadBalancingPolicy.class), // not the default policy + mock(TimestampGenerator.class), + Optional.empty()); + JsonNode lb = report(r).get("load-balancing-policy"); + assertThat(lb.get("type").asText()).isEqualTo("custom"); + assertThat(lb.get("name").asText()).isNotEmpty(); + } + + @Test + public void should_report_explicit_local_dc() throws Exception { + DefaultDriverConfigReporter r = + defaultsReporter(map -> map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_DATACENTER, "dc1")); + JsonNode nodeLocation = report(r).get("node-location-preference"); + assertThat(nodeLocation.get("type").asText()).isEqualTo("dc"); + assertThat(nodeLocation.get("local-dc").asText()).isEqualTo("dc1"); + assertThat(nodeLocation.has("local-rack")).isFalse(); + } + + @Test + public void should_report_explicit_local_dc_and_rack() throws Exception { + DefaultDriverConfigReporter r = + defaultsReporter( + map -> { + map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_DATACENTER, "dc1"); + map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_RACK, "rack1"); + }); + JsonNode nodeLocation = report(r).get("node-location-preference"); + assertThat(nodeLocation.get("type").asText()).isEqualTo("rack"); + assertThat(nodeLocation.get("local-dc").asText()).isEqualTo("dc1"); + assertThat(nodeLocation.get("local-rack").asText()).isEqualTo("rack1"); + } + + @Test + public void should_report_local_dc_set_via_session_builder() throws Exception { + // SessionBuilder.withLocalDatacenter(...), not the config option: surfaced through + // InternalDriverContext.getLocalDatacenter(), which the reporter must consult. + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty(), + /* programmaticLocalDc= */ "dc-programmatic"); + JsonNode nodeLocation = report(r).get("node-location-preference"); + assertThat(nodeLocation.get("type").asText()).isEqualTo("dc"); + assertThat(nodeLocation.get("local-dc").asText()).isEqualTo("dc-programmatic"); + } + + @Test + public void should_prefer_programmatic_local_dc_over_config_option() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults( + map -> map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_DATACENTER, "dc-config")), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty(), + /* programmaticLocalDc= */ "dc-programmatic"); + JsonNode nodeLocation = report(r).get("node-location-preference"); + assertThat(nodeLocation.get("type").asText()).isEqualTo("dc"); + assertThat(nodeLocation.get("local-dc").asText()).isEqualTo("dc-programmatic"); + } + + @Test + public void should_report_rack_auto_when_only_rack_is_configured() throws Exception { + // Rack configured explicitly, but no DC (neither programmatically nor via config): the DC will + // be inferred, so this must not silently drop the explicitly-configured rack. + DefaultDriverConfigReporter r = + defaultsReporter(map -> map.put(TypedDriverOption.LOAD_BALANCING_LOCAL_RACK, "rack1")); + JsonNode nodeLocation = report(r).get("node-location-preference"); + assertThat(nodeLocation.get("type").asText()).isEqualTo("rack-auto"); + assertThat(nodeLocation.get("local-rack").asText()).isEqualTo("rack1"); + assertThat(nodeLocation.has("local-dc")).isFalse(); + } + + @Test + public void should_report_server_side_timestamps_as_disabled_client_timestamps() + throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> {}), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(ServerSideTimestampGenerator.class), + Optional.empty()); + assertThat(report(r).get("query-defaults").get("client-timestamps").asBoolean()).isFalse(); + } + + @Test + public void should_report_tls_enabled_with_hostname_verification() throws Exception { + DefaultDriverConfigReporter r = + reporterWith( + defaults(map -> map.put(TypedDriverOption.SSL_HOSTNAME_VALIDATION, true)), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.of(mock(SslEngineFactory.class))); + JsonNode tls = report(r).get("tls"); + assertThat(tls.get("enabled").asBoolean()).isTrue(); + assertThat(tls.get("hostname-verification").asBoolean()).isTrue(); + } + + @Test + public void should_report_socket_overrides() throws Exception { + DefaultDriverConfigReporter r = + defaultsReporter( + map -> { + map.put(TypedDriverOption.SOCKET_KEEP_ALIVE, true); + map.put(TypedDriverOption.SOCKET_RECEIVE_BUFFER_SIZE, 65535); + map.put(TypedDriverOption.SOCKET_LINGER_INTERVAL, 5); + }); + JsonNode socket = report(r).get("socket"); + assertThat(socket.get("keep-alive").asBoolean()).isTrue(); + assertThat(socket.get("receive-buffer").get("size-bytes").asInt()).isEqualTo(65535); + assertThat(socket.get("linger").get("interval-s").asInt()).isEqualTo(5); + } + + @Test + public void should_omit_page_when_unbounded() throws Exception { + DefaultDriverConfigReporter r = + defaultsReporter(map -> map.put(TypedDriverOption.REQUEST_PAGE_SIZE, 0)); + // The schema has no "unbounded" sentinel: the whole page group is omitted instead. + assertThat(report(r).get("query-defaults").has("page")).isFalse(); + } + + @Test + public void should_report_bounded_page_size() throws Exception { + DefaultDriverConfigReporter r = + defaultsReporter(map -> map.put(TypedDriverOption.REQUEST_PAGE_SIZE, 5000)); + assertThat(report(r).get("query-defaults").get("page").get("size").asInt()).isEqualTo(5000); + } + + // ==================== helpers ==================== + + private JsonNode report(DefaultDriverConfigReporter reporter) throws Exception { + return MAPPER.readTree(reporter.buildJson()); + } + + /** A real default execution profile with the given customizations applied. */ + private DriverExecutionProfile defaults(Consumer customizer) { + OptionsMap map = OptionsMap.driverDefaults(); + customizer.accept(map); + return DriverConfigLoader.fromMap(map).getInitialConfig().getDefaultProfile(); + } + + /** Reporter over default config + the Java-default policy set. */ + private DefaultDriverConfigReporter defaultsReporter(Consumer customizer) { + return reporterWith( + defaults(customizer), + mock(ExponentialReconnectionPolicy.class), + mock(DefaultRetryPolicy.class), + mock(NoSpeculativeExecutionPolicy.class), + mock(DefaultLoadBalancingPolicy.class), + mock(TimestampGenerator.class), + Optional.empty()); + } + + private DefaultDriverConfigReporter reporterWith( + DriverExecutionProfile profile, + ReconnectionPolicy reconnection, + RetryPolicy retry, + SpeculativeExecutionPolicy speculative, + LoadBalancingPolicy loadBalancing, + TimestampGenerator timestamps, + Optional ssl) { + return reporterWith( + profile, reconnection, retry, speculative, loadBalancing, timestamps, ssl, null); + } + + /** Same as the 7-arg overload, with an optional programmatic ({@code withLocalDatacenter}) DC. */ + private DefaultDriverConfigReporter reporterWith( + DriverExecutionProfile profile, + ReconnectionPolicy reconnection, + RetryPolicy retry, + SpeculativeExecutionPolicy speculative, + LoadBalancingPolicy loadBalancing, + TimestampGenerator timestamps, + Optional ssl, + String programmaticLocalDc) { + InternalDriverContext ctx = mock(InternalDriverContext.class); + DriverConfig config = mock(DriverConfig.class); + when(ctx.getConfig()).thenReturn(config); + when(config.getDefaultProfile()).thenReturn(profile); + when(ctx.getReconnectionPolicy()).thenReturn(reconnection); + when(ctx.getRetryPolicy(DriverExecutionProfile.DEFAULT_NAME)).thenReturn(retry); + when(ctx.getSpeculativeExecutionPolicy(DriverExecutionProfile.DEFAULT_NAME)) + .thenReturn(speculative); + when(ctx.getLoadBalancingPolicy(DriverExecutionProfile.DEFAULT_NAME)).thenReturn(loadBalancing); + when(ctx.getTimestampGenerator()).thenReturn(timestamps); + when(ctx.getSslEngineFactory()).thenReturn(ssl); + when(ctx.getLocalDatacenter(DriverExecutionProfile.DEFAULT_NAME)) + .thenReturn(programmaticLocalDc); + return new DefaultDriverConfigReporter(ctx); + } + + /** A minimal {@link DriverContext} good enough to construct a real built-in policy instance. */ + private DriverContext policyConstructionContext() { + DriverContext ctx = mock(DriverContext.class); + DriverConfig config = mock(DriverConfig.class); + DriverExecutionProfile profile = defaults(map -> {}); + when(ctx.getConfig()).thenReturn(config); + when(config.getDefaultProfile()).thenReturn(profile); + when(config.getProfile(DriverExecutionProfile.DEFAULT_NAME)).thenReturn(profile); + when(ctx.getSessionName()).thenReturn("test-session"); + return ctx; + } +} diff --git a/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingCcmIT.java b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingCcmIT.java new file mode 100644 index 00000000000..a4546b2efa6 --- /dev/null +++ b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingCcmIT.java @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.core.config; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.cql.Row; +import com.datastax.oss.driver.api.testinfra.ccm.CcmRule; +import com.datastax.oss.driver.api.testinfra.requirement.BackendRequirement; +import com.datastax.oss.driver.api.testinfra.requirement.BackendType; +import com.datastax.oss.driver.api.testinfra.session.SessionRule; +import com.datastax.oss.driver.api.testinfra.session.SessionUtils; +import com.datastax.oss.driver.categories.ParallelizableTests; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.rules.RuleChain; +import org.junit.rules.TestRule; + +/** + * Verifies client-configuration reporting end-to-end against a live ScyllaDB (via CCM), by reading + * back what the server actually stored in {@code system.clients.client_options}. + * + *

This is the real-server counterpart of {@code ClientConfigReportingSimulacronIT}: it confirms + * (a) that ScyllaDB accepts the extra {@code STARTUP} keys and stores them, (b) that {@code + * SESSION_ID} is present on every one of the session's connections with a single shared value, and + * (c) that {@code DRIVER_CONFIG} is stored for exactly one connection (the control connection). + * + *

ScyllaDB-only: {@code system.clients.client_options} is a Scylla feature. + */ +@Category(ParallelizableTests.class) +@BackendRequirement( + type = BackendType.SCYLLA, + minInclusive = "2026.1.0", + description = "system.clients.client_options is a ScyllaDB feature") +public class ClientConfigReportingCcmIT { + + private static final String DRIVER_NAME = "ScyllaDB Java Driver"; + + private static final CcmRule CCM_RULE = CcmRule.getInstance(); + + private static final SessionRule SESSION_RULE = + SessionRule.builder(CCM_RULE) + .withConfigLoader( + SessionUtils.configLoaderBuilder() + .withBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, true) + .withDuration(DefaultDriverOption.REQUEST_TIMEOUT, Duration.ofSeconds(30)) + .build()) + .build(); + + @ClassRule + public static final TestRule CHAIN = RuleChain.outerRule(CCM_RULE).around(SESSION_RULE); + + @Test + public void should_store_session_id_on_all_connections_and_driver_config_on_control() { + CqlSession session = SESSION_RULE.session(); + + // The session opens a control connection plus at least one pooled connection; wait until the + // server reflects them (system.clients is updated asynchronously as connections are set up). + await() + .atMost(60, TimeUnit.SECONDS) + .pollInterval(500, TimeUnit.MILLISECONDS) + .until(() -> driverConnections(session).size() >= 2); + + List rows = driverConnections(session); + + // Evidence for the record: dump exactly what the server stored per connection. + System.out.println("system.clients rows for this driver session:"); + for (Row row : rows) { + System.out.printf( + " %s:%s stage=%s client_options=%s%n", + row.getObject("address"), + row.getObject("port"), + row.getString("connection_stage"), + row.getMap("client_options", String.class, String.class)); + } + + // (b) Every connection carries SESSION_ID, and all of them share a single value (one session). + Set sessionIds = + rows.stream().map(row -> clientOptions(row).get("SESSION_ID")).collect(Collectors.toSet()); + assertThat(sessionIds).doesNotContainNull().hasSize(1); + + // (c) DRIVER_CONFIG is stored for exactly one connection (the control connection), and it is + // the + // compact JSON blob the driver reports. + List driverConfigs = + rows.stream() + .map(row -> clientOptions(row).get("DRIVER_CONFIG")) + .filter(java.util.Objects::nonNull) + .collect(Collectors.toList()); + assertThat(driverConfigs).hasSize(1); + assertThat(driverConfigs.get(0)).contains("\"version\""); + } + + /** + * The rows in {@code system.clients} that belong to this driver session's connections: this + * driver, in a {@code READY} state, and carrying the reporting {@code SESSION_ID}. Transient + * protocol-version negotiation attempts (no driver identity, closed immediately) are excluded, + * and their absence here is itself the confirmation that they leave no lingering session rows. + */ + private List driverConnections(CqlSession session) { + return session + .execute( + "SELECT address, port, connection_stage, driver_name, client_options FROM system.clients") + .all() + .stream() + .filter(row -> DRIVER_NAME.equals(row.getString("driver_name"))) + .filter(row -> "READY".equals(row.getString("connection_stage"))) + .filter(row -> clientOptions(row).containsKey("SESSION_ID")) + .collect(Collectors.toList()); + } + + private Map clientOptions(Row row) { + Map options = row.getMap("client_options", String.class, String.class); + return options == null ? java.util.Collections.emptyMap() : options; + } +} diff --git a/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingSimulacronIT.java b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingSimulacronIT.java new file mode 100644 index 00000000000..1003686313a --- /dev/null +++ b/integration-tests/src/test/java/com/datastax/oss/driver/core/config/ClientConfigReportingSimulacronIT.java @@ -0,0 +1,179 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.core.config; + +import static com.datastax.oss.driver.internal.core.context.DefaultDriverConfigReporter.DRIVER_CONFIG_KEY; +import static com.datastax.oss.driver.internal.core.context.DefaultDriverConfigReporter.SESSION_ID_KEY; +import static com.datastax.oss.driver.internal.core.context.StartupOptionsBuilder.CLIENT_ID_KEY; +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +import com.datastax.oss.driver.api.testinfra.session.SessionUtils; +import com.datastax.oss.driver.api.testinfra.simulacron.SimulacronRule; +import com.datastax.oss.driver.categories.ParallelizableTests; +import com.datastax.oss.protocol.internal.request.Register; +import com.datastax.oss.protocol.internal.request.Startup; +import com.datastax.oss.simulacron.common.cluster.ClusterSpec; +import com.datastax.oss.simulacron.common.cluster.QueryLog; +import java.net.SocketAddress; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * End-to-end check of client-configuration reporting against a mock server, asserting on the actual + * CQL {@code STARTUP} frames the driver sends. + * + *

Simulacron records every inbound frame with its originating client connection, so we can + * verify that when {@code advanced.client-config-reporting.enabled} is: + * + *

+ * + *

The control connection is identified independently of the reported options: it is the only + * connection that issues a {@code REGISTER} frame (to subscribe to cluster events). + * + *

Assertions are scoped to the session's real connections, identified by the {@code CLIENT_ID} + * startup option that the driver always sends (independently of config reporting). This excludes + * the short-lived connections opened by protocol-version negotiation: when the protocol version is + * not pinned (the default), the driver first tries higher versions (DSE_V2, DSE_V1, V5) that + * ScyllaDB rejects; those attempts are torn down during negotiation before the driver ever reaches + * the point of sending a {@code STARTUP} that carries its full identifying options (including + * {@code CLIENT_ID}), so filtering on {@code CLIENT_ID}'s presence naturally excludes them. + */ +@Category(ParallelizableTests.class) +public class ClientConfigReportingSimulacronIT { + + // A single node yields one dedicated control connection plus a pool connection (local.size + // defaults to 1), i.e. at least two distinct session connections of which only the control one + // registers for events. + @ClassRule + public static final SimulacronRule SIMULACRON_RULE = + new SimulacronRule(ClusterSpec.builder().withNodes(1)); + + @Before + public void clearLogs() { + SIMULACRON_RULE.cluster().clearLogs(); + } + + @Test + public void should_report_session_id_on_all_connections_and_driver_config_only_on_control() { + DriverConfigLoader loader = + SessionUtils.configLoaderBuilder() + .withBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, true) + .build(); + try (CqlSession session = SessionUtils.newSession(SIMULACRON_RULE, loader)) { + awaitControlAndPoolConnected(); + + SocketAddress controlConnection = controlConnection().orElseThrow(AssertionError::new); + List startups = sessionStartups(); + + // Sanity: we are actually observing more than one connection (control + at least one pool), + // otherwise the "only on the control connection" assertion below would be vacuous. + assertThat(distinctConnections(startups)).isGreaterThanOrEqualTo(2); + + // SESSION_ID is present on every session connection and has the same value everywhere. + assertThat(startups).allSatisfy(log -> assertThat(options(log)).containsKey(SESSION_ID_KEY)); + assertThat(startups.stream().map(log -> options(log).get(SESSION_ID_KEY)).distinct()) + .hasSize(1); + + // DRIVER_CONFIG is present on exactly one connection, and that connection is the control one. + List withDriverConfig = + startups.stream() + .filter(log -> options(log).containsKey(DRIVER_CONFIG_KEY)) + .collect(Collectors.toList()); + assertThat(withDriverConfig).hasSize(1); + assertThat(withDriverConfig.get(0).getConnection()).isEqualTo(controlConnection); + } + } + + @Test + public void should_report_nothing_when_disabled() { + DriverConfigLoader loader = + SessionUtils.configLoaderBuilder() + .withBoolean(DefaultDriverOption.CLIENT_CONFIG_REPORTING_ENABLED, false) + .build(); + try (CqlSession session = SessionUtils.newSession(SIMULACRON_RULE, loader)) { + awaitControlAndPoolConnected(); + + List startups = sessionStartups(); + assertThat(distinctConnections(startups)).isGreaterThanOrEqualTo(2); + + // Neither option is sent on any session connection: zero change on the wire when disabled. + assertThat(startups) + .allSatisfy( + log -> + assertThat(options(log)) + .doesNotContainKey(SESSION_ID_KEY) + .doesNotContainKey(DRIVER_CONFIG_KEY)); + } + } + + /** + * The {@code STARTUP} frames of the session's real connections (control + pool), identified by + * the always-present {@code CLIENT_ID} option; excludes protocol-version negotiation attempts. + */ + private List sessionStartups() { + return SIMULACRON_RULE.cluster().getLogs().getQueryLogs().stream() + .filter(log -> log.getFrame().message instanceof Startup) + .filter(log -> options(log).containsKey(CLIENT_ID_KEY)) + .collect(Collectors.toList()); + } + + /** The {@code STARTUP} options carried by a recorded frame. */ + private Map options(QueryLog log) { + return ((Startup) log.getFrame().message).options; + } + + /** The connection that issued a {@code REGISTER} frame, i.e. the control connection. */ + private Optional controlConnection() { + return SIMULACRON_RULE.cluster().getLogs().getQueryLogs().stream() + .filter(log -> log.getFrame().message instanceof Register) + .map(QueryLog::getConnection) + .findFirst(); + } + + private long distinctConnections(List logs) { + return logs.stream().map(QueryLog::getConnection).distinct().count(); + } + + /** + * Waits until both the control connection (identified by its {@code REGISTER}) and at least one + * pool connection have sent their {@code STARTUP}, since pool connections may finish initializing + * slightly after the session builder returns. + */ + private void awaitControlAndPoolConnected() { + await() + .pollInterval(100, TimeUnit.MILLISECONDS) + .atMost(30, TimeUnit.SECONDS) + .until( + () -> controlConnection().isPresent() && distinctConnections(sessionStartups()) >= 2); + } +}