/* Ċ Ġ* ĠCopyright Ġ 2 0 1 8 ĠNet flix , ĠInc . Ċ Ġ* Ċ Ġ* Ġ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 Ġcom . netflix . spin n aker . or ca . sql . cleanup Ċ Ċ import Ġcom . netflix . spect ator . api . Counter Ċ import Ġcom . netflix . spect ator . api . Id Ċ import Ġcom . netflix . spect ator . api . Long Task Timer Ċ import Ġcom . netflix . spect ator . api . Registry Ċ import Ġcom . netflix . spin n aker . k ork . core . Retry Support Ċ import Ġcom . netflix . spin n aker . or ca . Execution Status Ċ import Ġcom . netflix . spin n aker . or ca . notifications . Abstract Polling Notification Agent Ċ import Ġcom . netflix . spin n aker . or ca . notifications . Notification Cluster Lock Ċ import Ġcom . netflix . spin n aker . or ca . sql . pipeline . persistence . transaction al Ċ import Ġorg . jooq . DSL Context Ċ import Ġorg . jooq . impl . DSL . count Ċ import Ġorg . jooq . impl . DSL . field Ċ import Ġorg . jooq . impl . DSL . table Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . springframework . beans . factory . annotation . Value Ċ import Ġorg . springframework . boot . autoconfigure . condition . Conditional On Expression Ċ import Ġorg . springframework . stereotype . Component Ċ import Ġjava . time . Clock Ċ import Ġjava . time . Instant Ċ import Ġjava . time . temporal . Ch rono Unit Ċ import Ġjava . util . concurrent . atomic . Atomic Integer Ċ Ċ @ Component Ċ @ Conditional On Expression ("\ ${ pol lers . old - pipeline - cleanup . enabled : false } Ġ&& Ġ! \ ${ execution - repository . redis . enabled : false }") Ċ class ĠOld Pipeline Cleanup Polling Notification Agent ( ĊĠ Ġcluster Lock : ĠNotification Cluster Lock , ĊĠ Ġprivate Ġval Ġj oo q : ĠDSL Context , ĊĠ Ġprivate Ġval Ġclock : ĠClock , ĊĠ Ġprivate Ġval Ġregistry : ĠRegistry , ĊĠ Ġ@ Value ("\ ${ pol lers . old - pipeline - cleanup . interval - ms : 3 6 0 0 0 0 0 }") Ġprivate Ġval Ġpolling Interval Ms : ĠLong , ĊĠ Ġ@ Value ("\ ${ pol lers . old - pipeline - cleanup . threshold - days : 3 0 }") Ġprivate Ġval Ġthreshold Days : ĠLong , ĊĠ Ġ@ Value ("\ ${ pol lers . old - pipeline - cleanup . minimum - pipeline - executions : 5 }") Ġprivate Ġval Ġminimum Pipeline Execution s : ĠInt , ĊĠ Ġ@ Value ("\ ${ pol lers . old - pipeline - cleanup . chunk - size : 1 }") Ġprivate Ġval Ġchunk Size : ĠInt Ċ ) Ġ: ĠAbstract Polling Notification Agent ( cluster Lock ) Ġ{ ĊĊĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠ Ġinternal Ġval Ġretry Support Ġ= ĠRetry Support () ĊĠ Ġ} ĊĊĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( Old Pipeline Cleanup Polling Notification Agent :: class . java ) ĊĊĠ Ġprivate Ġval Ġdeleted Id : ĠId ĊĠ Ġprivate Ġval Ġerrors Counter : ĠCounter ĊĠ Ġprivate Ġval Ġinvocation Timer : ĠLong Task Timer ĊĊĠ Ġprivate Ġval Ġcompleted Statuses Ġ= ĠExecution Status . COMPLETED . map Ġ{ Ġit . toString () Ġ} ĊĊĠ Ġinit Ġ{ ĊĠĠĠ Ġdeleted Id Ġ= Ġregistry . create Id (" pol lers . old Pipeline Cleanup . deleted ") ĊĠĠĠ Ġerrors Counter Ġ= Ġregistry . counter (" pol lers . old Pipeline Cleanup . errors ") ĊĊĠĠĠ Ġinvocation Timer Ġ= Ġcom . netflix . spect ator . api . patterns . Long Task Timer . get ( ĊĠĠĠĠĠ Ġregistry , Ġregistry . create Id (" pol lers . old Pipeline Cleanup . timing ") ĊĠĠĠ Ġ) ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġget Polling Interval (): ĠLong Ġ{ ĊĠĠĠ Ġreturn Ġpolling Interval Ms ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġget Notification Type (): ĠString Ġ{ ĊĠĠĠ Ġreturn ĠOld Pipeline Cleanup Polling Notification Agent :: class . java . simple Name ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġtick () Ġ{ ĊĠĠĠ Ġval Ġtimer Id Ġ= Ġinvocation Timer . start () ĊĠĠĠ Ġval ĠstartTime Ġ= ĠSystem . currentTimeMillis () ĊĊĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠ Ġlog . info (" Agent Ġ{} Ġstarted ", Ġnotification Type ) ĊĠĠĠĠĠ Ġperform Cleanup () ĊĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠ Ġlog . error (" Agent Ġ{} Ġfailed Ġto Ġperform Ġcleanup ", Ġe ) ĊĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠ Ġlog . info (" Agent Ġ{} Ġcompleted Ġin Ġ{} ms ", Ġnotification Type , ĠSystem . currentTimeMillis () Ġ- ĠstartTime ) ĊĠĠĠĠĠ Ġinvocation Timer . stop ( timer Id ) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġprivate Ġfun Ġperform Cleanup () Ġ{ ĊĠĠĠ Ġval Ġthreshold Millis Ġ= ĠInstant . of Epoch Milli ( clock . millis ()). minus ( threshold Days , ĠCh rono Unit . DAYS ). to Epoch Milli () ĊĊĠĠĠ Ġval Ġpipeline Config sWith Old Execution s Ġ= Ġj oo q ĊĠĠĠĠĠ Ġ. select ( field (" application "), Ġfield (" config _ id "), Ġcount ( field (" id ")). ` as ` (" count ")) ĊĠĠĠĠĠ Ġ. from ( table (" pipelines ")) ĊĠĠĠĠĠ Ġ. where ( ĊĠĠĠĠĠĠĠ Ġfield (" build _ time "). le ( threshold Millis ). and ( ĊĠĠĠĠĠĠĠĠĠ Ġ" status ĠIN Ġ(${ completed Statuses . join ToString (",") Ġ{ Ġ"' $ it '" Ġ} })" ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠ Ġ) ĊĠĠĠĠĠ Ġ. groupBy ( field (" application "), Ġfield (" config _ id ")) ĊĠĠĠĠĠ Ġ. having ( count ( field (" id ")). gt ( minimum Pipeline Execution s )) ĊĠĠĠĠĠ Ġ. fetch () ĊĊĠĠĠ Ġpipeline Config sWith Old Execution s . forEach Ġ{ ĊĠĠĠĠĠ Ġval Ġapplication Ġ= Ġit . getValue ( field (" application ")) Ġas ĠString ? Ġ?: Ġreturn @ forEach ĊĠĠĠĠĠ Ġval Ġpipeline Config Id Ġ= Ġit . getValue ( field (" config _ id ")) Ġas ĠString ? Ġ?: Ġreturn @ forEach ĊĊĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠ Ġval ĠstartTime Ġ= ĠSystem . currentTimeMillis () ĊĊĠĠĠĠĠĠĠ Ġlog . debug (" Clean ing Ġup Ġold Ġpipelines Ġfor Ġ$ application Ġ( pipeline Config Id : Ġ$ pipeline Config Id )") ĊĠĠĠĠĠĠĠ Ġval Ġdeleted Pipeline Count Ġ= Ġperform Cleanup ( application , Ġpipeline Config Id , Ġthreshold Millis ) ĊĠĠĠĠĠĠĠ Ġlog . debug ( ĊĠĠĠĠĠĠĠĠĠ Ġ" Clean ed Ġup Ġ{} Ġold Ġpipelines Ġfor Ġ{} Ġin Ġ{} ms Ġ( pipeline Config Id : Ġ{}) ", ĊĠĠĠĠĠĠĠĠĠ Ġdeleted Pipeline Count , ĊĠĠĠĠĠĠĠĠĠ Ġapplication , ĊĠĠĠĠĠĠĠĠĠ ĠSystem . currentTimeMillis () Ġ- ĠstartTime , ĊĠĠĠĠĠĠĠĠĠ Ġpipeline Config Id ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . error (" Failed Ġto Ġcleanup Ġold Ġpipelines Ġfor Ġ$ application Ġ( pipeline Config Id : Ġ$ pipeline Config Id )", Ġe ) ĊĠĠĠĠĠĠĠ Ġerrors Counter . increment () ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠCleanup Ġany Ġexecution Ġof Ġ[ pipeline Config Id ] Ġin Ġ[ application ] Ġthat Ġis Ġolder Ġthan Ġ[ threshold Millis ]. ĊĠĠ Ġ* ĊĠĠ Ġ* Ġ_ Only _ Ġexecution s Ġin Ġa Ġcompleted Ġstatus Ġare Ġcandidates Ġfor Ġcleanup . ĊĠĠ Ġ* ĊĠĠ Ġ* ĠIn Ġorder Ġto Ġavoid Ġremoving Ġ_ all _ Ġexecution s Ġof Ġa Ġparticular Ġpipeline Ġconfiguration , Ġ[ minimum Pipeline Execution s ] Ġcan ĊĠĠ Ġ* Ġbe Ġused Ġto Ġindicate Ġthat Ġa Ġspecific Ġnumber Ġof Ġthe Ġmost Ġrecent Ġexecution s Ġshould Ġbe Ġpreserved Ġeven Ġthough Ġthey Ġare ĊĠĠ Ġ* Ġolder Ġthan Ġ[ threshold Millis ]. ĊĠĠ Ġ*/ ĊĠ Ġprivate Ġfun Ġperform Cleanup ( ĊĠĠĠ Ġapplication : ĠString , ĊĠĠĠ Ġpipeline Config Id : ĠString , ĊĠĠĠ Ġthreshold Millis : ĠLong ĊĠ Ġ): ĠInt Ġ{ ĊĠĠĠ Ġval Ġdeleted Execution Count Ġ= ĠAtomicInteger () ĊĊĠĠĠ Ġval Ġexecution sTo Remove Ġ= Ġj oo q ĊĠĠĠĠĠ Ġ. select ( field (" id ")) ĊĠĠĠĠĠ Ġ. from ( table (" pipelines ")) ĊĠĠĠĠĠ Ġ. where ( ĊĠĠĠĠĠĠĠ Ġfield (" build _ time "). le ( threshold Millis ). and ( ĊĠĠĠĠĠĠĠĠĠ Ġ" status ĠIN Ġ(${ completed Statuses . join ToString (",") Ġ{ Ġ"' $ it '" Ġ} })" ĊĠĠĠĠĠĠĠ Ġ). and ( ĊĠĠĠĠĠĠĠĠĠ Ġfield (" application "). eq ( application ) ĊĠĠĠĠĠĠĠ Ġ). and ( ĊĠĠĠĠĠĠĠĠĠ Ġfield (" config _ id "). eq ( pipeline Config Id ) ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠ Ġ) ĊĠĠĠĠĠ Ġ. orderBy ( field (" build _ time "). desc ()) ĊĠĠĠĠĠ Ġ. limit ( minimum Pipeline Execution s , ĠInt . MAX _ VALUE ) ĊĠĠĠĠĠ Ġ. fetch () ĊĠĠĠĠĠ Ġ. map Ġ{ Ġit . getValue ( field (" id ")) Ġ} ĊĊĠĠĠ Ġexecution sTo Remove . chunk ed ( chunk Size ). forEach Ġ{ Ġids Ġ-> ĊĠĠĠĠĠ Ġdeleted Execution Count . add AndGet ( ids . size ) ĊĠĠĠĠĠ Ġj oo q . transaction al ( retry Support ) Ġ{ ĊĠĠĠĠĠĠĠ Ġit . delete ( table (" correlation _ ids ")). where ( ĊĠĠĠĠĠĠĠĠĠ Ġ" pipeline _ id ĠIN Ġ(${ ids . join ToString (",") Ġ{ Ġ"' $ it '" Ġ} })" ĊĠĠĠĠĠĠĠ Ġ). execute () ĊĠĠĠĠĠĠĠ Ġit . delete ( table (" pipeline _ stages ")). where ( ĊĠĠĠĠĠĠĠĠĠ Ġ" execution _ id ĠIN Ġ(${ ids . join ToString (",") Ġ{ Ġ"' $ it '" Ġ} })" ĊĠĠĠĠĠĠĠ Ġ). execute () ĊĠĠĠĠĠĠĠ Ġit . delete ( table (" pipelines ")). where ( ĊĠĠĠĠĠĠĠĠĠ Ġ" id ĠIN Ġ(${ ids . join ToString (",") Ġ{ Ġ"' $ it '" Ġ} })" ĊĠĠĠĠĠĠĠ Ġ). execute () ĊĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠ Ġregistry . counter ( deleted Id . with Tag (" application ", Ġapplication )). add ( ids . size . toDouble ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġreturn Ġdeleted Execution Count . toInt () ĊĠ Ġ} Ċ } Ċ