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/conf/timely-balancer.yml

@@ -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/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

@@ -72,7 +72,7 @@ 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 DistributedAtomicValue nonCachedMetricsIP;
@@ -198,53 +198,44 @@ public class BalancedMetricResolver implements MetricResolver {
     }
 
     private void readNonCachedMetricsIP() {
-        try {
-            nonCachedMetricsLocalLock.readLock().lock();
-            Set<String> currentNonCachedMetricsDistributed;
-            try {
-                nonCachedMetricsIPRWLock.readLock().acquire();
-                byte[] currentNonCachedMetricsDistributedBytes = nonCachedMetricsIP.get().postValue();
-                if (currentNonCachedMetricsDistributedBytes == null) {
-                    currentNonCachedMetricsDistributed = new TreeSet<>();
-                } else {
-                    try {
-                        currentNonCachedMetricsDistributed = SerializationUtils
-                                .deserialize(currentNonCachedMetricsDistributedBytes);
-                    } catch (Exception e) {
-                        LOG.error(e.getMessage());
-                        currentNonCachedMetricsDistributed = new TreeSet<>();
-                    }
-                }
-            } finally {
-                nonCachedMetricsIPRWLock.readLock().release();
-            }
 
-            if (nonCachedMetrics.containsAll(currentNonCachedMetricsDistributed)) {
-                LOG.info("local nonCachedMetrics already contains {}", currentNonCachedMetricsDistributed);
-            } else {
-                nonCachedMetricsLocalLock.readLock().unlock();
-                nonCachedMetricsLocalLock.writeLock().lock();
-                currentNonCachedMetricsDistributed.removeAll(nonCachedMetrics);
-                LOG.info("Adding {} to local nonCachedMetrics", currentNonCachedMetricsDistributed);
-                nonCachedMetrics.addAll(currentNonCachedMetricsDistributed);
-                balancerLock.writeLock().lock();
+        Set<String> currentNonCachedMetricsDistributed = new TreeSet<>();
+        try {
+            nonCachedMetricsIPRWLock.readLock().acquire();
+            byte[] currentNonCachedMetricsDistributedBytes = nonCachedMetricsIP.get().postValue();
+            if (currentNonCachedMetricsDistributedBytes != null) {
                 try {
-                    for (String m : currentNonCachedMetricsDistributed) {
-                        metricToHostMap.remove(m);
-                    }
-                } finally {
-                    balancerLock.writeLock().unlock();
+                    currentNonCachedMetricsDistributed = SerializationUtils
+                            .deserialize(currentNonCachedMetricsDistributedBytes);
+                } catch (Exception e) {
+                    LOG.error(e.getMessage());
+                    currentNonCachedMetricsDistributed = new TreeSet<>();
                 }
             }
         } catch (Exception e) {
             LOG.error(e.getMessage(), e);
         } finally {
-            if (nonCachedMetricsLocalLock.isWriteLockedByCurrentThread()) {
-                nonCachedMetricsLocalLock.writeLock().unlock();
-            } else {
-                nonCachedMetricsLocalLock.readLock().unlock();
+            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();
+        }
     }
 
     private void startLeaderLatch(CuratorFramework curatorFramework) {
@@ -649,14 +640,6 @@ public class BalancedMetricResolver implements MetricResolver {
                 } finally {
                     nonCachedMetricsLocalLock.writeLock().unlock();
                 }
-                balancerLock.writeLock().lock();
-                try {
-                    for (String m : nonCachedMetricsUpdate) {
-                        metricToHostMap.remove(m);
-                    }
-                } finally {
-                    balancerLock.writeLock().unlock();
-                }
 
                 try {
                     nonCachedMetricsIPRWLock.writeLock().acquire();

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/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

@@ -676,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

@@ -36,9 +36,11 @@ 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;
@@ -559,7 +561,6 @@ public class DataStoreCache {
                 return false;
             }
         }
-
         return true;
     }
 
@@ -632,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/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) {