diff --git a/client/src/main/java/com/metamx/druid/client/cache/Cache.java b/client/src/main/java/com/metamx/druid/client/cache/Cache.java index 9bf0cde33e7..6e9463deb56 100644 --- a/client/src/main/java/com/metamx/druid/client/cache/Cache.java +++ b/client/src/main/java/com/metamx/druid/client/cache/Cache.java @@ -19,8 +19,11 @@ package com.metamx.druid.client.cache; +import com.google.common.base.Charsets; import com.google.common.base.Preconditions; +import com.google.common.primitives.Ints; +import java.nio.ByteBuffer; import java.util.Arrays; import java.util.Map; @@ -48,6 +51,14 @@ public interface Cache this.key = key; } + public byte[] toByteArray() { + final byte[] nsBytes = this.namespace.getBytes(Charsets.UTF_8); + return ByteBuffer.allocate(Ints.BYTES + nsBytes.length + this.key.length) + .putInt(nsBytes.length) + .put(nsBytes) + .put(this.key).array(); + } + @Override public boolean equals(Object o) { diff --git a/client/src/main/java/com/metamx/druid/client/cache/MemcachedCache.java b/client/src/main/java/com/metamx/druid/client/cache/MemcachedCache.java index 86cbf5153e5..ed7f5292e8a 100644 --- a/client/src/main/java/com/metamx/druid/client/cache/MemcachedCache.java +++ b/client/src/main/java/com/metamx/druid/client/cache/MemcachedCache.java @@ -20,9 +20,10 @@ package com.metamx.druid.client.cache; import com.google.common.base.Function; +import com.google.common.base.Preconditions; import com.google.common.base.Throwables; import com.google.common.collect.Maps; -import net.iharder.base64.Base64; +import com.google.common.primitives.Ints; import net.spy.memcached.AddrUtil; import net.spy.memcached.ConnectionFactoryBuilder; import net.spy.memcached.DefaultHashAlgorithm; @@ -31,9 +32,12 @@ import net.spy.memcached.MemcachedClient; import net.spy.memcached.MemcachedClientIF; import net.spy.memcached.internal.BulkFuture; import net.spy.memcached.transcoders.SerializingTranscoder; +import org.apache.commons.codec.digest.DigestUtils; import javax.annotation.Nullable; import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Map; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; @@ -62,6 +66,7 @@ public class MemcachedCache implements Cache .build(), AddrUtil.getAddresses(config.getHosts()) ), + config.getMemcachedPrefix(), config.getTimeout(), config.getExpiration() ); @@ -72,6 +77,7 @@ public class MemcachedCache implements Cache private final int timeout; private final int expiration; + private final String memcachedPrefix; private final MemcachedClientIF client; @@ -79,10 +85,15 @@ public class MemcachedCache implements Cache private final AtomicLong missCount = new AtomicLong(0); private final AtomicLong timeoutCount = new AtomicLong(0); - MemcachedCache(MemcachedClientIF client, int timeout, int expiration) { + MemcachedCache(MemcachedClientIF client, String memcachedPrefix, int timeout, int expiration) { + Preconditions.checkArgument(memcachedPrefix.length() <= MAX_PREFIX_LENGTH, + "memcachedPrefix length [%d] exceeds maximum length [%d]", + memcachedPrefix.length(), + MAX_PREFIX_LENGTH); this.timeout = timeout; this.expiration = expiration; this.client = client; + this.memcachedPrefix = memcachedPrefix; } @Override @@ -101,7 +112,7 @@ public class MemcachedCache implements Cache @Override public byte[] get(NamedKey key) { - Future future = client.asyncGet(computeKeyString(key)); + Future future = client.asyncGet(computeKeyHash(memcachedPrefix, key)); try { byte[] bytes = (byte[]) future.get(timeout, TimeUnit.MILLISECONDS); if(bytes != null) { @@ -110,7 +121,7 @@ public class MemcachedCache implements Cache else { missCount.incrementAndGet(); } - return bytes; + return bytes == null ? null : deserializeValue(key, bytes); } catch(TimeoutException e) { timeoutCount.incrementAndGet(); @@ -129,7 +140,30 @@ public class MemcachedCache implements Cache @Override public void put(NamedKey key, byte[] value) { - client.set(computeKeyString(key), expiration, value); + client.set(computeKeyHash(memcachedPrefix, key), expiration, serializeValue(key, value)); + } + + private static byte[] serializeValue(NamedKey key, byte[] value) { + byte[] keyBytes = key.toByteArray(); + return ByteBuffer.allocate(Ints.BYTES + keyBytes.length + value.length) + .putInt(keyBytes.length) + .put(keyBytes) + .put(value) + .array(); + } + + private static byte[] deserializeValue(NamedKey key, byte[] bytes) { + ByteBuffer buf = ByteBuffer.wrap(bytes); + + final int keyLength = buf.getInt(); + byte[] keyBytes = new byte[keyLength]; + buf.get(keyBytes); + byte[] value = new byte[buf.remaining()]; + buf.get(value); + + Preconditions.checkState(Arrays.equals(keyBytes, key.toByteArray()), + "Keys do not match, possible hash collision?"); + return value; } @Override @@ -144,7 +178,7 @@ public class MemcachedCache implements Cache @Nullable NamedKey input ) { - return computeKeyString(input); + return computeKeyHash(memcachedPrefix, input); } } ); @@ -163,9 +197,11 @@ public class MemcachedCache implements Cache Map results = Maps.newHashMap(); for(Map.Entry entry : some.entrySet()) { + final NamedKey key = keyLookup.get(entry.getKey()); + final byte[] value = (byte[]) entry.getValue(); results.put( - keyLookup.get(entry.getKey()), - (byte[])entry.getValue() + key, + value == null ? null : deserializeValue(key, value) ); } @@ -186,7 +222,15 @@ public class MemcachedCache implements Cache // no resources to cleanup } - private static String computeKeyString(NamedKey key) { - return key.namespace + ":" + Base64.encodeBytes(key.key, Base64.DONT_BREAK_LINES); + public static final int MAX_PREFIX_LENGTH = + MemcachedClientIF.MAX_KEY_LENGTH + - 40 // length of namespace hash + - 40 // length of key hash + - 2 // length of separators + ; + + private static String computeKeyHash(String memcachedPrefix, NamedKey key) { + // hash keys to keep things under 250 characters for memcached + return memcachedPrefix + ":" + DigestUtils.sha1Hex(key.namespace) + ":" + DigestUtils.sha1Hex(key.key); } } diff --git a/client/src/main/java/com/metamx/druid/client/cache/MemcachedCacheConfig.java b/client/src/main/java/com/metamx/druid/client/cache/MemcachedCacheConfig.java index 83f626d8641..c2a277fb72f 100644 --- a/client/src/main/java/com/metamx/druid/client/cache/MemcachedCacheConfig.java +++ b/client/src/main/java/com/metamx/druid/client/cache/MemcachedCacheConfig.java @@ -18,4 +18,7 @@ public abstract class MemcachedCacheConfig @Config("${prefix}.maxObjectSize") public abstract int getMaxObjectSize(); + + @Config("${prefix}.memcachedPrefix") + public abstract String getMemcachedPrefix(); } diff --git a/client/src/test/java/com/metamx/druid/client/cache/MapCacheBrokerTest.java b/client/src/test/java/com/metamx/druid/client/cache/MapCacheTest.java similarity index 98% rename from client/src/test/java/com/metamx/druid/client/cache/MapCacheBrokerTest.java rename to client/src/test/java/com/metamx/druid/client/cache/MapCacheTest.java index 78f071ca539..23a3bd1d641 100644 --- a/client/src/test/java/com/metamx/druid/client/cache/MapCacheBrokerTest.java +++ b/client/src/test/java/com/metamx/druid/client/cache/MapCacheTest.java @@ -26,7 +26,7 @@ import org.junit.Test; /** */ -public class MapCacheBrokerTest +public class MapCacheTest { private static final byte[] HI = "hi".getBytes(); private static final byte[] HO = "ho".getBytes(); diff --git a/client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheBrokerBenchmark.java b/client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheBenchmark.java similarity index 95% rename from client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheBrokerBenchmark.java rename to client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheBenchmark.java index d87dfd5f7a1..3a746c9484b 100644 --- a/client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheBrokerBenchmark.java +++ b/client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheBenchmark.java @@ -17,7 +17,7 @@ import java.util.Map; import java.util.Random; import java.util.concurrent.TimeUnit; -public class MemcachedCacheBrokerBenchmark extends SimpleBenchmark +public class MemcachedCacheBenchmark extends SimpleBenchmark { private static final String BASE_KEY = "test_2012-11-26T00:00:00.000Z_2012-11-27T00:00:00.000Z_2012-11-27T04:11:25.979Z_"; public static final String NAMESPACE = "default"; @@ -56,6 +56,7 @@ public class MemcachedCacheBrokerBenchmark extends SimpleBenchmark cache = new MemcachedCache( client, + "druid-memcached-benchmark", 30000, // 30 seconds 3600 // 1 hour ); @@ -113,6 +114,6 @@ public class MemcachedCacheBrokerBenchmark extends SimpleBenchmark } public static void main(String[] args) throws Exception { - Runner.main(MemcachedCacheBrokerBenchmark.class, args); + Runner.main(MemcachedCacheBenchmark.class, args); } } diff --git a/client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheBrokerTest.java b/client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheTest.java similarity index 99% rename from client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheBrokerTest.java rename to client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheTest.java index 87c1dcdd9f4..287d208db62 100644 --- a/client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheBrokerTest.java +++ b/client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheTest.java @@ -50,7 +50,7 @@ import java.util.concurrent.TimeoutException; /** */ -public class MemcachedCacheBrokerTest +public class MemcachedCacheTest { private static final byte[] HI = "hi".getBytes(); private static final byte[] HO = "ho".getBytes(); @@ -60,7 +60,7 @@ public class MemcachedCacheBrokerTest public void setUp() throws Exception { MemcachedClientIF client = new MockMemcachedClient(); - cache = new MemcachedCache(client, 500, 3600); + cache = new MemcachedCache(client, "druid-memcached-test", 500, 3600); } @Test