Update version of collectd

This reverts commit fcfc3f7bc0.

analytics/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.0.1-SNAPSHOT</version>
+    <version>1.1.0</version>
   </parent>
   <artifactId>timely-analytics</artifactId>
   <!--

analytics/src/main/java/timely/analytics/flink/SubscriptionSource.java

@@ -96,7 +96,7 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
 
             @Override
             public void onOpen(Session session, EndpointConfig config) {
-                session.addMessageHandler(new MessageHandler.Whole<String>() {
+                session.addMessageHandler(String.class, new MessageHandler.Whole<String>() {
 
                     private final SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMdd-HH");
                     private long lastWatermarkTime = 0L;

balancer/bin/timely-balancer.sh

@@ -8,11 +8,6 @@ else
         TCNATIVE_SUFFIX="so"
 fi
 
-# netty tcnative file reference
-#netty-tcnative-1.1.33.Fork18-linux-x86_64-fedora.jar -> libnetty-tcnative.so
-#netty-tcnative-1.1.33.Fork18-linux-x86_64.jar -> libnetty-tcnative.so
-#netty-tcnative-1.1.33.Fork18-osx-x86_64.jar -> libnetty-tcnative.jnilib
-
 THIS_DIR="${THIS_SCRIPT%/*}"
 NATIVE_DIR="${THIS_DIR}/META-INF/native"
 BASE_DIR=${THIS_DIR}/..
@@ -32,14 +27,16 @@ fi
 mkdir -p ${NATIVE_DIR}
 
 pushd ${BASE_DIR}/bin
-$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${NATIVE_SUFFIX}
-$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${NATIVE_SUFFIX}
+$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${TCNATIVE_SUFFIX}
+$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
 popd
 
-export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*"
+export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
 JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}"
 JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
 JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
+JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}/libnetty-${TCNATIVE_SUFFIX}"
+
 #JVM_ARGS="${JVM_ARGS} -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005"
 
 echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.balancer.Balancer --spring.config.name=timely --spring.profiles.active=balancer"

balancer/conf/log4j2-spring.xml

@@ -22,5 +22,8 @@
       <AppenderRef ref="Console" level="warn" />
       <AppenderRef ref="Rolling" level="trace" />
     </Root>
+    <Logger name="timely" additivity="false">
+      <AppenderRef ref="Console" level="debug" />
+    </Logger>
   </Loggers>
 </Configuration>

balancer/conf/timely-balancer.yml

@@ -13,7 +13,7 @@ timely-balancer:
       certificate-file:
       key-file:
       #key-password:
-      use-generated-keypair: false
+      use-generated-keypair: true
       trust-store-file:
       use-openssl: true
       #use-ciphers:
@@ -31,7 +31,7 @@ timely-balancer:
     nonCachedMetrics:
   server:
     ip: 127.0.0.1
-    tcp-port: 4242
+    tcpPort: 4242
     numTcpPools: 1
     tcpBufferSize: 100
     tcpClientPool:
@@ -51,7 +51,7 @@ timely-balancer:
       jmxEnabled: true
       testOnBorrow: false
       testOnReturn: false
-    udp-port: 4245
+    udpPort: 4245
     udpClientPool:
       minIdlePerKey: 0
       maxIdlePerKey: 40

balancer/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.0.1-SNAPSHOT</version>
+    <version>1.1.0</version>
   </parent>
   <artifactId>timely-balancer</artifactId>
   <dependencies>
@@ -201,11 +201,6 @@
       <artifactId>spring-boot-test-autoconfigure</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>timely</groupId>
-      <artifactId>timely-common</artifactId>
-      <scope>test</scope>
-    </dependency>
   </dependencies>
   <build>
     <plugins>

balancer/src/main/dist/assembly.xml

@@ -33,7 +33,7 @@
       <directory>${project.basedir}/bin</directory>
       <outputDirectory>bin</outputDirectory>
       <includes>
-        <include>*</include>
+        <include>**/*</include>
       </includes>
      </fileSet>
     <fileSet>

balancer/src/main/java/timely/balancer/Balancer.java

@@ -1,5 +1,9 @@
 package timely.balancer;
 
+import static timely.Server.SERVICE_DISCOVERY_PATH;
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS;
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS_LOCK_PATH;
+
 import java.io.File;
 import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
@@ -52,6 +56,12 @@ import io.netty.handler.stream.ChunkedWriteHandler;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.NettyRuntime;
 import io.netty.util.internal.SystemPropertyUtil;
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryForever;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.data.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.builder.SpringApplicationBuilder;
@@ -115,9 +125,13 @@ public class Balancer {
     protected Channel httpChannelHandle = null;
     protected Channel wsChannelHandle = null;
     protected List<Channel> udpChannelHandleList = new ArrayList<>();
+    private CuratorFramework curatorFramework;
     protected volatile boolean shutdown = false;
     private static final int DEFAULT_EVENT_LOOP_THREADS;
 
+    private String[] zkPaths = new String[] { LEADER_LATCH_PATH, ASSIGNMENTS_LAST_UPDATED_PATH, ASSIGNMENTS_LOCK_PATH,
+            SERVICE_DISCOVERY_PATH, NON_CACHED_METRICS, NON_CACHED_METRICS_LOCK_PATH };
+
     static {
         DEFAULT_EVENT_LOOP_THREADS = Math.max(1,
                 SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
@@ -239,6 +253,10 @@ public class Balancer {
             LOG.info("Closing applicationContext");
             applicationContext.close();
         }
+
+        if (curatorFramework != null) {
+            curatorFramework.close();
+        }
         this.shutdown = true;
         LOG.info("Server shut down.");
     }
@@ -320,8 +338,30 @@ public class Balancer {
                 ssl.getKeyFile(), ssl.getKeyType(), ssl.getKeyPassword());
     }
 
+    private void ensureZkPaths(CuratorFramework curatorFramework, String[] paths) {
+        for (String s : paths) {
+            try {
+                Stat stat = curatorFramework.checkExists().forPath(s);
+                if (stat == null) {
+                    curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT)
+                            .forPath(s);
+                }
+            } catch (Exception e) {
+                LOG.info(e.getMessage());
+            }
+
+        }
+    }
+
     public void run() throws Exception {
 
+        // start curator framework
+        RetryPolicy retryPolicy = new RetryForever(1000);
+        curatorFramework = CuratorFrameworkFactory.newClient(balancerConfig.getZooKeeper().getServers(), 30000, 1000,
+                retryPolicy);
+        curatorFramework.start();
+        ensureZkPaths(curatorFramework, zkPaths);
+
         final boolean useEpoll = useEpoll();
         Class<? extends ServerSocketChannel> channelClass;
         Class<? extends Channel> datagramChannelClass;
@@ -355,7 +395,7 @@ public class Balancer {
             tcpClientPools.add(new TcpClientPool(this.balancerConfig));
         }
         HealthChecker healthChecker = new HealthChecker(this.balancerConfig, tcpClientPools.get(0));
-        this.metricResolver = new BalancedMetricResolver(this.balancerConfig, healthChecker);
+        this.metricResolver = new BalancedMetricResolver(curatorFramework, this.balancerConfig, healthChecker);
 
         final ServerBootstrap tcpServer = new ServerBootstrap();
         tcpServer.group(tcpBossGroup, tcpWorkerGroup);

balancer/src/main/java/timely/balancer/connection/TimelyBalancedHost.java

@@ -1,5 +1,7 @@
 package timely.balancer.connection;
 
+import java.util.Timer;
+
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.apache.commons.lang3.builder.ToStringBuilder;
@@ -8,8 +10,6 @@ import org.slf4j.LoggerFactory;
 import timely.balancer.ArrivalRate;
 import timely.balancer.configuration.BalancerConfiguration;
 
-import java.util.Timer;
-
 public class TimelyBalancedHost {
 
     private static final Logger LOG = LoggerFactory.getLogger(TimelyBalancedHost.class);

balancer/src/main/java/timely/balancer/netty/ws/WsClientHandler.java

@@ -4,7 +4,6 @@ import java.util.concurrent.TimeUnit;
 
 import javax.websocket.CloseReason;
 import javax.websocket.EndpointConfig;
-import javax.websocket.MessageHandler;
 import javax.websocket.Session;
 
 import io.netty.channel.ChannelHandlerContext;
@@ -33,7 +32,7 @@ public class WsClientHandler extends ClientHandler {
 
     @Override
     public void onOpen(Session session, EndpointConfig config) {
-        session.addMessageHandler((MessageHandler.Whole<String>) message -> {
+        session.addMessageHandler(String.class, message -> {
             ctx.writeAndFlush(new TextWebSocketFrame(message));
             LOG.debug("Message received on Websocket session {}: {}", session.getId(), message);
         });

balancer/src/main/java/timely/balancer/resolver/BalancedMetricResolver.java

@@ -4,6 +4,8 @@ import static timely.Server.SERVICE_DISCOVERY_PATH;
 import static timely.balancer.Balancer.ASSIGNMENTS_LAST_UPDATED_PATH;
 import static timely.balancer.Balancer.ASSIGNMENTS_LOCK_PATH;
 import static timely.balancer.Balancer.LEADER_LATCH_PATH;
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS;
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS_LOCK_PATH;
 
 import java.nio.charset.Charset;
 import java.util.ArrayList;
@@ -20,6 +22,8 @@ import java.util.Set;
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -27,10 +31,10 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
 import com.csvreader.CsvReader;
 import com.csvreader.CsvWriter;
 import org.apache.commons.lang.StringUtils;
-import org.apache.curator.RetryPolicy;
+import org.apache.commons.lang3.SerializationUtils;
 import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
+import org.apache.curator.framework.recipes.atomic.DistributedAtomicValue;
 import org.apache.curator.framework.recipes.cache.TreeCache;
 import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
 import org.apache.curator.framework.recipes.cache.TreeCacheListener;
@@ -38,7 +42,7 @@ import org.apache.curator.framework.recipes.leader.LeaderLatch;
 import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
 import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
 import org.apache.curator.framework.state.ConnectionState;
-import org.apache.curator.retry.RetryUntilElapsed;
+import org.apache.curator.retry.RetryForever;
 import org.apache.curator.x.discovery.ServiceCache;
 import org.apache.curator.x.discovery.ServiceDiscovery;
 import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
@@ -49,7 +53,7 @@ import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.CreateMode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import timely.ServerDetails;
@@ -68,46 +72,40 @@ public class BalancedMetricResolver implements MetricResolver {
     private Random r = new Random();
     final private HealthChecker healthChecker;
     private Timer timer = new Timer("RebalanceTimer", true);
-    private Timer arrivalRateTimer = new Timer("AriivalRateTimerResolver", true);
+    private Timer arrivalRateTimer = new Timer("ArrivalRateTimerResolver", true);
     private int roundRobinCounter = 0;
     private Set<String> nonCachedMetrics = new HashSet<>();
-    private ReentrantReadWriteLock nonCachedMetricsLock = new ReentrantReadWriteLock();
+    private DistributedAtomicValue nonCachedMetricsIP;
+    private ReentrantReadWriteLock nonCachedMetricsLocalLock = new ReentrantReadWriteLock();
+    private InterProcessReadWriteLock nonCachedMetricsIPRWLock;
     private BalancerConfiguration balancerConfig;
-    private CuratorFramework curatorFramework;
-    private RetryPolicy retryPolicy = new RetryUntilElapsed(60000, 1000);
     private LeaderLatch leaderLatch;
     private AtomicBoolean isLeader = new AtomicBoolean(false);
-    private InterProcessReadWriteLock assignmentsLock;
+    private InterProcessReadWriteLock assignmentsIPRWLock;
     private DistributedAtomicLong assignmentsLastUpdatedInHdfs;
     private AtomicLong assignmentsLastUpdatedLocal = new AtomicLong(0);
 
-    private String[] zkPaths = new String[] { LEADER_LATCH_PATH, ASSIGNMENTS_LAST_UPDATED_PATH, ASSIGNMENTS_LOCK_PATH,
-            SERVICE_DISCOVERY_PATH };
-
     private enum BalanceType {
-        HIGH_LOW, HIGH_AVG, AVG_LOW;
+        HIGH_LOW, HIGH_AVG, AVG_LOW
     }
 
-    public BalancedMetricResolver(BalancerConfiguration balancerConfig, HealthChecker healthChecker) {
+    public BalancedMetricResolver(CuratorFramework curatorFramework, BalancerConfiguration balancerConfig,
+            HealthChecker healthChecker) {
         this.balancerConfig = balancerConfig;
         this.healthChecker = healthChecker;
 
-        // start curator framework
-        curatorFramework = CuratorFrameworkFactory.newClient(balancerConfig.getZooKeeper().getServers(), 30000, 1000,
-                retryPolicy);
-        curatorFramework.start();
-        ensureZkPaths(curatorFramework, zkPaths);
-        assignmentsLock = new InterProcessReadWriteLock(curatorFramework, ASSIGNMENTS_LOCK_PATH);
+        assignmentsIPRWLock = new InterProcessReadWriteLock(curatorFramework, ASSIGNMENTS_LOCK_PATH);
+        testIPRWLock(curatorFramework, assignmentsIPRWLock, ASSIGNMENTS_LOCK_PATH);
         startLeaderLatch(curatorFramework);
         startServiceListener(curatorFramework);
         assignmentsLastUpdatedInHdfs = new DistributedAtomicLong(curatorFramework, ASSIGNMENTS_LAST_UPDATED_PATH,
-                retryPolicy);
+                new RetryForever(1000));
 
-        TreeCacheListener listener = new TreeCacheListener() {
+        TreeCacheListener assignmentListener = new TreeCacheListener() {
 
             @Override
             public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {
-                LOG.info("Handling event {}. assignmentsLastUpdatedInHdfs:{}", event.getType().toString(),
+                LOG.info("Handling assignments event {}. assignmentsLastUpdatedInHdfs:{}", event.getType().toString(),
                         new Date(assignmentsLastUpdatedInHdfs.get().postValue()));
                 if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
                     long lastLocalUpdate = assignmentsLastUpdatedLocal.get();
@@ -125,20 +123,43 @@ public class BalancedMetricResolver implements MetricResolver {
         };
 
         try {
-            TreeCache treeCache = new TreeCache(curatorFramework, ASSIGNMENTS_LAST_UPDATED_PATH);
-            treeCache.getListenable().addListener(listener);
-            treeCache.start();
+            TreeCache assignmentTreeCache = new TreeCache(curatorFramework, ASSIGNMENTS_LAST_UPDATED_PATH);
+            assignmentTreeCache.getListenable().addListener(assignmentListener);
+            assignmentTreeCache.start();
         } catch (Exception e) {
             LOG.error(e.getMessage(), e);
         }
 
-        nonCachedMetricsLock.writeLock().lock();
+        nonCachedMetricsIPRWLock = new InterProcessReadWriteLock(curatorFramework, NON_CACHED_METRICS_LOCK_PATH);
+        testIPRWLock(curatorFramework, nonCachedMetricsIPRWLock, NON_CACHED_METRICS_LOCK_PATH);
+        nonCachedMetricsIP = new DistributedAtomicValue(curatorFramework, NON_CACHED_METRICS, new RetryForever(1000));
+        TreeCacheListener nonCachedMetricsListener = new TreeCacheListener() {
+
+            @Override
+            public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {
+                if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
+                    LOG.info("Handling nonCachedMetricsIP event {}", event.getType().toString());
+                    readNonCachedMetricsIP();
+                }
+            }
+        };
+
         try {
-            nonCachedMetrics.addAll(balancerConfig.getCache().getNonCachedMetrics());
-        } finally {
-            nonCachedMetricsLock.writeLock().unlock();
+            TreeCache nonCachedMetricsTreeCache = new TreeCache(curatorFramework, NON_CACHED_METRICS);
+            nonCachedMetricsTreeCache.getListenable().addListener(nonCachedMetricsListener);
+            nonCachedMetricsTreeCache.start();
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
         }
 
+        nonCachedMetricsLocalLock.writeLock().lock();
+        try {
+            addNonCachedMetrics(balancerConfig.getCache().getNonCachedMetrics());
+        } finally {
+            nonCachedMetricsLocalLock.writeLock().unlock();
+        }
+        readNonCachedMetricsIP();
+
         readAssignmentsFromHdfs();
 
         timer.schedule(new TimerTask() {
@@ -176,17 +197,44 @@ public class BalancedMetricResolver implements MetricResolver {
         }, 10000, 60000);
     }
 
-    private void ensureZkPaths(CuratorFramework curatorFramework, String[] paths) {
-        for (String s : paths) {
-            try {
-                Stat stat = curatorFramework.checkExists().forPath(s);
-                if (stat == null) {
-                    curatorFramework.create().creatingParentContainersIfNeeded().forPath(s);
-                }
-            } catch (Exception e) {
-                LOG.info(e.getMessage());
-            }
+    private void readNonCachedMetricsIP() {
 
+        Set<String> currentNonCachedMetricsDistributed = new TreeSet<>();
+        try {
+            nonCachedMetricsIPRWLock.readLock().acquire();
+            byte[] currentNonCachedMetricsDistributedBytes = nonCachedMetricsIP.get().postValue();
+            if (currentNonCachedMetricsDistributedBytes != null) {
+                try {
+                    currentNonCachedMetricsDistributed = SerializationUtils
+                            .deserialize(currentNonCachedMetricsDistributedBytes);
+                } catch (Exception e) {
+                    LOG.error(e.getMessage());
+                    currentNonCachedMetricsDistributed = new TreeSet<>();
+                }
+            }
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        } finally {
+            try {
+                nonCachedMetricsIPRWLock.readLock().release();
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            }
+        }
+
+        nonCachedMetricsLocalLock.writeLock().lock();
+        try {
+            if (nonCachedMetrics.containsAll(currentNonCachedMetricsDistributed)) {
+                LOG.info("local nonCachedMetrics already contains {}", currentNonCachedMetricsDistributed);
+            } else {
+                currentNonCachedMetricsDistributed.removeAll(nonCachedMetrics);
+                LOG.info("Adding {} to local nonCachedMetrics", currentNonCachedMetricsDistributed);
+                nonCachedMetrics.addAll(currentNonCachedMetricsDistributed);
+            }
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        } finally {
+            nonCachedMetricsLocalLock.writeLock().unlock();
         }
     }
 
@@ -582,6 +630,70 @@ public class BalancedMetricResolver implements MetricResolver {
         return numReassigned;
     }
 
+    private void addNonCachedMetrics(Collection<String> nonCachedMetricsUpdate) {
+        if (!nonCachedMetricsUpdate.isEmpty()) {
+            try {
+                LOG.info("Adding {} to local nonCachedMetrics", nonCachedMetricsUpdate);
+                nonCachedMetricsLocalLock.writeLock().lock();
+                try {
+                    nonCachedMetrics.addAll(nonCachedMetricsUpdate);
+                } finally {
+                    nonCachedMetricsLocalLock.writeLock().unlock();
+                }
+
+                try {
+                    nonCachedMetricsIPRWLock.writeLock().acquire();
+                    byte[] currentNonCachedMetricsDistributedBytes = nonCachedMetricsIP.get().postValue();
+                    Set<String> currentNonCachedMetricsIP;
+                    if (currentNonCachedMetricsDistributedBytes == null) {
+                        currentNonCachedMetricsIP = new TreeSet<>();
+                    } else {
+                        currentNonCachedMetricsIP = SerializationUtils
+                                .deserialize(currentNonCachedMetricsDistributedBytes);
+                    }
+                    if (currentNonCachedMetricsIP.containsAll(nonCachedMetricsUpdate)) {
+                        LOG.info("nonCachedMetricsIP already contains {}", nonCachedMetricsUpdate);
+                    } else {
+                        nonCachedMetricsUpdate.removeAll(currentNonCachedMetricsIP);
+                        LOG.info("Adding {} to nonCachedMetricsIP", nonCachedMetricsUpdate);
+                        TreeSet<String> updateSet = new TreeSet<>();
+                        updateSet.addAll(currentNonCachedMetricsIP);
+                        updateSet.addAll(nonCachedMetricsUpdate);
+                        byte[] updateValue = SerializationUtils.serialize(updateSet);
+                        nonCachedMetricsIP.trySet(updateValue);
+                        if (!nonCachedMetricsIP.get().succeeded()) {
+                            nonCachedMetricsIP.forceSet(updateValue);
+                        }
+                    }
+                } finally {
+                    nonCachedMetricsIPRWLock.writeLock().release();
+                }
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            }
+        }
+    }
+
+    private void testIPRWLock(CuratorFramework curatorFramework, InterProcessReadWriteLock lock, String path) {
+        try {
+            lock.writeLock().acquire(10, TimeUnit.SECONDS);
+        } catch (Exception e1) {
+            try {
+                curatorFramework.delete().deletingChildrenIfNeeded().forPath(path);
+                curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT)
+                        .forPath(path);
+            } catch (Exception e2) {
+                LOG.info(e2.getMessage(), e2);
+            }
+        } finally {
+            try {
+                lock.writeLock().release();
+            } catch (Exception e3) {
+                LOG.error(e3.getMessage());
+            }
+        }
+    }
+
     private boolean shouldCache(String metricName) {
 
         if (StringUtils.isBlank(metricName)) {
@@ -589,6 +701,14 @@ public class BalancedMetricResolver implements MetricResolver {
         } else {
             balancerLock.readLock().lock();
             try {
+                nonCachedMetricsLocalLock.readLock().lock();
+                try {
+                    if (nonCachedMetrics.contains(metricName)) {
+                        return false;
+                    }
+                } finally {
+                    nonCachedMetricsLocalLock.readLock().unlock();
+                }
                 if (metricToHostMap.containsKey(metricName)) {
                     return true;
                 }
@@ -596,27 +716,17 @@ public class BalancedMetricResolver implements MetricResolver {
                 balancerLock.readLock().unlock();
             }
 
-            nonCachedMetricsLock.readLock().lock();
+            nonCachedMetricsLocalLock.writeLock().lock();
             try {
-                if (nonCachedMetrics.contains(metricName)) {
-                    return false;
-                }
-
                 for (String r : nonCachedMetrics) {
                     if (metricName.matches(r)) {
                         LOG.info("Adding {} to list of non-cached metrics", metricName);
-                        nonCachedMetricsLock.readLock().unlock();
-                        nonCachedMetricsLock.writeLock().lock();
-                        nonCachedMetrics.add(metricName);
+                        addNonCachedMetrics(Collections.singleton(metricName));
                         return false;
                     }
                 }
             } finally {
-                if (nonCachedMetricsLock.isWriteLockedByCurrentThread()) {
-                    nonCachedMetricsLock.writeLock().unlock();
-                } else {
-                    nonCachedMetricsLock.readLock().unlock();
-                }
+                nonCachedMetricsLocalLock.writeLock().unlock();
             }
             return true;
         }
@@ -768,7 +878,8 @@ public class BalancedMetricResolver implements MetricResolver {
 
         Map<String, TimelyBalancedHost> assignedMetricToHostMap = new TreeMap<>();
         try {
-            assignmentsLock.readLock().acquire();
+            assignmentsIPRWLock.readLock().acquire();
+            balancerLock.writeLock().lock();
             Configuration configuration = new Configuration();
             FileSystem fs = FileSystem.get(configuration);
             Path assignmentFile = new Path(balancerConfig.getAssignmentFile());
@@ -798,16 +909,24 @@ public class BalancedMetricResolver implements MetricResolver {
                 }
             }
 
-            balancerLock.writeLock().lock();
+            metricToHostMap.clear();
+            for (Map.Entry<String, TimelyBalancedHost> e : assignedMetricToHostMap.entrySet()) {
+                if (shouldCache(e.getKey())) {
+                    metricToHostMap.put(e.getKey(), e.getValue());
+                }
+            }
+
+            nonCachedMetricsLocalLock.readLock().lock();
             try {
-                metricToHostMap.clear();
-                for (Map.Entry<String, TimelyBalancedHost> e : assignedMetricToHostMap.entrySet()) {
-                    if (shouldCache(e.getKey())) {
-                        metricToHostMap.put(e.getKey(), e.getValue());
+                Iterator<Map.Entry<String, TimelyBalancedHost>> itr = metricToHostMap.entrySet().iterator();
+                while (itr.hasNext()) {
+                    Map.Entry<String, TimelyBalancedHost> e = itr.next();
+                    if (nonCachedMetrics.contains(e.getKey())) {
+                        itr.remove();
                     }
                 }
             } finally {
-                balancerLock.writeLock().unlock();
+                nonCachedMetricsLocalLock.readLock().unlock();
             }
             assignmentsLastUpdatedLocal.set(assignmentsLastUpdatedInHdfs.get().postValue());
             LOG.info("Read {} assignments from hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
@@ -815,8 +934,9 @@ public class BalancedMetricResolver implements MetricResolver {
         } catch (Exception e) {
             LOG.error(e.getMessage(), e);
         } finally {
+            balancerLock.writeLock().unlock();
             try {
-                assignmentsLock.readLock().release();
+                assignmentsIPRWLock.readLock().release();
             } catch (Exception e) {
                 LOG.error(e.getMessage(), e);
             }
@@ -827,8 +947,9 @@ public class BalancedMetricResolver implements MetricResolver {
 
         CsvWriter writer = null;
         try {
-            assignmentsLock.writeLock().acquire();
+            assignmentsIPRWLock.writeLock().acquire();
             balancerLock.readLock().lock();
+            nonCachedMetricsLocalLock.readLock().lock();
             try {
                 if (!metricToHostMap.isEmpty()) {
                     Configuration configuration = new Configuration();
@@ -845,12 +966,14 @@ public class BalancedMetricResolver implements MetricResolver {
                     writer.write("tcpPort");
                     writer.endRecord();
                     for (Map.Entry<String, TimelyBalancedHost> e : metricToHostMap.entrySet()) {
-                        writer.write(e.getKey());
-                        writer.write(e.getValue().getHost());
-                        writer.write(Integer.toString(e.getValue().getTcpPort()));
-                        writer.endRecord();
-                        LOG.trace("Saving assigment: {} to {}:{}", e.getKey(), e.getValue().getHost(),
-                                e.getValue().getTcpPort());
+                        if (!nonCachedMetrics.contains(e.getKey())) {
+                            writer.write(e.getKey());
+                            writer.write(e.getValue().getHost());
+                            writer.write(Integer.toString(e.getValue().getTcpPort()));
+                            writer.endRecord();
+                            LOG.trace("Saving assigment: {} to {}:{}", e.getKey(), e.getValue().getHost(),
+                                    e.getValue().getTcpPort());
+                        }
                     }
 
                     long now = System.currentTimeMillis();
@@ -863,6 +986,7 @@ public class BalancedMetricResolver implements MetricResolver {
                             metricToHostMap.size(), new Date(assignmentsLastUpdatedLocal.get()));
                 }
             } finally {
+                nonCachedMetricsLocalLock.readLock().unlock();
                 balancerLock.readLock().unlock();
             }
         } catch (Exception e) {
@@ -872,7 +996,7 @@ public class BalancedMetricResolver implements MetricResolver {
                 writer.close();
             }
             try {
-                assignmentsLock.writeLock().release();
+                assignmentsIPRWLock.writeLock().release();
             } catch (Exception e) {
                 LOG.error(e.getMessage(), e);
             }
@@ -890,4 +1014,4 @@ public class BalancedMetricResolver implements MetricResolver {
             }
         }
     }
-}
+}

client/pom.xml

@@ -4,12 +4,12 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.0.1-SNAPSHOT</version>
+    <version>1.1.0</version>
   </parent>
   <artifactId>timely-client</artifactId>
   <properties>
     <!-- PEP 440 compliant version gets substituted into src/main/python/template/setup.py -->
-    <python.client.version>0.0.7.dev0</python.client.version>
+    <python.client.version>1.1.0</python.client.version>
   </properties>
   <dependencies>
     <dependency>
@@ -107,4 +107,43 @@
       </plugin>
     </plugins>
   </build>
+  <profiles>
+    <profile>
+      <id>python</id>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.codehaus.mojo</groupId>
+            <artifactId>exec-maven-plugin</artifactId>
+            <executions>
+              <execution>
+                <id>python-dist</id>
+                <goals>
+                  <goal>exec</goal>
+                </goals>
+                <phase>package</phase>
+                <configuration>
+                  <executable>python</executable>
+                  <workingDirectory>${project.basedir}/src/main/python</workingDirectory>
+                  <commandlineArgs>setup.py sdist --dist-dir ${project.build.directory}</commandlineArgs>
+                </configuration>
+              </execution>
+              <execution>
+                <id>python-clean</id>
+                <goals>
+                  <goal>exec</goal>
+                </goals>
+                <phase>package</phase>
+                <configuration>
+                  <executable>rm</executable>
+                  <workingDirectory>${project.basedir}/src/main/python</workingDirectory>
+                  <commandlineArgs>-rf timely.egg-info setup.py</commandlineArgs>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+  </profiles>
 </project>

client/src/main/python/timely/TimeDateRange.py

@@ -2,7 +2,6 @@ import time
 from datetime import datetime
 from datetime import timedelta
 from datetime import tzinfo
-from pandas.tslib import Timedelta
 import pandas
 
 class TimeDateError(ValueError):

collectd-nsq-plugin/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.0.1-SNAPSHOT</version>
+    <version>1.1.0</version>
   </parent>
   <artifactId>collectd-nsq-plugin</artifactId>
   <dependencies>

collectd-plugin-parent/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.0.1-SNAPSHOT</version>
+    <version>1.1.0</version>
   </parent>
   <artifactId>collectd-plugin-parent</artifactId>
   <dependencies>

collectd-timely-plugin/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.0.1-SNAPSHOT</version>
+    <version>1.1.0</version>
   </parent>
   <artifactId>collectd-timely-plugin</artifactId>
   <dependencies>

common/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.0.1-SNAPSHOT</version>
+    <version>1.1.0</version>
   </parent>
   <artifactId>timely-common</artifactId>
   <dependencies>

grafana/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.0.1-SNAPSHOT</version>
+    <version>1.1.0</version>
   </parent>
   <artifactId>timely-app</artifactId>
   <packaging>pom</packaging>

grafana/timely-app/dist/plugin.json

@@ -23,8 +23,8 @@
       {"name": "Project site", "url": "https://github.com/NationalSecurityAgency/timely"},
       {"name": "License & Terms", "url": "https://raw.githubusercontent.com/NationalSecurityAgency/timely/master/LICENSE"}
     ],
-    "version": "0.0.6-SNAPSHOT",
-    "updated": "2017-08-11"
+    "version": "1.1.0",
+    "updated": "2019-01-15"
   },
 
   "includes": [

grafana/timely-app/package.json

@@ -1,7 +1,7 @@
 {
   "name": "timely-app",
   "private": true,
-  "version": "0.0.2",
+  "version": "1.1.0",
   "description": "",
   "main": "index.js",
   "scripts": {

grafana/timely-app/plugin.json

@@ -23,8 +23,8 @@
       {"name": "Project site", "url": "https://github.com/NationalSecurityAgency/timely"},
       {"name": "License & Terms", "url": "https://raw.githubusercontent.com/NationalSecurityAgency/timely/master/LICENSE"}
     ],
-    "version": "0.0.6-SNAPSHOT",
-    "updated": "2017-08-11"
+    "version": "1.1.0",
+    "updated": "2019-01-15"
   },
 
   "includes": [

pom.xml

@@ -3,7 +3,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>timely</groupId>
   <artifactId>timely</artifactId>
-  <version>1.0.1-SNAPSHOT</version>
+  <version>1.1.0</version>
   <packaging>pom</packaging>
   <name>Timely</name>
   <description>A secure and scalable time series database based on Apache Accumulo</description>
@@ -43,7 +43,7 @@
     <version.accumulo>1.7.4</version.accumulo>
     <version.buildnumber-maven>1.4</version.buildnumber-maven>
     <version.caffeine>2.3.3</version.caffeine>
-    <version.collectd>5.5.1</version.collectd>
+    <version.collectd>5.7.1</version.collectd>
     <version.commons-collections4>4.1</version.commons-collections4>
     <version.commons-configuration>1.10</version.commons-configuration>
     <version.commons-io>2.6</version.commons-io>
@@ -59,7 +59,7 @@
     <version.hadoop>2.7.3</version.hadoop>
     <version.httpcomponents>4.5.2</version.httpcomponents>
     <version.impsort>1.2.0</version.impsort>
-    <version.jackson>2.9.7</version.jackson>
+    <version.jackson>2.9.8</version.jackson>
     <version.jacoco>0.7.6.201602180812</version.jacoco>
     <version.javacsv>2.0</version.javacsv>
     <version.javassist>3.21.0-GA</version.javassist>

server/bin/timely-server.sh

@@ -8,11 +8,6 @@ else
         TCNATIVE_SUFFIX="so"
 fi
 
-# netty tcnative file reference
-#netty-tcnative-1.1.33.Fork18-linux-x86_64-fedora.jar -> libnetty-tcnative.so
-#netty-tcnative-1.1.33.Fork18-linux-x86_64.jar -> libnetty-tcnative.so
-#netty-tcnative-1.1.33.Fork18-osx-x86_64.jar -> libnetty-tcnative.jnilib
-
 THIS_DIR="${THIS_SCRIPT%/*}"
 NATIVE_DIR="${THIS_DIR}/META-INF/native"
 BASE_DIR=${THIS_DIR}/..
@@ -32,14 +27,15 @@ fi
 mkdir -p ${NATIVE_DIR}
 
 pushd ${BASE_DIR}/bin
-$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${NATIVE_SUFFIX}
-$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${NATIVE_SUFFIX}
+$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${TCNATIVE_SUFFIX}
+$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
 popd
 
 export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*"
 JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}"
 JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
 JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
+JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}/libnetty-${TCNATIVE_SUFFIX}"
 
 echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.Server --spring.config.name=timely"
 $JAVA_HOME/bin/java ${JVM_ARGS} timely.Server --spring.config.name=timely

server/conf/log4j2-spring.xml

@@ -26,12 +26,12 @@
     </Console>
   </Appenders>
   <Loggers>
-    <Logger name="timely.subscription" level="trace" additivity="false">
-      <AppenderRef ref="Subscription" />
-    </Logger>
-    <Root level="trace" includeLocation="false">
+    <Root level="debug" includeLocation="false">
       <AppenderRef ref="Console" level="warn" />
-      <AppenderRef ref="Rolling" level="trace" />
+      <AppenderRef ref="Rolling" level="debug" />
     </Root>
+    <Logger name="timely" additivity="false">
+      <AppenderRef ref="Console" level="debug" />
+    </Logger>
   </Loggers>
 </Configuration>

server/conf/timely.yml

@@ -4,14 +4,14 @@ timely:
   metric-age-off-days:
     default: 7
   cache:
-    enabled:
+    enabled: true
     age-off-hours:
-      default: 30
+      default: 26
   metrics-report-ignored-tags:
   accumulo:
-    instance-name:
-    zookeepers:
-    username:
+    instance-name: accumulo
+    zookeepers: localhost:2181
+    username: root
     password:
     write:
       latency: 5s
@@ -22,23 +22,23 @@ timely:
       max-downsample-memory: 10000000
   security:
     session-max-age: 86400
-    allow-anonymous-access: false
-    ssl:
+    allow-anonymous-access: true
+    serverSsl:
       certificate-file:
       key-file:
       key-password:
-      use-generated-keypair: false
+      use-generated-keypair: true
       trust-store-file:
       use-openssl: true
       #use-ciphers:
   server:
-    ip:
-    tcp-port:
-    udp-port:
+    ip: 0.0.0.0
+    tcp-port: 4242
+    udp-port: 4542
   http:
-    host:
-    ip:
-    port:
+    host: localhost
+    ip: 0.0.0.0
+    port: 4243
     redirect-path: /secure-me
     strict-transport-max-age: 608400
     cors:
@@ -49,8 +49,8 @@ timely:
       allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
       allowed-headers: content-type
   websocket:
-    ip:
-    port:
+    ip: 0.0.0.0
+    port: 4244
     timeout: 60
     subscription-lag: 120
     scanner-batch-size: 10000

server/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.0.1-SNAPSHOT</version>
+    <version>1.1.0</version>
   </parent>
   <artifactId>timely-server</artifactId>
   <dependencies>

server/src/main/dist/assembly.xml

@@ -33,7 +33,7 @@
       <directory>${project.basedir}/bin</directory>
       <outputDirectory>bin</outputDirectory>
       <includes>
-        <include>*</include>
+        <include>**/*</include>
       </includes>
      </fileSet>
     <fileSet>

server/src/main/java/timely/Server.java

@@ -1,5 +1,8 @@
 package timely;
 
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS;
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS_LOCK_PATH;
+
 import java.io.File;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -63,6 +66,7 @@ import org.apache.curator.x.discovery.ServiceDiscovery;
 import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
 import org.apache.curator.x.discovery.ServiceInstance;
 import org.apache.curator.x.discovery.ServiceInstanceBuilder;
+import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.data.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -125,6 +129,7 @@ public class Server {
     private static final String OS_VERSION = "os.version";
     private static final String WS_PATH = "/websocket";
 
+    private CuratorFramework curatorFramework;
     protected static final CountDownLatch LATCH = new CountDownLatch(1);
     static ConfigurableApplicationContext applicationContext;
 
@@ -146,6 +151,9 @@ public class Server {
     protected volatile boolean shutdown = false;
     private final int DEFAULT_EVENT_LOOP_THREADS;
 
+    private String[] zkPaths = new String[] { SERVICE_DISCOVERY_PATH, NON_CACHED_METRICS,
+            NON_CACHED_METRICS_LOCK_PATH };
+
     private static boolean useEpoll() {
 
         // Should we just return true if this is Linux and if we get an error
@@ -175,13 +183,8 @@ public class Server {
         }
     }
 
-    public void registerService() {
+    public void registerService(CuratorFramework curatorFramework) {
         try {
-            RetryPolicy retryPolicy = new RetryForever(1000);
-            CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(config.getAccumulo().getZookeepers(),
-                    30000, 10000, retryPolicy);
-            curatorFramework.start();
-
             try {
                 Stat stat = curatorFramework.checkExists().forPath(SERVICE_DISCOVERY_PATH);
                 if (stat == null) {
@@ -369,6 +372,10 @@ public class Server {
             LOG.error("Error closing WebSocketRequestDecoder during shutdown", e);
         }
 
+        if (curatorFramework != null) {
+            curatorFramework.close();
+        }
+
         if (applicationContext != null) {
             LOG.info("Closing applicationContext");
             applicationContext.close();
@@ -390,13 +397,33 @@ public class Server {
         this.config = conf;
     }
 
+    private void ensureZkPaths(CuratorFramework curatorFramework, String[] paths) {
+        for (String s : paths) {
+            try {
+                Stat stat = curatorFramework.checkExists().forPath(s);
+                if (stat == null) {
+                    curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT)
+                            .forPath(s);
+                }
+            } catch (Exception e) {
+                LOG.info(e.getMessage());
+            }
+        }
+    }
+
     public void run() throws Exception {
 
+        RetryPolicy retryPolicy = new RetryForever(1000);
+        curatorFramework = CuratorFrameworkFactory.newClient(config.getAccumulo().getZookeepers(), 30000, 10000,
+                retryPolicy);
+        curatorFramework.start();
+        ensureZkPaths(curatorFramework, zkPaths);
+
         int nettyThreads = Math.max(1,
                 SystemPropertyUtil.getInt("io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2));
         dataStore = DataStoreFactory.create(config, nettyThreads);
         if (config.getCache().isEnabled()) {
-            dataStoreCache = new DataStoreCache(config);
+            dataStoreCache = new DataStoreCache(curatorFramework, config);
             dataStoreCache.setInternalMetrics(dataStore.getInternalMetrics());
             dataStore.setCache(dataStoreCache);
         }
@@ -500,7 +527,7 @@ public class Server {
             udpServer.option(EpollChannelOption.SO_REUSEPORT, true);
             udpChannelHandleList.add(udpServer.bind(udpIp, udpPort).sync().channel());
         }
-        registerService();
+        registerService(curatorFramework);
         shutdownHook();
         LOG.info(
                 "Server started. Listening on {}:{} for TCP traffic, {}:{} for HTTP traffic, {}:{} for WebSocket traffic, and {}:{} for UDP traffic",
@@ -649,7 +676,8 @@ public class Server {
                 ch.pipeline().addLast("suggest", new WSSuggestRequestHandler(dataStore));
                 ch.pipeline().addLast("version", new WSVersionRequestHandler());
                 ch.pipeline().addLast("put", new WSMetricPutHandler(dataStore));
-                ch.pipeline().addLast("create", new WSCreateSubscriptionRequestHandler(dataStore, config));
+                ch.pipeline().addLast("create",
+                        new WSCreateSubscriptionRequestHandler(dataStore, dataStoreCache, config));
                 ch.pipeline().addLast("add", new WSAddSubscriptionRequestHandler());
                 ch.pipeline().addLast("remove", new WSRemoveSubscriptionRequestHandler());
                 ch.pipeline().addLast("close", new WSCloseSubscriptionRequestHandler());

server/src/main/java/timely/netty/websocket/WebSocketRequestDecoder.java

@@ -7,6 +7,7 @@ import io.netty.handler.codec.MessageToMessageDecoder;
 import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
 import io.netty.handler.ssl.SslCompletionEvent;
 import io.netty.handler.timeout.IdleState;
 import io.netty.handler.timeout.IdleStateEvent;
@@ -99,10 +100,11 @@ public class WebSocketRequestDecoder extends MessageToMessageDecoder<WebSocketFr
                 }
             }
         } else if (evt instanceof SslCompletionEvent) {
-            SslCompletionEvent ssl = (SslCompletionEvent) evt;
-            if (!ssl.isSuccess()) {
-                LOG.error("SSL error: {}", ssl.getClass().getSimpleName(), ssl.cause());
-            }
+            LOG.debug("{}", ((SslCompletionEvent) evt).getClass().getSimpleName());
+        } else if (evt instanceof WebSocketServerProtocolHandler.ServerHandshakeStateEvent) {
+            // The handshake completed succesfully and the channel was upgraded to
+            // websockets
+            LOG.trace("SSL handshake completed successfully, upgraded channel to websockets");
         } else {
             LOG.warn("Received unhandled user event {}", evt);
         }

server/src/main/java/timely/netty/websocket/subscription/WSCreateSubscriptionRequestHandler.java

@@ -9,6 +9,7 @@ import org.slf4j.LoggerFactory;
 import timely.api.request.subscription.CreateSubscription;
 import timely.configuration.Configuration;
 import timely.store.DataStore;
+import timely.store.cache.DataStoreCache;
 import timely.subscription.Subscription;
 import timely.subscription.SubscriptionRegistry;
 
@@ -16,10 +17,12 @@ public class WSCreateSubscriptionRequestHandler extends SimpleChannelInboundHand
 
     private static final Logger LOG = LoggerFactory.getLogger(WSCreateSubscriptionRequestHandler.class);
     private final DataStore store;
+    private final DataStoreCache cache;
     private final Configuration conf;
 
-    public WSCreateSubscriptionRequestHandler(DataStore store, Configuration conf) {
+    public WSCreateSubscriptionRequestHandler(DataStore store, DataStoreCache cache, Configuration conf) {
         this.store = store;
+        this.cache = cache;
         this.conf = conf;
     }
 
@@ -27,11 +30,11 @@ public class WSCreateSubscriptionRequestHandler extends SimpleChannelInboundHand
     protected void channelRead0(ChannelHandlerContext ctx, CreateSubscription create) throws Exception {
         final String subscriptionId = create.getSubscriptionId();
         SubscriptionRegistry.get().put(subscriptionId,
-                new Subscription(subscriptionId, create.getSessionId(), store, ctx, this.conf));
+                new Subscription(subscriptionId, create.getSessionId(), store, cache, ctx, this.conf));
 
         // Store the session id as an attribute on the context.
         ctx.channel().attr(SubscriptionRegistry.SUBSCRIPTION_ID_ATTR).set(subscriptionId);
-        LOG.info("Created subscription {} on channel {}", subscriptionId, ctx);
+        LOG.info("[{}] Created subscription on channel {}", subscriptionId, ctx);
 
         ctx.channel().closeFuture().addListener(new ChannelFutureListener() {
 
@@ -39,7 +42,7 @@ public class WSCreateSubscriptionRequestHandler extends SimpleChannelInboundHand
             public void operationComplete(ChannelFuture future) throws Exception {
                 Subscription s = SubscriptionRegistry.get().remove(subscriptionId);
                 if (null != s) {
-                    LOG.info("Channel closed, closing subscriptions for subscriptionId: " + subscriptionId);
+                    LOG.info("[{}] Channel closed, closing subscriptions", subscriptionId);
                     s.close();
                 }
             }

server/src/main/java/timely/store/DataStore.java

@@ -36,8 +36,8 @@ public interface DataStore {
 
     Set<Tag> getColumnFamilies(String metric, Map<String, String> tags) throws TableNotFoundException;
 
-    Scanner createScannerForMetric(String sessionId, String metric, Map<String, String> tags, long startTime,
-            long endTime, int lag, int scannerBatchSize, int scannerReadAhead) throws TimelyException;
+    Scanner createScannerForMetric(String sessionId, String metric, Map<String, String> tags, int scannerBatchSize,
+            int scannerReadAhead) throws TimelyException;
 
     void setCache(DataStoreCache cache);
 

server/src/main/java/timely/store/DataStoreImpl.java

@@ -982,8 +982,8 @@ public class DataStoreImpl implements DataStore {
         }
     }
 
-    public Scanner createScannerForMetric(String sessionId, String metric, Map<String, String> tags, long startTime,
-            long endTime, int lag, int scannerBatchSize, int scannerReadAhead) throws TimelyException {
+    public Scanner createScannerForMetric(String sessionId, String metric, Map<String, String> tags,
+            int scannerBatchSize, int scannerReadAhead) throws TimelyException {
         try {
             Authorizations auths = null;
             try {

server/src/main/java/timely/store/cache/DataStoreCache.java

@@ -7,11 +7,14 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.Timer;
 import java.util.TimerTask;
+import java.util.TreeSet;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.StampedLock;
 
 import io.netty.handler.codec.http.HttpResponseStatus;
@@ -21,12 +24,23 @@ import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.accumulo.core.security.Authorizations;
+import org.apache.commons.lang3.SerializationUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.atomic.DistributedAtomicValue;
+import org.apache.curator.framework.recipes.cache.TreeCache;
+import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
+import org.apache.curator.framework.recipes.cache.TreeCacheListener;
+import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
+import org.apache.curator.retry.RetryForever;
+import org.apache.zookeeper.CreateMode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import timely.adapter.accumulo.MetricAdapter;
 import timely.api.request.AuthenticatedRequest;
 import timely.api.request.timeseries.QueryRequest;
 import timely.api.response.CacheResponse;
+import timely.api.response.MetricResponse;
 import timely.api.response.TimelyException;
 import timely.api.response.timeseries.QueryResponse;
 import timely.auth.AuthCache;
@@ -45,12 +59,17 @@ import timely.store.iterators.RateIterator;
 public class DataStoreCache {
 
     private static final Logger LOG = LoggerFactory.getLogger(DataStoreCache.class);
+    public static final String NON_CACHED_METRICS = "/timely/server/non_cached_metrics/data";
+    public static final String NON_CACHED_METRICS_LOCK_PATH = "/timely/server/non_cached_metrics/lock";
+
     public static final String DEFAULT_AGEOFF_KEY = "default";
 
     private Map<String, Map<TaggedMetric, GorillaStore>> gorillaMap = new HashMap<>();
     private StampedLock gorillaMapLock = new StampedLock();
 
     private Set<String> nonCachedMetrics = Collections.synchronizedSet(new HashSet<>());
+    private DistributedAtomicValue nonCachedMetricsIP = null;
+    private InterProcessReadWriteLock nonCachedMetricsIPRWLock;
     private long maxUniqueTagSets;
     private boolean anonAccessAllowed;
     private Map<String, Long> minimumAgeOff;
@@ -58,11 +77,16 @@ public class DataStoreCache {
     private int flushBatch = 0;
     private int numBatches = 5;
     private InternalMetrics internalMetrics;
-
     private Timer maintenanceTimer = new Timer("DataStoreCacheTimer");
 
-    public DataStoreCache(Configuration conf) {
-        nonCachedMetrics.addAll(conf.getCache().getNonCachedMetrics());
+    public DataStoreCache(CuratorFramework curatorFramework, Configuration conf) {
+        if (curatorFramework != null) {
+            addNonCachedMetricsListener(curatorFramework);
+        }
+        LOG.info("Adding initial values from configuration");
+        addNonCachedMetrics(conf.getCache().getNonCachedMetrics());
+        LOG.info("Reading initial values from nonCachedMetricsIP");
+        readNonCachedMetricsIP();
         maxUniqueTagSets = conf.getCache().getMaxUniqueTagSets();
         anonAccessAllowed = conf.getSecurity().isAllowAnonymousAccess();
         Map<String, Integer> cacheAgeOff = conf.getCache().getMetricAgeOffHours();
@@ -142,6 +166,124 @@ public class DataStoreCache {
         }, (60 * 1000), (60 * 1000));
     }
 
+    private void addNonCachedMetrics(Collection<String> nonCachedMetricsUpdate) {
+        if (!nonCachedMetricsUpdate.isEmpty()) {
+            try {
+                LOG.info("Adding {} to local nonCachedMetrics", nonCachedMetricsUpdate);
+                nonCachedMetrics.addAll(nonCachedMetricsUpdate);
+                if (nonCachedMetricsIP != null) {
+                    try {
+                        nonCachedMetricsIPRWLock.writeLock().acquire();
+                        byte[] currentNonCachedMetricsDistributedBytes = nonCachedMetricsIP.get().postValue();
+                        Set<String> currentNonCachedMetricsIP;
+                        if (currentNonCachedMetricsDistributedBytes == null) {
+                            currentNonCachedMetricsIP = new TreeSet<>();
+                        } else {
+                            try {
+                                currentNonCachedMetricsIP = SerializationUtils
+                                        .deserialize(currentNonCachedMetricsDistributedBytes);
+                            } catch (Exception e) {
+                                LOG.error(e.getMessage());
+                                currentNonCachedMetricsIP = new TreeSet<>();
+                            }
+                        }
+                        if (currentNonCachedMetricsIP.containsAll(nonCachedMetricsUpdate)) {
+                            LOG.info("nonCachedMetricsIP already contains {}", nonCachedMetricsUpdate);
+                        } else {
+                            nonCachedMetricsUpdate.removeAll(currentNonCachedMetricsIP);
+                            LOG.info("Adding {} to nonCachedMetricsIP", nonCachedMetricsUpdate);
+                            TreeSet<String> updateSet = new TreeSet<>();
+                            updateSet.addAll(currentNonCachedMetricsIP);
+                            updateSet.addAll(nonCachedMetricsUpdate);
+                            byte[] updateValue = SerializationUtils.serialize(updateSet);
+                            nonCachedMetricsIP.trySet(updateValue);
+                            if (!nonCachedMetricsIP.get().succeeded()) {
+                                nonCachedMetricsIP.forceSet(updateValue);
+                            }
+                        }
+                    } finally {
+                        nonCachedMetricsIPRWLock.writeLock().release();
+                    }
+                }
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            }
+        }
+    }
+
+    private void addNonCachedMetricsListener(CuratorFramework curatorFramework) {
+        nonCachedMetricsIPRWLock = new InterProcessReadWriteLock(curatorFramework, NON_CACHED_METRICS_LOCK_PATH);
+        testIPRWLock(curatorFramework, nonCachedMetricsIPRWLock, NON_CACHED_METRICS_LOCK_PATH);
+        nonCachedMetricsIP = new DistributedAtomicValue(curatorFramework, NON_CACHED_METRICS, new RetryForever(1000));
+        TreeCacheListener nonCachedMetricsListener = new TreeCacheListener() {
+
+            @Override
+            public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {
+                if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
+                    LOG.info("Handling nonCachedMetricsIP event {}", event.getType().toString());
+                    readNonCachedMetricsIP();
+                }
+            }
+        };
+
+        try {
+            TreeCache nonCachedMetricsTreeCache = new TreeCache(curatorFramework, NON_CACHED_METRICS);
+            nonCachedMetricsTreeCache.getListenable().addListener(nonCachedMetricsListener);
+            nonCachedMetricsTreeCache.start();
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        }
+    }
+
+    private void readNonCachedMetricsIP() {
+        try {
+            nonCachedMetricsIPRWLock.readLock().acquire();
+            byte[] nonCachedMetricsUpdateBytes = nonCachedMetricsIP.get().postValue();
+            Set<String> nonCachedMetricsUpdate;
+            if (nonCachedMetricsUpdateBytes == null) {
+                nonCachedMetricsUpdate = new TreeSet<>();
+            } else {
+                nonCachedMetricsUpdate = SerializationUtils.deserialize(nonCachedMetricsIP.get().postValue());
+            }
+
+            if (nonCachedMetrics.containsAll(nonCachedMetricsUpdate)) {
+                LOG.info("local nonCachedMetrics already contains {}", nonCachedMetricsUpdate);
+            } else {
+                nonCachedMetricsUpdate.removeAll(nonCachedMetrics);
+                LOG.info("Adding {} to local nonCachedMetrics", nonCachedMetricsUpdate);
+                nonCachedMetrics.addAll(nonCachedMetricsUpdate);
+            }
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        } finally {
+            try {
+                nonCachedMetricsIPRWLock.readLock().release();
+            } catch (Exception e) {
+                LOG.error(e.getMessage());
+            }
+        }
+    }
+
+    private void testIPRWLock(CuratorFramework curatorFramework, InterProcessReadWriteLock lock, String path) {
+        try {
+            lock.writeLock().acquire(10, TimeUnit.SECONDS);
+        } catch (Exception e1) {
+            try {
+                curatorFramework.delete().deletingChildrenIfNeeded().forPath(path);
+                curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT)
+                        .forPath(path);
+            } catch (Exception e2) {
+                LOG.info(e2.getMessage(), e2);
+            }
+        } finally {
+            try {
+                lock.writeLock().release();
+            } catch (Exception e3) {
+                LOG.error(e3.getMessage());
+            }
+        }
+    }
+
     public void setInternalMetrics(InternalMetrics internalMetrics) {
         this.internalMetrics = internalMetrics;
     }
@@ -204,7 +346,7 @@ public class DataStoreCache {
                     LOG.info("Cache of metric {} has {} tag variations.  Discontinuing cache.", entry1.getKey(),
                             numberTagVariations);
                     metricItr.remove();
-                    nonCachedMetrics.add(entry1.getKey());
+                    addNonCachedMetrics(Collections.singleton(entry1.getKey()));
                 }
             }
         } finally {
@@ -396,6 +538,10 @@ public class DataStoreCache {
 
     private boolean shouldCache(Metric metric) {
         String metricName = metric.getName();
+        if (nonCachedMetrics.contains(metricName)) {
+            return false;
+        }
+
         long stamp = gorillaMapLock.readLock();
         try {
             if (gorillaMap.containsKey(metricName)) {
@@ -405,17 +551,14 @@ public class DataStoreCache {
             gorillaMapLock.unlockRead(stamp);
         }
 
-        if (nonCachedMetrics.contains(metricName)) {
-            return false;
-        }
-
+        Set<String> tempNonCachedMetrics = new LinkedHashSet<>();
         synchronized (nonCachedMetrics) {
-            for (String r : nonCachedMetrics) {
-                if (metricName.matches(r)) {
-                    LOG.info("Adding {} to list of non-cached metrics", metricName);
-                    nonCachedMetrics.add(metricName);
-                    return false;
-                }
+            tempNonCachedMetrics.addAll(nonCachedMetrics);
+        }
+        for (String r : tempNonCachedMetrics) {
+            if (metricName.matches(r)) {
+                addNonCachedMetrics(Collections.singleton(metricName));
+                return false;
             }
         }
         return true;
@@ -490,6 +633,39 @@ public class DataStoreCache {
         }
     }
 
+    public List<MetricResponse> getMetricsFromCache(String metric, Map<String, String> tags, long begin, long end,
+            String sessionId) {
+
+        List<MetricResponse> metricResponses = new ArrayList<>();
+        QueryRequest.SubQuery subQuery = new QueryRequest.SubQuery();
+        subQuery.setMetric(metric);
+        if (tags != null) {
+            for (Map.Entry<String, String> t : tags.entrySet()) {
+                subQuery.addTag(t.getKey(), t.getValue());
+            }
+        }
+        VisibilityFilter visFilter = new VisibilityFilter(getSessionAuthorizations(sessionId));
+        DataStoreCacheIterator itr = new DataStoreCacheIterator(this, visFilter, subQuery, begin, end);
+        try {
+            itr.seek(new Range(subQuery.getMetric()), null, true);
+            while (itr.hasTop()) {
+                Key k = itr.getTopKey();
+                Value v = itr.getTopValue();
+                Metric m = MetricAdapter.parse(k, v);
+                MetricResponse mr = new MetricResponse();
+                mr.setMetric(metric);
+                mr.setTags(m.getTags());
+                mr.setTimestamp(m.getValue().getTimestamp());
+                mr.setValue(m.getValue().getMeasure());
+                metricResponses.add(mr);
+                itr.next();
+            }
+        } catch (IOException e) {
+            LOG.error(e.getMessage(), e);
+        }
+        return metricResponses;
+    }
+
     protected SortedKeyValueIterator<Key, Value> setupIterator(QueryRequest query, QueryRequest.SubQuery subQuery,
             Authorizations authorizations, long ageOffForMetric) throws TimelyException {
 

server/src/main/java/timely/subscription/MetricScanner.java

@@ -3,6 +3,7 @@ package timely.subscription;
 import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
 
 import java.lang.Thread.UncaughtExceptionHandler;
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -22,6 +23,7 @@ import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
 import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import timely.adapter.accumulo.MetricAdapter;
@@ -31,6 +33,7 @@ import timely.api.response.TimelyException;
 import timely.model.Metric;
 import timely.model.Tag;
 import timely.store.DataStore;
+import timely.store.cache.DataStoreCache;
 import timely.util.JsonUtil;
 
 public class MetricScanner extends Thread implements UncaughtExceptionHandler {
@@ -43,79 +46,143 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
     private final ChannelHandlerContext ctx;
     private volatile boolean closed = false;
     private final long delay;
-    private final String name;
+    private String scannerInfo;
     private final int lag;
     private final String subscriptionId;
     private final String metric;
-    private long startTime;
+    private long beginTime;
     private long endTime;
     private final Subscription subscription;
-    private MetricResponses responses = new MetricResponses();
-    private final ScheduledFuture<?> flusher;
     private final int subscriptionBatchSize;
+    private MetricResponses responses = new MetricResponses();
+    private ScheduledFuture<?> flusher = null;
     private List<Range> ranges = null;
     private Iterator<Range> rangeItr = null;
     private DataStore store = null;
+    private DataStoreCache cache = null;
     private Set<Tag> colFamValues = null;
     private boolean completedResponseSent = false;
+    private boolean done = false;
 
-    public MetricScanner(Subscription sub, String subscriptionId, String sessionId, DataStore store, String metric,
-            Map<String, String> tags, long startTime, long endTime, long delay, int lag, ChannelHandlerContext ctx,
-            int scannerBatchSize, int flushIntervalSeconds, int scannerReadAhead, int subscriptionBatchSize)
-            throws TimelyException {
+    public MetricScanner(Subscription sub, String subscriptionId, String sessionId, DataStore store,
+            DataStoreCache cache, String metric, Map<String, String> tags, long beginTime, long endTime, long delay,
+            int lag, ChannelHandlerContext ctx, int scannerBatchSize, int flushIntervalSeconds, int scannerReadAhead,
+            int subscriptionBatchSize) throws TimelyException {
         this.setDaemon(true);
         this.setUncaughtExceptionHandler(this);
         this.store = store;
+        this.cache = cache;
         this.subscription = sub;
         this.ctx = ctx;
         this.lag = lag;
         this.metric = metric;
-        this.startTime = startTime;
+        this.beginTime = beginTime;
         this.endTime = endTime;
-        this.scanner = this.store.createScannerForMetric(sessionId, metric, tags, startTime, endTime, lag,
-                scannerBatchSize, scannerReadAhead);
-
-        if (0 == startTime) {
-            this.startTime = (System.currentTimeMillis() - this.store.getAgeOffForMetric(metric) - 1000);
-            LOG.debug("Overriding zero start time to {} due to age off configuration", startTime);
-        }
-        long endTimeStamp = (endTime == 0) ? (System.currentTimeMillis() - (lag * 1000)) : endTime;
-
-        try {
-            this.colFamValues = this.store.getColumnFamilies(metric, tags);
-            ranges = this.store.getQueryRanges(metric, this.startTime, endTimeStamp, colFamValues);
-            rangeItr = ranges.iterator();
-            if (rangeItr.hasNext()) {
-                Range r = rangeItr.next();
-                this.scanner.setRange(r);
-            }
-        } catch (TableNotFoundException e) {
-            throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), "Error in MetricScanner",
-                    e.getMessage(), e);
-        }
-        this.iter = scanner.iterator();
         this.delay = delay;
         this.subscriptionId = subscriptionId;
         this.subscriptionBatchSize = subscriptionBatchSize;
-        ToStringBuilder buf = new ToStringBuilder(this);
+
+        ToStringBuilder buf = new ToStringBuilder(this, ToStringStyle.JSON_STYLE);
         buf.append("sessionId", sessionId);
+        buf.append("subscriptionId", subscriptionId);
         buf.append("metric", metric);
-        buf.append("startTime", startTime);
+        buf.append("startTime", beginTime);
         buf.append("endTime", endTime);
         buf.append("delayTime", delay);
         if (null != tags) {
             buf.append("tags", tags.toString());
         }
-        name = buf.toString();
-        this.setName("Metric Scanner " + name);
-        this.flusher = this.ctx.executor().scheduleAtFixedRate(() -> {
-            flush();
-        }, flushIntervalSeconds, flushIntervalSeconds, TimeUnit.SECONDS);
-        LOG.trace("Created MetricScanner: {}", name);
+        scannerInfo = buf.toString();
+        LOG.info("[{}] Setting up MetricScanner: {}", subscriptionId, scannerInfo);
+
+        if (0 == beginTime) {
+            this.beginTime = (System.currentTimeMillis() - this.store.getAgeOffForMetric(metric) - 1000);
+            LOG.debug("[{}] Overriding zero start time to {} due to age off configuration", subscriptionId, beginTime);
+        }
+        long endTimeStamp = (endTime == 0) ? (System.currentTimeMillis() - (lag * 1000)) : endTime;
+
+        try {
+            this.colFamValues = this.store.getColumnFamilies(metric, tags);
+            this.ranges = new ArrayList<>();
+            if (this.cache == null) {
+                LOG.debug("[{}] Cache not enabled, adding complete range", subscriptionId);
+                ranges.addAll(this.store.getQueryRanges(metric, this.beginTime, endTimeStamp, colFamValues));
+            } else {
+                long oldestTsForMetric = this.cache.getOldestTimestamp(metric);
+                // metric is cached on this server and requested range ends somewhere in the
+                // cache
+                if (oldestTsForMetric < Long.MAX_VALUE && endTimeStamp > oldestTsForMetric) {
+                    List<MetricResponse> metricsFromCache;
+                    if (this.beginTime < oldestTsForMetric) {
+                        // add range this.beginTime to oldestTsForMetric - 1 because this range is not
+                        // cached
+                        ranges.addAll(
+                                this.store.getQueryRanges(metric, this.beginTime, oldestTsForMetric - 1, colFamValues));
+                        metricsFromCache = cache.getMetricsFromCache(metric, tags, oldestTsForMetric, endTimeStamp,
+                                sessionId);
+                        LOG.debug("[{}] MetricScanner request partially fulfilled from cache: {} metrics",
+                                subscriptionId, metricsFromCache.size());
+                    } else {
+                        metricsFromCache = cache.getMetricsFromCache(metric, tags, this.beginTime, endTimeStamp,
+                                sessionId);
+                        LOG.debug("[{}] Websocket request completely fulfilled from cache: {} metrics [{}]",
+                                subscriptionId, metricsFromCache.size());
+                    }
+
+                    if (metricsFromCache.isEmpty()) {
+                        // no metrics from cache, clear ranges and add whole range
+                        ranges.clear();
+                        ranges.addAll(this.store.getQueryRanges(metric, this.beginTime, endTimeStamp, colFamValues));
+                        LOG.debug("[{}] MetricScanner request got no metrics from cache, re-adding complete range",
+                                subscriptionId);
+                    } else {
+                        for (MetricResponse r : metricsFromCache) {
+                            responses.addResponse(r);
+                        }
+                        flush();
+                    }
+                } else {
+                    LOG.debug("[{}] MetricScanner request range not cached, adding complete range", subscriptionId);
+                    ranges.addAll(this.store.getQueryRanges(metric, this.beginTime, endTimeStamp, colFamValues));
+                }
+            }
+
+            if (ranges.isEmpty() && endTime == 0) {
+                // request fulfilled from cache up until current, but user asked for continuous
+                // data
+                // must add a small range otherwise we will be scanning all data regardless of
+                // metric, tag
+                ranges.addAll(this.store.getQueryRanges(metric, endTimeStamp, endTimeStamp + 1, colFamValues));
+            }
+
+            if (ranges.isEmpty()) {
+                this.scanner = null;
+                this.done = true;
+            } else {
+                this.scanner = this.store.createScannerForMetric(sessionId, metric, tags, scannerBatchSize,
+                        scannerReadAhead);
+                rangeItr = ranges.iterator();
+                if (rangeItr.hasNext()) {
+                    Range r = rangeItr.next();
+                    this.scanner.setRange(r);
+                }
+            }
+        } catch (TableNotFoundException e) {
+            throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
+                    "[" + subscriptionId + "] Error in MetricScanner", e.getMessage(), e);
+        }
+
+        if (!done) {
+            this.iter = scanner.iterator();
+            this.setName("Scan:" + subscriptionId);
+            this.flusher = this.ctx.executor().scheduleAtFixedRate(() -> {
+                flush();
+            }, flushIntervalSeconds, flushIntervalSeconds, TimeUnit.SECONDS);
+        }
     }
 
     public synchronized void flush() {
-        LOG.info("Flush called");
+        LOG.debug("[{}] Flush called", subscriptionId);
         synchronized (responses) {
             if (responses.size() > 0) {
                 try {
@@ -123,7 +190,7 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
                     this.ctx.writeAndFlush(new TextWebSocketFrame(json));
                     responses.clear();
                 } catch (JsonProcessingException e) {
-                    LOG.error("Error serializing metrics: " + responses, e);
+                    LOG.error("[" + subscriptionId + "] Error serializing metrics: " + responses, e);
                 }
             }
         }
@@ -133,7 +200,6 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
     public void run() {
         Metric m = null;
         try {
-            boolean done = false;
             while (!done && !closed) {
 
                 if (this.iter.hasNext()) {
@@ -145,7 +211,8 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
                         }
                         this.responses.addResponse(MetricResponse.fromMetric(m, this.subscriptionId));
                     } catch (Exception e1) {
-                        LOG.error("Error {} parsing metric at key: {}", e1.getMessage(), e.getKey().toString());
+                        LOG.error("[{}] Error {} parsing metric at key: {}", subscriptionId, e1.getMessage(),
+                                e.getKey().toString());
                     }
                 } else if (rangeItr.hasNext()) {
                     // set next range on the scanner
@@ -158,9 +225,10 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
                     long endTimeStamp = (System.currentTimeMillis() - (lag * 1000));
                     this.scanner.close();
                     if (null == m) {
-                        LOG.debug("No results found, waiting {}ms to retry with new end time {}.", delay, endTimeStamp);
+                        LOG.debug("[{}] No results found, waiting {}ms to retry with new end time {}. [{}]",
+                                subscriptionId, delay, endTimeStamp);
                         sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
-                        ranges = this.store.getQueryRanges(metric, startTime, endTimeStamp, colFamValues);
+                        ranges = this.store.getQueryRanges(metric, beginTime, endTimeStamp, colFamValues);
                         rangeItr = ranges.iterator();
                         if (rangeItr.hasNext()) {
                             Range r = rangeItr.next();
@@ -169,8 +237,9 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
                         }
                     } else {
                         // Reset the starting range to the last key returned
-                        LOG.debug("Exhausted scanner, last metric returned was {}", m);
-                        LOG.debug("Waiting {}ms to retry with new end time {}.", delay, endTimeStamp);
+                        LOG.debug("[{}] Exhausted scanner, last metric returned was {} [{}]", subscriptionId, m);
+                        LOG.debug("[{}] Waiting {}ms to retry with new end time {}. [{}]", subscriptionId, delay,
+                                endTimeStamp);
                         sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
                         ranges = this.store.getQueryRanges(metric, m.getValue().getTimestamp() + 1, endTimeStamp,
                                 colFamValues);
@@ -187,22 +256,24 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
                     done = true;
                 }
                 if (done) {
-                    LOG.debug("Exhausted scanner for subscription {}", this.subscriptionId);
+                    LOG.debug("[{}] Exhausted scanner", subscriptionId);
                     sendCompletedResponse();
                 }
             }
         } catch (Throwable e) {
-            LOG.error("Error in metric scanner, closing.", e);
+            LOG.error("[" + subscriptionId + "] Error in metric scanner, closing.", e);
         } finally {
             close();
-            this.scanner.close();
+            if (this.scanner != null) {
+                this.scanner.close();
+            }
             subscription.scannerComplete(metric);
         }
     }
 
     synchronized private void sendCompletedResponse() {
         if (!completedResponseSent) {
-            LOG.debug("Sending completed response for subscription {}", this.subscriptionId);
+            LOG.info("[{}] Sending completed response", subscriptionId);
             final MetricResponse completedResponse = new MetricResponse();
             completedResponse.setSubscriptionId(this.subscriptionId);
             completedResponse.setComplete(true);
@@ -214,16 +285,18 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
 
     public void close() {
         if (!closed) {
-            LOG.info("Marking metric scanner closed: {}", name);
+            LOG.debug("[{}] Marking metric scanner closed", subscriptionId);
             sendCompletedResponse();
-            this.flusher.cancel(false);
+            if (this.flusher != null) {
+                this.flusher.cancel(false);
+            }
             this.closed = true;
         }
     }
 
     @Override
     public void uncaughtException(Thread t, Throwable e) {
-        LOG.error("Error during metric scanner " + name, e);
+        LOG.error("[" + subscriptionId + "] Error during metric scanner", e);
         this.close();
     }
 }

server/src/main/java/timely/subscription/Subscription.java

@@ -13,6 +13,7 @@ import org.slf4j.LoggerFactory;
 import timely.api.response.TimelyException;
 import timely.configuration.Configuration;
 import timely.store.DataStore;
+import timely.store.cache.DataStoreCache;
 
 public class Subscription {
 
@@ -21,6 +22,7 @@ public class Subscription {
 
     private final String sessionId;
     private final DataStore store;
+    private final DataStoreCache cache;
     private final ChannelHandlerContext ctx;
     private final ScheduledFuture<?> ping;
     private final int lag;
@@ -31,11 +33,12 @@ public class Subscription {
     private final String subscriptionId;
     private LinkedList<String> deadMetricScannerQueue = new LinkedList<>();
 
-    public Subscription(String subscriptionId, String sessionId, DataStore store, ChannelHandlerContext ctx,
-            Configuration conf) {
+    public Subscription(String subscriptionId, String sessionId, DataStore store, DataStoreCache cache,
+            ChannelHandlerContext ctx, Configuration conf) {
         this.subscriptionId = subscriptionId;
         this.sessionId = sessionId;
         this.store = store;
+        this.cache = cache;
         this.ctx = ctx;
         this.lag = conf.getWebsocket().getSubscriptionLag();
         this.scannerBatchSize = conf.getWebsocket().getScannerBatchSize();
@@ -45,7 +48,7 @@ public class Subscription {
         // send a websocket ping at half the timeout interval.
         int rate = conf.getWebsocket().getTimeout() / 2;
         this.ping = this.ctx.executor().scheduleAtFixedRate(() -> {
-            LOG.trace("Sending ping on channel {}", ctx.channel());
+            LOG.trace("[{}] Sending ping on channel {}", subscriptionId, ctx.channel());
             ctx.writeAndFlush(new PingWebSocketFrame());
             cleanupCompletedMetrics();
         }, rate, rate, TimeUnit.SECONDS);
@@ -53,9 +56,9 @@ public class Subscription {
 
     public void addMetric(String metric, Map<String, String> tags, long startTime, long endTime, long delay)
             throws TimelyException {
-        LOG.debug("Adding metric scanner for subscription {}", this.subscriptionId);
-        MetricScanner m = new MetricScanner(this, this.subscriptionId, this.sessionId, store, metric, tags, startTime,
-                endTime, delay, lag, ctx, scannerBatchSize, flushIntervalSeconds, scannerReadAhead,
+        LOG.debug("[{}] Adding metric scanner", this.subscriptionId);
+        MetricScanner m = new MetricScanner(this, this.subscriptionId, this.sessionId, store, cache, metric, tags,
+                startTime, endTime, delay, lag, ctx, scannerBatchSize, flushIntervalSeconds, scannerReadAhead,
                 subscriptionBatchSize);
         METRICS.put(metric, m);
         m.start();
@@ -80,20 +83,20 @@ public class Subscription {
             try {
                 m.join();
             } catch (InterruptedException e) {
-                LOG.error("Error closing MetricScanner", e);
+                LOG.error("[" + this.subscriptionId + "] Error closing MetricScanner", e);
             }
         }
     }
 
     public void close() {
-        LOG.info("Closing subscriptions for {}", this.subscriptionId);
+        LOG.info("[{}] Closing subscriptions", this.subscriptionId);
         this.ping.cancel(false);
         METRICS.forEach((k, v) -> {
             v.close();
             try {
                 v.join();
             } catch (InterruptedException e) {
-                LOG.error("Error closing MetricScanner", e);
+                LOG.error("[" + this.subscriptionId + "] Error closing MetricScanner", e);
             }
         });
         METRICS.clear();

server/src/test/java/timely/store/cache/TestDataStoreCacheIterator.java

@@ -43,7 +43,7 @@ public class TestDataStoreCacheIterator {
 
     private DataStoreCache getMetricMemoryStore1(long baseTime) throws TimelyException {
 
-        DataStoreCache mmStore = new DataStoreCache(configuration);
+        DataStoreCache mmStore = new DataStoreCache(null, configuration);
 
         Map<String, String> tags = new HashMap<>();
         tags.put("part", "webservice");
@@ -69,7 +69,7 @@ public class TestDataStoreCacheIterator {
 
     private DataStoreCache getMetricMemoryStore2(long baseTime) throws TimelyException {
 
-        DataStoreCache mmStore = new DataStoreCache(configuration);
+        DataStoreCache mmStore = new DataStoreCache(null, configuration);
 
         int increment = 10;
         Map<String, String> tags = new HashMap<>();

server/src/test/java/timely/store/cache/TestMemoryDataStore.java

@@ -27,7 +27,7 @@ public class TestMemoryDataStore {
 
     private DataStoreCache getMetricMemoryStore1(long baseTimestamp) throws TimelyException {
 
-        DataStoreCache mmStore = new DataStoreCache(configuration);
+        DataStoreCache mmStore = new DataStoreCache(null, configuration);
 
         long timestamp = baseTimestamp + 10000;
         Map<String, String> tags = new HashMap<>();
@@ -61,7 +61,7 @@ public class TestMemoryDataStore {
 
     private DataStoreCache getMetricMemoryStore2(long baseTimestamp) throws TimelyException {
 
-        DataStoreCache mmStore = new DataStoreCache(configuration);
+        DataStoreCache mmStore = new DataStoreCache(null, configuration);
 
         Map<String, String> tags = new HashMap<>();
         tags.put("part", "webservice");
@@ -154,7 +154,7 @@ public class TestMemoryDataStore {
 
     @Test
     public void TestExtentOfStorage() {
-        DataStoreCache mmStore = new DataStoreCache(configuration);
+        DataStoreCache mmStore = new DataStoreCache(null, configuration);
 
         HashMap<String, String> tags = new HashMap<>();
         tags.put("host", "localhost");

server/src/test/java/timely/test/integration/client/WebSocketClientIT.java

@@ -93,7 +93,7 @@ public class WebSocketClientIT extends OneWaySSLBase {
 
             @Override
             public void onOpen(Session session, EndpointConfig config) {
-                session.addMessageHandler(new MessageHandler.Whole<String>() {
+                session.addMessageHandler(String.class, new MessageHandler.Whole<String>() {
 
                     @Override
                     public void onMessage(String message) {