Spring cloud version upgraded, loadBalancer replaced ribbon

linkis-commons/linkis-module/pom.xml

@@ -266,6 +266,10 @@
       <artifactId>jedis</artifactId>
       <version>${jedis.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.springframework.cloud</groupId>
+      <artifactId>spring-cloud-openfeign-core</artifactId>
+    </dependency>
   </dependencies>
 
   <build>

linkis-commons/linkis-module/src/main/java/org/apache/linkis/DataWorkCloudApplication.java

@@ -41,6 +41,7 @@
 import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
 import org.springframework.cloud.context.config.annotation.RefreshScope;
 import org.springframework.cloud.context.scope.refresh.RefreshScopeRefreshedEvent;
+import org.springframework.cloud.openfeign.EnableFeignClients;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.ConfigurableApplicationContext;
@@ -66,6 +67,7 @@
 @SpringBootApplication(scanBasePackages = {"org.apache.linkis", "com.webank.wedatasphere"})
 @EnableDiscoveryClient
 @RefreshScope
+@EnableFeignClients
 public class DataWorkCloudApplication extends SpringBootServletInitializer {
   private static final Log logger = LogFactory.getLog(DataWorkCloudApplication.class);
 

linkis-commons/linkis-module/src/main/java/org/apache/linkis/swagger/SwaggerBeanPostProcessor.java

@@ -0,0 +1,65 @@
+/*
+ * 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.linkis.swagger;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.util.ReflectionUtils;
+import org.springframework.web.servlet.mvc.method.RequestMappingInfoHandlerMapping;
+
+import java.lang.reflect.Field;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import springfox.documentation.spring.web.plugins.WebFluxRequestHandlerProvider;
+import springfox.documentation.spring.web.plugins.WebMvcRequestHandlerProvider;
+
+@Configuration
+public class SwaggerBeanPostProcessor implements BeanPostProcessor {
+  @Override
+  public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+    if (bean instanceof WebMvcRequestHandlerProvider
+        || bean instanceof WebFluxRequestHandlerProvider) {
+      List<RequestMappingInfoHandlerMapping> handlerMappings = getHandlerMappings(bean);
+      customizeSpringfoxHandlerMappings(handlerMappings);
+    }
+    return bean;
+  }
+
+  private <T extends RequestMappingInfoHandlerMapping> void customizeSpringfoxHandlerMappings(
+      List<T> mappings) {
+    List<T> copy =
+        mappings.stream()
+            .filter(mapping -> mapping.getPatternParser() == null)
+            .collect(Collectors.toList());
+    mappings.clear();
+    mappings.addAll(copy);
+  }
+
+  @SuppressWarnings("unchecked")
+  private List<RequestMappingInfoHandlerMapping> getHandlerMappings(Object bean) {
+    try {
+      Field field = ReflectionUtils.findField(bean.getClass(), "handlerMappings");
+      field.setAccessible(true);
+      return (List<RequestMappingInfoHandlerMapping>) field.get(bean);
+    } catch (IllegalArgumentException | IllegalAccessException e) {
+      throw new IllegalStateException(e);
+    }
+  }
+}

linkis-commons/linkis-rpc/src/main/java/org/apache/linkis/rpc/conf/DynamicFeignClient.java

@@ -0,0 +1,126 @@
+/*
+ * 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.linkis.rpc.conf;
+
+import org.apache.linkis.DataWorkCloudApplication;
+
+import org.apache.commons.lang3.StringUtils;
+
+import org.springframework.cloud.openfeign.FeignClientBuilder;
+import org.springframework.cloud.openfeign.FeignClientFactoryBean;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+@Component
+public class DynamicFeignClient<T> {
+
+  private FeignClientBuilder feignClientBuilder;
+
+  private final ConcurrentHashMap<String, T> CACHE_BEAN = new ConcurrentHashMap();
+
+  public DynamicFeignClient() {
+    this.feignClientBuilder =
+        new FeignClientBuilder(DataWorkCloudApplication.getApplicationContext());
+  }
+
+  public T getFeignClient(final Class<T> type, final String serviceName) {
+    return getFeignClient(type, serviceName, null);
+  }
+
+  public T getFeignClient(
+      final Class<T> type, final Class<?> fallbackFactory, final String serviceName) {
+    return getFeignClient(type, fallbackFactory, serviceName, null);
+  }
+
+  public T getFeignClient(
+      final Class<T> type,
+      final FeignClientFactoryBean clientFactoryBean,
+      final String serviceName) {
+    return getFeignClient(type, clientFactoryBean, serviceName, null);
+  }
+
+  public T getFeignClient(final Class<T> type, String serviceName, final String serviceUrl) {
+    String k = serviceName;
+    if (StringUtils.isNotEmpty(serviceUrl)) {
+      k = serviceUrl;
+    }
+    return CACHE_BEAN.computeIfAbsent(
+        k,
+        (t) -> {
+          FeignClientBuilder.Builder<T> builder =
+              this.feignClientBuilder.forType(type, serviceName);
+          if (StringUtils.isNotEmpty(serviceUrl)) {
+            builder.url(serviceUrl);
+          }
+          return builder.build();
+        });
+  }
+
+  public T getFeignClient(
+      final Class<T> type,
+      final Class<?> fallbackFactory,
+      final String serviceName,
+      final String serviceUrl) {
+    String k = serviceName;
+    if (StringUtils.isNotEmpty(serviceUrl)) {
+      k = serviceUrl;
+    }
+    return CACHE_BEAN.computeIfAbsent(
+        k,
+        (t) -> {
+          FeignClientFactoryBean feignClientFactoryBean = new FeignClientFactoryBean();
+          feignClientFactoryBean.setFallbackFactory(fallbackFactory);
+          FeignClientBuilder.Builder<T> builder =
+              this.feignClientBuilder.forType(type, feignClientFactoryBean, serviceName);
+          if (StringUtils.isNotEmpty(serviceUrl)) {
+            builder.url(serviceUrl);
+          }
+          return builder.build();
+        });
+  }
+
+  public T getFeignClient(
+      final Class<T> type,
+      final FeignClientFactoryBean clientFactoryBean,
+      final String serviceName,
+      final String serviceUrl) {
+    String k = serviceName;
+    if (StringUtils.isNotEmpty(serviceUrl)) {
+      k = serviceUrl;
+    }
+    return CACHE_BEAN.computeIfAbsent(
+        k,
+        (t) -> {
+          FeignClientBuilder.Builder<T> builder =
+              this.feignClientBuilder.forType(type, clientFactoryBean, serviceName);
+          if (StringUtils.isNotEmpty(serviceUrl)) {
+            builder.url(serviceUrl);
+          }
+          return builder.build();
+        });
+  }
+
+  private T getFromCache(final String serviceName, final String serviceUrl) {
+    if (StringUtils.isNotEmpty(serviceUrl)) {
+      return CACHE_BEAN.get(serviceUrl);
+    } else {
+      return CACHE_BEAN.get(serviceName);
+    }
+  }
+}

linkis-commons/linkis-rpc/src/main/java/org/apache/linkis/rpc/conf/EurekaClientCacheManualRefresher.java

@@ -0,0 +1,112 @@
+/*
+ * 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.linkis.rpc.conf;
+
+import org.apache.commons.lang.exception.ExceptionUtils;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ReflectionUtils;
+
+import java.lang.reflect.Field;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.netflix.discovery.DiscoveryClient;
+import com.netflix.discovery.TimedSupervisorTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Component
+public class EurekaClientCacheManualRefresher {
+  private static final Logger logger =
+      LoggerFactory.getLogger(EurekaClientCacheManualRefresher.class);
+  private final AtomicBoolean isRefreshing = new AtomicBoolean(false);
+  private final ExecutorService refreshExecutor = Executors.newSingleThreadExecutor();
+  private final String cacheRefreshTaskField = "cacheRefreshTask";
+  private TimedSupervisorTask cacheRefreshTask;
+
+  private long lastRefreshMillis = 0;
+  private final Duration refreshIntervalDuration = Duration.ofSeconds(3);
+
+  @Autowired private BeanFactory beanFactory;
+
+  public void refreshOnExceptions(Exception e, List<Class<? extends Exception>> clazzs) {
+    if (null == clazzs || clazzs.size() == 0) {
+      throw new IllegalArgumentException();
+    }
+
+    if (clazzs.stream()
+        .anyMatch(
+            clazz -> clazz.isInstance(e) || clazz.isInstance(ExceptionUtils.getRootCause(e)))) {
+      refresh();
+    }
+  }
+
+  public void refresh() {
+    if (isRefreshing.compareAndSet(false, true)) {
+      refreshExecutor.execute(
+          () -> {
+            try {
+              if (System.currentTimeMillis()
+                  <= lastRefreshMillis + refreshIntervalDuration.toMillis()) {
+                logger.warn(
+                    "Not manually refresh eureka client cache as refresh interval was not exceeded:{}",
+                    refreshIntervalDuration.getSeconds());
+                return;
+              }
+
+              if (null == cacheRefreshTask) {
+                Field field =
+                    ReflectionUtils.findField(DiscoveryClient.class, cacheRefreshTaskField);
+                if (null != field) {
+                  ReflectionUtils.makeAccessible(field);
+                  DiscoveryClient discoveryClient = beanFactory.getBean(DiscoveryClient.class);
+                  cacheRefreshTask =
+                      (TimedSupervisorTask) ReflectionUtils.getField(field, discoveryClient);
+                }
+              }
+
+              if (null == cacheRefreshTask) {
+                logger.error(
+                    "Field ({}) not found in class '{}'",
+                    cacheRefreshTaskField,
+                    DiscoveryClient.class.getSimpleName());
+                return;
+              }
+
+              lastRefreshMillis = System.currentTimeMillis();
+              cacheRefreshTask.run();
+              logger.info(
+                  "Manually refresh eureka client cache completed(DiscoveryClient.cacheRefreshTask#run())");
+            } catch (Exception e) {
+              logger.error("An exception occurred when manually refresh eureka client cache", e);
+            } finally {
+              isRefreshing.set(false);
+            }
+          });
+    } else {
+      logger.warn(
+          "Not manually refresh eureka client cache as another thread is refreshing it already");
+    }
+  }
+}

linkis-commons/linkis-rpc/src/main/java/org/apache/linkis/rpc/conf/FeignRequestInterceptor.java

@@ -0,0 +1,64 @@
+/*
+ * 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.linkis.rpc.conf;
+
+import org.apache.linkis.rpc.BaseRPCSender;
+import org.apache.linkis.rpc.constant.RpcConstant;
+import org.apache.linkis.server.BDPJettyServerHelper;
+import org.apache.linkis.server.Message;
+import org.apache.linkis.server.security.SSOUtils$;
+import org.apache.linkis.server.security.SecurityFilter$;
+
+import org.springframework.stereotype.Component;
+
+import java.io.UnsupportedEncodingException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+import scala.Tuple2;
+
+import feign.RequestInterceptor;
+import feign.RequestTemplate;
+
+@Component
+public class FeignRequestInterceptor implements RequestInterceptor {
+
+  @Override
+  public void apply(RequestTemplate requestTemplate) {
+    Map<String, Collection<String>> headers = new HashMap<>(requestTemplate.headers());
+    headers.put(
+        RpcConstant.LINKIS_LOAD_BALANCER_TYPE,
+        Arrays.asList(RpcConstant.LINKIS_LOAD_BALANCER_TYPE_RPC));
+    Tuple2<String, String> userTicketKV =
+        SSOUtils$.MODULE$.getUserTicketKV(SecurityFilter$.MODULE$.OTHER_SYSTEM_IGNORE_UM_USER());
+    headers.put(userTicketKV._1, Arrays.asList(userTicketKV._2));
+    try {
+      String body =
+          new String(
+              requestTemplate.body(),
+              org.apache.linkis.common.conf.Configuration.BDP_ENCODING().getValue());
+      Message message = BDPJettyServerHelper.gson().fromJson(body, Message.class);
+      headers.put(
+          RpcConstant.FIXED_INSTANCE, Arrays.asList(BaseRPCSender.getFixedInstanceInfo(message)));
+      requestTemplate.headers(headers);
+    } catch (UnsupportedEncodingException e) {
+    }
+  }
+}