CASSJAVA-92: Local DC provided for nodetool clientstats #2036
Conversation
lukasz-antoniak
force-pushed
the
CASSJAVA-92
branch
from
d637315 to
1afcf84
Compare
| DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER); // DC from configuration | ||
| } | ||
| return dc; | ||
| } |
There was a problem hiding this comment.
DefaultDriverContext already defines lazy instantiation for (and access to) the startup options map for a given run. Rather than splitting the logic for determining the contents of a STARTUP message between DefaultDriverContext and this class the majority of the logic in this class should be consolidated into the existing DefaultDriverContext methods.
There was a problem hiding this comment.
I have noticed that and though that other entries of the STARTUP options are added here. The justification of the logic would be that all "dedicated" properties for STARTUP message are lazily instantiated where you pointed out, whereas all properties taken from other components (e.g. compression) are automatically injected in build() method.
There was a problem hiding this comment.
Okay, as I look at this again you're right, there's something of a bifurcation here already. The entries returned by DefaultDriverContext.buildStartupOptions() are more static key/value pairs, mostly (exclusively?) pairs that were used by Insights. Nearly all of those should be removed as part of CASSJAVA-73; driver name and version will stay but the rest should disappear.
So how should we format this data? That question is still under discussion in CASSJAVA-92... we probably need to settle on what the data should look like and then adjust this impl accordingly.
| @@ -119,6 +122,12 @@ public Map<String, String> build() { | |||
| if (applicationVersion != null) { | |||
| builder.put(APPLICATION_VERSION_KEY, applicationVersion); | |||
| } | |||
| if (applicationLocalDc == null) { | |||
| applicationLocalDc = localDc(config); | |||
There was a problem hiding this comment.
config here is just the default profile but any profile can define a local datacenter... which in turn means we need to think about how execution profiles interact with this functionality.
I'm going to raise this point on the JIRA ticket; I think it warrants further discussion.
core/src/main/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilder.java
Outdated
Show resolved
Hide resolved
There was a problem hiding this comment.
I believe all my concerns have been addressed by changes in this PR and discussion on CASSJAVA-92.
| /** Returns the local datacenter name, if known; empty otherwise. */ | ||
| @Nullable | ||
| String getLocalDatacenter(); | ||
| } |
There was a problem hiding this comment.
Oooooh, I like this very much... being explicit about whether an LBP cares about this (and building that into the type system) seems desirable.
Note that we're not saying any specific LBP will take action in a particular way based on this information. Presumably all we can say of an LBP that implements this interface is that it cares about a "local" LBP in some way. That's what we're communicating back to the server... but it's worth noting that this information might be used in different ways by different load balancers.
| return first ? null : result.toString(); | ||
| } | ||
|
|
||
| private String getLocalDc(LoadBalancingPolicy loadBalancingPolicy) { |
There was a problem hiding this comment.
Not essential by any means, but probably a good place to replace null with an Optional return type
| result.append(entry.getKey()).append(": ").append(dc); | ||
| } | ||
| } | ||
| return first ? null : result.toString(); |
There was a problem hiding this comment.
Semi-nit: this is the kind of thing the streaming API is really good at:
// do not cache local DC as it can change within LBP implementation
localDcs().ifPresent(s -> builder.put(DRIVER_LOCAL_DC, s));
...
private Optional<String> localDcs() {
Joiner joiner = Joiner.on(": ");
Map<String, Optional<String>> lbpToDc = Maps.transformValues(context.getLoadBalancingPolicies(), this::getLocalDc);
if (lbpToDc.isEmpty())
return Optional.empty();
return Optional.of(lbpToDc.entrySet().stream()
.filter(e -> e.getValue().isPresent())
.map(entry -> joiner.join(entry.getKey(), entry.getValue().get()))
.collect(Collectors.joining(", ")));
}
private Optional<String> getLocalDc(LoadBalancingPolicy loadBalancingPolicy) {
return (loadBalancingPolicy instanceof LocalDcAwareLoadBalancingPolicy) ?
Optional.of(((LocalDcAwareLoadBalancingPolicy) loadBalancingPolicy).getLocalDatacenter()) :
Optional.empty();
}There was a problem hiding this comment.
I applied your changes and also added on more unit test to make sure that non-local aware LBPs are skipped.
core/src/main/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilder.java
Outdated
Show resolved
Hide resolved
|
@aratno If you have some cycles do you mind taking a look at this and giving me a second 👍 ? I know you and @smiklosovic are still in conversation about the server-side of this but I think we're in a good spot to at least call the client-side done. |
lukasz-antoniak
force-pushed
the
CASSJAVA-92
branch
from
f517822 to
3ce91e3
Compare
There was a problem hiding this comment.
Left a couple suggestions, also waiting for CI: https://ci-cassandra.apache.org/job/cassandra-java-driver/job/PR-2036/7/
| return Optional.of( | ||
| "{" | ||
| + lbpToBag.entrySet().stream() | ||
| .filter(e -> e.getValue().isPresent()) | ||
| .map( | ||
| entry -> | ||
| joiner.join(Strings.doubleQuote(entry.getKey()), entry.getValue().get())) | ||
| .collect(Collectors.joining(", ")) | ||
| + "}"); |
There was a problem hiding this comment.
Why not depend on Jackson to do the serialization?
I was thinking you could add a DriverBaggage class with String toJson() and static DriverBaggage put(String key, DriverBaggage value), and in StartupOptionsBuilder here add all the load balancing policies' baggage. Then we wouldn't have the implicit "String that's actually JSON" type.
There was a problem hiding this comment.
I completely agree with @aratno on this one. I'd actually go a bit further; I'd argue it's the responsibility of individual LBPs to return a collection of name/value pairs as a Map with absolutely no notion of JSON formatting. It's then the responsibility of StartupOptionsBuilder (or any tool that wants to format these values in any way, either as JSON or something else) to put them into the appropriate format for their use case.
As a proof-of-concept the following implementation passes the test you added to DseStartupOptionsBuilderTest @lukasz-antoniak . I'm not saying this has to be the impl... I'm just providing it as a concrete example of what I'm talking about (and I think what @aratno is talking about as well, although I don't want to speak for him):
diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/loadbalancing/LocalDcAwareLoadBalancingPolicy.java b/core/src/main/java/com/datastax/oss/driver/api/core/loadbalancing/LocalDcAwareLoadBalancingPolicy.java
index d5604b4bf..55ed157e6 100644
--- a/core/src/main/java/com/datastax/oss/driver/api/core/loadbalancing/LocalDcAwareLoadBalancingPolicy.java
+++ b/core/src/main/java/com/datastax/oss/driver/api/core/loadbalancing/LocalDcAwareLoadBalancingPolicy.java
@@ -20,6 +20,8 @@ package com.datastax.oss.driver.api.core.loadbalancing;
import edu.umd.cs.findbugs.annotations.NonNull;
import edu.umd.cs.findbugs.annotations.Nullable;
+import java.util.Map;
+
/** Load balancing policy taking into account local datacenter of the application. */
public interface LocalDcAwareLoadBalancingPolicy extends LoadBalancingPolicy {
@@ -29,5 +31,5 @@ public interface LocalDcAwareLoadBalancingPolicy extends LoadBalancingPolicy {
/** Returns JSON string containing all properties that impact C* node connectivity. */
@NonNull
- String getStartupConfiguration();
+ Map<String, ?> getStartupConfiguration();
}
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilder.java b/core/src/main/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilder.java
index f64472761..55412fcf5 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilder.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilder.java
@@ -23,16 +23,21 @@ import com.datastax.oss.driver.api.core.loadbalancing.LoadBalancingPolicy;
import com.datastax.oss.driver.api.core.loadbalancing.LocalDcAwareLoadBalancingPolicy;
import com.datastax.oss.driver.api.core.session.Session;
import com.datastax.oss.driver.api.core.uuid.Uuids;
+import com.datastax.oss.driver.internal.core.type.codec.extras.json.JsonCodec;
import com.datastax.oss.driver.internal.core.util.Strings;
import com.datastax.oss.driver.shaded.guava.common.base.Joiner;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
import com.datastax.oss.driver.shaded.guava.common.collect.Maps;
import com.datastax.oss.protocol.internal.request.Startup;
import com.datastax.oss.protocol.internal.util.collection.NullAllowingImmutableMap;
+import com.fasterxml.jackson.databind.ObjectMapper;
import edu.umd.cs.findbugs.annotations.Nullable;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
import net.jcip.annotations.Immutable;
@Immutable
@@ -154,21 +159,24 @@ public class StartupOptionsBuilder {
}
private Optional<String> driverBaggage() {
- Joiner joiner = Joiner.on(": ");
- Map<String, Optional<String>> lbpToBag =
- Maps.transformValues(context.getLoadBalancingPolicies(), this::getDriverBaggage);
- return Optional.of(
- "{"
- + lbpToBag.entrySet().stream()
- .filter(e -> e.getValue().isPresent())
- .map(
- entry ->
- joiner.join(Strings.doubleQuote(entry.getKey()), entry.getValue().get()))
- .collect(Collectors.joining(", "))
- + "}");
+ ImmutableMap.Builder builder = new ImmutableMap.Builder();
+ for (Map.Entry<String,LoadBalancingPolicy> entry : context.getLoadBalancingPolicies().entrySet()) {
+ this.getDriverBaggage(entry.getValue()).ifPresent(baggage -> {
+ builder.put(entry.getKey(), baggage);
+ });
+ }
+ ObjectMapper mapper = new ObjectMapper();
+ try {
+
+ return Optional.of(mapper.writeValueAsString(builder.build()));
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ return Optional.empty();
+ }
}
- private Optional<String> getDriverBaggage(LoadBalancingPolicy loadBalancingPolicy) {
+ private Optional<Map<String,?>> getDriverBaggage(LoadBalancingPolicy loadBalancingPolicy) {
if (loadBalancingPolicy instanceof LocalDcAwareLoadBalancingPolicy) {
LocalDcAwareLoadBalancingPolicy dcAwareLbp =
(LocalDcAwareLoadBalancingPolicy) loadBalancingPolicy;
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/BasicLoadBalancingPolicy.java b/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/BasicLoadBalancingPolicy.java
index 777fa66ce..e62c724db 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/BasicLoadBalancingPolicy.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/BasicLoadBalancingPolicy.java
@@ -46,9 +46,12 @@ import com.datastax.oss.driver.internal.core.util.collection.CompositeQueryPlan;
import com.datastax.oss.driver.internal.core.util.collection.LazyQueryPlan;
import com.datastax.oss.driver.internal.core.util.collection.QueryPlan;
import com.datastax.oss.driver.internal.core.util.collection.SimpleQueryPlan;
+import com.datastax.oss.driver.shaded.guava.common.base.Joiner;
import com.datastax.oss.driver.shaded.guava.common.base.Predicates;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
import com.datastax.oss.driver.shaded.guava.common.collect.Lists;
import com.datastax.oss.driver.shaded.guava.common.collect.Sets;
+import com.fasterxml.jackson.databind.ObjectMapper;
import edu.umd.cs.findbugs.annotations.NonNull;
import edu.umd.cs.findbugs.annotations.Nullable;
import java.nio.ByteBuffer;
@@ -65,6 +68,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntUnaryOperator;
import java.util.stream.Collectors;
import net.jcip.annotations.ThreadSafe;
+import org.apache.tinkerpop.shaded.kryo.util.ObjectMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -165,43 +169,20 @@ public class BasicLoadBalancingPolicy implements LocalDcAwareLoadBalancingPolicy
@NonNull
@Override
- public String getStartupConfiguration() {
- StringBuilder builder = new StringBuilder();
- builder
- .append("{")
- .append(Strings.doubleQuote(BasicLoadBalancingPolicy.class.getSimpleName()))
- .append(":")
- .append("{")
- .append(Strings.doubleQuote("localDc"))
- .append(":")
- .append(Strings.doubleQuoteNullable(localDc));
+ public Map<String, ?> getStartupConfiguration() {
+
+ ImmutableMap.Builder builder = new ImmutableMap.Builder<>();
+ builder.put("localDc", localDc);
if (!preferredRemoteDcs.isEmpty()) {
- builder
- .append(",")
- .append(Strings.doubleQuote("preferredRemoteDcs"))
- .append(":[")
- .append(
- preferredRemoteDcs.stream()
- .map(Strings::doubleQuote)
- .collect(Collectors.joining(", ")))
- .append("]");
+ builder.put("preferredRemoteDcs", preferredRemoteDcs);
}
if (allowDcFailoverForLocalCl) {
- builder
- .append(",")
- .append(Strings.doubleQuote("allowDcFailoverForLocalCl"))
- .append(":")
- .append(allowDcFailoverForLocalCl);
+ builder.put("allowDcFailoverForLocalCl", allowDcFailoverForLocalCl);
}
if (maxNodesPerRemoteDc > 0) {
- builder
- .append(",")
- .append(Strings.doubleQuote("maxNodesPerRemoteDc"))
- .append(":")
- .append(maxNodesPerRemoteDc);
+ builder.put("maxNodesPerRemoteDc", maxNodesPerRemoteDc);
}
- builder.append("}}");
- return builder.toString();
+ return ImmutableMap.of(BasicLoadBalancingPolicy.class.getSimpleName(), builder.build());
}
/** @return The nodes currently considered as live. */
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/DefaultLoadBalancingPolicy.java b/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/DefaultLoadBalancingPolicy.java
index 533239a6d..146b4b41c 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/DefaultLoadBalancingPolicy.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/DefaultLoadBalancingPolicy.java
@@ -34,6 +34,7 @@ import com.datastax.oss.driver.internal.core.util.ArrayUtils;
import com.datastax.oss.driver.internal.core.util.collection.QueryPlan;
import com.datastax.oss.driver.internal.core.util.collection.SimpleQueryPlan;
import com.datastax.oss.driver.shaded.guava.common.annotations.VisibleForTesting;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
import com.datastax.oss.driver.shaded.guava.common.collect.MapMaker;
import edu.umd.cs.findbugs.annotations.NonNull;
import edu.umd.cs.findbugs.annotations.Nullable;
@@ -353,12 +354,8 @@ public class DefaultLoadBalancingPolicy extends BasicLoadBalancingPolicy impleme
@NonNull
@Override
- public String getStartupConfiguration() {
- String result = super.getStartupConfiguration();
- result =
- result.replaceFirst(
- BasicLoadBalancingPolicy.class.getSimpleName(),
- DefaultLoadBalancingPolicy.class.getSimpleName());
- return result;
+ public Map<String, ?> getStartupConfiguration() {
+ Map<String, ?> parent = super.getStartupConfiguration();
+ return ImmutableMap.of(DefaultLoadBalancingPolicy.class.getSimpleName(), parent.get(BasicLoadBalancingPolicy.class.getSimpleName()));
}
}
diff --git a/core/src/test/java/com/datastax/dse/driver/internal/core/context/DseStartupOptionsBuilderTest.java b/core/src/test/java/com/datastax/dse/driver/internal/core/context/DseStartupOptionsBuilderTest.java
index 15e4bcbfc..99175265b 100644
--- a/core/src/test/java/com/datastax/dse/driver/internal/core/context/DseStartupOptionsBuilderTest.java
+++ b/core/src/test/java/com/datastax/dse/driver/internal/core/context/DseStartupOptionsBuilderTest.java
@@ -339,9 +339,9 @@ public class DseStartupOptionsBuilderTest {
assertThat(startup.options)
.containsEntry(
StartupOptionsBuilder.DRIVER_BAGGAGE,
- "{\"oltp\": {\"DefaultLoadBalancingPolicy\":{"
+ "{\"oltp\":{\"DefaultLoadBalancingPolicy\":{"
+ "\"localDc\":\"dc1\","
- + "\"preferredRemoteDcs\":[\"dc2\", \"dc3\"],"
+ + "\"preferredRemoteDcs\":[\"dc2\",\"dc3\"],"
+ "\"allowDcFailoverForLocalCl\":true,"
+ "\"maxNodesPerRemoteDc\":2}}}");
}There are some changes to the test code itself but they're only minor whitespace changes; the substance is still very much intact. Regardless this demonstrates the key aspect of the change; LBPs provide metadata as Maps and StartupOptionsBuilder handles the JSON conversion by passing everything off to Jackson.
| builder | ||
| .append("{") | ||
| .append(Strings.doubleQuote(BasicLoadBalancingPolicy.class.getSimpleName())) | ||
| .append(":") | ||
| .append("{") | ||
| .append(Strings.doubleQuote("localDc")) | ||
| .append(":") | ||
| .append(Strings.doubleQuoteNullable(localDc)); |
There was a problem hiding this comment.
Same comment as above about using Jackson for JSON encoding
| /** Returns JSON string containing all properties that impact C* node connectivity. */ | ||
| @NonNull | ||
| String getStartupConfiguration(); |
There was a problem hiding this comment.
See comment below about String vs DriverBaggage. Could hoist this up to LoadBalancingPolicy with an empty default as well.
|
So, I think this addresses the serialization changes referenced above. But we are seeing a host of IT failures with errors similar to the following: Underlying problem here seems to be a variation on something we discussed earlier; some startup options aren't available at context creation time (when StartupOptionsBuilder is created). Let's focus specifically on local DC; it's computed based on the nodes in the cluster and is computed when the LBP is initialized. But that doesn't happen until we create the connection pool which makes sense (given that we don't have full info about available nodes until much beyond that point). Upshot is I don't think we can handle local DC in StartupOptionsBuilder... it's gonna have to be computed elsewhere and added to the STARTUP message at generation time. |
|
I am not sure why all tests passed when I was running tests locally. Apologies for that. Indeed test |
|
Thanks for the fixes @lukasz-antoniak. I can confirm that these fixes address the ConnectionInitExceptions I mentioned in my prior message. There's still something that has been gnawing at me about this impl but despite a hefty amount of effort I haven't been able to cleanly articulate my concern. I think at the core it's the idea that we implement an extra interface (LocalDcAwareLoadBalancingPolicy) here to indicate that an LBP is aware of it's local DC, a change that in general I completely agree with. But then we say that this functionality can only be determined in some cases (i.e. if a user configures a DC in some way). This covers the default case but that's only due to the implementation of DefaultLoadBalancingPolicy. We have an extension of the default policy named DcInferringLoadBalancingPolicy which will try to "infer" the local DC (based on contact points) if nothing is specified. This LBP may or may not have a value... again, it depends entirely on the impl. It's a similar story for custom LBP impls. A reasonable question to ask here is "if we change LocalDcAwareLoadBalancingPolicy.getLocalDatacenter() to return an Optional rather than a String doesn't that address the problem?" Maybe it does... and I think that might be a very reasonable change. But it's not completely satisfying; we're still saying we have this whole interface because an LBP is aware of it's local DC... until it's not. I'm still not explaining this very well, and for that I apologize. @aratno does any of this resonate for you or am I just nuts? |
|
After delegating the logic of providing startup configuration to a implementations of |
| @@ -76,6 +76,12 @@ default Optional<RequestTracker> getRequestTracker() { | |||
| */ | |||
| void init(@NonNull Map<UUID, Node> nodes, @NonNull DistanceReporter distanceReporter); | |||
|
|
|||
| /** Returns map containing details that impact C* node connectivity. */ | |||
| @NonNull | |||
| default Optional<Map<String, ?>> getStartupConfiguration() { | |||
There was a problem hiding this comment.
Why not make this a Map and default to empty map, instead of using Optional?
Could also make DriverBaggage its own class, since it's used in a few different places and we expect it to be used even more than LoadBalancingPolicies.
There was a problem hiding this comment.
I changed the code to use a Map. For now I did not create new class, because once we use it in different places, maybe the lifecycle will be different. Let us change it once need comes.
lukasz-antoniak
force-pushed
the
CASSJAVA-92
branch
from
0484676 to
90a8bac
Compare
| @@ -119,6 +129,8 @@ public Map<String, String> build() { | |||
| if (applicationVersion != null) { | |||
| builder.put(APPLICATION_VERSION_KEY, applicationVersion); | |||
| } | |||
| // do not cache local DC as it can change within LBP implementation | |||
There was a problem hiding this comment.
Maybe I'm missing something but... isn't this exactly what we're doing here? We're in the middle of StartupOptionsBuilder.build() which is only called when DefaultDriverContext is building a new StartupOptions instance... and it only does that once (lazily, the first time it's needed). So if information becomes available later... how is it ever updated here?
Bad thing is the creation of Startup instances only takes whatever the context gives you for startup options... so there isn't a whole lot of room to work with here. You could probably fix this by making DefaultDriverContext.getStartupOptions() do something beyond just returning the static list of startup options... maybe you could query the LBPs there and update any information (if updates are to be had).
| + "\"localDc\":\"dc1\"," | ||
| + "\"preferredRemoteDcs\":[\"dc2\",\"dc3\"]," | ||
| + "\"allowDcFailoverForLocalCl\":true," | ||
| + "\"maxNodesPerRemoteDc\":2}}}"); | ||
| } | ||
| } |
There was a problem hiding this comment.
Semi-nit: there's nothing DSE-specific about this test that I can see. Shouldn't this just live in StartupOptionsBuilderTest?
There was a problem hiding this comment.
I don't know that I want to hold this up much longer but I believe the current impl still doesn't support the idea of refreshing LBP-provided information when it becomes available later. We can take a position that we simply don't care about that... but my understanding was that we did want to fix that and I don't think this does it quite yet.
There was a problem hiding this comment.
Sorry for the duplication... apparently I don't know how to use Github anymore. :(
I don't know that I want to hold this up much longer but I believe the current impl still doesn't support the idea of refreshing LBP-provided information when it becomes available later. We can take a position that we simply don't care about that... but my understanding was that we did want to fix that and I don't think this does it quite yet.
Fixes: CASSJAVA-92