-
Notifications
You must be signed in to change notification settings - Fork 15
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #1 from ContainerTag/HB-2599-Replacing-redis-with-…
…aerospike Hb 2599 replacing redis with aerospike
- Loading branch information
Showing
17 changed files
with
642 additions
and
240 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
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
37 changes: 37 additions & 0 deletions
37
src/main/java/org/prebid/cache/listners/AerospikeReadListener.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,37 @@ | ||
| package org.prebid.cache.listners; | ||
|
|
||
| import com.aerospike.client.AerospikeException; | ||
| import com.aerospike.client.Key; | ||
| import com.aerospike.client.Record; | ||
| import com.aerospike.client.listener.RecordListener; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import reactor.core.publisher.MonoSink; | ||
|
|
||
| import java.util.Objects; | ||
|
|
||
| @Slf4j | ||
| public class AerospikeReadListener implements RecordListener { | ||
| private final static String NAME = "cache"; | ||
| private final MonoSink<String> sink; | ||
| private final String recordKeyId; | ||
|
|
||
| public AerospikeReadListener(MonoSink<String> sink, String recordKeyId) { | ||
| this.sink = sink; | ||
| this.recordKeyId = recordKeyId; | ||
| } | ||
|
|
||
| @Override | ||
| public void onSuccess(Key key, Record record) { | ||
| if (Objects.nonNull(record)) { | ||
| sink.success(record.getString(NAME)); | ||
| } else { | ||
| sink.success(); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(AerospikeException exception) { | ||
| log.error("Error when reading record with keyId {}", recordKeyId); | ||
| sink.error(exception); | ||
| } | ||
| } |
29 changes: 29 additions & 0 deletions
29
src/main/java/org/prebid/cache/listners/AerospikeWriteListener.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,29 @@ | ||
| package org.prebid.cache.listners; | ||
|
|
||
| import com.aerospike.client.AerospikeException; | ||
| import com.aerospike.client.Key; | ||
| import com.aerospike.client.listener.WriteListener; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import reactor.core.publisher.MonoSink; | ||
|
|
||
| @Slf4j | ||
| public class AerospikeWriteListener implements WriteListener { | ||
| private final MonoSink<String> sink; | ||
| private final String keyId; | ||
|
|
||
| public AerospikeWriteListener(MonoSink<String> sink, String keyId) { | ||
| this.sink = sink; | ||
| this.keyId = keyId; | ||
| } | ||
|
|
||
| @Override | ||
| public void onSuccess(Key key) { | ||
| sink.success(key.userKey.toString()); | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(AerospikeException exception) { | ||
| log.error("Error when writing record with keyID : {}", keyId); | ||
| sink.error(exception); | ||
| } | ||
| } |
100 changes: 100 additions & 0 deletions
100
src/main/java/org/prebid/cache/repository/aerospike/AerospikePropertyConfiguration.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,100 @@ | ||
| package org.prebid.cache.repository.aerospike; | ||
|
|
||
| import com.aerospike.client.AerospikeClient; | ||
| import com.aerospike.client.Host; | ||
| import com.aerospike.client.async.AsyncClientPolicy; | ||
| import com.aerospike.client.async.EventLoops; | ||
| import com.aerospike.client.async.EventPolicy; | ||
| import com.aerospike.client.async.NettyEventLoops; | ||
| import com.aerospike.client.policy.Policy; | ||
| import io.netty.channel.EventLoopGroup; | ||
| import io.netty.channel.nio.NioEventLoopGroup; | ||
| import lombok.AllArgsConstructor; | ||
| import lombok.Data; | ||
| import lombok.NoArgsConstructor; | ||
| import org.springframework.boot.context.properties.ConfigurationProperties; | ||
| import org.springframework.boot.context.properties.EnableConfigurationProperties; | ||
| import org.springframework.context.annotation.Bean; | ||
| import org.springframework.context.annotation.Configuration; | ||
|
|
||
| import javax.validation.constraints.NotNull; | ||
| import java.util.Arrays; | ||
|
|
||
| import static java.util.Objects.requireNonNull; | ||
|
|
||
| @Data | ||
| @NoArgsConstructor | ||
| @AllArgsConstructor | ||
| @Configuration | ||
| @EnableConfigurationProperties | ||
| @ConfigurationProperties(prefix = "spring.aerospike") | ||
| public class AerospikePropertyConfiguration { | ||
| private String host; | ||
| private Integer port; | ||
| private String password; | ||
| private Integer cores; | ||
| private Long firstBackoff; | ||
| private Long maxBackoff; | ||
| private int maxRetry; | ||
| private String namespace; | ||
|
|
||
| private static final int DEFAULT_PORT = 3000; | ||
|
|
||
| public static Host[] extractHosts(@NotNull String hostList) { | ||
| return Arrays.stream(hostList.split(",")) | ||
| .map(host -> { | ||
| String[] params = host.split(":"); | ||
| String hostname = requireNonNull(params[0]); | ||
| int port = DEFAULT_PORT; | ||
| if (params.length == 2) { | ||
| port = Integer.parseInt(params[1]); | ||
| } | ||
| return new Host(hostname, port); | ||
| }) | ||
| .toArray(Host[]::new); | ||
| } | ||
|
|
||
| public static boolean isAerospikeCluster(@NotNull String hostList) { | ||
| if (hostList.split(",").length > 1) { | ||
| return true; | ||
| } | ||
| return false; | ||
| } | ||
|
|
||
| @Bean | ||
| Policy readPolicy() { | ||
| return new Policy(); | ||
| } | ||
|
|
||
| @Bean | ||
| EventPolicy eventPolicy() { | ||
| return new EventPolicy(); | ||
| } | ||
|
|
||
| @Bean | ||
| EventLoopGroup eventGroup() { | ||
| return new NioEventLoopGroup(cores); | ||
| } | ||
|
|
||
| @Bean | ||
| EventLoops eventLoops() { | ||
| return new NettyEventLoops(eventPolicy(), eventGroup()); | ||
| } | ||
|
|
||
| @Bean | ||
| AsyncClientPolicy clientPolicy() { | ||
| AsyncClientPolicy clientPolicy = new AsyncClientPolicy(); | ||
| clientPolicy.eventLoops = eventLoops(); | ||
| return clientPolicy; | ||
| } | ||
|
|
||
| @Bean(destroyMethod = "close") | ||
| AerospikeClient client() { | ||
| if (isAerospikeCluster(getHost())) { | ||
| return new AerospikeClient(clientPolicy(), extractHosts(getHost())); | ||
| } else { | ||
| return new AerospikeClient(clientPolicy(), getHost(), getPort()); | ||
| } | ||
| } | ||
|
|
||
| } |
95 changes: 95 additions & 0 deletions
95
src/main/java/org/prebid/cache/repository/aerospike/AerospikeRepositoryImpl.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,95 @@ | ||
| package org.prebid.cache.repository.aerospike; | ||
|
|
||
| import com.aerospike.client.AerospikeClient; | ||
| import com.aerospike.client.AerospikeException; | ||
| import com.aerospike.client.Bin; | ||
| import com.aerospike.client.Key; | ||
| import com.aerospike.client.ResultCode; | ||
| import com.aerospike.client.async.EventLoops; | ||
| import com.aerospike.client.policy.Policy; | ||
| import com.aerospike.client.policy.WritePolicy; | ||
| import lombok.RequiredArgsConstructor; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.prebid.cache.exceptions.PayloadWrapperPropertyException; | ||
| import org.prebid.cache.helpers.Json; | ||
| import org.prebid.cache.listners.AerospikeReadListener; | ||
| import org.prebid.cache.listners.AerospikeWriteListener; | ||
| import org.prebid.cache.model.PayloadWrapper; | ||
| import org.prebid.cache.repository.ReactiveRepository; | ||
| import reactor.core.publisher.Mono; | ||
| import reactor.retry.Retry; | ||
|
|
||
| import javax.validation.constraints.NotNull; | ||
| import java.time.Duration; | ||
| import java.util.Arrays; | ||
| import java.util.List; | ||
| import java.util.Objects; | ||
|
|
||
| @Slf4j | ||
| @RequiredArgsConstructor | ||
| public class AerospikeRepositoryImpl implements ReactiveRepository<PayloadWrapper, String> { | ||
| @NotNull | ||
| private final AerospikePropertyConfiguration configuration; | ||
| @NotNull | ||
| private final AerospikeClient client; | ||
| @NotNull | ||
| private final EventLoops eventLoops; | ||
| @NotNull | ||
| private final Policy policy; | ||
| private WritePolicy writePolicy; | ||
|
|
||
| private final static String BIN_NAME="cache"; | ||
|
|
||
| @Override | ||
| public Mono save(final PayloadWrapper wrapper) { | ||
| long expiry; | ||
| String normalizedId; | ||
| WritePolicy policy = writePolicy(); | ||
|
|
||
| try { | ||
| expiry = wrapper.getExpiry(); | ||
| normalizedId = wrapper.getNormalizedId(); | ||
| policy.expiration = (int) expiry; | ||
| } catch (PayloadWrapperPropertyException e) { | ||
| e.printStackTrace(); | ||
| return Mono.empty(); | ||
| } | ||
|
|
||
| return Mono.<String>create(sink -> client.put(eventLoops.next(), | ||
| new AerospikeWriteListener(sink, normalizedId), policy, | ||
| new Key(configuration.getNamespace(), "", normalizedId), | ||
| new Bin(BIN_NAME, Json.toJson(wrapper)))).map(payload -> wrapper) | ||
| .retryWhen(getRetryPolicy()); | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<PayloadWrapper> findById(String id) { | ||
| return Mono.<String>create(sink -> client.get(eventLoops.next(), | ||
| new AerospikeReadListener(sink, id), | ||
| policy, new Key(configuration.getNamespace(), "", id))) | ||
| .map(json -> Json.createPayloadFromJson(json, PayloadWrapper.class)) | ||
| .retryWhen(getRetryPolicy()); | ||
| } | ||
|
|
||
| private WritePolicy writePolicy() { | ||
| if (Objects.isNull(writePolicy)) { | ||
| writePolicy = new WritePolicy(); | ||
| } | ||
| return writePolicy; | ||
| } | ||
|
|
||
| private List<Integer> getRetryCodes() { | ||
| return Arrays.asList(ResultCode.GENERATION_ERROR, ResultCode.KEY_EXISTS_ERROR, ResultCode.KEY_NOT_FOUND_ERROR); | ||
| } | ||
|
|
||
| private Retry<Object> getRetryPolicy() { | ||
| Duration firstBackoff = Duration.ofMillis(configuration.getFirstBackoff()); | ||
| Duration maxBackoff = Duration.ofMillis(configuration.getMaxBackoff()); | ||
|
|
||
| return Retry.onlyIf(context -> context.exception() instanceof AerospikeException | ||
| && getRetryCodes().contains(((AerospikeException) context.exception()).getResultCode()) | ||
| ).doOnRetry(context -> log.warn("Retrying context {}", context)) | ||
| .retryMax(configuration.getMaxRetry()) | ||
| .exponentialBackoffWithJitter(firstBackoff, maxBackoff); | ||
| } | ||
| } |
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,5 +1,5 @@ | ||
|
|
||
| --- | ||
| cache.profiles.active: redis | ||
| cache.redis.classname.canonical: org.prebid.cache.repository.redis.RedisRepositoryImpl | ||
| cache.redis.property.configuration.classname: RedisPropertyConfiguration | ||
| cache.profiles.active: aerospike | ||
| cache.aerospike.classname.canonical: org.prebid.cache.repository.aerospike.AerospikeRepositoryImpl | ||
| cache.aerospike.property.configuration.classname: AerospikePropertyConfiguration |
Oops, something went wrong.