Skip to content

Commit

Permalink
refactor: add support for client timeouts
Browse files Browse the repository at this point in the history
  • Loading branch information
@aaron-steinfeld
aaron-steinfeld committed Jun 7, 2024
1 parent 31fc31e commit 42d3589
Show file tree
Hide file tree
Showing 9 changed files with 177 additions and 108 deletions.
11 changes: 11 additions & 0 deletions object-store/src/main/java/org/hypertrace/config/objectstore/ClientConfig.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
package org.hypertrace.config.objectstore;

import java.time.Duration;
import lombok.Value;

@Value
public class ClientConfig {
Duration timeout;

public static ClientConfig DEFAULT = new ClientConfig(Duration.ofSeconds(5));
}
66 changes: 28 additions & 38 deletions ...re/src/main/java/org/hypertrace/config/objectstore/ContextuallyIdentifiedObjectStore.java
View file
Original file line number Diff line number Diff line change
@@ -1,41 +1,50 @@
package org.hypertrace.config.objectstore;

import com.google.protobuf.Value;
import io.grpc.Status;
import java.util.Optional;
import java.util.function.Function;
import org.hypertrace.config.service.change.event.api.ConfigChangeEventGenerator;
import org.hypertrace.config.service.v1.ConfigServiceGrpc.ConfigServiceBlockingStub;
import org.hypertrace.config.service.v1.ContextSpecificConfig;
import org.hypertrace.config.service.v1.GetConfigRequest;
import org.hypertrace.config.service.v1.GetConfigResponse;
import org.hypertrace.core.grpcutils.context.RequestContext;

public abstract class ContextuallyIdentifiedObjectStore<T> {
private final ConfigServiceBlockingStub configServiceBlockingStub;
private final String resourceNamespace;
private final String resourceName;
private final Optional<ConfigChangeEventGenerator> configChangeEventGenerator;
private final ClientConfig clientConfig;

protected ContextuallyIdentifiedObjectStore(
ConfigServiceBlockingStub configServiceBlockingStub,
String resourceNamespace,
String resourceName,
ConfigChangeEventGenerator configChangeEventGenerator) {
this(
configServiceBlockingStub,
resourceNamespace,
resourceName,
configChangeEventGenerator,
ClientConfig.DEFAULT);
}

protected ContextuallyIdentifiedObjectStore(
ConfigServiceBlockingStub configServiceBlockingStub,
String resourceNamespace,
String resourceName,
ConfigChangeEventGenerator configChangeEventGenerator,
ClientConfig clientConfig) {
this.configServiceBlockingStub = configServiceBlockingStub;
this.resourceNamespace = resourceNamespace;
this.resourceName = resourceName;
this.configChangeEventGenerator = Optional.of(configChangeEventGenerator);
this.configChangeEventGenerator = Optional.ofNullable(configChangeEventGenerator);
this.clientConfig = clientConfig;
}

protected ContextuallyIdentifiedObjectStore(
ConfigServiceBlockingStub configServiceBlockingStub,
String resourceNamespace,
String resourceName) {
this.configServiceBlockingStub = configServiceBlockingStub;
this.resourceNamespace = resourceNamespace;
this.resourceName = resourceName;
this.configChangeEventGenerator = Optional.empty();
this(configServiceBlockingStub, resourceNamespace, resourceName, null);
}

protected abstract Optional<T> buildDataFromValue(Value value);
Expand All @@ -59,32 +68,9 @@ private IdentifiedObjectStore<T> buildObjectStoreForContext(RequestContext conte
}

public Optional<ConfigObject<T>> getObject(RequestContext context) {
String configContext = this.getConfigContextFromRequestContext(context);
try {
GetConfigResponse getConfigResponse =
context.call(
() ->
this.configServiceBlockingStub.getConfig(
GetConfigRequest.newBuilder()
.setResourceName(this.resourceName)
.setResourceNamespace(this.resourceNamespace)
.addContexts(configContext)
.build()));

ContextSpecificConfig contextSpecificConfig =
ContextSpecificConfig.newBuilder()
.setContext(configContext)
.setConfig(getConfigResponse.getConfig())
.setCreationTimestamp(getConfigResponse.getCreationTimestamp())
.setUpdateTimestamp(getConfigResponse.getUpdateTimestamp())
.build();
return ConfigObjectImpl.tryBuild(contextSpecificConfig, this::buildDataFromValue);
} catch (Exception exception) {
if (Status.fromThrowable(exception).equals(Status.NOT_FOUND)) {
return Optional.empty();
}
throw exception;
}
return this.buildObjectStoreForContext(context)
.getObject(context, this.getConfigContextFromRequestContext(context))
.map(Function.identity());
}

public Optional<T> getData(RequestContext context) {
Expand All @@ -107,13 +93,17 @@ private class ContextAwareIdentifiedObjectStoreDelegate extends IdentifiedObject
private final RequestContext requestContext;

ContextAwareIdentifiedObjectStoreDelegate(RequestContext requestContext) {
super(configServiceBlockingStub, resourceNamespace, resourceName);
this.requestContext = requestContext;
this(requestContext, null);
}

ContextAwareIdentifiedObjectStoreDelegate(
RequestContext requestContext, ConfigChangeEventGenerator configChangeEventGenerator) {
super(configServiceBlockingStub, resourceNamespace, resourceName, configChangeEventGenerator);
super(
configServiceBlockingStub,
resourceNamespace,
resourceName,
configChangeEventGenerator,
clientConfig);
this.requestContext = requestContext;
}

Expand Down
70 changes: 48 additions & 22 deletions object-store/src/main/java/org/hypertrace/config/objectstore/DefaultObjectStore.java
View file
Original file line number Diff line number Diff line change
@@ -1,8 +1,10 @@
package org.hypertrace.config.objectstore;

import com.google.protobuf.Value;
import io.grpc.Deadline;
import io.grpc.Status;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.hypertrace.config.service.change.event.api.ConfigChangeEventGenerator;
import org.hypertrace.config.service.v1.ConfigServiceGrpc.ConfigServiceBlockingStub;
Expand All @@ -26,26 +28,39 @@ public abstract class DefaultObjectStore<T> {
private final String resourceNamespace;
private final String resourceName;
private final Optional<ConfigChangeEventGenerator> configChangeEventGeneratorOptional;
private final ClientConfig clientConfig;

protected DefaultObjectStore(
ConfigServiceBlockingStub configServiceBlockingStub,
String resourceNamespace,
String resourceName,
ConfigChangeEventGenerator configChangeEventGenerator) {
ConfigChangeEventGenerator configChangeEventGenerator,
ClientConfig clientConfig) {
this.configServiceBlockingStub = configServiceBlockingStub;
this.resourceNamespace = resourceNamespace;
this.resourceName = resourceName;
this.configChangeEventGeneratorOptional = Optional.of(configChangeEventGenerator);
this.configChangeEventGeneratorOptional = Optional.ofNullable(configChangeEventGenerator);
this.clientConfig = clientConfig;
}

protected DefaultObjectStore(
ConfigServiceBlockingStub configServiceBlockingStub,
String resourceNamespace,
String resourceName,
ConfigChangeEventGenerator configChangeEventGenerator) {
this(
configServiceBlockingStub,
resourceNamespace,
resourceName,
configChangeEventGenerator,
ClientConfig.DEFAULT);
}

protected DefaultObjectStore(
ConfigServiceBlockingStub configServiceBlockingStub,
String resourceNamespace,
String resourceName) {
this.configServiceBlockingStub = configServiceBlockingStub;
this.resourceNamespace = resourceNamespace;
this.resourceName = resourceName;
this.configChangeEventGeneratorOptional = Optional.empty();
this(configServiceBlockingStub, resourceNamespace, resourceName, null);
}

protected abstract Optional<T> buildDataFromValue(Value value);
Expand All @@ -65,11 +80,13 @@ public Optional<ConfigObject<T>> getObject(RequestContext context) {
GetConfigResponse getConfigResponse =
context.call(
() ->
this.configServiceBlockingStub.getConfig(
GetConfigRequest.newBuilder()
.setResourceName(this.resourceName)
.setResourceNamespace(this.resourceNamespace)
.build()));
this.configServiceBlockingStub
.withDeadline(getDeadline())
.getConfig(
GetConfigRequest.newBuilder()
.setResourceName(this.resourceName)
.setResourceNamespace(this.resourceNamespace)
.build()));

return ConfigObjectImpl.tryBuild(
getConfigResponse.getConfig(),
Expand All @@ -90,6 +107,7 @@ public Optional<T> getData(RequestContext context) {
context.call(
() ->
this.configServiceBlockingStub
.withDeadline(getDeadline())
.getConfig(
GetConfigRequest.newBuilder()
.setResourceName(this.resourceName)
Expand All @@ -110,12 +128,14 @@ public ConfigObject<T> upsertObject(RequestContext context, T data) {
UpsertConfigResponse response =
context.call(
() ->
this.configServiceBlockingStub.upsertConfig(
UpsertConfigRequest.newBuilder()
.setResourceName(this.resourceName)
.setResourceNamespace(this.resourceNamespace)
.setConfig(this.buildValueFromData(data))
.build()));
this.configServiceBlockingStub
.withDeadline(getDeadline())
.upsertConfig(
UpsertConfigRequest.newBuilder()
.setResourceName(this.resourceName)
.setResourceNamespace(this.resourceNamespace)
.setConfig(this.buildValueFromData(data))
.build()));

ConfigObject<T> upsertedObject =
ConfigObjectImpl.tryBuild(response, this::buildDataFromValue)
Expand Down Expand Up @@ -153,11 +173,13 @@ public Optional<ConfigObject<T>> deleteObject(RequestContext context) {
context
.call(
() ->
this.configServiceBlockingStub.deleteConfig(
DeleteConfigRequest.newBuilder()
.setResourceName(this.resourceName)
.setResourceNamespace(this.resourceNamespace)
.build()))
this.configServiceBlockingStub
.withDeadline(getDeadline())
.deleteConfig(
DeleteConfigRequest.newBuilder()
.setResourceName(this.resourceName)
.setResourceNamespace(this.resourceNamespace)
.build()))
.getDeletedConfig();
ConfigObject<T> object =
ConfigObjectImpl.tryBuild(deletedConfig, this::buildDataFromValue)
Expand All @@ -176,4 +198,8 @@ public Optional<ConfigObject<T>> deleteObject(RequestContext context) {
throw exception;
}
}

protected Deadline getDeadline() {
return Deadline.after(this.clientConfig.getTimeout().toMillis(), TimeUnit.MILLISECONDS);
}
}
Loading

0 comments on commit 42d3589

Please sign in to comment.