Skip to content
This repository was archived by the owner on Jul 22, 2026. It is now read-only.
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 24 additions & 6 deletions d2/src/main/java/com/linkedin/d2/balancer/D2ClientBuilder.java
Original file line number Diff line number Diff line change
Expand Up @@ -52,11 +52,13 @@
import com.linkedin.d2.discovery.event.ServiceDiscoveryEventEmitter;
import com.linkedin.d2.discovery.stores.zk.ZKPersistentConnection;
import com.linkedin.d2.discovery.stores.zk.ZooKeeper;
import com.linkedin.d2.jmx.XdsServerMetricsProvider;
import com.linkedin.d2.jmx.XdsClientOtelMetricsProvider;
import com.linkedin.d2.jmx.DegraderLoadBalancerStrategyV3OtelMetricsProvider;
import com.linkedin.d2.jmx.JmxManager;
import com.linkedin.d2.xds.XdsClientValidator;
import com.linkedin.d2.jmx.NoOpJmxManager;
import com.linkedin.d2.jmx.RelativeLoadBalancerStrategyOtelMetricsProvider;
import com.linkedin.d2.jmx.XdsClientOtelMetricsProvider;
import com.linkedin.d2.jmx.XdsServerMetricsProvider;
import com.linkedin.d2.xds.XdsClientValidator;
import com.linkedin.r2.transport.common.TransportClientFactory;
import com.linkedin.r2.transport.http.client.HttpClientFactory;
import com.linkedin.r2.util.NamedThreadFactory;
Expand Down Expand Up @@ -251,7 +253,9 @@ public D2Client build()
_config.d2CalleeInfoRecorder,
_config.enableIndisDownstreamServicesFetcher,
_config.indisDownstreamServicesFetchTimeout,
_config.xdsClientOtelMetricsProvider
_config.xdsClientOtelMetricsProvider,
_config.relativeLoadBalancerStrategyOtelMetricsProvider,
_config.degraderLoadBalancerStrategyV3OtelMetricsProvider
);

final LoadBalancerWithFacilitiesFactory loadBalancerFactory = (_config.lbWithFacilitiesFactory == null) ?
Expand Down Expand Up @@ -873,6 +877,18 @@ public D2ClientBuilder setXdsClientOtelMetricsProvider(XdsClientOtelMetricsProvi
return this;
}

public D2ClientBuilder setRelativeLoadBalancerStrategyOtelMetricsProvider(
RelativeLoadBalancerStrategyOtelMetricsProvider relativeLoadBalancerStrategyOtelMetricsProvider) {
_config.relativeLoadBalancerStrategyOtelMetricsProvider = relativeLoadBalancerStrategyOtelMetricsProvider;
return this;
}

public D2ClientBuilder setDegraderLoadBalancerStrategyV3OtelMetricsProvider(
DegraderLoadBalancerStrategyV3OtelMetricsProvider degraderLoadBalancerStrategyV3OtelMetricsProvider) {
_config.degraderLoadBalancerStrategyV3OtelMetricsProvider = degraderLoadBalancerStrategyV3OtelMetricsProvider;
return this;
}

public D2ClientBuilder setLoadBalanceStreamException(boolean loadBalanceStreamException) {
_config.loadBalanceStreamException = loadBalanceStreamException;
return this;
Expand Down Expand Up @@ -965,7 +981,8 @@ private Map<String, LoadBalancerStrategyFactory<?>> createDefaultLoadBalancerStr
loadBalancerStrategyFactories.putIfAbsent("random", randomStrategyFactory);

final DegraderLoadBalancerStrategyFactoryV3 degraderStrategyFactoryV3 = new DegraderLoadBalancerStrategyFactoryV3(
_config.healthCheckOperations, _config._executorService, _config.eventEmitter, Collections.emptyList());
_config.healthCheckOperations, _config._executorService, _config.eventEmitter, Collections.emptyList(),
_config.degraderLoadBalancerStrategyV3OtelMetricsProvider);
loadBalancerStrategyFactories.putIfAbsent("degrader", degraderStrategyFactoryV3);
loadBalancerStrategyFactories.putIfAbsent("degraderV2", degraderStrategyFactoryV3);
loadBalancerStrategyFactories.putIfAbsent("degraderV3", degraderStrategyFactoryV3);
Expand All @@ -976,7 +993,8 @@ private Map<String, LoadBalancerStrategyFactory<?>> createDefaultLoadBalancerStr
// TODO: create StateUpdater.LoadBalanceConfig and pass it to the RelativeLoadBalancerStrategyFactory
final RelativeLoadBalancerStrategyFactory relativeLoadBalancerStrategyFactory = new RelativeLoadBalancerStrategyFactory(
_config._executorService, _config.healthCheckOperations, Collections.emptyList(), _config.eventEmitter,
SystemClock.instance(), _config.loadBalanceStreamException, _config.enableRelativeStrategyDeferredAllocation);
SystemClock.instance(), _config.loadBalanceStreamException, _config.enableRelativeStrategyDeferredAllocation,
_config.relativeLoadBalancerStrategyOtelMetricsProvider);
loadBalancerStrategyFactories.putIfAbsent(RelativeLoadBalancerStrategy.RELATIVE_LOAD_BALANCER_STRATEGY_NAME,
relativeLoadBalancerStrategyFactory);
}
Expand Down
196 changes: 196 additions & 0 deletions d2/src/main/java/com/linkedin/d2/balancer/D2ClientConfig.java
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,10 @@
import com.linkedin.d2.jmx.NoOpJmxManager;
import com.linkedin.d2.jmx.XdsClientOtelMetricsProvider;
import com.linkedin.d2.jmx.NoOpXdsClientOtelMetricsProvider;
import com.linkedin.d2.jmx.DegraderLoadBalancerStrategyV3OtelMetricsProvider;
import com.linkedin.d2.jmx.NoOpDegraderLoadBalancerStrategyV3OtelMetricsProvider;
import com.linkedin.d2.jmx.RelativeLoadBalancerStrategyOtelMetricsProvider;
import com.linkedin.d2.jmx.NoOpRelativeLoadBalancerStrategyOtelMetricsProvider;
import com.linkedin.r2.transport.common.TransportClientFactory;
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
import java.time.Duration;
Expand All @@ -54,14 +58,23 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLParameters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.linkedin.d2.xds.XdsClientValidator.DEFAULT_MINIMUM_JAVA_VERSION;

public class D2ClientConfig
{
private static final Logger LOG = LoggerFactory.getLogger(D2ClientConfig.class);

// Guards so the deprecated-forwarder OTel-NoOp warnings are emitted at most once per JVM each.
private static final AtomicBoolean PRE_XDS_OTEL_FORWARDER_WARNED = new AtomicBoolean(false);
private static final AtomicBoolean PRE_STRATEGY_OTEL_FORWARDER_WARNED = new AtomicBoolean(false);

// default values for some configs, to be shared with other classes
public static final String D2_JMX_MANAGER_PREFIX_DEFAULT = "UnknownPrefix";
public static final int DEFAULT_RETRY_LIMIT = 3;
Expand Down Expand Up @@ -189,6 +202,19 @@ public class D2ClientConfig
* Defaults to no-op implementation; can be overridden to enable metric tracking.
*/
public XdsClientOtelMetricsProvider xdsClientOtelMetricsProvider = new NoOpXdsClientOtelMetricsProvider();

/**
* Provider for OpenTelemetry metrics collection for RelativeLoadBalancerStrategy operations.
* Defaults to no-op implementation; can be overridden to enable metric tracking.
*/
public RelativeLoadBalancerStrategyOtelMetricsProvider relativeLoadBalancerStrategyOtelMetricsProvider = new NoOpRelativeLoadBalancerStrategyOtelMetricsProvider();

/**
* Provider for OpenTelemetry metrics collection for DegraderLoadBalancerStrategyV3 operations.
* Defaults to no-op implementation; can be overridden to enable metric tracking.
*/
public DegraderLoadBalancerStrategyV3OtelMetricsProvider degraderLoadBalancerStrategyV3OtelMetricsProvider = new NoOpDegraderLoadBalancerStrategyV3OtelMetricsProvider();

public boolean loadBalanceStreamException = false;
public boolean enablePotentialClientsCache = false;
public boolean enableRelativeStrategyDeferredAllocation = false;
Expand Down Expand Up @@ -358,8 +384,25 @@ public D2ClientConfig()
enableIndisDownstreamServicesFetcher,
indisDownstreamServicesFetchTimeout,
new NoOpXdsClientOtelMetricsProvider());
// Deprecated ctor: all three OTel providers defaulted to NoOp; warn once per JVM.
if (PRE_XDS_OTEL_FORWARDER_WARNED.compareAndSet(false, true))
{
LOG.warn("Deprecated D2ClientConfig ctor: XDS, Relative LB, and Degrader LB OTel providers default to NoOp. "
+ "Use the ctor with all three providers or set them on D2ClientConfig / D2ClientBuilder.");
}
}

/**
* Backward-compatible overload preserving the constructor signature that took only an
* {@link XdsClientOtelMetricsProvider}, so that any in-package callers updated for that signature
* continue to compile now that the relative- and degrader-strategy OTel providers have been
* added. New callers should use the constructor that accepts all three OTel metrics providers.
*
* @deprecated Use the constructor that additionally takes
* {@link RelativeLoadBalancerStrategyOtelMetricsProvider} and
* {@link DegraderLoadBalancerStrategyV3OtelMetricsProvider}.
*/
@Deprecated
D2ClientConfig(String zkHosts,
String xdsServer,
String hostName,
Expand Down Expand Up @@ -445,6 +488,157 @@ public D2ClientConfig()
Boolean enableIndisDownstreamServicesFetcher,
Duration indisDownstreamServicesFetchTimeout,
XdsClientOtelMetricsProvider xdsClientOtelMetricsProvider)
{
this(zkHosts, xdsServer, hostName, zkSessionTimeoutInMs, zkStartupTimeoutInMs, lbWaitTimeout, lbWaitUnit,
flagFile, basePath, fsBasePath, indisFsBasePath, componentFactory, clientFactories, lbWithFacilitiesFactory,
sslContext, grpcSslContext, sslParameters, isSSLEnabled, shutdownAsynchronously, isSymlinkAware,
clientServicesConfig, d2ServicePath, useNewEphemeralStoreWatcher, healthCheckOperations, executorService,
retry, restRetryEnabled, streamRetryEnabled, retryLimit, retryUpdateIntervalMs, retryAggregatedIntervalNum,
warmUp, warmUpTimeoutSeconds, indisWarmUpTimeoutSeconds, warmUpConcurrentRequests,
indisWarmUpConcurrentRequests, downstreamServicesFetcher, indisDownstreamServicesFetcher,
backupRequestsEnabled, backupRequestsStrategyStatsConsumer,
backupRequestsLatencyNotificationInterval,
backupRequestsLatencyNotificationIntervalUnit,
enableBackupRequestsClientAsync,
backupRequestsExecutorService,
emitter,
partitionAccessorRegistry,
zooKeeperDecorator,
enableSaveUriDataOnDisk,
loadBalancerStrategyFactories,
requestTimeoutHandlerEnabled,
sslSessionValidatorFactory,
zkConnection,
startUpExecutorService,
indisStartUpExecutorService,
jmxManager,
d2JmxManagerPrefix,
zookeeperReadWindowMs,
enableRelativeLoadBalancer,
deterministicSubsettingMetadataProvider,
canaryDistributionProvider,
enableClusterFailout,
failoutConfigProviderFactory,
failoutRedirectStrategy,
serviceDiscoveryEventEmitter,
dualReadStateManager,
xdsExecutorService,
xdsStreamReadyTimeout,
dualReadNewLbExecutor,
xdsChannelLoadBalancingPolicy,
xdsChannelLoadBalancingPolicyConfig,
subscribeToUriGlobCollection,
xdsServerMetricsProvider,
loadBalanceStreamException,
enablePotentialClientsCache,
xdsInitialResourceVersionsEnabled,
disableDetectLiRawD2Client,
isLiRawD2Client,
xdsStreamMaxRetryBackoffSeconds,
xdsChannelKeepAliveTimeMins,
xdsMinimumJavaVersion,
actionOnPrecheckFailure,
d2CalleeInfoRecorder,
enableIndisDownstreamServicesFetcher,
indisDownstreamServicesFetchTimeout,
xdsClientOtelMetricsProvider,
new NoOpRelativeLoadBalancerStrategyOtelMetricsProvider(),
new NoOpDegraderLoadBalancerStrategyV3OtelMetricsProvider());
// Deprecated ctor: Relative/Degrader LB OTel providers defaulted to NoOp; warn once per JVM.
if (PRE_STRATEGY_OTEL_FORWARDER_WARNED.compareAndSet(false, true))
{
LOG.warn("Deprecated D2ClientConfig ctor: Relative and Degrader LB OTel providers default to NoOp. "
+ "Use the ctor that takes both providers or set them on D2ClientConfig / D2ClientBuilder.");
}
}

D2ClientConfig(String zkHosts,
String xdsServer,
String hostName,
long zkSessionTimeoutInMs,
long zkStartupTimeoutInMs,
long lbWaitTimeout,
TimeUnit lbWaitUnit,
String flagFile,
String basePath,
String fsBasePath,
String indisFsBasePath,
ComponentFactory componentFactory,
Map<String, TransportClientFactory> clientFactories,
LoadBalancerWithFacilitiesFactory lbWithFacilitiesFactory,
SSLContext sslContext,
SslContext grpcSslContext,
SSLParameters sslParameters,
boolean isSSLEnabled,
boolean shutdownAsynchronously,
boolean isSymlinkAware,
Map<String, Map<String, Object>> clientServicesConfig,
String d2ServicePath,
boolean useNewEphemeralStoreWatcher,
HealthCheckOperations healthCheckOperations,
ScheduledExecutorService executorService,
boolean retry,
boolean restRetryEnabled,
boolean streamRetryEnabled,
int retryLimit,
long retryUpdateIntervalMs,
int retryAggregatedIntervalNum,
boolean warmUp,
int warmUpTimeoutSeconds,
int indisWarmUpTimeoutSeconds,
int warmUpConcurrentRequests,
int indisWarmUpConcurrentRequests,
DownstreamServicesFetcher downstreamServicesFetcher,
DownstreamServicesFetcher indisDownstreamServicesFetcher,
boolean backupRequestsEnabled,
BackupRequestsStrategyStatsConsumer backupRequestsStrategyStatsConsumer,
long backupRequestsLatencyNotificationInterval,
TimeUnit backupRequestsLatencyNotificationIntervalUnit,
boolean enableBackupRequestsClientAsync,
ScheduledExecutorService backupRequestsExecutorService,
EventEmitter emitter,
PartitionAccessorRegistry partitionAccessorRegistry,
Function<ZooKeeper, ZooKeeper> zooKeeperDecorator,
boolean enableSaveUriDataOnDisk,
Map<String, LoadBalancerStrategyFactory<? extends LoadBalancerStrategy>> loadBalancerStrategyFactories,
boolean requestTimeoutHandlerEnabled,
SslSessionValidatorFactory sslSessionValidatorFactory,
ZKPersistentConnection zkConnection,
ScheduledExecutorService startUpExecutorService,
ScheduledExecutorService indisStartUpExecutorService,
JmxManager jmxManager,
String d2JmxManagerPrefix,
int zookeeperReadWindowMs,
boolean enableRelativeLoadBalancer,
DeterministicSubsettingMetadataProvider deterministicSubsettingMetadataProvider,
CanaryDistributionProvider canaryDistributionProvider,
boolean enableClusterFailout,
FailoutConfigProviderFactory failoutConfigProviderFactory,
FailoutRedirectStrategy failoutRedirectStrategy,
ServiceDiscoveryEventEmitter serviceDiscoveryEventEmitter,
DualReadStateManager dualReadStateManager,
ScheduledExecutorService xdsExecutorService,
Long xdsStreamReadyTimeout,
ExecutorService dualReadNewLbExecutor,
String xdsChannelLoadBalancingPolicy,
Map<String, ?> xdsChannelLoadBalancingPolicyConfig,
boolean subscribeToUriGlobCollection,
XdsServerMetricsProvider xdsServerMetricsProvider,
boolean loadBalanceStreamException,
boolean enablePotentialClientsCache,
boolean xdsInitialResourceVersionsEnabled,
boolean disableDetectLiRawD2Client,
boolean isLiRawD2Client,
Integer xdsStreamMaxRetryBackoffSeconds,
Long xdsChannelKeepAliveTimeMins,
String xdsMinimumJavaVersion,
XdsClientValidator.ActionOnPrecheckFailure actionOnPrecheckFailure,
D2CalleeInfoRecorder d2CalleeInfoRecorder,
Boolean enableIndisDownstreamServicesFetcher,
Duration indisDownstreamServicesFetchTimeout,
XdsClientOtelMetricsProvider xdsClientOtelMetricsProvider,
RelativeLoadBalancerStrategyOtelMetricsProvider relativeLoadBalancerStrategyOtelMetricsProvider,
DegraderLoadBalancerStrategyV3OtelMetricsProvider degraderLoadBalancerStrategyV3OtelMetricsProvider)
{
this.zkHosts = zkHosts;
this.xdsServer = xdsServer;
Expand Down Expand Up @@ -531,5 +725,7 @@ public D2ClientConfig()
this.indisDownstreamServicesFetchTimeout = indisDownstreamServicesFetchTimeout;
this.enableIndisDownstreamServicesFetcher = enableIndisDownstreamServicesFetcher;
this.xdsClientOtelMetricsProvider = xdsClientOtelMetricsProvider;
this.relativeLoadBalancerStrategyOtelMetricsProvider = relativeLoadBalancerStrategyOtelMetricsProvider;
this.degraderLoadBalancerStrategyV3OtelMetricsProvider = degraderLoadBalancerStrategyV3OtelMetricsProvider;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
/*
Copyright (c) 2026 LinkedIn Corp.

Licensed 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.linkedin.d2.balancer.clients;

/**
* Per-call duration listener for {@link TrackerClient}.
*
* <p>Primitive-{@code long} specialization of a {@code BiConsumer<Long, PerCallDurationSemantics>}
* to avoid boxing the duration on every completed call. Equivalent in spirit to
* {@link java.util.function.LongConsumer}, extended with a second
* {@link PerCallDurationSemantics} argument that the JDK has no built-in functional interface for.
*
* <p><b>Threading and contract.</b> The listener fires <em>synchronously</em> on the thread that
* completes the underlying transport call &mdash; for the REST path it runs immediately before the
* wrapped {@code TransportCallback} is invoked, and for the streaming path it runs from the entity
* stream's {@code onDone}/{@code onError} (typically a transport / event-loop thread). Therefore
* implementations <b>MUST NOT block</b> (no I/O, no locks held by slow code paths, no waiting on
* other threads) and <b>MUST NOT throw checked work</b> onto the caller. Any blocking behaviour
* here directly stalls request completion and downstream user callbacks.
*
* <p>Implementations should also be cheap and allocation-free where possible; they are on the
* per-request hot path. Exceptions thrown from {@link #accept(long, PerCallDurationSemantics)}
* are caught by {@link TrackerClient} and rate-limited in the logs &mdash; they will not propagate
* to the wrapped callback, but they will cause the duration sample to be dropped.
*/
@FunctionalInterface
public interface PerCallDurationListener
{
/**
* Records the perceived per-call duration for one completed call.
*
* @param durationMs duration in milliseconds the server was perceived to have contributed
* @param semantics what {@code durationMs} represents (full round-trip vs. streaming TTFB);
* see {@link PerCallDurationSemantics}
*/
void accept(long durationMs, PerCallDurationSemantics semantics);
}
Loading
Loading