[#1191] add weightedResponseTime loadbalancer

integration-tests/discovery-nacos-tests/account-provider/src/main/java/com/huaweicloud/crossappsample/AuthController.java

@@ -19,8 +19,8 @@
 import com.huaweicloud.common.configration.dynamic.BlackWhiteListProperties;
 import com.huaweicloud.common.context.InvocationContext;
 import com.huaweicloud.common.context.InvocationContextHolder;
+import com.huaweicloud.governance.GovernanceConst;
 import com.huaweicloud.governance.authentication.AuthHandlerBoot;
-import com.huaweicloud.governance.authentication.Const;
 import com.huaweicloud.governance.authentication.ProviderAuthPreHandlerInterceptor;
 import com.huaweicloud.governance.authentication.securityPolicy.SecurityPolicyProperties;
 
@@ -51,7 +51,7 @@ public String checkToken() {
     }
 
     InvocationContext invocationContext = InvocationContextHolder.getOrCreateInvocationContext();
-    if (StringUtils.isEmpty(invocationContext.getContext(Const.AUTH_TOKEN))) {
+    if (StringUtils.isEmpty(invocationContext.getContext(GovernanceConst.AUTH_TOKEN))) {
       return null;
     }
     return "success";
@@ -69,7 +69,7 @@ public String checkTokenSecurity() {
     }
 
     InvocationContext invocationContext = InvocationContextHolder.getOrCreateInvocationContext();
-    if (StringUtils.isEmpty(invocationContext.getContext(Const.AUTH_TOKEN))) {
+    if (StringUtils.isEmpty(invocationContext.getContext(GovernanceConst.AUTH_TOKEN))) {
       return null;
     }
     return "success";

integration-tests/discovery-servicecomb-tests/account-provider/src/main/java/com/huaweicloud/crossappsample/AuthController.java

@@ -17,8 +17,8 @@
 package com.huaweicloud.crossappsample;
 
 import com.huaweicloud.common.configration.dynamic.BlackWhiteListProperties;
+import com.huaweicloud.governance.GovernanceConst;
 import com.huaweicloud.governance.authentication.AuthHandlerBoot;
-import com.huaweicloud.governance.authentication.Const;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
@@ -51,7 +51,7 @@ public String checkToken() {
     }
 
     InvocationContext invocationContext = InvocationContextHolder.getOrCreateInvocationContext();
-    if (StringUtils.isEmpty(invocationContext.getContext(Const.AUTH_TOKEN))) {
+    if (StringUtils.isEmpty(invocationContext.getContext(GovernanceConst.AUTH_TOKEN))) {
       return null;
     }
     return "success";
@@ -69,7 +69,7 @@ public String checkTokenSecurity() {
     }
 
     InvocationContext invocationContext = InvocationContextHolder.getOrCreateInvocationContext();
-    if (StringUtils.isEmpty(invocationContext.getContext(Const.AUTH_TOKEN))) {
+    if (StringUtils.isEmpty(invocationContext.getContext(GovernanceConst.AUTH_TOKEN))) {
       return null;
     }
     return "success";

spring-cloud-huawei-common/src/main/java/com/huaweicloud/common/configration/dynamic/LoadBalancerProperties.java

@@ -24,6 +24,8 @@ public class LoadBalancerProperties {
 
   public static final String RULE_RANDOM = "Random";
 
+  public static final String RULE_WEIGHTED_RESPONSE = "WeightedResponse";
+
   private String rule = RULE_ROUND_ROBIN;
 
   public String getRule() {

spring-cloud-huawei-governance/src/main/java/com/huaweicloud/governance/authentication/Const.java → spring-cloud-huawei-governance/src/main/java/com/huaweicloud/governance/GovernanceConst.java

@@ -15,9 +15,9 @@
  * limitations under the License.
  */
 
-package com.huaweicloud.governance.authentication;
+package com.huaweicloud.governance;
 
-public class Const {
+public class GovernanceConst {
   public static final String AUTH_TOKEN = "x-auth-token";
 
   public static final String INSTANCE_PUBKEY_PRO = "publickey";
@@ -29,4 +29,6 @@ public class Const {
   public static final String AUTH_TOKEN_HEADER_KEY = "spring.cloud.servicecomb.webmvc.publicKey.headerTokenKey";
 
   public static final String AUTH_API_PATH_WHITELIST = "spring.cloud.servicecomb.webmvc.publicKey.apiPathWhitelist";
+
+  public static final String CONTEXT_CURRENT_INSTANCE = "x-current-instance";
 }

spring-cloud-huawei-governance/src/main/java/com/huaweicloud/governance/adapters/feign/GovernanceFeignBlockingLoadBalancerClient.java

@@ -27,6 +27,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.servicecomb.governance.handler.FaultInjectionHandler;
 import org.apache.servicecomb.governance.handler.InstanceBulkheadHandler;
@@ -65,13 +66,16 @@
 import com.huaweicloud.common.context.InvocationStage;
 import com.huaweicloud.common.disovery.InstanceIDAdapter;
 import com.huaweicloud.common.event.EventManager;
+import com.huaweicloud.governance.GovernanceConst;
 import com.huaweicloud.governance.adapters.loadbalancer.RetryContext;
+import com.huaweicloud.governance.adapters.loadbalancer.weightedResponseTime.ServiceInstanceMetrics;
 import com.huaweicloud.governance.event.InstanceIsolatedEvent;
 
 import io.github.resilience4j.bulkhead.Bulkhead;
 import io.github.resilience4j.bulkhead.BulkheadFullException;
 import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
 import io.github.resilience4j.circuitbreaker.CircuitBreaker;
+import io.github.resilience4j.core.metrics.Metrics.Outcome;
 import io.github.resilience4j.decorators.Decorators;
 import io.github.resilience4j.decorators.Decorators.DecorateCheckedSupplier;
 import io.github.resilience4j.retry.Retry;
@@ -140,16 +144,30 @@ public Response execute(Request request, Request.Options options) {
     InvocationContext context = InvocationContextHolder.getOrCreateInvocationContext();
     String stageName = context.getInvocationStage().recordStageBegin(stageId(originalUri, request));
 
+    long time = System.currentTimeMillis();
     try {
       Response response = decorateWithFault(request, options, originalUri);
       context.getInvocationStage().recordStageEnd(stageName);
+      if (response.status() == 200) {
+        metricsRecord(Outcome.SUCCESS, context, time);
+      } else {
+        metricsRecord(Outcome.ERROR, context, time);
+      }
       return response;
     } catch (Throwable error) {
       context.getInvocationStage().recordStageEnd(stageName);
+      metricsRecord(Outcome.ERROR, context, time);
       throw error;
     }
   }
 
+  private void metricsRecord(Outcome outcome, InvocationContext context, long time) {
+    if (context.getLocalContext(GovernanceConst.CONTEXT_CURRENT_INSTANCE) != null) {
+      ServiceInstanceMetrics.getMetrics(context.getLocalContext(GovernanceConst.CONTEXT_CURRENT_INSTANCE))
+          .record((System.currentTimeMillis() - time), TimeUnit.MILLISECONDS, outcome);
+    }
+  }
+
   private String stageId(URI uri, Request request) {
     return InvocationStage.STAGE_FEIGN + " " + request.httpMethod().name() + " " + uri.getPath();
   }

spring-cloud-huawei-governance/src/main/java/com/huaweicloud/governance/adapters/gateway/RetryGlobalFilter.java

@@ -20,6 +20,7 @@
 import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR;
 
 import java.net.URI;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.servicecomb.governance.handler.RetryHandler;
 import org.apache.servicecomb.governance.marker.GovernanceRequestExtractor;
@@ -33,8 +34,11 @@
 
 import com.huaweicloud.common.context.InvocationContext;
 import com.huaweicloud.common.context.InvocationContextHolder;
+import com.huaweicloud.governance.GovernanceConst;
 import com.huaweicloud.governance.adapters.loadbalancer.RetryContext;
+import com.huaweicloud.governance.adapters.loadbalancer.weightedResponseTime.ServiceInstanceMetrics;
 
+import io.github.resilience4j.core.metrics.Metrics.Outcome;
 import io.github.resilience4j.reactor.retry.RetryOperator;
 import io.github.resilience4j.retry.Retry;
 import reactor.core.publisher.Mono;
@@ -61,10 +65,21 @@ public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
   private Mono<Void> addRetry(ServerWebExchange exchange, GovernanceRequestExtractor governanceRequest,
       Mono<Void> toRun) {
     Retry retry = retryHandler.getActuator(governanceRequest);
+    InvocationContext invocationContext = exchange.getAttribute(InvocationContextHolder.ATTRIBUTE_KEY);
+    long time = System.currentTimeMillis();
     if (retry == null) {
-      return toRun;
+      return toRun
+          .doOnSuccess(resp -> {
+            if (invocationContext != null) {
+              metricsRecord(Outcome.SUCCESS, invocationContext, time);
+            }
+          })
+          .doOnError(resp -> {
+            if (invocationContext != null) {
+              metricsRecord(Outcome.ERROR, invocationContext, time);
+            }
+          }).then();
     }
-    InvocationContext invocationContext = exchange.getAttribute(InvocationContextHolder.ATTRIBUTE_KEY);
     if (invocationContext != null) {
       RetryPolicy retryPolicy = retryHandler.matchPolicy(governanceRequest);
       RetryContext retryContext = new RetryContext(retryPolicy.getRetryOnSame());
@@ -82,9 +97,26 @@ private Mono<Void> addRetry(ServerWebExchange exchange, GovernanceRequestExtract
             Mono.defer(() -> Mono.just(exchange.getResponse().getStatusCode() == null ? 0
                 : exchange.getResponse().getStatusCode().value())))
         .transformDeferred(RetryOperator.of(retry))
+        .doOnSuccess(resp -> {
+          if (invocationContext != null) {
+            metricsRecord(Outcome.SUCCESS, invocationContext, time);
+          }
+        })
+        .doOnError(resp -> {
+          if (invocationContext != null) {
+            metricsRecord(Outcome.ERROR, invocationContext, time);
+          }
+        })
         .then();
   }
 
+  private void metricsRecord(Outcome outcome, InvocationContext context, long time) {
+    if (context.getLocalContext(GovernanceConst.CONTEXT_CURRENT_INSTANCE) != null) {
+      ServiceInstanceMetrics.getMetrics(context.getLocalContext(GovernanceConst.CONTEXT_CURRENT_INSTANCE))
+          .record((System.currentTimeMillis() - time), TimeUnit.MILLISECONDS, outcome);
+    }
+  }
+
   private void reset(ServerWebExchange exchange) {
     ServerWebExchangeUtils.reset(exchange);
   }

spring-cloud-huawei-governance/src/main/java/com/huaweicloud/governance/adapters/loadbalancer/RetryAwareLoadBalancer.java

@@ -19,6 +19,7 @@
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
 
 import org.apache.servicecomb.governance.handler.LoadBalanceHandler;
 import org.apache.servicecomb.governance.marker.GovernanceRequestExtractor;
@@ -38,6 +39,8 @@
 import com.huaweicloud.common.configration.dynamic.LoadBalancerProperties;
 import com.huaweicloud.common.context.InvocationContext;
 import com.huaweicloud.common.context.InvocationContextHolder;
+import com.huaweicloud.governance.GovernanceConst;
+import com.huaweicloud.governance.adapters.loadbalancer.weightedResponseTime.WeightedResponseTimeLoadBalancer;
 
 import reactor.core.publisher.Mono;
 
@@ -55,6 +58,8 @@ public class RetryAwareLoadBalancer implements ReactorServiceInstanceLoadBalance
 
   private final Map<String, ReactorServiceInstanceLoadBalancer> loadBalancers = new ConcurrentHashMap<>();
 
+  private final ThreadLocalRandom threadLocalRandom = ThreadLocalRandom.current();
+
   public RetryAwareLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider,
       String serviceId, LoadBalancerProperties loadBalancerProperties, LoadBalanceHandler loadBalanceHandler) {
     this.serviceInstanceListSupplierProvider = serviceInstanceListSupplierProvider;
@@ -71,15 +76,9 @@ public Mono<Response<ServiceInstance>> choose(Request request) {
     LoadBalance loadBalance = loadBalanceHandler.getActuator(governanceRequest);
     String rule = loadBalance != null ? loadBalance.getRule() : loadBalancerProperties.getRule();
     if (context.getLocalContext(RetryContext.RETRY_CONTEXT) == null) {
-      ReactorServiceInstanceLoadBalancer loadBalancer = loadBalancers.computeIfAbsent(rule,
-          key -> {
-            if (LoadBalancerProperties.RULE_RANDOM.equals(key)) {
-              return new RandomLoadBalancer(this.serviceInstanceListSupplierProvider, this.serviceId);
-            } else {
-              return new RoundRobinLoadBalancer(this.serviceInstanceListSupplierProvider, this.serviceId);
-            }
-          });
-      return loadBalancer.choose(request);
+      ReactorServiceInstanceLoadBalancer loadBalancer = loadBalancers.computeIfAbsent(rule, this::builedLoadBalancer);
+      return loadBalancer.choose(request).doOnSuccess(r ->
+          context.putLocalContext(GovernanceConst.CONTEXT_CURRENT_INSTANCE, r.getServer()));
     }
 
     RetryContext retryContext = context.getLocalContext(RetryContext.RETRY_CONTEXT);
@@ -88,17 +87,25 @@ public Mono<Response<ServiceInstance>> choose(Request request) {
       return Mono.just(new DefaultResponse(retryContext.getLastServer()));
     }
 
-    ReactorServiceInstanceLoadBalancer loadBalancer = loadBalancers.computeIfAbsent(rule,
-        key -> {
-          if (LoadBalancerProperties.RULE_RANDOM.equals(key)) {
-            return new RandomLoadBalancer(this.serviceInstanceListSupplierProvider, this.serviceId);
-          } else {
-            return new RoundRobinLoadBalancer(this.serviceInstanceListSupplierProvider, this.serviceId);
-          }
-        });
-    return loadBalancer.choose(request).doOnSuccess(r -> retryContext.setLastServer(r.getServer()));
+    ReactorServiceInstanceLoadBalancer loadBalancer = loadBalancers.computeIfAbsent(rule, this::builedLoadBalancer);
+    return loadBalancer.choose(request).doOnSuccess(r -> {
+      context.putLocalContext(GovernanceConst.CONTEXT_CURRENT_INSTANCE, r.getServer());
+      retryContext.setLastServer(r.getServer());
+    });
   }
 
+  private ReactorServiceInstanceLoadBalancer builedLoadBalancer(String key) {
+    if (LoadBalancerProperties.RULE_RANDOM.equals(key)) {
+      return new RandomLoadBalancer(this.serviceInstanceListSupplierProvider, this.serviceId);
+    } else if (LoadBalancerProperties.RULE_WEIGHTED_RESPONSE.equals(key)) {
+      return new WeightedResponseTimeLoadBalancer(this.serviceId, threadLocalRandom.nextInt(1000),
+          this.serviceInstanceListSupplierProvider);
+    } else {
+      return new RoundRobinLoadBalancer(this.serviceInstanceListSupplierProvider, this.serviceId);
+    }
+  }
+
+
   @SuppressWarnings("rawtypes")
   private InvocationContext getOrCreateInvocationContext(Request request) {
     Object context = request.getContext();

spring-cloud-huawei-governance/src/main/java/com/huaweicloud/governance/adapters/loadbalancer/weightedResponseTime/ServerMetrics.java

@@ -0,0 +1,39 @@
+/*
+
+ * Copyright (C) 2020-2024 Huawei Technologies Co., Ltd. All rights reserved.
+
+ * Licensed 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 com.huaweicloud.governance.adapters.loadbalancer.weightedResponseTime;
+
+import java.util.concurrent.TimeUnit;
+
+import io.github.resilience4j.core.metrics.FixedSizeSlidingWindowMetrics;
+import io.github.resilience4j.core.metrics.Metrics.Outcome;
+import io.github.resilience4j.core.metrics.Snapshot;
+
+/**
+ * ServiceCombServer states
+ */
+public class ServerMetrics {
+  private final FixedSizeSlidingWindowMetrics metrics = new FixedSizeSlidingWindowMetrics(50);
+
+  public void record(long duration, TimeUnit durationUnit, Outcome outcome) {
+    metrics.record(duration, durationUnit, outcome);
+  }
+
+  public Snapshot getSnapshot() {
+    return metrics.getSnapshot();
+  }
+}

spring-cloud-huawei-governance/src/main/java/com/huaweicloud/governance/adapters/loadbalancer/weightedResponseTime/ServiceInstanceMetrics.java

@@ -0,0 +1,53 @@
+/*
+
+ * Copyright (C) 2020-2024 Huawei Technologies Co., Ltd. All rights reserved.
+
+ * Licensed 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 com.huaweicloud.governance.adapters.loadbalancer.weightedResponseTime;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.cloud.client.ServiceInstance;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+
+public class ServiceInstanceMetrics {
+  private static final Logger LOGGER = LoggerFactory.getLogger(ServiceInstanceMetrics.class);
+
+  public static final Cache<String, ServerMetrics> INSTANCE_SERVER_METRICS_MAP = CacheBuilder.newBuilder()
+      .expireAfterAccess(12 * 60 * 60 * 1000, TimeUnit.MILLISECONDS)
+      .build();
+
+  public static ServerMetrics getMetrics(ServiceInstance instance) {
+    try {
+      return INSTANCE_SERVER_METRICS_MAP.get(buildKey(instance), ServerMetrics::new);
+    } catch (Exception e) {
+      LOGGER.error("serverMetrics load failed.");
+      return new ServerMetrics();
+    }
+  }
+
+  private static String buildKey(ServiceInstance instance) {
+    if (StringUtils.isEmpty(instance.getInstanceId())) {
+      String result = instance.getHost() + ":" + instance.getPort();
+      return result.replaceAll("[^0-9a-zA-Z]", "-");
+    }
+    return instance.getInstanceId();
+  }
+}