Upgrade Azure Storage SDK to a modern version

config/cdConfig.js

@@ -5,7 +5,8 @@ const config = require('painless-config')
 
 const cd_azblob = {
   connection: config.get('CRAWLER_AZBLOB_CONNECTION_STRING'),
-  container: config.get('CRAWLER_AZBLOB_CONTAINER_NAME')
+  container: config.get('CRAWLER_AZBLOB_CONTAINER_NAME'),
+  account: config.get('CRAWLER_AZBLOB_ACCOUNT_NAME'),
 }
 
 const githubToken = config.get('CRAWLER_GITHUB_TOKEN')
@@ -110,6 +111,7 @@ module.exports = {
     },
     azqueue: {
       connectionString: cd_azblob.connection,
+      account: cd_azblob.account,
       queueName: config.get('CRAWLER_HARVESTS_QUEUE_NAME') || 'harvests'
     },
     'cd(azblob)': cd_azblob,

ghcrawler/providers/queuing/storageQueue.js

@@ -1,52 +1,46 @@
 // Copyright (c) Microsoft Corporation and others. Made available under the MIT license.
 // SPDX-License-Identifier: MIT
 
+const { QueueServiceClient } = require('@azure/storage-queue')
 const qlimit = require('qlimit')
 const { cloneDeep } = require('lodash')
 
 class StorageQueue {
+  /**
+   * @param {QueueServiceClient} client
+   * @param {string} name
+   * @param {string} queueName
+   * @param {object} formatter
+   * @param {object} options
+   */
   constructor(client, name, queueName, formatter, options) {
-    this.client = client
     this.name = name
     this.queueName = queueName
     this.messageFormatter = formatter
     this.options = options
     this.logger = options.logger
+    this.queueClient = client.getQueueClient(this.queueName)
   }
 
   async subscribe() {
-    return new Promise((resolve, reject) => {
-      this.client.createQueueIfNotExists(this.queueName, error => {
-        if (error) {
-          return reject(error)
-        }
-        this.logger.info(`Subscribed to ${this.queueName} using Queue Storage`)
-        resolve()
-      })
-    })
+    await this.queueClient.createIfNotExists()
+    this.logger.info(`Subscribed to ${this.queueName} using Queue Storage`)
   }
 
   async unsubscribe() {
-    return
+    // No specific unsubscribe logic for Azure Queue Storage
   }
 
   async push(requests, option) {
     requests = Array.isArray(requests) ? requests : [requests]
     return Promise.all(
       requests.map(
-        qlimit(this.options.parallelPush || 1)(request => {
+        qlimit(this.options.parallelPush || 1)(async request => {
           const body = JSON.stringify(request)
-          return new Promise((resolve, reject) => {
-            this.client.createMessage(this.queueName, body, option, (error, queueMessageResult) => {
-              if (error) {
-                return reject(error)
-              }
-              this._log('Queued', request)
-              resolve(this._buildMessageReceipt(queueMessageResult, request))
-            })
-          })
-        })
-      )
+          const queueMessageResult = await this.queueClient.sendMessage(body)
+          this._log('Queued', request)
+          return this._buildMessageReceipt(queueMessageResult, request)
+        }))
     )
   }
 
@@ -56,47 +50,32 @@ class StorageQueue {
   }
 
   async pop() {
-    const msgOptions = { numOfMessages: 1, visibilityTimeout: this.options.visibilityTimeout || 60 * 60 }
-    return new Promise((resolve, reject) => {
-      this.client.getMessages(this.queueName, msgOptions, (error, result) => {
-        if (error) {
-          return reject(error)
-        }
-        const message = result[0]
-        if (!message) {
-          this.logger.verbose('No messages to receive')
-          return resolve(null)
-        }
-        if (this.options.maxDequeueCount && message.dequeueCount > this.options.maxDequeueCount) {
-          this.logger.verbose('maxDequeueCount exceeded')
-          this.client.deleteMessage(this.queueName, message.messageId, message.popReceipt, error => {
-            if (error) return reject(error)
-            resolve(null)
-          })
-        } else {
-          message.body = JSON.parse(message.messageText)
-          const request = this.messageFormatter(message)
-          request._message = message
-          this._log('Popped', message.body)
-          resolve(request)
-        }
-      })
-    })
+    const msgOptions = { numberOfMessages: 1, visibilityTimeout: this.options.visibilityTimeout || 60 * 60 }
+    const response = await this.queueClient.receiveMessages(msgOptions)
+    const message = response.receivedMessageItems[0]
+    if (!message) {
+      this.logger.verbose('No messages to receive')
+      return null
+    }
+    if (this.options.maxDequeueCount && message.dequeueCount > this.options.maxDequeueCount) {
+      this.logger.verbose('maxDequeueCount exceeded')
+      await this.queueClient.deleteMessage(message.messageId, message.popReceipt)
+      return null
+    } else {
+      message.body = JSON.parse(message.messageText)
+      const request = this.messageFormatter(message)
+      request._message = message
+      this._log('Popped', message.body)
+      return request
+    }
   }
 
   async done(request) {
     if (!request || !request._message) {
       return
     }
-    return new Promise((resolve, reject) => {
-      this.client.deleteMessage(this.queueName, request._message.messageId, request._message.popReceipt, error => {
-        if (error) {
-          return reject(error)
-        }
-        this._log('ACKed', request._message.body)
-        resolve()
-      })
-    })
+    await this.queueClient.deleteMessage(request._message.messageId, request._message.popReceipt)
+    this._log('ACKed', request._message.body)
   }
 
   async defer(request) {
@@ -110,47 +89,21 @@ class StorageQueue {
     await this.updateVisibilityTimeout(request)
   }
 
-  updateVisibilityTimeout(request, visibilityTimeout = 0) {
-    return new Promise((resolve, reject) => {
-      // visibilityTimeout is updated to 0 to unlock/unlease the message
-      this.client.updateMessage(
-        this.queueName,
-        request._message.messageId,
-        request._message.popReceipt,
-        visibilityTimeout,
-        (error, result) => {
-          if (error) {
-            return reject(error)
-          }
-          this._log('NAKed', request._message.body)
-          resolve(this._buildMessageReceipt(result, request._message.body))
-        }
-      )
+  async updateVisibilityTimeout(request, visibilityTimeout = 0) {
+    await this.queueClient.updateMessage(request._message.messageId, request._message.popReceipt, {
+      visibilityTimeout
     })
+    this._log('NAKed', request._message.body)
   }
 
   async flush() {
-    return new Promise((resolve, reject) => {
-      this.client.deleteQueue(this.queueName, error => {
-        if (error) return reject(error)
-        this.client.createQueueIfNotExists(this.queueName, error => {
-          if (error) return reject(error)
-          resolve()
-        })
-      })
-    })
+    await this.queueClient.clearMessages()
+    this.logger.info(`Flushed all messages from ${this.queueName}`)
   }
 
   async getInfo() {
-    return new Promise(resolve => {
-      this.client.getQueueMetadata(this.queueName, (result, error) => {
-        if (error) {
-          this.logger.error(error)
-          resolve(null)
-        }
-        resolve({ count: result[0].approximateMessageCount })
-      })
-    })
+    const properties = await this.queueClient.getProperties()
+    return { count: properties.approximateMessagesCount }
   }
 
   getName() {

ghcrawler/providers/queuing/storageQueueManager.js

@@ -2,14 +2,31 @@
 // SPDX-License-Identifier: MIT
 
 const AttenuatedQueue = require('./attenuatedQueue')
-const AzureStorage = require('azure-storage')
+const { QueueServiceClient } = require('@azure/storage-queue')
 const Request = require('../../lib/request')
 const StorageQueue = require('./storageQueue')
+const { DefaultAzureCredential } = require('@azure/identity')
 
 class StorageQueueManager {
-  constructor(connectionString) {
-    const retryOperations = new AzureStorage.ExponentialRetryPolicyFilter()
-    this.client = AzureStorage.createQueueService(connectionString).withFilter(retryOperations)
+  constructor(connectionString, options) {
+    const pipelineOptions = {
+      retryOptions: {
+        maxTries: 3,
+        retryDelayInMs: 1000,
+        maxRetryDelayInMs: 120 * 1000,
+        tryTimeoutInMs: 30000,
+        retryPolicyType: StorageRetryPolicyType.EXPONENTIAL
+      }
+    }
+    if (connectionString) {
+      this.client = QueueServiceClient.fromConnectionString(connectionString, pipelineOptions)
+    } else {
+      this.client = new QueueServiceClient(
+        `https://${options.account}.queue.core.windows.net`,
+        new DefaultAzureCredential(),
+        pipelineOptions
+      )
+    }
   }
 
   createQueueClient(name, formatter, options) {

ghcrawler/providers/storage/azureBlobFactory.js

@@ -1,15 +1,44 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // SPDX-License-Identifier: MIT
 
-const AzureStorage = require('azure-storage')
+// @ts-check
+const { BlobServiceClient, StorageRetryPolicyType } = require('@azure/storage-blob')
 const AzureStorageDocStore = require('./storageDocStore')
+const { DefaultAzureCredential } = require ('@azure/identity');
 
+/**
+ * @param {object} options
+ * @param {string} options.account
+ * @param {string} options.connection
+ * @param {string} options.container
+ * @param {object} options.logger
+ */
 module.exports = options => {
   options.logger.info('creating azure storage store')
-  const { account, key, connection, container } = options
-  const retryOperations = new AzureStorage.ExponentialRetryPolicyFilter()
-  const blobService = connection
-    ? AzureStorage.createBlobService(connection).withFilter(retryOperations)
-    : AzureStorage.createBlobService(account, key).withFilter(retryOperations)
-  return new AzureStorageDocStore(blobService, container, options)
+  const { account, connection, container } = options
+
+  const pipelineOptions = {
+    retryOptions: {
+      maxTries: 3,
+      retryDelayInMs: 1000,
+      maxRetryDelayInMs: 120 * 1000,
+      tryTimeoutInMs: 30000,
+      retryPolicyType: StorageRetryPolicyType.EXPONENTIAL
+    }
+  }
+
+  let blobServiceClient
+  if (connection) {
+    options.logger.info('using connection string')
+    blobServiceClient = BlobServiceClient.fromConnectionString(connection, pipelineOptions)
+  } else if (account) {
+    options.logger.info('using default credentials')
+    blobServiceClient = new BlobServiceClient(`https://${account}.blob.core.windows.net`, new DefaultAzureCredential(), pipelineOptions)
+  } else {
+    throw new Error('either connection or account must be provided')
+  }
+
+  const containerClient = blobServiceClient.getContainerClient(container)
+
+  return new AzureStorageDocStore(containerClient, options)
 }