import { AbortController } from "abort-controller" ; import { Gaxios , GaxiosOptions , GaxiosPromise , GaxiosResponse } from "gaxios" ; import { cloudbilling_v1 , cloudfunctions_v1 , google , GoogleApis , pubsub_v1 } from "googleapis" ; import * as https from "https" ; import * as util from "util" ; import { caches } from "../cache" ; import { CostMetric , CostSnapshot } from "../cost" ; import { FaastError , FaastErrorNames } from "../error" ; import { log } from "../log" ; import { packer , PackerResult } from "../packer" ; import { CleanupOptions , commonDefaults , CommonOptions , FunctionStats , PollResult , ProviderImpl , UUID } from "../provider" ; import { serialize } from "../serialize" ; import { computeHttpResponseBytes , hasExpired , keysOf , sleep , uuidv4Pattern } from "../shared" ; import { retryOp , throttle } from "../throttle" ; import { FunctionCall , WrapperOptions } from "../wrapper" ; import { publishPubSub , receiveMessages } from "./google-queue" ; import { shouldRetryRequest } from "./google-shared" ; import * as googleTrampolineHttps from "./google-trampoline-https" ; import * as googleTrampolineQueue from "./google-trampoline-queue" ; import CloudFunctions = cloudfunctions_v1 ; import PubSubApi = pubsub_v1 ; import CloudBilling = cloudbilling_v1 ; const gaxios = new Gaxios ( { retryConfig : { retry : 3 , noResponseRetries : 3 , shouldRetry : shouldRetryRequest ( log . info ) } } ) ; /**
 * Valid Google Cloud
 *  https . Only
 * some of these
 *  https .
 *  public export type GoogleRegion = "asia-east1" | "asia-east2" | "asia-northeast1" | "asia-south1" | "asia-southeast1" | "australia-southeast1" | "europe-north1" | "europe-west1" | "europe-west2" | "europe-west3" | "europe-west4" | "europe-west6" | "northamerica-northeast1" | "southamerica-east1" | "us-central1" | "us-east1" | "us-east4" | "us-west1" | "us-west2" ; const GoogleCloudFunctionsMemorySizes = [ 128 , 256 , 512 , 1024 , 2048 ] ; /**
 * Google-specific options for  faastGoogle .
 *  public export interface GoogleOptions extends CommonOptions { /**
     * The region to create resources in. Garbage collection is also limited to
     * this region. Default: `"us-central1"`.
     */ region ? : GoogleRegion ; /**
     * Additional options to pass to Google Cloud Function creation. See
     *  https .
     *  remarks googleCloudFunctionOptions ? : CloudFunctions . Schema$CloudFunction ; internal _gcWorker ? : ( resources : GoogleResources , services : GoogleServices ) => Promise < void > ; } export interface GoogleResources { trampoline : string ; requestQueueTopic ? : string ; requestSubscription ? : string ; responseQueueTopic ? : string ; responseSubscription ? : string ; region : string ; } export interface GoogleCloudPricing { perInvocation : number ; perGhzSecond : number ; perGbSecond : number ; perGbOutboundData : number ; perGbPubSub : number ; } export class GoogleMetrics { outboundBytes = 0 ; pubSubBytes = 0 ; } export interface GoogleServices { readonly cloudFunctions : CloudFunctions . Cloudfunctions ; readonly pubsub : PubSubApi . Pubsub ; readonly google : GoogleApis ; readonly cloudBilling : CloudBilling . Cloudbilling ; } internal export interface GoogleState { resources : GoogleResources ; services : GoogleServices ; url ? : string ; project : string ; functionName : string ; metrics : GoogleMetrics ; options : Required < GoogleOptions > ; gcPromise ? : Promise < void > ; } export function defaultGcWorker ( resources , services ) { return deleteResources ( services , resources , log . gc ) ; } export const defaults = { ... commonDefaults , region : "us-central1" , googleCloudFunctionOptions : { } , _gcWorker : defaultGcWorker } ; export const GoogleImpl = { name : "google" , initialize , defaults , cleanup , costSnapshot , logUrl , invoke , poll , responseQueueId } ; export async function initializeGoogleServices ( ) { const auth = await google . auth . getClient ( { scopes : [ "https://www.googleapis.com/auth/cloud-platform" ] } ) ; google . options ( { auth , retryConfig : { retry : 8 , retryDelay : 250 , noResponseRetries : 3 , shouldRetry : shouldRetryRequest ( log . info ) } } ) ; return { cloudFunctions : google . cloudfunctions ( "v1" ) , pubsub : google . pubsub ( "v1" ) , cloudBilling : google . cloudbilling ( "v1" ) , google } ; } interface PollOptions { maxRetries ? : number ; delay ? : ( retries : number ) => Promise < void > ; } interface PollConfig < T > extends PollOptions { request : ( ) => Promise < T > ; checkDone : ( result : T ) => boolean ; } async function defaultPollDelay ( retries ) { if ( retries > 5 ) { await sleep ( 5 * 1000 ) ; } await sleep ( ( retries + 1 ) * 500 ) ; } async function pollOperation < T > ( { request , checkDone , delay = defaultPollDelay , maxRetries = 50 } ) { let retries = 0 ; await delay ( retries ) ; while ( true ) { log . info ( `Polling...` ) ; const result = await request ( ) ; if ( checkDone ( result ) ) { log . info ( `Done.` ) ; return result ; } if ( retries ++ >= maxRetries ) { throw new FaastError ( `Timed out after ${ retries } attempts.` ) ; } await delay ( retries ) ; } } async function quietly < T > ( promise ) { try { const result = await promise ; return result . data ; } catch ( err ) { return ; } } const throttleGoogleWrite = throttle ( { concurrency : 4 , rate : 3 , retry : ( err , n ) => { const { message } = err as Error ; return ( n < 6 && ( message . match ( /Build failed/ ) !== null || message . match ( /Quota/ ) !== null || message . match ( /load attempt timed out/ ) !== null || message . match ( /ECONNRESET/ ) !== null || message . match ( /failed on loading user code/ ) != null ) ) ; } } , < T > ( op ) => op ( ) ) ; async function waitFor ( api , response ) { return throttleGoogleWrite ( async ( ) => { let operation ; try { operation = await response ( ) ; } catch ( err ) { throw new FaastError ( err , "could not get operation" ) ; } const operationName = operation . data . name ! ; try { return pollOperation ( { request : ( ) => quietly ( api . operations . get ( { name : operationName } ) ) , checkDone : result => { if ( ! result ) { return false ; } if ( result . error ) { const underlying = new FaastError ( result . error . message ?? undefined ) ; underlying . stack = "" ; throw new FaastError ( underlying , "Error polling operation" ) ; } return result . done || false ; } } ) ; } catch ( err ) { throw new FaastError ( err , "poll operation failed" ) ; } } ) ; } async function deleteFunction ( api , path ) { try { return await waitFor ( api , ( ) => api . projects . locations . functions . delete ( { name : path } ) ) ; } catch ( err ) { if ( err . message . match ( /does not exist/ ) ) { return ; } throw err ; } } export async function initialize ( fmodule , nonce , options ) { log . info ( `Create google cloud function` ) ; const services = await initializeGoogleServices ( ) ; const project = await google . auth . getProjectId ( ) ; const { cloudFunctions , pubsub } = services ; const { region } = options ; const location = `projects/${ project }/locations/${ region }` ; const functionName = "faast-" + nonce ; const { timeout } = options ; const { wrapperVerbose } = options . debugOptions ; async function createCodeBundle ( ) { const wrapperOptions = { childProcessTimeoutMs : Math . max ( 1000 , ( timeout - 5 ) * 1000 ) , wrapperVerbose } ; const { archive } = await googlePacker ( fmodule , options , wrapperOptions , functionName ) ; const uploadUrlResponse = await throttleGoogleWrite ( ( ) => cloudFunctions . projects . locations . functions . generateUploadUrl ( { parent : location } ) ) ; const uploadResult = await uploadZip ( uploadUrlResponse . data . uploadUrl ! , archive ) ; log . info ( `Upload zip file response: ${ uploadResult ?. statusText }` ) ; return uploadUrlResponse . data . uploadUrl ; } const trampoline = `projects/${ project }/locations/${ region }/functions/${ functionName }` ; const resources = { trampoline , region } ; const state = { resources , services , project , functionName , metrics : new GoogleMetrics ( ) , options } ; const { gc , retentionInDays , _gcWorker : gcWorker } = options ; if ( gc === "auto" || gc === "force" ) { log . gc ( `Starting garbage collector` ) ; state . gcPromise = collectGarbage ( gcWorker , services , project , retentionInDays ) . catch ( err => { log . gc ( `Garbage collection error: ${ err }` ) ; } ) ; } const pricingPromise = getGoogleCloudFunctionsPricing ( services . cloudBilling , region ) ; const { mode } = options ; const responseQueuePromise = ( async ( ) => { const topic = await pubsub . projects . topics . create ( { name : getResponseQueueTopic ( project , functionName ) } ) ; resources . responseQueueTopic = topic . data . name ?? undefined ; resources . responseSubscription = getResponseSubscription ( project , functionName ) ; log . info ( `Creating response queue subscription` ) ; await pubsub . projects . subscriptions . create ( { name : resources . responseSubscription , requestBody : { topic : resources . responseQueueTopic } } ) ; } ) ( ) ; let requestQueuePromise ; if ( mode === "queue" ) { log . info ( `Initializing queue` ) ; resources . requestQueueTopic = getRequestQueueTopic ( project , functionName ) ; requestQueuePromise = pubsub . projects . topics . create ( { name : resources . requestQueueTopic } ) ; resources . requestSubscription = getRequestSubscription ( project , functionName , region ) ; } const sourceUploadUrl = await createCodeBundle ( ) ; const { memorySize , googleCloudFunctionOptions , env } = options ; if ( ! GoogleCloudFunctionsMemorySizes . find ( size => size === memorySize ) ) { log . warn ( `Invalid memorySize ${ memorySize } for Google Cloud Functions` ) ; } const requestBody = { name : trampoline , entryPoint : "trampoline" , timeout : `${ timeout }s` , availableMemoryMb : memorySize , sourceUploadUrl , environmentVariables : env , runtime : "nodejs14" , ... googleCloudFunctionOptions } ; if ( mode === "queue" ) { await requestQueuePromise ; requestBody . eventTrigger = { eventType : "providers/cloud.pubsub/eventTypes/topic.publish" , resource : resources . requestQueueTopic } ; } else { requestBody . httpsTrigger = { } ; } log . info ( `Create function at ${ location }` ) ; log . info ( `Request body: %O` , requestBody ) ; await retryOp ( 3 , async ( ) => { try { log . info ( `create function ${ requestBody . name } [${ options . description }]` ) ; await waitFor ( cloudFunctions , ( ) => cloudFunctions . projects . locations . functions . create ( { location , requestBody } ) ) ; await cloudFunctions . projects . locations . functions . setIamPolicy ( { resource : trampoline , requestBody : { policy : { bindings : [ { members : [ "allUsers" ] , role : "roles/cloudfunctions.invoker" } ] } } } ) ; } catch ( err ) { await deleteFunction ( cloudFunctions , trampoline ) . catch ( ( ) => { } ) ; throw new FaastError ( { cause : err , name : FaastErrorNames . ECREATE } , "failed to create google cloud function" ) ; } } ) ; if ( mode === "https" || mode === "auto" ) { try { const func = await cloudFunctions . projects . locations . functions . get ( { name : trampoline } ) ; if ( ! func . data . httpsTrigger ) { throw new FaastError ( "Could not get http trigger url" ) ; } const { url } = func . data . httpsTrigger ! ; if ( ! url ) { throw new FaastError ( "Could not get http trigger url" ) ; } log . info ( `Function URL: ${ url }` ) ; state . url = url ; } catch ( err ) { throw new FaastError ( err , `Could not get function ${ trampoline } or its url, despite it being created` ) ; } } await pricingPromise ; await responseQueuePromise ; return state ; } function getRequestQueueTopic ( project , functionName ) { return `projects/${ project }/topics/${ functionName }-Requests` ; } export function getResponseQueueTopic ( project , functionName ) { return `projects/${ project }/topics/${ functionName }-Responses` ; } export function getResponseSubscription ( project , functionName ) { return `projects/${ project }/subscriptions/${ functionName }-Responses` ; } export function getRequestSubscription ( project , functionName , region ) { return `projects/${ project }/subscriptions/gcf-${ functionName }-${ region }-${ functionName }-Requests` ; } const agent = new https . Agent ( { keepAlive : true , timeout : 0 , maxSockets : 1000 } ) ; async function callFunctionHttps ( url , call , metrics , cancel ) { const source = new AbortController ( ) ; try { const axiosConfig = { method : "POST" , url , headers : { "Content-Type" : "application/json" } , body : serialize ( call ) , signal : source . signal , responseType : "json" , retry : false , agent } ; const rawResponse = await Promise . race ( [ gaxios . request < void > ( axiosConfig ) , cancel ] ) ; if ( ! rawResponse ) { log . info ( `cancelling gcp invoke` ) ; source . abort ( ) ; return ; } try { metrics . outboundBytes += computeHttpResponseBytes ( rawResponse ! . headers ) ; } catch ( err ) { throw new FaastError ( err , `Could not parse ${ util . inspect ( rawResponse . data ) }` ) ; } } catch ( err ) { const { response } = err ; if ( response ) { if ( response . status === 503 ) { throw new FaastError ( { cause : err , name : FaastErrorNames . EMEMORY } , "google cloud function: possibly out of memory" ) ; } throw new FaastError ( err , `when invoking google cloud function: %s\nDetails: %s` , response . statusText , response . data ) ; } throw new FaastError ( err , `when invoking google cloud function` ) ; } } async function invoke ( state , call , cancel ) { const { options , resources , services , url , metrics } = state ; switch ( options . mode ) { case "auto" : case "https" : return callFunctionHttps ( url ! , call , metrics , cancel ) ; case "queue" : const { requestQueueTopic } = resources ; const { pubsub } = services ; const serialized = serialize ( call ) ; return publishPubSub ( pubsub , requestQueueTopic ! , serialized ) ; } } function poll ( state , cancel ) { return receiveMessages ( state . services . pubsub , state . resources . responseSubscription ! , state . metrics , cancel ) ; } function responseQueueId ( state ) { return state . resources . responseQueueTopic ! ; } async function deleteResources ( services , resources , output = log . info ) { const { trampoline , requestQueueTopic , requestSubscription , responseSubscription , responseQueueTopic , region , ... rest } = resources ; const _exhaustiveCheck = { } ; const { cloudFunctions , pubsub } = services ; const check = async < T > ( request ) => { try { await request ; } catch ( err ) { if ( err . message . match ( /Resource not found/ ) ) { return ; } throw err ; } } ; if ( responseSubscription ) { await check ( pubsub . projects . subscriptions . delete ( { subscription : responseSubscription } ) ) ; output ( `Deleted response subscription: ${ responseSubscription }` ) ; } if ( responseQueueTopic ) { await check ( pubsub . projects . topics . delete ( { topic : responseQueueTopic } ) ) ; output ( `Deleted response queue topic: ${ responseQueueTopic }` ) ; } if ( requestSubscription ) { await check ( pubsub . projects . subscriptions . delete ( { subscription : requestSubscription } ) ) ; output ( `Deleted response subscription: ${ requestSubscription }` ) ; } if ( requestQueueTopic ) { await check ( pubsub . projects . topics . delete ( { topic : requestQueueTopic } ) ) ; output ( `Deleted request queue topic: ${ requestQueueTopic }` ) ; } if ( trampoline ) { await check ( deleteFunction ( cloudFunctions , trampoline ) ) ; output ( `Deleted function ${ trampoline }` ) ; } } export async function cleanup ( state , options ) { log . info ( `google cleanup starting.` ) ; if ( state . gcPromise ) { log . info ( `Waiting for garbage collection...` ) ; await state . gcPromise ; log . info ( `Garbage collection done.` ) ; } if ( options . deleteResources ) { try { await deleteResources ( state . services , state . resources ) ; } catch ( err ) { throw new FaastError ( err , "delete resources failed" ) ; } } log . info ( `google cleanup done.` ) ; } let garbageCollectorRunning = false ; async function collectGarbage ( gcWorker , services , proj , retentionInDays ) { if ( gcWorker === defaultGcWorker ) { if ( garbageCollectorRunning ) { return ; } garbageCollectorRunning = true ; } try { const { cloudFunctions } = services ; let pageToken ; let promises = [ ] ; const scheduleDeleteResources = throttle ( { concurrency : 5 , rate : 5 , burst : 2 } , async ( gServices , fn ) => { const { region , name , project } = parseFunctionName ( fn . name ! ) ! ; const resources = { region , trampoline : fn . name ! , requestQueueTopic : getRequestQueueTopic ( project , name ) , requestSubscription : getRequestSubscription ( project , name , region ) , responseQueueTopic : getResponseQueueTopic ( project , name ) , responseSubscription : getResponseSubscription ( project , name ) } ; await gcWorker ( resources , gServices ) ; } ) ; const fnPattern = new RegExp ( `/functions/faast-${ uuidv4Pattern }$` ) ; do { const funcListResponse = await cloudFunctions . projects . locations . functions . list ( { parent : `projects/${ proj }/locations/-` , pageToken } ) ; pageToken = funcListResponse . data . nextPageToken ?? undefined ; const garbageFunctions = ( funcListResponse . data . functions || [ ] ) . filter ( fn => hasExpired ( fn . updateTime , retentionInDays ) ) . filter ( fn => fn . name ! . match ( fnPattern ) ) ; promises = garbageFunctions . map ( fn => scheduleDeleteResources ( services , fn ) ) ; } while ( pageToken ) ; await Promise . all ( promises ) ; } finally { if ( gcWorker === defaultGcWorker ) { garbageCollectorRunning = false ; } } } function parseFunctionName ( path ) { const match = path . match ( /^projects\/(.*)\/locations\/(.*)\/functions\/(.*)$/ ) ; return match && { project : match [ 1 ] , region : match [ 2 ] , name : match [ 3 ] } ; } async function uploadZip ( url , zipStream ) { const config = { method : "PUT" , url , body : zipStream , headers : { "content-type" : "application/zip" , "x-goog-content-length-range" : "0,104857600" } } ; return gaxios . request ( config ) ; } export async function googlePacker ( functionModule , options , wrapperOptions , FunctionName ) { const { mode } = options ; const trampolineModule = mode === "queue" ? googleTrampolineQueue : googleTrampolineHttps ; return packer ( trampolineModule , functionModule , options , wrapperOptions , FunctionName ) ; } let getGooglePrice ; function ensureGooglePriceCache ( cloudBilling ) { if ( getGooglePrice ) { return ; } getGooglePrice = throttle ( { concurrency : 1 , rate : 3 , memoize : true , cache : caches . googlePrices } , async ( region , serviceName , description , conversionFactor ) => { try { const skusResponse = await cloudBilling . services . skus . list ( { parent : serviceName } ) ; const { skus = [ ] } = skusResponse . data ; const matchingSkus = skus . filter ( sku => sku . description === description ) ; log . provider ( `matching SKUs: ${ util . inspect ( matchingSkus , { depth : null } ) }` ) ; const regionOrGlobalSku = matchingSkus . find ( sku => sku . serviceRegions ! . find ( r => r === region ) ) ?? matchingSkus . find ( sku => sku . serviceRegions ! . find ( r => r === "global" ) ) ; const pexp = regionOrGlobalSku ! . pricingInfo ! [ 0 ] . pricingExpression ! ; const prices = pexp . tieredRates ! . map ( rate => Number ( rate . unitPrice ! . units ?? "0" ) + rate . unitPrice ! . nanos ! / 1e9 ) ; const price = Math . max ( ... prices ) * ( conversionFactor / pexp . baseUnitConversionFactor ! ) ; log . provider ( `Found price for ${ serviceName }, ${ description }, ${ region }: ${ price }` ) ; return price ; } catch ( err ) { throw new FaastError ( err , `failed to get google pricing for "${ description }"` ) ; } } ) ; } let googleServices ; const listGoogleServices = throttle ( { concurrency : 1 } , async ( cloudBilling ) => { if ( googleServices ) { return googleServices ; } const response = await cloudBilling . services . list ( ) ; googleServices = response . data . services ! ; return googleServices ; } ) ; async function getGoogleCloudFunctionsPricing ( cloudBilling , region ) { const services = await listGoogleServices ( cloudBilling ) ; ensureGooglePriceCache ( cloudBilling ) ; const getPricing = ( serviceName , description , conversionFactor = 1 ) => { const service = services . find ( s => s . displayName === serviceName ) ! ; return getGooglePrice ! ( region , service . name ! , description , conversionFactor ) ; } ; return { perInvocation : await getPricing ( "Cloud Functions" , "Invocations" ) , perGhzSecond : await getPricing ( "Cloud Functions" , "CPU Time" ) , perGbSecond : await getPricing ( "Cloud Functions" , "Memory Time" , 2 ** 30 ) , perGbOutboundData : await getPricing ( "Cloud Functions" , `Network Egress from ${ region }` , 2 ** 30 ) , perGbPubSub : await getPricing ( "Cloud Pub/Sub" , "Message Delivery Basic" , 2 ** 30 ) } ; } const gcfProvisonableMemoryTable = { 128 : 0.2 , 256 : 0.4 , 512 : 0.8 , 1024 : 1.4 , 2048 : 2.4 } ; async function costSnapshot ( state , stats ) { const costs = new CostSnapshot ( "google" , state . options , stats ) ; const { memorySize = defaults . memorySize } = state . options ; const provisionableSizes = keysOf ( gcfProvisonableMemoryTable ) . map ( n => Number ( n ) ) . sort ( ( a , b ) => a - b ) ; const provisionedMb = provisionableSizes . find ( size => memorySize <= size ) ; if ( ! provisionedMb ) { log . warn ( `Could not determine provisioned memory or CPU for requested memory size ${ memorySize }` ) ; } const provisionedGhz = gcfProvisonableMemoryTable [ provisionedMb ! ] ; const billedTimeStats = stats . estimatedBilledTime ; const seconds = ( billedTimeStats . mean / 1000 ) * billedTimeStats . samples ; const { region } = state . resources ; const prices = await getGoogleCloudFunctionsPricing ( state . services . cloudBilling , region ) ; const provisionedGb = provisionedMb ! / 1024 ; const functionCallDuration = new CostMetric ( { name : "functionCallDuration" , pricing : prices . perGbSecond * provisionedGb + prices . perGhzSecond * provisionedGhz , unit : "second" , measured : seconds , comment : `https://cloud.google.com/functions/pricing#compute_time (${ provisionedMb } MB, ${ provisionedGhz } GHz)` } ) ; costs . push ( functionCallDuration ) ; const functionCallRequests = new CostMetric ( { name : "functionCallRequests" , pricing : prices . perInvocation , measured : stats . invocations , unit : "request" , comment : "https://cloud.google.com/functions/pricing#invocations" } ) ; costs . push ( functionCallRequests ) ; const outboundDataTransfer = new CostMetric ( { name : "outboundDataTransfer" , pricing : prices . perGbOutboundData , measured : state . metrics . outboundBytes / 2 ** 30 , unit : "GB" , comment : "https://cloud.google.com/functions/pricing#networking" } ) ; costs . push ( outboundDataTransfer ) ; const pubsub = new CostMetric ( { name : "pubsub" , pricing : prices . perGbPubSub , measured : state . metrics . pubSubBytes / 2 ** 30 , unit : "GB" , comment : "https://cloud.google.com/pubsub/pricing" } ) ; costs . push ( pubsub ) ; return costs ; } export function logUrl ( state ) { const { project , functionName } = state ; return `https://console.cloud.google.com/logs/viewer?project=${ project }&resource=cloud_function%2Ffunction_name%2F${ functionName }` ; }