[type:feature] peak ewma lb

shenyu-loadbalancer/src/main/java/org/apache/shenyu/loadbalancer/spi/PeakEWMALoadBalancer.java

@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.loadbalancer.spi;
+
+import org.apache.shenyu.loadbalancer.entity.Upstream;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * <p>
+ * PeakEwmaLoadBalance is designed to converge quickly when encountering slow endpoints.
+ * It is quick to react to latency spikes recovering only cautiously.Peak EWMA takes
+ * history into account,so that slow behavior is penalized relative to the
+ * supplied `decayTime`.
+ * if there are multiple invokers and the same cost,then randomly called,which doesn't care
+ * about weight.
+ * <p>
+ * Inspiration drawn from:
+ * https://github.com/twitter/finagle/blob/1bc837c4feafc0096e43c0e98516a8e1c50c4421
+ * /finagle-core/src/main/scala/com/twitter/finagle/loadbalancer/PeakEwma.scala
+ *
+ * https://github.com/apache/dubbo-spi-extensions/blob/efd18a63468f817a7581fea44e9e2e3f35d9c9ba
+ * /dubbo-cluster-extensions/dubbo-cluster-loadbalance-peakewma/src/main/java/org/apache
+ * /dubbo/rpc/cluster/loadbalance/PeakEwmaLoadBalance.java#L46
+ */
+public class PeakEWMALoadBalancer extends AbstractLoadBalancer {
+
+    private static final double PENALTY = Long.MAX_VALUE >> 16;
+
+    private static final double ZERO_COST = 1E-6;
+
+    private static double decayTime = 10000;
+
+    @Override
+    protected Upstream doSelect(List<Upstream> upstreamList, String ip) {
+        double minResponse = Double.MAX_VALUE;
+
+        List<Integer> selectInvokerIndexList = new ArrayList<>(upstreamList.size());
+
+        for (int i = 0; i < upstreamList.size(); i++) {
+            Metric metric = new Metric(upstreamList.get(i));
+            double estimateResponse = metric.getCost();
+
+            if (estimateResponse < minResponse) {
+                selectInvokerIndexList.clear();
+                selectInvokerIndexList.add(i);
+                minResponse = estimateResponse;
+            } else if (estimateResponse == minResponse) {
+                selectInvokerIndexList.add(i);
+            }
+        }
+
+        return upstreamList.get(selectInvokerIndexList.get(ThreadLocalRandom.current().nextInt(selectInvokerIndexList.size())));
+    }
+
+    protected static class Metric {
+
+        /**
+         *  last timestamp in Millis we observed an runningTime
+         */
+        private volatile long lastUpdateTime;
+
+        /**
+         * ewma of rtt, sensitive to peaks.
+         */
+        private volatile double cost;
+
+        private Upstream upstream;
+
+        private long invokeOffset;
+
+        private long invokeElapsedOffset;
+
+        //lock for get and set cost
+        ReentrantLock ewmaLock = new ReentrantLock();
+
+        public Metric(Upstream upstream) {
+            this.upstream = upstream;
+            this.lastUpdateTime = System.currentTimeMillis();
+            this.cost = 0.0;
+            this.invokeOffset = 0;
+            this.invokeElapsedOffset = 0;
+        }
+
+        private void observe() {
+            double rtt = 0;
+            long succeed = this.upstream.getSucceeded().get() - this.invokeOffset;
+            if (succeed != 0) {
+                rtt = (this.upstream.getSucceededElapsed().get() * 1.0 - this.invokeElapsedOffset) / succeed;
+            }
+
+            final long currentTime = System.currentTimeMillis();
+            long td = Math.max(currentTime - lastUpdateTime, 0);
+            double w = Math.exp(-td / decayTime);
+            if (rtt > cost) {
+                cost = rtt;
+            } else {
+                cost = cost * w + rtt * (1.0 - w);
+            }
+
+            lastUpdateTime = currentTime;
+
+//            invokeOffset = upstream.getTotal();
+//            invokeElapsedOffset = upstream.getTotalElapsed();
+
+        }
+
+        private double getCost() {
+            ewmaLock.lock();
+            observe();
+            int active = 0;
+            if (upstream.isHealthy()) {
+                active = 1;
+            }
+
+            ewmaLock.unlock();
+
+            double costTemp = cost;
+
+            //If we don't have any latency history, we penalize the host on the first probe.
+            return (costTemp < ZERO_COST && active != 0) ? PENALTY + active : costTemp * (active + 1);
+        }
+
+    }
+
+
+}