rrdcontext.c 150 KB

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