[Artifacts] zip creation + blob storage upload functionality (#1488)

* Artifact zip creation + blob storage upload functionality

* Fix lint

* PR feedback

packages/artifact/src/internal/shared/config.ts

@@ -1,3 +1,9 @@
+// Used for controlling the highWaterMark value of the zip that is being streamed
+// The same value is used as the chunk size that is use during upload to blob storage
+export function getUploadChunkSize(): number {
+  return 8 * 1024 * 1024 // 8 MB Chunks
+}
+
 export function getRuntimeToken(): string {
   const token = process.env['ACTIONS_RUNTIME_TOKEN']
   if (!token) {

packages/artifact/src/internal/upload/blob-upload.ts

@@ -0,0 +1,73 @@
+import {BlobClient, BlockBlobUploadStreamOptions} from '@azure/storage-blob'
+import {TransferProgressEvent} from '@azure/core-http'
+import {ZipUploadStream} from './zip'
+import {getUploadChunkSize} from '../shared/config'
+import * as core from '@actions/core'
+
+export interface BlobUploadResponse {
+  /**
+   * If the upload was successful or not
+   */
+  isSuccess: boolean
+
+  /**
+   * The total reported upload size in bytes. Empty if the upload failed
+   */
+  uploadSize?: number
+}
+
+export async function uploadZipToBlobStorage(
+  authenticatedUploadURL: string,
+  zipUploadStream: ZipUploadStream
+): Promise<BlobUploadResponse> {
+  let uploadByteCount = 0
+
+  const maxBuffers = 5
+  const bufferSize = getUploadChunkSize()
+  const blobClient = new BlobClient(authenticatedUploadURL)
+  const blockBlobClient = blobClient.getBlockBlobClient()
+
+  core.debug(
+    `Uploading artifact zip to blob storage with maxBuffers: ${maxBuffers}, bufferSize: ${bufferSize}`
+  )
+
+  const uploadCallback = (progress: TransferProgressEvent): void => {
+    core.info(`Uploaded bytes ${progress.loadedBytes}`)
+    uploadByteCount = progress.loadedBytes
+  }
+
+  const options: BlockBlobUploadStreamOptions = {
+    blobHTTPHeaders: {blobContentType: 'zip'},
+    onProgress: uploadCallback
+  }
+
+  try {
+    await blockBlobClient.uploadStream(
+      zipUploadStream,
+      bufferSize,
+      maxBuffers,
+      options
+    )
+  } catch (error) {
+    core.info(`Failed to upload artifact zip to blob storage, error: ${error}`)
+    return {
+      isSuccess: false
+    }
+  }
+
+  if (uploadByteCount === 0) {
+    core.warning(`No data was uploaded to blob storage. Reported upload byte count is 0`)
+    return {
+        isSuccess: false,
+    }
+  }
+
+  core.info(
+    `Successfully uploaded all artifact file content. Total reported size: ${uploadByteCount}`
+  )
+
+  return {
+    isSuccess: true,
+    uploadSize: uploadByteCount
+  }
+}

packages/artifact/src/internal/upload/upload-artifact.ts

@@ -11,6 +11,8 @@ import {
 } from './upload-zip-specification'
 import {getBackendIdsFromToken} from '../shared/util'
 import {CreateArtifactRequest} from 'src/generated'
+import {uploadZipToBlobStorage} from './blob-upload'
+import {createZipUploadStream} from './zip'
 
 export async function uploadArtifact(
   name: string,
@@ -32,6 +34,8 @@ export async function uploadArtifact(
     }
   }
 
+  const zipUploadStream = await createZipUploadStream(zipSpecification)
+
   // get the IDs needed for the artifact creation
   const backendIds = getBackendIdsFromToken()
   if (!backendIds.workflowRunBackendId || !backendIds.workflowJobRunBackendId) {
@@ -52,7 +56,7 @@ export async function uploadArtifact(
   const createArtifactReq: CreateArtifactRequest = {
     workflowRunBackendId: backendIds.workflowRunBackendId,
     workflowJobRunBackendId: backendIds.workflowJobRunBackendId,
-    name,
+    name: name,
     version: 4
   }
 
@@ -72,14 +76,20 @@ export async function uploadArtifact(
     }
   }
 
-  // TODO - Implement upload functionality
+  // Upload zip to blob storage
+  const uploadResult = await uploadZipToBlobStorage(createArtifactResp.signedUploadUrl, zipUploadStream)
+  if (uploadResult.isSuccess === false) {
+    return {
+      success: false
+    }
+  }
 
   // finalize the artifact
   const finalizeArtifactResp = await artifactClient.FinalizeArtifact({
     workflowRunBackendId: backendIds.workflowRunBackendId,
     workflowJobRunBackendId: backendIds.workflowJobRunBackendId,
-    name,
-    size: '0' // TODO - Add size
+    name: name,
+    size: uploadResult.uploadSize!.toString()
   })
   if (!finalizeArtifactResp.ok) {
     core.warning(`Failed to finalize artifact`)
@@ -88,11 +98,9 @@ export async function uploadArtifact(
     }
   }
 
-  const uploadResponse: UploadResponse = {
+  return {
     success: true,
-    size: 0,
+    size: uploadResult.uploadSize,
     id: parseInt(finalizeArtifactResp.artifactId) // TODO - will this be a problem due to the id being a bigint?
   }
-
-  return uploadResponse
 }

packages/artifact/src/internal/upload/zip.ts

@@ -0,0 +1,95 @@
+import * as stream from 'stream'
+import * as archiver from 'archiver'
+import * as core from '@actions/core'
+import {createReadStream} from 'fs'
+import {UploadZipSpecification} from './upload-zip-specification'
+import {getUploadChunkSize} from '../shared/config'
+
+// Custom stream transformer so we can set the highWaterMark property
+// See https://github.com/nodejs/node/issues/8855
+export class ZipUploadStream extends stream.Transform {
+  constructor(bufferSize: number) {
+    super({
+      highWaterMark: bufferSize
+    })
+  }
+
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  _transform(chunk: any, enc: any, cb: any): void {
+    cb(null, chunk)
+  }
+}
+
+export async function createZipUploadStream(
+  uploadSpecification: UploadZipSpecification[]
+): Promise<ZipUploadStream> {
+  const zip = archiver.create('zip', {
+    zlib: {level: 9} // Sets the compression level.
+    // Available options are 0-9
+    // 0 => no compression
+    // 1 => fastest with low compression
+    // 9 => highest compression ratio but the slowest
+  })
+
+  // register callbacks for various events during the zip lifecycle
+  zip.on('error', zipErrorCallback)
+  zip.on('warning', zipWarningCallback)
+  zip.on('finish', zipFinishCallback)
+  zip.on('end', zipEndCallback)
+
+  for (const file of uploadSpecification) {
+    if (file.sourcePath !== null) {
+      // Add a normal file to the zip
+      zip.append(createReadStream(file.sourcePath), {
+        name: file.destinationPath
+      })
+    } else {
+      // Add a directory to the zip
+      zip.append('', {name: file.destinationPath})
+    }
+  }
+
+  const bufferSize = getUploadChunkSize()
+  const zipUploadStream = new ZipUploadStream(bufferSize)
+
+  core.debug(
+    `Zip write high watermark value ${zipUploadStream.writableHighWaterMark}`
+  )
+  core.debug(
+    `Zip read high watermark value ${zipUploadStream.readableHighWaterMark}`
+  )
+
+  zip.pipe(zipUploadStream)
+  zip.finalize()
+
+  return zipUploadStream
+}
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+const zipErrorCallback = (error: any): void => {
+  core.error('An error has occurred while creating the zip file for upload')
+  core.info(error)
+
+  throw new Error('An error has occurred during zip creation for the artifact')
+}
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+const zipWarningCallback = (error: any): void => {
+  if (error.code === 'ENOENT') {
+    core.warning('ENOENT warning during artifact zip creation. No such file or directory')
+    core.info(error)
+  } else {
+    core.warning(
+      `A non-blocking warning has occurred during artifact zip creation: ${error.code}`
+    )
+    core.info(error)
+  }
+}
+
+const zipFinishCallback = (): void => {
+  core.debug('Zip stream for upload has finished.')
+}
+
+const zipEndCallback = (): void => {
+  core.debug('Zip stream for upload has ended.')
+}