/* Ċ Ġ* ĠTencent Ġis Ġplease d Ġto Ġsupport Ġthe Ġopen Ġsource Ġcommunity Ġby Ġmaking ĠB K - CI Ġ èĵĿ é² ¸ æĮģ ç»Ń éĽĨ æĪĲ å¹³åı° Ġavailable . ĠĠĊ Ġ* Ċ Ġ* ĠCopyright Ġ( C ) Ġ 2 0 1 9 ĠTH L ĠA 2 9 ĠLimited , Ġa ĠTencent Ġcompany . Ġ ĠAll Ġrights Ġreserved . Ċ Ġ* Ċ Ġ* ĠB K - CI Ġ èĵĿ é² ¸ æĮģ ç»Ń éĽĨ æĪĲ å¹³åı° Ġis Ġlicensed Ġunder Ġthe ĠMIT Ġlicense . Ċ Ġ* Ċ Ġ* ĠA Ġcopy Ġof Ġthe ĠMIT ĠLicense Ġis Ġincluded Ġin Ġthis Ġfile . Ċ Ġ* Ċ Ġ* Ċ Ġ* ĠTerms Ġof Ġthe ĠMIT ĠLicense : Ċ Ġ* Ġ------------------------------------------------ --- Ċ Ġ* ĠPermission Ġis Ġhereby Ġgranted , Ġfree Ġof Ġcharge , Ġto Ġany Ġperson Ġobtaining Ġa Ġcopy Ġof Ġthis Ġsoftware Ġand Ġassociated Ġdocumentation Ġfiles Ġ( the Ġ" Software "), Ġto Ġdeal Ġin Ġthe ĠSoftware Ġwithout Ġrestriction , Ġincluding Ġwithout Ġlimitation Ġthe Ġrights Ġto Ġuse , Ġcopy , Ġmodify , Ġmerge , Ġpublish , Ġdistribute , Ġsublicense , Ġand / or Ġsell Ġcopies Ġof Ġthe ĠSoftware , Ġand Ġto Ġpermit Ġpersons Ġto Ġwhom Ġthe ĠSoftware Ġis Ġfurnished Ġto Ġdo Ġso , Ġsubject Ġto Ġthe Ġfollowing Ġconditions : Ċ Ġ* Ċ Ġ* ĠThe Ġabove Ġcopyright Ġnotice Ġand Ġthis Ġpermission Ġnotice Ġshall Ġbe Ġincluded Ġin Ġall Ġcopies Ġor Ġsubstantial Ġportions Ġof Ġthe ĠSoftware . Ċ Ġ* Ċ Ġ* ĠTHE ĠSOFTWARE ĠIS ĠPROVIDED Ġ" AS ĠIS ", ĠWITHOUT ĠWARRANTY ĠOF ĠANY ĠKIND , ĠEXPRESS ĠOR ĠIMPLIED , ĠINCLUDING ĠBUT ĠNOT ĠLIMITED ĠTO ĠTHE ĠWARRANTIES ĠOF ĠMERCHANTABILITY , ĠFITNESS ĠFOR ĠA ĠPARTICULAR ĠPURPOSE ĠAND ĠNONINFRINGEMENT . ĠIN ĠNO ĠEVENT ĠSHALL ĠTHE ĠAUTHORS ĠOR ĠCOPYRIGHT ĠHOLDERS ĠBE ĠLIABLE ĠFOR ĠANY ĠCLAIM , ĠDAMAGES ĠOR ĠOTHER ĠLIABILITY , ĠWHETHER ĠIN ĠAN ĠACTION ĠOF ĠCONTRACT , ĠTORT ĠOR ĠOTHERWISE , ĠARISING ĠFROM , ĠOUT ĠOF ĠOR ĠIN ĠCONNECTION ĠWITH ĠTHE ĠSOFTWARE ĠOR ĠTHE ĠUSE ĠOR ĠOTHER ĠDEALINGS ĠIN ĠTHE ĠSOFTWARE . Ċ Ġ* Ċ Ġ* Ċ Ġ*/ Ċ Ċ package Ġcom . tencent . bk repo . monitor . export Ċ Ċ import Ġcom . tencent . bk repo . monitor . metrics . Metric Info Ċ import Ġcom . tencent . bk repo . monitor . service . Metric Source Service Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . springframework . http . MediaType Ċ import Ġorg . springframework . stereotype . Component Ċ import Ġorg . springframework . web . reactive . function . Body Inser ters Ċ import Ġorg . springframework . web . reactive . function . client . Web Client Ċ import Ġreactor . core . publisher . Flux Ċ Ċ @ Component Ċ class ĠIn flux Metrics Exporter ( ĊĠĠĠ Ġmetric Source Service : ĠMetric Source Service , ĊĠĠĠ Ġprivate Ġval Ġin flux Export Properties : ĠIn flux Export Properties Ċ ) Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġconverter Ġ= ĠIn flux Metrics Converter () ĊĠĠĠ Ġprivate Ġval Ġweb Client : ĠWeb Client ĊĠĠĠ Ġprivate Ġvar Ġdatabase Exists : ĠBoolean Ġ= Ġfalse ĊĠĠĠ Ġprivate Ġvar Ġin flux Endpoint : ĠString Ġ= Ġ"" ĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġwith ( influx Export Properties ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġin flux Endpoint Ġ= Ġ"/ write ? consistency =" Ġ+ Ġconsistency . toLowerCase () Ġ+ Ġ"& precision = ms & db =" Ġ+ Ġdb ĊĠĠĠĠĠĠĠĠĠĠĠ Ġretention Policy ?. let Ġ{ Ġin flux Endpoint Ġ+= Ġ"& rp =$ retention Policy " Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbuilder Ġ= ĠWeb Client . builder (). baseUrl ( uri ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! username . isNull Or Blank ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbuilder . default Headers Ġ{ Ġit . set Basic Auth ( username . or Empty (), Ġpassword . or Empty ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġweb Client Ġ= Ġbuilder . build () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmetric Source Service . get Merged Source (). window ( step ). subscribe Ġ{ Ġexport Metric Source ( it ) Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġexport Metric Source ( metric Source : ĠFlux < Metric Info >) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! influx Export Properties . enabled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġcreate Database If Ne cessary () ĊĠĠĠĠĠĠĠ Ġval Ġstring Source Ġ= Ġmetric Source . map Ġ{ Ġconverter . convert ( it ) Ġ} ĊĠĠĠĠĠĠĠ Ġweb Client . post () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. uri ( influx Endpoint ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. contentType ( MediaType . TEXT _ PLAIN ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. body ( Body Inser ters . from Publisher ( string Source , ĠString :: class . java )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. exchange () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġclient Response Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient Response . body To Mono ( String :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Success Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( client Response . statusCode (). isError ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Failed Ġto Ġexport Ġmetrics Ġto Ġin flux ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" HttpStatusCode Ġ= Ġ{}", Ġclient Response . statusCode ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" HttpHeaders Ġ= Ġ{}", Ġclient Response . headers (). as HttpHeaders ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" ResponseBody Ġ= Ġ{}", Ġit ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. subscribe () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate Database If Ne cessary () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! influx Export Properties . auto Create Db Ġ|| Ġdatabase Exists ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġcreate Database Query Ġ= ĠCreate Database QueryBuilder ( influx Export Properties . db ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. set Retention Duration ( influx Export Properties . retention Duration ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. set RetentionPolicy Name ( influx Export Properties . retention Policy ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. set Retention Replication Factor ( influx Export Properties . retention Replication Factor ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. set Retention Shard Duration ( influx Export Properties . retention Shard Duration ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠĠĠĠĠ Ġweb Client . post () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. uri ("/ query ? q =$ create Database Query ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. exchange () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġclient Response Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient Response . body To Mono ( String :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Success Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( client Response . statusCode (). isError ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Unable Ġto Ġcreate Ġdatabase Ġ'{} '", Ġin flux Export Properties . db ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" HttpStatusCode Ġ= Ġ{}", Ġclient Response . statusCode ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" HttpHeaders Ġ= Ġ{}", Ġclient Response . headers (). as HttpHeaders ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" ResponseBody Ġ= Ġ{}", Ġit ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" In flux Ġdatabase Ġ{} Ġis Ġready Ġto Ġreceive Ġmetrics ", Ġin flux Export Properties . db ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdatabase Exists Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. block () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( In flux Metrics Exporter :: class . java ) ĊĠĠĠ Ġ} Ċ } Ċ