PubSub init retry and additional logs (#451)

* Update pubSubUtil.js

* updates

* Update dataManager.js

* Update dataManager.js

* updates

* Update ds-api-mgr-role-definition.yaml

* Update pubSubUtil.js

Author: mservidio

api/config/ds-api-mgr-role-definition.yaml

@@ -1,6 +1,6 @@
 title: "Datashare Manager Role"
 description: "Custom Datashare API role to interface with GCP services"
-stage: "BETA"
+stage: "GA"
 includedPermissions:
 - bigquery.datasets.create
 - bigquery.datasets.delete
@@ -20,7 +20,6 @@ includedPermissions:
 - bigquery.tables.updateData
 - compute.projects.get
 - iam.serviceAccounts.signBlob
-- pubsub.snapshots.seek
 - pubsub.subscriptions.consume
 - pubsub.subscriptions.create
 - pubsub.subscriptions.get

api/deploy.sh

@@ -61,7 +61,8 @@ gcloud run deploy ds-api \
   --image gcr.io/${PROJECT_ID}/ds-api:${TAG} \
   --platform gke \
   --service-account ${SERVICE_ACCOUNT_NAME} \
-  --update-env-vars=PROJECT_ID="${PROJECT_ID}"
+  --update-env-vars=PROJECT_ID="${PROJECT_ID}" \
+  --use-http2
 
 gcloud run services update-traffic ds-api \
     --cluster $CLUSTER \

api/v1alpha/package-lock.json

@@ -5,6 +5,7 @@
     "license": "Apache-2.0",
     "dependencies": {
         "@hapi/joi": "^16.1.7",
+        "async-retry": "^1.3.1",
         "axios": "^0.21.1",
         "cds-shared": "file:../../shared",
         "compression": "^1.7.4",

api/v1alpha/package.json

@@ -23,6 +23,7 @@ const metaManager = require('../lib/metaManager');
 const datasetManager = require('../datasets/dataManager');
 const procurementManager = require('../procurements/dataManager');
 const fs = require('fs');
+const retry = require('async-retry');
 
 require.extensions['.sql'] = function (module, filename) {
     module.exports = fs.readFileSync(filename, 'utf8');
@@ -241,6 +242,26 @@ async function setupDatasharePrerequisites(projectId) {
     });
 }
 
+/**
+ * Starts the PubSub listener, and retries if launch failure
+ */
+async function startPubSubListener() {
+    function logRetry(error, attempt) {
+        console.warn(`PubSub listener performing retry attempt: ${attempt} after ${error}`);
+    }
+
+    // eslint-disable-next-line no-unused-vars
+    await retry(async bail => {
+        await initializePubSubListener();
+    }, {
+        // retries: 5,
+        forever: true,
+        minTimeout: 30000,
+        maxTimeout: 60000,
+        onRetry: logRetry
+    })
+}
+
 /**
  * Initializes PubSub listener for entitlement auto approvals
  */
@@ -348,6 +369,13 @@ async function initializePubSubListener() {
             let subscription = pubSubUtil.getSubscription(subscriptionName, subscriberOptions);
             subscription.on('message', messageHandler);
             subscription.on('error', errorHandler);
+            subscription.on('close', () => { console.error('Subscription closed') });
+            subscription.detached((cb) => {
+                console.warn('Subscription detached');
+                if (cb) {
+                    console.error(cb);
+                }
+            });
         } catch (err) {
             console.error(err);
         }
@@ -362,6 +390,6 @@ async function initializePubSubListener() {
 module.exports = {
     initializeSchema,
     syncResources,
-    initializePubSubListener
+    startPubSubListener
 };
 

api/v1alpha/src/admin/dataManager.js

@@ -302,11 +302,7 @@ app.listen(PORT, () => {
     console.log("Listening on port " + PORT + ". Press Ctrl+C to quit.");
 });
 
-try {
-    const adminMgr = require('./admin/dataManager');
-    adminMgr.initializePubSubListener();
-} catch (err) {
-    console.error(err);
-}
+const adminMgr = require('./admin/dataManager');
+adminMgr.startPubSubListener();
 
 module.exports = app;

api/v1alpha/src/index.js

@@ -92,14 +92,10 @@ class PubSubUtil {
      * this function will only work if you have the necessary permissions on the topic
      */
     async checkIfSubscriptionExists(topicName, subscriptionProjectId, subscriptionName) {
-        const topic = this.pubsub.topic(topicName);
-        const formattedSubscription = this.client.subscriptionPath(subscriptionProjectId, subscriptionName);
-        const subscription = topic.subscription(formattedSubscription);
+        const formattedSubscription = `projects/${subscriptionProjectId}/subscriptions/${subscriptionName}`;
+        const subscription = this.getSubscription(formattedSubscription);
         const response = await subscription.exists();
         const exists = response[0];
-        if (this.VERBOSE_MODE) {
-            console.log(`Checking if subscription exists: '${topicName}': ${formattedSubscription}`);
-        }
         return exists;
     }