Tag version 1.2.7

Fixed #194, corrected datasource.js to look at wildcard tag value

analytics/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>1.2.5</version>
+    <version>1.2.7</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.7</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.7</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.7</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.7</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.7</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.7</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.7</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.7</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.7</version>
   </parent>
   <artifactId>timely-app</artifactId>
   <packaging>pom</packaging>

grafana/timely-app/dist/datasource/datasource.js

@@ -509,7 +509,7 @@ System.register(['lodash', 'angular', '../../../app/core/utils/datemath'], funct
                 } else {
                   return target.metric === metricData.metric && _.every(target.tags, function (tagV, tagK) {
                     interpolatedTagValue = this.templateSrv.replace(tagV, options.scopedVars, 'pipe');
-                    return metricData.tags[tagK] === interpolatedTagValue || interpolatedTagValue === "*";
+                    return metricData.tags[tagK] === interpolatedTagValue || interpolatedTagValue === ".*";
                   }.bind(this));
                 }
               }.bind(this));

grafana/timely-app/dist/datasource/datasource.js.map

@@ -1 +1 @@
-{"version":3,"sources":["../../src/datasource/datasource.js"],"names":["_","angular","dateMath","TimelyDatasource","instanceSettings","$q","backendSrv","templateSrv","contextSrv","type","name","jsonData","httpsPort","timelyHost","wsPort","basicAuths","url","_makeHttpsUrl","withCredentials","basicAuth","supportMetrics","tagKeys","aggregatorsPromise","filterTypesPromise","options","start","convertToTSDBTime","rangeRaw","from","end","to","qs","each","targets","target","metric","push","convertTargetToQuery","bind","queries","compact","isEmpty","d","defer","resolve","data","promise","groupByTags","query","filters","length","val","tagk","tags","key","performTimeSeriesQuery","then","response","metricToTargetMapping","mapMetricsToTargets","result","map","metricData","index","_saveTagKeys","transformMetricData","datasourceRequest","method","status","message","title","eventList","aggregator","annotation","results","annotationObject","annotations","isGlobal","globalAnnotations","event","description","time","Math","floor","startTime","text","notes","when","interpolated","replace","err","reject","responseTransform","value","metrics_regex","tag_names_regex","tag_values_regex","tag_names_suggest_regex","tag_values_suggest_regex","metrics_query","match","_performSuggestQuery","tag_names_query","_performMetricKeyLookup","tag_values_query","_performMetricKeyValueLookup","tag_names_suggest_query","tag_values_suggest_query","msResolution","responsePromises","reqBody","showQuery","_addCredentialOptions","headers","Promise","all","queryResponses","combinedResponse","x","concat","Object","keys","aggregateTags","tag","_get","q","max","keysArray","split","trim","keysQuery","splice","join","m","limit","tagvs","r","indexOf","tagks","tagv","relativeUrl","params","isArray","sort","md","metricLabel","createMetricLabel","dps","v","k","datapoints","alias","scopedVars","clone","label","tagData","toPairs","has","hide","shouldComputeRate","rate","rateOptions","counter","isCounter","counterMax","parseInt","counterResetValue","resetValue","disableDownsampling","interval","downsampleInterval","parseFloat","downsample","downsampleAggregator","downsampleFillPolicy","copy","filter_key","filter","tag_key","metrics","interpolatedTagValue","findIndex","every","tagV","tagK","date","roundUp","parse","valueOf"],"mappings":";;;;;;;;;;;;;;;AAAOA,O;;AACAC,a;;AACKC,c;;;;;;;;;;;;;;;;;;;;;kCAECC,gB;;AAEX;AACA,kCAAYC,gBAAZ,EAA8BC,EAA9B,EAAkCC,UAAlC,EAA8CC,WAA9C,EAA2DC,UAA3D,EAAuE;AAAA;;AACrE,eAAKC,IAAL,GAAYL,iBAAiBK,IAA7B;AACA,eAAKC,IAAL,GAAYN,iBAAiBM,IAA7B;;AAEA,eAAKC,QAAL,GAAgBP,iBAAiBO,QAAjB,IAA6B,EAA7C;AACA,eAAKC,SAAL,GAAiBR,iBAAiBO,QAAjB,CAA0BC,SAA3C;AACA,eAAKC,UAAL,GAAkBT,iBAAiBO,QAAjB,CAA0BE,UAA5C;AACA,eAAKC,MAAL,GAAcV,iBAAiBO,QAAjB,CAA0BG,MAAxC;AACA,cAAI,CAACV,iBAAiBO,QAAjB,CAA0BI,UAA/B,EAA0C;AACxCX,6BAAiBO,QAAjB,CAA0BI,UAA1B,GAAuC,KAAvC;AACD;;AAED,eAAKC,GAAL,GAAW,KAAKC,aAAL,EAAX;AACA,eAAKC,eAAL,GAAuB,IAAvB;AACA,eAAKC,SAAL,GAAiB,KAAjB;;AAEA,eAAKC,cAAL,GAAsB,IAAtB;AACA,eAAKC,OAAL,GAAe,EAAf;;AAEA,eAAKhB,EAAL,GAAUA,EAAV;AACA,eAAKC,UAAL,GAAkBA,UAAlB;AACA,eAAKC,WAAL,GAAmBA,WAAnB;AACA,eAAKC,UAAL,GAAkBA,UAAlB;;AAEA,eAAKc,kBAAL,GAA0B,IAA1B;AACA,eAAKC,kBAAL,GAA0B,IAA1B;AAED;;AAED;;;;;gCACMC,O,EAAQ;AACZ,gBAAIC,QAAQ,KAAKC,iBAAL,CAAuBF,QAAQG,QAAR,CAAiBC,IAAxC,EAA8C,KAA9C,CAAZ;AACA,gBAAIC,MAAM,KAAKH,iBAAL,CAAuBF,QAAQG,QAAR,CAAiBG,EAAxC,EAA4C,IAA5C,CAAV;AACA,gBAAIC,KAAK,EAAT;;AAEA/B,cAAEgC,IAAF,CAAOR,QAAQS,OAAf,EAAwB,UAASC,MAAT,EAAiB;AACvC,kBAAI,CAACA,OAAOC,MAAZ,EAAoB;AAAE;AAAS;AAC/BJ,iBAAGK,IAAH,CAAQ,KAAKC,oBAAL,CAA0BH,MAA1B,EAAkCV,OAAlC,CAAR;AACD,aAHuB,CAGtBc,IAHsB,CAGjB,IAHiB,CAAxB;;AAKA,gBAAIC,UAAUvC,EAAEwC,OAAF,CAAUT,EAAV,CAAd;;AAEA;AACA,gBAAI/B,EAAEyC,OAAF,CAAUF,OAAV,CAAJ,EAAwB;AACtB,kBAAIG,IAAI,KAAKrC,EAAL,CAAQsC,KAAR,EAAR;AACAD,gBAAEE,OAAF,CAAU,EAAEC,MAAM,EAAR,EAAV;AACA,qBAAOH,EAAEI,OAAT;AACD;;AAED,gBAAIC,cAAc,EAAlB;AACA/C,cAAEgC,IAAF,CAAOO,OAAP,EAAgB,UAASS,KAAT,EAAgB;AAC9B,kBAAIA,MAAMC,OAAN,IAAiBD,MAAMC,OAAN,CAAcC,MAAd,GAAuB,CAA5C,EAA+C;AAC7ClD,kBAAEgC,IAAF,CAAOgB,MAAMC,OAAb,EAAsB,UAASE,GAAT,EAAc;AAClCJ,8BAAYI,IAAIC,IAAhB,IAAwB,IAAxB;AACD,iBAFD;AAGD,eAJD,MAIO;AACLpD,kBAAEgC,IAAF,CAAOgB,MAAMK,IAAb,EAAmB,UAASF,GAAT,EAAcG,GAAd,EAAmB;AACpCP,8BAAYO,GAAZ,IAAmB,IAAnB;AACD,iBAFD;AAGD;AACF,aAVD;;AAYA,mBAAO,KAAKC,sBAAL,CAA4BhB,OAA5B,EAAqCd,KAArC,EAA4CI,GAA5C,EAAiD2B,IAAjD,CAAsD,UAASC,QAAT,EAAmB;AAC9E,kBAAIC,wBAAwB,KAAKC,mBAAL,CAAyBF,SAASZ,IAAlC,EAAwCrB,OAAxC,CAA5B;AACA,kBAAIoC,SAAS5D,EAAE6D,GAAF,CAAMJ,SAASZ,IAAf,EAAqB,UAASiB,UAAT,EAAqBC,KAArB,EAA4B;AAC5DA,wBAAQL,sBAAsBK,KAAtB,CAAR;AACA,oBAAIA,UAAU,CAAC,CAAf,EAAkB;AAChBA,0BAAQ,CAAR;AACD;AACD,qBAAKC,YAAL,CAAkBF,UAAlB;;AAEA,uBAAO,KAAKG,mBAAL,CAAyBH,UAAzB,EAAqCf,WAArC,EAAkDvB,QAAQS,OAAR,CAAgB8B,KAAhB,CAAlD,EAA0EvC,OAA1E,CAAP;AACD,eARiC,CAQhCc,IARgC,CAQ3B,IAR2B,CAArB,CAAb;AASA,qBAAO,EAAEO,MAAMe,MAAR,EAAP;AACD,aAZ4D,CAY3DtB,IAZ2D,CAYtD,IAZsD,CAAtD,CAAP;AAaD;;;2CAGe;AACd,mBAAO,KAAKhC,UAAL,CAAgB4D,iBAAhB,CAAkC;AACvClD,mBAAK,KAAKA,GAAL,GAAW,UADuB;AAEvCmD,sBAAQ;AAF+B,aAAlC,EAGJX,IAHI,CAGC,oBAAY;AAClB,kBAAIC,SAASW,MAAT,KAAoB,GLine truncated
+{"version":3,"sources":["../../src/datasource/datasource.js"],"names":["_","angular","dateMath","TimelyDatasource","instanceSettings","$q","backendSrv","templateSrv","contextSrv","type","name","jsonData","httpsPort","timelyHost","wsPort","basicAuths","url","_makeHttpsUrl","withCredentials","basicAuth","supportMetrics","tagKeys","aggregatorsPromise","filterTypesPromise","options","start","convertToTSDBTime","rangeRaw","from","end","to","qs","each","targets","target","metric","push","convertTargetToQuery","bind","queries","compact","isEmpty","d","defer","resolve","data","promise","groupByTags","query","filters","length","val","tagk","tags","key","performTimeSeriesQuery","then","response","metricToTargetMapping","mapMetricsToTargets","result","map","metricData","index","_saveTagKeys","transformMetricData","datasourceRequest","method","status","message","title","eventList","aggregator","annotation","results","annotationObject","annotations","isGlobal","globalAnnotations","event","description","time","Math","floor","startTime","text","notes","when","interpolated","replace","err","reject","responseTransform","value","metrics_regex","tag_names_regex","tag_values_regex","tag_names_suggest_regex","tag_values_suggest_regex","metrics_query","match","_performSuggestQuery","tag_names_query","_performMetricKeyLookup","tag_values_query","_performMetricKeyValueLookup","tag_names_suggest_query","tag_values_suggest_query","msResolution","responsePromises","reqBody","showQuery","_addCredentialOptions","headers","Promise","all","queryResponses","combinedResponse","x","concat","Object","keys","aggregateTags","tag","_get","q","max","keysArray","split","trim","keysQuery","splice","join","m","limit","tagvs","r","indexOf","tagks","tagv","relativeUrl","params","isArray","sort","md","metricLabel","createMetricLabel","dps","v","k","datapoints","alias","scopedVars","clone","label","tagData","toPairs","has","hide","shouldComputeRate","rate","rateOptions","counter","isCounter","counterMax","parseInt","counterResetValue","resetValue","disableDownsampling","interval","downsampleInterval","parseFloat","downsample","downsampleAggregator","downsampleFillPolicy","copy","filter_key","filter","tag_key","metrics","interpolatedTagValue","findIndex","every","tagV","tagK","date","roundUp","parse","valueOf"],"mappings":";;;;;;;;;;;;;;;AAAOA,O;;AACAC,a;;AACKC,c;;;;;;;;;;;;;;;;;;;;;kCAECC,gB;;AAEX;AACA,kCAAYC,gBAAZ,EAA8BC,EAA9B,EAAkCC,UAAlC,EAA8CC,WAA9C,EAA2DC,UAA3D,EAAuE;AAAA;;AACrE,eAAKC,IAAL,GAAYL,iBAAiBK,IAA7B;AACA,eAAKC,IAAL,GAAYN,iBAAiBM,IAA7B;;AAEA,eAAKC,QAAL,GAAgBP,iBAAiBO,QAAjB,IAA6B,EAA7C;AACA,eAAKC,SAAL,GAAiBR,iBAAiBO,QAAjB,CAA0BC,SAA3C;AACA,eAAKC,UAAL,GAAkBT,iBAAiBO,QAAjB,CAA0BE,UAA5C;AACA,eAAKC,MAAL,GAAcV,iBAAiBO,QAAjB,CAA0BG,MAAxC;AACA,cAAI,CAACV,iBAAiBO,QAAjB,CAA0BI,UAA/B,EAA0C;AACxCX,6BAAiBO,QAAjB,CAA0BI,UAA1B,GAAuC,KAAvC;AACD;;AAED,eAAKC,GAAL,GAAW,KAAKC,aAAL,EAAX;AACA,eAAKC,eAAL,GAAuB,IAAvB;AACA,eAAKC,SAAL,GAAiB,KAAjB;;AAEA,eAAKC,cAAL,GAAsB,IAAtB;AACA,eAAKC,OAAL,GAAe,EAAf;;AAEA,eAAKhB,EAAL,GAAUA,EAAV;AACA,eAAKC,UAAL,GAAkBA,UAAlB;AACA,eAAKC,WAAL,GAAmBA,WAAnB;AACA,eAAKC,UAAL,GAAkBA,UAAlB;;AAEA,eAAKc,kBAAL,GAA0B,IAA1B;AACA,eAAKC,kBAAL,GAA0B,IAA1B;AAED;;AAED;;;;;gCACMC,O,EAAQ;AACZ,gBAAIC,QAAQ,KAAKC,iBAAL,CAAuBF,QAAQG,QAAR,CAAiBC,IAAxC,EAA8C,KAA9C,CAAZ;AACA,gBAAIC,MAAM,KAAKH,iBAAL,CAAuBF,QAAQG,QAAR,CAAiBG,EAAxC,EAA4C,IAA5C,CAAV;AACA,gBAAIC,KAAK,EAAT;;AAEA/B,cAAEgC,IAAF,CAAOR,QAAQS,OAAf,EAAwB,UAASC,MAAT,EAAiB;AACvC,kBAAI,CAACA,OAAOC,MAAZ,EAAoB;AAAE;AAAS;AAC/BJ,iBAAGK,IAAH,CAAQ,KAAKC,oBAAL,CAA0BH,MAA1B,EAAkCV,OAAlC,CAAR;AACD,aAHuB,CAGtBc,IAHsB,CAGjB,IAHiB,CAAxB;;AAKA,gBAAIC,UAAUvC,EAAEwC,OAAF,CAAUT,EAAV,CAAd;;AAEA;AACA,gBAAI/B,EAAEyC,OAAF,CAAUF,OAAV,CAAJ,EAAwB;AACtB,kBAAIG,IAAI,KAAKrC,EAAL,CAAQsC,KAAR,EAAR;AACAD,gBAAEE,OAAF,CAAU,EAAEC,MAAM,EAAR,EAAV;AACA,qBAAOH,EAAEI,OAAT;AACD;;AAED,gBAAIC,cAAc,EAAlB;AACA/C,cAAEgC,IAAF,CAAOO,OAAP,EAAgB,UAASS,KAAT,EAAgB;AAC9B,kBAAIA,MAAMC,OAAN,IAAiBD,MAAMC,OAAN,CAAcC,MAAd,GAAuB,CAA5C,EAA+C;AAC7ClD,kBAAEgC,IAAF,CAAOgB,MAAMC,OAAb,EAAsB,UAASE,GAAT,EAAc;AAClCJ,8BAAYI,IAAIC,IAAhB,IAAwB,IAAxB;AACD,iBAFD;AAGD,eAJD,MAIO;AACLpD,kBAAEgC,IAAF,CAAOgB,MAAMK,IAAb,EAAmB,UAASF,GAAT,EAAcG,GAAd,EAAmB;AACpCP,8BAAYO,GAAZ,IAAmB,IAAnB;AACD,iBAFD;AAGD;AACF,aAVD;;AAYA,mBAAO,KAAKC,sBAAL,CAA4BhB,OAA5B,EAAqCd,KAArC,EAA4CI,GAA5C,EAAiD2B,IAAjD,CAAsD,UAASC,QAAT,EAAmB;AAC9E,kBAAIC,wBAAwB,KAAKC,mBAAL,CAAyBF,SAASZ,IAAlC,EAAwCrB,OAAxC,CAA5B;AACA,kBAAIoC,SAAS5D,EAAE6D,GAAF,CAAMJ,SAASZ,IAAf,EAAqB,UAASiB,UAAT,EAAqBC,KAArB,EAA4B;AAC5DA,wBAAQL,sBAAsBK,KAAtB,CAAR;AACA,oBAAIA,UAAU,CAAC,CAAf,EAAkB;AAChBA,0BAAQ,CAAR;AACD;AACD,qBAAKC,YAAL,CAAkBF,UAAlB;;AAEA,uBAAO,KAAKG,mBAAL,CAAyBH,UAAzB,EAAqCf,WAArC,EAAkDvB,QAAQS,OAAR,CAAgB8B,KAAhB,CAAlD,EAA0EvC,OAA1E,CAAP;AACD,eARiC,CAQhCc,IARgC,CAQ3B,IAR2B,CAArB,CAAb;AASA,qBAAO,EAAEO,MAAMe,MAAR,EAAP;AACD,aAZ4D,CAY3DtB,IAZ2D,CAYtD,IAZsD,CAAtD,CAAP;AAaD;;;2CAGe;AACd,mBAAO,KAAKhC,UAAL,CAAgB4D,iBAAhB,CAAkC;AACvClD,mBAAK,KAAKA,GAAL,GAAW,UADuB;AAEvCmD,sBAAQ;AAF+B,aAAlC,EAGJX,IAHI,CAGC,oBAAY;AAClB,kBAAIC,SAASW,MAAT,KAAoB,GLine truncated

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.7",
+    "updated": "2020-02-21"
   },
 
   "includes": [

grafana/timely-app/package.json

@@ -1,7 +1,7 @@
 {
   "name": "timely-app",
   "private": true,
-  "version": "1.2.5",
+  "version": "1.2.7",
   "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.7",
+    "updated": "2020-02-21"
   },
 
   "includes": [

grafana/timely-app/src/datasource/datasource.js

@@ -446,7 +446,7 @@ export class TimelyDatasource {
         } else {
           return target.metric === metricData.metric && _.every(target.tags, function(tagV, tagK) {
                 interpolatedTagValue = this.templateSrv.replace(tagV, options.scopedVars, 'pipe');
-                return metricData.tags[tagK] === interpolatedTagValue || interpolatedTagValue === "*";
+                return metricData.tags[tagK] === interpolatedTagValue || interpolatedTagValue === ".*";
               }.bind(this));
         }
       }.bind(this));

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.7</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.10.0</version.jackson>
+    <version.jackson.databind>2.10.0</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.7</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;
     }