import * as azure from '@azure/storage-blob' ; import { logger , dateToUnixTimeSeconds } from '../utils' ; import { BadPathError , InvalidInputError , DoesNotExist , ConflictError , PreconditionFailedError } from '../errors' ; import { PerformWriteArgs , WriteResult , PerformDeleteArgs , PerformRenameArgs , PerformStatArgs , StatResult , PerformReadArgs , ReadResult , PerformListFilesArgs , ListFilesStatResult , ListFileStatResult , DriverStatics , DriverModel , DriverModelTestMethods } from '../driverModel' ; import { Readable } from 'stream' ; import { BlobGetPropertiesHeaders , BlobProperties } from '@azure/storage-blob/typings/src/generated/src/models' ; export interface AZ_CONFIG_TYPE { azCredentials : { accountName ? : string ; accountKey ? : string ; } ; bucket : string ; pageSize ? : number ; readURL ? : string ; cacheControl ? : string ; } class AzDriver implements DriverModel , DriverModelTestMethods { container : azure . ContainerURL ; accountName : string ; bucket : string ; pageSize : number ; readURL ? : string ; cacheControl ? : string ; initPromise : Promise < void > ; supportsETagMatching = true ; static getConfigInformation ( ) { const envVars = { } ; const azCredentials = { } ; if ( process . env [ 'GAIA_AZURE_ACCOUNT_NAME' ] ) { envVars [ 'azCredentials' ] = azCredentials ; azCredentials [ 'accountName' ] = process . env [ 'GAIA_AZURE_ACCOUNT_NAME' ] ; } if ( process . env [ 'GAIA_AZURE_ACCOUNT_KEY' ] ) { envVars [ 'azCredentials' ] = azCredentials ; azCredentials [ 'accountKey' ] = process . env [ 'GAIA_AZURE_ACCOUNT_KEY' ] ; } return { defaults : { azCredentials : { accountName : undefined as any , accountKey : undefined as any } } , envVars } ; } constructor ( config ) { this . bucket = config . bucket ; this . pageSize = config . pageSize ? config . pageSize : 100 ; this . accountName = config . azCredentials . accountName ; this . readURL = config . readURL ; this . cacheControl = config . cacheControl ; const sharedKeyCredential = new azure . SharedKeyCredential ( config . azCredentials . accountName , config . azCredentials . accountKey ) ; const pipeline = azure . StorageURL . newPipeline ( sharedKeyCredential ) ; const service = new azure . ServiceURL ( this . getServiceUrl ( ) , pipeline ) ; this . container = azure . ContainerURL . fromServiceURL ( service , this . bucket ) ; this . initPromise = this . createContainerIfNotExists ( ) ; } async createContainerIfNotExists ( ) { try { await this . container . create ( azure . Aborter . none , { access : 'blob' } ) ; logger . info ( `Create container ${ this . bucket } successfully` ) ; } catch ( error ) { if ( error . body && error . body . Code === 'ContainerAlreadyExists' ) { logger . info ( 'Container initialized.' ) ; } else { logger . error ( `Failed to create container: ${ error }` ) ; throw error ; } } } async deleteEmptyBucket ( ) { const files = await this . listFiles ( { pathPrefix : undefined as string } ) ; if ( files . entries . length > 0 ) { throw new Error ( 'Tried deleting non-empty bucket' ) ; } await this . container . delete ( azure . Aborter . none ) ; } ensureInitialized ( ) { return this . initPromise ; } dispose ( ) { return Promise . resolve ( ) ; } static isPathValid ( path ) { return ! path . includes ( '..' ) ; } getServiceUrl ( ) { return `https://${ this . accountName }.blob.core.windows.net` ; } getReadURLPrefix ( ) { return `${ this . getServiceUrl ( ) }/${ this . bucket }/` ; } async listBlobs ( prefix , page ? , pageSize ? ) { const blobs = await this . container . listBlobFlatSegment ( azure . Aborter . none , page || undefined , { prefix : prefix , maxresults : pageSize || this . pageSize } ) ; const items = blobs . segment . blobItems ; const entries = items . map ( e => { const fileStat = AzDriver . parseFileStat ( e . properties ) ; const result = { ... fileStat , exists : true , name : e . name . slice ( prefix . length + 1 ) } ; return result ; } ) ; return { entries , page : blobs . nextMarker || null } ; } async listFiles ( args ) { try { const files = await this . listBlobs ( args . pathPrefix , args . page , args . pageSize ) ; return { entries : files . entries . map ( f => f . name ) , page : files . page } ; } catch ( error ) { logger . debug ( `Failed to list files: ${ error }` ) ; throw error ; } } async listFilesStat ( args ) { try { return await this . listBlobs ( args . pathPrefix , args . page , args . pageSize ) ; } catch ( error ) { logger . debug ( `Failed to list files: ${ error }` ) ; throw error ; } } async performWrite ( args ) { if ( ! AzDriver . isPathValid ( args . path ) ) { throw new BadPathError ( 'Invalid Path' ) ; } if ( args . contentType && args . contentType . length > 1024 ) { throw new InvalidInputError ( 'Invalid content-type' ) ; } const azBlob = `${ args . storageTopLevel }/${ args . path }` ; const blobURL = azure . BlobURL . fromContainerURL ( this . container , azBlob ) ; const blockBlobURL = azure . BlockBlobURL . fromBlobURL ( blobURL ) ; const defaultBufferSize = 1024 * 1024 ; const bufferSize = Math . min ( defaultBufferSize , args . contentLength ) || defaultBufferSize ; see https const maxBuffers = 1 ; const options = { blobHTTPHeaders : { blobContentType : args . contentType , blobCacheControl : this . cacheControl || undefined } , accessConditions : { modifiedAccessConditions : { ifMatch : args . ifMatch , ifNoneMatch : args . ifNoneMatch } } } ; try { const uploadResult = await azure . uploadStreamToBlockBlob ( azure . Aborter . none , args . stream , blockBlobURL , bufferSize , maxBuffers , options ) ; const readURL = this . getReadURLPrefix ( ) ; const publicURL = `${ readURL }${ azBlob }` ; logger . debug ( `Storing ${ azBlob } in ${ this . bucket }, URL: ${ publicURL }` ) ; return { publicURL , etag : uploadResult . eTag . replace ( /^"|"$/g , '' ) } ; } catch ( error ) { logger . error ( `failed to store ${ azBlob } in ${ this . bucket }: ${ error }` ) ; if ( error . body && error . body . Code === 'ConditionNotMet' ) { throw new PreconditionFailedError ( 'The provided ETag does not match that of the resource on the server' ) ; } if ( error . body && error . body . Code === 'BlobAlreadyExists' ) { throw new PreconditionFailedError ( 'The entity you are trying to create already exists' ) ; } if ( error . body && error . body . Code === 'InvalidBlockList' ) { throw new ConflictError ( 'Likely failed due to concurrent PUTs to the same file' ) ; } throw new Error ( 'Azure storage failure: failed to store' + ` ${ azBlob } in container ${ this . bucket }: ${ error }` ) ; } } async performDelete ( args ) { if ( ! AzDriver . isPathValid ( args . path ) ) { throw new BadPathError ( 'Invalid Path' ) ; } const azBlob = `${ args . storageTopLevel }/${ args . path }` ; const blobURL = azure . BlobURL . fromContainerURL ( this . container , azBlob ) ; const blockBlobURL = azure . BlockBlobURL . fromBlobURL ( blobURL ) ; try { await blockBlobURL . delete ( azure . Aborter . none ) ; } catch ( error ) { if ( error . statusCode === 404 ) { throw new DoesNotExist ( 'File does not exist' ) ; } logger . error ( `failed to delete ${ azBlob } in ${ this . bucket }: ${ error }` ) ; throw new Error ( 'Azure storage failure: failed to delete' + ` ${ azBlob } in container ${ this . bucket }: ${ error }` ) ; } } async performRead ( args ) { if ( ! AzDriver . isPathValid ( args . path ) ) { throw new BadPathError ( 'Invalid Path' ) ; } const azBlob = `${ args . storageTopLevel }/${ args . path }` ; const blobURL = azure . BlobURL . fromContainerURL ( this . container , azBlob ) ; const blockBlobURL = azure . BlockBlobURL . fromBlobURL ( blobURL ) ; try { const offset = 0 ; const downloadResult = await blockBlobURL . download ( azure . Aborter . none , offset ) ; const dataStream = downloadResult . readableStreamBody as Readable ; const fileStat = AzDriver . parseFileStat ( downloadResult ) ; const result = { ... fileStat , exists : true , data : dataStream } ; return result ; } catch ( error ) { if ( error . statusCode === 404 ) { throw new DoesNotExist ( 'File does not exist' ) ; } logger . error ( `failed to read ${ azBlob } in ${ this . bucket }: ${ error }` ) ; throw new Error ( 'Azure storage failure: failed to read' + ` ${ azBlob } in container ${ this . bucket }: ${ error }` ) ; } } static parseFileStat ( properties ) { let lastModified ; if ( properties . lastModified ) { lastModified = dateToUnixTimeSeconds ( properties . lastModified ) ; } let etag = ( properties as BlobProperties ) . etag || ( properties as BlobGetPropertiesHeaders ) . eTag ; etag = etag . replace ( /^"|"$/g , '' ) ; const result = { exists : true , etag , contentLength : properties . contentLength , contentType : properties . contentType , lastModifiedDate : lastModified } ; return result ; } async performStat ( args ) { if ( ! AzDriver . isPathValid ( args . path ) ) { throw new BadPathError ( 'Invalid Path' ) ; } const azBlob = `${ args . storageTopLevel }/${ args . path }` ; const blobURL = azure . BlobURL . fromContainerURL ( this . container , azBlob ) ; const blockBlobURL = azure . BlockBlobURL . fromBlobURL ( blobURL ) ; try { const propertiesResult = await blockBlobURL . getProperties ( azure . Aborter . none ) ; const result = AzDriver . parseFileStat ( propertiesResult ) ; return result ; } catch ( error ) { if ( error . statusCode === 404 ) { const result = { exists : false } as StatResult ; return result ; } logger . error ( `failed to stat ${ azBlob } in ${ this . bucket }: ${ error }` ) ; throw new Error ( 'Azure storage failure: failed to stat' + ` ${ azBlob } in container ${ this . bucket }: ${ error }` ) ; } } async performRename ( args ) { if ( ! AzDriver . isPathValid ( args . path ) ) { throw new BadPathError ( 'Invalid original path' ) ; } if ( ! AzDriver . isPathValid ( args . newPath ) ) { throw new BadPathError ( 'Invalid new path' ) ; } const origAzBlob = `${ args . storageTopLevel }/${ args . path }` ; const origBlobUrl = azure . BlobURL . fromContainerURL ( this . container , origAzBlob ) ; const origBlockBlobURL = azure . BlockBlobURL . fromBlobURL ( origBlobUrl ) ; const newAzBlob = `${ args . storageTopLevel }/${ args . newPath }` ; const newBlobURL = azure . BlobURL . fromContainerURL ( this . container , newAzBlob ) ; const newBlockBlobURL = azure . BlockBlobURL . fromBlobURL ( newBlobURL ) ; try { const copyResult = await newBlockBlobURL . startCopyFromURL ( azure . Aborter . none , origBlockBlobURL . url ) ; if ( copyResult . copyStatus !== 'success' ) { throw new Error ( `Expected copy status to be success, got ${ copyResult . copyStatus }` ) ; } await origBlockBlobURL . delete ( azure . Aborter . none ) ; } catch ( error ) { if ( error . statusCode === 404 ) { throw new DoesNotExist ( 'File does not exist' ) ; } logger . error ( `failed to rename ${ origAzBlob } to ${ newAzBlob } in ${ this . bucket }: ${ error }` ) ; throw new Error ( 'Azure storage failure: failed to rename' + ` ${ origAzBlob } to ${ newAzBlob } in container ${ this . bucket }: ${ error }` ) ; } } } const driver = AzDriver ; export default driver ;