[#1510] Fixing the issue where microservices cannot be registered when enabling RBAC authentication in a dual-engine disaster recovery scenario. (#1511)

Author: chengyouling

pom.xml

@@ -38,7 +38,7 @@
      import spring-framework-bom. No need configure spring version. -->
     <spring-cloud.version>2024.0.2</spring-cloud.version>
     <spring-boot.version>3.4.12</spring-boot.version>
-    <servicecomb.version>2.8.27</servicecomb.version>
+    <servicecomb.version>2.8.29</servicecomb.version>
   </properties>
 
   <modules>

spring-cloud-huawei-service-engine/service-engine-common/src/main/java/com/huaweicloud/service/engine/common/disovery/ServiceCenterUtils.java

@@ -21,6 +21,7 @@
 import java.util.List;
 import java.util.Map;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.servicecomb.foundation.auth.AuthHeaderProvider;
 import org.apache.servicecomb.http.client.auth.RequestAuthHeaderProvider;
 import org.apache.servicecomb.http.client.common.HttpConfiguration.SSLProperties;
@@ -32,30 +33,56 @@
 import org.springframework.core.env.Environment;
 
 import com.huaweicloud.common.event.EventManager;
+import com.huaweicloud.common.util.URLUtil;
 import com.huaweicloud.service.engine.common.configration.bootstrap.DiscoveryBootstrapProperties;
 import com.huaweicloud.service.engine.common.configration.bootstrap.ServiceCombSSLProperties;
 import com.huaweicloud.service.engine.common.transport.TransportUtils;
-import com.huaweicloud.common.util.URLUtil;
 
 public class ServiceCenterUtils {
   private static final Logger LOGGER = LoggerFactory.getLogger(ServiceCenterUtils.class);
 
+  private static volatile ServiceCenterAddressManager serviceAddressManager;
+
   public static ServiceCenterAddressManager createAddressManager(DiscoveryBootstrapProperties discoveryProperties) {
-    List<String> addresses = URLUtil.dealMultiUrl(discoveryProperties.getAddress());
-    LOGGER.info("initialize discovery server={}", addresses);
-    return new ServiceCenterAddressManager("default", addresses, EventManager.getEventBus());
+    if (serviceAddressManager == null) {
+      synchronized (ServiceCenterUtils.class) {
+        if (serviceAddressManager == null) {
+          List<String> addresses = URLUtil.dealMultiUrl(discoveryProperties.getAddress());
+          LOGGER.info("initialize discovery server={}", addresses);
+          serviceAddressManager = new ServiceCenterAddressManager("default", addresses,
+              getDataCenterRegion(discoveryProperties), getDataCenterZone(discoveryProperties),
+              EventManager.getEventBus());
+        }
+      }
+    }
+    return serviceAddressManager;
+  }
+
+  private static String getDataCenterRegion(DiscoveryBootstrapProperties discoveryProperties) {
+    if (discoveryProperties.getDatacenter() == null) {
+      return "";
+    }
+    String region = discoveryProperties.getDatacenter().getRegion();
+    return StringUtils.isEmpty(region) ? "" : region;
+  }
+
+  private static String getDataCenterZone(DiscoveryBootstrapProperties discoveryProperties) {
+    if (discoveryProperties.getDatacenter() == null) {
+      return "";
+    }
+    String zone = discoveryProperties.getDatacenter().getAvailableZone();
+    return StringUtils.isEmpty(zone) ? "" : zone;
   }
 
   // add other headers needed for registration by new ServiceCenterClient(...)
   public static ServiceCenterClient serviceCenterClient(DiscoveryBootstrapProperties discoveryProperties,
-      ServiceCombSSLProperties serviceCombSSLProperties, List<AuthHeaderProvider> authHeaderProviders,
-      Environment env) {
+      ServiceCombSSLProperties serviceCombSSLProperties, List<AuthHeaderProvider> authHeaderProviders, Environment env) {
     ServiceCenterAddressManager addressManager = createAddressManager(discoveryProperties);
     SSLProperties sslProperties = TransportUtils
         .createSSLProperties(addressManager.sslEnabled(), serviceCombSSLProperties);
     return new ServiceCenterClient(addressManager, sslProperties,
-        getRequestAuthHeaderProvider(authHeaderProviders),
-        "default", new HashMap<>(), env).setEventBus(EventManager.getEventBus());
+        getRequestAuthHeaderProvider(authHeaderProviders), "default", new HashMap<>(), env)
+        .setEventBus(EventManager.getEventBus());
   }
 
   public static ServiceCenterWatch serviceCenterWatch(DiscoveryBootstrapProperties discoveryProperties,
@@ -71,8 +98,9 @@ public static ServiceCenterWatch serviceCenterWatch(DiscoveryBootstrapProperties
 
   private static RequestAuthHeaderProvider getRequestAuthHeaderProvider(List<AuthHeaderProvider> authHeaderProviders) {
     return signRequest -> {
+      String host = signRequest != null && signRequest.getEndpoint() != null ? signRequest.getEndpoint().getHost() : "";
       Map<String, String> headers = new HashMap<>();
-      authHeaderProviders.forEach(provider -> headers.putAll(provider.authHeaders()));
+      authHeaderProviders.forEach(provider -> headers.putAll(provider.authHeaders(host)));
       return headers;
     };
   }

spring-cloud-huawei-service-engine/service-engine-common/src/main/java/com/huaweicloud/service/engine/common/transport/AkSkRequestAuthHeaderProvider.java

@@ -41,7 +41,7 @@ public AkSkRequestAuthHeaderProvider(ServiceCombAkSkProperties serviceCombAkSkPr
   }
 
   @Override
-  public Map<String, String> authHeaders() {
+  public Map<String, String> authHeaders(String host) {
     if (isAKSKNotEnabled(serviceCombAkSkProperties)) {
       return Collections.emptyMap();
     }

spring-cloud-huawei-service-engine/service-engine-common/src/main/java/com/huaweicloud/service/engine/common/transport/RBACRequestAuthHeaderProvider.java

@@ -17,6 +17,7 @@
 
 package com.huaweicloud.service.engine.common.transport;
 
+import java.net.URI;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -26,8 +27,7 @@
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.servicecomb.foundation.auth.AuthHeaderProvider;
-import org.apache.servicecomb.http.client.event.EngineConnectChangedEvent;
-import org.apache.servicecomb.service.center.client.OperationEvents;
+import org.apache.servicecomb.http.client.event.OperationEvents;
 import org.apache.servicecomb.service.center.client.ServiceCenterClient;
 import org.apache.servicecomb.service.center.client.model.RbacTokenRequest;
 import org.apache.servicecomb.service.center.client.model.RbacTokenResponse;
@@ -41,13 +41,12 @@
 import com.google.common.eventbus.Subscribe;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.huaweicloud.common.event.EventManager;
 import com.huaweicloud.service.engine.common.configration.bootstrap.BootstrapProperties;
 import com.huaweicloud.service.engine.common.configration.bootstrap.DiscoveryBootstrapProperties;
-import com.huaweicloud.service.engine.common.configration.bootstrap.MicroserviceProperties;
 import com.huaweicloud.service.engine.common.configration.bootstrap.ServiceCombRBACProperties;
 import com.huaweicloud.service.engine.common.configration.bootstrap.ServiceCombSSLProperties;
 import com.huaweicloud.service.engine.common.disovery.ServiceCenterUtils;
-import com.huaweicloud.common.event.EventManager;
 
 import jakarta.ws.rs.core.Response.Status;
 
@@ -59,37 +58,26 @@ public class RBACRequestAuthHeaderProvider implements AuthHeaderProvider {
   // e.g. not found:  will query token after token expired period
   public static final String INVALID_TOKEN = "invalid";
 
-  private static final String UN_AUTHORIZED_CODE_HALF_OPEN = "401302";
-
   public static final String CACHE_KEY = "token";
 
   public static final String AUTH_HEADER = "Authorization";
 
   private static final long TOKEN_REFRESH_TIME_IN_SECONDS = 20 * 60 * 1000;
 
-  private final DiscoveryBootstrapProperties discoveryProperties;
-
-  private final ServiceCombSSLProperties serviceCombSSLProperties;
+  private static final Object LOCK = new Object();
 
   private final ServiceCombRBACProperties serviceCombRBACProperties;
 
-  private final MicroserviceProperties microserviceProperties;
-
   private ExecutorService executorService;
 
   private LoadingCache<String, String> cache;
 
-  private String lastErrorCode = "401302";
-
-  private int lastStatusCode = 401;
-
   private ServiceCenterClient serviceCenterClient;
 
   public RBACRequestAuthHeaderProvider(BootstrapProperties bootstrapProperties, Environment env) {
-    this.discoveryProperties = bootstrapProperties.getDiscoveryBootstrapProperties();
-    this.serviceCombSSLProperties = bootstrapProperties.getServiceCombSSLProperties();
+    DiscoveryBootstrapProperties discoveryProperties = bootstrapProperties.getDiscoveryBootstrapProperties();
+    ServiceCombSSLProperties serviceCombSSLProperties = bootstrapProperties.getServiceCombSSLProperties();
     this.serviceCombRBACProperties = bootstrapProperties.getServiceCombRBACProperties();
-    this.microserviceProperties = bootstrapProperties.getMicroserviceProperties();
 
     if (enabled()) {
       serviceCenterClient = ServiceCenterUtils.serviceCenterClient(discoveryProperties,
@@ -98,38 +86,41 @@ public RBACRequestAuthHeaderProvider(BootstrapProperties bootstrapProperties, En
 
       executorService = Executors.newFixedThreadPool(1, t -> new Thread(t, "rbac-executor"));
       cache = CacheBuilder.newBuilder()
-          .maximumSize(1)
+          .maximumSize(10)
           .refreshAfterWrite(refreshTime(), TimeUnit.MILLISECONDS)
           .build(new CacheLoader<String, String>() {
             @Override
             public String load(String key) {
-              return createHeaders();
+              return createHeaders(key);
             }
 
             @Override
             public ListenableFuture<String> reload(String key, String oldValue) {
-              return Futures.submit(() -> createHeaders(), executorService);
+              return Futures.submit(() -> createHeaders(key), executorService);
             }
           });
     }
   }
 
   @Subscribe
-  public void onNotPermittedEvent(OperationEvents.UnAuthorizedOperationEvent event) {
-    this.executorService.submit(this::retryRefresh);
+  public void onUnAuthorizedOperationEvent(OperationEvents.UnAuthorizedOperationEvent event) {
+    LOGGER.warn("address {} unAuthorized, refresh cache token!", event.getAddress());
+    cache.refresh(getHostByAddress(event.getAddress()));
   }
 
-  @Subscribe
-  public void onEngineConnectChangedEvent(EngineConnectChangedEvent event) {
-    cache.refresh(CACHE_KEY);
+  private static String getHostByAddress(String address) {
+    try {
+      URI uri = URI.create(address);
+      return uri.getHost();
+    } catch (Exception e) {
+      LOGGER.error("get host by address [{}] error!", address, e);
+      return CACHE_KEY;
+    }
   }
 
-  protected String createHeaders() {
-    LOGGER.info("start to create RBAC headers");
-
-    RbacTokenResponse rbacTokenResponse = callCreateHeaders();
-    lastErrorCode = rbacTokenResponse.getErrorCode();
-    lastStatusCode = rbacTokenResponse.getStatusCode();
+  protected String createHeaders(String key) {
+    LOGGER.info("start to create server [{}] RBAC headers", key);
+    RbacTokenResponse rbacTokenResponse = callCreateHeaders(key);
 
     if (Status.UNAUTHORIZED.getStatusCode() == rbacTokenResponse.getStatusCode()
         || Status.FORBIDDEN.getStatusCode() == rbacTokenResponse.getStatusCode()) {
@@ -140,51 +131,68 @@ protected String createHeaders() {
       // service center not support, do not try
       LOGGER.warn("service center do not support RBAC token, you should not config account info");
       return INVALID_TOKEN;
+    } else if (Status.INTERNAL_SERVER_ERROR.getStatusCode() == rbacTokenResponse.getStatusCode()) {
+      // return null for server_error, so the token information can be re-fetched on the next call.
+      // It will prompt 'CacheLoader returned null for key xxx'
+      LOGGER.warn("service center query RBAC token error!");
+      return null;
     }
 
-    LOGGER.info("refresh token successfully {}", rbacTokenResponse.getStatusCode());
+    LOGGER.info("refresh server [{}] token successfully {}", key, rbacTokenResponse.getStatusCode());
     return rbacTokenResponse.getToken();
   }
 
-  protected RbacTokenResponse callCreateHeaders() {
+  protected RbacTokenResponse callCreateHeaders(String host) {
     RbacTokenRequest request = new RbacTokenRequest();
     request.setName(serviceCombRBACProperties.getName());
     request.setPassword(serviceCombRBACProperties.getPassword());
-
-    return serviceCenterClient.queryToken(request);
+    try {
+      return serviceCenterClient.queryToken(request, host);
+    } catch (Exception e) {
+      LOGGER.error("query token from server [{}] error!", host, e);
+    }
+    RbacTokenResponse response = new RbacTokenResponse();
+    response.setStatusCode(Status.INTERNAL_SERVER_ERROR.getStatusCode());
+    return response;
   }
 
   protected long refreshTime() {
     return TOKEN_REFRESH_TIME_IN_SECONDS;
   }
 
+  /**
+   * Retrieve the corresponding engine token cache information based on the host in the request
+   * to resolve cross-engine authentication issues in dual-engine disaster recovery scenarios.
+   *
+   * @param host host
+   * @return token info
+   */
   @Override
-  public Map<String, String> authHeaders() {
+  public Map<String, String> authHeaders(String host) {
     if (!enabled()) {
       return Collections.emptyMap();
     }
-
-    try {
-      String header = cache.get(CACHE_KEY);
-      if (!StringUtils.isEmpty(header)) {
-        Map<String, String> tokens = new HashMap<>(1);
-        tokens.put(AUTH_HEADER, "Bearer " + header);
-        return tokens;
+    String address = host;
+    if (StringUtils.isEmpty(address)) {
+      address = CACHE_KEY;
+    }
+    synchronized (LOCK) {
+      try {
+        String header = cache.get(address);
+        if (!StringUtils.isEmpty(header)) {
+          Map<String, String> tokens = new HashMap<>(1);
+          tokens.put(AUTH_HEADER, "Bearer " + header);
+          return tokens;
+        }
+      } catch (Exception e) {
+        LOGGER.error("Get auth headers failed", e);
       }
-    } catch (Exception e) {
-      LOGGER.error("Get auth headers failed", e);
+      return Collections.emptyMap();
     }
-    return Collections.emptyMap();
   }
 
   private boolean enabled() {
     return !StringUtils.isEmpty(serviceCombRBACProperties.getName()) && !StringUtils
         .isEmpty(serviceCombRBACProperties.getPassword());
   }
-
-  private void retryRefresh() {
-    if (Status.UNAUTHORIZED.getStatusCode() == lastStatusCode && UN_AUTHORIZED_CODE_HALF_OPEN.equals(lastErrorCode)) {
-      cache.refresh(microserviceProperties.getName());
-    }
-  }
 }