/* Ċ Ġ* ĠCopyright Ġ 2 0 1 7 Ġ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 . q . metrics Ċ Ċ import Ġcom . netflix . spect ator . api . Counter Ċ import Ġcom . netflix . spect ator . api . Registry Ċ import Ġcom . netflix . spin n aker . or ca . q . Application Aware Ċ import Ġcom . netflix . spin n aker . or ca . q . Queue Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . springframework . beans . factory . annotation . Autowired Ċ import Ġorg . springframework . beans . factory . annotation . Qualifier Ċ import Ġorg . springframework . boot . autoconfigure . condition . Conditional On Bean Ċ import Ġorg . springframework . context . Application Listener Ċ import Ġorg . springframework . scheduling . annotation . Scheduled Ċ import Ġorg . springframework . stereotype . Component Ċ import Ġjava . time . Clock Ċ import Ġjava . time . Duration Ċ import Ġjava . time . Instant Ċ import Ġjava . util . concurrent . atomic . Atomic Reference Ċ import Ġjavax . annotation . Post Construct Ċ Ċ /** Ċ Ġ* ĠMonitor s Ġa Ġqueue Ġand Ġgenerates ĠAtlas Ġmetrics . Ċ Ġ*/ Ċ @ Component Ċ @ Conditional On Bean ( Monitor able Queue :: class ) Ċ open Ġclass ĠAtlas Queue Monitor Ċ @ Autowired Ġconstructor ( ĊĠ Ġ@ Qualifier (" queue Impl ") Ġprivate Ġval Ġqueue : ĠMonitor able Queue , ĊĠ Ġprivate Ġval Ġregistry : ĠRegistry , ĊĠ Ġprivate Ġval Ġclock : ĠClock Ċ ) Ġ: ĠApplication Listener < Queue Event > Ġ{ ĊĊĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( java Class ) ĊĊĠ Ġoverride Ġfun Ġon Application Event ( event : ĠQueue Event ) Ġ{ ĊĠĠĠ Ġwhen Ġ( event ) Ġ{ ĊĠĠĠĠĠ Ġis ĠQueue Pol led Ġ-> Ġ_ last Queue Poll . set ( event . instant ) ĊĠĠĠĠĠ Ġis ĠRetry Pol led Ġ-> Ġ_ last Retry Poll . set ( event . instant ) ĊĠĠĠĠĠ Ġis ĠMessage P ushed Ġ-> Ġevent . counter . increment () ĊĠĠĠĠĠ Ġis ĠMessage Ack nowled ged Ġ-> Ġevent . counter . increment () ĊĠĠĠĠĠ Ġis ĠMessage Ret ried Ġ-> Ġevent . counter . increment () ĊĠĠĠĠĠ Ġis ĠMessage Dead Ġ-> Ġevent . counter . increment () ĊĠĠĠĠĠ Ġis ĠMessage Duplicate Ġ-> Ġevent . counter . increment () ĊĠĠĠĠĠ Ġis ĠLock Failed Ġ-> Ġevent . counter . increment () ĊĠĠĠĠĠ Ġis ĠMessage Res chedule d Ġ-> Ġevent . counter . increment () ĊĠĠĠĠĠ Ġis ĠMessage NotFound Ġ-> Ġevent . counter . increment () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ@ Scheduled ( fixed Delay String Ġ= Ġ"\ ${ queue . depth . metric . frequency : 3 0 0 0 0 }") ĊĠ Ġfun Ġpoll Queue State () Ġ{ ĊĠĠĠ Ġ_ last State . set ( queue . read State ()) ĊĠ Ġ} ĊĊĠ Ġ@ Post Construct Ġfun Ġregister G au ges () Ġ{ ĊĠĠĠ Ġlog . info (" Monitor able Ġqueue Ġimplementation Ġ$ queue Ġfound . ĠExport ing Ġmetrics Ġto ĠAtlas .") ĊĊĠĠĠ Ġregistry . gauge (" queue . depth ", Ġthis , Ġ{ ĊĠĠĠĠĠ Ġit . last State . depth . toDouble () ĊĠĠĠ Ġ}) ĊĠĠĠ Ġregistry . gauge (" queue . un ack ed . depth ", Ġthis , Ġ{ ĊĠĠĠĠĠ Ġit . last State . un ack ed . toDouble () ĊĠĠĠ Ġ}) ĊĠĠĠ Ġregistry . gauge (" queue . ready . depth ", Ġthis , Ġ{ ĊĠĠĠĠĠ Ġit . last State . ready . toDouble () ĊĠĠĠ Ġ}) ĊĠĠĠ Ġregistry . gauge (" queue . orphan ed . messages ", Ġthis , Ġ{ ĊĠĠĠĠĠ Ġit . last State . orphan ed . toDouble () ĊĠĠĠ Ġ}) ĊĠĠĠ Ġregistry . gauge (" queue . last . poll . age ", Ġthis , Ġ{ ĊĠĠĠĠĠ ĠDuration ĊĠĠĠĠĠĠĠ Ġ. between ( it . last Queue Poll , Ġclock . instant ()) ĊĠĠĠĠĠĠĠ Ġ. to Millis () ĊĠĠĠĠĠĠĠ Ġ. toDouble () ĊĠĠĠ Ġ}) ĊĠĠĠ Ġregistry . gauge (" queue . last . retry . check . age ", Ġthis , Ġ{ ĊĠĠĠĠĠ ĠDuration ĊĠĠĠĠĠĠĠ Ġ. between ( it . last Retry Poll , Ġclock . instant ()) ĊĠĠĠĠĠĠĠ Ġ. to Millis () ĊĠĠĠĠĠĠĠ Ġ. toDouble () ĊĠĠĠ Ġ}) ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠThe Ġlast Ġtime Ġthe Ġ[ Queue . poll ] Ġmethod Ġwas Ġexecuted . ĊĠĠ Ġ*/ ĊĠ Ġval Ġlast Queue Poll : ĠInstant ĊĠĠĠ Ġget () Ġ= Ġ_ last Queue Poll . get () ĊĠ Ġprivate Ġval Ġ_ last Queue Poll Ġ= ĠAtomic Reference < Instant >( clock . instant ()) ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠThe Ġtime Ġthe Ġlast Ġ[ Queue . retry ] Ġmethod Ġwas Ġexecuted . ĊĠĠ Ġ*/ ĊĠ Ġval Ġlast Retry Poll : ĠInstant ĊĠĠĠ Ġget () Ġ= Ġ_ last Retry Poll . get () ĊĠ Ġprivate Ġval Ġ_ last Retry Poll Ġ= ĠAtomic Reference < Instant >( clock . instant ()) ĊĊĠ Ġval Ġlast State : ĠQueue State ĊĠĠĠ Ġget () Ġ= Ġ_ last State . get () ĊĠ Ġprivate Ġval Ġ_ last State Ġ= ĠAtomic Reference < Queue State >( Queue State ( 0 , Ġ 0 , Ġ 0 )) ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠCount Ġof Ġmessages Ġpushed Ġto Ġthe Ġqueue . ĊĠĠ Ġ*/ ĊĠ Ġprivate Ġval ĠMessage P ushed . counter : ĠCounter ĊĠĠĠ Ġget () Ġ= Ġwhen Ġ( payload ) Ġ{ ĊĠĠĠĠĠ Ġis ĠApplication Aware Ġ-> Ġregistry . counter (" queue . pus hed . messages ", Ġ" application ", Ġpayload . application ) ĊĠĠĠĠĠ Ġelse Ġ-> Ġregistry . counter (" queue . pus hed . messages ") ĊĠĠĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠCount Ġof Ġmessages Ġsuccessfully Ġprocessed Ġand Ġacknowled ged . ĊĠĠ Ġ*/ ĊĠ Ġprivate Ġval ĠMessage Ack nowled ged . counter : ĠCounter ĊĠĠĠ Ġget () Ġ= Ġregistry . counter (" queue . ack nowled ged . messages ") ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠCount Ġof Ġmessages Ġthat Ġhave Ġbeen Ġretried . ĠThis Ġdoes Ġnot Ġmean Ġunique ĊĠĠ Ġ* Ġmessages , Ġso Ġretry ing Ġthe Ġsame Ġmessage Ġagain Ġwill Ġstill Ġincrement Ġthis ĊĠĠ Ġ* Ġcount . ĊĠĠ Ġ*/ ĊĠ Ġprivate Ġval ĠMessage Ret ried . counter : ĠCounter ĊĠĠĠ Ġget () Ġ= Ġregistry . counter (" queue . ret ried . messages ") ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠCount Ġof Ġmessages Ġthat Ġhave Ġexceeded Ġ[ Queue . max Retries ] Ġretry ĊĠĠ Ġ* Ġattempts Ġand Ġhave Ġbeen Ġsent Ġto Ġthe Ġdead Ġmessage Ġhandler . ĊĠĠ Ġ*/ ĊĠ Ġprivate Ġval ĠMessage Dead . counter : ĠCounter ĊĠĠĠ Ġget () Ġ= Ġregistry . counter (" queue . dead . messages ") ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠCount Ġof Ġmessages Ġthat Ġhave Ġbeen Ġpushed Ġor Ġre - deliver ed Ġwhile Ġan Ġidentical ĊĠĠ Ġ* Ġmessage Ġis Ġalready Ġon Ġthe Ġqueue . ĊĠĠ Ġ*/ ĊĠ Ġprivate Ġval ĠMessage Duplicate . counter : ĠCounter ĊĠĠĠ Ġget () Ġ= Ġregistry . counter (" queue . duplicate . messages ", Ġ" message Type ", Ġpayload . java Class . simple Name ) ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠCount Ġof Ġattempted Ġmessage Ġreads Ġthat Ġfailed Ġto Ġacquire Ġa Ġlock Ġ( in Ġother ĊĠĠ Ġ* Ġwords , Ġmultiple ĠO rc a Ġinstance Ġtried Ġto Ġread Ġthe Ġsame Ġmessage ). ĊĠĠ Ġ*/ ĊĠ Ġprivate Ġval ĠLock Failed . counter : ĠCounter ĊĠĠĠ Ġget () Ġ= Ġregistry . counter (" queue . lock . failed ") ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠCount Ġof Ġattempted Ġmessage Ġre scheduling Ġthat Ġsucceeded Ġ( in Ġother Ġwords , ĊĠĠ Ġ* Ġthat Ġmessage Ġexisted Ġon Ġthe Ġqueue ). ĊĠĠ Ġ*/ ĊĠ Ġprivate Ġval ĠMessage Res chedule d . counter : ĠCounter ĊĠĠĠ Ġget () Ġ= Ġregistry . counter (" queue . re schedule . succeeded ") ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠCount Ġof Ġattempted Ġmessage Ġre scheduling Ġthat Ġfailed Ġ( in Ġother Ġwords , ĊĠĠ Ġ* Ġthat Ġmessage Ġdid Ġnot Ġexist Ġon Ġthe Ġqueue ). ĊĠĠ Ġ*/ ĊĠ Ġprivate Ġval ĠMessage NotFound . counter : ĠCounter ĊĠĠĠ Ġget () Ġ= Ġregistry . counter (" queue . message . not found ") Ċ } Ċ