fix TopUtil concurrency problems and http reporter support dynamic fi…

…ltering configurations fix TopUtili bug (#54) Http reporter support dynamic filtering configurations #55
sofastack · May 26, 2019 · bf70e43 · bf70e43
1 parent e8cba52
commit bf70e43
Show file tree

Hide file tree

Showing 35 changed files with 543 additions and 56 deletions.
diff --git a/README.md b/README.md
@@ -32,10 +32,10 @@ SOFALookout 是一个利用多维度的 metrics 对目标系统进行度量和
 - 服务器端代码编译，进入server子目录，Maven 3.2.5+, JDK Version 1.8+；
 ## 样例工程
 
-样例工程演示了如何快速使用 SOFALookout，[详细可参考](https://github.com/alipay/sofa-lookout/wiki/useguide-samples)。
+样例工程演示了如何快速使用 SOFALookout，[详细可参考](https://github.com/sofastack/sofa-lookout/wiki/useguide-samples)。
 
 ## 贡献
 如何参与 SOFALookout [代码贡献](./CONTRIBUTING.md)
 
 ## 开源许可
-SOFALookout 基于 [Apache License 2.0](./LICENSE) 协议，SOFALookout 依赖了一些三方组件，它们的开源协议参见[依赖组件版权说明](https://github.com/alipay/sofa-lookout/wiki/NOTICE)。
+SOFALookout 基于 [Apache License 2.0](./LICENSE) 协议，SOFALookout 依赖了一些三方组件，它们的开源协议参见[依赖组件版权说明](https://github.com/sofastack/sofa-lookout/wiki/NOTICE)。
diff --git a/README_EN.md b/README_EN.md
@@ -5,7 +5,7 @@
 ![license](https://img.shields.io/badge/license-Apache--2.0-green.svg)
 ![maven](https://img.shields.io/github/release/alipay/sofa-lookout.svg)
 
-Visit [WIKI](https://github.com/alipay/sofa-lookout/wiki) for the full documentation, examples, and guides.
+Visit [WIKI](https://github.com/sofastack/sofa-lookout/wiki) for the full documentation, examples, and guides.
 
 Lookout can help you to measure and monitor the status of the target system with its multi-dimensional metrics.
 Lookout's multi-dimensional metrics refer to the [Metrics 2.0](http://metrics20.org/) standard.

diff --git a/client/lookout-api/pom.xml b/client/lookout-api/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <groupId>com.alipay.sofa.lookout</groupId>
         <artifactId>lookout-client-parent</artifactId>
-        <version>1.5.3</version>
+        <version>1.5.4</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

diff --git a/client/lookout-api/src/main/java/com/alipay/lookout/common/top/AbstractTopMetric.java b/client/lookout-api/src/main/java/com/alipay/lookout/common/top/AbstractTopMetric.java
@@ -21,6 +21,8 @@
 
 import java.util.Comparator;
 import java.util.TreeSet;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * Created by kevin.luy@alipay.com on 2017/3/31.
@@ -44,6 +46,7 @@ public int compare(TopUtil.Entry<Id, Long> o1,
                                                                       };
     private final TopUtil.Order                       order;
     private long                                      lastRolledStamp = -1l;
+    private final Lock                                lock            = new ReentrantLock();
 
     AbstractTopMetric(Registry registry, Id id, int maxNumber, TopUtil.Order order) {
         this.maxNumber = maxNumber;
@@ -57,32 +60,35 @@ protected void pushAsync(Long value, Id timerId) {
         TopUtil.executor.execute(new Runnable() {
             @Override
             public void run() {
-                pushSafe(set, entry);
+                lock.lock();
+                try {
+                    pushSafe(set, entry);
+                } finally {
+                    lock.unlock();
+                }
             }
         });
     }
 
     private void pushSafe(TreeSet<TopUtil.Entry<Id, Long>> set, TopUtil.Entry<Id, Long> e) {
-        synchronized (set) {
-            if (!set.isEmpty()) {
-                boolean replaceable = false;
-                TopUtil.Entry<Id, Long> boundaryTarget = null;
-                if (order == TopUtil.Order.DESC) {
-                    boundaryTarget = set.first();
-                    replaceable = boundaryTarget.getValue() < e.getValue();
-                } else {
-                    boundaryTarget = set.last();
-                    replaceable = boundaryTarget.getValue() > e.getValue();
-                }
-                if (replaceable & set.size() >= maxNumber) {
-                    remove(set, boundaryTarget);
-                }
-                if (!replaceable & set.size() >= maxNumber) {
-                    return;//不add
-                }
+        if (!set.isEmpty()) {
+            boolean replaceable = false;
+            TopUtil.Entry<Id, Long> boundaryTarget = null;
+            if (order == TopUtil.Order.DESC) {
+                boundaryTarget = set.first();
+                replaceable = boundaryTarget.getValue() < e.getValue();
+            } else {
+                boundaryTarget = set.last();
+                replaceable = boundaryTarget.getValue() > e.getValue();
+            }
+            if (replaceable & set.size() >= maxNumber) {
+                remove(set, boundaryTarget);
+            }
+            if (!replaceable & set.size() >= maxNumber) {
+                return;//不add
             }
-            add(set, e);
         }
+        add(set, e);
     }
 
     private void remove(TreeSet<TopUtil.Entry<Id, Long>> set, TopUtil.Entry<Id, Long> boundaryTarget) {

diff --git a/client/lookout-api/src/main/java/com/alipay/lookout/common/top/TopUtil.java b/client/lookout-api/src/main/java/com/alipay/lookout/common/top/TopUtil.java
@@ -73,11 +73,11 @@ public static TopGauger topGauger(final Registry registry, final Id id, final in
         if (registry instanceof NoopRegistry || (id == NoopRegistry.INSTANCE.createId(null))) {
             return NoopTopGauger.INSTANCE;
         }
-        Id key = id.withTag(TOP_NUM_TAG_KEY, String.valueOf(maxNumber));
+        final Id key = id.withTag(TOP_NUM_TAG_KEY, String.valueOf(maxNumber));
         TopGauger topGauger = computeIfAbsent(cache, key, new Function<Object, TopGauger>() {
             @Override
             public TopGauger apply(Object obj) {
-                return new DefaultTopGauger(registry, id, maxNumber, order);
+                return new DefaultTopGauger(registry, key, maxNumber, order);
             }
         });
         return topGauger == null ? NoopTopGauger.INSTANCE : topGauger;
@@ -109,9 +109,7 @@ public enum Order {
      * @param <K>
      * @param <V>
      */
-    static class Entry<K, V>
-
-    {
+    static class Entry<K, V> {
         private final K key;
         private V       value;
 

diff --git a/client/lookout-client/pom.xml b/client/lookout-client/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>com.alipay.sofa.lookout</groupId>
         <artifactId>lookout-client-parent</artifactId>
-        <version>1.5.3</version>
+        <version>1.5.4</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

diff --git a/client/lookout-common/pom.xml b/client/lookout-common/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>com.alipay.sofa.lookout</groupId>
         <artifactId>lookout-client-parent</artifactId>
-        <version>1.5.3</version>
+        <version>1.5.4</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

diff --git a/client/lookout-common/src/test/java/com/alipay/lookout/common/utils/ClassUtilTest.java b/client/lookout-common/src/test/java/com/alipay/lookout/common/utils/ClassUtilTest.java
@@ -35,4 +35,9 @@ public void testNewInstance() {
             new Object[] { 5 });
         Assert.assertEquals("5", integer.toString());
     }
+
+    @Test(expected = IllegalStateException.class)
+    public void testNewInstanceException() {
+        ClassUtil.newInstance(ArrayList.class.getName() + "x", null, null);
+    }
 }
diff --git a/client/lookout-core/pom.xml b/client/lookout-core/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>com.alipay.sofa.lookout</groupId>
         <artifactId>lookout-client-parent</artifactId>
-        <version>1.5.3</version>
+        <version>1.5.4</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

diff --git a/client/lookout-ext-jvm/pom.xml b/client/lookout-ext-jvm/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>com.alipay.sofa.lookout</groupId>
         <artifactId>lookout-client-parent</artifactId>
-        <version>1.5.3</version>
+        <version>1.5.4</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

diff --git a/client/lookout-ext-os/pom.xml b/client/lookout-ext-os/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <groupId>com.alipay.sofa.lookout</groupId>
         <artifactId>lookout-client-parent</artifactId>
-        <version>1.5.3</version>
+        <version>1.5.4</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

diff --git a/client/lookout-reg-dropwizard/pom.xml b/client/lookout-reg-dropwizard/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>com.alipay.sofa.lookout</groupId>
         <artifactId>lookout-client-parent</artifactId>
-        <version>1.5.3</version>
+        <version>1.5.4</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

diff --git a/client/lookout-reg-prometheus/pom.xml b/client/lookout-reg-prometheus/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>com.alipay.sofa.lookout</groupId>
         <artifactId>lookout-client-parent</artifactId>
-        <version>1.5.3</version>
+        <version>1.5.4</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

diff --git a/client/lookout-reg-server/pom.xml b/client/lookout-reg-server/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>com.alipay.sofa.lookout</groupId>
         <artifactId>lookout-client-parent</artifactId>
-        <version>1.5.3</version>
+        <version>1.5.4</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

diff --git a/client/lookout-reg-server/src/main/java/com/alipay/lookout/remote/report/HttpObserver.java b/client/lookout-reg-server/src/main/java/com/alipay/lookout/remote/report/HttpObserver.java
@@ -158,14 +158,27 @@ public void update(List<LookoutMeasurement> measures, Map<String, String> metada
             logger.debug("## no gateway address found, drop metrics:\n{}\n", measures.toString());
             return;
         }
-        logger.debug(">> send metrics to {}:\n{}\n", address, measures.toString());
-        List<List<LookoutMeasurement>> batches = getBatches(measures,
+
+        //Filter measures by config
+        List<LookoutMeasurement> filteredMeasures = filter(measures);
+        if (filteredMeasures.isEmpty()) {
+            return;
+        }
+        logger.debug(">> send metrics to {}:\n{}\n", address, filteredMeasures);
+        List<List<LookoutMeasurement>> batches = getBatches(filteredMeasures,
             lookoutConfig.getInt(LOOKOUT_REPORT_BATCH_SIZE, DEFAULT_REPORT_BATCH_SIZE));
         for (List<LookoutMeasurement> batch : batches) {
             reportBatch(batch, metadata, address);
         }
     }
 
+    private List<LookoutMeasurement> filter(List<LookoutMeasurement> measures) {
+        if (httpRequestProcessor instanceof ReportDecider) {
+            return ((ReportDecider) httpRequestProcessor).filter(measures);
+        }
+        return measures;
+    }
+
     /**
      * Get a list of all measurements and break them into batches.
      *
@@ -255,17 +268,16 @@ void sendHttpDataSilently(HttpRequest httpRequest, Map<String, String> metadata)
         } catch (Throwable e) {
             if (e instanceof UnknownHostException || e instanceof ConnectException) {
                 addressService.clearAddressCache();
-                logger.info(">>WARNING: lookout agent:{} err?cause:{}", httpRequest.toString(),
-                    e.getMessage());
+                logger
+                    .info(">>WARNING: lookout agent:{} err?cause:{}", httpRequest, e.getMessage());
             } else if (e instanceof SocketTimeoutException) {
                 registry().counter(
                     registry().createId("lookout.client.report.fail.count").withTag("err",
                         "socket_timeout")).inc();
             } else {
                 registry().counter(registry().createId("lookout.client.report.fail.count")).inc();
             }
-            logger.info(">>WARNING: lookout agent:{} fail!cause:{}", httpRequest.toString(),
-                e.getMessage());
+            logger.info(">>WARNING: lookout agent:{} fail!cause:{}", httpRequest, e.getMessage());
         }
     }
 

diff --git a/...reg-server/src/main/java/com/alipay/lookout/remote/report/poller/MetricsHttpExporter.java b/...reg-server/src/main/java/com/alipay/lookout/remote/report/poller/MetricsHttpExporter.java
@@ -18,6 +18,7 @@
 
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
+import com.alipay.lookout.common.log.LookoutLoggerFactory;
 import com.alipay.lookout.common.utils.CommonUtil;
 import com.google.common.collect.Sets;
 import com.sun.net.httpserver.HttpExchange;
@@ -26,6 +27,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.http.NameValuePair;
 import org.apache.http.client.utils.URIBuilder;
+import org.slf4j.Logger;
 
 import java.io.IOException;
 import java.io.OutputStream;
@@ -48,6 +50,8 @@
  * @since 2018/7/17
  */
 public class MetricsHttpExporter {
+    static final Logger            logger          = LookoutLoggerFactory
+                                                       .getLogger(MetricsHttpExporter.class);
     private static final Charset   UTF8            = Charset.forName("UTF-8");
     private static final int       DEFAULT_BACKLOG = 2;
     private final PollerController controller;
@@ -146,6 +150,9 @@ public void handle(HttpExchange exchange)
 
                                                        // if (oldRate != newStep || oldSlotCount != newSlotCount) {
                                                        // }
+                                                   } catch (Throwable e) {
+                                                       logger.warn("pull metrics failed."
+                                                                   + e.getMessage());
                                                    } finally {
                                                        exchange.close();
                                                    }