/* Ċ Ġ* ĠĠ ĠCopyright Ġ 2 0 2 0 ĠAmazon . com , ĠInc . Ġor Ġits Ġaffiliates . ĠAll ĠRights ĠReserved . Ċ Ġ* Ċ Ġ* ĠĠ ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "). Ċ Ġ* ĠĠ ĠYou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ Ġ* ĠĠ ĠA Ġcopy Ġof Ġthe ĠLicense Ġis Ġlocated Ġat Ċ Ġ* Ċ Ġ* ĠĠĠĠĠĠ Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠĠ Ġor Ġin Ġthe Ġ" license " Ġfile Ġaccompanying Ġthis Ġfile . ĠThis Ġfile Ġ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 . amazon . opend ist ro fore lastic search . alert ing . transport Ċ Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . action . Index Monitor Action Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . action . Index Monitor Request Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . action . Index Monitor Response Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . core . Scheduled Job Indices Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . core . model . Scheduled Job Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . core . model . Scheduled Job . Companion . SCHEDUL ED _ JOB S _ INDEX Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . core . model . Scheduled Job . Companion . SCHEDUL ED _ JOB _ TYPE Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . core . model . Search Input Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . model . Monitor Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . settings . Alert ing Settings Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . settings . Alert ing Settings . Companion . ALERT ING _ MAX _ MON IT ORS Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . settings . Alert ing Settings . Companion . FILTER _ BY _ BACKEND _ RO LES Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . settings . Alert ing Settings . Companion . INDEX _ TIMEOUT Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . settings . Alert ing Settings . Companion . MAX _ ACTION _ TH RO TT LE _ VALUE Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . settings . Alert ing Settings . Companion . REQUEST _ TIMEOUT Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . settings . Destination Settings . Companion . ALLOW _ LIST Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . util . Alert ing Exception Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . util . Index Utils Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . util . add User Backend Roles Filter Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . util . check Filter ByUser Backend Roles Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . util . check User Filter By Permissions Ċ import Ġcom . amazon . opend ist ro fore lastic search . alert ing . util . is AD Monitor Ċ import Ġcom . amazon . opend ist ro fore lastic search . commons . Config Constants Ċ import Ġcom . amazon . opend ist ro fore lastic search . commons . auth user . User Ċ import Ġorg . apache . logging . log 4 j . Log Manager Ċ import Ġorg . elasticsearch . Elasticsearch Security Exception Ċ import Ġorg . elasticsearch . Elasticsearch Status Exception Ċ import Ġorg . elasticsearch . action . ActionListener Ċ import Ġorg . elasticsearch . action . admin . indices . create . Create Index Response Ċ import Ġorg . elasticsearch . action . get . GetRequest Ċ import Ġorg . elasticsearch . action . get . GetResponse Ċ import Ġorg . elasticsearch . action . index . Index Request Ċ import Ġorg . elasticsearch . action . index . Index Response Ċ import Ġorg . elasticsearch . action . search . Search Request Ċ import Ġorg . elasticsearch . action . search . Search Response Ċ import Ġorg . elasticsearch . action . support . Action Filters Ċ import Ġorg . elasticsearch . action . support . Handled Transport Action Ċ import Ġorg . elasticsearch . action . support . master . Ack nowled ged Response Ċ import Ġorg . elasticsearch . client . Client Ċ import Ġorg . elasticsearch . cluster . service . Cluster Service Ċ import Ġorg . elasticsearch . common . inject . Inject Ċ import Ġorg . elasticsearch . common . settings . Settings Ċ import Ġorg . elasticsearch . common . unit . Time Value Ċ import Ġorg . elasticsearch . common . x content . Logging De precation Handler Ċ import Ġorg . elasticsearch . common . x content . Named X Content Registry Ċ import Ġorg . elasticsearch . common . x content . To X Content Ċ import Ġorg . elasticsearch . common . x content . X Content Factory . json Builder Ċ import Ġorg . elasticsearch . common . x content . X Content Helper Ċ import Ġorg . elasticsearch . common . x content . X ContentType Ċ import Ġorg . elasticsearch . index . query . Query Builders Ċ import Ġorg . elasticsearch . rest . Rest Request Ċ import Ġorg . elasticsearch . rest . Rest Status Ċ import Ġorg . elasticsearch . search . builder . Search Source Builder Ċ import Ġorg . elasticsearch . tasks . Task Ċ import Ġorg . elasticsearch . transport . Transport Service Ċ import Ġjava . io . IOException Ċ import Ġjava . time . Duration Ċ Ċ private Ġval Ġlog Ġ= ĠLog Manager . getLogger ( Transport Index Monitor Action :: class . java ) Ċ Ċ class ĠTransport Index Monitor Action Ġ@ Inject Ġconstructor ( ĊĠĠĠ Ġtransport Service : ĠTransport Service , ĊĠĠĠ Ġval Ġclient : ĠClient , ĊĠĠĠ Ġaction Filters : ĠAction Filters , ĊĠĠĠ Ġval Ġscheduled Job Indices : ĠScheduled Job Indices , ĊĠĠĠ Ġval Ġcluster Service : ĠCluster Service , ĊĠĠĠ Ġval Ġsettings : ĠSettings , ĊĠĠĠ Ġval Ġx Content Registry : ĠNamed X Content Registry Ċ ) Ġ: ĠHand led Transport Action < Index Monitor Request , ĠIndex Monitor Response >( ĊĠĠĠĠĠĠĠ ĠIndex Monitor Action . NAME , Ġtransport Service , Ġaction Filters , Ġ:: Index Monitor Request Ċ ) Ġ{ ĊĊĠĠĠ Ġ@ Volatile Ġprivate Ġvar Ġmax Monitor s Ġ= ĠAL ERT ING _ MAX _ MON IT ORS . get ( settings ) ĊĠĠĠ Ġ@ Volatile Ġprivate Ġvar Ġrequest Timeout Ġ= ĠREQUEST _ TIMEOUT . get ( settings ) ĊĠĠĠ Ġ@ Volatile Ġprivate Ġvar Ġindex Timeout Ġ= ĠINDEX _ TIMEOUT . get ( settings ) ĊĠĠĠ Ġ@ Volatile Ġprivate Ġvar Ġmax Action Throttle Ġ= ĠMAX _ ACTION _ TH RO TT LE _ VALUE . get ( settings ) ĊĠĠĠ Ġ@ Volatile Ġprivate Ġvar Ġallow List Ġ= ĠALLOW _ LIST . get ( settings ) ĊĠĠĠ Ġ@ Volatile Ġprivate Ġvar Ġfilter By Enabled Ġ= ĠAlert ing Settings . FILTER _ BY _ BACKEND _ RO LES . get ( settings ) ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġcluster Service . cluster Settings . add Settings Update Consumer ( ALERT ING _ MAX _ MON IT ORS ) Ġ{ Ġmax Monitor s Ġ= Ġit Ġ} ĊĠĠĠĠĠĠĠ Ġcluster Service . cluster Settings . add Settings Update Consumer ( REQUEST _ TIMEOUT ) Ġ{ Ġrequest Timeout Ġ= Ġit Ġ} ĊĠĠĠĠĠĠĠ Ġcluster Service . cluster Settings . add Settings Update Consumer ( INDEX _ TIMEOUT ) Ġ{ Ġindex Timeout Ġ= Ġit Ġ} ĊĠĠĠĠĠĠĠ Ġcluster Service . cluster Settings . add Settings Update Consumer ( MAX _ ACTION _ TH RO TT LE _ VALUE ) Ġ{ Ġmax Action Throttle Ġ= Ġit Ġ} ĊĠĠĠĠĠĠĠ Ġcluster Service . cluster Settings . add Settings Update Consumer ( ALLOW _ LIST ) Ġ{ Ġallow List Ġ= Ġit Ġ} ĊĠĠĠĠĠĠĠ Ġcluster Service . cluster Settings . add Settings Update Consumer ( FILTER _ BY _ BACKEND _ RO LES ) Ġ{ Ġfilter By Enabled Ġ= Ġit Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġdo Execute ( task : ĠTask , Ġrequest : ĠIndex Monitor Request , Ġaction Listener : ĠActionListener < Index Monitor Response >) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġuser Str Ġ= Ġclient . thread Pool (). thread Context . get Transient < String >( Config Constants . OPEN DIST RO _ SECURITY _ USER _ INFO _ THREAD _ CONTEXT ) ĊĠĠĠĠĠĠĠ Ġlog . debug (" User Ġand Ġroles Ġstring Ġfrom Ġthread Ġcontext : Ġ$ user Str ") ĊĠĠĠĠĠĠĠ Ġval Ġuser : ĠUser ? Ġ= ĠUser . parse ( user Str ) ĊĊĠĠĠĠĠĠĠ Ġif Ġ(! check Filter ByUser Backend Roles ( filter By Enabled , Ġuser , Ġaction Listener )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ(! is AD Monitor ( request . monitor )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheck Indices And Execute ( client , Ġaction Listener , Ġrequest , Ġuser ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġcheck Ġif Ġuser Ġhas Ġaccess Ġto Ġany Ġanomaly Ġdetector Ġfor ĠAD Ġmonitor ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheck An omaly Detector And Execute ( client , Ġaction Listener , Ġrequest , Ġuser ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ ĠCheck Ġif Ġuser Ġhas Ġpermissions Ġto Ġread Ġthe Ġconfigured Ġindices Ġon Ġthe Ġmonitor Ġand ĊĠĠĠĠ Ġ* Ġ Ġthen Ġcreate Ġmonitor . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġcheck Indices And Execute ( ĊĠĠĠĠĠĠĠ Ġclient : ĠClient , ĊĠĠĠĠĠĠĠ Ġaction Listener : ĠActionListener < Index Monitor Response >, ĊĠĠĠĠĠĠĠ Ġrequest : ĠIndex Monitor Request , ĊĠĠĠĠĠĠĠ Ġuser : ĠUser ? ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġindices Ġ= Ġmutable ListOf < String >() ĊĠĠĠĠĠĠĠ Ġval Ġsearch Inputs Ġ= Ġrequest . monitor . inputs . filter Ġ{ Ġit . name () Ġ== ĠSearch Input . SEARCH _ FIELD Ġ} ĊĠĠĠĠĠĠĠ Ġsearch Inputs . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsearch Input Ġ= Ġit Ġas ĠSearch Input ĊĠĠĠĠĠĠĠĠĠĠĠ Ġindices . addAll ( search Input . indices ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġsearch Request Ġ= ĠSearch Request (). indices (* indices . to Typed Array ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. source ( Search Source Builder . search Source (). size ( 1 ). query ( Query Builders . match All Query ())) ĊĠĠĠĠĠĠĠ Ġclient . search ( search Request , Ġobject Ġ: ĠActionListener < Search Response > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Response ( search Response : ĠSearch Response ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠUser Ġhas Ġread Ġaccess Ġto Ġconfigured Ġindices Ġin Ġthe Ġmonitor , Ġnow Ġcreate Ġmonitor Ġwith Ġout Ġuser Ġcontext . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient . thread Pool (). thread Context . stash Context (). use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠIndex Monitor Handler ( client , Ġaction Listener , Ġrequest , Ġuser ). resolve User And Start () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ ĠDue Ġto Ġbelow Ġissue Ġwith Ġsecurity Ġplugin , Ġwe Ġget Ġsecurity _ exception Ġwhen Ġinvalid Ġindex Ġname Ġis Ġmentioned . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ Ġhttps :// github . com / opend ist ro - for - elasticsearch / security / issues / 7 1 8 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( t : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( t Ġis ĠElasticsearch Security Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtrue Ġ-> ĠElasticsearch Status Exception (" User Ġdoesn 't Ġhave Ġread Ġpermissions Ġfor Ġone Ġor Ġmore Ġconfigured Ġindex Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"$ indices ", ĠRest Status . FOR BID DEN ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfalse Ġ-> Ġt ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠIt 's Ġno Ġreasonable Ġto Ġcreate ĠAD Ġmonitor Ġif Ġthe Ġuser Ġhas Ġno Ġaccess Ġto Ġany Ġdetector . ĠOtherwise ĊĠĠĠĠ Ġ* Ġthe Ġmonitor Ġwill Ġnot Ġget Ġany Ġanomaly Ġresult . ĠSo Ġwe Ġwill Ġcheck Ġuser Ġhas Ġaccess Ġto Ġat Ġleast Ġ 1 ĊĠĠĠĠ Ġ* Ġanomaly Ġdetector Ġif Ġthey Ġneed Ġto Ġcreate ĠAD Ġmonitor . ĊĠĠĠĠ Ġ* ĠAs Ġanomaly Ġdetector Ġindex Ġis Ġsystem Ġindex , Ġcommon Ġuser Ġhas Ġno Ġpermission Ġto Ġquery . ĠSo Ġwe Ġneed ĊĠĠĠĠ Ġ* Ġto Ġsend ĠREST ĠAPI Ġcall Ġto ĠAD ĠREST ĠAPI . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġcheck An omaly Detector And Execute ( ĊĠĠĠĠĠĠĠ Ġclient : ĠClient , ĊĠĠĠĠĠĠĠ Ġaction Listener : ĠActionListener < Index Monitor Response >, ĊĠĠĠĠĠĠĠ Ġrequest : ĠIndex Monitor Request , ĊĠĠĠĠĠĠĠ Ġuser : ĠUser ? ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġclient . thread Pool (). thread Context . stash Context (). use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠIndex Monitor Handler ( client , Ġaction Listener , Ġrequest , Ġuser ). resolve User And Start For AD () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinner Ġclass ĠIndex Monitor Handler ( ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġclient : ĠClient , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġaction Listener : ĠActionListener < Index Monitor Response >, ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġrequest : ĠIndex Monitor Request , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġuser : ĠUser ? ĊĠĠĠ Ġ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġfun Ġresolve User And Start () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( user Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠSecurity Ġis Ġdisabled , Ġadd Ġempty Ġuser Ġto ĠMonitor . Ġuser Ġis Ġnull Ġfor Ġolder Ġversions . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequest . monitor Ġ= Ġrequest . monitor ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. copy ( user Ġ= ĠUser ("", ĠlistOf (), ĠlistOf (), ĠlistOf ())) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequest . monitor Ġ= Ġrequest . monitor ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. copy ( user Ġ= ĠUser ( user . name , Ġuser . backend Roles , Ġuser . roles , Ġuser . custom Att Names )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun Ġresolve User And Start For AD () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( user Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠSecurity Ġis Ġdisabled , Ġadd Ġempty Ġuser Ġto ĠMonitor . Ġuser Ġis Ġnull Ġfor Ġolder Ġversions . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequest . monitor Ġ= Ġrequest . monitor ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. copy ( user Ġ= ĠUser ("", ĠlistOf (), ĠlistOf (), ĠlistOf ())) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequest . monitor Ġ= Ġrequest . monitor ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. copy ( user Ġ= ĠUser ( user . name , Ġuser . backend Roles , Ġuser . roles , Ġuser . custom Att Names )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsearch Source Builder Ġ= ĠSearch Source Builder (). size ( 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġadd User Backend Roles Filter ( user , Ġsearch Source Builder ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsearch Request Ġ= ĠSearch Request (). indices (". opend ist ro - anomaly - detector s "). source ( search Source Builder ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient . search ( search Request , Ġobject Ġ: ĠActionListener < Search Response > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Response ( response : ĠSearch Response ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtotal Hits Ġ= Ġresponse ?. hits ?. total Hits ?. value ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( total Hits Ġ!= Ġnull Ġ&& Ġtotal Hits Ġ> Ġ 0 L ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠElasticsearch Status Exception (" User Ġhas Ġno Ġavailable Ġdetector s ", ĠRest Status . NOT _ FOUND ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( t : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( t )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( ex )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun Ġstart () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! scheduled Job Indices . scheduled Job Index Exists ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġscheduled Job Indices . init Scheduled Job Index ( object Ġ: ĠActionListener < Create Index Response > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Response ( response : ĠCreate Index Response ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠonCreate Mappings Response ( response ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( t : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( t )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ(! Index Utils . scheduled Job Index Updated ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠIndex Utils . update Index Mapping ( SCHEDUL ED _ JOB S _ INDEX , ĠS CH ED UL ED _ JOB _ TYPE , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠScheduled Job Indices . scheduled Job Mappings (), Ġcluster Service . state (), Ġclient . admin (). indices (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġobject Ġ: ĠActionListener < Ack nowled ged Response > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Response ( response : ĠAck nowled ged Response ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠonUpdate Mappings Response ( response ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( t : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( t )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprepare Monitor Indexing () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠThis Ġfunction Ġpre pa res Ġfor Ġindexing Ġa Ġnew Ġmonitor . ĊĠĠĠĠĠĠĠĠ Ġ* ĠIf Ġthis Ġis Ġan Ġupdate Ġrequest Ġwe Ġcan Ġsimply Ġupdate Ġthe Ġmonitor . ĠOtherwise Ġwe Ġfirst Ġcheck Ġto Ġsee Ġhow Ġmany Ġmonitor s Ġalready Ġexist , ĊĠĠĠĠĠĠĠĠ Ġ* Ġand Ġcompare Ġthis Ġto Ġthe Ġ[ max Monitor Count ]. ĠRequests Ġthat Ġb reach Ġthis Ġthreshold Ġwill Ġbe Ġrejected . ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġprepare Monitor Indexing () Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠBelow Ġcheck Ġneeds Ġto Ġbe Ġasync Ġoperations Ġand Ġneeds Ġto Ġbe Ġref actored Ġissue # 2 6 9 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġcheck For Dis allowed Dest inations ( allow List ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvalidate Action Throttle ( request . monitor , Ġmax Action Throttle , ĠTime Value . time Value Minutes ( 1 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠRuntimeException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( e )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( request . method Ġ== ĠRest Request . Method . PUT ) Ġreturn Ġupdate Monitor () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġquery Ġ= ĠQuery Builders . bool Query (). filter ( Query Builders . term Query ("${ Monitor . MONITOR _ TYPE }. type ", ĠMonitor . MONITOR _ TYPE )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsearch Source Ġ= ĠSearch Source Builder (). query ( query ). timeout ( request Timeout ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsearch Request Ġ= ĠSearch Request ( SCHEDUL ED _ JOB S _ INDEX ). source ( search Source ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclient . search ( search Request , Ġobject Ġ: ĠActionListener < Search Response > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Response ( search Response : ĠSearch Response ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Search Response ( search Response ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( t : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( t )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġvalidate Action Throttle ( monitor : ĠMonitor , ĠmaxValue : ĠTime Value , Ġmin Value : ĠTime Value ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmonitor . triggers . forEach Ġ{ Ġtrigger Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtrigger . actions . forEach Ġ{ Ġaction Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( action . throttle Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequire ( Time Value ( Duration . of ( action . throttle . value . to Long (), Ġaction . throttle . unit ). to Millis ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. compareTo ( max Value ) Ġ<= Ġ 0 , Ġ{ Ġ" Can Ġonly Ġset Ġthrottle Ġperiod Ġless Ġthan Ġor Ġequal Ġto Ġ$ max Value " Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequire ( Time Value ( Duration . of ( action . throttle . value . to Long (), Ġaction . throttle . unit ). to Millis ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. compareTo ( min Value ) Ġ>= Ġ 0 , Ġ{ Ġ" Can Ġonly Ġset Ġthrottle Ġperiod Ġgreater Ġthan Ġor Ġequal Ġto Ġ$ min Value " Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠAfter Ġsearching Ġfor Ġall Ġexisting Ġmonitor s Ġwe Ġvalidate Ġthe Ġsystem Ġcan Ġsupport Ġanother Ġmonitor Ġto Ġbe Ġcreated . ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġon Search Response ( response : ĠSearch Response ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtotal Hits Ġ= Ġresponse . hits . total Hits ?. value ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( total Hits Ġ!= Ġnull Ġ&& Ġtotal Hits Ġ>= Ġmax Monitor s ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" This Ġrequest Ġwould Ġcreate Ġmore Ġthan Ġthe Ġallowed Ġmonitor s Ġ[$ max Monitor s ]. ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠAlert ing Exception . wrap ( IllegalArgumentException ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" This Ġrequest Ġwould Ġcreate Ġmore Ġthan Ġthe Ġallowed Ġmonitor s Ġ[$ max Monitor s ]. ")) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġindex Monitor () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun ĠonCreate Mappings Response ( response : ĠCreate Index Response ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( response . is Ack nowled ged ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Created Ġ$ SCHEDUL ED _ JOB S _ INDEX Ġwith Ġmappings .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprepare Monitor Indexing () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠIndex Utils . scheduled Job Index Updated () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Create Ġ$ SCHEDUL ED _ JOB S _ INDEX Ġmappings Ġcall Ġnot Ġacknowled ged .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( Elasticsearch Status Exception ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Create Ġ$ SCHEDUL ED _ JOB S _ INDEX Ġmappings Ġcall Ġnot Ġacknowled ged ", ĠRest Status . INTERNAL _ SERVER _ ERROR )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun ĠonUpdate Mappings Response ( response : ĠAck nowled ged Response ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( response . is Ack nowled ged ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Updated Ġ Ġ${ Scheduled Job . SCHEDUL ED _ JOB S _ INDEX } Ġwith Ġmappings .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠIndex Utils . scheduled Job Index Updated () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprepare Monitor Indexing () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Update Ġ${ Scheduled Job . SCHEDUL ED _ JOB S _ INDEX } Ġmappings Ġcall Ġnot Ġacknowled ged .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( Elasticsearch Status Exception ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Updated Ġ${ Scheduled Job . SCHEDUL ED _ JOB S _ INDEX } Ġmappings Ġcall Ġnot Ġacknowled ged .", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRest Status . INTERNAL _ SERVER _ ERROR )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġindex Monitor () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequest . monitor Ġ= Ġrequest . monitor . copy ( schema Version Ġ= ĠIndex Utils . scheduled Job Index Schema Version ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġindex Request Ġ= ĠIndex Request ( SCHEDUL ED _ JOB S _ INDEX ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set Refresh Policy ( request . refresh Policy ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. source ( request . monitor . to X Content ( json Builder (), ĠTo X Content . Map Params ( map Of (" with _ type " Ġto Ġ" true ")) )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set If Seq No ( request . seq No ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set If Primary Term ( request . primary Term ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. timeout ( index Timeout ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclient . index ( index Request , Ġobject Ġ: ĠActionListener < Index Response > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Response ( response : ĠIndex Response ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfailure Re asons Ġ= Ġcheck Shards Failure ( response ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( failure Re asons Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠAlert ing Exception . wrap ( Elasticsearch Status Exception ( failure Re asons . toString (), Ġresponse . status ()))) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Response ( Index Monitor Response ( response . id , Ġresponse . version , Ġresponse . seq No , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresponse . primary Term , ĠRest Status . CREATED , Ġrequest . monitor )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( t : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( t )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġupdate Monitor () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġget Request Ġ= ĠGet Request ( SCHEDUL ED _ JOB S _ INDEX , Ġrequest . monitor Id ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclient . get ( getRequest , Ġobject Ġ: ĠActionListener < GetResponse > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Response ( response : ĠGet Response ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! response . is Exists ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠElasticsearch Status Exception (" Monitor Ġwith Ġ${ request . monitor Id } Ġis Ġnot Ġfound ", ĠRest Status . NOT _ FOUND ))) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġx cp Ġ= ĠX Content Helper . create Parser ( x Content Registry , ĠLogging De precation Handler . INSTANCE , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresponse . source As Bytes Ref , ĠX ContentType . JSON ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmonitor Ġ= ĠScheduled Job . parse ( x cp , Ġresponse . id , Ġresponse . version ) Ġas ĠMonitor ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon GetResponse ( monitor ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( t : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( t )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġon GetResponse ( current Monitor : ĠMonitor ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! check User Filter By Permissions ( filter By Enabled , Ġuser , Ġcurrent Monitor . user , Ġaction Listener , Ġ" monitor ", Ġrequest . monitor Id )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIf Ġboth Ġare Ġenabled , Ġuse Ġthe Ġcurrent Ġexisting Ġmonitor Ġenabled Ġtime , Ġotherwise Ġthe Ġnext Ġexecution Ġwill Ġbe ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġincorrect . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( request . monitor . enabled Ġ&& Ġcurrent Monitor . enabled ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequest . monitor Ġ= Ġrequest . monitor . copy ( enabled Time Ġ= Ġcurrent Monitor . enabled Time ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequest . monitor Ġ= Ġrequest . monitor . copy ( schema Version Ġ= ĠIndex Utils . scheduled Job Index Schema Version ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġindex Request Ġ= ĠIndex Request ( SCHEDUL ED _ JOB S _ INDEX ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set Refresh Policy ( request . refresh Policy ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. source ( request . monitor . to X Content ( json Builder (), ĠTo X Content . Map Params ( map Of (" with _ type " Ġto Ġ" true ")) )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. id ( request . monitor Id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set If Seq No ( request . seq No ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set If Primary Term ( request . primary Term ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. timeout ( index Timeout ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġclient . index ( index Request , Ġobject Ġ: ĠActionListener < Index Response > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Response ( response : ĠIndex Response ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfailure Re asons Ġ= Ġcheck Shards Failure ( response ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( failure Re asons Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠAlert ing Exception . wrap ( Elasticsearch Status Exception ( failure Re asons . toString (), Ġresponse . status ()))) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Response ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠIndex Monitor Response ( response . id , Ġresponse . version , Ġresponse . seq No , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresponse . primary Term , ĠRest Status . CREATED , Ġrequest . monitor ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( t : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Listener . on Failure ( Alert ing Exception . wrap ( t )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġcheck Shards Failure ( response : ĠIndex Response ): ĠString ? Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfailure Re asons Ġ= ĠStringBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( response . shard Info . failed Ġ> Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresponse . shard Info . failures . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġentry Ġ-> Ġfailure Re asons . append ( entry . reason ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfailure Re asons . toString () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ