-
Notifications
You must be signed in to change notification settings - Fork 10
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add non rx based caching attribute client (#200)
* Update gradle locks * trivy * Add non rx based caching attribute client * remove * update * update * fix --------- Co-authored-by: kishansairam9 <[email protected]>
- Loading branch information
1 parent
f13783d
commit 32a2977
Showing
14 changed files
with
437 additions
and
13 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,2 @@ | ||
| # libssl3 | ||
| CVE-2023-5678 exp:2023-12-31 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
128 changes: 128 additions & 0 deletions
128
.../main/java/org/hypertrace/core/attribute/service/client/AttributeServiceCachedClient.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,128 @@ | ||
| package org.hypertrace.core.attribute.service.client; | ||
|
|
||
| import com.google.common.cache.Cache; | ||
| import com.google.common.cache.CacheBuilder; | ||
| import com.google.common.cache.CacheLoader; | ||
| import com.google.common.cache.LoadingCache; | ||
| import com.google.common.collect.ImmutableTable; | ||
| import com.google.common.collect.Table; | ||
| import com.google.common.util.concurrent.ThreadFactoryBuilder; | ||
| import io.grpc.Channel; | ||
| import java.util.Collections; | ||
| import java.util.List; | ||
| import java.util.Optional; | ||
| import java.util.concurrent.Executors; | ||
| import java.util.concurrent.ThreadFactory; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.function.Function; | ||
| import javax.annotation.Nonnull; | ||
| import lombok.Value; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.hypertrace.core.attribute.service.client.config.AttributeServiceCachedClientConfig; | ||
| import org.hypertrace.core.attribute.service.v1.AttributeMetadata; | ||
| import org.hypertrace.core.attribute.service.v1.AttributeServiceGrpc; | ||
| import org.hypertrace.core.attribute.service.v1.AttributeServiceGrpc.AttributeServiceBlockingStub; | ||
| import org.hypertrace.core.attribute.service.v1.GetAttributesRequest; | ||
| import org.hypertrace.core.grpcutils.client.ClientCallCredentialsProvider; | ||
| import org.hypertrace.core.grpcutils.client.RequestContextClientCallCredsProviderFactory; | ||
| import org.hypertrace.core.grpcutils.context.ContextualKey; | ||
| import org.hypertrace.core.grpcutils.context.RequestContext; | ||
| import org.hypertrace.core.serviceframework.metrics.PlatformMetricsRegistry; | ||
|
|
||
| @Slf4j | ||
| public class AttributeServiceCachedClient { | ||
| private final LoadingCache<ContextualKey<Void>, Table<String, String, AttributeMetadata>> cache; | ||
| private final Cache<String, AttributeScopeAndKey> scopeAndKeyLookup; | ||
| private final AttributeServiceBlockingStub attributeServiceBlockingStub; | ||
| private final long deadlineMs; | ||
| private final ClientCallCredentialsProvider callCredentialsProvider; | ||
|
|
||
| AttributeServiceCachedClient(Channel channel, AttributeServiceCachedClientConfig clientConfig) { | ||
| this( | ||
| channel, | ||
| clientConfig, | ||
| RequestContextClientCallCredsProviderFactory.getClientCallCredsProvider()); | ||
| } | ||
|
|
||
| AttributeServiceCachedClient( | ||
| Channel channel, | ||
| AttributeServiceCachedClientConfig clientConfig, | ||
| ClientCallCredentialsProvider callCredentialsProvider) { | ||
| this.callCredentialsProvider = callCredentialsProvider; | ||
| this.attributeServiceBlockingStub = AttributeServiceGrpc.newBlockingStub(channel); | ||
| deadlineMs = clientConfig.getDeadline().toMillis(); | ||
| cache = | ||
| CacheBuilder.newBuilder() | ||
| .maximumSize(clientConfig.getMaxSize()) | ||
| .refreshAfterWrite(clientConfig.getRefreshAfterWrite()) | ||
| .expireAfterAccess(clientConfig.getExpireAfterAccess()) | ||
| .build( | ||
| CacheLoader.asyncReloading( | ||
| CacheLoader.from(this::loadTable), | ||
| Executors.newFixedThreadPool( | ||
| clientConfig.getExecutorThreads(), this.buildThreadFactory()))); | ||
| PlatformMetricsRegistry.registerCache( | ||
| clientConfig.getCacheMetricsName(), cache, Collections.emptyMap()); | ||
| scopeAndKeyLookup = | ||
| CacheBuilder.newBuilder().expireAfterWrite(clientConfig.getExpireAfterAccess()).build(); | ||
| } | ||
|
|
||
| public Optional<AttributeMetadata> get( | ||
| @Nonnull RequestContext requestContext, | ||
| @Nonnull String attributeScope, | ||
| @Nonnull String attributeKey) { | ||
| return Optional.ofNullable(getTable(requestContext).get(attributeScope, attributeKey)); | ||
| } | ||
|
|
||
| public Optional<AttributeMetadata> getById( | ||
| @Nonnull RequestContext requestContext, @Nonnull String attributeId) { | ||
| Table<String, String, AttributeMetadata> table = getTable(requestContext); | ||
| return Optional.ofNullable(scopeAndKeyLookup.getIfPresent(attributeId)) | ||
| .map(scopeAndKey -> table.get(scopeAndKey.getScope(), scopeAndKey.getKey())); | ||
| } | ||
|
|
||
| public List<AttributeMetadata> getAllInScope( | ||
| @Nonnull RequestContext requestContext, @Nonnull String attributeScope) { | ||
| return List.copyOf(getTable(requestContext).row(attributeScope).values()); | ||
| } | ||
|
|
||
| private Table<String, String, AttributeMetadata> getTable(RequestContext requestContext) { | ||
| return cache.getUnchecked(requestContext.buildInternalContextualKey()); | ||
| } | ||
|
|
||
| private Table<String, String, AttributeMetadata> loadTable(ContextualKey<Void> contextualKey) { | ||
| List<AttributeMetadata> attributeMetadataList = | ||
| contextualKey | ||
| .getContext() | ||
| .call( | ||
| () -> | ||
| attributeServiceBlockingStub | ||
| .withDeadlineAfter(deadlineMs, TimeUnit.MILLISECONDS) | ||
| .withCallCredentials(callCredentialsProvider.get()) | ||
| .getAttributes(GetAttributesRequest.getDefaultInstance())) | ||
| .getAttributesList(); | ||
| attributeMetadataList.forEach( | ||
| attributeMetadata -> | ||
| scopeAndKeyLookup.put( | ||
| attributeMetadata.getId(), | ||
| new AttributeScopeAndKey( | ||
| attributeMetadata.getScopeString(), attributeMetadata.getKey()))); | ||
| return attributeMetadataList.stream() | ||
| .collect( | ||
| ImmutableTable.toImmutableTable( | ||
| AttributeMetadata::getScopeString, AttributeMetadata::getKey, Function.identity())); | ||
| } | ||
|
|
||
| private ThreadFactory buildThreadFactory() { | ||
| return new ThreadFactoryBuilder() | ||
| .setDaemon(true) | ||
| .setNameFormat("attribute-service-cached-client-%d") | ||
| .build(); | ||
| } | ||
|
|
||
| @Value | ||
| private static class AttributeScopeAndKey { | ||
| String scope; | ||
| String key; | ||
| } | ||
| } |
52 changes: 52 additions & 0 deletions
52
...g/hypertrace/core/attribute/service/client/config/AttributeServiceCachedClientConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,52 @@ | ||
| package org.hypertrace.core.attribute.service.client.config; | ||
|
|
||
| import com.typesafe.config.Config; | ||
| import java.time.Duration; | ||
| import lombok.Value; | ||
| import org.hypertrace.core.attribute.service.client.AttributeServiceCachedClient; | ||
|
|
||
| @Value | ||
| public class AttributeServiceCachedClientConfig { | ||
| private static final String DEADLINE_CONFIG_KEY = "deadline"; | ||
| private static final String CACHE_MAX_SIZE_CONFIG_KEY = "maxSize"; | ||
| private static final String CACHE_REFRESH_AFTER_WRITE_CONFIG_KEY = "refreshAfterWriteDuration"; | ||
| private static final String CACHE_EXPIRE_AFTER_ACCESS_CONFIG_KEY = "expireAfterAccessDuration"; | ||
| private static final String CACHE_EXECUTOR_THREADS_CONFIG_KEY = "executorThreads"; | ||
|
|
||
| Duration deadline; | ||
| long maxSize; | ||
| Duration refreshAfterWrite; | ||
| Duration expireAfterAccess; | ||
| int executorThreads; | ||
| String cacheMetricsName; | ||
|
|
||
| public static AttributeServiceCachedClientConfig from(Config attributeServiceConfig) { | ||
| return from(attributeServiceConfig, AttributeServiceCachedClient.class.getName()); | ||
| } | ||
|
|
||
| public static AttributeServiceCachedClientConfig from( | ||
| Config attributeServiceConfig, String cacheMetricsName) { | ||
| Duration deadline = | ||
| attributeServiceConfig.hasPath(DEADLINE_CONFIG_KEY) | ||
| ? attributeServiceConfig.getDuration(DEADLINE_CONFIG_KEY) | ||
| : Duration.ofSeconds(30); | ||
| long maxSize = | ||
| attributeServiceConfig.hasPath(CACHE_MAX_SIZE_CONFIG_KEY) | ||
| ? attributeServiceConfig.getLong(CACHE_MAX_SIZE_CONFIG_KEY) | ||
| : 1000; | ||
| Duration refreshAfterWrite = | ||
| attributeServiceConfig.hasPath(CACHE_REFRESH_AFTER_WRITE_CONFIG_KEY) | ||
| ? attributeServiceConfig.getDuration(CACHE_REFRESH_AFTER_WRITE_CONFIG_KEY) | ||
| : Duration.ofMinutes(15); | ||
| Duration expireAfterWrite = | ||
| attributeServiceConfig.hasPath(CACHE_EXPIRE_AFTER_ACCESS_CONFIG_KEY) | ||
| ? attributeServiceConfig.getDuration(CACHE_EXPIRE_AFTER_ACCESS_CONFIG_KEY) | ||
| : Duration.ofHours(1); | ||
| int executorThreads = | ||
| attributeServiceConfig.hasPath(CACHE_EXECUTOR_THREADS_CONFIG_KEY) | ||
| ? attributeServiceConfig.getInt(CACHE_EXECUTOR_THREADS_CONFIG_KEY) | ||
| : 1; | ||
| return new AttributeServiceCachedClientConfig( | ||
| deadline, maxSize, refreshAfterWrite, expireAfterWrite, executorThreads, cacheMetricsName); | ||
| } | ||
| } |
Oops, something went wrong.