Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

memcached cache: switch to AWS elasticache-java-cluster-client and add TLS support #14827

Merged
merged 21 commits into from
Oct 2, 2023
Merged

memcached cache: switch to AWS elasticache-java-cluster-client and add TLS support #14827

Show file tree
Hide file tree
Changes from 11 commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
5c4cae2
Elasticache client
pagrawal10 Aug 15, 2023
1bbabc4
Update license for aws elasticache client
pagrawal10 Aug 16, 2023
52dba0a
Add sslconnection and autodiscovery configs
pagrawal10 Aug 21, 2023
9deeb1d
PR comments
pagrawal10 Aug 22, 2023
d72e3c1
Modified description
pagrawal10 Aug 23, 2023
2a1e535
Update documentation as per PR review
pagrawal10 Aug 25, 2023
10916ba
Updated docs
pagrawal10 Aug 28, 2023
4889f7e
Fix lint error
pagrawal10 Aug 29, 2023
1bd5efe
Add skipTlsHostnameVerification option
pagrawal10 Sep 4, 2023
dd10972
codestyle fixes
pagrawal10 Sep 5, 2023
e419b24
throw error on invalid client mode
pagrawal10 Sep 5, 2023
16ada09
Refactor connectionFactoryBuilder
pagrawal10 Sep 7, 2023
4f1507e
fix checkstyle issues
pagrawal10 Sep 12, 2023
d37b266
Fix cfbuilder.toString()
pagrawal10 Sep 12, 2023
c70573d
check sslcontext is passed to connectionFactory
pagrawal10 Sep 12, 2023
11490d4
refactor unit tests
pagrawal10 Sep 12, 2023
cc1cdb5
revert custom.css changes
pagrawal10 Sep 12, 2023
70fee95
Merge branch 'master' into pagrawal/memcache
pagrawal10 Sep 12, 2023
3e88827
checkstyle fixes
pagrawal10 Sep 20, 2023
f8dbaff
minor nits in toString()
pagrawal10 Sep 21, 2023
1b5a8ce
Update custom.css
pagrawal10 Oct 2, 2023
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
23 changes: 13 additions & 10 deletions docs/configuration/index.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2103,16 +2103,19 @@ In addition to the normal cache metrics, the caffeine cache implementation also

Uses memcached as cache backend. This allows all processes to share the same cache.

|Property|Description|Default|
|--------|-----------|-------|
|`druid.cache.expiration`|Memcached [expiration time](https://code.google.com/p/memcached/wiki/NewCommands#Standard_Protocol).|2592000 (30 days)|
|`druid.cache.timeout`|Maximum time in milliseconds to wait for a response from Memcached.|500|
|`druid.cache.hosts`|Comma separated list of Memcached hosts `<host:port>`.|none|
|`druid.cache.maxObjectSize`|Maximum object size in bytes for a Memcached object.|52428800 (50 MiB)|
|`druid.cache.memcachedPrefix`|Key prefix for all keys in Memcached.|druid|
|`druid.cache.numConnections`|Number of memcached connections to use.|1|
|`druid.cache.protocol`|Memcached communication protocol. Can be binary or text.|binary|
|`druid.cache.locator`|Memcached locator. Can be consistent or array_mod.|consistent|
| Property | Description | Default |
|-------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------|
| `druid.cache.expiration` | Memcached [expiration time](https://code.google.com/p/memcached/wiki/NewCommands#Standard_Protocol). | 2592000 (30 days) |
| `druid.cache.timeout` | Maximum time in milliseconds to wait for a response from Memcached. | 500 |
| `druid.cache.hosts` | Comma separated list of Memcached hosts `<host:port>`. Need to specify all nodes when `druid.cache.clientMode` is set to static. Dynamic mode [automatically identifies nodes in your cluster](https://docs.aws.amazon.com/AmazonElastiCache/latest/mem-ug/AutoDiscovery.html) so just specifying the configuration endpoint and port is fine. | none |
| `druid.cache.maxObjectSize` | Maximum object size in bytes for a Memcached object. | 52428800 (50 MiB) |
| `druid.cache.memcachedPrefix` | Key prefix for all keys in Memcached. | druid |
| `druid.cache.numConnections` | Number of memcached connections to use. | 1 |
| `druid.cache.protocol` | Memcached communication protocol. Can be binary or text. | binary |
| `druid.cache.locator` | Memcached locator. Can be consistent or array_mod. | consistent |
| `druid.cache.enableTls` | Enable TLS based connection for Memcached client. Boolean | false |
| `druid.cache.clientMode` | Client Mode. Static mode requires the user to specify individual cluster nodes. Dynamic mode uses [AutoDiscovery](https://docs.aws.amazon.com/AmazonElastiCache/latest/mem-ug/AutoDiscovery.HowAutoDiscoveryWorks.html) feature of AWS Memcached. String. ["static"](https://docs.aws.amazon.com/AmazonElastiCache/latest/mem-ug/AutoDiscovery.Manual.html) or ["dynamic"](https://docs.aws.amazon.com/AmazonElastiCache/latest/mem-ug/AutoDiscovery.Using.ModifyApp.Java.html) | static |
| `druid.cache.skipTlsHostnameVerification` | Skip TLS Hostname Verification. Boolean. | true |

#### Hybrid

Expand Down
6 changes: 3 additions & 3 deletions licenses.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1638,13 +1638,13 @@ libraries:

---

name: Spymemcached
name: aws-elasticache-cluster-client-memcached-for-java
license_category: binary
module: java-core
license_name: Apache License version 2.0
version: 2.12.3
version: 1.2.0
libraries:
- net.spy: spymemcached
- com.amazonaws: elasticache-java-cluster-client

---

Expand Down
6 changes: 3 additions & 3 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -766,9 +766,9 @@
<version>3.3.6</version>
</dependency>
<dependency>
<groupId>net.spy</groupId>
<artifactId>spymemcached</artifactId>
<version>2.12.3</version>
<groupId>com.amazonaws</groupId>
<artifactId>elasticache-java-cluster-client</artifactId>
<version>1.2.0</version>
xvrl marked this conversation as resolved.
Show resolved Hide resolved
</dependency>
<dependency>
<groupId>org.antlr</groupId>
Expand Down
4 changes: 2 additions & 2 deletions server/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,8 +134,8 @@
<artifactId>tesla-aether</artifactId>
</dependency>
<dependency>
<groupId>net.spy</groupId>
<artifactId>spymemcached</artifactId>
<groupId>com.amazonaws</groupId>
<artifactId>elasticache-java-cluster-client</artifactId>
</dependency>
<dependency>
<groupId>org.lz4</groupId>
Expand Down
48 changes: 38 additions & 10 deletions server/src/main/java/org/apache/druid/client/cache/MemcachedCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,12 @@
import com.google.common.collect.Maps;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hashing;
import java.security.KeyManagementException;
import java.security.KeyStore;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManagerFactory;
import net.spy.memcached.AddrUtil;
import net.spy.memcached.ClientMode;
import net.spy.memcached.ConnectionFactory;
import net.spy.memcached.ConnectionFactoryBuilder;
import net.spy.memcached.FailureMode;
Expand All @@ -56,11 +61,9 @@
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
Expand Down Expand Up @@ -339,7 +342,7 @@ public void updateHistogram(String name, int amount)
}
};

final ConnectionFactory connectionFactory = new MemcachedCustomConnectionFactoryBuilder()
ConnectionFactoryBuilder connectionFactoryBuilder = new MemcachedCustomConnectionFactoryBuilder()
// 1000 repetitions gives us good distribution with murmur3_128
// (approx < 5% difference in counts across nodes, with 5 cache nodes)
.setKetamaNodeRepetitions(1000)
Expand All @@ -355,9 +358,28 @@ public void updateHistogram(String name, int amount)
.setReadBufferSize(config.getReadBufferSize())
.setOpQueueFactory(opQueueFactory)
.setMetricCollector(metricCollector)
.setEnableMetrics(MetricType.DEBUG) // Not as scary as it sounds
.build();

.setEnableMetrics(MetricType.DEBUG); // Not as scary as it sounds
if (config.enableTls()) {
// Build SSLContext
TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
tmf.init((KeyStore) null);
SSLContext sslContext = SSLContext.getInstance("TLS");
sslContext.init(null, tmf.getTrustManagers(), null);
// Create the client in TLS mode
connectionFactoryBuilder.setSSLContext(sslContext);
}
if ("dynamic".equals(config.getClientMode())) {
connectionFactoryBuilder.setClientMode(ClientMode.Dynamic);
connectionFactoryBuilder.setHostnameForTlsVerification(config.getHosts().split(",")[0]);
} else if ("static".equals(config.getClientMode())) {
connectionFactoryBuilder.setClientMode(ClientMode.Static);
xvrl marked this conversation as resolved.
Show resolved Hide resolved
} else {
throw new RuntimeException("Invalid value provided for `druid.cache.clientMode`. Value must be 'static' or 'dynamic'.");
pagrawal10 marked this conversation as resolved.
Show resolved Hide resolved
}
if (config.skipTlsHostnameVerification()) {
connectionFactoryBuilder.setSkipTlsHostnameVerification(true);
}
final ConnectionFactory connectionFactory = connectionFactoryBuilder.build();
final List<InetSocketAddress> hosts = AddrUtil.getAddresses(config.getHosts());


Expand Down Expand Up @@ -389,7 +411,13 @@ public MemcachedClientIF get()

return new MemcachedCache(clientSupplier, config, monitor);
}
catch (IOException e) {
catch (IOException | NoSuchAlgorithmException e) {
throw new RuntimeException(e);
}
catch (KeyStoreException e) {
throw new RuntimeException(e);
}
catch (KeyManagementException e) {
throw new RuntimeException(e);
}
}
Expand Down
24 changes: 24 additions & 0 deletions server/src/main/java/org/apache/druid/client/cache/MemcachedCacheConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,15 @@ public class MemcachedCacheConfig
@JsonProperty
private String locator = "consistent";

@JsonProperty
private boolean enableTls = false;

@JsonProperty
private String clientMode = "static";

@JsonProperty
private boolean skipTlsHostnameVerification = true;

public int getExpiration()
{
return expiration;
Expand Down Expand Up @@ -112,4 +121,19 @@ public String getLocator()
{
return locator;
}

public boolean enableTls()
{
return enableTls;
}

public String getClientMode()
{
return clientMode;
}

public boolean skipTlsHostnameVerification()
{
return skipTlsHostnameVerification;
}
}
46 changes: 38 additions & 8 deletions server/src/test/java/org/apache/druid/client/cache/MemcachedCacheTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,14 +29,7 @@
import com.google.inject.Injector;
import com.google.inject.Module;
import com.google.inject.name.Names;
import net.spy.memcached.BroadcastOpFactory;
import net.spy.memcached.CASResponse;
import net.spy.memcached.CASValue;
import net.spy.memcached.CachedData;
import net.spy.memcached.ConnectionObserver;
import net.spy.memcached.MemcachedClientIF;
import net.spy.memcached.MemcachedNode;
import net.spy.memcached.NodeLocator;
import net.spy.memcached.*;
import net.spy.memcached.internal.BulkFuture;
import net.spy.memcached.internal.BulkGetCompletionListener;
import net.spy.memcached.internal.OperationFuture;
Expand Down Expand Up @@ -224,6 +217,37 @@ public void emit(Event event)
}
}

@Test
public void testSslConnection()
{
final MemcachedCacheConfig config = new MemcachedCacheConfig()
{
@Override
public boolean enableTls()
{
return true;
}

@Override
public String getHosts()
{
return "localhost:9999";
}
};
// Static Mode
Assert.assertEquals(config.getClientMode(), "static");
xvrl marked this conversation as resolved.
Show resolved Hide resolved
MemcachedCache client = new MemcachedCache(
Suppliers.ofInstance(
StupidResourceHolder.create(new MockMemcachedClient())
),
config,
NOOP_MONITOR
);
Assert.assertNull(client.get(new Cache.NamedKey("a", HI)));
put(client, "a", HI, 1);
Assert.assertEquals(1, get(client, "a", HI));
xvrl marked this conversation as resolved.
Show resolved Hide resolved
}

@Test
public void testSanity()
{
Expand Down Expand Up @@ -316,6 +340,12 @@ public Collection<SocketAddress> getAvailableServers()
throw new UnsupportedOperationException("not implemented");
}

@Override
public boolean refreshCertificate()
{
return true;
}

@Override
public Collection<SocketAddress> getUnavailableServers()
{
Expand Down
Loading