rrdcontext.c 147 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235323632373238323932403241324232433244324532463247324832493250325132523253325432553256325732583259326032613262326332643265326632673268326932703271327232733274327532763277327832793280328132823283328432853286328732883289329032913292329332943295329632973298329933003301330233033304330533063307330833093310331133123313331433153316331733183319332033213322332333243325332633273328332933303331333233333334333533363337333833393340334133423343334433453346334733483349335033513352335333543355335633573358335933603361336233633364336533663367336833693370337133723373337433753376337733783379338033813382338333843385338633873388338933903391339233933394339533963397339833993400340134023403340434053406340734083409341034113412341334143415341634173418341934203421342234233424342534263427342834293430343134323433343434353436343734383439344034413442344334443445344634473448344934503451345234533454345534563457345834593460346134623463346434653466346734683469347034713472347334743475347634773478347934803481348234833484348534863487348834893490349134923493349434953496349734983499350035013502350335043505350635073508350935103511351235133514351535163517351835193520352135223523352435253526352735283529353035313532353335343535353635373538353935403541354235433544354535463547354835493550355135523553355435553556355735583559356035613562356335643565356635673568356935703571357235733574357535763577357835793580358135823583358435853586358735883589359035913592359335943595359635973598359936003601360236033604360536063607360836093610361136123613361436153616361736183619362036213622362336243625362636273628362936303631363236333634363536363637363836393640364136423643364436453646364736483649365036513652365336543655365636573658365936603661366236633664366536663667366836693670367136723673367436753676367736783679368036813682368336843685368636873688368936903691369236933694369536963697369836993700370137023703370437053706370737083709371037113712371337143715371637173718371937203721372237233724372537263727372837293730373137323733373437353736373737383739374037413742374337443745374637473748374937503751375237533754375537563757375837593760376137623763376437653766376737683769377037713772377337743775377637773778377937803781378237833784378537863787378837893790379137923793379437953796379737983799380038013802380338043805380638073808380938103811381238133814381538163817381838193820382138223823382438253826382738283829383038313832383338343835383638373838383938403841384238433844384538463847384838493850385138523853385438553856385738583859386038613862386338643865386638673868386938703871387238733874387538763877387838793880388138823883388438853886388738883889389038913892389338943895389638973898389939003901390239033904390539063907390839093910391139123913391439153916
  1. // SPDX-License-Identifier: GPL-3.0-or-later
  2. #include "rrdcontext.h"
  3. #include "sqlite/sqlite_context.h"
  4. #include "aclk/schema-wrappers/context.h"
  5. #include "aclk/aclk_contexts_api.h"
  6. #include "aclk/aclk.h"
  7. #include "storage_engine.h"
  8. #define MESSAGES_PER_BUNDLE_TO_SEND_TO_HUB_PER_HOST 5000
  9. #define FULL_RETENTION_SCAN_DELAY_AFTER_DB_ROTATION_SECS 120
  10. #define RRDCONTEXT_WORKER_THREAD_HEARTBEAT_USEC (1000 * USEC_PER_MS)
  11. #define RRDCONTEXT_MINIMUM_ALLOWED_PRIORITY 10
  12. #define LOG_TRANSITIONS false
  13. #define WORKER_JOB_HOSTS 1
  14. #define WORKER_JOB_CHECK 2
  15. #define WORKER_JOB_SEND 3
  16. #define WORKER_JOB_DEQUEUE 4
  17. #define WORKER_JOB_RETENTION 5
  18. #define WORKER_JOB_QUEUED 6
  19. #define WORKER_JOB_CLEANUP 7
  20. #define WORKER_JOB_CLEANUP_DELETE 8
  21. #define WORKER_JOB_PP_METRIC 9 // post-processing metrics
  22. #define WORKER_JOB_PP_INSTANCE 10 // post-processing instances
  23. #define WORKER_JOB_PP_CONTEXT 11 // post-processing contexts
  24. #define WORKER_JOB_HUB_QUEUE_SIZE 12
  25. #define WORKER_JOB_PP_QUEUE_SIZE 13
  26. typedef enum {
  27. RRD_FLAG_NONE = 0,
  28. RRD_FLAG_DELETED = (1 << 0), // this is a deleted object (metrics, instances, contexts)
  29. RRD_FLAG_COLLECTED = (1 << 1), // this object is currently being collected
  30. RRD_FLAG_UPDATED = (1 << 2), // this object has updates to propagate
  31. RRD_FLAG_ARCHIVED = (1 << 3), // this object is not currently being collected
  32. RRD_FLAG_OWN_LABELS = (1 << 4), // this instance has its own labels - not linked to an RRDSET
  33. RRD_FLAG_LIVE_RETENTION = (1 << 5), // we have got live retention from the database
  34. RRD_FLAG_QUEUED_FOR_HUB = (1 << 6), // this context is currently queued to be dispatched to hub
  35. RRD_FLAG_QUEUED_FOR_POST_PROCESSING = (1 << 7), // this context is currently queued to be post-processed
  36. RRD_FLAG_HIDDEN = (1 << 8), // don't expose this to the hub or the API
  37. RRD_FLAG_UPDATE_REASON_TRIGGERED = (1 << 9), // the update was triggered by the child object
  38. RRD_FLAG_UPDATE_REASON_LOAD_SQL = (1 << 10), // this object has just been loaded from SQL
  39. RRD_FLAG_UPDATE_REASON_NEW_OBJECT = (1 << 11), // this object has just been created
  40. RRD_FLAG_UPDATE_REASON_UPDATED_OBJECT = (1 << 12), // we received an update on this object
  41. RRD_FLAG_UPDATE_REASON_CHANGED_LINKING = (1 << 13), // an instance or a metric switched RRDSET or RRDDIM
  42. RRD_FLAG_UPDATE_REASON_CHANGED_UUID = (1 << 14), // an instance or a metric changed UUID
  43. RRD_FLAG_UPDATE_REASON_CHANGED_NAME = (1 << 15), // an instance or a metric changed name
  44. RRD_FLAG_UPDATE_REASON_CHANGED_UNITS = (1 << 16), // this context or instance changed units
  45. RRD_FLAG_UPDATE_REASON_CHANGED_TITLE = (1 << 17), // this context or instance changed title
  46. RRD_FLAG_UPDATE_REASON_CHANGED_FAMILY = (1 << 18), // the context or the instance changed family
  47. RRD_FLAG_UPDATE_REASON_CHANGED_CHART_TYPE = (1 << 19), // this context or instance changed chart type
  48. RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY = (1 << 20), // this context or instance changed its priority
  49. RRD_FLAG_UPDATE_REASON_CHANGED_UPDATE_EVERY = (1 << 21), // the instance or the metric changed update frequency
  50. RRD_FLAG_UPDATE_REASON_ZERO_RETENTION = (1 << 22), // this object has not retention
  51. RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T = (1 << 23), // this object changed its oldest time in the db
  52. RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T = (1 << 24), // this object change its latest time in the db
  53. RRD_FLAG_UPDATE_REASON_STOPPED_BEING_COLLECTED = (1 << 25), // this object has stopped being collected
  54. RRD_FLAG_UPDATE_REASON_STARTED_BEING_COLLECTED = (1 << 26), // this object has started being collected
  55. RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD = (1 << 27), // this context belongs to a host that just disconnected
  56. RRD_FLAG_UPDATE_REASON_DB_ROTATION = (1 << 28), // this context changed because of a db rotation
  57. RRD_FLAG_UPDATE_REASON_UNUSED = (1 << 29), // this context is not used anymore
  58. RRD_FLAG_UPDATE_REASON_CHANGED_FLAGS = (1 << 30), // this context is not used anymore
  59. RRD_FLAG_UPDATE_REASON_UPDATED_RETENTION = (1 << 31), // this object has updated retention
  60. } RRD_FLAGS;
  61. #define RRD_FLAG_ALL_UPDATE_REASONS ( \
  62. RRD_FLAG_UPDATE_REASON_TRIGGERED \
  63. |RRD_FLAG_UPDATE_REASON_LOAD_SQL \
  64. |RRD_FLAG_UPDATE_REASON_NEW_OBJECT \
  65. |RRD_FLAG_UPDATE_REASON_UPDATED_OBJECT \
  66. |RRD_FLAG_UPDATE_REASON_CHANGED_LINKING \
  67. |RRD_FLAG_UPDATE_REASON_CHANGED_UUID \
  68. |RRD_FLAG_UPDATE_REASON_CHANGED_NAME \
  69. |RRD_FLAG_UPDATE_REASON_CHANGED_UNITS \
  70. |RRD_FLAG_UPDATE_REASON_CHANGED_TITLE \
  71. |RRD_FLAG_UPDATE_REASON_CHANGED_FAMILY \
  72. |RRD_FLAG_UPDATE_REASON_CHANGED_CHART_TYPE \
  73. |RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY \
  74. |RRD_FLAG_UPDATE_REASON_CHANGED_UPDATE_EVERY \
  75. |RRD_FLAG_UPDATE_REASON_ZERO_RETENTION \
  76. |RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T \
  77. |RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T \
  78. |RRD_FLAG_UPDATE_REASON_STOPPED_BEING_COLLECTED \
  79. |RRD_FLAG_UPDATE_REASON_STARTED_BEING_COLLECTED \
  80. |RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD \
  81. |RRD_FLAG_UPDATE_REASON_DB_ROTATION \
  82. |RRD_FLAG_UPDATE_REASON_UNUSED \
  83. |RRD_FLAG_UPDATE_REASON_CHANGED_FLAGS \
  84. )
  85. #define RRD_FLAGS_ALLOWED_EXTERNALLY_ON_NEW_OBJECTS ( \
  86. RRD_FLAG_ARCHIVED \
  87. |RRD_FLAG_HIDDEN \
  88. |RRD_FLAG_ALL_UPDATE_REASONS \
  89. )
  90. #define RRD_FLAGS_REQUIRED_FOR_DELETIONS ( \
  91. RRD_FLAG_DELETED \
  92. |RRD_FLAG_LIVE_RETENTION \
  93. )
  94. #define RRD_FLAGS_PREVENTING_DELETIONS ( \
  95. RRD_FLAG_QUEUED_FOR_HUB \
  96. |RRD_FLAG_COLLECTED \
  97. |RRD_FLAG_QUEUED_FOR_POST_PROCESSING \
  98. )
  99. // get all the flags of an object
  100. #define rrd_flags_get(obj) __atomic_load_n(&((obj)->flags), __ATOMIC_SEQ_CST)
  101. // check if ANY of the given flags (bits) is set
  102. #define rrd_flag_check(obj, flag) (rrd_flags_get(obj) & (flag))
  103. // check if ALL the given flags (bits) are set
  104. #define rrd_flag_check_all(obj, flag) (rrd_flag_check(obj, flag) == (flag))
  105. // set one or more flags (bits)
  106. #define rrd_flag_set(obj, flag) __atomic_or_fetch(&((obj)->flags), flag, __ATOMIC_SEQ_CST)
  107. // clear one or more flags (bits)
  108. #define rrd_flag_clear(obj, flag) __atomic_and_fetch(&((obj)->flags), ~(flag), __ATOMIC_SEQ_CST)
  109. // replace the flags of an object, with the supplied ones
  110. #define rrd_flags_replace(obj, all_flags) __atomic_store_n(&((obj)->flags), all_flags, __ATOMIC_SEQ_CST)
  111. static inline void
  112. rrd_flag_add_remove_atomic(RRD_FLAGS *flags, RRD_FLAGS check, RRD_FLAGS conditionally_add, RRD_FLAGS always_remove) {
  113. RRD_FLAGS expected, desired;
  114. do {
  115. expected = *flags;
  116. desired = expected;
  117. desired &= ~(always_remove);
  118. if(!(expected & check))
  119. desired |= (check | conditionally_add);
  120. } while(!__atomic_compare_exchange_n(flags, &expected, desired, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST));
  121. }
  122. #define rrd_flag_set_collected(obj) \
  123. rrd_flag_add_remove_atomic(&((obj)->flags) \
  124. /* check this flag */ \
  125. , RRD_FLAG_COLLECTED \
  126. \
  127. /* add these flags together with the above, if the above is not already set */ \
  128. , RRD_FLAG_UPDATE_REASON_STARTED_BEING_COLLECTED | RRD_FLAG_UPDATED \
  129. \
  130. /* always remove these flags */ \
  131. , RRD_FLAG_ARCHIVED \
  132. | RRD_FLAG_DELETED \
  133. | RRD_FLAG_UPDATE_REASON_STOPPED_BEING_COLLECTED \
  134. | RRD_FLAG_UPDATE_REASON_ZERO_RETENTION \
  135. | RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD \
  136. )
  137. #define rrd_flag_set_archived(obj) \
  138. rrd_flag_add_remove_atomic(&((obj)->flags) \
  139. /* check this flag */ \
  140. , RRD_FLAG_ARCHIVED \
  141. \
  142. /* add these flags together with the above, if the above is not already set */ \
  143. , RRD_FLAG_UPDATE_REASON_STOPPED_BEING_COLLECTED | RRD_FLAG_UPDATED \
  144. \
  145. /* always remove these flags */ \
  146. , RRD_FLAG_COLLECTED \
  147. | RRD_FLAG_DELETED \
  148. | RRD_FLAG_UPDATE_REASON_STARTED_BEING_COLLECTED \
  149. | RRD_FLAG_UPDATE_REASON_ZERO_RETENTION \
  150. )
  151. #define rrd_flag_set_deleted(obj, reason) \
  152. rrd_flag_add_remove_atomic(&((obj)->flags) \
  153. /* check this flag */ \
  154. , RRD_FLAG_DELETED \
  155. \
  156. /* add these flags together with the above, if the above is not already set */ \
  157. , RRD_FLAG_UPDATE_REASON_ZERO_RETENTION | RRD_FLAG_UPDATED | (reason) \
  158. \
  159. /* always remove these flags */ \
  160. , RRD_FLAG_ARCHIVED \
  161. | RRD_FLAG_COLLECTED \
  162. )
  163. #define rrd_flag_is_collected(obj) rrd_flag_check(obj, RRD_FLAG_COLLECTED)
  164. #define rrd_flag_is_archived(obj) rrd_flag_check(obj, RRD_FLAG_ARCHIVED)
  165. #define rrd_flag_is_deleted(obj) rrd_flag_check(obj, RRD_FLAG_DELETED)
  166. #define rrd_flag_is_updated(obj) rrd_flag_check(obj, RRD_FLAG_UPDATED)
  167. // mark an object as updated, providing reasons (additional bits)
  168. #define rrd_flag_set_updated(obj, reason) rrd_flag_set(obj, RRD_FLAG_UPDATED | (reason))
  169. // clear an object as being updated, clearing also all the reasons
  170. #define rrd_flag_unset_updated(obj) rrd_flag_clear(obj, RRD_FLAG_UPDATED | RRD_FLAG_ALL_UPDATE_REASONS)
  171. static struct rrdcontext_reason {
  172. RRD_FLAGS flag;
  173. const char *name;
  174. usec_t delay_ut;
  175. } rrdcontext_reasons[] = {
  176. // context related
  177. { RRD_FLAG_UPDATE_REASON_TRIGGERED, "triggered transition", 65 * USEC_PER_SEC },
  178. { RRD_FLAG_UPDATE_REASON_NEW_OBJECT, "object created", 65 * USEC_PER_SEC },
  179. { RRD_FLAG_UPDATE_REASON_UPDATED_OBJECT, "object updated", 65 * USEC_PER_SEC },
  180. { RRD_FLAG_UPDATE_REASON_LOAD_SQL, "loaded from sql", 65 * USEC_PER_SEC },
  181. { RRD_FLAG_UPDATE_REASON_CHANGED_TITLE, "changed title", 65 * USEC_PER_SEC },
  182. { RRD_FLAG_UPDATE_REASON_CHANGED_UNITS, "changed units", 65 * USEC_PER_SEC },
  183. { RRD_FLAG_UPDATE_REASON_CHANGED_FAMILY, "changed family", 65 * USEC_PER_SEC },
  184. { RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY, "changed priority", 65 * USEC_PER_SEC },
  185. { RRD_FLAG_UPDATE_REASON_ZERO_RETENTION, "has no retention", 65 * USEC_PER_SEC },
  186. { RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T, "updated first_time_t", 65 * USEC_PER_SEC },
  187. { RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T, "updated last_time_t", 65 * USEC_PER_SEC },
  188. { RRD_FLAG_UPDATE_REASON_CHANGED_CHART_TYPE, "changed chart type", 65 * USEC_PER_SEC },
  189. { RRD_FLAG_UPDATE_REASON_STOPPED_BEING_COLLECTED, "stopped collected", 65 * USEC_PER_SEC },
  190. { RRD_FLAG_UPDATE_REASON_STARTED_BEING_COLLECTED, "started collected", 5 * USEC_PER_SEC },
  191. { RRD_FLAG_UPDATE_REASON_UNUSED, "unused", 5 * USEC_PER_SEC },
  192. // not context related
  193. { RRD_FLAG_UPDATE_REASON_CHANGED_UUID, "changed uuid", 65 * USEC_PER_SEC },
  194. { RRD_FLAG_UPDATE_REASON_CHANGED_UPDATE_EVERY, "changed updated every",65 * USEC_PER_SEC },
  195. { RRD_FLAG_UPDATE_REASON_CHANGED_LINKING, "changed rrd link", 65 * USEC_PER_SEC },
  196. { RRD_FLAG_UPDATE_REASON_CHANGED_NAME, "changed name", 65 * USEC_PER_SEC },
  197. { RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD, "child disconnected", 65 * USEC_PER_SEC },
  198. { RRD_FLAG_UPDATE_REASON_DB_ROTATION, "db rotation", 65 * USEC_PER_SEC },
  199. { RRD_FLAG_UPDATE_REASON_CHANGED_FLAGS, "changed flags", 65 * USEC_PER_SEC },
  200. { RRD_FLAG_UPDATE_REASON_UPDATED_RETENTION, "updated retention", 65 * USEC_PER_SEC },
  201. // terminator
  202. { 0, NULL, 0 },
  203. };
  204. typedef struct rrdmetric {
  205. uuid_t uuid;
  206. STRING *id;
  207. STRING *name;
  208. RRDDIM *rrddim;
  209. time_t first_time_t;
  210. time_t last_time_t;
  211. RRD_FLAGS flags;
  212. struct rrdinstance *ri;
  213. } RRDMETRIC;
  214. typedef struct rrdinstance {
  215. uuid_t uuid;
  216. STRING *id;
  217. STRING *name;
  218. STRING *title;
  219. STRING *units;
  220. STRING *family;
  221. uint32_t priority;
  222. RRDSET_TYPE chart_type;
  223. RRD_FLAGS flags; // flags related to this instance
  224. time_t first_time_t;
  225. time_t last_time_t;
  226. int update_every; // data collection frequency
  227. RRDSET *rrdset; // pointer to RRDSET when collected, or NULL
  228. DICTIONARY *rrdlabels; // linked to RRDSET->chart_labels or own version
  229. struct rrdcontext *rc;
  230. DICTIONARY *rrdmetrics;
  231. struct {
  232. uint32_t collected_metrics_count; // a temporary variable to detect BEGIN/END without SET
  233. // don't use it for other purposes
  234. // it goes up and then resets to zero, on every iteration
  235. } internal;
  236. } RRDINSTANCE;
  237. typedef struct rrdcontext {
  238. uint64_t version;
  239. STRING *id;
  240. STRING *title;
  241. STRING *units;
  242. STRING *family;
  243. uint32_t priority;
  244. RRDSET_TYPE chart_type;
  245. RRD_FLAGS flags;
  246. time_t first_time_t;
  247. time_t last_time_t;
  248. VERSIONED_CONTEXT_DATA hub;
  249. DICTIONARY *rrdinstances;
  250. RRDHOST *rrdhost;
  251. struct {
  252. RRD_FLAGS queued_flags; // the last flags that triggered the post-processing
  253. usec_t queued_ut; // the last time this was queued
  254. usec_t dequeued_ut; // the last time we sent (or deduplicated) this context
  255. size_t executions; // how many times this context has been processed
  256. } pp;
  257. struct {
  258. RRD_FLAGS queued_flags; // the last flags that triggered the queueing
  259. usec_t queued_ut; // the last time this was queued
  260. usec_t delay_calc_ut; // the last time we calculated the scheduled_dispatched_ut
  261. usec_t scheduled_dispatch_ut; // the time it was/is scheduled to be sent
  262. usec_t dequeued_ut; // the last time we sent (or deduplicated) this context
  263. size_t dispatches; // the number of times this has been dispatched to hub
  264. } queue;
  265. netdata_mutex_t mutex;
  266. } RRDCONTEXT;
  267. // ----------------------------------------------------------------------------
  268. // helper one-liners for RRDMETRIC
  269. static void rrdmetric_update_retention(RRDMETRIC *rm);
  270. static inline RRDMETRIC *rrdmetric_acquired_value(RRDMETRIC_ACQUIRED *rma) {
  271. return dictionary_acquired_item_value((DICTIONARY_ITEM *)rma);
  272. }
  273. static inline RRDMETRIC_ACQUIRED *rrdmetric_acquired_dup(RRDMETRIC_ACQUIRED *rma) {
  274. RRDMETRIC *rm = rrdmetric_acquired_value(rma);
  275. return (RRDMETRIC_ACQUIRED *)dictionary_acquired_item_dup(rm->ri->rrdmetrics, (DICTIONARY_ITEM *)rma);
  276. }
  277. static inline void rrdmetric_release(RRDMETRIC_ACQUIRED *rma) {
  278. RRDMETRIC *rm = rrdmetric_acquired_value(rma);
  279. dictionary_acquired_item_release(rm->ri->rrdmetrics, (DICTIONARY_ITEM *)rma);
  280. }
  281. const char *rrdmetric_acquired_id(RRDMETRIC_ACQUIRED *rma) {
  282. RRDMETRIC *rm = rrdmetric_acquired_value(rma);
  283. return string2str(rm->id);
  284. }
  285. const char *rrdmetric_acquired_name(RRDMETRIC_ACQUIRED *rma) {
  286. RRDMETRIC *rm = rrdmetric_acquired_value(rma);
  287. return string2str(rm->name);
  288. }
  289. NETDATA_DOUBLE rrdmetric_acquired_last_stored_value(RRDMETRIC_ACQUIRED *rma) {
  290. RRDMETRIC *rm = rrdmetric_acquired_value(rma);
  291. if(rm->rrddim)
  292. return rm->rrddim->last_stored_value;
  293. return NAN;
  294. }
  295. // ----------------------------------------------------------------------------
  296. // helper one-liners for RRDINSTANCE
  297. static inline RRDINSTANCE *rrdinstance_acquired_value(RRDINSTANCE_ACQUIRED *ria) {
  298. return dictionary_acquired_item_value((DICTIONARY_ITEM *)ria);
  299. }
  300. static inline RRDINSTANCE_ACQUIRED *rrdinstance_acquired_dup(RRDINSTANCE_ACQUIRED *ria) {
  301. RRDINSTANCE *ri = rrdinstance_acquired_value(ria);
  302. return (RRDINSTANCE_ACQUIRED *)dictionary_acquired_item_dup(ri->rc->rrdinstances, (DICTIONARY_ITEM *)ria);
  303. }
  304. static inline void rrdinstance_release(RRDINSTANCE_ACQUIRED *ria) {
  305. RRDINSTANCE *ri = rrdinstance_acquired_value(ria);
  306. dictionary_acquired_item_release(ri->rc->rrdinstances, (DICTIONARY_ITEM *)ria);
  307. }
  308. const char *rrdinstance_acquired_id(RRDINSTANCE_ACQUIRED *ria) {
  309. RRDINSTANCE *ri = rrdinstance_acquired_value(ria);
  310. return string2str(ri->id);
  311. }
  312. const char *rrdinstance_acquired_name(RRDINSTANCE_ACQUIRED *ria) {
  313. RRDINSTANCE *ri = rrdinstance_acquired_value(ria);
  314. return string2str(ri->name);
  315. }
  316. DICTIONARY *rrdinstance_acquired_labels(RRDINSTANCE_ACQUIRED *ria) {
  317. RRDINSTANCE *ri = rrdinstance_acquired_value(ria);
  318. return ri->rrdlabels;
  319. }
  320. DICTIONARY *rrdinstance_acquired_functions(RRDINSTANCE_ACQUIRED *ria) {
  321. RRDINSTANCE *ri = rrdinstance_acquired_value(ria);
  322. if(!ri->rrdset) return NULL;
  323. return ri->rrdset->functions_view;
  324. }
  325. // ----------------------------------------------------------------------------
  326. // helper one-liners for RRDCONTEXT
  327. static inline RRDCONTEXT *rrdcontext_acquired_value(RRDCONTEXT_ACQUIRED *rca) {
  328. return dictionary_acquired_item_value((DICTIONARY_ITEM *)rca);
  329. }
  330. const char *rrdcontext_acquired_id(RRDCONTEXT_ACQUIRED *rca) {
  331. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  332. return string2str(rc->id);
  333. }
  334. static inline RRDCONTEXT_ACQUIRED *rrdcontext_acquired_dup(RRDCONTEXT_ACQUIRED *rca) {
  335. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  336. return (RRDCONTEXT_ACQUIRED *)dictionary_acquired_item_dup((DICTIONARY *)rc->rrdhost->rrdctx, (DICTIONARY_ITEM *)rca);
  337. }
  338. static inline void rrdcontext_release(RRDCONTEXT_ACQUIRED *rca) {
  339. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  340. dictionary_acquired_item_release((DICTIONARY *)rc->rrdhost->rrdctx, (DICTIONARY_ITEM *)rca);
  341. }
  342. static void rrdcontext_recalculate_context_retention(RRDCONTEXT *rc, RRD_FLAGS reason, bool worker_jobs);
  343. static void rrdcontext_recalculate_host_retention(RRDHOST *host, RRD_FLAGS reason, bool worker_jobs);
  344. #define rrdcontext_version_hash(host) rrdcontext_version_hash_with_callback(host, NULL, false, NULL)
  345. static uint64_t rrdcontext_version_hash_with_callback(RRDHOST *host, void (*callback)(RRDCONTEXT *, bool, void *), bool snapshot, void *bundle);
  346. static void rrdcontext_garbage_collect_single_host(RRDHOST *host, bool worker_jobs);
  347. static void rrdcontext_garbage_collect_for_all_hosts(void);
  348. #define rrdcontext_lock(rc) netdata_mutex_lock(&((rc)->mutex))
  349. #define rrdcontext_unlock(rc) netdata_mutex_unlock(&((rc)->mutex))
  350. // ----------------------------------------------------------------------------
  351. // Forward definitions
  352. static uint64_t rrdcontext_get_next_version(RRDCONTEXT *rc);
  353. static bool check_if_cloud_version_changed_unsafe(RRDCONTEXT *rc, bool sending __maybe_unused);
  354. static void rrdcontext_message_send_unsafe(RRDCONTEXT *rc, bool snapshot __maybe_unused, void *bundle __maybe_unused);
  355. static void rrdcontext_delete_from_sql_unsafe(RRDCONTEXT *rc);
  356. static void rrdcontext_dequeue_from_post_processing(RRDCONTEXT *rc);
  357. static void rrdcontext_queue_for_post_processing(RRDCONTEXT *rc, const char *function, RRD_FLAGS flags);
  358. static void rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAGS reason, bool worker_jobs);
  359. static void rrdmetric_trigger_updates(RRDMETRIC *rm, const char *function);
  360. static void rrdinstance_trigger_updates(RRDINSTANCE *ri, const char *function);
  361. static void rrdcontext_trigger_updates(RRDCONTEXT *rc, const char *function);
  362. // ----------------------------------------------------------------------------
  363. // visualizing flags
  364. static void rrd_flags_to_buffer(RRD_FLAGS flags, BUFFER *wb) {
  365. if(flags & RRD_FLAG_QUEUED_FOR_HUB)
  366. buffer_strcat(wb, "QUEUED ");
  367. if(flags & RRD_FLAG_DELETED)
  368. buffer_strcat(wb, "DELETED ");
  369. if(flags & RRD_FLAG_COLLECTED)
  370. buffer_strcat(wb, "COLLECTED ");
  371. if(flags & RRD_FLAG_UPDATED)
  372. buffer_strcat(wb, "UPDATED ");
  373. if(flags & RRD_FLAG_ARCHIVED)
  374. buffer_strcat(wb, "ARCHIVED ");
  375. if(flags & RRD_FLAG_OWN_LABELS)
  376. buffer_strcat(wb, "OWN_LABELS ");
  377. if(flags & RRD_FLAG_LIVE_RETENTION)
  378. buffer_strcat(wb, "LIVE_RETENTION ");
  379. if(flags & RRD_FLAG_HIDDEN)
  380. buffer_strcat(wb, "HIDDEN ");
  381. if(flags & RRD_FLAG_QUEUED_FOR_POST_PROCESSING)
  382. buffer_strcat(wb, "PENDING_UPDATES ");
  383. }
  384. static void rrd_reasons_to_buffer(RRD_FLAGS flags, BUFFER *wb) {
  385. for(int i = 0, added = 0; rrdcontext_reasons[i].name ; i++) {
  386. if (flags & rrdcontext_reasons[i].flag) {
  387. if (added)
  388. buffer_strcat(wb, ", ");
  389. buffer_strcat(wb, rrdcontext_reasons[i].name);
  390. added++;
  391. }
  392. }
  393. }
  394. // ----------------------------------------------------------------------------
  395. // RRDMETRIC
  396. // free the contents of RRDMETRIC.
  397. // RRDMETRIC itself is managed by DICTIONARY - no need to free it here.
  398. static void rrdmetric_free(RRDMETRIC *rm) {
  399. string_freez(rm->id);
  400. string_freez(rm->name);
  401. rm->id = NULL;
  402. rm->name = NULL;
  403. rm->ri = NULL;
  404. }
  405. // called when this rrdmetric is inserted to the rrdmetrics dictionary of a rrdinstance
  406. // the constructor of the rrdmetric object
  407. static void rrdmetric_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *rrdinstance) {
  408. RRDMETRIC *rm = value;
  409. // link it to its parent
  410. rm->ri = rrdinstance;
  411. // remove flags that we need to figure out at runtime
  412. rm->flags = rm->flags & RRD_FLAGS_ALLOWED_EXTERNALLY_ON_NEW_OBJECTS; // no need for atomics
  413. // signal the react callback to do the job
  414. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_NEW_OBJECT);
  415. }
  416. // called when this rrdmetric is deleted from the rrdmetrics dictionary of a rrdinstance
  417. // the destructor of the rrdmetric object
  418. static void rrdmetric_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *rrdinstance __maybe_unused) {
  419. RRDMETRIC *rm = value;
  420. internal_error(rm->rrddim, "RRDMETRIC: '%s' is freed but there is a RRDDIM linked to it.", string2str(rm->id));
  421. // free the resources
  422. rrdmetric_free(rm);
  423. }
  424. // called when the same rrdmetric is inserted again to the rrdmetrics dictionary of a rrdinstance
  425. // while this is called, the dictionary is write locked, but there may be other users of the object
  426. static bool rrdmetric_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *old_value, void *new_value, void *rrdinstance __maybe_unused) {
  427. RRDMETRIC *rm = old_value;
  428. RRDMETRIC *rm_new = new_value;
  429. internal_error(rm->id != rm_new->id,
  430. "RRDMETRIC: '%s' cannot change id to '%s'",
  431. string2str(rm->id), string2str(rm_new->id));
  432. if(uuid_compare(rm->uuid, rm_new->uuid) != 0) {
  433. char uuid1[UUID_STR_LEN], uuid2[UUID_STR_LEN];
  434. uuid_unparse(rm->uuid, uuid1);
  435. uuid_unparse(rm_new->uuid, uuid2);
  436. internal_error(true, "RRDMETRIC: '%s' of instance '%s' changed uuid from '%s' to '%s'", string2str(rm->id), string2str(rm->ri->id), uuid1, uuid2);
  437. uuid_copy(rm->uuid, rm_new->uuid);
  438. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_UUID);
  439. }
  440. if(rm->rrddim && rm_new->rrddim && rm->rrddim != rm_new->rrddim) {
  441. rm->rrddim = rm_new->rrddim;
  442. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_LINKING);
  443. }
  444. if(rm->rrddim && uuid_compare(rm->uuid, rm->rrddim->metric_uuid) != 0) {
  445. char uuid1[UUID_STR_LEN], uuid2[UUID_STR_LEN];
  446. uuid_unparse(rm->uuid, uuid1);
  447. uuid_unparse(rm_new->uuid, uuid2);
  448. internal_error(true, "RRDMETRIC: '%s' is linked to RRDDIM '%s' but they have different UUIDs. RRDMETRIC has '%s', RRDDIM has '%s'", string2str(rm->id), rrddim_id(rm->rrddim), uuid1, uuid2);
  449. }
  450. if(rm->rrddim != rm_new->rrddim)
  451. rm->rrddim = rm_new->rrddim;
  452. if(rm->name != rm_new->name) {
  453. STRING *old = rm->name;
  454. rm->name = string_dup(rm_new->name);
  455. string_freez(old);
  456. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_NAME);
  457. }
  458. if(!rm->first_time_t || (rm_new->first_time_t && rm_new->first_time_t < rm->first_time_t)) {
  459. rm->first_time_t = rm_new->first_time_t;
  460. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
  461. }
  462. if(!rm->last_time_t || (rm_new->last_time_t && rm_new->last_time_t > rm->last_time_t)) {
  463. rm->last_time_t = rm_new->last_time_t;
  464. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
  465. }
  466. rrd_flag_set(rm, rm_new->flags & RRD_FLAGS_ALLOWED_EXTERNALLY_ON_NEW_OBJECTS); // no needs for atomics on rm_new
  467. if(rrd_flag_is_collected(rm) && rrd_flag_is_archived(rm))
  468. rrd_flag_set_collected(rm);
  469. if(rrd_flag_check(rm, RRD_FLAG_UPDATED))
  470. rrd_flag_set(rm, RRD_FLAG_UPDATE_REASON_UPDATED_OBJECT);
  471. rrdmetric_free(rm_new);
  472. // the react callback will continue from here
  473. return rrd_flag_is_updated(rm);
  474. }
  475. // this is called after the insert or the conflict callbacks,
  476. // but the dictionary is now unlocked
  477. static void rrdmetric_react_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *rrdinstance __maybe_unused) {
  478. RRDMETRIC *rm = value;
  479. rrdmetric_trigger_updates(rm, __FUNCTION__ );
  480. }
  481. static void rrdmetrics_create_in_rrdinstance(RRDINSTANCE *ri) {
  482. if(unlikely(!ri)) return;
  483. if(likely(ri->rrdmetrics)) return;
  484. ri->rrdmetrics = dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE);
  485. dictionary_register_insert_callback(ri->rrdmetrics, rrdmetric_insert_callback, ri);
  486. dictionary_register_delete_callback(ri->rrdmetrics, rrdmetric_delete_callback, ri);
  487. dictionary_register_conflict_callback(ri->rrdmetrics, rrdmetric_conflict_callback, ri);
  488. dictionary_register_react_callback(ri->rrdmetrics, rrdmetric_react_callback, ri);
  489. }
  490. static void rrdmetrics_destroy_from_rrdinstance(RRDINSTANCE *ri) {
  491. if(unlikely(!ri || !ri->rrdmetrics)) return;
  492. dictionary_destroy(ri->rrdmetrics);
  493. ri->rrdmetrics = NULL;
  494. }
  495. // trigger post-processing of the rrdmetric, escalating changes to the rrdinstance it belongs
  496. static void rrdmetric_trigger_updates(RRDMETRIC *rm, const char *function) {
  497. if(unlikely(rrd_flag_is_collected(rm)) && (!rm->rrddim || rrd_flag_check(rm, RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD)))
  498. rrd_flag_set_archived(rm);
  499. if(rrd_flag_is_updated(rm) || !rrd_flag_check(rm, RRD_FLAG_LIVE_RETENTION)) {
  500. rrd_flag_set_updated(rm->ri, RRD_FLAG_UPDATE_REASON_TRIGGERED);
  501. rrdcontext_queue_for_post_processing(rm->ri->rc, function, rm->flags);
  502. }
  503. }
  504. // ----------------------------------------------------------------------------
  505. // RRDMETRIC HOOKS ON RRDDIM
  506. static inline void rrdmetric_from_rrddim(RRDDIM *rd) {
  507. if(unlikely(!rd->rrdset))
  508. fatal("RRDMETRIC: rrddim '%s' does not have a rrdset.", rrddim_id(rd));
  509. if(unlikely(!rd->rrdset->rrdhost))
  510. fatal("RRDMETRIC: rrdset '%s' does not have a rrdhost", rrdset_id(rd->rrdset));
  511. if(unlikely(!rd->rrdset->rrdinstance))
  512. fatal("RRDMETRIC: rrdset '%s' does not have a rrdinstance", rrdset_id(rd->rrdset));
  513. RRDINSTANCE *ri = rrdinstance_acquired_value(rd->rrdset->rrdinstance);
  514. RRDMETRIC trm = {
  515. .id = string_dup(rd->id),
  516. .name = string_dup(rd->name),
  517. .flags = RRD_FLAG_NONE, // no need for atomics
  518. .rrddim = rd,
  519. };
  520. uuid_copy(trm.uuid, rd->metric_uuid);
  521. RRDMETRIC_ACQUIRED *rma = (RRDMETRIC_ACQUIRED *)dictionary_set_and_acquire_item(ri->rrdmetrics, string2str(trm.id), &trm, sizeof(trm));
  522. if(rd->rrdmetric)
  523. rrdmetric_release(rd->rrdmetric);
  524. rd->rrdmetric = rma;
  525. }
  526. #define rrddim_get_rrdmetric(rd) rrddim_get_rrdmetric_with_trace(rd, __FUNCTION__)
  527. static inline RRDMETRIC *rrddim_get_rrdmetric_with_trace(RRDDIM *rd, const char *function) {
  528. if(unlikely(!rd->rrdmetric)) {
  529. error("RRDMETRIC: RRDDIM '%s' is not linked to an RRDMETRIC at %s()", rrddim_id(rd), function);
  530. return NULL;
  531. }
  532. RRDMETRIC *rm = rrdmetric_acquired_value(rd->rrdmetric);
  533. if(unlikely(!rm)) {
  534. error("RRDMETRIC: RRDDIM '%s' lost the link to its RRDMETRIC at %s()", rrddim_id(rd), function);
  535. return NULL;
  536. }
  537. if(unlikely(rm->rrddim != rd))
  538. fatal("RRDMETRIC: '%s' is not linked to RRDDIM '%s' at %s()", string2str(rm->id), rrddim_id(rd), function);
  539. return rm;
  540. }
  541. static inline void rrdmetric_rrddim_is_freed(RRDDIM *rd) {
  542. RRDMETRIC *rm = rrddim_get_rrdmetric(rd);
  543. if(unlikely(!rm)) return;
  544. if(unlikely(rrd_flag_is_collected(rm)))
  545. rrd_flag_set_archived(rm);
  546. rm->rrddim = NULL;
  547. rrdmetric_trigger_updates(rm, __FUNCTION__ );
  548. rrdmetric_release(rd->rrdmetric);
  549. rd->rrdmetric = NULL;
  550. }
  551. static inline void rrdmetric_updated_rrddim_flags(RRDDIM *rd) {
  552. RRDMETRIC *rm = rrddim_get_rrdmetric(rd);
  553. if(unlikely(!rm)) return;
  554. if(unlikely(rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED|RRDDIM_FLAG_OBSOLETE))) {
  555. if(unlikely(rrd_flag_is_collected(rm)))
  556. rrd_flag_set_archived(rm);
  557. }
  558. rrdmetric_trigger_updates(rm, __FUNCTION__ );
  559. }
  560. static inline void rrdmetric_collected_rrddim(RRDDIM *rd) {
  561. RRDMETRIC *rm = rrddim_get_rrdmetric(rd);
  562. if(unlikely(!rm)) return;
  563. if(unlikely(!rrd_flag_is_collected(rm)))
  564. rrd_flag_set_collected(rm);
  565. // we use this variable to detect BEGIN/END without SET
  566. rm->ri->internal.collected_metrics_count++;
  567. rrdmetric_trigger_updates(rm, __FUNCTION__ );
  568. }
  569. // ----------------------------------------------------------------------------
  570. // RRDINSTANCE
  571. static void rrdinstance_free(RRDINSTANCE *ri) {
  572. if(rrd_flag_check(ri, RRD_FLAG_OWN_LABELS))
  573. dictionary_destroy(ri->rrdlabels);
  574. rrdmetrics_destroy_from_rrdinstance(ri);
  575. string_freez(ri->id);
  576. string_freez(ri->name);
  577. string_freez(ri->title);
  578. string_freez(ri->units);
  579. string_freez(ri->family);
  580. ri->id = NULL;
  581. ri->name = NULL;
  582. ri->title = NULL;
  583. ri->units = NULL;
  584. ri->family = NULL;
  585. ri->rc = NULL;
  586. ri->rrdlabels = NULL;
  587. ri->rrdmetrics = NULL;
  588. ri->rrdset = NULL;
  589. }
  590. static void rrdinstance_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *rrdcontext) {
  591. static STRING *ml_anomaly_rates_id = NULL;
  592. if(unlikely(!ml_anomaly_rates_id))
  593. ml_anomaly_rates_id = string_strdupz(ML_ANOMALY_RATES_CHART_ID);
  594. RRDINSTANCE *ri = value;
  595. // link it to its parent
  596. ri->rc = rrdcontext;
  597. ri->flags = ri->flags & RRD_FLAGS_ALLOWED_EXTERNALLY_ON_NEW_OBJECTS; // no need for atomics
  598. if(!ri->name)
  599. ri->name = string_dup(ri->id);
  600. if(ri->rrdset) {
  601. ri->rrdlabels = ri->rrdset->rrdlabels;
  602. ri->flags &= ~RRD_FLAG_OWN_LABELS; // no need of atomics at the constructor
  603. }
  604. else {
  605. ri->rrdlabels = rrdlabels_create();
  606. ri->flags |= RRD_FLAG_OWN_LABELS; // no need of atomics at the constructor
  607. }
  608. if(ri->rrdset) {
  609. if(unlikely(rrdset_flag_check(ri->rrdset, RRDSET_FLAG_HIDDEN)))
  610. ri->flags |= RRD_FLAG_HIDDEN; // no need of atomics at the constructor
  611. else
  612. ri->flags &= ~RRD_FLAG_HIDDEN; // no need of atomics at the constructor
  613. }
  614. // we need this when loading from SQL
  615. if(unlikely(ri->id == ml_anomaly_rates_id))
  616. ri->flags |= RRD_FLAG_HIDDEN; // no need of atomics at the constructor
  617. rrdmetrics_create_in_rrdinstance(ri);
  618. // signal the react callback to do the job
  619. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_NEW_OBJECT);
  620. }
  621. static void rrdinstance_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *rrdcontext __maybe_unused) {
  622. RRDINSTANCE *ri = (RRDINSTANCE *)value;
  623. internal_error(ri->rrdset, "RRDINSTANCE: '%s' is freed but there is a RRDSET linked to it.", string2str(ri->id));
  624. rrdinstance_free(ri);
  625. }
  626. static bool rrdinstance_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *old_value, void *new_value, void *rrdcontext __maybe_unused) {
  627. RRDINSTANCE *ri = (RRDINSTANCE *)old_value;
  628. RRDINSTANCE *ri_new = (RRDINSTANCE *)new_value;
  629. internal_error(ri->id != ri_new->id,
  630. "RRDINSTANCE: '%s' cannot change id to '%s'",
  631. string2str(ri->id), string2str(ri_new->id));
  632. if(uuid_compare(ri->uuid, ri_new->uuid) != 0) {
  633. uuid_copy(ri->uuid, ri_new->uuid);
  634. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_UUID);
  635. }
  636. if(ri->rrdset && ri_new->rrdset && ri->rrdset != ri_new->rrdset) {
  637. ri->rrdset = ri_new->rrdset;
  638. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_LINKING);
  639. }
  640. if(ri->rrdset && uuid_compare(ri->uuid, ri->rrdset->chart_uuid) != 0) {
  641. char uuid1[UUID_STR_LEN], uuid2[UUID_STR_LEN];
  642. uuid_unparse(ri->uuid, uuid1);
  643. uuid_unparse(ri->rrdset->chart_uuid, uuid2);
  644. internal_error(true, "RRDINSTANCE: '%s' is linked to RRDSET '%s' but they have different UUIDs. RRDINSTANCE has '%s', RRDSET has '%s'", string2str(ri->id), rrdset_id(ri->rrdset), uuid1, uuid2);
  645. }
  646. if(ri->name != ri_new->name) {
  647. STRING *old = ri->name;
  648. ri->name = string_dup(ri_new->name);
  649. string_freez(old);
  650. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_NAME);
  651. }
  652. if(ri->title != ri_new->title) {
  653. STRING *old = ri->title;
  654. ri->title = string_dup(ri_new->title);
  655. string_freez(old);
  656. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_TITLE);
  657. }
  658. if(ri->units != ri_new->units) {
  659. STRING *old = ri->units;
  660. ri->units = string_dup(ri_new->units);
  661. string_freez(old);
  662. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_UNITS);
  663. }
  664. if(ri->family != ri_new->family) {
  665. STRING *old = ri->family;
  666. ri->family = string_dup(ri_new->family);
  667. string_freez(old);
  668. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_FAMILY);
  669. }
  670. if(ri->chart_type != ri_new->chart_type) {
  671. ri->chart_type = ri_new->chart_type;
  672. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_CHART_TYPE);
  673. }
  674. if(ri->priority != ri_new->priority) {
  675. ri->priority = ri_new->priority;
  676. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY);
  677. }
  678. if(ri->update_every != ri_new->update_every) {
  679. ri->update_every = ri_new->update_every;
  680. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_UPDATE_EVERY);
  681. }
  682. if(ri->rrdset != ri_new->rrdset) {
  683. ri->rrdset = ri_new->rrdset;
  684. if(ri->rrdset && rrd_flag_check(ri, RRD_FLAG_OWN_LABELS)) {
  685. DICTIONARY *old = ri->rrdlabels;
  686. ri->rrdlabels = ri->rrdset->rrdlabels;
  687. rrd_flag_clear(ri, RRD_FLAG_OWN_LABELS);
  688. rrdlabels_destroy(old);
  689. }
  690. else if(!ri->rrdset && !rrd_flag_check(ri, RRD_FLAG_OWN_LABELS)) {
  691. ri->rrdlabels = rrdlabels_create();
  692. rrd_flag_set(ri, RRD_FLAG_OWN_LABELS);
  693. }
  694. }
  695. if(ri->rrdset) {
  696. if(unlikely(rrdset_flag_check(ri->rrdset, RRDSET_FLAG_HIDDEN)))
  697. rrd_flag_set(ri, RRD_FLAG_HIDDEN);
  698. else
  699. rrd_flag_clear(ri, RRD_FLAG_HIDDEN);
  700. }
  701. rrd_flag_set(ri, ri_new->flags & RRD_FLAGS_ALLOWED_EXTERNALLY_ON_NEW_OBJECTS); // no need for atomics on ri_new
  702. if(rrd_flag_is_collected(ri) && rrd_flag_is_archived(ri))
  703. rrd_flag_set_collected(ri);
  704. if(rrd_flag_is_updated(ri))
  705. rrd_flag_set(ri, RRD_FLAG_UPDATE_REASON_UPDATED_OBJECT);
  706. // free the new one
  707. rrdinstance_free(ri_new);
  708. // the react callback will continue from here
  709. return rrd_flag_is_updated(ri);
  710. }
  711. static void rrdinstance_react_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *rrdcontext __maybe_unused) {
  712. RRDINSTANCE *ri = value;
  713. rrdinstance_trigger_updates(ri, __FUNCTION__ );
  714. }
  715. void rrdinstances_create_in_rrdcontext(RRDCONTEXT *rc) {
  716. if(unlikely(!rc || rc->rrdinstances)) return;
  717. rc->rrdinstances = dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE);
  718. dictionary_register_insert_callback(rc->rrdinstances, rrdinstance_insert_callback, rc);
  719. dictionary_register_delete_callback(rc->rrdinstances, rrdinstance_delete_callback, rc);
  720. dictionary_register_conflict_callback(rc->rrdinstances, rrdinstance_conflict_callback, rc);
  721. dictionary_register_react_callback(rc->rrdinstances, rrdinstance_react_callback, rc);
  722. }
  723. void rrdinstances_destroy_from_rrdcontext(RRDCONTEXT *rc) {
  724. if(unlikely(!rc || !rc->rrdinstances)) return;
  725. dictionary_destroy(rc->rrdinstances);
  726. rc->rrdinstances = NULL;
  727. }
  728. static void rrdinstance_trigger_updates(RRDINSTANCE *ri, const char *function) {
  729. RRDSET *st = ri->rrdset;
  730. if(likely(st)) {
  731. if(unlikely((unsigned int) st->priority != ri->priority)) {
  732. ri->priority = st->priority;
  733. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY);
  734. }
  735. if(unlikely(st->update_every != ri->update_every)) {
  736. ri->update_every = st->update_every;
  737. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_UPDATE_EVERY);
  738. }
  739. }
  740. else if(unlikely(rrd_flag_is_collected(ri))) {
  741. // there is no rrdset, but we have it as collected!
  742. rrd_flag_set_archived(ri);
  743. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_LINKING);
  744. }
  745. if(rrd_flag_is_updated(ri) || !rrd_flag_check(ri, RRD_FLAG_LIVE_RETENTION)) {
  746. rrd_flag_set_updated(ri->rc, RRD_FLAG_UPDATE_REASON_TRIGGERED);
  747. rrdcontext_queue_for_post_processing(ri->rc, function, ri->flags);
  748. }
  749. }
  750. // ----------------------------------------------------------------------------
  751. // RRDINSTANCE HOOKS ON RRDSET
  752. static inline void rrdinstance_from_rrdset(RRDSET *st) {
  753. RRDCONTEXT trc = {
  754. .id = string_dup(st->context),
  755. .title = string_dup(st->title),
  756. .units = string_dup(st->units),
  757. .family = string_dup(st->family),
  758. .priority = st->priority,
  759. .chart_type = st->chart_type,
  760. .flags = RRD_FLAG_NONE, // no need for atomics
  761. .rrdhost = st->rrdhost,
  762. };
  763. RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_set_and_acquire_item((DICTIONARY *)st->rrdhost->rrdctx, string2str(trc.id), &trc, sizeof(trc));
  764. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  765. RRDINSTANCE tri = {
  766. .id = string_dup(st->id),
  767. .name = string_dup(st->name),
  768. .units = string_dup(st->units),
  769. .family = string_dup(st->family),
  770. .title = string_dup(st->title),
  771. .chart_type = st->chart_type,
  772. .priority = st->priority,
  773. .update_every = st->update_every,
  774. .flags = RRD_FLAG_NONE, // no need for atomics
  775. .rrdset = st,
  776. };
  777. uuid_copy(tri.uuid, st->chart_uuid);
  778. RRDINSTANCE_ACQUIRED *ria = (RRDINSTANCE_ACQUIRED *)dictionary_set_and_acquire_item(rc->rrdinstances, string2str(tri.id), &tri, sizeof(tri));
  779. RRDCONTEXT_ACQUIRED *rca_old = st->rrdcontext;
  780. RRDINSTANCE_ACQUIRED *ria_old = st->rrdinstance;
  781. st->rrdcontext = rca;
  782. st->rrdinstance = ria;
  783. if(rca == rca_old) {
  784. rrdcontext_release(rca_old);
  785. rca_old = NULL;
  786. }
  787. if(ria == ria_old) {
  788. rrdinstance_release(ria_old);
  789. ria_old = NULL;
  790. }
  791. if(rca_old && ria_old) {
  792. // Oops! The chart changed context!
  793. // RRDCONTEXT *rc_old = rrdcontext_acquired_value(rca_old);
  794. RRDINSTANCE *ri_old = rrdinstance_acquired_value(ria_old);
  795. // migrate all dimensions to the new metrics
  796. RRDDIM *rd;
  797. rrddim_foreach_read(rd, st) {
  798. if (!rd->rrdmetric) continue;
  799. RRDMETRIC *rm_old = rrdmetric_acquired_value(rd->rrdmetric);
  800. rrd_flags_replace(rm_old, RRD_FLAG_DELETED|RRD_FLAG_UPDATED|RRD_FLAG_LIVE_RETENTION|RRD_FLAG_UPDATE_REASON_UNUSED|RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  801. rm_old->rrddim = NULL;
  802. rm_old->first_time_t = 0;
  803. rm_old->last_time_t = 0;
  804. rrdmetric_release(rd->rrdmetric);
  805. rd->rrdmetric = NULL;
  806. rrdmetric_from_rrddim(rd);
  807. }
  808. rrddim_foreach_done(rd);
  809. // mark the old instance, ready to be deleted
  810. if(!rrd_flag_check(ri_old, RRD_FLAG_OWN_LABELS))
  811. ri_old->rrdlabels = rrdlabels_create();
  812. rrd_flags_replace(ri_old, RRD_FLAG_OWN_LABELS|RRD_FLAG_DELETED|RRD_FLAG_UPDATED|RRD_FLAG_LIVE_RETENTION|RRD_FLAG_UPDATE_REASON_UNUSED|RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  813. ri_old->rrdset = NULL;
  814. ri_old->first_time_t = 0;
  815. ri_old->last_time_t = 0;
  816. rrdinstance_trigger_updates(ri_old, __FUNCTION__ );
  817. rrdinstance_release(ria_old);
  818. /*
  819. // trigger updates on the old context
  820. if(!dictionary_entries(rc_old->rrdinstances) && !dictionary_stats_referenced_items(rc_old->rrdinstances)) {
  821. rrdcontext_lock(rc_old);
  822. rc_old->flags = ((rc_old->flags & RRD_FLAG_QUEUED)?RRD_FLAG_QUEUED:RRD_FLAG_NONE)|RRD_FLAG_DELETED|RRD_FLAG_UPDATED|RRD_FLAG_LIVE_RETENTION|RRD_FLAG_UPDATE_REASON_UNUSED|RRD_FLAG_UPDATE_REASON_ZERO_RETENTION;
  823. rc_old->first_time_t = 0;
  824. rc_old->last_time_t = 0;
  825. rrdcontext_unlock(rc_old);
  826. rrdcontext_trigger_updates(rc_old, __FUNCTION__ );
  827. }
  828. else
  829. rrdcontext_trigger_updates(rc_old, __FUNCTION__ );
  830. */
  831. rrdcontext_release(rca_old);
  832. rca_old = NULL;
  833. ria_old = NULL;
  834. }
  835. if(rca_old || ria_old)
  836. fatal("RRDCONTEXT: cannot switch rrdcontext without switching rrdinstance too");
  837. }
  838. #define rrdset_get_rrdinstance(st) rrdset_get_rrdinstance_with_trace(st, __FUNCTION__);
  839. static inline RRDINSTANCE *rrdset_get_rrdinstance_with_trace(RRDSET *st, const char *function) {
  840. if(unlikely(!st->rrdinstance)) {
  841. error("RRDINSTANCE: RRDSET '%s' is not linked to an RRDINSTANCE at %s()", rrdset_id(st), function);
  842. return NULL;
  843. }
  844. RRDINSTANCE *ri = rrdinstance_acquired_value(st->rrdinstance);
  845. if(unlikely(!ri)) {
  846. error("RRDINSTANCE: RRDSET '%s' lost its link to an RRDINSTANCE at %s()", rrdset_id(st), function);
  847. return NULL;
  848. }
  849. if(unlikely(ri->rrdset != st))
  850. fatal("RRDINSTANCE: '%s' is not linked to RRDSET '%s' at %s()", string2str(ri->id), rrdset_id(st), function);
  851. return ri;
  852. }
  853. static inline void rrdinstance_rrdset_is_freed(RRDSET *st) {
  854. RRDINSTANCE *ri = rrdset_get_rrdinstance(st);
  855. if(unlikely(!ri)) return;
  856. rrd_flag_set_archived(ri);
  857. if(!rrd_flag_check(ri, RRD_FLAG_OWN_LABELS)) {
  858. ri->rrdlabels = rrdlabels_create();
  859. rrdlabels_copy(ri->rrdlabels, st->rrdlabels);
  860. rrd_flag_set(ri, RRD_FLAG_OWN_LABELS);
  861. }
  862. ri->rrdset = NULL;
  863. rrdinstance_trigger_updates(ri, __FUNCTION__ );
  864. rrdinstance_release(st->rrdinstance);
  865. st->rrdinstance = NULL;
  866. rrdcontext_release(st->rrdcontext);
  867. st->rrdcontext = NULL;
  868. }
  869. static inline void rrdinstance_rrdset_has_updated_retention(RRDSET *st) {
  870. RRDINSTANCE *ri = rrdset_get_rrdinstance(st);
  871. if(unlikely(!ri)) return;
  872. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_UPDATED_RETENTION);
  873. rrdinstance_trigger_updates(ri, __FUNCTION__ );
  874. }
  875. static inline void rrdinstance_updated_rrdset_name(RRDSET *st) {
  876. // the chart may not be initialized when this is called
  877. if(unlikely(!st->rrdinstance)) return;
  878. RRDINSTANCE *ri = rrdset_get_rrdinstance(st);
  879. if(unlikely(!ri)) return;
  880. if(st->name != ri->name) {
  881. STRING *old = ri->name;
  882. ri->name = string_dup(st->name);
  883. string_freez(old);
  884. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_NAME);
  885. rrdinstance_trigger_updates(ri, __FUNCTION__ );
  886. }
  887. }
  888. static inline void rrdinstance_updated_rrdset_flags_no_action(RRDINSTANCE *ri, RRDSET *st) {
  889. if(unlikely(ri->rrdset != st))
  890. fatal("RRDCONTEXT: instance '%s' is not linked to chart '%s' on host '%s'",
  891. string2str(ri->id), rrdset_id(st), rrdhost_hostname(st->rrdhost));
  892. bool st_is_hidden = rrdset_flag_check(st, RRDSET_FLAG_HIDDEN);
  893. bool ri_is_hidden = rrd_flag_check(ri, RRD_FLAG_HIDDEN);
  894. if(unlikely(st_is_hidden != ri_is_hidden)) {
  895. if (unlikely(st_is_hidden && !ri_is_hidden))
  896. rrd_flag_set_updated(ri, RRD_FLAG_HIDDEN | RRD_FLAG_UPDATE_REASON_CHANGED_FLAGS);
  897. else if (unlikely(!st_is_hidden && ri_is_hidden)) {
  898. rrd_flag_clear(ri, RRD_FLAG_HIDDEN);
  899. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_FLAGS);
  900. }
  901. }
  902. }
  903. static inline void rrdinstance_updated_rrdset_flags(RRDSET *st) {
  904. RRDINSTANCE *ri = rrdset_get_rrdinstance(st);
  905. if(unlikely(!ri)) return;
  906. if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_ARCHIVED|RRDSET_FLAG_OBSOLETE)))
  907. rrd_flag_set_archived(ri);
  908. rrdinstance_updated_rrdset_flags_no_action(ri, st);
  909. rrdinstance_trigger_updates(ri, __FUNCTION__ );
  910. }
  911. static inline void rrdinstance_collected_rrdset(RRDSET *st) {
  912. RRDINSTANCE *ri = rrdset_get_rrdinstance(st);
  913. if(unlikely(!ri)) return;
  914. rrdinstance_updated_rrdset_flags_no_action(ri, st);
  915. if(unlikely(ri->internal.collected_metrics_count && !rrd_flag_is_collected(ri)))
  916. rrd_flag_set_collected(ri);
  917. // we use this variable to detect BEGIN/END without SET
  918. ri->internal.collected_metrics_count = 0;
  919. rrdinstance_trigger_updates(ri, __FUNCTION__ );
  920. }
  921. // ----------------------------------------------------------------------------
  922. // RRDCONTEXT
  923. static void rrdcontext_freez(RRDCONTEXT *rc) {
  924. string_freez(rc->id);
  925. string_freez(rc->title);
  926. string_freez(rc->units);
  927. string_freez(rc->family);
  928. }
  929. static void rrdcontext_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *rrdhost) {
  930. RRDHOST *host = (RRDHOST *)rrdhost;
  931. RRDCONTEXT *rc = (RRDCONTEXT *)value;
  932. rc->rrdhost = host;
  933. rc->flags = rc->flags & RRD_FLAGS_ALLOWED_EXTERNALLY_ON_NEW_OBJECTS; // no need for atomics at constructor
  934. if(rc->hub.version) {
  935. // we are loading data from the SQL database
  936. if(rc->version)
  937. error("RRDCONTEXT: context '%s' is already initialized with version %"PRIu64", but it is loaded again from SQL with version %"PRIu64"", string2str(rc->id), rc->version, rc->hub.version);
  938. // IMPORTANT
  939. // replace all string pointers in rc->hub with our own versions
  940. // the originals are coming from a tmp allocation of sqlite
  941. string_freez(rc->id);
  942. rc->id = string_strdupz(rc->hub.id);
  943. rc->hub.id = string2str(rc->id);
  944. string_freez(rc->title);
  945. rc->title = string_strdupz(rc->hub.title);
  946. rc->hub.title = string2str(rc->title);
  947. string_freez(rc->units);
  948. rc->units = string_strdupz(rc->hub.units);
  949. rc->hub.units = string2str(rc->units);
  950. string_freez(rc->family);
  951. rc->family = string_strdupz(rc->hub.family);
  952. rc->hub.family = string2str(rc->family);
  953. rc->chart_type = rrdset_type_id(rc->hub.chart_type);
  954. rc->hub.chart_type = rrdset_type_name(rc->chart_type);
  955. rc->version = rc->hub.version;
  956. rc->priority = rc->hub.priority;
  957. rc->first_time_t = (time_t)rc->hub.first_time_t;
  958. rc->last_time_t = (time_t)rc->hub.last_time_t;
  959. if(rc->hub.deleted || !rc->hub.first_time_t)
  960. rrd_flag_set_deleted(rc, RRD_FLAG_NONE);
  961. else {
  962. if (rc->last_time_t == 0)
  963. rrd_flag_set_collected(rc);
  964. else
  965. rrd_flag_set_archived(rc);
  966. }
  967. rc->flags |= RRD_FLAG_UPDATE_REASON_LOAD_SQL; // no need for atomics at constructor
  968. }
  969. else {
  970. // we are adding this context now for the first time
  971. rc->version = now_realtime_sec();
  972. }
  973. rrdinstances_create_in_rrdcontext(rc);
  974. netdata_mutex_init(&rc->mutex);
  975. // signal the react callback to do the job
  976. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_NEW_OBJECT);
  977. }
  978. static void rrdcontext_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *rrdhost __maybe_unused) {
  979. RRDCONTEXT *rc = (RRDCONTEXT *)value;
  980. rrdinstances_destroy_from_rrdcontext(rc);
  981. netdata_mutex_destroy(&rc->mutex);
  982. rrdcontext_freez(rc);
  983. }
  984. static bool rrdcontext_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *old_value, void *new_value, void *rrdhost __maybe_unused) {
  985. RRDCONTEXT *rc = (RRDCONTEXT *)old_value;
  986. RRDCONTEXT *rc_new = (RRDCONTEXT *)new_value;
  987. //current rc is not archived, new_rc is archived, don't merge
  988. if (!rrd_flag_is_archived(rc) && rrd_flag_is_archived(rc_new)) {
  989. rrdcontext_freez(rc_new);
  990. return false;
  991. }
  992. rrdcontext_lock(rc);
  993. if(rc->title != rc_new->title) {
  994. STRING *old_title = rc->title;
  995. if (rrd_flag_is_archived(rc) && !rrd_flag_is_archived(rc_new))
  996. rc->title = string_dup(rc_new->title);
  997. else
  998. rc->title = string_2way_merge(rc->title, rc_new->title);
  999. string_freez(old_title);
  1000. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_TITLE);
  1001. }
  1002. if(rc->units != rc_new->units) {
  1003. STRING *old_units = rc->units;
  1004. rc->units = string_dup(rc_new->units);
  1005. string_freez(old_units);
  1006. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_UNITS);
  1007. }
  1008. if(rc->family != rc_new->family) {
  1009. STRING *old_family = rc->family;
  1010. if (rrd_flag_is_archived(rc) && !rrd_flag_is_archived(rc_new))
  1011. rc->family = string_dup(rc_new->family);
  1012. else
  1013. rc->family = string_2way_merge(rc->family, rc_new->family);
  1014. string_freez(old_family);
  1015. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_FAMILY);
  1016. }
  1017. if(rc->chart_type != rc_new->chart_type) {
  1018. rc->chart_type = rc_new->chart_type;
  1019. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_CHART_TYPE);
  1020. }
  1021. if(rc->priority != rc_new->priority) {
  1022. rc->priority = rc_new->priority;
  1023. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY);
  1024. }
  1025. rrd_flag_set(rc, rc_new->flags & RRD_FLAGS_ALLOWED_EXTERNALLY_ON_NEW_OBJECTS); // no need for atomics on rc_new
  1026. if(rrd_flag_is_collected(rc) && rrd_flag_is_archived(rc))
  1027. rrd_flag_set_collected(rc);
  1028. if(rrd_flag_is_updated(rc))
  1029. rrd_flag_set(rc, RRD_FLAG_UPDATE_REASON_UPDATED_OBJECT);
  1030. rrdcontext_unlock(rc);
  1031. // free the resources of the new one
  1032. rrdcontext_freez(rc_new);
  1033. // the react callback will continue from here
  1034. return rrd_flag_is_updated(rc);
  1035. }
  1036. static void rrdcontext_react_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *rrdhost __maybe_unused) {
  1037. RRDCONTEXT *rc = (RRDCONTEXT *)value;
  1038. rrdcontext_trigger_updates(rc, __FUNCTION__ );
  1039. }
  1040. static void rrdcontext_trigger_updates(RRDCONTEXT *rc, const char *function) {
  1041. if(rrd_flag_is_updated(rc) || !rrd_flag_check(rc, RRD_FLAG_LIVE_RETENTION))
  1042. rrdcontext_queue_for_post_processing(rc, function, rc->flags);
  1043. }
  1044. static void rrdcontext_hub_queue_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *nothing __maybe_unused) {
  1045. RRDCONTEXT *rc = context;
  1046. rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_HUB);
  1047. rc->queue.queued_ut = now_realtime_usec();
  1048. rc->queue.queued_flags = rrd_flags_get(rc);
  1049. }
  1050. static void rrdcontext_hub_queue_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *nothing __maybe_unused) {
  1051. RRDCONTEXT *rc = context;
  1052. rrd_flag_clear(rc, RRD_FLAG_QUEUED_FOR_HUB);
  1053. }
  1054. static bool rrdcontext_hub_queue_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *new_context __maybe_unused, void *nothing __maybe_unused) {
  1055. // context and new_context are the same
  1056. // we just need to update the timings
  1057. RRDCONTEXT *rc = context;
  1058. rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_HUB);
  1059. rc->queue.queued_ut = now_realtime_usec();
  1060. rc->queue.queued_flags |= rrd_flags_get(rc);
  1061. return true;
  1062. }
  1063. static void rrdcontext_post_processing_queue_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *nothing __maybe_unused) {
  1064. RRDCONTEXT *rc = context;
  1065. rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_POST_PROCESSING);
  1066. rc->pp.queued_flags = rc->flags;
  1067. rc->pp.queued_ut = now_realtime_usec();
  1068. }
  1069. static void rrdcontext_post_processing_queue_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *nothing __maybe_unused) {
  1070. RRDCONTEXT *rc = context;
  1071. rrd_flag_clear(rc, RRD_FLAG_QUEUED_FOR_POST_PROCESSING);
  1072. rc->pp.dequeued_ut = now_realtime_usec();
  1073. }
  1074. static bool rrdcontext_post_processing_queue_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *new_context __maybe_unused, void *nothing __maybe_unused) {
  1075. RRDCONTEXT *rc = context;
  1076. bool changed = false;
  1077. if(!(rc->flags & RRD_FLAG_QUEUED_FOR_POST_PROCESSING)) {
  1078. rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_POST_PROCESSING);
  1079. changed = true;
  1080. }
  1081. if(rc->pp.queued_flags != rc->flags) {
  1082. rc->pp.queued_flags |= rc->flags;
  1083. changed = true;
  1084. }
  1085. return changed;
  1086. }
  1087. void rrdhost_create_rrdcontexts(RRDHOST *host) {
  1088. if(unlikely(!host)) return;
  1089. if(likely(host->rrdctx)) return;
  1090. host->rrdctx = (RRDCONTEXTS *)dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE);
  1091. dictionary_register_insert_callback((DICTIONARY *)host->rrdctx, rrdcontext_insert_callback, host);
  1092. dictionary_register_delete_callback((DICTIONARY *)host->rrdctx, rrdcontext_delete_callback, host);
  1093. dictionary_register_conflict_callback((DICTIONARY *)host->rrdctx, rrdcontext_conflict_callback, host);
  1094. dictionary_register_react_callback((DICTIONARY *)host->rrdctx, rrdcontext_react_callback, host);
  1095. host->rrdctx_hub_queue = (RRDCONTEXTS *)dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_VALUE_LINK_DONT_CLONE);
  1096. dictionary_register_insert_callback((DICTIONARY *)host->rrdctx_hub_queue, rrdcontext_hub_queue_insert_callback, NULL);
  1097. dictionary_register_delete_callback((DICTIONARY *)host->rrdctx_hub_queue, rrdcontext_hub_queue_delete_callback, NULL);
  1098. dictionary_register_conflict_callback((DICTIONARY *)host->rrdctx_hub_queue, rrdcontext_hub_queue_conflict_callback, NULL);
  1099. host->rrdctx_post_processing_queue = (RRDCONTEXTS *)dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_VALUE_LINK_DONT_CLONE);
  1100. dictionary_register_insert_callback((DICTIONARY *)host->rrdctx_post_processing_queue, rrdcontext_post_processing_queue_insert_callback, NULL);
  1101. dictionary_register_delete_callback((DICTIONARY *)host->rrdctx_post_processing_queue, rrdcontext_post_processing_queue_delete_callback, NULL);
  1102. dictionary_register_conflict_callback((DICTIONARY *)host->rrdctx_post_processing_queue, rrdcontext_post_processing_queue_conflict_callback, NULL);
  1103. }
  1104. void rrdhost_destroy_rrdcontexts(RRDHOST *host) {
  1105. if(unlikely(!host)) return;
  1106. if(unlikely(!host->rrdctx)) return;
  1107. DICTIONARY *old;
  1108. if(host->rrdctx_hub_queue) {
  1109. old = (DICTIONARY *)host->rrdctx_hub_queue;
  1110. host->rrdctx_hub_queue = NULL;
  1111. RRDCONTEXT *rc;
  1112. dfe_start_write(old, rc) {
  1113. dictionary_del(old, string2str(rc->id));
  1114. }
  1115. dfe_done(rc);
  1116. dictionary_destroy(old);
  1117. }
  1118. if(host->rrdctx_post_processing_queue) {
  1119. old = (DICTIONARY *)host->rrdctx_post_processing_queue;
  1120. host->rrdctx_post_processing_queue = NULL;
  1121. RRDCONTEXT *rc;
  1122. dfe_start_write(old, rc) {
  1123. dictionary_del(old, string2str(rc->id));
  1124. }
  1125. dfe_done(rc);
  1126. dictionary_destroy(old);
  1127. }
  1128. old = (DICTIONARY *)host->rrdctx;
  1129. host->rrdctx = NULL;
  1130. dictionary_destroy(old);
  1131. }
  1132. // ----------------------------------------------------------------------------
  1133. // public API
  1134. void rrdcontext_updated_rrddim(RRDDIM *rd) {
  1135. rrdmetric_from_rrddim(rd);
  1136. }
  1137. void rrdcontext_removed_rrddim(RRDDIM *rd) {
  1138. rrdmetric_rrddim_is_freed(rd);
  1139. }
  1140. void rrdcontext_updated_rrddim_algorithm(RRDDIM *rd) {
  1141. rrdmetric_updated_rrddim_flags(rd);
  1142. }
  1143. void rrdcontext_updated_rrddim_multiplier(RRDDIM *rd) {
  1144. rrdmetric_updated_rrddim_flags(rd);
  1145. }
  1146. void rrdcontext_updated_rrddim_divisor(RRDDIM *rd) {
  1147. rrdmetric_updated_rrddim_flags(rd);
  1148. }
  1149. void rrdcontext_updated_rrddim_flags(RRDDIM *rd) {
  1150. rrdmetric_updated_rrddim_flags(rd);
  1151. }
  1152. void rrdcontext_collected_rrddim(RRDDIM *rd) {
  1153. rrdmetric_collected_rrddim(rd);
  1154. }
  1155. void rrdcontext_updated_rrdset(RRDSET *st) {
  1156. rrdinstance_from_rrdset(st);
  1157. }
  1158. void rrdcontext_removed_rrdset(RRDSET *st) {
  1159. rrdinstance_rrdset_is_freed(st);
  1160. }
  1161. void rrdcontext_updated_retention_rrdset(RRDSET *st) {
  1162. rrdinstance_rrdset_has_updated_retention(st);
  1163. }
  1164. void rrdcontext_updated_rrdset_name(RRDSET *st) {
  1165. rrdinstance_updated_rrdset_name(st);
  1166. }
  1167. void rrdcontext_updated_rrdset_flags(RRDSET *st) {
  1168. rrdinstance_updated_rrdset_flags(st);
  1169. }
  1170. void rrdcontext_collected_rrdset(RRDSET *st) {
  1171. rrdinstance_collected_rrdset(st);
  1172. }
  1173. void rrdcontext_host_child_connected(RRDHOST *host) {
  1174. (void)host;
  1175. // no need to do anything here
  1176. ;
  1177. }
  1178. int rrdcontext_find_dimension_uuid(RRDSET *st, const char *id, uuid_t *store_uuid) {
  1179. if(!st->rrdhost) return 1;
  1180. if(!st->context) return 2;
  1181. RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_get_and_acquire_item((DICTIONARY *)st->rrdhost->rrdctx, string2str(st->context));
  1182. if(!rca) return 3;
  1183. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  1184. RRDINSTANCE_ACQUIRED *ria = (RRDINSTANCE_ACQUIRED *)dictionary_get_and_acquire_item(rc->rrdinstances, string2str(st->id));
  1185. if(!ria) {
  1186. rrdcontext_release(rca);
  1187. return 4;
  1188. }
  1189. RRDINSTANCE *ri = rrdinstance_acquired_value(ria);
  1190. RRDMETRIC_ACQUIRED *rma = (RRDMETRIC_ACQUIRED *)dictionary_get_and_acquire_item(ri->rrdmetrics, id);
  1191. if(!rma) {
  1192. rrdinstance_release(ria);
  1193. rrdcontext_release(rca);
  1194. return 5;
  1195. }
  1196. RRDMETRIC *rm = rrdmetric_acquired_value(rma);
  1197. uuid_copy(*store_uuid, rm->uuid);
  1198. rrdmetric_release(rma);
  1199. rrdinstance_release(ria);
  1200. rrdcontext_release(rca);
  1201. return 0;
  1202. }
  1203. int rrdcontext_find_chart_uuid(RRDSET *st, uuid_t *store_uuid) {
  1204. if(!st->rrdhost) return 1;
  1205. if(!st->context) return 2;
  1206. RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_get_and_acquire_item((DICTIONARY *)st->rrdhost->rrdctx, string2str(st->context));
  1207. if(!rca) return 3;
  1208. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  1209. RRDINSTANCE_ACQUIRED *ria = (RRDINSTANCE_ACQUIRED *)dictionary_get_and_acquire_item(rc->rrdinstances, string2str(st->id));
  1210. if(!ria) {
  1211. rrdcontext_release(rca);
  1212. return 4;
  1213. }
  1214. RRDINSTANCE *ri = rrdinstance_acquired_value(ria);
  1215. uuid_copy(*store_uuid, ri->uuid);
  1216. rrdinstance_release(ria);
  1217. rrdcontext_release(rca);
  1218. return 0;
  1219. }
  1220. void rrdcontext_host_child_disconnected(RRDHOST *host) {
  1221. rrdcontext_recalculate_host_retention(host, RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD, false);
  1222. }
  1223. static usec_t rrdcontext_next_db_rotation_ut = 0;
  1224. void rrdcontext_db_rotation(void) {
  1225. // called when the db rotates its database
  1226. rrdcontext_next_db_rotation_ut = now_realtime_usec() + FULL_RETENTION_SCAN_DELAY_AFTER_DB_ROTATION_SECS * USEC_PER_SEC;
  1227. }
  1228. int rrdcontext_foreach_instance_with_rrdset_in_context(RRDHOST *host, const char *context, int (*callback)(RRDSET *st, void *data), void *data) {
  1229. if(unlikely(!host || !context || !*context || !callback))
  1230. return -1;
  1231. RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_get_and_acquire_item((DICTIONARY *)host->rrdctx, context);
  1232. if(unlikely(!rca)) return -1;
  1233. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  1234. if(unlikely(!rc)) return -1;
  1235. int ret = 0;
  1236. RRDINSTANCE *ri;
  1237. dfe_start_read(rc->rrdinstances, ri) {
  1238. if(ri->rrdset) {
  1239. int r = callback(ri->rrdset, data);
  1240. if(r >= 0) ret += r;
  1241. else {
  1242. ret = r;
  1243. break;
  1244. }
  1245. }
  1246. }
  1247. dfe_done(ri);
  1248. rrdcontext_release(rca);
  1249. return ret;
  1250. }
  1251. // ----------------------------------------------------------------------------
  1252. // ACLK interface
  1253. static bool rrdhost_check_our_claim_id(const char *claim_id) {
  1254. if(!localhost->aclk_state.claimed_id) return false;
  1255. return (strcasecmp(claim_id, localhost->aclk_state.claimed_id) == 0) ? true : false;
  1256. }
  1257. static RRDHOST *rrdhost_find_by_node_id(const char *node_id) {
  1258. uuid_t uuid;
  1259. if (uuid_parse(node_id, uuid))
  1260. return NULL;
  1261. RRDHOST *host = NULL;
  1262. rrd_rdlock();
  1263. rrdhost_foreach_read(host) {
  1264. if(!host->node_id) continue;
  1265. if(uuid_compare(uuid, *host->node_id) == 0)
  1266. break;
  1267. }
  1268. rrd_unlock();
  1269. return host;
  1270. }
  1271. void rrdcontext_hub_checkpoint_command(void *ptr) {
  1272. struct ctxs_checkpoint *cmd = ptr;
  1273. if(!rrdhost_check_our_claim_id(cmd->claim_id)) {
  1274. error("RRDCONTEXT: received checkpoint command for claim_id '%s', node id '%s', but this is not our claim id. Ours '%s', received '%s'. Ignoring command.",
  1275. cmd->claim_id, cmd->node_id,
  1276. localhost->aclk_state.claimed_id?localhost->aclk_state.claimed_id:"NOT SET",
  1277. cmd->claim_id);
  1278. return;
  1279. }
  1280. RRDHOST *host = rrdhost_find_by_node_id(cmd->node_id);
  1281. if(!host) {
  1282. error("RRDCONTEXT: received checkpoint command for claim id '%s', node id '%s', but there is no node with such node id here. Ignoring command.",
  1283. cmd->claim_id, cmd->node_id);
  1284. return;
  1285. }
  1286. if(rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS)) {
  1287. info("RRDCONTEXT: received checkpoint command for claim id '%s', node id '%s', while node '%s' has an active context streaming.",
  1288. cmd->claim_id, cmd->node_id, rrdhost_hostname(host));
  1289. // disable it temporarily, so that our worker will not attempt to send messages in parallel
  1290. rrdhost_flag_clear(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS);
  1291. }
  1292. uint64_t our_version_hash = rrdcontext_version_hash(host);
  1293. if(cmd->version_hash != our_version_hash) {
  1294. error("RRDCONTEXT: received version hash %"PRIu64" for host '%s', does not match our version hash %"PRIu64". Sending snapshot of all contexts.",
  1295. cmd->version_hash, rrdhost_hostname(host), our_version_hash);
  1296. #ifdef ENABLE_ACLK
  1297. // prepare the snapshot
  1298. char uuid[UUID_STR_LEN];
  1299. uuid_unparse_lower(*host->node_id, uuid);
  1300. contexts_snapshot_t bundle = contexts_snapshot_new(cmd->claim_id, uuid, our_version_hash);
  1301. // do a deep scan on every metric of the host to make sure all our data are updated
  1302. rrdcontext_recalculate_host_retention(host, RRD_FLAG_NONE, false);
  1303. // calculate version hash and pack all the messages together in one go
  1304. our_version_hash = rrdcontext_version_hash_with_callback(host, rrdcontext_message_send_unsafe, true, bundle);
  1305. // update the version
  1306. contexts_snapshot_set_version(bundle, our_version_hash);
  1307. // send it
  1308. aclk_send_contexts_snapshot(bundle);
  1309. #endif
  1310. }
  1311. internal_error(true, "RRDCONTEXT: host '%s' enabling streaming of contexts", rrdhost_hostname(host));
  1312. rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS);
  1313. char node_str[UUID_STR_LEN];
  1314. uuid_unparse_lower(*host->node_id, node_str);
  1315. log_access("ACLK REQ [%s (%s)]: STREAM CONTEXTS ENABLED", node_str, rrdhost_hostname(host));
  1316. }
  1317. void rrdcontext_hub_stop_streaming_command(void *ptr) {
  1318. struct stop_streaming_ctxs *cmd = ptr;
  1319. if(!rrdhost_check_our_claim_id(cmd->claim_id)) {
  1320. error("RRDCONTEXT: received stop streaming command for claim_id '%s', node id '%s', but this is not our claim id. Ours '%s', received '%s'. Ignoring command.",
  1321. cmd->claim_id, cmd->node_id,
  1322. localhost->aclk_state.claimed_id?localhost->aclk_state.claimed_id:"NOT SET",
  1323. cmd->claim_id);
  1324. return;
  1325. }
  1326. RRDHOST *host = rrdhost_find_by_node_id(cmd->node_id);
  1327. if(!host) {
  1328. error("RRDCONTEXT: received stop streaming command for claim id '%s', node id '%s', but there is no node with such node id here. Ignoring command.",
  1329. cmd->claim_id, cmd->node_id);
  1330. return;
  1331. }
  1332. if(!rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS)) {
  1333. error("RRDCONTEXT: received stop streaming command for claim id '%s', node id '%s', but node '%s' does not have active context streaming. Ignoring command.",
  1334. cmd->claim_id, cmd->node_id, rrdhost_hostname(host));
  1335. return;
  1336. }
  1337. internal_error(true, "RRDCONTEXT: host '%s' disabling streaming of contexts", rrdhost_hostname(host));
  1338. rrdhost_flag_clear(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS);
  1339. }
  1340. // ----------------------------------------------------------------------------
  1341. // web API
  1342. struct rrdcontext_to_json {
  1343. BUFFER *wb;
  1344. RRDCONTEXT_TO_JSON_OPTIONS options;
  1345. time_t after;
  1346. time_t before;
  1347. SIMPLE_PATTERN *chart_label_key;
  1348. SIMPLE_PATTERN *chart_labels_filter;
  1349. SIMPLE_PATTERN *chart_dimensions;
  1350. size_t written;
  1351. time_t now;
  1352. time_t combined_first_time_t;
  1353. time_t combined_last_time_t;
  1354. RRD_FLAGS combined_flags;
  1355. };
  1356. static inline int rrdmetric_to_json_callback(const DICTIONARY_ITEM *item, void *value, void *data) {
  1357. const char *id = dictionary_acquired_item_name(item);
  1358. struct rrdcontext_to_json * t = data;
  1359. RRDMETRIC *rm = value;
  1360. BUFFER *wb = t->wb;
  1361. RRDCONTEXT_TO_JSON_OPTIONS options = t->options;
  1362. time_t after = t->after;
  1363. time_t before = t->before;
  1364. if(unlikely(rrd_flag_is_deleted(rm) && !(options & RRDCONTEXT_OPTION_SHOW_DELETED)))
  1365. return 0;
  1366. if(after && (!rm->last_time_t || after > rm->last_time_t))
  1367. return 0;
  1368. if(before && (!rm->first_time_t || before < rm->first_time_t))
  1369. return 0;
  1370. if(t->chart_dimensions
  1371. && !simple_pattern_matches(t->chart_dimensions, string2str(rm->id))
  1372. && !simple_pattern_matches(t->chart_dimensions, string2str(rm->name)))
  1373. return 0;
  1374. if(t->written) {
  1375. buffer_strcat(wb, ",\n");
  1376. t->combined_first_time_t = MIN(t->combined_first_time_t, rm->first_time_t);
  1377. t->combined_last_time_t = MAX(t->combined_last_time_t, rm->last_time_t);
  1378. t->combined_flags |= rrd_flags_get(rm);
  1379. }
  1380. else {
  1381. buffer_strcat(wb, "\n");
  1382. t->combined_first_time_t = rm->first_time_t;
  1383. t->combined_last_time_t = rm->last_time_t;
  1384. t->combined_flags = rrd_flags_get(rm);
  1385. }
  1386. buffer_sprintf(wb, "\t\t\t\t\t\t\"%s\": {", id);
  1387. if(options & RRDCONTEXT_OPTION_SHOW_UUIDS) {
  1388. char uuid[UUID_STR_LEN];
  1389. uuid_unparse(rm->uuid, uuid);
  1390. buffer_sprintf(wb, "\n\t\t\t\t\t\t\t\"uuid\":\"%s\",", uuid);
  1391. }
  1392. buffer_sprintf(wb,
  1393. "\n\t\t\t\t\t\t\t\"name\":\"%s\""
  1394. ",\n\t\t\t\t\t\t\t\"first_time_t\":%lld"
  1395. ",\n\t\t\t\t\t\t\t\"last_time_t\":%lld"
  1396. ",\n\t\t\t\t\t\t\t\"collected\":%s"
  1397. , string2str(rm->name)
  1398. , (long long)rm->first_time_t
  1399. , rrd_flag_is_collected(rm) ? (long long)t->now : (long long)rm->last_time_t
  1400. , rrd_flag_is_collected(rm) ? "true" : "false"
  1401. );
  1402. if(options & RRDCONTEXT_OPTION_SHOW_DELETED) {
  1403. buffer_sprintf(wb,
  1404. ",\n\t\t\t\t\t\t\t\"deleted\":%s"
  1405. , rrd_flag_is_deleted(rm) ? "true" : "false"
  1406. );
  1407. }
  1408. if(options & RRDCONTEXT_OPTION_SHOW_FLAGS) {
  1409. buffer_strcat(wb, ",\n\t\t\t\t\t\t\t\"flags\":\"");
  1410. rrd_flags_to_buffer(rrd_flags_get(rm), wb);
  1411. buffer_strcat(wb, "\"");
  1412. }
  1413. buffer_strcat(wb, "\n\t\t\t\t\t\t}");
  1414. t->written++;
  1415. return 1;
  1416. }
  1417. static inline int rrdinstance_to_json_callback(const DICTIONARY_ITEM *item, void *value, void *data) {
  1418. const char *id = dictionary_acquired_item_name(item);
  1419. struct rrdcontext_to_json *t_parent = data;
  1420. RRDINSTANCE *ri = value;
  1421. BUFFER *wb = t_parent->wb;
  1422. RRDCONTEXT_TO_JSON_OPTIONS options = t_parent->options;
  1423. time_t after = t_parent->after;
  1424. time_t before = t_parent->before;
  1425. bool has_filter = t_parent->chart_label_key || t_parent->chart_labels_filter || t_parent->chart_dimensions;
  1426. if(unlikely(rrd_flag_is_deleted(ri) && !(options & RRDCONTEXT_OPTION_SHOW_DELETED)))
  1427. return 0;
  1428. if(after && (!ri->last_time_t || after > ri->last_time_t))
  1429. return 0;
  1430. if(before && (!ri->first_time_t || before < ri->first_time_t))
  1431. return 0;
  1432. if(t_parent->chart_label_key && !rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, t_parent->chart_label_key, '\0'))
  1433. return 0;
  1434. if(t_parent->chart_labels_filter && !rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, t_parent->chart_labels_filter, ':'))
  1435. return 0;
  1436. time_t first_time_t = ri->first_time_t;
  1437. time_t last_time_t = ri->last_time_t;
  1438. RRD_FLAGS flags = rrd_flags_get(ri);
  1439. BUFFER *wb_metrics = NULL;
  1440. if(options & RRDCONTEXT_OPTION_SHOW_METRICS || t_parent->chart_dimensions) {
  1441. wb_metrics = buffer_create(4096);
  1442. struct rrdcontext_to_json t_metrics = {
  1443. .wb = wb_metrics,
  1444. .options = options,
  1445. .chart_label_key = t_parent->chart_label_key,
  1446. .chart_labels_filter = t_parent->chart_labels_filter,
  1447. .chart_dimensions = t_parent->chart_dimensions,
  1448. .after = after,
  1449. .before = before,
  1450. .written = 0,
  1451. .now = t_parent->now,
  1452. };
  1453. dictionary_walkthrough_read(ri->rrdmetrics, rrdmetric_to_json_callback, &t_metrics);
  1454. if(has_filter && !t_metrics.written) {
  1455. buffer_free(wb_metrics);
  1456. return 0;
  1457. }
  1458. first_time_t = t_metrics.combined_first_time_t;
  1459. last_time_t = t_metrics.combined_last_time_t;
  1460. flags = t_metrics.combined_flags;
  1461. }
  1462. if(t_parent->written) {
  1463. buffer_strcat(wb, ",\n");
  1464. t_parent->combined_first_time_t = MIN(t_parent->combined_first_time_t, first_time_t);
  1465. t_parent->combined_last_time_t = MAX(t_parent->combined_last_time_t, last_time_t);
  1466. t_parent->combined_flags |= flags;
  1467. }
  1468. else {
  1469. buffer_strcat(wb, "\n");
  1470. t_parent->combined_first_time_t = first_time_t;
  1471. t_parent->combined_last_time_t = last_time_t;
  1472. t_parent->combined_flags = flags;
  1473. }
  1474. buffer_sprintf(wb, "\t\t\t\t\"%s\": {", id);
  1475. if(options & RRDCONTEXT_OPTION_SHOW_UUIDS) {
  1476. char uuid[UUID_STR_LEN];
  1477. uuid_unparse(ri->uuid, uuid);
  1478. buffer_sprintf(wb,"\n\t\t\t\t\t\"uuid\":\"%s\",", uuid);
  1479. }
  1480. buffer_sprintf(wb,
  1481. "\n\t\t\t\t\t\"name\":\"%s\""
  1482. ",\n\t\t\t\t\t\"context\":\"%s\""
  1483. ",\n\t\t\t\t\t\"title\":\"%s\""
  1484. ",\n\t\t\t\t\t\"units\":\"%s\""
  1485. ",\n\t\t\t\t\t\"family\":\"%s\""
  1486. ",\n\t\t\t\t\t\"chart_type\":\"%s\""
  1487. ",\n\t\t\t\t\t\"priority\":%u"
  1488. ",\n\t\t\t\t\t\"update_every\":%d"
  1489. ",\n\t\t\t\t\t\"first_time_t\":%lld"
  1490. ",\n\t\t\t\t\t\"last_time_t\":%lld"
  1491. ",\n\t\t\t\t\t\"collected\":%s"
  1492. , string2str(ri->name)
  1493. , string2str(ri->rc->id)
  1494. , string2str(ri->title)
  1495. , string2str(ri->units)
  1496. , string2str(ri->family)
  1497. , rrdset_type_name(ri->chart_type)
  1498. , ri->priority
  1499. , ri->update_every
  1500. , (long long)first_time_t
  1501. , (flags & RRD_FLAG_COLLECTED) ? (long long)t_parent->now : (long long)last_time_t
  1502. , (flags & RRD_FLAG_COLLECTED) ? "true" : "false"
  1503. );
  1504. if(options & RRDCONTEXT_OPTION_SHOW_DELETED) {
  1505. buffer_sprintf(wb,
  1506. ",\n\t\t\t\t\t\"deleted\":%s"
  1507. , rrd_flag_is_deleted(ri) ? "true" : "false"
  1508. );
  1509. }
  1510. if(options & RRDCONTEXT_OPTION_SHOW_FLAGS) {
  1511. buffer_strcat(wb, ",\n\t\t\t\t\t\"flags\":\"");
  1512. rrd_flags_to_buffer(rrd_flags_get(ri), wb);
  1513. buffer_strcat(wb, "\"");
  1514. }
  1515. if(options & RRDCONTEXT_OPTION_SHOW_LABELS && ri->rrdlabels && dictionary_entries(ri->rrdlabels)) {
  1516. buffer_sprintf(wb, ",\n\t\t\t\t\t\"labels\": {\n");
  1517. rrdlabels_to_buffer(ri->rrdlabels, wb, "\t\t\t\t\t\t", ":", "\"", ",\n", NULL, NULL, NULL, NULL);
  1518. buffer_strcat(wb, "\n\t\t\t\t\t}");
  1519. }
  1520. if(wb_metrics) {
  1521. buffer_sprintf(wb, ",\n\t\t\t\t\t\"dimensions\": {");
  1522. buffer_fast_strcat(wb, buffer_tostring(wb_metrics), buffer_strlen(wb_metrics));
  1523. buffer_strcat(wb, "\n\t\t\t\t\t}");
  1524. buffer_free(wb_metrics);
  1525. }
  1526. buffer_strcat(wb, "\n\t\t\t\t}");
  1527. t_parent->written++;
  1528. return 1;
  1529. }
  1530. static inline int rrdcontext_to_json_callback(const DICTIONARY_ITEM *item, void *value, void *data) {
  1531. const char *id = dictionary_acquired_item_name(item);
  1532. struct rrdcontext_to_json *t_parent = data;
  1533. RRDCONTEXT *rc = value;
  1534. BUFFER *wb = t_parent->wb;
  1535. RRDCONTEXT_TO_JSON_OPTIONS options = t_parent->options;
  1536. time_t after = t_parent->after;
  1537. time_t before = t_parent->before;
  1538. bool has_filter = t_parent->chart_label_key || t_parent->chart_labels_filter || t_parent->chart_dimensions;
  1539. if(unlikely(rrd_flag_check(rc, RRD_FLAG_HIDDEN) && !(options & RRDCONTEXT_OPTION_SHOW_HIDDEN)))
  1540. return 0;
  1541. if(unlikely(rrd_flag_is_deleted(rc) && !(options & RRDCONTEXT_OPTION_SHOW_DELETED)))
  1542. return 0;
  1543. if(options & RRDCONTEXT_OPTION_DEEPSCAN)
  1544. rrdcontext_recalculate_context_retention(rc, RRD_FLAG_NONE, false);
  1545. if(after && (!rc->last_time_t || after > rc->last_time_t))
  1546. return 0;
  1547. if(before && (!rc->first_time_t || before < rc->first_time_t))
  1548. return 0;
  1549. time_t first_time_t = rc->first_time_t;
  1550. time_t last_time_t = rc->last_time_t;
  1551. RRD_FLAGS flags = rrd_flags_get(rc);
  1552. BUFFER *wb_instances = NULL;
  1553. if((options & (RRDCONTEXT_OPTION_SHOW_LABELS|RRDCONTEXT_OPTION_SHOW_INSTANCES|RRDCONTEXT_OPTION_SHOW_METRICS))
  1554. || t_parent->chart_label_key
  1555. || t_parent->chart_labels_filter
  1556. || t_parent->chart_dimensions) {
  1557. wb_instances = buffer_create(4096);
  1558. struct rrdcontext_to_json t_instances = {
  1559. .wb = wb_instances,
  1560. .options = options,
  1561. .chart_label_key = t_parent->chart_label_key,
  1562. .chart_labels_filter = t_parent->chart_labels_filter,
  1563. .chart_dimensions = t_parent->chart_dimensions,
  1564. .after = after,
  1565. .before = before,
  1566. .written = 0,
  1567. .now = t_parent->now,
  1568. };
  1569. dictionary_walkthrough_read(rc->rrdinstances, rrdinstance_to_json_callback, &t_instances);
  1570. if(has_filter && !t_instances.written) {
  1571. buffer_free(wb_instances);
  1572. return 0;
  1573. }
  1574. first_time_t = t_instances.combined_first_time_t;
  1575. last_time_t = t_instances.combined_last_time_t;
  1576. flags = t_instances.combined_flags;
  1577. }
  1578. if(t_parent->written)
  1579. buffer_strcat(wb, ",\n");
  1580. else
  1581. buffer_strcat(wb, "\n");
  1582. if(options & RRDCONTEXT_OPTION_SKIP_ID)
  1583. buffer_sprintf(wb, "\t\t\{");
  1584. else
  1585. buffer_sprintf(wb, "\t\t\"%s\": {", id);
  1586. rrdcontext_lock(rc);
  1587. buffer_sprintf(wb,
  1588. "\n\t\t\t\"title\":\"%s\""
  1589. ",\n\t\t\t\"units\":\"%s\""
  1590. ",\n\t\t\t\"family\":\"%s\""
  1591. ",\n\t\t\t\"chart_type\":\"%s\""
  1592. ",\n\t\t\t\"priority\":%u"
  1593. ",\n\t\t\t\"first_time_t\":%lld"
  1594. ",\n\t\t\t\"last_time_t\":%lld"
  1595. ",\n\t\t\t\"collected\":%s"
  1596. , string2str(rc->title)
  1597. , string2str(rc->units)
  1598. , string2str(rc->family)
  1599. , rrdset_type_name(rc->chart_type)
  1600. , rc->priority
  1601. , (long long)first_time_t
  1602. , (flags & RRD_FLAG_COLLECTED) ? (long long)t_parent->now : (long long)last_time_t
  1603. , (flags & RRD_FLAG_COLLECTED) ? "true" : "false"
  1604. );
  1605. if(options & RRDCONTEXT_OPTION_SHOW_DELETED) {
  1606. buffer_sprintf(wb,
  1607. ",\n\t\t\t\"deleted\":%s"
  1608. , rrd_flag_is_deleted(rc) ? "true" : "false"
  1609. );
  1610. }
  1611. if(options & RRDCONTEXT_OPTION_SHOW_FLAGS) {
  1612. buffer_strcat(wb, ",\n\t\t\t\"flags\":\"");
  1613. rrd_flags_to_buffer(rrd_flags_get(rc), wb);
  1614. buffer_strcat(wb, "\"");
  1615. }
  1616. if(options & RRDCONTEXT_OPTION_SHOW_QUEUED) {
  1617. buffer_strcat(wb, ",\n\t\t\t\"queued_reasons\":\"");
  1618. rrd_reasons_to_buffer(rc->queue.queued_flags, wb);
  1619. buffer_strcat(wb, "\"");
  1620. buffer_sprintf(wb,
  1621. ",\n\t\t\t\"last_queued\":%llu"
  1622. ",\n\t\t\t\"scheduled_dispatch\":%llu"
  1623. ",\n\t\t\t\"last_dequeued\":%llu"
  1624. ",\n\t\t\t\"dispatches\":%zu"
  1625. ",\n\t\t\t\"hub_version\":%"PRIu64""
  1626. ",\n\t\t\t\"version\":%"PRIu64""
  1627. , rc->queue.queued_ut / USEC_PER_SEC
  1628. , rc->queue.scheduled_dispatch_ut / USEC_PER_SEC
  1629. , rc->queue.dequeued_ut / USEC_PER_SEC
  1630. , rc->queue.dispatches
  1631. , rc->hub.version
  1632. , rc->version
  1633. );
  1634. buffer_strcat(wb, ",\n\t\t\t\"pp_reasons\":\"");
  1635. rrd_reasons_to_buffer(rc->pp.queued_flags, wb);
  1636. buffer_strcat(wb, "\"");
  1637. buffer_sprintf(wb,
  1638. ",\n\t\t\t\"pp_last_queued\":%llu"
  1639. ",\n\t\t\t\"pp_last_dequeued\":%llu"
  1640. ",\n\t\t\t\"pp_executed\":%zu"
  1641. , rc->pp.queued_ut / USEC_PER_SEC
  1642. , rc->pp.dequeued_ut / USEC_PER_SEC
  1643. , rc->pp.executions
  1644. );
  1645. }
  1646. rrdcontext_unlock(rc);
  1647. if(wb_instances) {
  1648. buffer_sprintf(wb, ",\n\t\t\t\"charts\": {");
  1649. buffer_fast_strcat(wb, buffer_tostring(wb_instances), buffer_strlen(wb_instances));
  1650. buffer_strcat(wb, "\n\t\t\t}");
  1651. buffer_free(wb_instances);
  1652. }
  1653. buffer_strcat(wb, "\n\t\t}");
  1654. t_parent->written++;
  1655. return 1;
  1656. }
  1657. int rrdcontext_to_json(RRDHOST *host, BUFFER *wb, time_t after, time_t before, RRDCONTEXT_TO_JSON_OPTIONS options, const char *context, SIMPLE_PATTERN *chart_label_key, SIMPLE_PATTERN *chart_labels_filter, SIMPLE_PATTERN *chart_dimensions) {
  1658. if(!host->rrdctx) {
  1659. error("%s(): request for host '%s' that does not have rrdcontexts initialized.", __FUNCTION__, rrdhost_hostname(host));
  1660. return HTTP_RESP_NOT_FOUND;
  1661. }
  1662. RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_get_and_acquire_item((DICTIONARY *)host->rrdctx, context);
  1663. if(!rca) return HTTP_RESP_NOT_FOUND;
  1664. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  1665. if(after != 0 && before != 0)
  1666. rrdr_relative_window_to_absolute(&after, &before);
  1667. struct rrdcontext_to_json t_contexts = {
  1668. .wb = wb,
  1669. .options = options|RRDCONTEXT_OPTION_SKIP_ID,
  1670. .chart_label_key = chart_label_key,
  1671. .chart_labels_filter = chart_labels_filter,
  1672. .chart_dimensions = chart_dimensions,
  1673. .after = after,
  1674. .before = before,
  1675. .written = 0,
  1676. .now = now_realtime_sec(),
  1677. };
  1678. rrdcontext_to_json_callback((DICTIONARY_ITEM *)rca, rc, &t_contexts);
  1679. rrdcontext_release(rca);
  1680. if(!t_contexts.written)
  1681. return HTTP_RESP_NOT_FOUND;
  1682. return HTTP_RESP_OK;
  1683. }
  1684. int rrdcontexts_to_json(RRDHOST *host, BUFFER *wb, time_t after, time_t before, RRDCONTEXT_TO_JSON_OPTIONS options, SIMPLE_PATTERN *chart_label_key, SIMPLE_PATTERN *chart_labels_filter, SIMPLE_PATTERN *chart_dimensions) {
  1685. if(!host->rrdctx) {
  1686. error("%s(): request for host '%s' that does not have rrdcontexts initialized.", __FUNCTION__, rrdhost_hostname(host));
  1687. return HTTP_RESP_NOT_FOUND;
  1688. }
  1689. char node_uuid[UUID_STR_LEN] = "";
  1690. if(host->node_id)
  1691. uuid_unparse(*host->node_id, node_uuid);
  1692. if(after != 0 && before != 0)
  1693. rrdr_relative_window_to_absolute(&after, &before);
  1694. buffer_sprintf(wb, "{\n"
  1695. "\t\"hostname\": \"%s\""
  1696. ",\n\t\"machine_guid\": \"%s\""
  1697. ",\n\t\"node_id\": \"%s\""
  1698. ",\n\t\"claim_id\": \"%s\""
  1699. , rrdhost_hostname(host)
  1700. , host->machine_guid
  1701. , node_uuid
  1702. , host->aclk_state.claimed_id ? host->aclk_state.claimed_id : ""
  1703. );
  1704. if(options & RRDCONTEXT_OPTION_SHOW_LABELS) {
  1705. buffer_sprintf(wb, ",\n\t\"host_labels\": {\n");
  1706. rrdlabels_to_buffer(host->rrdlabels, wb, "\t\t", ":", "\"", ",\n", NULL, NULL, NULL, NULL);
  1707. buffer_strcat(wb, "\n\t}");
  1708. }
  1709. buffer_sprintf(wb, ",\n\t\"contexts\": {");
  1710. struct rrdcontext_to_json t_contexts = {
  1711. .wb = wb,
  1712. .options = options,
  1713. .chart_label_key = chart_label_key,
  1714. .chart_labels_filter = chart_labels_filter,
  1715. .chart_dimensions = chart_dimensions,
  1716. .after = after,
  1717. .before = before,
  1718. .written = 0,
  1719. .now = now_realtime_sec(),
  1720. };
  1721. dictionary_walkthrough_read((DICTIONARY *)host->rrdctx, rrdcontext_to_json_callback, &t_contexts);
  1722. // close contexts, close main
  1723. buffer_strcat(wb, "\n\t}\n}");
  1724. return HTTP_RESP_OK;
  1725. }
  1726. // ----------------------------------------------------------------------------
  1727. // weights API
  1728. static void metric_entry_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *data __maybe_unused) {
  1729. struct metric_entry *t = value;
  1730. t->rca = rrdcontext_acquired_dup(t->rca);
  1731. t->ria = rrdinstance_acquired_dup(t->ria);
  1732. t->rma = rrdmetric_acquired_dup(t->rma);
  1733. }
  1734. static void metric_entry_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *data __maybe_unused) {
  1735. struct metric_entry *t = value;
  1736. rrdcontext_release(t->rca);
  1737. rrdinstance_release(t->ria);
  1738. rrdmetric_release(t->rma);
  1739. }
  1740. static bool metric_entry_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *old_value __maybe_unused, void *new_value __maybe_unused, void *data __maybe_unused) {
  1741. fatal("RRDCONTEXT: %s() detected a conflict on a metric pointer!", __FUNCTION__);
  1742. return false;
  1743. }
  1744. DICTIONARY *rrdcontext_all_metrics_to_dict(RRDHOST *host, SIMPLE_PATTERN *contexts) {
  1745. if(!host || !host->rrdctx)
  1746. return NULL;
  1747. DICTIONARY *dict = dictionary_create(DICT_OPTION_SINGLE_THREADED|DICT_OPTION_DONT_OVERWRITE_VALUE);
  1748. dictionary_register_insert_callback(dict, metric_entry_insert_callback, NULL);
  1749. dictionary_register_delete_callback(dict, metric_entry_delete_callback, NULL);
  1750. dictionary_register_conflict_callback(dict, metric_entry_conflict_callback, NULL);
  1751. RRDCONTEXT *rc;
  1752. dfe_start_reentrant((DICTIONARY *)host->rrdctx, rc) {
  1753. if(rrd_flag_is_deleted(rc))
  1754. continue;
  1755. if(contexts && !simple_pattern_matches(contexts, string2str(rc->id)))
  1756. continue;
  1757. RRDINSTANCE *ri;
  1758. dfe_start_read(rc->rrdinstances, ri) {
  1759. if(rrd_flag_is_deleted(ri))
  1760. continue;
  1761. RRDMETRIC *rm;
  1762. dfe_start_read(ri->rrdmetrics, rm) {
  1763. if(rrd_flag_is_deleted(rm))
  1764. continue;
  1765. struct metric_entry tmp = {
  1766. .rca = (RRDCONTEXT_ACQUIRED *)rc_dfe.item,
  1767. .ria = (RRDINSTANCE_ACQUIRED *)ri_dfe.item,
  1768. .rma = (RRDMETRIC_ACQUIRED *)rm_dfe.item,
  1769. };
  1770. char buffer[20 + 1];
  1771. ssize_t len = snprintfz(buffer, 20, "%p", rm);
  1772. dictionary_set_advanced(dict, buffer, len + 1, &tmp, sizeof(struct metric_entry), NULL);
  1773. }
  1774. dfe_done(rm);
  1775. }
  1776. dfe_done(ri);
  1777. }
  1778. dfe_done(rc);
  1779. return dict;
  1780. }
  1781. // ----------------------------------------------------------------------------
  1782. // query API
  1783. typedef struct query_target_locals {
  1784. time_t start_s;
  1785. QUERY_TARGET *qt;
  1786. RRDSET *st;
  1787. const char *hosts;
  1788. const char *contexts;
  1789. const char *charts;
  1790. const char *dimensions;
  1791. const char *chart_label_key;
  1792. const char *charts_labels_filter;
  1793. long long after;
  1794. long long before;
  1795. bool match_ids;
  1796. bool match_names;
  1797. RRDHOST *host;
  1798. RRDCONTEXT_ACQUIRED *rca;
  1799. RRDINSTANCE_ACQUIRED *ria;
  1800. size_t metrics_skipped_due_to_not_matching_timeframe;
  1801. } QUERY_TARGET_LOCALS;
  1802. static __thread QUERY_TARGET thread_query_target = {};
  1803. void query_target_release(QUERY_TARGET *qt) {
  1804. if(unlikely(!qt)) return;
  1805. simple_pattern_free(qt->hosts.pattern);
  1806. qt->hosts.pattern = NULL;
  1807. simple_pattern_free(qt->contexts.pattern);
  1808. qt->contexts.pattern = NULL;
  1809. simple_pattern_free(qt->instances.pattern);
  1810. qt->instances.pattern = NULL;
  1811. simple_pattern_free(qt->instances.chart_label_key_pattern);
  1812. qt->instances.chart_label_key_pattern = NULL;
  1813. simple_pattern_free(qt->instances.charts_labels_filter_pattern);
  1814. qt->instances.charts_labels_filter_pattern = NULL;
  1815. simple_pattern_free(qt->query.pattern);
  1816. qt->query.pattern = NULL;
  1817. // release the query
  1818. for(size_t i = 0, used = qt->query.used; i < used ;i++) {
  1819. string_freez(qt->query.array[i].dimension.id);
  1820. qt->query.array[i].dimension.id = NULL;
  1821. string_freez(qt->query.array[i].dimension.name);
  1822. qt->query.array[i].dimension.name = NULL;
  1823. string_freez(qt->query.array[i].chart.id);
  1824. qt->query.array[i].chart.id = NULL;
  1825. string_freez(qt->query.array[i].chart.name);
  1826. qt->query.array[i].chart.name = NULL;
  1827. for(size_t tier = 0; tier < storage_tiers ;tier++) {
  1828. if(qt->query.array[i].tiers[tier].db_metric_handle) {
  1829. STORAGE_ENGINE *eng = qt->query.array[i].tiers[tier].eng;
  1830. eng->api.metric_release(qt->query.array[i].tiers[tier].db_metric_handle);
  1831. qt->query.array[i].tiers[tier].db_metric_handle = NULL;
  1832. }
  1833. }
  1834. }
  1835. // release the metrics
  1836. for(size_t i = 0, used = qt->metrics.used; i < used ;i++) {
  1837. rrdmetric_release(qt->metrics.array[i]);
  1838. qt->metrics.array[i] = NULL;
  1839. }
  1840. // release the instances
  1841. for(size_t i = 0, used = qt->instances.used; i < used ;i++) {
  1842. rrdinstance_release(qt->instances.array[i]);
  1843. qt->instances.array[i] = NULL;
  1844. }
  1845. // release the contexts
  1846. for(size_t i = 0, used = qt->contexts.used; i < used ;i++) {
  1847. rrdcontext_release(qt->contexts.array[i]);
  1848. qt->contexts.array[i] = NULL;
  1849. }
  1850. // release the hosts
  1851. for(size_t i = 0, used = qt->hosts.used; i < used ;i++) {
  1852. qt->hosts.array[i] = NULL;
  1853. }
  1854. qt->query.used = 0;
  1855. qt->metrics.used = 0;
  1856. qt->instances.used = 0;
  1857. qt->contexts.used = 0;
  1858. qt->hosts.used = 0;
  1859. qt->db.minimum_latest_update_every = 0;
  1860. qt->db.first_time_t = 0;
  1861. qt->db.last_time_t = 0;
  1862. qt->id[0] = '\0';
  1863. qt->used = false;
  1864. }
  1865. void query_target_free(void) {
  1866. if(thread_query_target.used)
  1867. query_target_release(&thread_query_target);
  1868. freez(thread_query_target.query.array);
  1869. thread_query_target.query.array = NULL;
  1870. thread_query_target.query.size = 0;
  1871. freez(thread_query_target.metrics.array);
  1872. thread_query_target.metrics.array = NULL;
  1873. thread_query_target.metrics.size = 0;
  1874. freez(thread_query_target.instances.array);
  1875. thread_query_target.instances.array = NULL;
  1876. thread_query_target.instances.size = 0;
  1877. freez(thread_query_target.contexts.array);
  1878. thread_query_target.contexts.array = NULL;
  1879. thread_query_target.contexts.size = 0;
  1880. freez(thread_query_target.hosts.array);
  1881. thread_query_target.hosts.array = NULL;
  1882. thread_query_target.hosts.size = 0;
  1883. }
  1884. static void query_target_add_metric(QUERY_TARGET_LOCALS *qtl, RRDMETRIC_ACQUIRED *rma, RRDINSTANCE *ri,
  1885. bool queryable_instance) {
  1886. QUERY_TARGET *qt = qtl->qt;
  1887. RRDMETRIC *rm = rrdmetric_acquired_value(rma);
  1888. if(rrd_flag_is_deleted(rm))
  1889. return;
  1890. if(qt->metrics.used == qt->metrics.size) {
  1891. qt->metrics.size = (qt->metrics.size) ? qt->metrics.size * 2 : 1;
  1892. qt->metrics.array = reallocz(qt->metrics.array, qt->metrics.size * sizeof(RRDMETRIC_ACQUIRED *));
  1893. }
  1894. qt->metrics.array[qt->metrics.used++] = rrdmetric_acquired_dup(rma);
  1895. if(!queryable_instance)
  1896. return;
  1897. time_t common_first_time_t = 0;
  1898. time_t common_last_time_t = 0;
  1899. time_t common_update_every = 0;
  1900. size_t tiers_added = 0;
  1901. struct {
  1902. STORAGE_ENGINE *eng;
  1903. STORAGE_METRIC_HANDLE *db_metric_handle;
  1904. time_t db_first_time_t;
  1905. time_t db_last_time_t;
  1906. time_t db_update_every;
  1907. } tier_retention[storage_tiers];
  1908. for (size_t tier = 0; tier < storage_tiers; tier++) {
  1909. STORAGE_ENGINE *eng = qtl->host->db[tier].eng;
  1910. tier_retention[tier].eng = eng;
  1911. tier_retention[tier].db_update_every = (time_t) (qtl->host->db[tier].tier_grouping * ri->update_every);
  1912. if(rm->rrddim && rm->rrddim->tiers[tier] && rm->rrddim->tiers[tier]->db_metric_handle)
  1913. tier_retention[tier].db_metric_handle = eng->api.metric_dup(rm->rrddim->tiers[tier]->db_metric_handle);
  1914. else
  1915. tier_retention[tier].db_metric_handle = eng->api.metric_get(qtl->host->db[tier].instance, &rm->uuid, NULL);
  1916. if(tier_retention[tier].db_metric_handle) {
  1917. tier_retention[tier].db_first_time_t = tier_retention[tier].eng->api.query_ops.oldest_time(tier_retention[tier].db_metric_handle);
  1918. tier_retention[tier].db_last_time_t = tier_retention[tier].eng->api.query_ops.latest_time(tier_retention[tier].db_metric_handle);
  1919. if(!common_first_time_t)
  1920. common_first_time_t = tier_retention[tier].db_first_time_t;
  1921. else if(tier_retention[tier].db_first_time_t)
  1922. common_first_time_t = MIN(common_first_time_t, tier_retention[tier].db_first_time_t);
  1923. if(!common_last_time_t)
  1924. common_last_time_t = tier_retention[tier].db_last_time_t;
  1925. else
  1926. common_last_time_t = MAX(common_last_time_t, tier_retention[tier].db_last_time_t);
  1927. if(!common_update_every)
  1928. common_update_every = tier_retention[tier].db_update_every;
  1929. else if(tier_retention[tier].db_update_every)
  1930. common_update_every = MIN(common_update_every, tier_retention[tier].db_update_every);
  1931. tiers_added++;
  1932. }
  1933. else {
  1934. tier_retention[tier].db_first_time_t = 0;
  1935. tier_retention[tier].db_last_time_t = 0;
  1936. tier_retention[tier].db_update_every = 0;
  1937. }
  1938. }
  1939. bool release_retention = true;
  1940. bool timeframe_matches =
  1941. (tiers_added
  1942. && (common_first_time_t - common_update_every * 2) <= qt->window.before
  1943. && (common_last_time_t + common_update_every * 2) >= qt->window.after
  1944. ) ? true : false;
  1945. if(timeframe_matches) {
  1946. RRDR_DIMENSION_FLAGS options = RRDR_DIMENSION_DEFAULT;
  1947. if (rrd_flag_check(rm, RRD_FLAG_HIDDEN)
  1948. || (rm->rrddim && rrddim_option_check(rm->rrddim, RRDDIM_OPTION_HIDDEN))) {
  1949. options |= RRDR_DIMENSION_HIDDEN;
  1950. options &= ~RRDR_DIMENSION_SELECTED;
  1951. }
  1952. if (qt->query.pattern) {
  1953. // we have a dimensions pattern
  1954. // lets see if this dimension is selected
  1955. if ((qtl->match_ids && simple_pattern_matches(qt->query.pattern, string2str(rm->id)))
  1956. || (qtl->match_names && simple_pattern_matches(qt->query.pattern, string2str(rm->name)))
  1957. ) {
  1958. // it matches the pattern
  1959. options |= (RRDR_DIMENSION_SELECTED | RRDR_DIMENSION_NONZERO);
  1960. options &= ~RRDR_DIMENSION_HIDDEN;
  1961. }
  1962. else {
  1963. // it does not match the pattern
  1964. options |= RRDR_DIMENSION_HIDDEN;
  1965. options &= ~RRDR_DIMENSION_SELECTED;
  1966. }
  1967. }
  1968. else {
  1969. // we don't have a dimensions pattern
  1970. // so this is a selected dimension
  1971. // if it is not hidden
  1972. if(!(options & RRDR_DIMENSION_HIDDEN))
  1973. options |= RRDR_DIMENSION_SELECTED;
  1974. }
  1975. if((options & RRDR_DIMENSION_HIDDEN) && (options & RRDR_DIMENSION_SELECTED))
  1976. options &= ~RRDR_DIMENSION_HIDDEN;
  1977. if(!(options & RRDR_DIMENSION_HIDDEN) || (qt->request.options & RRDR_OPTION_PERCENTAGE)) {
  1978. // we have a non-hidden dimension
  1979. // let's add it to the query metrics
  1980. if(ri->rrdset)
  1981. ri->rrdset->last_accessed_time = qtl->start_s;
  1982. if (qt->query.used == qt->query.size) {
  1983. qt->query.size = (qt->query.size) ? qt->query.size * 2 : 1;
  1984. qt->query.array = reallocz(qt->query.array, qt->query.size * sizeof(QUERY_METRIC));
  1985. }
  1986. QUERY_METRIC *qm = &qt->query.array[qt->query.used++];
  1987. qm->dimension.options = options;
  1988. qm->link.host = qtl->host;
  1989. qm->link.rca = qtl->rca;
  1990. qm->link.ria = qtl->ria;
  1991. qm->link.rma = rma;
  1992. qm->chart.id = string_dup(ri->id);
  1993. qm->chart.name = string_dup(ri->name);
  1994. qm->dimension.id = string_dup(rm->id);
  1995. qm->dimension.name = string_dup(rm->name);
  1996. if (!qt->db.first_time_t || common_first_time_t < qt->db.first_time_t)
  1997. qt->db.first_time_t = common_first_time_t;
  1998. if (!qt->db.last_time_t || common_last_time_t > qt->db.last_time_t)
  1999. qt->db.last_time_t = common_last_time_t;
  2000. for (size_t tier = 0; tier < storage_tiers; tier++) {
  2001. qm->tiers[tier].eng = tier_retention[tier].eng;
  2002. qm->tiers[tier].db_metric_handle = tier_retention[tier].db_metric_handle;
  2003. qm->tiers[tier].db_first_time_t = tier_retention[tier].db_first_time_t;
  2004. qm->tiers[tier].db_last_time_t = tier_retention[tier].db_last_time_t;
  2005. qm->tiers[tier].db_update_every = tier_retention[tier].db_update_every;
  2006. }
  2007. release_retention = false;
  2008. }
  2009. }
  2010. else
  2011. qtl->metrics_skipped_due_to_not_matching_timeframe++;
  2012. if(release_retention) {
  2013. // cleanup anything we allocated to the retention we will not use
  2014. for(size_t tier = 0; tier < storage_tiers ;tier++) {
  2015. if (tier_retention[tier].db_metric_handle)
  2016. tier_retention[tier].eng->api.metric_release(tier_retention[tier].db_metric_handle);
  2017. }
  2018. }
  2019. }
  2020. static void query_target_add_instance(QUERY_TARGET_LOCALS *qtl, RRDINSTANCE_ACQUIRED *ria, bool queryable_instance) {
  2021. QUERY_TARGET *qt = qtl->qt;
  2022. RRDINSTANCE *ri = rrdinstance_acquired_value(ria);
  2023. if(rrd_flag_is_deleted(ri))
  2024. return;
  2025. if(qt->instances.used == qt->instances.size) {
  2026. qt->instances.size = (qt->instances.size) ? qt->instances.size * 2 : 1;
  2027. qt->instances.array = reallocz(qt->instances.array, qt->instances.size * sizeof(RRDINSTANCE_ACQUIRED *));
  2028. }
  2029. qtl->ria = qt->instances.array[qt->instances.used++] = rrdinstance_acquired_dup(ria);
  2030. if(qt->db.minimum_latest_update_every == 0 || ri->update_every < qt->db.minimum_latest_update_every)
  2031. qt->db.minimum_latest_update_every = ri->update_every;
  2032. if(queryable_instance) {
  2033. if ((qt->instances.chart_label_key_pattern && !rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, qt->instances.chart_label_key_pattern, ':')) ||
  2034. (qt->instances.charts_labels_filter_pattern && !rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, qt->instances.charts_labels_filter_pattern, ':')))
  2035. queryable_instance = false;
  2036. }
  2037. size_t added = 0;
  2038. if(unlikely(qt->request.rma)) {
  2039. query_target_add_metric(qtl, qt->request.rma, ri, queryable_instance);
  2040. added++;
  2041. }
  2042. else {
  2043. RRDMETRIC *rm;
  2044. dfe_start_read(ri->rrdmetrics, rm) {
  2045. query_target_add_metric(qtl, (RRDMETRIC_ACQUIRED *) rm_dfe.item, ri, queryable_instance);
  2046. added++;
  2047. }
  2048. dfe_done(rm);
  2049. }
  2050. if(!added) {
  2051. qt->instances.used--;
  2052. rrdinstance_release(ria);
  2053. }
  2054. }
  2055. static void query_target_add_context(QUERY_TARGET_LOCALS *qtl, RRDCONTEXT_ACQUIRED *rca) {
  2056. QUERY_TARGET *qt = qtl->qt;
  2057. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  2058. if(rrd_flag_is_deleted(rc))
  2059. return;
  2060. if(qt->contexts.used == qt->contexts.size) {
  2061. qt->contexts.size = (qt->contexts.size) ? qt->contexts.size * 2 : 1;
  2062. qt->contexts.array = reallocz(qt->contexts.array, qt->contexts.size * sizeof(RRDCONTEXT_ACQUIRED *));
  2063. }
  2064. qtl->rca = qt->contexts.array[qt->contexts.used++] = rrdcontext_acquired_dup(rca);
  2065. size_t added = 0;
  2066. if(unlikely(qt->request.ria)) {
  2067. query_target_add_instance(qtl, qt->request.ria, true);
  2068. added++;
  2069. }
  2070. else if(unlikely(qtl->st && qtl->st->rrdcontext == rca && qtl->st->rrdinstance)) {
  2071. query_target_add_instance(qtl, qtl->st->rrdinstance, true);
  2072. added++;
  2073. }
  2074. else {
  2075. RRDINSTANCE *ri;
  2076. dfe_start_read(rc->rrdinstances, ri) {
  2077. bool queryable_instance = false;
  2078. if(!qt->instances.pattern
  2079. || (qtl->match_ids && simple_pattern_matches(qt->instances.pattern, string2str(ri->id)))
  2080. || (qtl->match_names && simple_pattern_matches(qt->instances.pattern, string2str(ri->name)))
  2081. )
  2082. queryable_instance = true;
  2083. query_target_add_instance(qtl, (RRDINSTANCE_ACQUIRED *)ri_dfe.item, queryable_instance);
  2084. added++;
  2085. }
  2086. dfe_done(ri);
  2087. }
  2088. if(!added) {
  2089. qt->contexts.used--;
  2090. rrdcontext_release(rca);
  2091. }
  2092. }
  2093. static void query_target_add_host(QUERY_TARGET_LOCALS *qtl, RRDHOST *host) {
  2094. QUERY_TARGET *qt = qtl->qt;
  2095. if(qt->hosts.used == qt->hosts.size) {
  2096. qt->hosts.size = (qt->hosts.size) ? qt->hosts.size * 2 : 1;
  2097. qt->hosts.array = reallocz(qt->hosts.array, qt->hosts.size * sizeof(RRDHOST *));
  2098. }
  2099. qtl->host = qt->hosts.array[qt->hosts.used++] = host;
  2100. // is the chart given valid?
  2101. if(unlikely(qtl->st && (!qtl->st->rrdinstance || !qtl->st->rrdcontext))) {
  2102. error("QUERY TARGET: RRDSET '%s' given, because it is not linked to rrdcontext structures. Switching to context query.", rrdset_name(qtl->st));
  2103. if(!is_valid_sp(qtl->charts))
  2104. qtl->charts = rrdset_name(qtl->st);
  2105. qtl->st = NULL;
  2106. }
  2107. size_t added = 0;
  2108. if(unlikely(qt->request.rca)) {
  2109. query_target_add_context(qtl, qt->request.rca);
  2110. added++;
  2111. }
  2112. else if(unlikely(qtl->st)) {
  2113. // single chart data queries
  2114. query_target_add_context(qtl, qtl->st->rrdcontext);
  2115. added++;
  2116. }
  2117. else {
  2118. // context pattern queries
  2119. RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_get_and_acquire_item((DICTIONARY *)qtl->host->rrdctx, qtl->contexts);
  2120. if(likely(rca)) {
  2121. // we found it!
  2122. query_target_add_context(qtl, rca);
  2123. rrdcontext_release(rca);
  2124. added++;
  2125. }
  2126. else {
  2127. // Probably it is a pattern, we need to search for it...
  2128. RRDCONTEXT *rc;
  2129. dfe_start_read((DICTIONARY *)qtl->host->rrdctx, rc) {
  2130. if(qt->contexts.pattern && !simple_pattern_matches(qt->contexts.pattern, string2str(rc->id)))
  2131. continue;
  2132. query_target_add_context(qtl, (RRDCONTEXT_ACQUIRED *)rc_dfe.item);
  2133. added++;
  2134. }
  2135. dfe_done(rc);
  2136. }
  2137. }
  2138. if(!added) {
  2139. qt->hosts.used--;
  2140. }
  2141. }
  2142. void query_target_generate_name(QUERY_TARGET *qt) {
  2143. char options_buffer[100 + 1];
  2144. web_client_api_request_v1_data_options_to_string(options_buffer, 100, qt->request.options);
  2145. char resampling_buffer[20 + 1] = "";
  2146. if(qt->request.resampling_time > 1)
  2147. snprintfz(resampling_buffer, 20, "/resampling:%lld", (long long)qt->request.resampling_time);
  2148. char tier_buffer[20 + 1] = "";
  2149. if(qt->request.options & RRDR_OPTION_SELECTED_TIER)
  2150. snprintfz(tier_buffer, 20, "/tier:%zu", qt->request.tier);
  2151. if(qt->request.st)
  2152. snprintfz(qt->id, MAX_QUERY_TARGET_ID_LENGTH, "chart://host:%s/instance:%s/dimensions:%s/after:%lld/before:%lld/points:%zu/group:%s%s/options:%s%s%s"
  2153. , rrdhost_hostname(qt->request.st->rrdhost)
  2154. , rrdset_name(qt->request.st)
  2155. , (qt->request.dimensions) ? qt->request.dimensions : "*"
  2156. , (long long)qt->request.after
  2157. , (long long)qt->request.before
  2158. , qt->request.points
  2159. , web_client_api_request_v1_data_group_to_string(qt->request.group_method)
  2160. , qt->request.group_options?qt->request.group_options:""
  2161. , options_buffer
  2162. , resampling_buffer
  2163. , tier_buffer
  2164. );
  2165. else if(qt->request.host && qt->request.rca && qt->request.ria && qt->request.rma)
  2166. snprintfz(qt->id, MAX_QUERY_TARGET_ID_LENGTH, "metric://host:%s/context:%s/instance:%s/dimension:%s/after:%lld/before:%lld/points:%zu/group:%s%s/options:%s%s%s"
  2167. , rrdhost_hostname(qt->request.host)
  2168. , rrdcontext_acquired_id(qt->request.rca)
  2169. , rrdinstance_acquired_id(qt->request.ria)
  2170. , rrdmetric_acquired_id(qt->request.rma)
  2171. , (long long)qt->request.after
  2172. , (long long)qt->request.before
  2173. , qt->request.points
  2174. , web_client_api_request_v1_data_group_to_string(qt->request.group_method)
  2175. , qt->request.group_options?qt->request.group_options:""
  2176. , options_buffer
  2177. , resampling_buffer
  2178. , tier_buffer
  2179. );
  2180. else
  2181. snprintfz(qt->id, MAX_QUERY_TARGET_ID_LENGTH, "context://host:%s/contexts:%s/instances:%s/dimensions:%s/after:%lld/before:%lld/points:%zu/group:%s%s/options:%s%s%s"
  2182. , (qt->request.host) ? rrdhost_hostname(qt->request.host) : ((qt->request.hosts) ? qt->request.hosts : "*")
  2183. , (qt->request.contexts) ? qt->request.contexts : "*"
  2184. , (qt->request.charts) ? qt->request.charts : "*"
  2185. , (qt->request.dimensions) ? qt->request.dimensions : "*"
  2186. , (long long)qt->request.after
  2187. , (long long)qt->request.before
  2188. , qt->request.points
  2189. , web_client_api_request_v1_data_group_to_string(qt->request.group_method)
  2190. , qt->request.group_options?qt->request.group_options:""
  2191. , options_buffer
  2192. , resampling_buffer
  2193. , tier_buffer
  2194. );
  2195. json_fix_string(qt->id);
  2196. }
  2197. QUERY_TARGET *query_target_create(QUERY_TARGET_REQUEST *qtr) {
  2198. QUERY_TARGET *qt = &thread_query_target;
  2199. if(qt->used)
  2200. fatal("QUERY TARGET: this query target is already used (%zu queries made with this QUERY_TARGET so far).", qt->queries);
  2201. qt->used = true;
  2202. qt->queries++;
  2203. // copy the request into query_thread_target
  2204. qt->request = *qtr;
  2205. query_target_generate_name(qt);
  2206. qt->window.after = qt->request.after;
  2207. qt->window.before = qt->request.before;
  2208. rrdr_relative_window_to_absolute(&qt->window.after, &qt->window.before);
  2209. // prepare our local variables - we need these across all these functions
  2210. QUERY_TARGET_LOCALS qtl = {
  2211. .qt = qt,
  2212. .start_s = now_realtime_sec(),
  2213. .host = qt->request.host,
  2214. .st = qt->request.st,
  2215. .hosts = qt->request.hosts,
  2216. .contexts = qt->request.contexts,
  2217. .charts = qt->request.charts,
  2218. .dimensions = qt->request.dimensions,
  2219. .chart_label_key = qt->request.chart_label_key,
  2220. .charts_labels_filter = qt->request.charts_labels_filter,
  2221. };
  2222. qt->db.minimum_latest_update_every = 0; // it will be updated by query_target_add_query()
  2223. // prepare all the patterns
  2224. qt->hosts.pattern = is_valid_sp(qtl.hosts) ? simple_pattern_create(qtl.hosts, ",|\t\r\n\f\v", SIMPLE_PATTERN_EXACT) : NULL;
  2225. qt->contexts.pattern = is_valid_sp(qtl.contexts) ? simple_pattern_create(qtl.contexts, ",|\t\r\n\f\v", SIMPLE_PATTERN_EXACT) : NULL;
  2226. qt->instances.pattern = is_valid_sp(qtl.charts) ? simple_pattern_create(qtl.charts, ",|\t\r\n\f\v", SIMPLE_PATTERN_EXACT) : NULL;
  2227. qt->query.pattern = is_valid_sp(qtl.dimensions) ? simple_pattern_create(qtl.dimensions, ",|\t\r\n\f\v", SIMPLE_PATTERN_EXACT) : NULL;
  2228. qt->instances.chart_label_key_pattern = is_valid_sp(qtl.chart_label_key) ? simple_pattern_create(qtl.chart_label_key, ",|\t\r\n\f\v", SIMPLE_PATTERN_EXACT) : NULL;
  2229. qt->instances.charts_labels_filter_pattern = is_valid_sp(qtl.charts_labels_filter) ? simple_pattern_create(qtl.charts_labels_filter, ",|\t\r\n\f\v", SIMPLE_PATTERN_EXACT) : NULL;
  2230. qtl.match_ids = qt->request.options & RRDR_OPTION_MATCH_IDS;
  2231. qtl.match_names = qt->request.options & RRDR_OPTION_MATCH_NAMES;
  2232. if(likely(!qtl.match_ids && !qtl.match_names))
  2233. qtl.match_ids = qtl.match_names = true;
  2234. // verify that the chart belongs to the host we are interested
  2235. if(qtl.st) {
  2236. if (!qtl.host) {
  2237. // It is NULL, set it ourselves.
  2238. qtl.host = qtl.st->rrdhost;
  2239. }
  2240. else if (unlikely(qtl.host != qtl.st->rrdhost)) {
  2241. // Oops! A different host!
  2242. error("QUERY TARGET: RRDSET '%s' given does not belong to host '%s'. Switching query host to '%s'",
  2243. rrdset_name(qtl.st), rrdhost_hostname(qtl.host), rrdhost_hostname(qtl.st->rrdhost));
  2244. qtl.host = qtl.st->rrdhost;
  2245. }
  2246. }
  2247. if(qtl.host) {
  2248. // single host query
  2249. query_target_add_host(&qtl, qtl.host);
  2250. qtl.hosts = rrdhost_hostname(qtl.host);
  2251. }
  2252. else {
  2253. // multi host query
  2254. rrd_rdlock();
  2255. rrdhost_foreach_read(qtl.host) {
  2256. if(!qt->hosts.pattern || simple_pattern_matches(qt->hosts.pattern, rrdhost_hostname(qtl.host)))
  2257. query_target_add_host(&qtl, qtl.host);
  2258. }
  2259. rrd_unlock();
  2260. }
  2261. // make sure everything is good
  2262. if(!qt->query.used || !qt->metrics.used || !qt->instances.used || !qt->contexts.used || !qt->hosts.used) {
  2263. internal_error(
  2264. true
  2265. , "QUERY TARGET: query '%s' does not have all the data required. "
  2266. "Matched %u hosts, %u contexts, %u instances, %u dimensions, %u metrics to query, "
  2267. "%zu metrics skipped because they don't have data in the desired time-frame. "
  2268. "Aborting it."
  2269. , qt->id
  2270. , qt->hosts.used
  2271. , qt->contexts.used
  2272. , qt->instances.used
  2273. , qt->metrics.used
  2274. , qt->query.used
  2275. , qtl.metrics_skipped_due_to_not_matching_timeframe
  2276. );
  2277. query_target_release(qt);
  2278. return NULL;
  2279. }
  2280. if(!query_target_calculate_window(qt)) {
  2281. query_target_release(qt);
  2282. return NULL;
  2283. }
  2284. return qt;
  2285. }
  2286. // ----------------------------------------------------------------------------
  2287. // load from SQL
  2288. static void rrdinstance_load_clabel(SQL_CLABEL_DATA *sld, void *data) {
  2289. RRDINSTANCE *ri = data;
  2290. rrdlabels_add(ri->rrdlabels, sld->label_key, sld->label_value, sld->label_source);
  2291. }
  2292. static void rrdinstance_load_dimension(SQL_DIMENSION_DATA *sd, void *data) {
  2293. RRDINSTANCE *ri = data;
  2294. RRDMETRIC trm = {
  2295. .id = string_strdupz(sd->id),
  2296. .name = string_strdupz(sd->name),
  2297. .flags = RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_LOAD_SQL, // no need for atomic
  2298. };
  2299. if(sd->hidden) trm.flags |= RRD_FLAG_HIDDEN;
  2300. uuid_copy(trm.uuid, sd->dim_id);
  2301. dictionary_set(ri->rrdmetrics, string2str(trm.id), &trm, sizeof(trm));
  2302. }
  2303. static void rrdinstance_load_chart_callback(SQL_CHART_DATA *sc, void *data) {
  2304. RRDHOST *host = data;
  2305. RRDCONTEXT tc = {
  2306. .id = string_strdupz(sc->context),
  2307. .title = string_strdupz(sc->title),
  2308. .units = string_strdupz(sc->units),
  2309. .family = string_strdupz(sc->family),
  2310. .priority = sc->priority,
  2311. .chart_type = sc->chart_type,
  2312. .flags = RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_LOAD_SQL, // no need for atomics
  2313. .rrdhost = host,
  2314. };
  2315. RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_set_and_acquire_item((DICTIONARY *)host->rrdctx, string2str(tc.id), &tc, sizeof(tc));
  2316. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  2317. RRDINSTANCE tri = {
  2318. .id = string_strdupz(sc->id),
  2319. .name = string_strdupz(sc->name),
  2320. .title = string_strdupz(sc->title),
  2321. .units = string_strdupz(sc->units),
  2322. .family = string_strdupz(sc->family),
  2323. .chart_type = sc->chart_type,
  2324. .priority = sc->priority,
  2325. .update_every = sc->update_every,
  2326. .flags = RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_LOAD_SQL, // no need for atomics
  2327. };
  2328. uuid_copy(tri.uuid, sc->chart_id);
  2329. RRDINSTANCE_ACQUIRED *ria = (RRDINSTANCE_ACQUIRED *)dictionary_set_and_acquire_item(rc->rrdinstances, sc->id, &tri, sizeof(tri));
  2330. RRDINSTANCE *ri = rrdinstance_acquired_value(ria);
  2331. ctx_get_dimension_list(&ri->uuid, rrdinstance_load_dimension, ri);
  2332. ctx_get_label_list(&ri->uuid, rrdinstance_load_clabel, ri);
  2333. rrdinstance_trigger_updates(ri, __FUNCTION__ );
  2334. rrdinstance_release(ria);
  2335. rrdcontext_release(rca);
  2336. }
  2337. static void rrdcontext_load_context_callback(VERSIONED_CONTEXT_DATA *ctx_data, void *data) {
  2338. RRDHOST *host = data;
  2339. (void)host;
  2340. RRDCONTEXT trc = {
  2341. .id = string_strdupz(ctx_data->id),
  2342. .flags = RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_LOAD_SQL, // no need for atomics
  2343. // no need to set more data here
  2344. // we only need the hub data
  2345. .hub = *ctx_data,
  2346. };
  2347. dictionary_set((DICTIONARY *)host->rrdctx, string2str(trc.id), &trc, sizeof(trc));
  2348. }
  2349. void rrdhost_load_rrdcontext_data(RRDHOST *host) {
  2350. if(host->rrdctx) return;
  2351. rrdhost_create_rrdcontexts(host);
  2352. ctx_get_context_list(&host->host_uuid, rrdcontext_load_context_callback, host);
  2353. ctx_get_chart_list(&host->host_uuid, rrdinstance_load_chart_callback, host);
  2354. RRDCONTEXT *rc;
  2355. dfe_start_read((DICTIONARY *)host->rrdctx, rc) {
  2356. rrdcontext_trigger_updates(rc, __FUNCTION__ );
  2357. }
  2358. dfe_done(rc);
  2359. rrdcontext_garbage_collect_single_host(host, false);
  2360. }
  2361. // ----------------------------------------------------------------------------
  2362. // version hash calculation
  2363. static uint64_t rrdcontext_version_hash_with_callback(
  2364. RRDHOST *host,
  2365. void (*callback)(RRDCONTEXT *, bool, void *),
  2366. bool snapshot,
  2367. void *bundle) {
  2368. if(unlikely(!host || !host->rrdctx)) return 0;
  2369. RRDCONTEXT *rc;
  2370. uint64_t hash = 0;
  2371. // loop through all contexts of the host
  2372. dfe_start_read((DICTIONARY *)host->rrdctx, rc) {
  2373. rrdcontext_lock(rc);
  2374. if(unlikely(rrd_flag_check(rc, RRD_FLAG_HIDDEN))) {
  2375. rrdcontext_unlock(rc);
  2376. continue;
  2377. }
  2378. if(unlikely(callback))
  2379. callback(rc, snapshot, bundle);
  2380. // skip any deleted contexts
  2381. if(unlikely(rrd_flag_is_deleted(rc))) {
  2382. rrdcontext_unlock(rc);
  2383. continue;
  2384. }
  2385. // we use rc->hub.* which has the latest
  2386. // metadata we have sent to the hub
  2387. // if a context is currently queued, rc->hub.* does NOT
  2388. // reflect the queued changes. rc->hub.* is updated with
  2389. // their metadata, after messages are dispatched to hub.
  2390. // when the context is being collected,
  2391. // rc->hub.last_time_t is already zero
  2392. hash += rc->hub.version + rc->hub.last_time_t - rc->hub.first_time_t;
  2393. rrdcontext_unlock(rc);
  2394. }
  2395. dfe_done(rc);
  2396. return hash;
  2397. }
  2398. // ----------------------------------------------------------------------------
  2399. // retention recalculation
  2400. static void rrdcontext_recalculate_context_retention(RRDCONTEXT *rc, RRD_FLAGS reason, bool worker_jobs) {
  2401. rrdcontext_post_process_updates(rc, true, reason, worker_jobs);
  2402. }
  2403. static void rrdcontext_recalculate_host_retention(RRDHOST *host, RRD_FLAGS reason, bool worker_jobs) {
  2404. if(unlikely(!host || !host->rrdctx)) return;
  2405. RRDCONTEXT *rc;
  2406. dfe_start_read((DICTIONARY *)host->rrdctx, rc) {
  2407. rrdcontext_recalculate_context_retention(rc, reason, worker_jobs);
  2408. }
  2409. dfe_done(rc);
  2410. }
  2411. static void rrdcontext_recalculate_retention_all_hosts(void) {
  2412. rrdcontext_next_db_rotation_ut = 0;
  2413. rrd_rdlock();
  2414. RRDHOST *host;
  2415. rrdhost_foreach_read(host) {
  2416. worker_is_busy(WORKER_JOB_RETENTION);
  2417. rrdcontext_recalculate_host_retention(host, RRD_FLAG_UPDATE_REASON_DB_ROTATION, true);
  2418. }
  2419. rrd_unlock();
  2420. }
  2421. // ----------------------------------------------------------------------------
  2422. // garbage collector
  2423. static void rrdmetric_update_retention(RRDMETRIC *rm) {
  2424. time_t min_first_time_t = LONG_MAX, max_last_time_t = 0;
  2425. if(rm->rrddim) {
  2426. min_first_time_t = rrddim_first_entry_t(rm->rrddim);
  2427. max_last_time_t = rrddim_last_entry_t(rm->rrddim);
  2428. }
  2429. #ifdef ENABLE_DBENGINE
  2430. else if (dbengine_enabled) {
  2431. RRDHOST *rrdhost = rm->ri->rc->rrdhost;
  2432. for (size_t tier = 0; tier < storage_tiers; tier++) {
  2433. if(!rrdhost->db[tier].instance) continue;
  2434. time_t first_time_t, last_time_t;
  2435. if (rrdeng_metric_retention_by_uuid(rrdhost->db[tier].instance, &rm->uuid, &first_time_t, &last_time_t) == 0) {
  2436. if (first_time_t < min_first_time_t)
  2437. min_first_time_t = first_time_t;
  2438. if (last_time_t > max_last_time_t)
  2439. max_last_time_t = last_time_t;
  2440. }
  2441. }
  2442. }
  2443. else {
  2444. // cannot get retention
  2445. return;
  2446. }
  2447. #endif
  2448. if(min_first_time_t == LONG_MAX)
  2449. min_first_time_t = 0;
  2450. if(min_first_time_t > max_last_time_t) {
  2451. internal_error(true, "RRDMETRIC: retention of '%s' is flipped", string2str(rm->id));
  2452. time_t tmp = min_first_time_t;
  2453. min_first_time_t = max_last_time_t;
  2454. max_last_time_t = tmp;
  2455. }
  2456. // check if retention changed
  2457. if (min_first_time_t != rm->first_time_t) {
  2458. rm->first_time_t = min_first_time_t;
  2459. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
  2460. }
  2461. if (max_last_time_t != rm->last_time_t) {
  2462. rm->last_time_t = max_last_time_t;
  2463. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
  2464. }
  2465. if(unlikely(!rm->first_time_t && !rm->last_time_t))
  2466. rrd_flag_set_deleted(rm, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  2467. rrd_flag_set(rm, RRD_FLAG_LIVE_RETENTION);
  2468. }
  2469. static inline bool rrdmetric_should_be_deleted(RRDMETRIC *rm) {
  2470. if(likely(!rrd_flag_check(rm, RRD_FLAGS_REQUIRED_FOR_DELETIONS)))
  2471. return false;
  2472. if(likely(rrd_flag_check(rm, RRD_FLAGS_PREVENTING_DELETIONS)))
  2473. return false;
  2474. if(likely(rm->rrddim))
  2475. return false;
  2476. rrdmetric_update_retention(rm);
  2477. if(rm->first_time_t || rm->last_time_t)
  2478. return false;
  2479. return true;
  2480. }
  2481. static inline bool rrdinstance_should_be_deleted(RRDINSTANCE *ri) {
  2482. if(likely(!rrd_flag_check(ri, RRD_FLAGS_REQUIRED_FOR_DELETIONS)))
  2483. return false;
  2484. if(likely(rrd_flag_check(ri, RRD_FLAGS_PREVENTING_DELETIONS)))
  2485. return false;
  2486. if(likely(ri->rrdset))
  2487. return false;
  2488. if(unlikely(dictionary_referenced_items(ri->rrdmetrics) != 0))
  2489. return false;
  2490. if(unlikely(dictionary_entries(ri->rrdmetrics) != 0))
  2491. return false;
  2492. if(ri->first_time_t || ri->last_time_t)
  2493. return false;
  2494. return true;
  2495. }
  2496. static inline bool rrdcontext_should_be_deleted(RRDCONTEXT *rc) {
  2497. if(likely(!rrd_flag_check(rc, RRD_FLAGS_REQUIRED_FOR_DELETIONS)))
  2498. return false;
  2499. if(likely(rrd_flag_check(rc, RRD_FLAGS_PREVENTING_DELETIONS)))
  2500. return false;
  2501. if(unlikely(dictionary_referenced_items(rc->rrdinstances) != 0))
  2502. return false;
  2503. if(unlikely(dictionary_entries(rc->rrdinstances) != 0))
  2504. return false;
  2505. if(unlikely(rc->first_time_t || rc->last_time_t))
  2506. return false;
  2507. return true;
  2508. }
  2509. void rrdcontext_delete_from_sql_unsafe(RRDCONTEXT *rc) {
  2510. // we need to refresh the string pointers in rc->hub
  2511. // in case the context changed values
  2512. rc->hub.id = string2str(rc->id);
  2513. rc->hub.title = string2str(rc->title);
  2514. rc->hub.units = string2str(rc->units);
  2515. rc->hub.family = string2str(rc->family);
  2516. // delete it from SQL
  2517. if(ctx_delete_context(&rc->rrdhost->host_uuid, &rc->hub) != 0)
  2518. error("RRDCONTEXT: failed to delete context '%s' version %"PRIu64" from SQL.", rc->hub.id, rc->hub.version);
  2519. }
  2520. static void rrdcontext_garbage_collect_single_host(RRDHOST *host, bool worker_jobs) {
  2521. internal_error(true, "RRDCONTEXT: garbage collecting context structures of host '%s'", rrdhost_hostname(host));
  2522. RRDCONTEXT *rc;
  2523. dfe_start_reentrant((DICTIONARY *)host->rrdctx, rc) {
  2524. if(unlikely(netdata_exit)) break;
  2525. if(worker_jobs) worker_is_busy(WORKER_JOB_CLEANUP);
  2526. rrdcontext_lock(rc);
  2527. RRDINSTANCE *ri;
  2528. dfe_start_reentrant(rc->rrdinstances, ri) {
  2529. if(unlikely(netdata_exit)) break;
  2530. RRDMETRIC *rm;
  2531. dfe_start_write(ri->rrdmetrics, rm) {
  2532. if(rrdmetric_should_be_deleted(rm)) {
  2533. if(worker_jobs) worker_is_busy(WORKER_JOB_CLEANUP_DELETE);
  2534. if(!dictionary_del(ri->rrdmetrics, string2str(rm->id)))
  2535. error("RRDCONTEXT: metric '%s' of instance '%s' of context '%s' of host '%s', failed to be deleted from rrdmetrics dictionary.",
  2536. string2str(rm->id),
  2537. string2str(ri->id),
  2538. string2str(rc->id),
  2539. rrdhost_hostname(host));
  2540. else
  2541. internal_error(
  2542. true,
  2543. "RRDCONTEXT: metric '%s' of instance '%s' of context '%s' of host '%s', deleted from rrdmetrics dictionary.",
  2544. string2str(rm->id),
  2545. string2str(ri->id),
  2546. string2str(rc->id),
  2547. rrdhost_hostname(host));
  2548. }
  2549. }
  2550. dfe_done(rm);
  2551. if(rrdinstance_should_be_deleted(ri)) {
  2552. if(worker_jobs) worker_is_busy(WORKER_JOB_CLEANUP_DELETE);
  2553. if(!dictionary_del(rc->rrdinstances, string2str(ri->id)))
  2554. error("RRDCONTEXT: instance '%s' of context '%s' of host '%s', failed to be deleted from rrdmetrics dictionary.",
  2555. string2str(ri->id),
  2556. string2str(rc->id),
  2557. rrdhost_hostname(host));
  2558. else
  2559. internal_error(
  2560. true,
  2561. "RRDCONTEXT: instance '%s' of context '%s' of host '%s', deleted from rrdmetrics dictionary.",
  2562. string2str(ri->id),
  2563. string2str(rc->id),
  2564. rrdhost_hostname(host));
  2565. }
  2566. }
  2567. dfe_done(ri);
  2568. if(unlikely(rrdcontext_should_be_deleted(rc))) {
  2569. if(worker_jobs) worker_is_busy(WORKER_JOB_CLEANUP_DELETE);
  2570. rrdcontext_dequeue_from_post_processing(rc);
  2571. rrdcontext_delete_from_sql_unsafe(rc);
  2572. if(!dictionary_del((DICTIONARY *)host->rrdctx, string2str(rc->id)))
  2573. error("RRDCONTEXT: context '%s' of host '%s', failed to be deleted from rrdmetrics dictionary.",
  2574. string2str(rc->id),
  2575. rrdhost_hostname(host));
  2576. else
  2577. internal_error(
  2578. true,
  2579. "RRDCONTEXT: context '%s' of host '%s', deleted from rrdmetrics dictionary.",
  2580. string2str(rc->id),
  2581. rrdhost_hostname(host));
  2582. fprintf(stderr, "RRDCONTEXT: deleted context '%s'", string2str(rc->id));
  2583. }
  2584. // the item is referenced in the dictionary
  2585. // so, it is still here to unlock, even if we have deleted it
  2586. rrdcontext_unlock(rc);
  2587. }
  2588. dfe_done(rc);
  2589. }
  2590. static void rrdcontext_garbage_collect_for_all_hosts(void) {
  2591. rrd_rdlock();
  2592. RRDHOST *host;
  2593. rrdhost_foreach_read(host) {
  2594. rrdcontext_garbage_collect_single_host(host, true);
  2595. }
  2596. rrd_unlock();
  2597. }
  2598. // ----------------------------------------------------------------------------
  2599. // post processing
  2600. static void rrdmetric_process_updates(RRDMETRIC *rm, bool force, RRD_FLAGS reason, bool worker_jobs) {
  2601. if(reason != RRD_FLAG_NONE)
  2602. rrd_flag_set_updated(rm, reason);
  2603. if(!force && !rrd_flag_is_updated(rm) && rrd_flag_check(rm, RRD_FLAG_LIVE_RETENTION) && !rrd_flag_check(rm, RRD_FLAG_UPDATE_REASON_UPDATED_RETENTION))
  2604. return;
  2605. if(worker_jobs)
  2606. worker_is_busy(WORKER_JOB_PP_METRIC);
  2607. if(reason == RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD) {
  2608. rrd_flag_set_archived(rm);
  2609. rrd_flag_set(rm, RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD);
  2610. }
  2611. rrdmetric_update_retention(rm);
  2612. rrd_flag_unset_updated(rm);
  2613. }
  2614. static void rrdinstance_post_process_updates(RRDINSTANCE *ri, bool force, RRD_FLAGS reason, bool worker_jobs) {
  2615. if(reason != RRD_FLAG_NONE)
  2616. rrd_flag_set_updated(ri, reason);
  2617. if(!force && !rrd_flag_is_updated(ri) && rrd_flag_check(ri, RRD_FLAG_LIVE_RETENTION))
  2618. return;
  2619. if(worker_jobs)
  2620. worker_is_busy(WORKER_JOB_PP_INSTANCE);
  2621. time_t min_first_time_t = LONG_MAX, max_last_time_t = 0;
  2622. size_t metrics_active = 0, metrics_deleted = 0;
  2623. bool live_retention = true, currently_collected = false;
  2624. if(dictionary_entries(ri->rrdmetrics) > 0) {
  2625. RRDMETRIC *rm;
  2626. dfe_start_read((DICTIONARY *)ri->rrdmetrics, rm) {
  2627. if(unlikely(netdata_exit)) break;
  2628. RRD_FLAGS reason_to_pass = reason;
  2629. if(rrd_flag_check(ri, RRD_FLAG_UPDATE_REASON_UPDATED_RETENTION))
  2630. reason_to_pass |= RRD_FLAG_UPDATE_REASON_UPDATED_RETENTION;
  2631. rrdmetric_process_updates(rm, force, reason_to_pass, worker_jobs);
  2632. if(unlikely(!rrd_flag_check(rm, RRD_FLAG_LIVE_RETENTION)))
  2633. live_retention = false;
  2634. if (unlikely((rrdmetric_should_be_deleted(rm)))) {
  2635. metrics_deleted++;
  2636. continue;
  2637. }
  2638. if(!currently_collected && rrd_flag_check(rm, RRD_FLAG_COLLECTED) && rm->first_time_t)
  2639. currently_collected = true;
  2640. metrics_active++;
  2641. if (rm->first_time_t && rm->first_time_t < min_first_time_t)
  2642. min_first_time_t = rm->first_time_t;
  2643. if (rm->last_time_t && rm->last_time_t > max_last_time_t)
  2644. max_last_time_t = rm->last_time_t;
  2645. }
  2646. dfe_done(rm);
  2647. }
  2648. if(unlikely(live_retention && !rrd_flag_check(ri, RRD_FLAG_LIVE_RETENTION)))
  2649. rrd_flag_set(ri, RRD_FLAG_LIVE_RETENTION);
  2650. else if(unlikely(!live_retention && rrd_flag_check(ri, RRD_FLAG_LIVE_RETENTION)))
  2651. rrd_flag_clear(ri, RRD_FLAG_LIVE_RETENTION);
  2652. if(unlikely(!metrics_active)) {
  2653. // no metrics available
  2654. if(ri->first_time_t) {
  2655. ri->first_time_t = 0;
  2656. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
  2657. }
  2658. if(ri->last_time_t) {
  2659. ri->last_time_t = 0;
  2660. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
  2661. }
  2662. rrd_flag_set_deleted(ri, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  2663. }
  2664. else {
  2665. // we have active metrics...
  2666. if (unlikely(min_first_time_t == LONG_MAX))
  2667. min_first_time_t = 0;
  2668. if (unlikely(min_first_time_t == 0 || max_last_time_t == 0)) {
  2669. if(ri->first_time_t) {
  2670. ri->first_time_t = 0;
  2671. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
  2672. }
  2673. if(ri->last_time_t) {
  2674. ri->last_time_t = 0;
  2675. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
  2676. }
  2677. if(likely(live_retention))
  2678. rrd_flag_set_deleted(ri, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  2679. }
  2680. else {
  2681. rrd_flag_clear(ri, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  2682. if (unlikely(ri->first_time_t != min_first_time_t)) {
  2683. ri->first_time_t = min_first_time_t;
  2684. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
  2685. }
  2686. if (unlikely(ri->last_time_t != max_last_time_t)) {
  2687. ri->last_time_t = max_last_time_t;
  2688. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
  2689. }
  2690. if(likely(currently_collected))
  2691. rrd_flag_set_collected(ri);
  2692. else
  2693. rrd_flag_set_archived(ri);
  2694. }
  2695. }
  2696. rrd_flag_unset_updated(ri);
  2697. }
  2698. static void rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAGS reason, bool worker_jobs) {
  2699. if(reason != RRD_FLAG_NONE)
  2700. rrd_flag_set_updated(rc, reason);
  2701. if(worker_jobs)
  2702. worker_is_busy(WORKER_JOB_PP_CONTEXT);
  2703. size_t min_priority = LONG_MAX;
  2704. time_t min_first_time_t = LONG_MAX, max_last_time_t = 0;
  2705. size_t instances_active = 0, instances_deleted = 0;
  2706. bool live_retention = true, currently_collected = false, hidden = true;
  2707. if(dictionary_entries(rc->rrdinstances) > 0) {
  2708. RRDINSTANCE *ri;
  2709. dfe_start_reentrant(rc->rrdinstances, ri) {
  2710. if(unlikely(netdata_exit)) break;
  2711. RRD_FLAGS reason_to_pass = reason;
  2712. if(rrd_flag_check(rc, RRD_FLAG_UPDATE_REASON_UPDATED_RETENTION))
  2713. reason_to_pass |= RRD_FLAG_UPDATE_REASON_UPDATED_RETENTION;
  2714. rrdinstance_post_process_updates(ri, force, reason_to_pass, worker_jobs);
  2715. if(unlikely(hidden && !rrd_flag_check(ri, RRD_FLAG_HIDDEN)))
  2716. hidden = false;
  2717. if(unlikely(live_retention && !rrd_flag_check(ri, RRD_FLAG_LIVE_RETENTION)))
  2718. live_retention = false;
  2719. if (unlikely(rrdinstance_should_be_deleted(ri))) {
  2720. instances_deleted++;
  2721. continue;
  2722. }
  2723. if(unlikely(!currently_collected && rrd_flag_is_collected(ri) && ri->first_time_t))
  2724. currently_collected = true;
  2725. internal_error(rc->units != ri->units,
  2726. "RRDCONTEXT: '%s' rrdinstance '%s' has different units, context '%s', instance '%s'",
  2727. string2str(rc->id), string2str(ri->id),
  2728. string2str(rc->units), string2str(ri->units));
  2729. instances_active++;
  2730. if (ri->priority >= RRDCONTEXT_MINIMUM_ALLOWED_PRIORITY && ri->priority < min_priority)
  2731. min_priority = ri->priority;
  2732. if (ri->first_time_t && ri->first_time_t < min_first_time_t)
  2733. min_first_time_t = ri->first_time_t;
  2734. if (ri->last_time_t && ri->last_time_t > max_last_time_t)
  2735. max_last_time_t = ri->last_time_t;
  2736. }
  2737. dfe_done(ri);
  2738. }
  2739. {
  2740. bool previous_hidden = rrd_flag_check(rc, RRD_FLAG_HIDDEN);
  2741. if (hidden != previous_hidden) {
  2742. if (hidden && !rrd_flag_check(rc, RRD_FLAG_HIDDEN))
  2743. rrd_flag_set(rc, RRD_FLAG_HIDDEN);
  2744. else if (!hidden && rrd_flag_check(rc, RRD_FLAG_HIDDEN))
  2745. rrd_flag_clear(rc, RRD_FLAG_HIDDEN);
  2746. }
  2747. bool previous_live_retention = rrd_flag_check(rc, RRD_FLAG_LIVE_RETENTION);
  2748. if (live_retention != previous_live_retention) {
  2749. if (live_retention && !rrd_flag_check(rc, RRD_FLAG_LIVE_RETENTION))
  2750. rrd_flag_set(rc, RRD_FLAG_LIVE_RETENTION);
  2751. else if (!live_retention && rrd_flag_check(rc, RRD_FLAG_LIVE_RETENTION))
  2752. rrd_flag_clear(rc, RRD_FLAG_LIVE_RETENTION);
  2753. }
  2754. }
  2755. rrdcontext_lock(rc);
  2756. rc->pp.executions++;
  2757. if(unlikely(!instances_active)) {
  2758. // we had some instances, but they are gone now...
  2759. if(rc->first_time_t) {
  2760. rc->first_time_t = 0;
  2761. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
  2762. }
  2763. if(rc->last_time_t) {
  2764. rc->last_time_t = 0;
  2765. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
  2766. }
  2767. rrd_flag_set_deleted(rc, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  2768. }
  2769. else {
  2770. // we have some active instances...
  2771. if (unlikely(min_first_time_t == LONG_MAX))
  2772. min_first_time_t = 0;
  2773. if (unlikely(min_first_time_t == 0 && max_last_time_t == 0)) {
  2774. if(rc->first_time_t) {
  2775. rc->first_time_t = 0;
  2776. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
  2777. }
  2778. if(rc->last_time_t) {
  2779. rc->last_time_t = 0;
  2780. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
  2781. }
  2782. rrd_flag_set_deleted(rc, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  2783. }
  2784. else {
  2785. rrd_flag_clear(rc, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  2786. if (unlikely(rc->first_time_t != min_first_time_t)) {
  2787. rc->first_time_t = min_first_time_t;
  2788. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
  2789. }
  2790. if (rc->last_time_t != max_last_time_t) {
  2791. rc->last_time_t = max_last_time_t;
  2792. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
  2793. }
  2794. if(likely(currently_collected))
  2795. rrd_flag_set_collected(rc);
  2796. else
  2797. rrd_flag_set_archived(rc);
  2798. }
  2799. if (min_priority != LONG_MAX && rc->priority != min_priority) {
  2800. rc->priority = min_priority;
  2801. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY);
  2802. }
  2803. }
  2804. if(unlikely(rrd_flag_is_updated(rc) && rc->rrdhost->rrdctx_hub_queue)) {
  2805. if(check_if_cloud_version_changed_unsafe(rc, false)) {
  2806. rc->version = rrdcontext_get_next_version(rc);
  2807. dictionary_set((DICTIONARY *)rc->rrdhost->rrdctx_hub_queue,
  2808. string2str(rc->id), rc, sizeof(*rc));
  2809. }
  2810. }
  2811. rrd_flag_unset_updated(rc);
  2812. rrdcontext_unlock(rc);
  2813. }
  2814. static void rrdcontext_queue_for_post_processing(RRDCONTEXT *rc, const char *function __maybe_unused, RRD_FLAGS flags __maybe_unused) {
  2815. if(unlikely(!rc->rrdhost->rrdctx_post_processing_queue)) return;
  2816. if(!rrd_flag_check(rc, RRD_FLAG_QUEUED_FOR_POST_PROCESSING)) {
  2817. dictionary_set((DICTIONARY *)rc->rrdhost->rrdctx_post_processing_queue,
  2818. string2str(rc->id),
  2819. rc,
  2820. sizeof(*rc));
  2821. #if(defined(NETDATA_INTERNAL_CHECKS) && defined(LOG_POST_PROCESSING_QUEUE_INSERTIONS))
  2822. {
  2823. BUFFER *wb_flags = buffer_create(1000);
  2824. rrd_flags_to_buffer(flags, wb_flags);
  2825. BUFFER *wb_reasons = buffer_create(1000);
  2826. rrd_reasons_to_buffer(flags, wb_reasons);
  2827. internal_error(true, "RRDCONTEXT: '%s' update triggered by function %s(), due to flags: %s, reasons: %s",
  2828. string2str(rc->id), function,
  2829. buffer_tostring(wb_flags),
  2830. buffer_tostring(wb_reasons));
  2831. buffer_free(wb_reasons);
  2832. buffer_free(wb_flags);
  2833. }
  2834. #endif
  2835. }
  2836. }
  2837. static void rrdcontext_dequeue_from_post_processing(RRDCONTEXT *rc) {
  2838. if(unlikely(!rc->rrdhost->rrdctx_post_processing_queue)) return;
  2839. dictionary_del((DICTIONARY *)rc->rrdhost->rrdctx_post_processing_queue, string2str(rc->id));
  2840. }
  2841. static void rrdcontext_post_process_queued_contexts(RRDHOST *host) {
  2842. if(unlikely(!host->rrdctx_post_processing_queue)) return;
  2843. RRDCONTEXT *rc;
  2844. dfe_start_reentrant((DICTIONARY *)host->rrdctx_post_processing_queue, rc) {
  2845. if(unlikely(netdata_exit)) break;
  2846. rrdcontext_dequeue_from_post_processing(rc);
  2847. rrdcontext_post_process_updates(rc, false, RRD_FLAG_NONE, true);
  2848. }
  2849. dfe_done(rc);
  2850. }
  2851. // ----------------------------------------------------------------------------
  2852. // dispatching contexts to cloud
  2853. static uint64_t rrdcontext_get_next_version(RRDCONTEXT *rc) {
  2854. time_t now = now_realtime_sec();
  2855. uint64_t version = MAX(rc->version, rc->hub.version);
  2856. version = MAX((uint64_t)now, version);
  2857. version++;
  2858. return version;
  2859. }
  2860. static void rrdcontext_message_send_unsafe(RRDCONTEXT *rc, bool snapshot __maybe_unused, void *bundle __maybe_unused) {
  2861. // save it, so that we know the last version we sent to hub
  2862. rc->version = rc->hub.version = rrdcontext_get_next_version(rc);
  2863. rc->hub.id = string2str(rc->id);
  2864. rc->hub.title = string2str(rc->title);
  2865. rc->hub.units = string2str(rc->units);
  2866. rc->hub.family = string2str(rc->family);
  2867. rc->hub.chart_type = rrdset_type_name(rc->chart_type);
  2868. rc->hub.priority = rc->priority;
  2869. rc->hub.first_time_t = rc->first_time_t;
  2870. rc->hub.last_time_t = rrd_flag_is_collected(rc) ? 0 : rc->last_time_t;
  2871. rc->hub.deleted = rrd_flag_is_deleted(rc) ? true : false;
  2872. #ifdef ENABLE_ACLK
  2873. struct context_updated message = {
  2874. .id = rc->hub.id,
  2875. .version = rc->hub.version,
  2876. .title = rc->hub.title,
  2877. .units = rc->hub.units,
  2878. .family = rc->hub.family,
  2879. .chart_type = rc->hub.chart_type,
  2880. .priority = rc->hub.priority,
  2881. .first_entry = rc->hub.first_time_t,
  2882. .last_entry = rc->hub.last_time_t,
  2883. .deleted = rc->hub.deleted,
  2884. };
  2885. if(likely(!rrd_flag_check(rc, RRD_FLAG_HIDDEN))) {
  2886. if (snapshot) {
  2887. if (!rc->hub.deleted)
  2888. contexts_snapshot_add_ctx_update(bundle, &message);
  2889. }
  2890. else
  2891. contexts_updated_add_ctx_update(bundle, &message);
  2892. }
  2893. #endif
  2894. // store it to SQL
  2895. if(rrd_flag_is_deleted(rc))
  2896. rrdcontext_delete_from_sql_unsafe(rc);
  2897. else if (ctx_store_context(&rc->rrdhost->host_uuid, &rc->hub) != 0)
  2898. error("RRDCONTEXT: failed to save context '%s' version %"PRIu64" to SQL.", rc->hub.id, rc->hub.version);
  2899. }
  2900. static bool check_if_cloud_version_changed_unsafe(RRDCONTEXT *rc, bool sending __maybe_unused) {
  2901. bool id_changed = false,
  2902. title_changed = false,
  2903. units_changed = false,
  2904. family_changed = false,
  2905. chart_type_changed = false,
  2906. priority_changed = false,
  2907. first_time_changed = false,
  2908. last_time_changed = false,
  2909. deleted_changed = false;
  2910. RRD_FLAGS flags = rrd_flags_get(rc);
  2911. if(unlikely(string2str(rc->id) != rc->hub.id))
  2912. id_changed = true;
  2913. if(unlikely(string2str(rc->title) != rc->hub.title))
  2914. title_changed = true;
  2915. if(unlikely(string2str(rc->units) != rc->hub.units))
  2916. units_changed = true;
  2917. if(unlikely(string2str(rc->family) != rc->hub.family))
  2918. family_changed = true;
  2919. if(unlikely(rrdset_type_name(rc->chart_type) != rc->hub.chart_type))
  2920. chart_type_changed = true;
  2921. if(unlikely(rc->priority != rc->hub.priority))
  2922. priority_changed = true;
  2923. if(unlikely((uint64_t)rc->first_time_t != rc->hub.first_time_t))
  2924. first_time_changed = true;
  2925. if(unlikely((uint64_t)((flags & RRD_FLAG_COLLECTED) ? 0 : rc->last_time_t) != rc->hub.last_time_t))
  2926. last_time_changed = true;
  2927. if(unlikely(((flags & RRD_FLAG_DELETED) ? true : false) != rc->hub.deleted))
  2928. deleted_changed = true;
  2929. if(unlikely(id_changed || title_changed || units_changed || family_changed || chart_type_changed || priority_changed || first_time_changed || last_time_changed || deleted_changed)) {
  2930. internal_error(LOG_TRANSITIONS,
  2931. "RRDCONTEXT: %s NEW VERSION '%s'%s of host '%s', version %"PRIu64", title '%s'%s, units '%s'%s, family '%s'%s, chart type '%s'%s, priority %u%s, first_time_t %ld%s, last_time_t %ld%s, deleted '%s'%s, (queued for %llu ms, expected %llu ms)",
  2932. sending?"SENDING":"QUEUE",
  2933. string2str(rc->id), id_changed ? " (CHANGED)" : "",
  2934. rrdhost_hostname(rc->rrdhost),
  2935. rc->version,
  2936. string2str(rc->title), title_changed ? " (CHANGED)" : "",
  2937. string2str(rc->units), units_changed ? " (CHANGED)" : "",
  2938. string2str(rc->family), family_changed ? " (CHANGED)" : "",
  2939. rrdset_type_name(rc->chart_type), chart_type_changed ? " (CHANGED)" : "",
  2940. rc->priority, priority_changed ? " (CHANGED)" : "",
  2941. rc->first_time_t, first_time_changed ? " (CHANGED)" : "",
  2942. (flags & RRD_FLAG_COLLECTED) ? 0 : rc->last_time_t, last_time_changed ? " (CHANGED)" : "",
  2943. (flags & RRD_FLAG_DELETED) ? "true" : "false", deleted_changed ? " (CHANGED)" : "",
  2944. sending ? (now_realtime_usec() - rc->queue.queued_ut) / USEC_PER_MS : 0,
  2945. sending ? (rc->queue.scheduled_dispatch_ut - rc->queue.queued_ut) / USEC_PER_MS : 0
  2946. );
  2947. return true;
  2948. }
  2949. return false;
  2950. }
  2951. static inline usec_t rrdcontext_calculate_queued_dispatch_time_ut(RRDCONTEXT *rc, usec_t now_ut) {
  2952. if(likely(rc->queue.delay_calc_ut >= rc->queue.queued_ut))
  2953. return rc->queue.scheduled_dispatch_ut;
  2954. RRD_FLAGS flags = rc->queue.queued_flags;
  2955. usec_t delay = LONG_MAX;
  2956. int i;
  2957. struct rrdcontext_reason *reason;
  2958. for(i = 0, reason = &rrdcontext_reasons[i]; reason->name ; reason = &rrdcontext_reasons[++i]) {
  2959. if(unlikely(flags & reason->flag)) {
  2960. if(reason->delay_ut < delay)
  2961. delay = reason->delay_ut;
  2962. }
  2963. }
  2964. if(unlikely(delay == LONG_MAX)) {
  2965. internal_error(true, "RRDCONTEXT: '%s', cannot find minimum delay of flags %x", string2str(rc->id), (unsigned int)flags);
  2966. delay = 60 * USEC_PER_SEC;
  2967. }
  2968. rc->queue.delay_calc_ut = now_ut;
  2969. usec_t dispatch_ut = rc->queue.scheduled_dispatch_ut = rc->queue.queued_ut + delay;
  2970. return dispatch_ut;
  2971. }
  2972. static void rrdcontext_dequeue_from_hub_queue(RRDCONTEXT *rc) {
  2973. dictionary_del((DICTIONARY *)rc->rrdhost->rrdctx_hub_queue, string2str(rc->id));
  2974. }
  2975. static void rrdcontext_dispatch_queued_contexts_to_hub(RRDHOST *host, usec_t now_ut) {
  2976. // check if we have received a streaming command for this host
  2977. if(!rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS) || !aclk_connected || !host->rrdctx_hub_queue)
  2978. return;
  2979. // check if there are queued items to send
  2980. if(!dictionary_entries((DICTIONARY *)host->rrdctx_hub_queue))
  2981. return;
  2982. if(!host->node_id)
  2983. return;
  2984. size_t messages_added = 0;
  2985. contexts_updated_t bundle = NULL;
  2986. RRDCONTEXT *rc;
  2987. dfe_start_reentrant((DICTIONARY *)host->rrdctx_hub_queue, rc) {
  2988. if(unlikely(netdata_exit)) break;
  2989. if(unlikely(messages_added >= MESSAGES_PER_BUNDLE_TO_SEND_TO_HUB_PER_HOST))
  2990. break;
  2991. worker_is_busy(WORKER_JOB_QUEUED);
  2992. usec_t dispatch_ut = rrdcontext_calculate_queued_dispatch_time_ut(rc, now_ut);
  2993. char *claim_id = get_agent_claimid();
  2994. if(unlikely(now_ut >= dispatch_ut) && claim_id) {
  2995. worker_is_busy(WORKER_JOB_CHECK);
  2996. rrdcontext_lock(rc);
  2997. if(check_if_cloud_version_changed_unsafe(rc, true)) {
  2998. worker_is_busy(WORKER_JOB_SEND);
  2999. #ifdef ENABLE_ACLK
  3000. if(!bundle) {
  3001. // prepare the bundle to send the messages
  3002. char uuid[UUID_STR_LEN];
  3003. uuid_unparse_lower(*host->node_id, uuid);
  3004. bundle = contexts_updated_new(claim_id, uuid, 0, now_ut);
  3005. }
  3006. #endif
  3007. // update the hub data of the context, give a new version, pack the message
  3008. // and save an update to SQL
  3009. rrdcontext_message_send_unsafe(rc, false, bundle);
  3010. messages_added++;
  3011. rc->queue.dispatches++;
  3012. rc->queue.dequeued_ut = now_ut;
  3013. }
  3014. else
  3015. rc->version = rc->hub.version;
  3016. // remove it from the queue
  3017. worker_is_busy(WORKER_JOB_DEQUEUE);
  3018. rrdcontext_dequeue_from_hub_queue(rc);
  3019. if(unlikely(rrdcontext_should_be_deleted(rc))) {
  3020. // this is a deleted context - delete it forever...
  3021. worker_is_busy(WORKER_JOB_CLEANUP_DELETE);
  3022. rrdcontext_dequeue_from_post_processing(rc);
  3023. rrdcontext_delete_from_sql_unsafe(rc);
  3024. STRING *id = string_dup(rc->id);
  3025. rrdcontext_unlock(rc);
  3026. // delete it from the master dictionary
  3027. if(!dictionary_del((DICTIONARY *)host->rrdctx, string2str(rc->id)))
  3028. error("RRDCONTEXT: '%s' of host '%s' failed to be deleted from rrdcontext dictionary.",
  3029. string2str(id), rrdhost_hostname(host));
  3030. string_freez(id);
  3031. }
  3032. else
  3033. rrdcontext_unlock(rc);
  3034. }
  3035. freez(claim_id);
  3036. }
  3037. dfe_done(rc);
  3038. #ifdef ENABLE_ACLK
  3039. if(!netdata_exit && bundle) {
  3040. // we have a bundle to send messages
  3041. // update the version hash
  3042. contexts_updated_update_version_hash(bundle, rrdcontext_version_hash(host));
  3043. // send it
  3044. aclk_send_contexts_updated(bundle);
  3045. }
  3046. else if(bundle)
  3047. contexts_updated_delete(bundle);
  3048. #endif
  3049. }
  3050. // ----------------------------------------------------------------------------
  3051. // worker thread
  3052. static void rrdcontext_main_cleanup(void *ptr) {
  3053. struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
  3054. static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
  3055. // custom code
  3056. worker_unregister();
  3057. static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
  3058. }
  3059. void *rrdcontext_main(void *ptr) {
  3060. netdata_thread_cleanup_push(rrdcontext_main_cleanup, ptr);
  3061. worker_register("RRDCONTEXT");
  3062. worker_register_job_name(WORKER_JOB_HOSTS, "hosts");
  3063. worker_register_job_name(WORKER_JOB_CHECK, "dedup checks");
  3064. worker_register_job_name(WORKER_JOB_SEND, "sent contexts");
  3065. worker_register_job_name(WORKER_JOB_DEQUEUE, "deduplicated contexts");
  3066. worker_register_job_name(WORKER_JOB_RETENTION, "metrics retention");
  3067. worker_register_job_name(WORKER_JOB_QUEUED, "queued contexts");
  3068. worker_register_job_name(WORKER_JOB_CLEANUP, "cleanups");
  3069. worker_register_job_name(WORKER_JOB_CLEANUP_DELETE, "deletes");
  3070. worker_register_job_name(WORKER_JOB_PP_METRIC, "check metrics");
  3071. worker_register_job_name(WORKER_JOB_PP_INSTANCE, "check instances");
  3072. worker_register_job_name(WORKER_JOB_PP_CONTEXT, "check contexts");
  3073. worker_register_job_custom_metric(WORKER_JOB_HUB_QUEUE_SIZE, "hub queue size", "contexts", WORKER_METRIC_ABSOLUTE);
  3074. worker_register_job_custom_metric(WORKER_JOB_PP_QUEUE_SIZE, "post processing queue size", "contexts", WORKER_METRIC_ABSOLUTE);
  3075. heartbeat_t hb;
  3076. heartbeat_init(&hb);
  3077. usec_t step = RRDCONTEXT_WORKER_THREAD_HEARTBEAT_USEC;
  3078. while (!netdata_exit) {
  3079. worker_is_idle();
  3080. heartbeat_next(&hb, step);
  3081. if(unlikely(netdata_exit)) break;
  3082. usec_t now_ut = now_realtime_usec();
  3083. if(rrdcontext_next_db_rotation_ut && now_ut > rrdcontext_next_db_rotation_ut) {
  3084. rrdcontext_recalculate_retention_all_hosts();
  3085. rrdcontext_garbage_collect_for_all_hosts();
  3086. rrdcontext_next_db_rotation_ut = 0;
  3087. }
  3088. size_t hub_queued_contexts_for_all_hosts = 0;
  3089. size_t pp_queued_contexts_for_all_hosts = 0;
  3090. rrd_rdlock();
  3091. RRDHOST *host;
  3092. rrdhost_foreach_read(host) {
  3093. if(unlikely(netdata_exit)) break;
  3094. worker_is_busy(WORKER_JOB_HOSTS);
  3095. if(host->rrdctx_post_processing_queue) {
  3096. pp_queued_contexts_for_all_hosts +=
  3097. dictionary_entries((DICTIONARY *)host->rrdctx_post_processing_queue);
  3098. rrdcontext_post_process_queued_contexts(host);
  3099. }
  3100. if(host->rrdctx_hub_queue) {
  3101. hub_queued_contexts_for_all_hosts += dictionary_entries((DICTIONARY *)host->rrdctx_hub_queue);
  3102. rrdcontext_dispatch_queued_contexts_to_hub(host, now_ut);
  3103. }
  3104. }
  3105. rrd_unlock();
  3106. worker_set_metric(WORKER_JOB_HUB_QUEUE_SIZE, (NETDATA_DOUBLE)hub_queued_contexts_for_all_hosts);
  3107. worker_set_metric(WORKER_JOB_PP_QUEUE_SIZE, (NETDATA_DOUBLE)pp_queued_contexts_for_all_hosts);
  3108. }
  3109. netdata_thread_cleanup_pop(1);
  3110. return NULL;
  3111. }