Tag version 1.2.6

Bump netty-all from 4.1.35.Final to 4.1.42.Final

analytics/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.2.5</version>
+    <version>1.2.6</version>
   </parent>
   <artifactId>timely-analytics</artifactId>
   <!--

balancer/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.2.5</version>
+    <version>1.2.6</version>
   </parent>
   <artifactId>timely-balancer</artifactId>
   <dependencies>

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

@@ -57,6 +57,7 @@ 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.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.curator.RetryPolicy;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
@@ -360,7 +361,9 @@ public class Balancer {
 
         // start curator framework
         RetryPolicy retryPolicy = new RetryForever(1000);
-        curatorFramework = CuratorFrameworkFactory.newClient(balancerConfig.getZooKeeper().getServers(), 30000, 1000,
+        int timeout = Long.valueOf(AccumuloConfiguration.getTimeInMillis(balancerConfig.getZooKeeper().getTimeout()))
+                .intValue();
+        curatorFramework = CuratorFrameworkFactory.newClient(balancerConfig.getZooKeeper().getServers(), timeout, 10000,
                 retryPolicy);
         curatorFramework.start();
         ensureZkPaths(curatorFramework, zkPaths);

balancer/src/main/java/timely/balancer/configuration/ZooKeeper.java

@@ -1,9 +1,14 @@
 package timely.balancer.configuration;
 
+import javax.validation.constraints.NotBlank;
+
 public class ZooKeeper {
 
     private String servers;
 
+    @NotBlank
+    private String timeout = "120s";
+
     public String getServers() {
         return servers;
     }
@@ -11,4 +16,12 @@ public class ZooKeeper {
     public void setServers(String servers) {
         this.servers = servers;
     }
+
+    public String getTimeout() {
+        return timeout;
+    }
+
+    public void setTimeout(String timeout) {
+        this.timeout = timeout;
+    }
 }

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

@@ -108,16 +108,7 @@ public class BalancedMetricResolver implements MetricResolver {
                 LOG.debug("Handling assignments event {}. assignmentsLastUpdatedInHdfs:{}", event.getType().toString(),
                         new Date(assignmentsLastUpdatedInHdfs.get().postValue()));
                 if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
-                    long lastLocalUpdate = assignmentsLastUpdatedLocal.get();
-                    long lastHdfsUpdate = assignmentsLastUpdatedInHdfs.get().postValue();
-                    if (lastHdfsUpdate > lastLocalUpdate) {
-                        LOG.debug("Reading assignments from hdfs lastHdfsUpdate ({}) > lastLocalUpdate ({})",
-                                new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
-                        readAssignmentsFromHdfs();
-                    } else {
-                        LOG.debug("Not reading assignments from hdfs lastHdfsUpdate ({}) <= lastLocalUpdate ({})",
-                                new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
-                    }
+                    readAssignmentsFromHdfs(true);
                 }
             }
         };
@@ -158,7 +149,7 @@ public class BalancedMetricResolver implements MetricResolver {
         }
         readNonCachedMetricsIP();
 
-        readAssignmentsFromHdfs();
+        readAssignmentsFromHdfs(false);
 
         timer.schedule(new TimerTask() {
 
@@ -183,10 +174,16 @@ public class BalancedMetricResolver implements MetricResolver {
                         long lastLocalUpdate = assignmentsLastUpdatedLocal.get();
                         long lastHdfsUpdate = assignmentsLastUpdatedInHdfs.get().postValue();
                         if (lastLocalUpdate > lastHdfsUpdate) {
-                            LOG.debug("Writing assignments to hdfs lastLocalUpdate ({}) > lastHdfsUpdate ({})",
+                            LOG.debug("Leader writing assignments to hdfs lastLocalUpdate ({}) > lastHdfsUpdate ({})",
+                                    new Date(lastLocalUpdate), new Date(lastHdfsUpdate));
+                            writeAssignmentsToHdfs();
+                        } else {
+                            LOG.trace(
+                                    "Leader not writing assignments to hdfs lastLocalUpdate ({}) <= lastHdfsUpdate ({})",
                                     new Date(lastLocalUpdate), new Date(lastHdfsUpdate));
-                            writeAssigmentsToHdfs();
                         }
+                    } else {
+                        readAssignmentsFromHdfs(true);
                     }
                 } catch (Exception e) {
                     LOG.error(e.getMessage(), e);
@@ -209,7 +206,7 @@ public class BalancedMetricResolver implements MetricResolver {
                     currentNonCachedMetricsDistributed = SerializationUtils
                             .deserialize(currentNonCachedMetricsDistributedBytes);
                 } catch (Exception e) {
-                    LOG.error(e.getMessage());
+                    LOG.error(e.getMessage(), e);
                     currentNonCachedMetricsDistributed = new TreeSet<>();
                 }
             }
@@ -250,7 +247,7 @@ public class BalancedMetricResolver implements MetricResolver {
                 public void isLeader() {
                     LOG.info("this balancer is the leader");
                     isLeader.set(true);
-                    writeAssigmentsToHdfs();
+                    writeAssignmentsToHdfs();
                 }
 
                 @Override
@@ -343,14 +340,16 @@ public class BalancedMetricResolver implements MetricResolver {
                                 assignMetric(s, h);
                                 LOG.debug("Assigned server removed.  Assigning {} to server {}:{}", s, h.getTcpPort());
                             }
-                            writeAssigmentsToHdfs();
                         }
 
                     } finally {
                         balancerLock.writeLock().unlock();
                     }
-                    if (isLeader.get() && rebalanceNeeded) {
-                        balance();
+                    if (isLeader.get()) {
+                        if (rebalanceNeeded) {
+                            balance();
+                        }
+                        writeAssignmentsToHdfs();
                     }
                 }
 
@@ -694,7 +693,7 @@ public class BalancedMetricResolver implements MetricResolver {
             try {
                 lock.writeLock().release();
             } catch (Exception e3) {
-                LOG.error(e3.getMessage());
+                LOG.error(e3.getMessage(), e3);
             }
         }
     }
@@ -879,7 +878,25 @@ public class BalancedMetricResolver implements MetricResolver {
         return tbh;
     }
 
-    private void readAssignmentsFromHdfs() {
+    private void readAssignmentsFromHdfs(boolean checkIfNecessary) {
+
+        try {
+            long lastLocalUpdate = assignmentsLastUpdatedLocal.get();
+            long lastHdfsUpdate = assignmentsLastUpdatedInHdfs.get().postValue();
+            if (checkIfNecessary) {
+                if (lastHdfsUpdate <= lastLocalUpdate) {
+                    LOG.debug("Not reading assignments from hdfs lastHdfsUpdate ({}) <= lastLocalUpdate ({})",
+                            new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
+                    return;
+                }
+            }
+            LOG.debug("Reading assignments from hdfs lastHdfsUpdate ({}) > lastLocalUpdate ({})",
+                    new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        }
+
+        // proceed with reading from HDFS
 
         Map<String, TimelyBalancedHost> assignedMetricToHostMap = new TreeMap<>();
         try {
@@ -919,8 +936,14 @@ public class BalancedMetricResolver implements MetricResolver {
 
             metricToHostMap.clear();
             for (Map.Entry<String, TimelyBalancedHost> e : assignedMetricToHostMap.entrySet()) {
-                if (shouldCache(e.getKey())) {
-                    metricToHostMap.put(e.getKey(), e.getValue());
+                if (StringUtils.isNotBlank(e.getKey()) && e.getValue() != null) {
+                    if (shouldCache(e.getKey())) {
+                        metricToHostMap.put(e.getKey(), e.getValue());
+                    }
+                } else {
+                    Exception e1 = new IllegalStateException(
+                            "Bad assignment metric:" + e.getKey() + " host:" + e.getValue());
+                    LOG.warn(e1.getMessage(), e1);
                 }
             }
 
@@ -951,7 +974,7 @@ public class BalancedMetricResolver implements MetricResolver {
         }
     }
 
-    private void writeAssigmentsToHdfs() {
+    private void writeAssignmentsToHdfs() {
 
         CsvWriter writer = null;
         try {
@@ -1016,12 +1039,17 @@ public class BalancedMetricResolver implements MetricResolver {
 
     private void assignMetric(String metric, TimelyBalancedHost tbh) {
         if (isLeader.get()) {
-            balancerLock.writeLock().lock();
-            try {
-                metricToHostMap.put(metric, tbh);
-                assignmentsLastUpdatedLocal.set(System.currentTimeMillis());
-            } finally {
-                balancerLock.writeLock().unlock();
+            if (StringUtils.isNotBlank(metric) && tbh != null) {
+                balancerLock.writeLock().lock();
+                try {
+                    metricToHostMap.put(metric, tbh);
+                    assignmentsLastUpdatedLocal.set(System.currentTimeMillis());
+                } finally {
+                    balancerLock.writeLock().unlock();
+                }
+            } else {
+                Exception e = new IllegalStateException("Bad assignment metric:" + metric + " host:" + tbh);
+                LOG.warn(e.getMessage(), e);
             }
         }
     }

client/pom.xml

@@ -4,13 +4,13 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.2.5</version>
+    <version>1.2.6</version>
   </parent>
   <artifactId>timely-client</artifactId>
   <properties>
     <!-- PEP 440 compliant version gets substituted into src/main/python/template/setup.py -->
     <!-- use .dev0 instead of -SNAPSHOT -->
-    <python.client.version>1.2.5</python.client.version>
+    <python.client.version>1.2.6</python.client.version>
   </properties>
   <dependencies>
     <dependency>

collectd-nsq-plugin/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.2.5</version>
+    <version>1.2.6</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.2.5</version>
+    <version>1.2.6</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.2.5</version>
+    <version>1.2.6</version>
   </parent>
   <artifactId>collectd-timely-plugin</artifactId>
   <dependencies>

common/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.2.5</version>
+    <version>1.2.6</version>
   </parent>
   <artifactId>timely-common</artifactId>
   <dependencies>

grafana-auth/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.2.5</version>
+    <version>1.2.6</version>
   </parent>
   <artifactId>grafana-auth</artifactId>
   <dependencies>

grafana/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.2.5</version>
+    <version>1.2.6</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": "1.2.5",
-    "updated": "2019-08-22"
+    "version": "1.2.6",
+    "updated": "2019-10-30"
   },
 
   "includes": [

grafana/timely-app/package.json

@@ -1,7 +1,7 @@
 {
   "name": "timely-app",
   "private": true,
-  "version": "1.2.5",
+  "version": "1.2.6",
   "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": "1.2.5",
-    "updated": "2019-08-22"
+    "version": "1.2.6",
+    "updated": "2019-10-30"
   },
 
   "includes": [

pom.xml

@@ -3,7 +3,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>timely</groupId>
   <artifactId>timely</artifactId>
-  <version>1.2.5</version>
+  <version>1.2.6</version>
   <packaging>pom</packaging>
   <name>Timely</name>
   <description>A secure and scalable time series database based on Apache Accumulo</description>
@@ -59,8 +59,8 @@
     <version.hadoop>2.7.7</version.hadoop>
     <version.httpcomponents>4.5.6</version.httpcomponents>
     <version.impsort>1.3.0</version.impsort>
-    <version.jackson>2.9.9</version.jackson>
-    <version.jackson.databind>2.9.9.3</version.jackson.databind>
+    <version.jackson>2.9.10</version.jackson>
+    <version.jackson.databind>2.9.10</version.jackson.databind>
     <version.jacoco>0.8.4</version.jacoco>
     <version.javacsv>2.0</version.javacsv>
     <version.javassist>3.24.1-GA</version.javassist>
@@ -77,7 +77,7 @@
     <version.maven-rpm>2.2.0</version.maven-rpm>
     <version.maven-shade>3.2.1</version.maven-shade>
     <version.maven-surefire>3.0.0-M3</version.maven-surefire>
-    <version.netty>4.1.35.Final</version.netty>
+    <version.netty>4.1.42.Final</version.netty>
     <version.netty-tcnative>2.0.25.Final</version.netty-tcnative>
     <version.os-maven>1.5.0.Final</version.os-maven>
     <version.powermock>1.6.4</version.powermock>

server/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.2.5</version>
+    <version>1.2.6</version>
   </parent>
   <artifactId>timely-server</artifactId>
   <dependencies>

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

@@ -60,6 +60,7 @@ 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.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.curator.RetryPolicy;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
@@ -417,7 +418,9 @@ public class Server {
     public void run() throws Exception {
 
         RetryPolicy retryPolicy = new RetryForever(1000);
-        curatorFramework = CuratorFrameworkFactory.newClient(config.getAccumulo().getZookeepers(), 30000, 10000,
+        int timeout = Long.valueOf(AccumuloConfiguration.getTimeInMillis(config.getAccumulo().getZookeeperTimeout()))
+                .intValue();
+        curatorFramework = CuratorFrameworkFactory.newClient(config.getAccumulo().getZookeepers(), timeout, 10000,
                 retryPolicy);
         curatorFramework.start();
         ensureZkPaths(curatorFramework, zkPaths);

server/src/main/java/timely/configuration/Accumulo.java

@@ -12,6 +12,8 @@ public class Accumulo {
     @NotBlank
     private String instanceName;
     @NotBlank
+    private String zookeeperTimeout = "120s";
+    @NotBlank
     private String username;
     @NotBlank
     private String password;
@@ -46,6 +48,14 @@ public class Accumulo {
         this.instanceName = instanceName;
     }
 
+    public String getZookeeperTimeout() {
+        return zookeeperTimeout;
+    }
+
+    public void setZookeeperTimeout(String zookeeperTimeout) {
+        this.zookeeperTimeout = zookeeperTimeout;
+    }
+
     public String getUsername() {
         return username;
     }

server/src/main/java/timely/configuration/Cache.java

@@ -13,6 +13,7 @@ public class Cache {
     private List<String> nonCachedMetrics = new ArrayList<>();
     private long maxUniqueTagSets = 50000;
     private long flushInterval = 5000;
+    private long staleCacheExpiration = 1800000;
 
     public HashMap<String, Integer> getMetricAgeOffHours() {
         return metricAgeOffHours;
@@ -57,4 +58,12 @@ public class Cache {
     public void setFlushInterval(long flushInterval) {
         this.flushInterval = flushInterval;
     }
+
+    public long getStaleCacheExpiration() {
+        return staleCacheExpiration;
+    }
+
+    public void setStaleCacheExpiration(long staleCacheExpiration) {
+        this.staleCacheExpiration = staleCacheExpiration;
+    }
 }

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

@@ -147,6 +147,7 @@ public class DataStoreImpl implements DataStore {
             Accumulo accumuloConf = conf.getAccumulo();
             properties.put("instance.name", accumuloConf.getInstanceName());
             properties.put("instance.zookeeper.host", accumuloConf.getZookeepers());
+            properties.put("instance.zookeeper.timeout", accumuloConf.getZookeeperTimeout());
             final ClientConfiguration aconf = ClientConfiguration.fromMap(properties);
             final Instance instance = new ZooKeeperInstance(aconf);
             connector = instance.getConnector(accumuloConf.getUsername(),

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

@@ -71,6 +71,7 @@ public class DataStoreCache {
     private Set<String> nonCachedMetrics = Collections.synchronizedSet(new HashSet<>());
     private DistributedAtomicValue nonCachedMetricsIP = null;
     private InterProcessReadWriteLock nonCachedMetricsIPRWLock;
+    private long staleCacheExpiration;
     private long maxUniqueTagSets;
     private final Security security;
     private Map<String, Long> minimumAgeOff;
@@ -89,6 +90,7 @@ public class DataStoreCache {
         LOG.info("Reading initial values from nonCachedMetricsIP");
         readNonCachedMetricsIP();
         maxUniqueTagSets = conf.getCache().getMaxUniqueTagSets();
+        staleCacheExpiration = conf.getCache().getStaleCacheExpiration();
         security = conf.getSecurity();
         Map<String, Integer> cacheAgeOff = conf.getCache().getMetricAgeOffHours();
         Map<String, Integer> accumuloAgeOff = conf.getMetricAgeOffDays();
@@ -118,6 +120,18 @@ public class DataStoreCache {
             }
         }, conf.getCache().getFlushInterval(), conf.getCache().getFlushInterval());
 
+        maintenanceTimer.schedule(new TimerTask() {
+
+            @Override
+            public void run() {
+                try {
+                    removeStaleMetrics();
+                } catch (Exception e) {
+                    LOG.error(e.getMessage(), e);
+                }
+            }
+        }, (300 * 1000), (300 * 1000));
+
         maintenanceTimer.schedule(new TimerTask() {
 
             @Override
@@ -436,6 +450,68 @@ public class DataStoreCache {
         }
     }
 
+    private void removeStaleMetrics() {
+
+        Set<String> metricsToRemove = new HashSet<>();
+        long stamp = gorillaMapLock.readLock();
+        try {
+            long now = System.currentTimeMillis();
+            for (Map.Entry<String, Map<TaggedMetric, GorillaStore>> entry : gorillaMap.entrySet()) {
+                String metricName = entry.getKey();
+                long totalMsecToMetric = 0;
+                long numStoresWithMultipleEntries = 0;
+                int tagVariations = entry.getValue().values().size();
+                for (GorillaStore store : entry.getValue().values()) {
+                    long numEntries = store.getNumEntries();
+                    long oldestTimestamp = store.getOldestTimestamp();
+                    long newestTimestamp = store.getNewestTimestamp();
+                    if (numEntries > 1) {
+                        long msecToMetric = (newestTimestamp - oldestTimestamp) / (numEntries - 1);
+                        totalMsecToMetric += msecToMetric;
+                        numStoresWithMultipleEntries++;
+                    }
+                }
+                if (numStoresWithMultipleEntries > 0) {
+                    long avgMsecToMetric = totalMsecToMetric / numStoresWithMultipleEntries;
+                    long ageOfNewest = now - getNewestTimestamp(metricName, stamp);
+                    // do not delete slow-arriving metrics based on this criteria
+                    // only consider metrics that should have arrived in staleCacheExpiration * 0.5
+                    LOG.trace("metric:{} tagVariations:{} avgMsecToMetric:{}", metricName, tagVariations,
+                            avgMsecToMetric);
+                    if (avgMsecToMetric < (staleCacheExpiration * 0.5)) {
+                        if (ageOfNewest > staleCacheExpiration) {
+                            LOG.info(
+                                    "Removing metric:{} tagVariations:{} avgMsecToMetric:{} ageOfNewestEntry:{} > staleCacheExpiration:{}",
+                                    metricName, tagVariations, avgMsecToMetric, ageOfNewest, staleCacheExpiration);
+                            metricsToRemove.add(metricName);
+                        }
+                    } else {
+                        LOG.trace(
+                                "Skipping staleness evaluation of metric:{} tagVariations:{} avgMsecToMetric:{} < (0.5 * staleCacheExpiration):{}",
+                                metricName, tagVariations, avgMsecToMetric, (staleCacheExpiration / 2));
+                    }
+                }
+            }
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        } finally {
+            gorillaMapLock.unlockRead(stamp);
+        }
+
+        if (metricsToRemove.size() > 0) {
+            stamp = gorillaMapLock.writeLock();
+            try {
+                for (String metricName : metricsToRemove) {
+                    gorillaMap.remove(metricName);
+                }
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            } finally {
+                gorillaMapLock.unlockWrite(stamp);
+            }
+        }
+    }
+
     public long getAgeOffForMetric(String metricName) {
         if (this.minimumAgeOff.containsKey(metricName)) {
             return this.minimumAgeOff.get(metricName);
@@ -511,6 +587,7 @@ public class DataStoreCache {
         try {
             metricMap = gorillaMap.get(metric);
             if (metricMap == null) {
+                LOG.info("Creating new cache for metric:{}", metric);
                 metricMap = new HashMap<>();
                 needWrite = true;
             }
@@ -768,9 +845,13 @@ public class DataStoreCache {
     }
 
     public long getNewestTimestamp(String metric) {
+        return getNewestTimestamp(metric, 0);
+    }
+
+    public long getNewestTimestamp(String metric, long lockStamp) {
 
         long newest = 0;
-        long stamp = gorillaMapLock.readLock();
+        long stamp = lockStamp == 0 ? gorillaMapLock.readLock() : lockStamp;
         try {
             Map<TaggedMetric, GorillaStore> gorillaStoreMap = gorillaMap.get(metric);
             for (Map.Entry<TaggedMetric, GorillaStore> entry : gorillaStoreMap.entrySet()) {
@@ -779,7 +860,9 @@ public class DataStoreCache {
                 }
             }
         } finally {
-            gorillaMapLock.unlockRead(stamp);
+            if (lockStamp == 0) {
+                gorillaMapLock.unlockRead(stamp);
+            }
         }
         return newest;
     }