Upgrade Azure Storage SDK to a modern version

config/cdConfig.js

@@ -5,7 +5,9 @@ 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'),
+  spnAuth: config.get('CRAWLER_AZBLOB_SPN_AUTH')
 }
 
 const githubToken = config.get('CRAWLER_GITHUB_TOKEN')
@@ -111,7 +113,9 @@ module.exports = {
     },
     azqueue: {
       connectionString: cd_azblob.connection,
-      queueName: config.get('CRAWLER_HARVESTS_QUEUE_NAME') || 'harvests'
+      account: cd_azblob.account,
+      queueName: config.get('CRAWLER_HARVESTS_QUEUE_NAME') || 'harvests',
+      spnAuth: config.get('CRAWLER_HARVESTS_QUEUE_SPN_AUTH')
     },
     'cd(azblob)': cd_azblob,
     'cd(file)': cd_file
@@ -135,7 +139,9 @@ module.exports = {
       maxDequeueCount: 5,
       attenuation: {
         ttl: 3000
-      }
+      },
+      spnAuth: config.get('CRAWLER_HARVESTS_QUEUE_SPN_AUTH'),
+      account: cd_azblob.account
     },
     appVersion: config.get('APP_VERSION'),
     buildsha: config.get('BUILD_SHA')

ghcrawler/providers/queuing/storageQueue.js

@@ -1,50 +1,46 @@
 // Copyright (c) Microsoft Corporation and others. Made available under the MIT license.
 // SPDX-License-Identifier: MIT
 
+// eslint-disable-next-line no-unused-vars
+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) {
+  async push(requests) {
     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 +52,51 @@ 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')
+      try {
+        await this.queueClient.deleteMessage(message.messageId, message.popReceipt)
+      } catch (error) {
+        this.logger.error(`Failed to delete message ${message.messageId} in storageQueue, error: ${error.message}`)
+        throw error
+      }
+      return null
+    } else {
+      try {
+        const decodedText = message.messageText
+          .replace(/"/g, '"')
+          .replace(/&/g, '&')
+          .replace(/'/g, "'")
+          .replace(/&lt;/g, '<')
+          .replace(/&gt;/g, '>')
+        message.body = JSON.parse(decodedText)
+      } catch (error) {
+        this.logger.error(`Failed to parse message ${message.messageId}:`)
+        this.logger.error(`Raw message: ${message.messageText}`)
+        this.logger.error(`Parse error: ${error.message}`)
+        await this.queueClient.deleteMessage(message.messageId, message.popReceipt)
+        return null
+      }
+      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 +110,30 @@ 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) {
+    const response = await this.queueClient.updateMessage(
+      request._message.messageId,
+      request._message.popReceipt,
+      undefined,
+      visibilityTimeout
+    )
+    this._log('NAKed', request._message.body)
+    return this._buildMessageReceipt(response, request)
   }
 
   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 })
-      })
-    })
+    try {
+      const properties = await this.queueClient.getProperties()
+      return { count: properties.approximateMessagesCount }
+    } catch (error) {
+      this.logger.error(error)
+      return null
+    }
   }
 
   getName() {

ghcrawler/providers/queuing/storageQueueManager.js

@@ -2,14 +2,35 @@
 // SPDX-License-Identifier: MIT
 
 const AttenuatedQueue = require('./attenuatedQueue')
-const AzureStorage = require('azure-storage')
+const { QueueServiceClient, StorageRetryPolicyType } = require('@azure/storage-queue')
 const Request = require('../../lib/request')
 const StorageQueue = require('./storageQueue')
+const { DefaultAzureCredential, ClientSecretCredential } = 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 {
+      const { account, spnAuth } = options
+      let credential
+      if (spnAuth) {
+        const authParsed = JSON.parse(spnAuth)
+        credential = new ClientSecretCredential(authParsed.tenantId, authParsed.clientId, authParsed.clientSecret)
+      } else {
+        credential = new DefaultAzureCredential()
+      }
+      this.client = new QueueServiceClient(`https://${account}.queue.core.windows.net`, credential, pipelineOptions)
+    }
   }
 
   createQueueClient(name, formatter, options) {

ghcrawler/providers/storage/azureBlobFactory.js

@@ -1,15 +1,51 @@
 // 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, ClientSecretCredential } = require('@azure/identity')
 
+/**
+ * @param {object} options
+ * @param {string} options.account
+ * @param {string} options.connection
+ * @param {string} options.container
+ * @param {object} options.logger
+ * @param {object} options.spnAuth
+ */
 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, spnAuth } = 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 {
+    let credential
+    if (spnAuth) {
+      const authParsed = JSON.parse(spnAuth)
+      credential = new ClientSecretCredential(authParsed.tenantId, authParsed.clientId, authParsed.clientSecret)
+      options.logger.info('using service principal credentials')
+    } else {
+      credential = new DefaultAzureCredential()
+      options.logger.info('using default credentials')
+    }
+    blobServiceClient = new BlobServiceClient(`https://${account}.blob.core.windows.net`, credential, pipelineOptions)
+  }
+
+  const containerClient = blobServiceClient.getContainerClient(container)
+
+  return new AzureStorageDocStore(containerClient, options)
 }