// ĠCopyright Ġ 2 0 2 0 ĠThe ĠCross - Media ĠMeasurement ĠAuthors Ċ // Ċ // ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ // Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ // ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ // Ċ // ĠĠĠĠĠ Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ // Ċ // ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ // Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ // ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ // ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ // Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ċ package Ġorg . wf an et . measurement . du ch y . daemon . her ald Ċ Ċ import Ġio . grpc . Status Ċ import Ġjava . util . logging . Level Ċ import Ġjava . util . logging . Logger Ċ import Ġkotlinx . coroutines . Dispatch ers Ċ import Ġkotlinx . coroutines . Global Scope Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . flow . catch Ċ import Ġkotlinx . coroutines . flow . collect Ċ import Ġkotlinx . coroutines . launch Ċ import Ġorg . wf an et . measurement . common . grpc . grpc StatusCode Ċ import Ġorg . wf an et . measurement . common . th rott ler . Th rott ler Ċ import Ġorg . wf an et . measurement . common . with Retries On Each Ċ import Ġorg . wf an et . measurement . du ch y . db . computation . Computation Protocol Stage Details Ċ import Ġorg . wf an et . measurement . du ch y . service . internal . computation . to GetToken Request Ċ import Ġorg . wf an et . measurement . internal . du ch y . Computation Details Ċ import Ġorg . wf an et . measurement . internal . du ch y . Comp ut ations Grpc Kt . Comp ut ations Coroutine Stub Ċ import Ġorg . wf an et . measurement . internal . du ch y . config . Protocols Setup Config Ċ import Ġorg . wf an et . measurement . internal . du ch y . protocol . Re quisition Key Ċ import Ġorg . wf an et . measurement . system . v 1 alpha . Global Computation Ċ import Ġorg . wf an et . measurement . system . v 1 alpha . Global Computation . State Ċ import Ġorg . wf an et . measurement . system . v 1 alpha . Global Comp ut ations Grpc Kt . Global Comp ut ations Coroutine Stub Ċ import Ġorg . wf an et . measurement . system . v 1 alpha . Stream Active Global Comp ut ations Request Ċ import Ġorg . wf an et . measurement . system . v 1 alpha . Stream Active Global Comp ut ations Response Ċ Ċ /** Ċ Ġ* ĠThe ĠHer ald Ġlooks Ġto Ġthe Ġking dom Ġfor Ġstatus Ġof Ġcomput ations . Ċ Ġ* Ċ Ġ* ĠIt Ġis Ġresponsible Ġfor Ġinserting Ġnew Ġcomput ations Ġinto Ġthe Ġdatabase , Ġand Ġfor Ġmoving Ġcomput ations Ċ Ġ* Ġout Ġof Ġthe ĠWAIT _ TO _ START Ġstage Ġonce Ġthe Ġking dom Ġhas Ġgotten Ġconfirmation Ġfrom Ġall Ġd uch ies Ġthat Ċ Ġ* Ġthey Ġare Ġable Ġto Ġstart Ġthe Ġcomputation . Ċ Ġ* Ċ Ġ* Ġ@ param Ġcomputation Storage Client Ġmanages Ġinteractions Ġwith Ġcomput ations Ġstorage Ġservice . Ċ Ġ* Ġ@ param Ġglobal Comp ut ations Client Ġstub Ġfor Ġcommunic ating Ġwith Ġthe ĠGlobal ĠComput ations ĠService Ċ Ġ*/ Ċ class ĠHer ald ( ĊĠ Ġother D uch ies In Computation : ĠList < String >, ĊĠ Ġprivate Ġval Ġcomputation Storage Client : ĠComput ations Coroutine Stub , ĊĠ Ġprivate Ġval Ġglobal Comp ut ations Client : ĠGlobal Comp ut ations Coroutine Stub , ĊĠ Ġprivate Ġval Ġprotocols Setup Config : ĠProto cols Setup Config , ĊĠ Ġprivate Ġval Ġblob Storage Bucket : ĠString Ġ= Ġ" computation - blob - storage ", ĊĠ Ġprivate Ġval Ġmax Start Attempts : ĠInt Ġ= Ġ 1 0 Ċ ) Ġ{ ĊĠ Ġprivate Ġval Ġcomputation Protocol Stage Details Ġ= ĊĠĠĠ ĠComp utation Protocol Stage Details ( other D uch ies In Computation ) ĊĊĠ Ġ// ĠIf Ġone Ġof Ġthe ĠGlobal Scope Ġcor outines Ġlaunched Ġby Ġ` start ` Ġfails , Ġit Ġpop ulates Ġthis . ĊĠ Ġprivate Ġlateinit Ġvar Ġstart Exception : ĠThrowable ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠSync s Ġthe Ġstatus Ġof Ġcomput ations Ġstored Ġat Ġthe Ġking dom Ġwith Ġthose Ġstored Ġlocally Ġcontin ually Ġin ĊĠĠ Ġ* Ġa Ġforever Ġloop . ĠThe Ġ[ polling Th rott ler ] Ġis Ġused Ġto Ġlimit Ġhow Ġoften Ġthe Ġking dom Ġand Ġlocal ĊĠĠ Ġ* Ġcomputation Ġstorage Ġservice Ġare Ġpol led . ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ@ param Ġpolling Th rott ler Ġth rott les Ġhow Ġoften Ġto Ġget Ġactive Ġcomput ations Ġfrom Ġthe ĠGlobal ĊĠĠ Ġ* ĠComp utation ĠService ĊĠĠ Ġ*/ ĊĠ Ġsuspend Ġfun Ġcontin ually Sync Statuses ( polling Th rott ler : ĠTh rott ler ) Ġ{ ĊĠĠĠ Ġlogger . info (" Starting ...") ĊĠĠĠ Ġ// ĠToken Ġsign ifying Ġthe Ġlast Ġcomputation Ġin Ġan Ġactive Ġstate Ġat Ġthe Ġking dom Ġthat Ġwas Ġprocessed Ġby ĊĠĠĠ Ġ// Ġthis Ġjob . ĠWhen Ġempty , Ġall Ġactive Ġcomput ations Ġat Ġthe Ġking dom Ġwill Ġbe Ġst reamed Ġin Ġthe ĊĠĠĠ Ġ// Ġresponse . ĠThe Ġfirst Ġexecution Ġof Ġthe Ġloop Ġwill Ġthen Ġcompare Ġall Ġactive Ġcomput ations Ġat ĊĠĠĠ Ġ// Ġthe Ġking dom Ġwith Ġall Ġactive Ġcomput ations Ġlocally . ĊĠĠĠ Ġvar Ġlast Processed Continu ationToken Ġ= Ġ"" ĊĊĠĠĠ Ġpolling Th rott ler . loop On Ready Ġ{ ĊĠĠĠĠĠ Ġlast Processed Continu ationToken Ġ= Ġsync Statuses ( last Processed Continu ationToken ) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠSync s Ġthe Ġstatus Ġof Ġcomput ations Ġstored Ġat Ġthe Ġking dom , Ġvia Ġthe Ġglobal Ġcomputation Ġservice , ĊĠĠ Ġ* Ġwith Ġthose Ġstored Ġlocally . ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ@ param Ġcontinu ationToken Ġthe Ġcontinuation Ġtoken Ġof Ġthe Ġlast Ġcomputation Ġin Ġthe Ġstream Ġwhich Ġwas ĊĠĠ Ġ* Ġprocessed Ġby Ġthe Ġher ald . ĊĠĠ Ġ* Ġ@ return Ġthe Ġcontinuation Ġtoken Ġof Ġthe Ġlast Ġcomputation Ġprocessed Ġin Ġthat Ġstream Ġof Ġactive ĊĠĠ Ġ* Ġcomput ations Ġfrom Ġthe Ġglobal Ġcomputation Ġservice . ĊĠĠ Ġ*/ ĊĠ Ġsuspend Ġfun Ġsync Statuses ( continu ationToken : ĠString ): ĠString Ġ{ ĊĠĠĠ Ġif Ġ( this :: start Exception . isInitialized ) Ġ{ ĊĠĠĠĠĠ Ġthrow Ġstart Exception ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġvar Ġlast Processed Continu ationToken Ġ= Ġcontinu ationToken ĊĠĠĠ Ġlogger . info (" Reading Ġstream Ġof Ġactive Ġcomput ations Ġsince Ġ$ continu ationToken .") ĊĠĠĠ Ġglobal Comp ut ations Client ĊĠĠĠĠĠ Ġ. stream Active Global Comp ut ations ( ĊĠĠĠĠĠĠĠ ĠStream Active Global Comp ut ations Request . newBuilder () ĊĠĠĠĠĠĠĠĠĠ Ġ. set Continu ationToken ( continu ationToken ) ĊĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠĠĠ Ġ) ĊĠĠĠĠĠ Ġ. with Retries On Each ( max Attempts Ġ= Ġ 3 , Ġretry Predicate Ġ= Ġ:: may Be Transient Grpc Error ) Ġ{ Ġresponse Ġ-> ĊĠĠĠĠĠĠĠ Ġprocess Global Computation Change ( response ) ĊĠĠĠĠĠĠĠ Ġlast Processed Continu ationToken Ġ= Ġresponse . continu ationToken ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ// ĠCancel Ġthe Ġflow Ġon Ġthe Ġfirst Ġerror , Ġbut Ġdon 't Ġactually Ġthrow Ġthe Ġerror . ĠThis Ġwill Ġkeep ĊĠĠĠĠĠ Ġ// Ġthe Ġcontinuation Ġtoken Ġat Ġthe Ġlast Ġsuccessfully Ġprocessed Ġitem . ĠA Ġlater Ġexecution Ġof ĊĠĠĠĠĠ Ġ// Ġsync Statuses () Ġmay Ġbe Ġsuccessful Ġif Ġthe Ġstate Ġat Ġthe Ġking dom Ġand / or Ġthis Ġd uch y Ġwas Ġupdated . ĊĠĠĠĠĠ Ġ. catch Ġ{ Ġe Ġ-> Ġlogger . log ( Level . SEVERE , Ġ" Exception :", Ġe ) Ġ} ĊĠĠĠĠĠ Ġ. collect () ĊĠĠĠ Ġreturn Ġlast Processed Continu ationToken ĊĠ Ġ} ĊĊĠ Ġprivate Ġsuspend Ġfun Ġprocess Global Computation Change ( ĊĠĠĠ Ġresponse : ĠStream Active Global Comp ut ations Response ĊĠ Ġ) Ġ{ ĊĠĠĠ Ġ// ĠTODO : Ġcreate Ġcomputation Ġaccording Ġto Ġthe Ġprotocol Ġspecified Ġby Ġthe Ġking dom . ĊĠĠĠ Ġval Ġglobal Id : ĠString Ġ= ĠcheckNotNull ( response . global Computation . key ?. global Computation Id ) ĊĠĠĠ Ġlogger . info ("[ id =$ global Id ]: ĠProcessing Ġupdated ĠGlobal Computation ") ĊĠĠĠ Ġwhen Ġ( val Ġstate Ġ= Ġresponse . global Computation . state ) Ġ{ ĊĠĠĠĠĠ Ġ// ĠCreate Ġa Ġnew Ġcomputation Ġif Ġit Ġis Ġnot Ġalready Ġpresent Ġin Ġthe Ġdatabase . ĊĠĠĠĠĠ ĠState . CONFIRM ING Ġ-> Ġcreate ( response . global Computation ) ĊĠĠĠĠĠ Ġ// ĠStart Ġthe Ġcomputation Ġif Ġit Ġis Ġin ĠWAIT _ TO _ START . ĊĠĠĠĠĠ Ġ// ĠTODO : ĠResume Ġa Ġcomputation Ġthat Ġwas Ġonce ĠS US PEN DED . ĊĠĠĠĠĠ ĠState . RUNNING Ġ-> Ġstart ( global Id ) ĊĠĠĠĠĠ ĠState . SUS PEN DED Ġ-> ĊĠĠĠĠĠĠĠ Ġlogger . warning (" Pause / Resume Ġof Ġcomput ations Ġbased Ġon Ġking dom Ġstate Ġnot Ġyet Ġsupported .") ĊĠĠĠĠĠ Ġelse Ġ-> Ġlogger . warning (" Unexpected Ġglobal Ġcomputation Ġstate Ġ'$ state '") ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĠCreates Ġa Ġnew Ġcomputation . Ġ*/ ĊĠ Ġ// ĠTODO : Ġcreate ĠComp utation Ġof Ġtype Ġmapping Ġthe Ġprotocol Ġspecified Ġby Ġthe Ġking dom . ĊĠ Ġprivate Ġsuspend Ġfun Ġcreate ( global Computation : ĠGlobal Computation ) Ġ{ ĊĠĠĠ Ġval Ġglobal Id : ĠString Ġ= ĠcheckNotNull ( global Computation . key ?. global Computation Id ) ĊĠĠĠ Ġlogger . info ("[ id =$ global Id ] ĠCreating ĠComp utation ") ĊĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠ Ġ// ĠTODO : Ġget Ġthe Ġprotocol Ġfrom Ġthe Ġking dom 's Ġresponse Ġand Ġcreate Ġa Ġcorresponding Ġcomputation . ĊĠĠĠĠĠ ĠLi quid Leg ions V 2 Starter . create Computation ( ĊĠĠĠĠĠĠĠ Ġcomputation Storage Client , ĊĠĠĠĠĠĠĠ Ġglobal Computation , ĊĠĠĠĠĠĠĠ Ġprotocols Setup Config , ĊĠĠĠĠĠĠĠ Ġblob Storage Bucket ĊĠĠĠĠĠ Ġ) ĊĠĠĠĠĠ Ġlogger . info ("[ id =$ global Id ]: ĠCreated ĠComp utation ") ĊĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠ Ġif Ġ( e . grpc StatusCode () Ġ== ĠStatus . Code . AL READY _ EXISTS ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlogger . info ("[ id =$ global Id ]: ĠComp utation Ġalready Ġexists ") ĊĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġthrow Ġe Ġ// Ġre throw Ġall Ġother Ġexceptions . ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠStarts Ġa Ġcomputation Ġthat Ġis Ġin ĠWAIT _ TO _ START . ĊĠĠ Ġ* ĊĠĠ Ġ* ĠThis Ġimmediately Ġattempts Ġonce Ġand Ġif Ġthat Ġfa iles , Ġla unches Ġa Ġcoroutine Ġto Ġcontinue Ġretry ing Ġin ĊĠĠ Ġ* Ġthe Ġbackground . ĊĠĠ Ġ*/ ĊĠ Ġprivate Ġsuspend Ġfun Ġstart ( global Id : ĠString ) Ġ{ ĊĠĠĠ Ġval Ġattempt : Ġsuspend Ġ() Ġ-> ĠBoolean Ġ= Ġ{ Ġrun Catch ing Ġ{ Ġstart Attempt ( global Id ) Ġ}. is Success Ġ} ĊĠĠĠ Ġif Ġ(! attempt ()) Ġ{ ĊĠĠĠĠĠ ĠGlobal Scope . launch ( Dispatch ers . IO ) Ġ{ ĊĠĠĠĠĠĠĠ Ġfor Ġ( i Ġin Ġ 2 .. max Start Attempts ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġlogger . info ("[ id =$ global Id ] ĠAttempt Ġ#$ i Ġto Ġstart ") ĊĠĠĠĠĠĠĠĠĠ Ġdelay ( time Millis Ġ= Ġmin Of (( 1 L Ġshl Ġi ) Ġ* Ġ 1 0 0 0 L , Ġ 6 0 _ 0 0 0 L )) ĊĠĠĠĠĠĠĠĠĠ Ġif Ġ( attempt ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ launch ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġmessage Ġ= Ġ"[ id =$ global Id ] ĠG iving Ġup Ġafter Ġ$ max Start Attempts Ġattempts Ġto Ġstart " ĊĠĠĠĠĠĠĠ Ġlogger . se vere ( message ) ĊĠĠĠĠĠĠĠ Ġstart Exception Ġ= ĠIllegalStateException ( message ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĠAt tempts Ġto Ġstart Ġa Ġcomputation Ġthat Ġis Ġin ĠWAIT _ TO _ START . Ġ*/ ĊĠ Ġprivate Ġsuspend Ġfun Ġstart Attempt ( global Id : ĠString ) Ġ{ ĊĠĠĠ Ġlogger . info ("[ id =$ global Id ]: ĠStarting ĠComp utation ") ĊĠĠĠ Ġval Ġtoken Ġ= Ġcomputation Storage Client . get Comp ut ationToken ( global Id . to GetToken Request ()). token ĊĠĠĠ Ġwhen Ġ( token . computation Details . protocol Case ) Ġ{ ĊĠĠĠĠĠ ĠComp utation Details . Protocol Case . LI Q UID _ LEG IONS _ V 2 Ġ-> ĊĠĠĠĠĠĠĠ ĠLi quid Leg ions V 2 Starter . start Computation ( ĊĠĠĠĠĠĠĠĠĠ Ġtoken , ĊĠĠĠĠĠĠĠĠĠ Ġcomputation Storage Client , ĊĠĠĠĠĠĠĠĠĠ Ġcomputation Protocol Stage Details , ĊĠĠĠĠĠĠĠĠĠ Ġlogger ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠ Ġelse Ġ-> Ġerror Ġ{ Ġ" Unknown Ġor Ġunsupported Ġprotocol ." Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠ Ġprivate Ġval Ġlogger : ĠLogger Ġ= ĠLogger . getLogger ( this :: class . java . name ) ĊĠ Ġ} Ċ } Ċ Ċ fun ĠGlobal Computation . to Re quisition Keys (): ĠList < Re quisition Key > Ġ= ĊĠ Ġmetric Re quis itions List . map Ġ{ ĊĠĠĠ ĠRe quisition Key . newBuilder () ĊĠĠĠĠĠ Ġ. apply Ġ{ ĊĠĠĠĠĠĠĠ Ġdata Provider Id Ġ= Ġit . dataProvider Id ĊĠĠĠĠĠĠĠ Ġcampaign Id Ġ= Ġit . campaign Id ĊĠĠĠĠĠĠĠ Ġmetric Re quisition Id Ġ= Ġit . metric Re quisition Id ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ. build () ĊĠ Ġ} Ċ Ċ /** ĠReturns Ġtrue Ġif Ġthe Ġerror Ġmay Ġbe Ġtransient , Ġi . e . Ġretry ing Ġthe Ġrequest Ġmay Ġsucceed . Ġ*/ Ċ fun Ġmay Be Transient Grpc Error ( error : ĠThrowable ): ĠBoolean Ġ{ ĊĠ Ġval ĠstatusCode Ġ= Ġerror . grpc StatusCode () Ġ?: Ġreturn Ġfalse ĊĠ Ġreturn Ġwhen Ġ( statusCode ) Ġ{ ĊĠĠĠ ĠStatus . Code . ABORT ED , ĊĠĠĠ ĠStatus . Code . DEAD LINE _ EX CE EDED , ĊĠĠĠ ĠStatus . Code . RESOURCE _ EX HA UST ED , ĊĠĠĠ ĠStatus . Code . UNKNOWN , ĊĠĠĠ ĠStatus . Code . UN AVAILABLE Ġ-> Ġtrue ĊĠĠĠ Ġelse Ġ-> Ġfalse ĊĠ Ġ} Ċ } Ċ