[#4598] Supports traffic warm-up when the service is just started

demo/demo-cse-v1/provider-canary/src/main/resources/application.yml

@@ -19,6 +19,10 @@
 servicecomb-config-order: 10
 
 servicecomb:
+  instance:
+    properties:
+      warmupTime: 60000
+      warmupCurve: 2
   service:
     application: demo-java-chassis-cse-v1
     name: provider

demo/demo-cse-v1/provider/src/main/resources/application.yml

@@ -19,6 +19,10 @@
 servicecomb-config-order: 10
 
 servicecomb:
+  instance:
+    properties:
+      warmupTime: 30000
+      warmupCurve: 2
   service:
     application: demo-java-chassis-cse-v1
     name: provider

demo/demo-cse-v1/test-client/src/main/java/org/apache/servicecomb/samples/HelloWorldIT.java

@@ -41,6 +41,7 @@ public void testRestTransport() throws Exception {
     testHelloWorldeEptyProtectionCloseWeightLess100();
     testHelloWorldEmptyProtectionCloseFallback();
     testHelloWorldEmptyProtectionCloseWeight100Two();
+    testHelloWorldWarmUp();
   }
 
   private void testHelloWorld() {
@@ -200,4 +201,25 @@ private void testHelloWorldEmptyProtectionCloseWeight100Two() {
 
     TestMgr.check(failCount == succCount, true);
   }
+
+  private void testHelloWorldWarmUp() throws InterruptedException {
+    int oldCount = 0;
+    int newCount = 0;
+
+    for (int i = 0; i < 100; i++) {
+      String result = template
+              .getForObject(Config.GATEWAY_URL + "/sayHello?name=World", String.class);
+      Thread.sleep(1000);
+      if (result.equals("\"Hello World\"")) {
+        oldCount++;
+      } else if (result.equals("\"Hello in canary World\"")) {
+        newCount++;
+      } else {
+        TestMgr.fail("not expected result testHelloWorldCanary");
+        return;
+      }
+    }
+
+    TestMgr.check(oldCount > newCount, true);
+  }
 }

foundations/foundation-registry/src/main/java/org/apache/servicecomb/registry/api/registry/MicroserviceInstance.java

@@ -25,6 +25,7 @@
 import org.apache.commons.configuration.Configuration;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.servicecomb.config.BootStrapProperties;
+import org.apache.servicecomb.registry.config.InstancePropertiesConst;
 import org.apache.servicecomb.registry.config.InstancePropertiesLoader;
 import org.apache.servicecomb.registry.definition.DefinitionConst;
 
@@ -188,6 +189,9 @@ public static MicroserviceInstance createFromDefinition(Configuration configurat
 
     // load properties
     Map<String, String> propertiesMap = InstancePropertiesLoader.INSTANCE.loadProperties(configuration);
+
+    // add register time for warm up
+    propertiesMap.put(InstancePropertiesConst.REGISTER_TIME_KEY, String.valueOf(System.currentTimeMillis()));
     microserviceInstance.setProperties(propertiesMap);
 
     // load data center information

foundations/foundation-registry/src/main/java/org/apache/servicecomb/registry/config/InstancePropertiesConst.java

@@ -0,0 +1,22 @@
+/*
+ * 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.servicecomb.registry.config;
+
+public class InstancePropertiesConst {
+  public static final String REGISTER_TIME_KEY = "registerTime";
+}

handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/ServerListFilterExt.java

@@ -34,12 +34,16 @@ public interface ServerListFilterExt {
 
   int ORDER_ZONE_AWARE = 200;
 
+  int ORDER_WARM_UP = Integer.MAX_VALUE;
+
   String EMPTY_INSTANCE_PROTECTION = "servicecomb.loadbalance.filter.isolation.emptyInstanceProtectionEnabled";
 
   String ISOLATION_FILTER_ENABLED = "servicecomb.loadbalance.filter.isolation.enabled";
 
   String ZONE_AWARE_FILTER_ENABLED = "servicecomb.loadbalance.filter.zoneaware.enabled";
 
+  String WARM_UP_FILTER_ENABLED = "servicecomb.loadbalance.filter.service.warmup.enabled";
+
   default int getOrder() {
     return ORDER_NORMAL;
   }

handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/filterext/WarmUpDiscoveryFilter.java

@@ -0,0 +1,215 @@
+/*
+ * 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.servicecomb.loadbalance.filterext;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.netflix.config.DynamicPropertyFactory;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.servicecomb.core.Invocation;
+import org.apache.servicecomb.loadbalance.ServerListFilterExt;
+import org.apache.servicecomb.loadbalance.ServiceCombServer;
+import org.apache.servicecomb.registry.config.InstancePropertiesConst;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class WarmUpDiscoveryFilter implements ServerListFilterExt {
+  private static final Logger LOGGER = LoggerFactory.getLogger(WarmUpDiscoveryFilter.class);
+
+  private static final String IGNORE_WARN_UP_TIME = "servicecomb.loadbalance.filter.service.warmup.ignoreWarmUpTime";
+
+  private static final int INSTANCE_WEIGHT = 100;
+
+  // Default time for warm up, the unit is milliseconds
+  private static final String DEFAULT_WARM_UP_TIME = "30000";
+
+  private static final String WARM_TIME_KEY = "warmupTime";
+
+  private static final String WARM_CURVE_KEY = "warmupCurve";
+
+  // Preheat calculates curve value
+  private static final String DEFAULT_WARM_UP_CURVE = "2";
+
+  // provider run time, the unit is milliseconds
+  private final long ignoreWarmUpTime;
+
+  private final Random random = new Random();
+
+  private final Map<String, Long> instanceInvokeTime = new ConcurrentHashMap<>();
+
+  public WarmUpDiscoveryFilter() {
+    ScheduledExecutorService executor = Executors.newScheduledThreadPool(1,
+        new ThreadFactoryBuilder()
+            .setNameFormat("warm-up-cache-refresh-%d")
+            .build());
+    ignoreWarmUpTime = getIgnoreWarmUpTime();
+    executor.scheduleAtFixedRate(this::refreshMapCache, 0, 10, TimeUnit.MINUTES);
+  }
+
+  private long getIgnoreWarmUpTime() {
+    return DynamicPropertyFactory.getInstance()
+        .getLongProperty(IGNORE_WARN_UP_TIME, 30 * 60 * 1000)
+        .get();
+  }
+
+  private void refreshMapCache() {
+    List<String> removeKeys = new ArrayList<>();
+    for (Map.Entry<String, Long> entry : instanceInvokeTime.entrySet()) {
+      if (System.currentTimeMillis() - entry.getValue() > ignoreWarmUpTime) {
+        removeKeys.add(entry.getKey());
+      }
+    }
+    if (CollectionUtils.isEmpty(removeKeys)) {
+      return;
+    }
+    removeKeys.forEach(instanceInvokeTime::remove);
+  }
+
+  @Override
+  public int getOrder() {
+    return ORDER_WARM_UP;
+  }
+
+  @Override
+  public boolean enabled() {
+    return DynamicPropertyFactory.getInstance()
+        .getBooleanProperty(WARM_UP_FILTER_ENABLED, false)
+        .get();
+  }
+
+  @Override
+  public List<ServiceCombServer> getFilteredListOfServers(List<ServiceCombServer> servers,
+      Invocation invocation) {
+    if (servers.size() <= 1) {
+      return servers;
+    }
+    List<ServiceCombServer> notInvokedInstances = new ArrayList<>();
+    if (CollectionUtils.isEmpty(existNeedWarmUpInstances(servers, notInvokedInstances))) {
+      return servers;
+    }
+
+    // If exist not invoked instances, choose one that let it into warm up.
+    if (!CollectionUtils.isEmpty(notInvokedInstances)) {
+      setInstanceStartInvokeTime(notInvokedInstances.get(0));
+      return Collections.singletonList(notInvokedInstances.get(0));
+    }
+    int[] weights = new int[servers.size()];
+    int totalWeight = 0;
+    int index = 0;
+    for (ServiceCombServer server : servers) {
+      weights[index] = calculate(server.getInstance().getProperties(), server.getInstance().getInstanceId());
+      totalWeight += weights[index++];
+    }
+    return chooseServer(totalWeight, weights, servers);
+  }
+
+  private List<ServiceCombServer> existNeedWarmUpInstances(List<ServiceCombServer> servers,
+      List<ServiceCombServer> notInvokedInstances) {
+    return servers.stream()
+            .filter(server -> isInstanceNeedWarmUp(server, notInvokedInstances))
+            .collect(Collectors.toList());
+  }
+
+  private boolean isInstanceNeedWarmUp(ServiceCombServer server, List<ServiceCombServer> notInvokedInstances) {
+    Map<String, String> properties = server.getInstance().getProperties();
+    String instanceId = server.getInstance().getInstanceId();
+    if (isUpIgnoreWarmUpTime(properties)) {
+      return false;
+    }
+    Long invokeTime = instanceInvokeTime.get(instanceId);
+
+    // instance have not been invoked need to be warn-up
+    if (invokeTime == null) {
+      notInvokedInstances.add(server);
+      return true;
+    }
+    final long warmUpTime = Long.parseLong(properties.getOrDefault(WARM_TIME_KEY, DEFAULT_WARM_UP_TIME));
+    return System.currentTimeMillis() - invokeTime < warmUpTime;
+  }
+
+  private boolean isUpIgnoreWarmUpTime(Map<String, String> properties) {
+    String registerTimeStr = properties.get(InstancePropertiesConst.REGISTER_TIME_KEY);
+    long registerTime = Long.parseLong(StringUtils.isEmpty(registerTimeStr) ? "0" : registerTimeStr);
+
+    // Provider run time greater than 30 minute, can be regarded as don't need warn up.
+    // To ensure that the consumer is restarted but the provider is not restarted.
+    return System.currentTimeMillis() - registerTime > ignoreWarmUpTime;
+  }
+
+  private List<ServiceCombServer> chooseServer(int totalWeight, int[] weights, List<ServiceCombServer> servers) {
+    if (totalWeight <= 0) {
+      return servers;
+    }
+    int position = random.nextInt(totalWeight);
+    for (int i = 0; i < weights.length; i++) {
+      position -= weights[i];
+      if (position < 0) {
+        if (LOGGER.isDebugEnabled()) {
+          LOGGER.debug("warm up choose service instance: " + servers.get(i).getInstance().getInstanceId());
+        }
+        return Collections.singletonList(servers.get(i));
+      }
+    }
+    return servers;
+  }
+
+  private void setInstanceStartInvokeTime(ServiceCombServer server) {
+    instanceInvokeTime.putIfAbsent(server.getInstance().getInstanceId(), System.currentTimeMillis());
+  }
+
+  private int calculate(Map<String, String> properties, String instanceId) {
+    if (instanceInvokeTime.get(instanceId) == null) {
+      return INSTANCE_WEIGHT;
+    }
+    final int warmUpCurve = Integer.parseInt(properties.getOrDefault(WARM_CURVE_KEY, DEFAULT_WARM_UP_CURVE));
+    final long warmUpTime = Long.parseLong(properties.getOrDefault(WARM_TIME_KEY, DEFAULT_WARM_UP_TIME));
+    return calculateWeight(instanceInvokeTime.get(instanceId), warmUpTime, warmUpCurve);
+  }
+
+  private int calculateWeight(long invokeStartTime, long warmUpTime, int warmUpCurve) {
+    if (warmUpTime <= 0) {
+      return INSTANCE_WEIGHT;
+    }
+    if (warmUpCurve <= 0) {
+      warmUpCurve = Integer.parseInt(DEFAULT_WARM_UP_CURVE);
+    }
+    // calculated in milliseconds
+    final long runTime = System.currentTimeMillis() - invokeStartTime;
+    if (runTime > 0 && runTime < warmUpTime) {
+      return calculateWarmUpWeight(runTime, warmUpTime, warmUpCurve);
+    }
+    return INSTANCE_WEIGHT;
+  }
+
+  private int calculateWarmUpWeight(double runtime, double warmUpTime, int warmUpCurve) {
+    final int round = (int) Math.round(Math.pow(runtime / warmUpTime, warmUpCurve) * INSTANCE_WEIGHT);
+    return round < 1 ? 1 : Math.min(round, INSTANCE_WEIGHT);
+  }
+}

handlers/handler-loadbalance/src/main/resources/META-INF/services/org.apache.servicecomb.loadbalance.ServerListFilterExt

@@ -16,4 +16,5 @@
 #
 
 org.apache.servicecomb.loadbalance.filterext.IsolationServerListFilterExt
-org.apache.servicecomb.loadbalance.filterext.ZoneAwareDiscoveryFilter
+org.apache.servicecomb.loadbalance.filterext.ZoneAwareDiscoveryFilter
+org.apache.servicecomb.loadbalance.filterext.WarmUpDiscoveryFilter

service-registry/registry-service-center/src/test/java/org/apache/servicecomb/serviceregistry/config/TestPropertiesLoader.java

@@ -26,6 +26,7 @@
 import org.apache.servicecomb.registry.api.registry.Microservice;
 import org.apache.servicecomb.registry.api.registry.MicroserviceFactory;
 import org.apache.servicecomb.registry.api.registry.MicroserviceInstance;
+import org.apache.servicecomb.registry.config.InstancePropertiesConst;
 import org.apache.servicecomb.serviceregistry.registry.LocalServiceRegistryFactory;
 import org.junit.Test;
 import org.junit.jupiter.api.Assertions;
@@ -89,6 +90,7 @@ public void testMicroservicePropertiesLoader() throws Exception {
   public void testInstancePropertiesLoader() {
     Microservice microservice = LocalServiceRegistryFactory.createLocal().getMicroservice();
     MicroserviceInstance instance = microservice.getInstance();
+    instance.getProperties().remove(InstancePropertiesConst.REGISTER_TIME_KEY);
     Map<String, String> expectedMap = new HashMap<>();
     expectedMap.put("key0", "value0");
     expectedMap.put("ek0", "ev0");