Implement recall mechanism for gateway data refreshing

components/apimgt/org.wso2.carbon.apimgt.api/src/main/java/org/wso2/carbon/apimgt/api/gateway/GatewayAPIDTO.java

@@ -36,6 +36,7 @@ public class GatewayAPIDTO implements Serializable {
     private String provider;
     private String tenantDomain;
     private String apiId;
+    private String revision;
     private String apiContext;
     private String apiDefinition;
     private String graphQLSchema;
@@ -235,4 +236,12 @@ public String getApiContext() {
     public void setApiContext(String apiContext) {
         this.apiContext = apiContext;
     }
+
+    public String getRevision() {
+        return revision;
+    }
+
+    public void setRevision(String revision) {
+        this.revision = revision;
+    }
 }

components/apimgt/org.wso2.carbon.apimgt.common.jms/pom.xml

@@ -38,6 +38,10 @@
             <groupId>org.wso2.carbon.apimgt</groupId>
             <artifactId>org.wso2.carbon.apimgt.impl</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.wso2.carbon.apimgt</groupId>
+            <artifactId>org.wso2.carbon.apimgt.keymgt</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

components/apimgt/org.wso2.carbon.apimgt.common.jms/src/main/java/org/wso2/carbon/apimgt/common/jms/JMSConnectionEventListener.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright (c) 2023, WSO2 LLC. (http://www.wso2.org) 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 org.wso2.carbon.apimgt.common.jms;
+
+/**
+ * This interface is used to listen to JMS connection events.
+ */
+public interface JMSConnectionEventListener {
+
+    /**
+     * This method will be called when the JMS connection is re-established after a disconnection.
+     */
+    void onReconnect();
+
+    /**
+     * This method will be called when the JMS connection is disconnected.
+     */
+    void onDisconnect();
+}

components/apimgt/org.wso2.carbon.apimgt.common.jms/src/main/java/org/wso2/carbon/apimgt/common/jms/JMSTaskManager.java

@@ -743,6 +743,10 @@
             setJmsTaskManagerState(STATE_FAILURE);
             log.error("JMS Connection failed : " + j.getMessage() + " - shutting down worker tasks");
 
+            if (messageListener instanceof JMSConnectionEventListener) {
+                ((JMSConnectionEventListener) messageListener).onDisconnect();
+            }
+
             int r = 1;
 
             long retryDuration = initialReconnectDuration;
@@ -782,6 +786,10 @@
                     isOnExceptionError = false;
                     log.info("Reconnection attempt: " + r + " for " + jmsConsumerName +
                             " was successful!");
+
+                    if (messageListener instanceof JMSConnectionEventListener) {
+                        ((JMSConnectionEventListener) messageListener).onReconnect();
+                    }
                 }
 
 

components/apimgt/org.wso2.carbon.apimgt.gateway/src/main/java/org/wso2/carbon/apimgt/gateway/InMemoryAPIDeployer.java

@@ -33,6 +33,7 @@
 import org.apache.synapse.transport.dynamicconfigurations.DynamicProfileReloaderHolder;
 import org.wso2.carbon.apimgt.api.APIManagementException;
 import org.wso2.carbon.apimgt.api.ExceptionCodes;
+import org.wso2.carbon.apimgt.api.APIManagementException;
 import org.wso2.carbon.apimgt.api.gateway.GatewayAPIDTO;
 import org.wso2.carbon.apimgt.api.gateway.GatewayContentDTO;
 import org.wso2.carbon.apimgt.api.gateway.GraphQLSchemaDTO;
@@ -60,6 +61,7 @@
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.UUID;
@@ -185,6 +187,11 @@
         return result;
     }
 
+    public boolean deployAllAPIsAtGatewayStartup(Set<String> assignedGatewayLabels, String tenantDomain)
+            throws ArtifactSynchronizerException {
+        return deployAllAPIs(assignedGatewayLabels, tenantDomain, false);
+    }
+
     /**
      * Deploy an API in the gateway using the deployAPI method in gateway admin.
      *
@@ -193,43 +200,70 @@
      * @return True if all API artifacts retrieved from the storage and successfully deployed without any error. else
      * false
      */
-    public boolean deployAllAPIsAtGatewayStartup(Set<String> assignedGatewayLabels, String tenantDomain)
-            throws ArtifactSynchronizerException {
+    public boolean deployAllAPIs(Set<String> assignedGatewayLabels, String tenantDomain,
+                                                 boolean redeployChangedAPIs) throws ArtifactSynchronizerException {
 
         boolean result = false;
-        try {
-            deployJWKSSynapseAPI(tenantDomain); // Deploy JWKS API
-        } catch (APIManagementException e) {
-            log.error("Error while deploying JWKS API for tenant domain :" + tenantDomain, e);
-        }
+        Map<String, org.wso2.carbon.apimgt.keymgt.model.entity.API> apiMap = null;
 
+        if (!redeployChangedAPIs) {
+            try {
+                deployJWKSSynapseAPI(tenantDomain); // Deploy JWKS API
+            } catch (APIManagementException e) {
+                log.error("Error while deploying JWKS API for tenant domain :" + tenantDomain, e);
+            }
+        }
         if (gatewayArtifactSynchronizerProperties.isRetrieveFromStorageEnabled()) {
             if (artifactRetriever != null) {
                 try {
                     int errorCount = 0;
                     String labelString = String.join("|", assignedGatewayLabels);
                     String encodedString = Base64.encodeBase64URLSafeString(labelString.getBytes());
+
                     APIGatewayAdmin apiGatewayAdmin = new APIGatewayAdmin();
-                    MessageContext.setCurrentMessageContext(org.wso2.carbon.apimgt.gateway.utils.GatewayUtils.createAxis2MessageContext());
+                    MessageContext.setCurrentMessageContext(
+                                    org.wso2.carbon.apimgt.gateway.utils.GatewayUtils.createAxis2MessageContext());
                     PrivilegedCarbonContext.startTenantFlow();
-                    PrivilegedCarbonContext.getThreadLocalCarbonContext().setTenantDomain(tenantDomain, true);
+                    PrivilegedCarbonContext.getThreadLocalCarbonContext()
+                            .setTenantDomain(tenantDomain, true);
                     List<String> gatewayRuntimeArtifacts = ServiceReferenceHolder.getInstance().getArtifactRetriever()
                             .retrieveAllArtifacts(encodedString, tenantDomain);
-                    if (gatewayRuntimeArtifacts.size() == 0) {
+                    if (gatewayRuntimeArtifacts.isEmpty()) {
                         return true;
                     }
+                    if (redeployChangedAPIs) {
+                        DataHolder dataHolder = DataHolder.getInstance();
+                        apiMap = dataHolder.getTenantAPIMap().get(tenantDomain);
+                    }
                     for (String runtimeArtifact : gatewayRuntimeArtifacts) {
                         GatewayAPIDTO gatewayAPIDTO = null;
                         try {
                             if (StringUtils.isNotEmpty(runtimeArtifact)) {
                                 gatewayAPIDTO = new Gson().fromJson(runtimeArtifact, GatewayAPIDTO.class);
-                                log.info("Deploying synapse artifacts of " + gatewayAPIDTO.getName());
-                                apiGatewayAdmin.deployAPI(gatewayAPIDTO);
-                                addDeployedCertificatesToAPIAssociation(gatewayAPIDTO);
-                                addDeployedGraphqlQLToAPI(gatewayAPIDTO);
-                                DataHolder.getInstance().addKeyManagerToAPIMapping(gatewayAPIDTO.getApiId(),
-                                        gatewayAPIDTO.getKeyManagers());
-                                DataHolder.getInstance().markAPIAsDeployed(gatewayAPIDTO);
+                                if (redeployChangedAPIs && apiMap != null) {
+                                    org.wso2.carbon.apimgt.keymgt.model.entity.API api =
+                                            apiMap.get(gatewayAPIDTO.getApiContext());
+                                    // Here, we redeploy APIs only if there is a new revision deployed in the
+                                    // Control Plane and not synced with the gateway due to connection issues.
+                                    if (api != null && api.getRevisionId() != null &&
+                                            (!api.getRevisionId().equalsIgnoreCase(gatewayAPIDTO.getRevision()))) {
+                                        DeployAPIInGatewayEvent deployAPIInGatewayEvent =
+                                                new DeployAPIInGatewayEvent(UUID.randomUUID().toString(),
+                                                        System.currentTimeMillis(),
+                                                        APIConstants.EventType.REMOVE_API_FROM_GATEWAY.name(),
+                                                        tenantDomain, api.getApiId(), api.getUuid(),
+                                                        assignedGatewayLabels, api.getName(), api.getVersion(),
+                                                        api.getApiProvider(), api.getApiType(), api.getContext());
+                                        unDeployAPI(deployAPIInGatewayEvent);
+                                        deployAPIFromDTO(gatewayAPIDTO, apiGatewayAdmin);
+                                    } else {
+                                        if (log.isDebugEnabled()) {
+                                            log.debug("API " + gatewayAPIDTO.getName() + " is already deployed");
+                                        }
+                                    }
+                                } else {
+                                    deployAPIFromDTO(gatewayAPIDTO, apiGatewayAdmin);
+                                }
                             }
                         } catch (AxisFault axisFault) {
                             log.error("Error in deploying " + gatewayAPIDTO.getName() + " to the Gateway ", axisFault);
@@ -264,6 +298,18 @@
         return result;
     }
 
+    private void deployAPIFromDTO(GatewayAPIDTO gatewayAPIDTO, APIGatewayAdmin apiGatewayAdmin) throws AxisFault {
+        log.info("Deploying synapse artifacts of API ID: " + gatewayAPIDTO.getApiId() +
+                " and Context: " + gatewayAPIDTO.getApiContext());
+        apiGatewayAdmin.deployAPI(gatewayAPIDTO);
+        addDeployedCertificatesToAPIAssociation(gatewayAPIDTO);
+        addDeployedGraphqlQLToAPI(gatewayAPIDTO);
+        DataHolder.getInstance().addKeyManagerToAPIMapping(gatewayAPIDTO.getApiId(),
+                gatewayAPIDTO.getKeyManagers());
+        DataHolder.getInstance().markAPIAsDeployed(gatewayAPIDTO);
+    }
+
+
     private void unDeployAPI(APIGatewayAdmin apiGatewayAdmin, DeployAPIInGatewayEvent gatewayEvent)
             throws AxisFault {
             if (gatewayArtifactSynchronizerProperties.isRetrieveFromStorageEnabled()) {

components/apimgt/org.wso2.carbon.apimgt.gateway/src/main/java/org/wso2/carbon/apimgt/gateway/internal/DataHolder.java

@@ -164,6 +164,7 @@ public void markAPIAsDeployed(GatewayAPIDTO gatewayAPIDTO) {
                 if (log.isDebugEnabled()) {
                     log.debug("API : " + api.getApiName() + "is deployed successfully");
                 }
+                api.setRevisionId(gatewayAPIDTO.getRevision());
             }
         }
     }

components/apimgt/org.wso2.carbon.apimgt.gateway/src/main/java/org/wso2/carbon/apimgt/gateway/internal/ServiceReferenceHolder.java

@@ -395,6 +395,10 @@
         return activeTenants.contains(tenantDomain);
     }
 
+    public Set<String> getActiveTenants() {
+        return activeTenants;
+    }
+
     public void setRedisCacheUtil(RedisCacheUtils redisCacheUtils) {
 
     }

components/apimgt/org.wso2.carbon.apimgt.gateway/src/main/java/org/wso2/carbon/apimgt/gateway/listeners/GatewayJMSMessageListener.java

@@ -27,6 +27,7 @@
 import org.apache.commons.logging.LogFactory;
 import org.wso2.carbon.apimgt.api.APIManagementException;
 import org.wso2.carbon.apimgt.api.model.APIStatus;
+import org.wso2.carbon.apimgt.common.jms.JMSConnectionEventListener;
 import org.wso2.carbon.apimgt.gateway.APILoggerManager;
 import org.wso2.carbon.apimgt.gateway.EndpointCertificateDeployer;
 import org.wso2.carbon.apimgt.gateway.GoogleAnalyticsConfigDeployer;
@@ -55,6 +56,7 @@
 import org.wso2.carbon.apimgt.impl.notifier.events.SubscriptionPolicyEvent;
 import org.wso2.carbon.apimgt.impl.notifier.events.KeyTemplateEvent;
 import org.wso2.carbon.apimgt.impl.utils.APIUtil;
+import org.wso2.carbon.apimgt.keymgt.SubscriptionDataHolder;
 import org.wso2.carbon.context.PrivilegedCarbonContext;
 
 import java.util.HashSet;
@@ -67,17 +69,25 @@
 import javax.jms.TextMessage;
 import javax.jms.Topic;
 
-public class GatewayJMSMessageListener implements MessageListener {
+public class GatewayJMSMessageListener implements MessageListener, JMSConnectionEventListener {
 
     private static final Log log = LogFactory.getLog(GatewayJMSMessageListener.class);
     private boolean debugEnabled = log.isDebugEnabled();
+    private boolean refreshOnReconnect = false;
     private InMemoryAPIDeployer inMemoryApiDeployer = new InMemoryAPIDeployer();
     private EventHubConfigurationDto eventHubConfigurationDto = ServiceReferenceHolder.getInstance()
             .getAPIManagerConfiguration().getEventHubConfigurationDto();
     private GatewayArtifactSynchronizerProperties gatewayArtifactSynchronizerProperties = ServiceReferenceHolder
             .getInstance().getAPIManagerConfiguration().getGatewayArtifactSynchronizerProperties();
     ExecutorService executor = Executors.newSingleThreadExecutor(r -> new Thread(r, "DeploymentThread"));
 
+    public GatewayJMSMessageListener() {
+    }
+
+    public GatewayJMSMessageListener(boolean refreshOnReconnect) {
+        this.refreshOnReconnect = refreshOnReconnect;
+    }
+
     public void onMessage(Message message) {
 
         try {
@@ -410,4 +420,43 @@
         ServiceReferenceHolder.getInstance().getSubscriptionsDataService()
                 .removeSubscription(apiKey, topicName, tenantDomain, subscriber);
     }
+
+    @Override
+    public void onReconnect() {
+        if (refreshOnReconnect) {
+            log.info("Refreshing gateway data stores and deployments.");
+            new Thread(() -> {
+                synchronized (this) {
+                    SubscriptionDataHolder.getInstance().refreshSubscriptionStore();
+                    redeployGatewayArtifacts();
+                }
+            }).start();
+        }
+    }
+
+    private void redeployGatewayArtifacts() {
+        Set<String> activeTenants = ServiceReferenceHolder.getInstance().getActiveTenants();
+        activeTenants.forEach(tenantDomain -> {
+            try {
+                new EndpointCertificateDeployer(tenantDomain).deployCertificatesAtStartup();
+                if (log.isDebugEnabled()) {
+                    log.debug("Redeploying artifacts for tenant: " + tenantDomain);
+                }
+                inMemoryApiDeployer.
+                        deployAllAPIs(gatewayArtifactSynchronizerProperties.getGatewayLabels(),
+                        tenantDomain, true);
+            } catch (ArtifactSynchronizerException e) {
+                log.error("Error while redeploying gateway artifacts for tenant: " + tenantDomain, e);
+            } catch (APIManagementException e) {
+                log.error("Error while redeploying endpoint certificates for tenant: " + tenantDomain, e);
+            }
+        });
+
+    }
+
+    @Override
+    public void onDisconnect() {
+        // We currently do not have any logic to execute for this scenario.
+        // Added in case we need to implement an operation in the future.
+    }
 }

components/apimgt/org.wso2.carbon.apimgt.gateway/src/main/java/org/wso2/carbon/apimgt/gateway/listeners/GatewayStartupListener.java

@@ -167,12 +167,12 @@ public void completedServerStartup() {
                 }
             }).start();
             SubscriptionDataHolder.getInstance().registerTenantSubscriptionStore(MultitenantConstants.SUPER_TENANT_DOMAIN_NAME);
+            try {
+                retrieveAllAPIMetadata();
+            } catch (DataLoadingException e) {
+                log.error("Error while loading All API Metadata", e);
+            }
             if (GatewayUtils.isOnDemandLoading()) {
-                try {
-                    retrieveAllAPIMetadata();
-                } catch (DataLoadingException e) {
-                    log.error("Error while loading All API Metadata", e);
-                }
                 try {
                     new EndpointCertificateDeployer().deployAllCertificatesAtStartup();
                 } catch (APIManagementException e) {
@@ -190,7 +190,8 @@ public void completedServerStartup() {
             jmsTransportHandlerForEventHub.subscribeForJmsEvents(APIConstants.TopicNames.TOPIC_CACHE_INVALIDATION,
                     new APIMgtGatewayCacheMessageListener());
             jmsTransportHandlerForEventHub
-                    .subscribeForJmsEvents(APIConstants.TopicNames.TOPIC_NOTIFICATION, new GatewayJMSMessageListener());
+                    .subscribeForJmsEvents(APIConstants.TopicNames.TOPIC_NOTIFICATION,
+                            new GatewayJMSMessageListener(true));
             jmsTransportHandlerForEventHub
                     .subscribeForJmsEvents(APIConstants.TopicNames.TOPIC_THROTTLE_DATA, new JMSMessageListener());
             jmsTransportHandlerForEventHub.subscribeForJmsEvents(APIConstants.TopicNames.TOPIC_ASYNC_WEBHOOKS_DATA,

components/apimgt/org.wso2.carbon.apimgt.keymgt/src/main/java/org/wso2/carbon/apimgt/keymgt/SubscriptionDataHolder.java

@@ -17,6 +17,8 @@
  */
 package org.wso2.carbon.apimgt.keymgt;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.wso2.carbon.apimgt.keymgt.model.SubscriptionDataStore;
 import org.wso2.carbon.apimgt.keymgt.model.impl.SubscriptionDataStoreImpl;
 
@@ -30,6 +32,7 @@
 
     protected Map<String, SubscriptionDataStore> subscriptionStore =
             new ConcurrentHashMap<>();
+    private static final Log log = LogFactory.getLog(SubscriptionDataHolder.class);
     private static SubscriptionDataHolder instance = new SubscriptionDataHolder();
 
     public static SubscriptionDataHolder getInstance() {
@@ -80,4 +83,15 @@
         return null;
     }
 
+    public void refreshSubscriptionStore() {
+        subscriptionStore.keySet().forEach(tenant -> {
+            // Cleaning the existing SubscriptionDataStore instance before re-population
+            subscriptionStore.put(tenant, new SubscriptionDataStoreImpl(tenant));
+            if (log.isDebugEnabled()) {
+                log.debug("Refreshing subscription data store for tenant: " + tenant);
+            }
+            initializeSubscriptionStore(tenant);
+        });
+    }
+
 }