rrdcontext.c 95 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726
  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_api.h"
  7. int rrdcontext_enabled = CONFIG_BOOLEAN_YES;
  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_SECS 1
  11. #define RRDCONTEXT_MINIMUM_ALLOWED_PRIORITY 10
  12. // #define LOG_TRANSITIONS 1
  13. // #define LOG_RRDINSTANCES 1
  14. typedef enum {
  15. RRD_FLAG_NONE = 0,
  16. RRD_FLAG_DELETED = (1 << 0), // this is a deleted object (metrics, instances, contexts)
  17. RRD_FLAG_COLLECTED = (1 << 1), // this object is currently being collected
  18. RRD_FLAG_UPDATED = (1 << 2), // this object has updates to propagate
  19. RRD_FLAG_ARCHIVED = (1 << 3), // this object is not currently being collected
  20. RRD_FLAG_OWN_LABELS = (1 << 4), // this instance has its own labels - not linked to an RRDSET
  21. RRD_FLAG_LIVE_RETENTION = (1 << 5), // we have got live retention from the database
  22. RRD_FLAG_QUEUED = (1 << 6), // this context is currently queued to be dispatched to hub
  23. RRD_FLAG_UPDATE_REASON_LOAD_SQL = (1 << 10), // this object has just been loaded from SQL
  24. RRD_FLAG_UPDATE_REASON_NEW_OBJECT = (1 << 11), // this object has just been created
  25. RRD_FLAG_UPDATE_REASON_UPDATED_OBJECT = (1 << 12), // we received an update on this object
  26. RRD_FLAG_UPDATE_REASON_CHANGED_LINKING = (1 << 13), // an instance or a metric switched RRDSET or RRDDIM
  27. RRD_FLAG_UPDATE_REASON_CHANGED_UUID = (1 << 14), // an instance or a metric changed UUID
  28. RRD_FLAG_UPDATE_REASON_CHANGED_NAME = (1 << 15), // an instance or a metric changed name
  29. RRD_FLAG_UPDATE_REASON_CHANGED_UNITS = (1 << 16), // this context or instance changed units
  30. RRD_FLAG_UPDATE_REASON_CHANGED_TITLE = (1 << 17), // this context or instance changed title
  31. RRD_FLAG_UPDATE_REASON_CHANGED_FAMILY = (1 << 18), // the context or the instance changed family
  32. RRD_FLAG_UPDATE_REASON_CHANGED_CHART_TYPE = (1 << 19), // this context or instance changed chart type
  33. RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY = (1 << 20), // this context or instance changed its priority
  34. RRD_FLAG_UPDATE_REASON_CHANGED_UPDATE_EVERY = (1 << 21), // the instance or the metric changed update frequency
  35. RRD_FLAG_UPDATE_REASON_ZERO_RETENTION = (1 << 22), // this object has not retention
  36. RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T = (1 << 23), // this object changed its oldest time in the db
  37. RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T = (1 << 24), // this object change its latest time in the db
  38. RRD_FLAG_UPDATE_REASON_STOPPED_BEING_COLLECTED = (1 << 25), // this object has stopped being collected
  39. RRD_FLAG_UPDATE_REASON_STARTED_BEING_COLLECTED = (1 << 26), // this object has started being collected
  40. RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD = (1 << 27), // this context belongs to a host that just disconnected
  41. RRD_FLAG_UPDATE_REASON_DB_ROTATION = (1 << 28), // this context changed because of a db rotation
  42. } RRD_FLAGS;
  43. #define RRD_FLAG_ALL_UPDATE_REASONS ( \
  44. RRD_FLAG_UPDATE_REASON_LOAD_SQL \
  45. |RRD_FLAG_UPDATE_REASON_NEW_OBJECT \
  46. |RRD_FLAG_UPDATE_REASON_UPDATED_OBJECT \
  47. |RRD_FLAG_UPDATE_REASON_CHANGED_LINKING \
  48. |RRD_FLAG_UPDATE_REASON_CHANGED_UUID \
  49. |RRD_FLAG_UPDATE_REASON_CHANGED_NAME \
  50. |RRD_FLAG_UPDATE_REASON_CHANGED_UNITS \
  51. |RRD_FLAG_UPDATE_REASON_CHANGED_TITLE \
  52. |RRD_FLAG_UPDATE_REASON_CHANGED_FAMILY \
  53. |RRD_FLAG_UPDATE_REASON_CHANGED_CHART_TYPE \
  54. |RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY \
  55. |RRD_FLAG_UPDATE_REASON_CHANGED_UPDATE_EVERY \
  56. |RRD_FLAG_UPDATE_REASON_ZERO_RETENTION \
  57. |RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T \
  58. |RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T \
  59. |RRD_FLAG_UPDATE_REASON_STOPPED_BEING_COLLECTED \
  60. |RRD_FLAG_UPDATE_REASON_STARTED_BEING_COLLECTED \
  61. |RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD \
  62. |RRD_FLAG_UPDATE_REASON_DB_ROTATION \
  63. )
  64. #define RRD_FLAGS_PROPAGATED_UPSTREAM ( \
  65. RRD_FLAG_COLLECTED \
  66. |RRD_FLAG_DELETED \
  67. |RRD_FLAG_LIVE_RETENTION \
  68. |RRD_FLAG_ALL_UPDATE_REASONS \
  69. )
  70. #define RRD_FLAGS_ALLOWED_EXTERNALLY_ON_NEW_OBJECTS ( \
  71. RRD_FLAG_ARCHIVED \
  72. |RRD_FLAG_ALL_UPDATE_REASONS \
  73. )
  74. #define RRD_FLAGS_PREVENTING_DELETIONS ( \
  75. RRD_FLAG_QUEUED \
  76. |RRD_FLAG_COLLECTED \
  77. |RRD_FLAG_UPDATE_REASON_LOAD_SQL \
  78. |RRD_FLAG_UPDATE_REASON_NEW_OBJECT \
  79. |RRD_FLAG_UPDATE_REASON_UPDATED_OBJECT \
  80. |RRD_FLAG_UPDATE_REASON_CHANGED_LINKING \
  81. )
  82. #define rrd_flag_set_updated(obj, reason) (obj)->flags |= (RRD_FLAG_UPDATED | (reason))
  83. #define rrd_flag_unset_updated(obj) (obj)->flags &= ~(RRD_FLAG_UPDATED | RRD_FLAG_ALL_UPDATE_REASONS)
  84. #define rrd_flag_set_collected(obj) do { \
  85. if(likely(!((obj)->flags & RRD_FLAG_COLLECTED))) \
  86. (obj)->flags |= (RRD_FLAG_COLLECTED | RRD_FLAG_UPDATE_REASON_STARTED_BEING_COLLECTED | RRD_FLAG_UPDATED); \
  87. if(likely( ((obj)->flags & (RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_STOPPED_BEING_COLLECTED)))) \
  88. (obj)->flags &= ~(RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_STOPPED_BEING_COLLECTED); \
  89. } while(0)
  90. #define rrd_flag_set_archived(obj) do { \
  91. if(likely(!((obj)->flags & RRD_FLAG_ARCHIVED))) \
  92. (obj)->flags |= (RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_STOPPED_BEING_COLLECTED | RRD_FLAG_UPDATED); \
  93. if(likely( ((obj)->flags & (RRD_FLAG_COLLECTED | RRD_FLAG_UPDATE_REASON_STARTED_BEING_COLLECTED)))) \
  94. (obj)->flags &= ~(RRD_FLAG_COLLECTED | RRD_FLAG_UPDATE_REASON_STARTED_BEING_COLLECTED); \
  95. } while(0)
  96. #define rrd_flag_is_collected(obj) ((obj)->flags & RRD_FLAG_COLLECTED)
  97. #define rrd_flag_is_archived(obj) ((obj)->flags & RRD_FLAG_ARCHIVED)
  98. static struct rrdcontext_reason {
  99. RRD_FLAGS flag;
  100. const char *name;
  101. usec_t delay_ut;
  102. } rrdcontext_reasons[] = {
  103. // context related
  104. { RRD_FLAG_UPDATE_REASON_NEW_OBJECT, "object created", 60 * USEC_PER_SEC },
  105. { RRD_FLAG_UPDATE_REASON_UPDATED_OBJECT, "object updated", 60 * USEC_PER_SEC },
  106. { RRD_FLAG_UPDATE_REASON_LOAD_SQL, "loaded from sql", 60 * USEC_PER_SEC },
  107. { RRD_FLAG_UPDATE_REASON_CHANGED_TITLE, "changed title", 30 * USEC_PER_SEC },
  108. { RRD_FLAG_UPDATE_REASON_CHANGED_UNITS, "changed units", 30 * USEC_PER_SEC },
  109. { RRD_FLAG_UPDATE_REASON_CHANGED_FAMILY, "changed family", 30 * USEC_PER_SEC },
  110. { RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY, "changed priority", 30 * USEC_PER_SEC },
  111. { RRD_FLAG_UPDATE_REASON_ZERO_RETENTION, "has no retention", 60 * USEC_PER_SEC },
  112. { RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T, "updated first_time_t", 30 * USEC_PER_SEC },
  113. { RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T, "updated last_time_t", 60 * USEC_PER_SEC },
  114. { RRD_FLAG_UPDATE_REASON_CHANGED_CHART_TYPE, "changed chart type", 30 * USEC_PER_SEC },
  115. { RRD_FLAG_UPDATE_REASON_STOPPED_BEING_COLLECTED, "stopped collected", 60 * USEC_PER_SEC },
  116. { RRD_FLAG_UPDATE_REASON_STARTED_BEING_COLLECTED, "started collected", 0 * USEC_PER_SEC },
  117. // not context related
  118. { RRD_FLAG_UPDATE_REASON_CHANGED_UUID, "changed uuid", 60 * USEC_PER_SEC },
  119. { RRD_FLAG_UPDATE_REASON_CHANGED_UPDATE_EVERY, "changed updated every",60 * USEC_PER_SEC },
  120. { RRD_FLAG_UPDATE_REASON_CHANGED_LINKING, "changed rrd link", 60 * USEC_PER_SEC },
  121. { RRD_FLAG_UPDATE_REASON_CHANGED_NAME, "changed name", 60 * USEC_PER_SEC },
  122. { RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD, "child disconnected", 30 * USEC_PER_SEC },
  123. { RRD_FLAG_UPDATE_REASON_DB_ROTATION, "db rotation", 60 * USEC_PER_SEC },
  124. // terminator
  125. { 0, NULL, 0 },
  126. };
  127. typedef struct rrdmetric {
  128. uuid_t uuid;
  129. STRING *id;
  130. STRING *name;
  131. RRDDIM *rrddim;
  132. time_t first_time_t;
  133. time_t last_time_t;
  134. RRD_FLAGS flags;
  135. struct rrdinstance *ri;
  136. usec_t created_ut; // the time this object was created
  137. } RRDMETRIC;
  138. typedef struct rrdinstance {
  139. uuid_t uuid;
  140. STRING *id;
  141. STRING *name;
  142. STRING *title;
  143. STRING *units;
  144. STRING *family;
  145. uint32_t priority;
  146. RRDSET_TYPE chart_type;
  147. RRD_FLAGS flags; // flags related to this instance
  148. time_t first_time_t;
  149. time_t last_time_t;
  150. int update_every; // data collection frequency
  151. RRDSET *rrdset; // pointer to RRDSET when collected, or NULL
  152. DICTIONARY *rrdlabels; // linked to RRDSET->state->chart_labels or own version
  153. struct rrdcontext *rc;
  154. DICTIONARY *rrdmetrics;
  155. } RRDINSTANCE;
  156. typedef struct rrdcontext {
  157. uint64_t version;
  158. STRING *id;
  159. STRING *title;
  160. STRING *units;
  161. STRING *family;
  162. uint32_t priority;
  163. RRDSET_TYPE chart_type;
  164. RRD_FLAGS flags;
  165. time_t first_time_t;
  166. time_t last_time_t;
  167. VERSIONED_CONTEXT_DATA hub;
  168. DICTIONARY *rrdinstances;
  169. RRDHOST *rrdhost;
  170. struct {
  171. RRD_FLAGS queued_flags; // the last flags that triggered the queueing
  172. usec_t queued_ut; // the last time this was queued
  173. usec_t delay_calc_ut; // the last time we calculated the scheduled_dispatched_ut
  174. usec_t scheduled_dispatch_ut; // the time it was/is scheduled to be sent
  175. usec_t dequeued_ut; // the last time we sent (or deduped) this context
  176. } queue;
  177. netdata_mutex_t mutex;
  178. } RRDCONTEXT;
  179. // ----------------------------------------------------------------------------
  180. // helper one-liners for RRDMETRIC
  181. static inline RRDMETRIC *rrdmetric_acquired_value(RRDMETRIC_ACQUIRED *rma) {
  182. return dictionary_acquired_item_value((DICTIONARY_ITEM *)rma);
  183. }
  184. static inline void rrdmetric_release(RRDMETRIC_ACQUIRED *rma) {
  185. RRDMETRIC *rm = rrdmetric_acquired_value(rma);
  186. dictionary_acquired_item_release(rm->ri->rrdmetrics, (DICTIONARY_ITEM *)rma);
  187. }
  188. // ----------------------------------------------------------------------------
  189. // helper one-liners for RRDINSTANCE
  190. static inline RRDINSTANCE_ACQUIRED *rrdinstance_dup(RRDINSTANCE_ACQUIRED *ria) {
  191. return (RRDINSTANCE_ACQUIRED *)dictionary_acquired_item_dup((DICTIONARY_ITEM *)ria);
  192. }
  193. static inline RRDINSTANCE *rrdinstance_acquired_value(RRDINSTANCE_ACQUIRED *ria) {
  194. return dictionary_acquired_item_value((DICTIONARY_ITEM *)ria);
  195. }
  196. static inline const char *rrdinstance_acquired_name(RRDINSTANCE_ACQUIRED *ria) {
  197. return dictionary_acquired_item_name((DICTIONARY_ITEM *)ria);
  198. }
  199. static inline void rrdinstance_release(RRDINSTANCE_ACQUIRED *ria) {
  200. RRDINSTANCE *ri = rrdinstance_acquired_value(ria);
  201. dictionary_acquired_item_release(ri->rc->rrdinstances, (DICTIONARY_ITEM *)ria);
  202. }
  203. // ----------------------------------------------------------------------------
  204. // helper one-liners for RRDCONTEXT
  205. static inline RRDCONTEXT_ACQUIRED *rrdcontext_dup(RRDCONTEXT_ACQUIRED *rca) {
  206. return (RRDCONTEXT_ACQUIRED *)dictionary_acquired_item_dup((DICTIONARY_ITEM *)rca);
  207. }
  208. static inline const char *rrdcontext_acquired_name(RRDCONTEXT_ACQUIRED *rca) {
  209. return dictionary_acquired_item_name((DICTIONARY_ITEM *)rca);
  210. }
  211. static inline RRDCONTEXT *rrdcontext_acquired_value(RRDCONTEXT_ACQUIRED *rca) {
  212. return dictionary_acquired_item_value((DICTIONARY_ITEM *)rca);
  213. }
  214. static inline RRDCONTEXT_ACQUIRED *rrdcontext_acquire(RRDHOST *host, const char *name) {
  215. return (RRDCONTEXT_ACQUIRED *)dictionary_get_and_acquire_item((DICTIONARY *)host->rrdctx, name);
  216. }
  217. static inline void rrdcontext_release(RRDCONTEXT_ACQUIRED *rca) {
  218. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  219. dictionary_acquired_item_release((DICTIONARY *)rc->rrdhost->rrdctx, (DICTIONARY_ITEM *)rca);
  220. }
  221. static void rrdcontext_recalculate_context_retention(RRDCONTEXT *rc, RRD_FLAGS reason, int job_id);
  222. static void rrdcontext_recalculate_host_retention(RRDHOST *host, RRD_FLAGS reason, int job_id);
  223. #define rrdcontext_version_hash(host) rrdcontext_version_hash_with_callback(host, NULL, false, NULL)
  224. static uint64_t rrdcontext_version_hash_with_callback(RRDHOST *host, void (*callback)(RRDCONTEXT *, bool, void *), bool snapshot, void *bundle);
  225. void rrdcontext_delete_from_sql_unsafe(RRDCONTEXT *rc);
  226. #define rrdcontext_lock(rc) netdata_mutex_lock(&((rc)->mutex))
  227. #define rrdcontext_unlock(rc) netdata_mutex_unlock(&((rc)->mutex))
  228. // ----------------------------------------------------------------------------
  229. // Updates triggers
  230. static void rrdmetric_trigger_updates(RRDMETRIC *rm, bool force, bool escalate);
  231. static void rrdinstance_trigger_updates(RRDINSTANCE *ri, bool force, bool escalate);
  232. static void rrdcontext_trigger_updates(RRDCONTEXT *rc, bool force, RRD_FLAGS reason);
  233. // ----------------------------------------------------------------------------
  234. // visualizing flags
  235. static void rrd_flags_to_buffer(RRD_FLAGS flags, BUFFER *wb) {
  236. if(flags & RRD_FLAG_QUEUED)
  237. buffer_strcat(wb, "QUEUED ");
  238. if(flags & RRD_FLAG_DELETED)
  239. buffer_strcat(wb, "DELETED ");
  240. if(flags & RRD_FLAG_COLLECTED)
  241. buffer_strcat(wb, "COLLECTED ");
  242. if(flags & RRD_FLAG_UPDATED)
  243. buffer_strcat(wb, "UPDATED ");
  244. if(flags & RRD_FLAG_ARCHIVED)
  245. buffer_strcat(wb, "ARCHIVED ");
  246. if(flags & RRD_FLAG_OWN_LABELS)
  247. buffer_strcat(wb, "OWN_LABELS ");
  248. if(flags & RRD_FLAG_LIVE_RETENTION)
  249. buffer_strcat(wb, "LIVE_RETENTION ");
  250. }
  251. static void rrd_reasons_to_buffer(RRD_FLAGS flags, BUFFER *wb) {
  252. for(int i = 0, added = 0; rrdcontext_reasons[i].name ; i++) {
  253. if (flags & rrdcontext_reasons[i].flag) {
  254. if (added)
  255. buffer_strcat(wb, ", ");
  256. buffer_strcat(wb, rrdcontext_reasons[i].name);
  257. added++;
  258. }
  259. }
  260. }
  261. // ----------------------------------------------------------------------------
  262. // logging of all data collected
  263. #ifdef LOG_TRANSITIONS
  264. static void log_transition(STRING *metric, STRING *instance, STRING *context, RRD_FLAGS flags, const char *msg) {
  265. BUFFER *wb = buffer_create(1000);
  266. buffer_sprintf(wb, "RRD TRANSITION: context '%s'", string2str(context));
  267. if(instance)
  268. buffer_sprintf(wb, ", instance '%s'", string2str(instance));
  269. if(metric)
  270. buffer_sprintf(wb, ", metric '%s'", string2str(metric));
  271. buffer_sprintf(wb, ", triggered by %s: ", msg);
  272. rrd_flags_to_buffer(flags, wb);
  273. buffer_strcat(wb, ", reasons: ");
  274. rrd_reasons_to_buffer(flags, wb);
  275. internal_error(true, "%s", buffer_tostring(wb));
  276. buffer_free(wb);
  277. }
  278. #else
  279. #define log_transition(metric, instance, context, flags, msg) debug_dummy()
  280. #endif
  281. #ifdef LOG_RRDINSTANCES
  282. static void rrdinstance_log(RRDINSTANCE *ri, const char *msg) {
  283. char uuid[UUID_STR_LEN];
  284. uuid_unparse(ri->uuid, uuid);
  285. BUFFER *wb = buffer_create(1000);
  286. buffer_sprintf(wb,
  287. "RRDINSTANCE: %s id '%s' (host '%s'), uuid '%s', name '%s', context '%s', title '%s', units '%s', family '%s', priority %zu, chart type '%s', update every %d, rrdset '%s', flags %s%s%s%s%s%s%s, first_time_t %ld, last_time_t %ld",
  288. msg,
  289. string2str(ri->id),
  290. ri->rc->rrdhost->hostname,
  291. uuid,
  292. string2str(ri->name),
  293. string2str(ri->rc->id),
  294. string2str(ri->title),
  295. string2str(ri->units),
  296. string2str(ri->family),
  297. ri->priority,
  298. rrdset_type_name(ri->chart_type),
  299. ri->update_every,
  300. ri->rrdset?ri->rrdset->id:"NONE",
  301. ri->flags & RRD_FLAG_DELETED ?"DELETED ":"",
  302. ri->flags & RRD_FLAG_UPDATED ?"UPDATED ":"",
  303. rrd_flag_is_collected(ri) ?"COLLECTED ":"",
  304. rrd_flag_is_archived(ri) ?"ARCHIVED ":"",
  305. ri->flags & RRD_FLAG_OWNLABELS ?"OWNLABELS ":"",
  306. ri->flags & RRD_FLAG_LIVE_RETENTION ?"LIVE ":"",
  307. ri->flags & RRD_FLAG_QUEUED ?"QUEUED ":"",
  308. ri->first_time_t,
  309. ri->last_time_t
  310. );
  311. buffer_strcat(wb, ", update reasons: { ");
  312. for(int i = 0, added = 0; rrdcontext_reasons[i].name ;i++)
  313. if(ri->flags & rrdcontext_reasons[i].flag) {
  314. if(added) buffer_strcat(wb, ", ");
  315. buffer_strcat(wb, rrdcontext_reasons[i].name);
  316. added++;
  317. }
  318. buffer_strcat(wb, " }");
  319. buffer_strcat(wb, ", labels: { ");
  320. if(ri->rrdlabels) {
  321. if(!rrdlabels_to_buffer(ri->rrdlabels, wb, "", "=", "'", ", ", NULL, NULL, NULL, NULL))
  322. buffer_strcat(wb, "EMPTY }");
  323. else
  324. buffer_strcat(wb, " }");
  325. }
  326. else
  327. buffer_strcat(wb, "NONE }");
  328. buffer_strcat(wb, ", metrics: { ");
  329. if(ri->rrdmetrics) {
  330. RRDMETRIC *v;
  331. int i = 0;
  332. dfe_start_read((DICTIONARY *)ri->rrdmetrics, v) {
  333. buffer_sprintf(wb, "%s%s", i?",":"", v_name);
  334. i++;
  335. }
  336. dfe_done(v);
  337. if(!i)
  338. buffer_strcat(wb, "EMPTY }");
  339. else
  340. buffer_strcat(wb, " }");
  341. }
  342. else
  343. buffer_strcat(wb, "NONE }");
  344. internal_error(true, "%s", buffer_tostring(wb));
  345. buffer_free(wb);
  346. }
  347. #else
  348. #define rrdinstance_log(ir, msg) debug_dummy()
  349. #endif
  350. // ----------------------------------------------------------------------------
  351. // RRDMETRIC
  352. static void rrdmetric_free(RRDMETRIC *rm) {
  353. string_freez(rm->id);
  354. string_freez(rm->name);
  355. rm->id = NULL;
  356. rm->name = NULL;
  357. rm->ri = NULL;
  358. }
  359. static void rrdmetric_update_retention(RRDMETRIC *rm) {
  360. time_t min_first_time_t = LONG_MAX, max_last_time_t = 0;
  361. if(rm->rrddim) {
  362. min_first_time_t = rrddim_first_entry_t(rm->rrddim);
  363. max_last_time_t = rrddim_last_entry_t(rm->rrddim);
  364. }
  365. #ifdef ENABLE_DBENGINE
  366. else {
  367. RRDHOST *rrdhost = rm->ri->rc->rrdhost;
  368. for (int tier = 0; tier < storage_tiers; tier++) {
  369. if(!rrdhost->storage_instance[tier]) continue;
  370. time_t first_time_t, last_time_t;
  371. if (rrdeng_metric_retention_by_uuid(rrdhost->storage_instance[tier], &rm->uuid, &first_time_t, &last_time_t) == 0) {
  372. if (first_time_t < min_first_time_t)
  373. min_first_time_t = first_time_t;
  374. if (last_time_t > max_last_time_t)
  375. max_last_time_t = last_time_t;
  376. }
  377. }
  378. }
  379. #endif
  380. if(min_first_time_t == LONG_MAX)
  381. min_first_time_t = 0;
  382. if(min_first_time_t > max_last_time_t) {
  383. internal_error(true, "RRDMETRIC: retention of '%s' is flipped", string2str(rm->id));
  384. time_t tmp = min_first_time_t;
  385. min_first_time_t = max_last_time_t;
  386. max_last_time_t = tmp;
  387. }
  388. // check if retention changed
  389. if (min_first_time_t != rm->first_time_t) {
  390. rm->first_time_t = min_first_time_t;
  391. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
  392. }
  393. if (max_last_time_t != rm->last_time_t) {
  394. rm->last_time_t = max_last_time_t;
  395. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
  396. }
  397. if(rm->first_time_t == 0 && rm->last_time_t == 0 && (!(rm->flags & RRD_FLAG_DELETED))) {
  398. rm->flags |= RRD_FLAG_DELETED;
  399. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  400. }
  401. rm->flags |= RRD_FLAG_LIVE_RETENTION;
  402. }
  403. // called when this rrdmetric is inserted to the rrdmetrics dictionary of a rrdinstance
  404. static void rrdmetric_insert_callback(const char *id __maybe_unused, void *value, void *data) {
  405. RRDMETRIC *rm = value;
  406. // link it to its parent
  407. rm->ri = data;
  408. // remove flags that we need to figure out at runtime
  409. rm->flags = rm->flags & RRD_FLAGS_ALLOWED_EXTERNALLY_ON_NEW_OBJECTS;
  410. rm->created_ut = now_realtime_usec();
  411. // signal the react callback to do the job
  412. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_NEW_OBJECT);
  413. }
  414. // called when this rrdmetric is deleted from the rrdmetrics dictionary of a rrdinstance
  415. static void rrdmetric_delete_callback(const char *id __maybe_unused, void *value, void *data __maybe_unused) {
  416. RRDMETRIC *rm = value;
  417. internal_error(rm->rrddim, "RRDMETRIC: '%s' is freed but there is a RRDDIM linked to it.", string2str(rm->id));
  418. // free the resources
  419. rrdmetric_free(rm);
  420. }
  421. // called when the same rrdmetric is inserted again to the rrdmetrics dictionary of a rrdinstance
  422. static void rrdmetric_conflict_callback(const char *id __maybe_unused, void *oldv, void *newv, void *data __maybe_unused) {
  423. RRDMETRIC *rm = oldv;
  424. RRDMETRIC *rm_new = newv;
  425. internal_error(rm->id != rm_new->id,
  426. "RRDMETRIC: '%s' cannot change id to '%s'",
  427. string2str(rm->id), string2str(rm_new->id));
  428. if(uuid_compare(rm->uuid, rm_new->uuid) != 0) {
  429. char uuid1[UUID_STR_LEN], uuid2[UUID_STR_LEN];
  430. uuid_unparse(rm->uuid, uuid1);
  431. uuid_unparse(rm_new->uuid, uuid2);
  432. internal_error(true, "RRDMETRIC: '%s' of instance '%s' changed uuid from '%s' to '%s'", string2str(rm->id), string2str(rm->ri->id), uuid1, uuid2);
  433. uuid_copy(rm->uuid, rm_new->uuid);
  434. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_UUID);
  435. }
  436. if(rm->rrddim && rm_new->rrddim && rm->rrddim != rm_new->rrddim) {
  437. rm->rrddim = rm_new->rrddim;
  438. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_LINKING);
  439. }
  440. if(rm->rrddim && uuid_compare(rm->uuid, rm->rrddim->metric_uuid) != 0) {
  441. char uuid1[UUID_STR_LEN], uuid2[UUID_STR_LEN];
  442. uuid_unparse(rm->uuid, uuid1);
  443. uuid_unparse(rm_new->uuid, uuid2);
  444. internal_error(true, "RRDMETRIC: '%s' is linked to RRDDIM '%s' but they have different UUIDs. RRDMETRIC has '%s', RRDDIM has '%s'", string2str(rm->id), rm->rrddim->id, uuid1, uuid2);
  445. }
  446. if(rm->rrddim != rm_new->rrddim)
  447. rm->rrddim = rm_new->rrddim;
  448. if(rm->name != rm_new->name) {
  449. STRING *old = rm->name;
  450. rm->name = string_dup(rm_new->name);
  451. string_freez(old);
  452. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_NAME);
  453. }
  454. if(!rm->first_time_t || (rm_new->first_time_t && rm_new->first_time_t < rm->first_time_t)) {
  455. rm->first_time_t = rm_new->first_time_t;
  456. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
  457. }
  458. if(!rm->last_time_t || (rm_new->last_time_t && rm_new->last_time_t > rm->last_time_t)) {
  459. rm->last_time_t = rm_new->last_time_t;
  460. rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
  461. }
  462. rm->flags |= (rm_new->flags & RRD_FLAGS_ALLOWED_EXTERNALLY_ON_NEW_OBJECTS);
  463. if(rrd_flag_is_collected(rm) && rrd_flag_is_archived(rm))
  464. rrd_flag_set_collected(rm);
  465. if(rm->flags & RRD_FLAG_UPDATED)
  466. rm->flags |= RRD_FLAG_UPDATE_REASON_UPDATED_OBJECT;
  467. rrdmetric_free(rm_new);
  468. // the react callback will continue from here
  469. }
  470. static void rrdmetric_react_callback(const char *id __maybe_unused, void *value, void *data __maybe_unused) {
  471. RRDMETRIC *rm = value;
  472. rrdmetric_trigger_updates(rm, false, true);
  473. }
  474. static void rrdmetrics_create(RRDINSTANCE *ri) {
  475. if(unlikely(!ri)) return;
  476. if(likely(ri->rrdmetrics)) return;
  477. ri->rrdmetrics = dictionary_create(DICTIONARY_FLAG_DONT_OVERWRITE_VALUE);
  478. dictionary_register_insert_callback(ri->rrdmetrics, rrdmetric_insert_callback, (void *)ri);
  479. dictionary_register_delete_callback(ri->rrdmetrics, rrdmetric_delete_callback, (void *)ri);
  480. dictionary_register_conflict_callback(ri->rrdmetrics, rrdmetric_conflict_callback, (void *)ri);
  481. dictionary_register_react_callback(ri->rrdmetrics, rrdmetric_react_callback, (void *)ri);
  482. }
  483. static void rrdmetrics_destroy(RRDINSTANCE *ri) {
  484. if(unlikely(!ri || !ri->rrdmetrics)) return;
  485. dictionary_destroy(ri->rrdmetrics);
  486. ri->rrdmetrics = NULL;
  487. }
  488. static inline bool rrdmetric_should_be_deleted(RRDMETRIC *rm) {
  489. if(likely(!(rm->flags & RRD_FLAG_DELETED)))
  490. return false;
  491. if(likely(!(rm->flags & RRD_FLAG_LIVE_RETENTION)))
  492. return false;
  493. if(unlikely(rm->flags & RRD_FLAGS_PREVENTING_DELETIONS))
  494. return false;
  495. if(likely(rm->rrddim))
  496. return false;
  497. if((now_realtime_usec() - rm->created_ut) < 600 * USEC_PER_SEC)
  498. return false;
  499. rrdmetric_update_retention(rm);
  500. if(rm->first_time_t || rm->last_time_t)
  501. return false;
  502. return true;
  503. }
  504. static void rrdmetric_trigger_updates(RRDMETRIC *rm, bool force, bool escalate) {
  505. if(likely(!force && !(rm->flags & RRD_FLAG_UPDATED))) return;
  506. if(unlikely(rrd_flag_is_collected(rm) && !rm->rrddim))
  507. rrd_flag_set_archived(rm);
  508. if(unlikely((rm->flags & RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD) && rrd_flag_is_collected(rm)))
  509. rrd_flag_set_archived(rm);
  510. rrdmetric_update_retention(rm);
  511. if(unlikely(escalate && rm->flags & RRD_FLAG_UPDATED)) {
  512. log_transition(rm->id, rm->ri->id, rm->ri->rc->id, rm->flags, "RRDMETRIC");
  513. rrdinstance_trigger_updates(rm->ri, true, true);
  514. rrd_flag_unset_updated(rm);
  515. }
  516. }
  517. static inline void rrdmetric_from_rrddim(RRDDIM *rd) {
  518. if(unlikely(!rd->rrdset))
  519. fatal("RRDMETRIC: rrddim '%s' does not have a rrdset.", rd->id);
  520. if(unlikely(!rd->rrdset->rrdhost))
  521. fatal("RRDMETRIC: rrdset '%s' does not have a rrdhost", rd->rrdset->id);
  522. if(unlikely(!rd->rrdset->rrdinstance))
  523. fatal("RRDMETRIC: rrdset '%s' does not have a rrdinstance", rd->rrdset->id);
  524. RRDINSTANCE *ri = rrdinstance_acquired_value(rd->rrdset->rrdinstance);
  525. if(unlikely(!ri->rrdmetrics))
  526. fatal("RRDMETRIC: rrdinstance '%s' does not have a rrdmetrics dictionary", string2str(ri->id));
  527. RRDMETRIC trm = {
  528. .id = string_strdupz(rd->id),
  529. .name = string_strdupz(rd->name),
  530. .flags = RRD_FLAG_NONE,
  531. .rrddim = rd,
  532. };
  533. uuid_copy(trm.uuid, rd->metric_uuid);
  534. RRDMETRIC_ACQUIRED *rma = (RRDMETRIC_ACQUIRED *)dictionary_set_and_acquire_item(ri->rrdmetrics, string2str(trm.id), &trm, sizeof(trm));
  535. if(rd->rrdmetric && rd->rrdmetric != rma)
  536. fatal("RRDMETRIC: dimension '%s' of chart '%s' changed rrdmetric!", rd->id, rd->rrdset->id);
  537. if(rd->rrdmetric)
  538. rrdmetric_release(rd->rrdmetric);
  539. rd->rrdmetric = rma;
  540. }
  541. #define rrddim_get_rrdmetric(rd) rrddim_get_rrdmetric_with_trace(rd, __FUNCTION__)
  542. static inline RRDMETRIC *rrddim_get_rrdmetric_with_trace(RRDDIM *rd, const char *function) {
  543. if(unlikely(!rd->rrdmetric))
  544. fatal("RRDMETRIC: RRDDIM '%s' is not linked to an RRDMETRIC at %s()", rd->id, function);
  545. RRDMETRIC *rm = rrdmetric_acquired_value(rd->rrdmetric);
  546. if(unlikely(rm->rrddim != rd))
  547. fatal("RRDMETRIC: '%s' is not linked to RRDDIM '%s' at %s()", string2str(rm->id), rd->id, function);
  548. return rm;
  549. }
  550. static inline void rrdmetric_rrddim_is_freed(RRDDIM *rd) {
  551. RRDMETRIC *rm = rrddim_get_rrdmetric(rd);
  552. if(unlikely(rrd_flag_is_collected(rm)))
  553. rrd_flag_set_archived(rm);
  554. rm->rrddim = NULL;
  555. rrdmetric_trigger_updates(rm, false, true);
  556. rrdmetric_release(rd->rrdmetric);
  557. rd->rrdmetric = NULL;
  558. }
  559. static inline void rrdmetric_updated_rrddim_flags(RRDDIM *rd) {
  560. RRDMETRIC *rm = rrddim_get_rrdmetric(rd);
  561. if(unlikely(rd->flags & (RRDDIM_FLAG_ARCHIVED | RRDDIM_FLAG_OBSOLETE))) {
  562. if(unlikely(rrd_flag_is_collected(rm)))
  563. rrd_flag_set_archived(rm);
  564. }
  565. rrdmetric_trigger_updates(rm, false, true);
  566. }
  567. static inline void rrdmetric_collected_rrddim(RRDDIM *rd) {
  568. RRDMETRIC *rm = rrddim_get_rrdmetric(rd);
  569. if(unlikely(!rrd_flag_is_collected(rm)))
  570. rrd_flag_set_collected(rm);
  571. rrdmetric_trigger_updates(rm, false, true);
  572. }
  573. // ----------------------------------------------------------------------------
  574. // RRDINSTANCE
  575. static void rrdinstance_free(RRDINSTANCE *ri) {
  576. if(ri->flags & RRD_FLAG_OWN_LABELS)
  577. dictionary_destroy(ri->rrdlabels);
  578. rrdmetrics_destroy(ri);
  579. string_freez(ri->id);
  580. string_freez(ri->name);
  581. string_freez(ri->title);
  582. string_freez(ri->units);
  583. string_freez(ri->family);
  584. ri->id = NULL;
  585. ri->name = NULL;
  586. ri->title = NULL;
  587. ri->units = NULL;
  588. ri->family = NULL;
  589. ri->rc = NULL;
  590. ri->rrdlabels = NULL;
  591. ri->rrdmetrics = NULL;
  592. ri->rrdset = NULL;
  593. }
  594. static void rrdinstance_insert_callback(const char *id __maybe_unused, void *value, void *data) {
  595. RRDINSTANCE *ri = value;
  596. // link it to its parent
  597. ri->rc = data;
  598. if(!ri->name)
  599. ri->name = string_dup(ri->id);
  600. if(ri->rrdset && ri->rrdset->state) {
  601. ri->rrdlabels = ri->rrdset->state->chart_labels;
  602. if(ri->flags & RRD_FLAG_OWN_LABELS)
  603. ri->flags &= ~RRD_FLAG_OWN_LABELS;
  604. }
  605. else {
  606. ri->rrdlabels = rrdlabels_create();
  607. ri->flags |= RRD_FLAG_OWN_LABELS;
  608. }
  609. rrdmetrics_create(ri);
  610. rrdinstance_log(ri, "INSERT");
  611. // signal the react callback to do the job
  612. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_NEW_OBJECT);
  613. }
  614. static void rrdinstance_delete_callback(const char *id, void *value, void *data) {
  615. (void)id;
  616. RRDCONTEXT *rc = data; (void)rc;
  617. RRDINSTANCE *ri = (RRDINSTANCE *)value;
  618. rrdinstance_log(ri, "DELETE");
  619. internal_error(ri->rrdset, "RRDINSTANCE: '%s' is freed but there is a RRDSET linked to it.", string2str(ri->id));
  620. rrdinstance_free(ri);
  621. }
  622. static void rrdinstance_conflict_callback(const char *id __maybe_unused, void *oldv, void *newv, void *data __maybe_unused) {
  623. RRDINSTANCE *ri = (RRDINSTANCE *)oldv;
  624. RRDINSTANCE *ri_new = (RRDINSTANCE *)newv;
  625. internal_error(ri->id != ri_new->id,
  626. "RRDINSTANCE: '%s' cannot change id to '%s'",
  627. string2str(ri->id), string2str(ri_new->id));
  628. if(uuid_compare(ri->uuid, ri_new->uuid) != 0) {
  629. uuid_copy(ri->uuid, ri_new->uuid);
  630. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_UUID);
  631. }
  632. if(ri->rrdset && ri_new->rrdset && ri->rrdset != ri_new->rrdset) {
  633. ri->rrdset = ri_new->rrdset;
  634. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_LINKING);
  635. }
  636. if(ri->rrdset && ri->rrdset->chart_uuid && uuid_compare(ri->uuid, *ri->rrdset->chart_uuid) != 0) {
  637. char uuid1[UUID_STR_LEN], uuid2[UUID_STR_LEN];
  638. uuid_unparse(ri->uuid, uuid1);
  639. uuid_unparse(*ri->rrdset->chart_uuid, uuid2);
  640. internal_error(true, "RRDINSTANCE: '%s' is linked to RRDSET '%s' but they have different UUIDs. RRDINSTANCE has '%s', RRDSET has '%s'", string2str(ri->id), ri->rrdset->id, uuid1, uuid2);
  641. }
  642. if(ri->name != ri_new->name) {
  643. STRING *old = ri->name;
  644. ri->name = string_dup(ri_new->name);
  645. string_freez(old);
  646. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_NAME);
  647. }
  648. if(ri->title != ri_new->title) {
  649. STRING *old = ri->title;
  650. ri->title = string_dup(ri_new->title);
  651. string_freez(old);
  652. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_TITLE);
  653. }
  654. if(ri->units != ri_new->units) {
  655. STRING *old = ri->units;
  656. ri->units = string_dup(ri_new->units);
  657. string_freez(old);
  658. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_UNITS);
  659. }
  660. if(ri->family != ri_new->family) {
  661. STRING *old = ri->family;
  662. ri->family = string_dup(ri_new->family);
  663. string_freez(old);
  664. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_FAMILY);
  665. }
  666. if(ri->chart_type != ri_new->chart_type) {
  667. ri->chart_type = ri_new->chart_type;
  668. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_CHART_TYPE);
  669. }
  670. if(ri->priority != ri_new->priority) {
  671. ri->priority = ri_new->priority;
  672. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY);
  673. }
  674. if(ri->update_every != ri_new->update_every) {
  675. ri->update_every = ri_new->update_every;
  676. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_UPDATE_EVERY);
  677. }
  678. if(ri->rrdset != ri_new->rrdset) {
  679. ri->rrdset = ri_new->rrdset;
  680. if(ri->flags & RRD_FLAG_OWN_LABELS) {
  681. DICTIONARY *old = ri->rrdlabels;
  682. ri->rrdlabels = ri->rrdset->state->chart_labels;
  683. ri->flags &= ~RRD_FLAG_OWN_LABELS;
  684. rrdlabels_destroy(old);
  685. }
  686. }
  687. ri->flags |= (ri_new->flags & RRD_FLAGS_ALLOWED_EXTERNALLY_ON_NEW_OBJECTS);
  688. if(rrd_flag_is_collected(ri) && rrd_flag_is_archived(ri))
  689. rrd_flag_set_collected(ri);
  690. if(ri->flags & RRD_FLAG_UPDATED)
  691. ri->flags |= RRD_FLAG_UPDATE_REASON_UPDATED_OBJECT;
  692. rrdinstance_log(ri, "CONFLICT");
  693. // free the new one
  694. rrdinstance_free(ri_new);
  695. // the react callback will continue from here
  696. }
  697. static void rrdinstance_react_callback(const char *id __maybe_unused, void *value, void *data __maybe_unused) {
  698. RRDINSTANCE *ri = value;
  699. rrdinstance_trigger_updates(ri, false, true);
  700. }
  701. void rrdinstances_create(RRDCONTEXT *rc) {
  702. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  703. return;
  704. if(unlikely(!rc || rc->rrdinstances)) return;
  705. rc->rrdinstances = dictionary_create(DICTIONARY_FLAG_DONT_OVERWRITE_VALUE);
  706. dictionary_register_insert_callback(rc->rrdinstances, rrdinstance_insert_callback, (void *)rc);
  707. dictionary_register_delete_callback(rc->rrdinstances, rrdinstance_delete_callback, (void *)rc);
  708. dictionary_register_conflict_callback(rc->rrdinstances, rrdinstance_conflict_callback, (void *)rc);
  709. dictionary_register_react_callback(rc->rrdinstances, rrdinstance_react_callback, (void *)rc);
  710. }
  711. void rrdinstances_destroy(RRDCONTEXT *rc) {
  712. if(unlikely(!rc || !rc->rrdinstances)) return;
  713. dictionary_destroy(rc->rrdinstances);
  714. rc->rrdinstances = NULL;
  715. }
  716. static inline bool rrdinstance_should_be_deleted(RRDINSTANCE *ri) {
  717. if(likely(!(ri->flags & RRD_FLAG_DELETED)))
  718. return false;
  719. if(likely(!(ri->flags & RRD_FLAG_LIVE_RETENTION)))
  720. return false;
  721. if(unlikely(ri->flags & RRD_FLAGS_PREVENTING_DELETIONS))
  722. return false;
  723. if(likely(ri->rrdset))
  724. return false;
  725. if(unlikely(dictionary_stats_referenced_items(ri->rrdmetrics) != 0))
  726. return false;
  727. if(unlikely(dictionary_stats_entries(ri->rrdmetrics) != 0))
  728. return false;
  729. if(ri->first_time_t || ri->last_time_t)
  730. return false;
  731. return true;
  732. }
  733. static void rrdinstance_trigger_updates(RRDINSTANCE *ri, bool force, bool escalate) {
  734. if(unlikely(!force && !(ri->flags & RRD_FLAG_UPDATED))) return;
  735. if(likely(ri->rrdset)) {
  736. if(unlikely(ri->rrdset->priority != ri->priority)) {
  737. ri->priority = ri->rrdset->priority;
  738. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY);
  739. }
  740. if(unlikely(ri->rrdset->update_every != ri->update_every)) {
  741. ri->update_every = ri->rrdset->update_every;
  742. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_UPDATE_EVERY);
  743. }
  744. }
  745. else if(unlikely(rrd_flag_is_collected(ri))) {
  746. rrd_flag_set_archived(ri);
  747. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_LINKING);
  748. }
  749. RRD_FLAGS combined_metrics_flags = RRD_FLAG_NONE;
  750. time_t min_first_time_t = LONG_MAX, max_last_time_t = 0;
  751. size_t metrics_active = 0, metrics_deleted = 0;
  752. {
  753. RRDMETRIC *rm;
  754. dfe_start_read((DICTIONARY *)ri->rrdmetrics, rm) {
  755. // find the combined flags of all the metrics
  756. combined_metrics_flags |= rm->flags & RRD_FLAGS_PROPAGATED_UPSTREAM;
  757. if (unlikely((rrdmetric_should_be_deleted(rm)))) {
  758. metrics_deleted++;
  759. continue;
  760. }
  761. metrics_active++;
  762. if (rm->first_time_t && rm->first_time_t < min_first_time_t)
  763. min_first_time_t = rm->first_time_t;
  764. if (rm->last_time_t && rm->last_time_t > max_last_time_t)
  765. max_last_time_t = rm->last_time_t;
  766. }
  767. dfe_done(rm);
  768. }
  769. // remove the deleted flag - we will recalculate it below
  770. ri->flags &= ~RRD_FLAG_DELETED;
  771. if(unlikely(!metrics_active && metrics_deleted)) {
  772. // we had some metrics, but there are gone now...
  773. ri->flags |= RRD_FLAG_DELETED;
  774. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  775. }
  776. else if(metrics_active) {
  777. // we have active metrics...
  778. if (unlikely(min_first_time_t == LONG_MAX))
  779. min_first_time_t = 0;
  780. if (unlikely(min_first_time_t == 0 || max_last_time_t == 0)) {
  781. ri->first_time_t = 0;
  782. ri->last_time_t = 0;
  783. if(unlikely(combined_metrics_flags & RRD_FLAG_LIVE_RETENTION)) {
  784. ri->flags |= RRD_FLAG_DELETED;
  785. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  786. }
  787. }
  788. else {
  789. if (unlikely(ri->first_time_t != min_first_time_t)) {
  790. ri->first_time_t = min_first_time_t;
  791. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
  792. }
  793. if (unlikely(ri->last_time_t != max_last_time_t)) {
  794. ri->last_time_t = max_last_time_t;
  795. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
  796. }
  797. }
  798. if(combined_metrics_flags & RRD_FLAG_LIVE_RETENTION)
  799. ri->flags |= RRD_FLAG_LIVE_RETENTION;
  800. else
  801. ri->flags &= ~RRD_FLAG_LIVE_RETENTION;
  802. if(likely(combined_metrics_flags & RRD_FLAG_COLLECTED))
  803. rrd_flag_set_collected(ri);
  804. else
  805. rrd_flag_set_archived(ri);
  806. }
  807. else {
  808. // no deleted metrics, no active metrics
  809. // just hanging there...
  810. rrd_flag_set_archived(ri);
  811. }
  812. if(unlikely(escalate && ri->flags & RRD_FLAG_UPDATED)) {
  813. log_transition(NULL, ri->id, ri->rc->id, ri->flags, "RRDINSTANCE");
  814. rrdcontext_trigger_updates(ri->rc, true, RRD_FLAG_NONE);
  815. rrd_flag_unset_updated(ri);
  816. }
  817. }
  818. static inline void rrdinstance_from_rrdset(RRDSET *st) {
  819. RRDCONTEXT trc = {
  820. .id = string_strdupz(st->context),
  821. .title = string_strdupz(st->title),
  822. .units = string_strdupz(st->units),
  823. .family = string_strdupz(st->family),
  824. .priority = st->priority,
  825. .chart_type = st->chart_type,
  826. .flags = RRD_FLAG_NONE,
  827. .rrdhost = st->rrdhost,
  828. };
  829. RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_set_and_acquire_item((DICTIONARY *)st->rrdhost->rrdctx, string2str(trc.id), &trc, sizeof(trc));
  830. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  831. RRDINSTANCE tri = {
  832. .id = string_strdupz(st->id),
  833. .name = string_strdupz(st->name),
  834. .units = string_strdupz(st->units),
  835. .family = string_strdupz(st->family),
  836. .title = string_strdupz(st->title),
  837. .chart_type = st->chart_type,
  838. .priority = st->priority,
  839. .update_every = st->update_every,
  840. .flags = RRD_FLAG_NONE,
  841. .rrdset = st,
  842. };
  843. uuid_copy(tri.uuid, *st->chart_uuid);
  844. RRDINSTANCE_ACQUIRED *ria = (RRDINSTANCE_ACQUIRED *)dictionary_set_and_acquire_item(rc->rrdinstances, string2str(tri.id), &tri, sizeof(tri));
  845. if(st->rrdinstance)
  846. rrdinstance_release(st->rrdinstance);
  847. st->rrdinstance = ria;
  848. if(st->rrdcontext && st->rrdcontext != rca) {
  849. // the chart changed context
  850. RRDCONTEXT *rc_old = rrdcontext_acquired_value(st->rrdcontext);
  851. dictionary_del(rc_old->rrdinstances, st->id);
  852. rrdcontext_trigger_updates(rc_old, true, RRD_FLAG_UPDATE_REASON_CHANGED_LINKING);
  853. }
  854. if(st->rrdcontext)
  855. rrdcontext_release(st->rrdcontext);
  856. st->rrdcontext = rca;
  857. }
  858. #define rrdset_get_rrdinstance(st) rrdset_get_rrdinstance_with_trace(st, __FUNCTION__);
  859. static inline RRDINSTANCE *rrdset_get_rrdinstance_with_trace(RRDSET *st, const char *function) {
  860. if(unlikely(!st->rrdinstance))
  861. fatal("RRDINSTANCE: RRDSET '%s' is not linked to an RRDINSTANCE at %s()", st->id, function);
  862. RRDINSTANCE *ri = rrdinstance_acquired_value(st->rrdinstance);
  863. if(unlikely(ri->rrdset != st))
  864. fatal("RRDINSTANCE: '%s' is not linked to RRDSET '%s' at %s()", string2str(ri->id), st->id, function);
  865. return ri;
  866. }
  867. static inline void rrdinstance_rrdset_is_freed(RRDSET *st) {
  868. RRDINSTANCE *ri = rrdset_get_rrdinstance(st);
  869. rrd_flag_set_archived(ri);
  870. if(!(ri->flags & RRD_FLAG_OWN_LABELS)) {
  871. ri->flags |= RRD_FLAG_OWN_LABELS;
  872. ri->rrdlabels = rrdlabels_create();
  873. rrdlabels_copy(ri->rrdlabels, st->state->chart_labels);
  874. }
  875. ri->rrdset = NULL;
  876. rrdinstance_trigger_updates(ri, false, true);
  877. rrdinstance_release(st->rrdinstance);
  878. st->rrdinstance = NULL;
  879. rrdcontext_release(st->rrdcontext);
  880. st->rrdcontext = NULL;
  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. STRING *old = ri->name;
  887. ri->name = string_strdupz(st->name);
  888. if(ri->name != old)
  889. rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_NAME);
  890. string_freez(old);
  891. rrdinstance_trigger_updates(ri, false, true);
  892. }
  893. static inline void rrdinstance_updated_rrdset_flags(RRDSET *st) {
  894. RRDINSTANCE *ri = rrdset_get_rrdinstance(st);
  895. if(unlikely(st->flags & (RRDSET_FLAG_ARCHIVED | RRDSET_FLAG_OBSOLETE)))
  896. rrd_flag_set_archived(ri);
  897. rrdinstance_trigger_updates(ri, false, true);
  898. }
  899. static inline void rrdinstance_collected_rrdset(RRDSET *st) {
  900. RRDINSTANCE *ri = rrdset_get_rrdinstance(st);
  901. if(unlikely(!rrd_flag_is_collected(ri)))
  902. rrd_flag_set_collected(ri);
  903. rrdinstance_trigger_updates(ri, false, true);
  904. }
  905. // ----------------------------------------------------------------------------
  906. // RRDCONTEXT
  907. static void rrdcontext_freez(RRDCONTEXT *rc) {
  908. string_freez(rc->id);
  909. string_freez(rc->title);
  910. string_freez(rc->units);
  911. string_freez(rc->family);
  912. }
  913. static uint64_t rrdcontext_get_next_version(RRDCONTEXT *rc) {
  914. time_t now = now_realtime_sec();
  915. uint64_t version = MAX(rc->version, rc->hub.version);
  916. version = MAX((uint64_t)now, version);
  917. version++;
  918. return version;
  919. }
  920. static void rrdcontext_message_send_unsafe(RRDCONTEXT *rc, bool snapshot __maybe_unused, void *bundle __maybe_unused) {
  921. // save it, so that we know the last version we sent to hub
  922. rc->version = rc->hub.version = rrdcontext_get_next_version(rc);
  923. rc->hub.id = string2str(rc->id);
  924. rc->hub.title = string2str(rc->title);
  925. rc->hub.units = string2str(rc->units);
  926. rc->hub.family = string2str(rc->family);
  927. rc->hub.chart_type = rrdset_type_name(rc->chart_type);
  928. rc->hub.priority = rc->priority;
  929. rc->hub.first_time_t = rc->first_time_t;
  930. rc->hub.last_time_t = rrd_flag_is_collected(rc) ? 0 : rc->last_time_t;
  931. rc->hub.deleted = (rc->flags & RRD_FLAG_DELETED) ? true : false;
  932. #ifdef ENABLE_ACLK
  933. struct context_updated message = {
  934. .id = rc->hub.id,
  935. .version = rc->hub.version,
  936. .title = rc->hub.title,
  937. .units = rc->hub.units,
  938. .family = rc->hub.family,
  939. .chart_type = rc->hub.chart_type,
  940. .priority = rc->hub.priority,
  941. .first_entry = rc->hub.first_time_t,
  942. .last_entry = rc->hub.last_time_t,
  943. .deleted = rc->hub.deleted,
  944. };
  945. if(snapshot)
  946. contexts_snapshot_add_ctx_update(bundle, &message);
  947. else
  948. contexts_updated_add_ctx_update(bundle, &message);
  949. #endif
  950. // store it to SQL
  951. if(rc->flags & RRD_FLAG_DELETED) {
  952. rrdcontext_delete_from_sql_unsafe(rc);
  953. }
  954. else {
  955. if (ctx_store_context(&rc->rrdhost->host_uuid, &rc->hub) != 0)
  956. error("RRDCONTEXT: failed to save context '%s' version %lu to SQL.", rc->hub.id, rc->hub.version);
  957. }
  958. }
  959. static bool check_if_cloud_version_changed_unsafe(RRDCONTEXT *rc, bool sending) {
  960. bool id_changed = false,
  961. title_changed = false,
  962. units_changed = false,
  963. family_changed = false,
  964. chart_type_changed = false,
  965. priority_changed = false,
  966. first_time_changed = false,
  967. last_time_changed = false,
  968. deleted_changed = false;
  969. if(unlikely(string2str(rc->id) != rc->hub.id))
  970. id_changed = true;
  971. if(unlikely(string2str(rc->title) != rc->hub.title))
  972. title_changed = true;
  973. if(unlikely(string2str(rc->units) != rc->hub.units))
  974. units_changed = true;
  975. if(unlikely(string2str(rc->family) != rc->hub.family))
  976. family_changed = true;
  977. if(unlikely(rrdset_type_name(rc->chart_type) != rc->hub.chart_type))
  978. chart_type_changed = true;
  979. if(unlikely(rc->priority != rc->hub.priority))
  980. priority_changed = true;
  981. if(unlikely((uint64_t)rc->first_time_t != rc->hub.first_time_t))
  982. first_time_changed = true;
  983. if(unlikely((uint64_t)(rrd_flag_is_collected(rc) ? 0 : rc->last_time_t) != rc->hub.last_time_t))
  984. last_time_changed = true;
  985. if(unlikely(((rc->flags & RRD_FLAG_DELETED) ? true : false) != rc->hub.deleted))
  986. deleted_changed = true;
  987. if(unlikely(id_changed || title_changed || units_changed || family_changed || chart_type_changed || priority_changed || first_time_changed || last_time_changed || deleted_changed)) {
  988. internal_error(true, "RRDCONTEXT: %s NEW VERSION '%s'%s, version %zu, 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)",
  989. sending?"SENDING":"QUEUE",
  990. string2str(rc->id), id_changed ? " (CHANGED)" : "",
  991. rc->version,
  992. string2str(rc->title), title_changed ? " (CHANGED)" : "",
  993. string2str(rc->units), units_changed ? " (CHANGED)" : "",
  994. string2str(rc->family), family_changed ? " (CHANGED)" : "",
  995. rrdset_type_name(rc->chart_type), chart_type_changed ? " (CHANGED)" : "",
  996. rc->priority, priority_changed ? " (CHANGED)" : "",
  997. rc->first_time_t, first_time_changed ? " (CHANGED)" : "",
  998. rrd_flag_is_collected(rc) ? 0 : rc->last_time_t, last_time_changed ? " (CHANGED)" : "",
  999. (rc->flags & RRD_FLAG_DELETED) ? "true" : "false", deleted_changed ? " (CHANGED)" : "",
  1000. sending ? (now_realtime_usec() - rc->queue.queued_ut) / USEC_PER_MS : 0,
  1001. sending ? (rc->queue.scheduled_dispatch_ut - rc->queue.queued_ut) / USEC_PER_SEC : 0
  1002. );
  1003. return true;
  1004. }
  1005. return false;
  1006. }
  1007. static void rrdcontext_insert_callback(const char *id, void *value, void *data) {
  1008. (void)id;
  1009. RRDHOST *host = (RRDHOST *)data;
  1010. RRDCONTEXT *rc = (RRDCONTEXT *)value;
  1011. rc->rrdhost = host;
  1012. if(rc->hub.version) {
  1013. // we are loading data from the SQL database
  1014. if(rc->version)
  1015. error("RRDCONTEXT: context '%s' is already initialized with version %lu, but it is loaded again from SQL with version %lu", string2str(rc->id), rc->version, rc->hub.version);
  1016. // IMPORTANT
  1017. // replace all string pointers in rc->hub with our own versions
  1018. // the originals are coming from a tmp allocation of sqlite
  1019. string_freez(rc->id);
  1020. rc->id = string_strdupz(rc->hub.id);
  1021. rc->hub.id = string2str(rc->id);
  1022. string_freez(rc->title);
  1023. rc->title = string_strdupz(rc->hub.title);
  1024. rc->hub.title = string2str(rc->title);
  1025. string_freez(rc->units);
  1026. rc->units = string_strdupz(rc->hub.units);
  1027. rc->hub.units = string2str(rc->units);
  1028. string_freez(rc->family);
  1029. rc->family = string_strdupz(rc->hub.family);
  1030. rc->hub.family = string2str(rc->family);
  1031. rc->chart_type = rrdset_type_id(rc->hub.chart_type);
  1032. rc->hub.chart_type = rrdset_type_name(rc->chart_type);
  1033. rc->version = rc->hub.version;
  1034. rc->priority = rc->hub.priority;
  1035. rc->first_time_t = rc->hub.first_time_t;
  1036. rc->last_time_t = rc->hub.last_time_t;
  1037. if(rc->hub.deleted)
  1038. rc->flags |= RRD_FLAG_DELETED;
  1039. else {
  1040. if (rc->last_time_t == 0)
  1041. rrd_flag_set_collected(rc);
  1042. else
  1043. rrd_flag_set_archived(rc);
  1044. }
  1045. }
  1046. else {
  1047. // we are adding this context now for the first time
  1048. rc->version = now_realtime_sec();
  1049. rc->flags = rc->flags & RRD_FLAGS_ALLOWED_EXTERNALLY_ON_NEW_OBJECTS;
  1050. }
  1051. rrdinstances_create(rc);
  1052. netdata_mutex_init(&rc->mutex);
  1053. // signal the react callback to do the job
  1054. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_NEW_OBJECT);
  1055. }
  1056. static void rrdcontext_delete_callback(const char *id, void *value, void *data) {
  1057. (void)id;
  1058. RRDHOST *host = (RRDHOST *)data;
  1059. (void)host;
  1060. RRDCONTEXT *rc = (RRDCONTEXT *)value;
  1061. rrdinstances_destroy(rc);
  1062. netdata_mutex_destroy(&rc->mutex);
  1063. rrdcontext_freez(rc);
  1064. }
  1065. static STRING *string_2way_merge(RRDCONTEXT *rc __maybe_unused, STRING *a, STRING *b) {
  1066. static STRING *X = NULL;
  1067. if(unlikely(!X)) {
  1068. // initialization
  1069. X = string_strdupz("[x]");
  1070. }
  1071. if(unlikely(a == b)) return string_dup(a);
  1072. if(unlikely(a == X)) return string_dup(a);
  1073. if(unlikely(b == X)) return string_dup(b);
  1074. if(unlikely(!a)) return string_dup(X);
  1075. if(unlikely(!b)) return string_dup(X);
  1076. size_t alen = string_length(a);
  1077. size_t blen = string_length(b);
  1078. size_t length = alen + blen + string_length(X) + 1;
  1079. char buf1[length + 1], buf2[length + 1], *dst1;
  1080. const char *s1, *s2;
  1081. s1 = string2str(a);
  1082. s2 = string2str(b);
  1083. dst1 = buf1;
  1084. for( ; *s1 && *s2 && *s1 == *s2 ;s1++, s2++)
  1085. *dst1++ = *s1;
  1086. *dst1 = '\0';
  1087. if(*s1 != '\0' || *s2 != '\0') {
  1088. *dst1++ = '[';
  1089. *dst1++ = 'x';
  1090. *dst1++ = ']';
  1091. s1 = &(string2str(a))[alen - 1];
  1092. s2 = &(string2str(b))[blen - 1];
  1093. char *dst2 = &buf2[length];
  1094. *dst2 = '\0';
  1095. for (; *s1 && *s2 && *s1 == *s2; s1--, s2--)
  1096. *(--dst2) = *s1;
  1097. strcpy(dst1, dst2);
  1098. }
  1099. return string_strdupz(buf1);
  1100. }
  1101. static void rrdcontext_conflict_callback(const char *id, void *oldv, void *newv, void *data) {
  1102. (void)id;
  1103. RRDHOST *host = (RRDHOST *)data;
  1104. (void)host;
  1105. RRDCONTEXT *rc = (RRDCONTEXT *)oldv;
  1106. RRDCONTEXT *rc_new = (RRDCONTEXT *)newv;
  1107. rrdcontext_lock(rc);
  1108. if(rc->title != rc_new->title) {
  1109. STRING *old_title = rc->title;
  1110. rc->title = string_2way_merge(rc, rc->title, rc_new->title);
  1111. string_freez(old_title);
  1112. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_TITLE);
  1113. }
  1114. if(rc->units != rc_new->units) {
  1115. STRING *old_units = rc->units;
  1116. rc->units = string_dup(rc_new->units);
  1117. string_freez(old_units);
  1118. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_UNITS);
  1119. }
  1120. if(rc->family != rc_new->family) {
  1121. STRING *old_family = rc->family;
  1122. rc->family = string_2way_merge(rc, rc->family, rc_new->family);
  1123. string_freez(old_family);
  1124. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_FAMILY);
  1125. }
  1126. if(rc->chart_type != rc_new->chart_type) {
  1127. rc->chart_type = rc_new->chart_type;
  1128. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_CHART_TYPE);
  1129. }
  1130. if(rc->priority != rc_new->priority) {
  1131. rc->priority = rc_new->priority;
  1132. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY);
  1133. }
  1134. rc->flags |= (rc_new->flags & RRD_FLAGS_ALLOWED_EXTERNALLY_ON_NEW_OBJECTS);
  1135. if(rrd_flag_is_collected(rc) && rrd_flag_is_archived(rc))
  1136. rrd_flag_set_collected(rc);
  1137. if(rc->flags & RRD_FLAG_UPDATED)
  1138. rc->flags |= RRD_FLAG_UPDATE_REASON_UPDATED_OBJECT;
  1139. rrdcontext_unlock(rc);
  1140. // free the resources of the new one
  1141. rrdcontext_freez(rc_new);
  1142. // the react callback will continue from here
  1143. }
  1144. static void rrdcontext_react_callback(const char *id __maybe_unused, void *value, void *data __maybe_unused) {
  1145. RRDCONTEXT *rc = (RRDCONTEXT *)value;
  1146. rrdcontext_trigger_updates(rc, false, RRD_FLAG_NONE);
  1147. }
  1148. void rrdhost_create_rrdcontexts(RRDHOST *host) {
  1149. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1150. return;
  1151. if(unlikely(!host)) return;
  1152. if(likely(host->rrdctx)) return;
  1153. host->rrdctx = (RRDCONTEXTS *)dictionary_create(DICTIONARY_FLAG_DONT_OVERWRITE_VALUE);
  1154. dictionary_register_insert_callback((DICTIONARY *)host->rrdctx, rrdcontext_insert_callback, (void *)host);
  1155. dictionary_register_delete_callback((DICTIONARY *)host->rrdctx, rrdcontext_delete_callback, (void *)host);
  1156. dictionary_register_conflict_callback((DICTIONARY *)host->rrdctx, rrdcontext_conflict_callback, (void *)host);
  1157. dictionary_register_react_callback((DICTIONARY *)host->rrdctx, rrdcontext_react_callback, (void *)host);
  1158. host->rrdctx_queue = (RRDCONTEXTS *)dictionary_create(DICTIONARY_FLAG_DONT_OVERWRITE_VALUE | DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE);
  1159. }
  1160. void rrdhost_destroy_rrdcontexts(RRDHOST *host) {
  1161. if(unlikely(!host)) return;
  1162. if(unlikely(!host->rrdctx)) return;
  1163. if(host->rrdctx_queue) {
  1164. dictionary_destroy((DICTIONARY *)host->rrdctx_queue);
  1165. host->rrdctx_queue = NULL;
  1166. }
  1167. dictionary_destroy((DICTIONARY *)host->rrdctx);
  1168. host->rrdctx = NULL;
  1169. }
  1170. static inline bool rrdcontext_should_be_deleted(RRDCONTEXT *rc) {
  1171. if(likely(!(rc->flags & RRD_FLAG_DELETED)))
  1172. return false;
  1173. if(likely(!(rc->flags & RRD_FLAG_LIVE_RETENTION)))
  1174. return false;
  1175. if(unlikely(rc->flags & RRD_FLAGS_PREVENTING_DELETIONS))
  1176. return false;
  1177. if(unlikely(dictionary_stats_referenced_items(rc->rrdinstances) != 0))
  1178. return false;
  1179. if(unlikely(dictionary_stats_entries(rc->rrdinstances) != 0))
  1180. return false;
  1181. if(unlikely(rc->first_time_t || rc->last_time_t))
  1182. return false;
  1183. return true;
  1184. }
  1185. static void rrdcontext_trigger_updates(RRDCONTEXT *rc, bool force, RRD_FLAGS reason) {
  1186. if(unlikely(!force && !(rc->flags & RRD_FLAG_UPDATED))) return;
  1187. rrdcontext_lock(rc);
  1188. if(reason)
  1189. rrd_flag_set_updated(rc, reason);
  1190. size_t min_priority = LONG_MAX;
  1191. RRD_FLAGS combined_instances_flags = RRD_FLAG_NONE;
  1192. time_t min_first_time_t = LONG_MAX, max_last_time_t = 0;
  1193. size_t instances_active = 0, instances_deleted = 0;
  1194. {
  1195. RRDINSTANCE *ri;
  1196. dfe_start_read(rc->rrdinstances, ri) {
  1197. // find the combined flags of all the metrics
  1198. combined_instances_flags |= ri->flags & RRD_FLAGS_PROPAGATED_UPSTREAM;
  1199. if (unlikely(rrdinstance_should_be_deleted(ri))) {
  1200. instances_deleted++;
  1201. continue;
  1202. }
  1203. internal_error(rc->units != ri->units,
  1204. "RRDCONTEXT: '%s' rrdinstance '%s' has different units, context '%s', instance '%s'",
  1205. string2str(rc->id), string2str(ri->id),
  1206. string2str(rc->units), string2str(ri->units));
  1207. instances_active++;
  1208. if (ri->priority >= RRDCONTEXT_MINIMUM_ALLOWED_PRIORITY && ri->priority < min_priority)
  1209. min_priority = ri->priority;
  1210. if (ri->first_time_t && ri->first_time_t < min_first_time_t)
  1211. min_first_time_t = ri->first_time_t;
  1212. if (ri->last_time_t && ri->last_time_t > max_last_time_t)
  1213. max_last_time_t = ri->last_time_t;
  1214. }
  1215. dfe_done(ri);
  1216. }
  1217. rc->flags &= ~RRD_FLAG_DELETED;
  1218. if(unlikely(!instances_active && instances_deleted)) {
  1219. // we had some instances, but they are gone now...
  1220. rc->flags |= RRD_FLAG_DELETED;
  1221. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  1222. }
  1223. else if(instances_active) {
  1224. // we have some active instances...
  1225. if (unlikely(min_first_time_t == LONG_MAX))
  1226. min_first_time_t = 0;
  1227. if (unlikely(min_first_time_t == 0 && max_last_time_t == 0)) {
  1228. rc->first_time_t = 0;
  1229. rc->last_time_t = 0;
  1230. if(unlikely(combined_instances_flags & RRD_FLAG_LIVE_RETENTION)) {
  1231. rc->flags |= RRD_FLAG_DELETED;
  1232. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
  1233. }
  1234. }
  1235. else {
  1236. if (unlikely(rc->first_time_t != min_first_time_t)) {
  1237. rc->first_time_t = min_first_time_t;
  1238. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
  1239. }
  1240. if (rc->last_time_t != max_last_time_t) {
  1241. rc->last_time_t = max_last_time_t;
  1242. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
  1243. }
  1244. }
  1245. if(combined_instances_flags & RRD_FLAG_LIVE_RETENTION)
  1246. rc->flags |= RRD_FLAG_LIVE_RETENTION;
  1247. else
  1248. rc->flags &= ~RRD_FLAG_LIVE_RETENTION;
  1249. if(likely(combined_instances_flags & RRD_FLAG_COLLECTED))
  1250. rrd_flag_set_collected(rc);
  1251. else
  1252. rrd_flag_set_archived(rc);
  1253. if (min_priority != LONG_MAX && rc->priority != min_priority) {
  1254. rc->priority = min_priority;
  1255. rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_PRIORITY);
  1256. }
  1257. }
  1258. else {
  1259. // no deleted instances, no active instances
  1260. // just hanging there...
  1261. rrd_flag_set_archived(rc);
  1262. }
  1263. if(unlikely(rc->flags & RRD_FLAG_UPDATED)) {
  1264. log_transition(NULL, NULL, rc->id, rc->flags, "RRDCONTEXT");
  1265. if(check_if_cloud_version_changed_unsafe(rc, false)) {
  1266. rc->version = rrdcontext_get_next_version(rc);
  1267. rc->queue.queued_ut = now_realtime_usec();
  1268. rc->queue.queued_flags |= rc->flags;
  1269. if(!(rc->flags & RRD_FLAG_QUEUED)) {
  1270. rc->flags |= RRD_FLAG_QUEUED;
  1271. dictionary_set((DICTIONARY *)rc->rrdhost->rrdctx_queue, string2str(rc->id), rc, sizeof(*rc));
  1272. }
  1273. }
  1274. rrd_flag_unset_updated(rc);
  1275. }
  1276. rrdcontext_unlock(rc);
  1277. }
  1278. // ----------------------------------------------------------------------------
  1279. // public API
  1280. void rrdcontext_updated_rrddim(RRDDIM *rd) {
  1281. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1282. return;
  1283. rrdmetric_from_rrddim(rd);
  1284. }
  1285. void rrdcontext_removed_rrddim(RRDDIM *rd) {
  1286. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1287. return;
  1288. rrdmetric_rrddim_is_freed(rd);
  1289. }
  1290. void rrdcontext_updated_rrddim_algorithm(RRDDIM *rd) {
  1291. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1292. return;
  1293. rrdmetric_updated_rrddim_flags(rd);
  1294. }
  1295. void rrdcontext_updated_rrddim_multiplier(RRDDIM *rd) {
  1296. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1297. return;
  1298. rrdmetric_updated_rrddim_flags(rd);
  1299. }
  1300. void rrdcontext_updated_rrddim_divisor(RRDDIM *rd) {
  1301. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1302. return;
  1303. rrdmetric_updated_rrddim_flags(rd);
  1304. }
  1305. void rrdcontext_updated_rrddim_flags(RRDDIM *rd) {
  1306. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1307. return;
  1308. rrdmetric_updated_rrddim_flags(rd);
  1309. }
  1310. void rrdcontext_collected_rrddim(RRDDIM *rd) {
  1311. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1312. return;
  1313. rrdmetric_collected_rrddim(rd);
  1314. }
  1315. void rrdcontext_updated_rrdset(RRDSET *st) {
  1316. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1317. return;
  1318. rrdinstance_from_rrdset(st);
  1319. }
  1320. void rrdcontext_removed_rrdset(RRDSET *st) {
  1321. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1322. return;
  1323. rrdinstance_rrdset_is_freed(st);
  1324. }
  1325. void rrdcontext_updated_rrdset_name(RRDSET *st) {
  1326. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1327. return;
  1328. rrdinstance_updated_rrdset_name(st);
  1329. }
  1330. void rrdcontext_updated_rrdset_flags(RRDSET *st) {
  1331. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1332. return;
  1333. rrdinstance_updated_rrdset_flags(st);
  1334. }
  1335. void rrdcontext_collected_rrdset(RRDSET *st) {
  1336. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1337. return;
  1338. rrdinstance_collected_rrdset(st);
  1339. }
  1340. void rrdcontext_host_child_connected(RRDHOST *host) {
  1341. (void)host;
  1342. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1343. return;
  1344. // no need to do anything here
  1345. ;
  1346. }
  1347. void rrdcontext_host_child_disconnected(RRDHOST *host) {
  1348. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1349. return;
  1350. rrdcontext_recalculate_host_retention(host, RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD, -1);
  1351. }
  1352. // ----------------------------------------------------------------------------
  1353. // ACLK interface
  1354. static bool rrdhost_check_our_claim_id(const char *claim_id) {
  1355. if(!localhost->aclk_state.claimed_id) return false;
  1356. return (strcasecmp(claim_id, localhost->aclk_state.claimed_id) == 0) ? true : false;
  1357. }
  1358. static RRDHOST *rrdhost_find_by_node_id(const char *node_id) {
  1359. uuid_t uuid;
  1360. uuid_parse(node_id, uuid);
  1361. RRDHOST *host = NULL;
  1362. rrd_rdlock();
  1363. rrdhost_foreach_read(host) {
  1364. if(!host->node_id) continue;
  1365. if(uuid_compare(uuid, *host->node_id) == 0)
  1366. break;
  1367. }
  1368. rrd_unlock();
  1369. return host;
  1370. }
  1371. void rrdcontext_hub_checkpoint_command(void *ptr) {
  1372. struct ctxs_checkpoint *cmd = ptr;
  1373. if(!rrdhost_check_our_claim_id(cmd->claim_id)) {
  1374. 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.",
  1375. cmd->claim_id, cmd->node_id,
  1376. localhost->aclk_state.claimed_id?localhost->aclk_state.claimed_id:"NOT SET",
  1377. cmd->claim_id);
  1378. return;
  1379. }
  1380. RRDHOST *host = rrdhost_find_by_node_id(cmd->node_id);
  1381. if(!host) {
  1382. error("RRDCONTEXT: received checkpoint command for claim id '%s', node id '%s', but there is no node with such node id here. Ignoring command.",
  1383. cmd->claim_id, cmd->node_id);
  1384. return;
  1385. }
  1386. if(rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS)) {
  1387. info("RRDCONTEXT: received checkpoint command for claim id '%s', node id '%s', while node '%s' has an active context streaming.",
  1388. cmd->claim_id, cmd->node_id, host->hostname);
  1389. // disable it temporarily, so that our worker will not attempt to send messages in parallel
  1390. rrdhost_flag_clear(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS);
  1391. }
  1392. uint64_t our_version_hash = rrdcontext_version_hash(host);
  1393. if(cmd->version_hash != our_version_hash) {
  1394. error("RRDCONTEXT: received version hash %lu for host '%s', does not match our version hash %lu. Sending snapshot of all contexts.",
  1395. cmd->version_hash, host->hostname, our_version_hash);
  1396. #ifdef ENABLE_ACLK
  1397. // prepare the snapshot
  1398. char uuid[UUID_STR_LEN];
  1399. uuid_unparse_lower(*host->node_id, uuid);
  1400. contexts_snapshot_t bundle = contexts_snapshot_new(cmd->claim_id, uuid, our_version_hash);
  1401. // do a deep scan on every metric of the host to make sure all our data are updated
  1402. rrdcontext_recalculate_host_retention(host, RRD_FLAG_NONE, -1);
  1403. // calculate version hash and pack all the messages together in one go
  1404. our_version_hash = rrdcontext_version_hash_with_callback(host, rrdcontext_message_send_unsafe, true, bundle);
  1405. // update the version
  1406. contexts_snapshot_set_version(bundle, our_version_hash);
  1407. // send it
  1408. aclk_send_contexts_snapshot(bundle);
  1409. #endif
  1410. }
  1411. internal_error(true, "RRDCONTEXT: host '%s' enabling streaming of contexts", host->hostname);
  1412. rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS);
  1413. }
  1414. void rrdcontext_hub_stop_streaming_command(void *ptr) {
  1415. struct stop_streaming_ctxs *cmd = ptr;
  1416. if(!rrdhost_check_our_claim_id(cmd->claim_id)) {
  1417. 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.",
  1418. cmd->claim_id, cmd->node_id,
  1419. localhost->aclk_state.claimed_id?localhost->aclk_state.claimed_id:"NOT SET",
  1420. cmd->claim_id);
  1421. return;
  1422. }
  1423. RRDHOST *host = rrdhost_find_by_node_id(cmd->node_id);
  1424. if(!host) {
  1425. 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.",
  1426. cmd->claim_id, cmd->node_id);
  1427. return;
  1428. }
  1429. if(!rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS)) {
  1430. error("RRDCONTEXT: received stop streaming command for claim id '%s', node id '%s', but node '%s' does not have active context streaming. Ignoring command.",
  1431. cmd->claim_id, cmd->node_id, host->hostname);
  1432. return;
  1433. }
  1434. internal_error(true, "RRDCONTEXT: host '%s' disabling streaming of contexts", host->hostname);
  1435. rrdhost_flag_clear(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS);
  1436. }
  1437. // ----------------------------------------------------------------------------
  1438. // web API
  1439. struct rrdcontext_to_json {
  1440. BUFFER *wb;
  1441. RRDCONTEXT_TO_JSON_OPTIONS options;
  1442. time_t after;
  1443. time_t before;
  1444. SIMPLE_PATTERN *chart_label_key;
  1445. SIMPLE_PATTERN *chart_labels_filter;
  1446. SIMPLE_PATTERN *chart_dimensions;
  1447. size_t written;
  1448. time_t now;
  1449. time_t combined_first_time_t;
  1450. time_t combined_last_time_t;
  1451. RRD_FLAGS combined_flags;
  1452. };
  1453. static inline int rrdmetric_to_json_callback(const char *id, void *value, void *data) {
  1454. struct rrdcontext_to_json * t = data;
  1455. RRDMETRIC *rm = value;
  1456. BUFFER *wb = t->wb;
  1457. RRDCONTEXT_TO_JSON_OPTIONS options = t->options;
  1458. time_t after = t->after;
  1459. time_t before = t->before;
  1460. if((rm->flags & RRD_FLAG_DELETED) && !(options & RRDCONTEXT_OPTION_SHOW_DELETED))
  1461. return 0;
  1462. if(after && (!rm->last_time_t || after > rm->last_time_t))
  1463. return 0;
  1464. if(before && (!rm->first_time_t || before < rm->first_time_t))
  1465. return 0;
  1466. if(t->chart_dimensions
  1467. && !simple_pattern_matches(t->chart_dimensions, string2str(rm->id))
  1468. && !simple_pattern_matches(t->chart_dimensions, string2str(rm->name)))
  1469. return 0;
  1470. if(t->written) {
  1471. buffer_strcat(wb, ",\n");
  1472. t->combined_first_time_t = MIN(t->combined_first_time_t, rm->first_time_t);
  1473. t->combined_last_time_t = MAX(t->combined_last_time_t, rm->last_time_t);
  1474. t->combined_flags |= rm->flags;
  1475. }
  1476. else {
  1477. buffer_strcat(wb, "\n");
  1478. t->combined_first_time_t = rm->first_time_t;
  1479. t->combined_last_time_t = rm->last_time_t;
  1480. t->combined_flags = rm->flags;
  1481. }
  1482. buffer_sprintf(wb, "\t\t\t\t\t\t\"%s\": {", id);
  1483. if(options & RRDCONTEXT_OPTION_SHOW_UUIDS) {
  1484. char uuid[UUID_STR_LEN];
  1485. uuid_unparse(rm->uuid, uuid);
  1486. buffer_sprintf(wb, "\n\t\t\t\t\t\t\t\"uuid\":\"%s\",", uuid);
  1487. }
  1488. buffer_sprintf(wb,
  1489. "\n\t\t\t\t\t\t\t\"name\":\"%s\""
  1490. ",\n\t\t\t\t\t\t\t\"first_time_t\":%ld"
  1491. ",\n\t\t\t\t\t\t\t\"last_time_t\":%ld"
  1492. ",\n\t\t\t\t\t\t\t\"collected\":%s"
  1493. , string2str(rm->name)
  1494. , rm->first_time_t
  1495. , rrd_flag_is_collected(rm) ? t->now : rm->last_time_t
  1496. , rm->flags & RRD_FLAG_COLLECTED ? "true" : "false"
  1497. );
  1498. if(options & RRDCONTEXT_OPTION_SHOW_DELETED) {
  1499. buffer_sprintf(wb,
  1500. ",\n\t\t\t\t\t\t\t\"deleted\":%s"
  1501. , rm->flags & RRD_FLAG_DELETED ? "true" : "false"
  1502. );
  1503. }
  1504. if(options & RRDCONTEXT_OPTION_SHOW_FLAGS) {
  1505. buffer_strcat(wb, ",\n\t\t\t\t\t\t\t\"flags\":\"");
  1506. rrd_flags_to_buffer(rm->flags, wb);
  1507. buffer_strcat(wb, "\"");
  1508. }
  1509. buffer_strcat(wb, "\n\t\t\t\t\t\t}");
  1510. t->written++;
  1511. return 1;
  1512. }
  1513. static inline int rrdinstance_to_json_callback(const char *id, void *value, void *data) {
  1514. struct rrdcontext_to_json * t = data;
  1515. RRDINSTANCE *ri = value;
  1516. BUFFER *wb = t->wb;
  1517. RRDCONTEXT_TO_JSON_OPTIONS options = t->options;
  1518. time_t after = t->after;
  1519. time_t before = t->before;
  1520. if((ri->flags & RRD_FLAG_DELETED) && !(options & RRDCONTEXT_OPTION_SHOW_DELETED))
  1521. return 0;
  1522. if(after && (!ri->last_time_t || after > ri->last_time_t))
  1523. return 0;
  1524. if(before && (!ri->first_time_t || before < ri->first_time_t))
  1525. return 0;
  1526. if(t->chart_label_key && !rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, t->chart_label_key, '\0'))
  1527. return 0;
  1528. if(t->chart_labels_filter && !rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, t->chart_labels_filter, ':'))
  1529. return 0;
  1530. time_t first_time_t = ri->first_time_t;
  1531. time_t last_time_t = ri->last_time_t;
  1532. RRD_FLAGS flags = ri->flags;
  1533. BUFFER *wb_metrics = NULL;
  1534. if(options & RRDCONTEXT_OPTION_SHOW_METRICS || t->chart_dimensions) {
  1535. wb_metrics = buffer_create(4096);
  1536. struct rrdcontext_to_json t_metrics = {
  1537. .wb = wb_metrics,
  1538. .options = options,
  1539. .chart_label_key = t->chart_label_key,
  1540. .chart_labels_filter = t->chart_labels_filter,
  1541. .chart_dimensions = t->chart_dimensions,
  1542. .after = after,
  1543. .before = before,
  1544. .written = 0,
  1545. .now = t->now,
  1546. };
  1547. dictionary_walkthrough_read(ri->rrdmetrics, rrdmetric_to_json_callback, &t_metrics);
  1548. if(!t_metrics.written) {
  1549. buffer_free(wb_metrics);
  1550. return 0;
  1551. }
  1552. first_time_t = t_metrics.combined_first_time_t;
  1553. last_time_t = t_metrics.combined_last_time_t;
  1554. flags = t_metrics.combined_flags;
  1555. }
  1556. if(t->written) {
  1557. buffer_strcat(wb, ",\n");
  1558. t->combined_first_time_t = MIN(t->combined_first_time_t, first_time_t);
  1559. t->combined_last_time_t = MAX(t->combined_last_time_t, last_time_t);
  1560. t->combined_flags |= flags;
  1561. }
  1562. else {
  1563. buffer_strcat(wb, "\n");
  1564. t->combined_first_time_t = first_time_t;
  1565. t->combined_last_time_t = last_time_t;
  1566. t->combined_flags = flags;
  1567. }
  1568. buffer_sprintf(wb, "\t\t\t\t\"%s\": {", id);
  1569. if(options & RRDCONTEXT_OPTION_SHOW_UUIDS) {
  1570. char uuid[UUID_STR_LEN];
  1571. uuid_unparse(ri->uuid, uuid);
  1572. buffer_sprintf(wb,"\n\t\t\t\t\t\"uuid\":\"%s\",", uuid);
  1573. }
  1574. buffer_sprintf(wb,
  1575. "\n\t\t\t\t\t\"name\":\"%s\""
  1576. ",\n\t\t\t\t\t\"context\":\"%s\""
  1577. ",\n\t\t\t\t\t\"title\":\"%s\""
  1578. ",\n\t\t\t\t\t\"units\":\"%s\""
  1579. ",\n\t\t\t\t\t\"family\":\"%s\""
  1580. ",\n\t\t\t\t\t\"chart_type\":\"%s\""
  1581. ",\n\t\t\t\t\t\"priority\":%u"
  1582. ",\n\t\t\t\t\t\"update_every\":%d"
  1583. ",\n\t\t\t\t\t\"first_time_t\":%ld"
  1584. ",\n\t\t\t\t\t\"last_time_t\":%ld"
  1585. ",\n\t\t\t\t\t\"collected\":%s"
  1586. , string2str(ri->name)
  1587. , string2str(ri->rc->id)
  1588. , string2str(ri->title)
  1589. , string2str(ri->units)
  1590. , string2str(ri->family)
  1591. , rrdset_type_name(ri->chart_type)
  1592. , ri->priority
  1593. , ri->update_every
  1594. , first_time_t
  1595. , (flags & RRD_FLAG_COLLECTED) ? t->now : last_time_t
  1596. , (flags & RRD_FLAG_COLLECTED) ? "true" : "false"
  1597. );
  1598. if(options & RRDCONTEXT_OPTION_SHOW_DELETED) {
  1599. buffer_sprintf(wb,
  1600. ",\n\t\t\t\t\t\"deleted\":%s"
  1601. , (ri->flags & RRD_FLAG_DELETED) ? "true" : "false"
  1602. );
  1603. }
  1604. if(options & RRDCONTEXT_OPTION_SHOW_FLAGS) {
  1605. buffer_strcat(wb, ",\n\t\t\t\t\t\"flags\":\"");
  1606. rrd_flags_to_buffer(ri->flags, wb);
  1607. buffer_strcat(wb, "\"");
  1608. }
  1609. if(options & RRDCONTEXT_OPTION_SHOW_LABELS && ri->rrdlabels && dictionary_stats_entries(ri->rrdlabels)) {
  1610. buffer_sprintf(wb, ",\n\t\t\t\t\t\"labels\": {\n");
  1611. rrdlabels_to_buffer(ri->rrdlabels, wb, "\t\t\t\t\t\t", ":", "\"", ",\n", NULL, NULL, NULL, NULL);
  1612. buffer_strcat(wb, "\n\t\t\t\t\t}");
  1613. }
  1614. if(wb_metrics) {
  1615. buffer_sprintf(wb, ",\n\t\t\t\t\t\"dimensions\": {");
  1616. buffer_fast_strcat(wb, buffer_tostring(wb_metrics), buffer_strlen(wb_metrics));
  1617. buffer_strcat(wb, "\n\t\t\t\t\t}");
  1618. buffer_free(wb_metrics);
  1619. }
  1620. buffer_strcat(wb, "\n\t\t\t\t}");
  1621. t->written++;
  1622. return 1;
  1623. }
  1624. static inline int rrdcontext_to_json_callback(const char *id, void *value, void *data) {
  1625. struct rrdcontext_to_json * t = data;
  1626. RRDCONTEXT *rc = value;
  1627. BUFFER *wb = t->wb;
  1628. RRDCONTEXT_TO_JSON_OPTIONS options = t->options;
  1629. time_t after = t->after;
  1630. time_t before = t->before;
  1631. if((rc->flags & RRD_FLAG_DELETED) && !(options & RRDCONTEXT_OPTION_SHOW_DELETED))
  1632. return 0;
  1633. if(options & RRDCONTEXT_OPTION_DEEPSCAN)
  1634. rrdcontext_recalculate_context_retention(rc, RRD_FLAG_NONE, -1);
  1635. if(after && (!rc->last_time_t || after > rc->last_time_t))
  1636. return 0;
  1637. if(before && (!rc->first_time_t || before < rc->first_time_t))
  1638. return 0;
  1639. time_t first_time_t = rc->first_time_t;
  1640. time_t last_time_t = rc->last_time_t;
  1641. RRD_FLAGS flags = rc->flags;
  1642. BUFFER *wb_instances = NULL;
  1643. if((options & (RRDCONTEXT_OPTION_SHOW_LABELS|RRDCONTEXT_OPTION_SHOW_INSTANCES|RRDCONTEXT_OPTION_SHOW_METRICS))
  1644. || t->chart_label_key
  1645. || t->chart_labels_filter
  1646. || t->chart_dimensions) {
  1647. wb_instances = buffer_create(4096);
  1648. struct rrdcontext_to_json t_instances = {
  1649. .wb = wb_instances,
  1650. .options = options,
  1651. .chart_label_key = t->chart_label_key,
  1652. .chart_labels_filter = t->chart_labels_filter,
  1653. .chart_dimensions = t->chart_dimensions,
  1654. .after = after,
  1655. .before = before,
  1656. .written = 0,
  1657. .now = t->now,
  1658. };
  1659. dictionary_walkthrough_read(rc->rrdinstances, rrdinstance_to_json_callback, &t_instances);
  1660. if(!t_instances.written) {
  1661. buffer_free(wb_instances);
  1662. return 0;
  1663. }
  1664. first_time_t = t_instances.combined_first_time_t;
  1665. last_time_t = t_instances.combined_last_time_t;
  1666. flags = t_instances.combined_flags;
  1667. }
  1668. if(t->written)
  1669. buffer_strcat(wb, ",\n");
  1670. else
  1671. buffer_strcat(wb, "\n");
  1672. if(options & RRDCONTEXT_OPTION_SKIP_ID)
  1673. buffer_sprintf(wb, "\t\t\{");
  1674. else
  1675. buffer_sprintf(wb, "\t\t\"%s\": {", id);
  1676. rrdcontext_lock(rc);
  1677. buffer_sprintf(wb,
  1678. "\n\t\t\t\"title\":\"%s\""
  1679. ",\n\t\t\t\"units\":\"%s\""
  1680. ",\n\t\t\t\"family\":\"%s\""
  1681. ",\n\t\t\t\"chart_type\":\"%s\""
  1682. ",\n\t\t\t\"priority\":%u"
  1683. ",\n\t\t\t\"first_time_t\":%ld"
  1684. ",\n\t\t\t\"last_time_t\":%ld"
  1685. ",\n\t\t\t\"collected\":%s"
  1686. , string2str(rc->title)
  1687. , string2str(rc->units)
  1688. , string2str(rc->family)
  1689. , rrdset_type_name(rc->chart_type)
  1690. , rc->priority
  1691. , first_time_t
  1692. , (flags & RRD_FLAG_COLLECTED) ? t->now : last_time_t
  1693. , (flags & RRD_FLAG_COLLECTED) ? "true" : "false"
  1694. );
  1695. if(options & RRDCONTEXT_OPTION_SHOW_DELETED) {
  1696. buffer_sprintf(wb,
  1697. ",\n\t\t\t\"deleted\":%s"
  1698. , (rc->flags & RRD_FLAG_DELETED) ? "true" : "false"
  1699. );
  1700. }
  1701. if(options & RRDCONTEXT_OPTION_SHOW_FLAGS) {
  1702. buffer_strcat(wb, ",\n\t\t\t\"flags\":\"");
  1703. rrd_flags_to_buffer(rc->flags, wb);
  1704. buffer_strcat(wb, "\"");
  1705. }
  1706. if(options & RRDCONTEXT_OPTION_SHOW_QUEUED) {
  1707. buffer_strcat(wb, ",\n\t\t\t\"queued_reasons\":\"");
  1708. rrd_reasons_to_buffer(rc->queue.queued_flags, wb);
  1709. buffer_strcat(wb, "\"");
  1710. buffer_sprintf(wb,
  1711. ",\n\t\t\t\"last_queued\":%llu"
  1712. ",\n\t\t\t\"scheduled_dispatch\":%llu"
  1713. ",\n\t\t\t\"last_dequeued\":%llu"
  1714. ",\n\t\t\t\"hub_version\":%lu"
  1715. ",\n\t\t\t\"version\":%lu"
  1716. , rc->queue.queued_ut / USEC_PER_SEC
  1717. , rc->queue.scheduled_dispatch_ut / USEC_PER_SEC
  1718. , rc->queue.dequeued_ut / USEC_PER_SEC
  1719. , rc->hub.version
  1720. , rc->version
  1721. );
  1722. }
  1723. rrdcontext_unlock(rc);
  1724. if(wb_instances) {
  1725. buffer_sprintf(wb, ",\n\t\t\t\"charts\": {");
  1726. buffer_fast_strcat(wb, buffer_tostring(wb_instances), buffer_strlen(wb_instances));
  1727. buffer_strcat(wb, "\n\t\t\t}");
  1728. buffer_free(wb_instances);
  1729. }
  1730. buffer_strcat(wb, "\n\t\t}");
  1731. t->written++;
  1732. return 1;
  1733. }
  1734. 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) {
  1735. RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_get_and_acquire_item((DICTIONARY *)host->rrdctx, context);
  1736. if(!rca) return HTTP_RESP_NOT_FOUND;
  1737. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  1738. if(after != 0 && before != 0) {
  1739. long long after_wanted = after;
  1740. long long before_wanted = before;
  1741. rrdr_relative_window_to_absolute(&after_wanted, &before_wanted);
  1742. after = after_wanted;
  1743. before = before_wanted;
  1744. }
  1745. struct rrdcontext_to_json t_contexts = {
  1746. .wb = wb,
  1747. .options = options|RRDCONTEXT_OPTION_SKIP_ID,
  1748. .chart_label_key = chart_label_key,
  1749. .chart_labels_filter = chart_labels_filter,
  1750. .chart_dimensions = chart_dimensions,
  1751. .after = after,
  1752. .before = before,
  1753. .written = 0,
  1754. .now = now_realtime_sec(),
  1755. };
  1756. rrdcontext_to_json_callback(context, rc, &t_contexts);
  1757. rrdcontext_release(rca);
  1758. if(!t_contexts.written)
  1759. return HTTP_RESP_NOT_FOUND;
  1760. return HTTP_RESP_OK;
  1761. }
  1762. 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) {
  1763. char node_uuid[UUID_STR_LEN] = "";
  1764. if(host->node_id)
  1765. uuid_unparse(*host->node_id, node_uuid);
  1766. if(after != 0 && before != 0) {
  1767. long long after_wanted = after;
  1768. long long before_wanted = before;
  1769. rrdr_relative_window_to_absolute(&after_wanted, &before_wanted);
  1770. after = after_wanted;
  1771. before = before_wanted;
  1772. }
  1773. buffer_sprintf(wb, "{\n"
  1774. "\t\"hostname\": \"%s\""
  1775. ",\n\t\"machine_guid\": \"%s\""
  1776. ",\n\t\"node_id\": \"%s\""
  1777. ",\n\t\"claim_id\": \"%s\""
  1778. , host->hostname
  1779. , host->machine_guid
  1780. , node_uuid
  1781. , host->aclk_state.claimed_id ? host->aclk_state.claimed_id : ""
  1782. );
  1783. if(options & RRDCONTEXT_OPTION_SHOW_LABELS) {
  1784. buffer_sprintf(wb, ",\n\t\"host_labels\": {\n");
  1785. rrdlabels_to_buffer(host->host_labels, wb, "\t\t", ":", "\"", ",\n", NULL, NULL, NULL, NULL);
  1786. buffer_strcat(wb, "\n\t}");
  1787. }
  1788. buffer_sprintf(wb, ",\n\t\"contexts\": {");
  1789. struct rrdcontext_to_json t_contexts = {
  1790. .wb = wb,
  1791. .options = options,
  1792. .chart_label_key = chart_label_key,
  1793. .chart_labels_filter = chart_labels_filter,
  1794. .chart_dimensions = chart_dimensions,
  1795. .after = after,
  1796. .before = before,
  1797. .written = 0,
  1798. .now = now_realtime_sec(),
  1799. };
  1800. dictionary_walkthrough_read((DICTIONARY *)host->rrdctx, rrdcontext_to_json_callback, &t_contexts);
  1801. // close contexts, close main
  1802. buffer_strcat(wb, "\n\t}\n}");
  1803. return HTTP_RESP_OK;
  1804. }
  1805. // ----------------------------------------------------------------------------
  1806. // load from SQL
  1807. static void rrdinstance_load_clabel(SQL_CLABEL_DATA *sld, void *data) {
  1808. RRDINSTANCE *ri = data;
  1809. rrdlabels_add(ri->rrdlabels, sld->label_key, sld->label_value, sld->label_source);
  1810. }
  1811. static void rrdinstance_load_dimension(SQL_DIMENSION_DATA *sd, void *data) {
  1812. RRDINSTANCE *ri = data;
  1813. RRDMETRIC trm = {
  1814. .id = string_strdupz(sd->id),
  1815. .name = string_strdupz(sd->name),
  1816. .flags = RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_LOAD_SQL,
  1817. };
  1818. uuid_copy(trm.uuid, sd->dim_id);
  1819. dictionary_set(ri->rrdmetrics, string2str(trm.id), &trm, sizeof(trm));
  1820. }
  1821. static void rrdinstance_load_chart_callback(SQL_CHART_DATA *sc, void *data) {
  1822. RRDHOST *host = data;
  1823. RRDCONTEXT tc = {
  1824. .id = string_strdupz(sc->context),
  1825. .title = string_strdupz(sc->title),
  1826. .units = string_strdupz(sc->units),
  1827. .family = string_strdupz(sc->family),
  1828. .priority = sc->priority,
  1829. .chart_type = sc->chart_type,
  1830. .flags = RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_LOAD_SQL,
  1831. .rrdhost = host,
  1832. };
  1833. RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_set_and_acquire_item((DICTIONARY *)host->rrdctx, string2str(tc.id), &tc, sizeof(tc));
  1834. RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
  1835. RRDINSTANCE tri = {
  1836. .id = string_strdupz(sc->id),
  1837. .name = string_strdupz(sc->name),
  1838. .title = string_strdupz(sc->title),
  1839. .units = string_strdupz(sc->units),
  1840. .family = string_strdupz(sc->family),
  1841. .chart_type = sc->chart_type,
  1842. .priority = sc->priority,
  1843. .update_every = sc->update_every,
  1844. .flags = RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_LOAD_SQL,
  1845. };
  1846. uuid_copy(tri.uuid, sc->chart_id);
  1847. RRDINSTANCE_ACQUIRED *ria = (RRDINSTANCE_ACQUIRED *)dictionary_set_and_acquire_item(rc->rrdinstances, sc->id, &tri, sizeof(tri));
  1848. RRDINSTANCE *ri = rrdinstance_acquired_value(ria);
  1849. ctx_get_dimension_list(&ri->uuid, rrdinstance_load_dimension, ri);
  1850. ctx_get_label_list(&ri->uuid, rrdinstance_load_clabel, ri);
  1851. rrdinstance_release(ria);
  1852. rrdcontext_release(rca);
  1853. }
  1854. static void rrdcontext_load_context_callback(VERSIONED_CONTEXT_DATA *ctx_data, void *data) {
  1855. RRDHOST *host = data;
  1856. (void)host;
  1857. RRDCONTEXT tmp = {
  1858. .id = string_strdupz(ctx_data->id),
  1859. // no need to set more data here
  1860. // we only need the hub data
  1861. .hub = *ctx_data,
  1862. };
  1863. dictionary_set((DICTIONARY *)host->rrdctx, string2str(tmp.id), &tmp, sizeof(tmp));
  1864. }
  1865. void rrdhost_load_rrdcontext_data(RRDHOST *host) {
  1866. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  1867. return;
  1868. if(host->rrdctx) return;
  1869. rrdhost_create_rrdcontexts(host);
  1870. ctx_get_context_list(&host->host_uuid, rrdcontext_load_context_callback, host);
  1871. ctx_get_chart_list(&host->host_uuid, rrdinstance_load_chart_callback, host);
  1872. }
  1873. // ----------------------------------------------------------------------------
  1874. // the worker thread
  1875. static inline usec_t rrdcontext_calculate_queued_dispatch_time_ut(RRDCONTEXT *rc, usec_t now_ut) {
  1876. if(likely(rc->queue.delay_calc_ut >= rc->queue.queued_ut))
  1877. return rc->queue.scheduled_dispatch_ut;
  1878. RRD_FLAGS flags = rc->queue.queued_flags;
  1879. usec_t delay = LONG_MAX;
  1880. int i;
  1881. struct rrdcontext_reason *reason;
  1882. for(i = 0, reason = &rrdcontext_reasons[i]; reason->name ; reason = &rrdcontext_reasons[++i]) {
  1883. if(unlikely(flags & reason->flag)) {
  1884. if(reason->delay_ut < delay)
  1885. delay = reason->delay_ut;
  1886. }
  1887. }
  1888. if(unlikely(delay == LONG_MAX)) {
  1889. internal_error(true, "RRDCONTEXT: '%s', cannot find minimum delay of flags %x", string2str(rc->id), (unsigned int)flags);
  1890. delay = 60 * USEC_PER_SEC;
  1891. }
  1892. rc->queue.delay_calc_ut = now_ut;
  1893. usec_t dispatch_ut = rc->queue.scheduled_dispatch_ut = rc->queue.queued_ut + delay;
  1894. return dispatch_ut;
  1895. }
  1896. #define WORKER_JOB_HOSTS 1
  1897. #define WORKER_JOB_CHECK 2
  1898. #define WORKER_JOB_SEND 3
  1899. #define WORKER_JOB_DEQUEUE 4
  1900. #define WORKER_JOB_RETENTION 5
  1901. #define WORKER_JOB_QUEUED 6
  1902. #define WORKER_JOB_CLEANUP 7
  1903. #define WORKER_JOB_CLEANUP_DELETE 8
  1904. static usec_t rrdcontext_next_db_rotation_ut = 0;
  1905. void rrdcontext_db_rotation(void) {
  1906. // called when the db rotates its database
  1907. rrdcontext_next_db_rotation_ut = now_realtime_usec() + FULL_RETENTION_SCAN_DELAY_AFTER_DB_ROTATION_SECS * USEC_PER_SEC;
  1908. }
  1909. static uint64_t rrdcontext_version_hash_with_callback(
  1910. RRDHOST *host,
  1911. void (*callback)(RRDCONTEXT *, bool, void *),
  1912. bool snapshot,
  1913. void *bundle) {
  1914. if(unlikely(!host || !host->rrdctx)) return 0;
  1915. RRDCONTEXT *rc;
  1916. uint64_t hash = 0;
  1917. // loop through all contexts of the host
  1918. dfe_start_read((DICTIONARY *)host->rrdctx, rc) {
  1919. rrdcontext_lock(rc);
  1920. // skip any deleted contexts
  1921. if(unlikely(rc->flags & RRD_FLAG_DELETED)) {
  1922. rrdcontext_unlock(rc);
  1923. continue;
  1924. }
  1925. if(unlikely(callback))
  1926. callback(rc, snapshot, bundle);
  1927. // we use rc->hub.* which has the latest
  1928. // metadata we have sent to the hub
  1929. // if a context is currently queued, rc->hub.* does NOT
  1930. // reflect the queued changes. rc->hub.* is updated with
  1931. // their metadata, after messages are dispatched to hub.
  1932. // when the context is being collected,
  1933. // rc->hub.last_time_t is already zero
  1934. hash += rc->hub.version + rc->hub.last_time_t - rc->hub.first_time_t;
  1935. rrdcontext_unlock(rc);
  1936. }
  1937. dfe_done(rc);
  1938. return hash;
  1939. }
  1940. static void rrdcontext_recalculate_context_retention(RRDCONTEXT *rc, RRD_FLAGS reason, int job_id) {
  1941. RRDINSTANCE *ri;
  1942. dfe_start_read(rc->rrdinstances, ri) {
  1943. RRDMETRIC *rm;
  1944. dfe_start_read(ri->rrdmetrics, rm) {
  1945. if(job_id >= 0)
  1946. worker_is_busy(job_id);
  1947. rrd_flag_set_updated(rm, reason);
  1948. rrdmetric_trigger_updates(rm, true, false);
  1949. }
  1950. dfe_done(rm);
  1951. rrdinstance_trigger_updates(ri, true, false);
  1952. }
  1953. dfe_done(ri);
  1954. rrdcontext_trigger_updates(rc, true, RRD_FLAG_NONE);
  1955. }
  1956. static void rrdcontext_recalculate_host_retention(RRDHOST *host, RRD_FLAGS reason, int job_id) {
  1957. if(unlikely(!host || !host->rrdctx)) return;
  1958. RRDCONTEXT *rc;
  1959. dfe_start_read((DICTIONARY *)host->rrdctx, rc) {
  1960. rrdcontext_recalculate_context_retention(rc, reason, job_id);
  1961. }
  1962. dfe_done(rc);
  1963. }
  1964. static void rrdcontext_recalculate_retention(int job_id) {
  1965. rrdcontext_next_db_rotation_ut = 0;
  1966. rrd_rdlock();
  1967. RRDHOST *host;
  1968. rrdhost_foreach_read(host) {
  1969. rrdcontext_recalculate_host_retention(host, RRD_FLAG_UPDATE_REASON_DB_ROTATION, job_id);
  1970. }
  1971. rrd_unlock();
  1972. }
  1973. void rrdcontext_delete_from_sql_unsafe(RRDCONTEXT *rc) {
  1974. // we need to refresh the string pointers in rc->hub
  1975. // in case the context changed values
  1976. rc->hub.id = string2str(rc->id);
  1977. rc->hub.title = string2str(rc->title);
  1978. rc->hub.units = string2str(rc->units);
  1979. rc->hub.family = string2str(rc->family);
  1980. // delete it from SQL
  1981. if(ctx_delete_context(&rc->rrdhost->host_uuid, &rc->hub) != 0)
  1982. error("RRDCONTEXT: failed to delete context '%s' version %lu from SQL.", rc->hub.id, rc->hub.version);
  1983. }
  1984. static void rrdcontext_garbage_collect(void) {
  1985. rrd_rdlock();
  1986. RRDHOST *host;
  1987. rrdhost_foreach_read(host) {
  1988. RRDCONTEXT *rc;
  1989. dfe_start_write((DICTIONARY *)host->rrdctx, rc) {
  1990. worker_is_busy(WORKER_JOB_CLEANUP);
  1991. rrdcontext_lock(rc);
  1992. if(unlikely(rrdcontext_should_be_deleted(rc))) {
  1993. worker_is_busy(WORKER_JOB_CLEANUP_DELETE);
  1994. rrdcontext_delete_from_sql_unsafe(rc);
  1995. if(dictionary_del_having_write_lock((DICTIONARY *)host->rrdctx, string2str(rc->id)) != 0)
  1996. error("RRDCONTEXT: '%s' of host '%s' failed to be deleted from rrdcontext dictionary.",
  1997. string2str(rc->id), host->hostname);
  1998. }
  1999. else {
  2000. RRDINSTANCE *ri;
  2001. dfe_start_write(rc->rrdinstances, ri) {
  2002. if(rrdinstance_should_be_deleted(ri)) {
  2003. worker_is_busy(WORKER_JOB_CLEANUP_DELETE);
  2004. dictionary_del_having_write_lock(rc->rrdinstances, string2str(ri->id));
  2005. }
  2006. else {
  2007. RRDMETRIC *rm;
  2008. dfe_start_write(ri->rrdmetrics, rm) {
  2009. if(rrdmetric_should_be_deleted(rm)) {
  2010. worker_is_busy(WORKER_JOB_CLEANUP_DELETE);
  2011. dictionary_del_having_write_lock(ri->rrdmetrics, string2str(rm->id));
  2012. }
  2013. }
  2014. dfe_done(rm);
  2015. }
  2016. }
  2017. dfe_done(ri);
  2018. }
  2019. // the item is referenced in the dictionary
  2020. // so, it is still here to unlock, even if we have deleted it
  2021. rrdcontext_unlock(rc);
  2022. }
  2023. dfe_done(rc);
  2024. }
  2025. rrd_unlock();
  2026. }
  2027. static void rrdcontext_main_cleanup(void *ptr) {
  2028. struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
  2029. static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
  2030. // custom code
  2031. static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
  2032. }
  2033. void *rrdcontext_main(void *ptr) {
  2034. if(unlikely(rrdcontext_enabled == CONFIG_BOOLEAN_NO))
  2035. return NULL;
  2036. worker_register("RRDCONTEXT");
  2037. worker_register_job_name(WORKER_JOB_HOSTS, "hosts");
  2038. worker_register_job_name(WORKER_JOB_CHECK, "dedup checks");
  2039. worker_register_job_name(WORKER_JOB_SEND, "sent contexts");
  2040. worker_register_job_name(WORKER_JOB_DEQUEUE, "deduped contexts");
  2041. worker_register_job_name(WORKER_JOB_RETENTION, "metrics retention");
  2042. worker_register_job_name(WORKER_JOB_QUEUED, "queued contexts");
  2043. worker_register_job_name(WORKER_JOB_CLEANUP, "cleanups");
  2044. worker_register_job_name(WORKER_JOB_CLEANUP_DELETE, "deletes");
  2045. netdata_thread_cleanup_push(rrdcontext_main_cleanup, ptr);
  2046. heartbeat_t hb;
  2047. heartbeat_init(&hb);
  2048. usec_t step = USEC_PER_SEC * RRDCONTEXT_WORKER_THREAD_HEARTBEAT_SECS;
  2049. while (!netdata_exit) {
  2050. worker_is_idle();
  2051. heartbeat_next(&hb, step);
  2052. if(!aclk_connected) continue;
  2053. usec_t now_ut = now_realtime_usec();
  2054. if(rrdcontext_next_db_rotation_ut && now_ut > rrdcontext_next_db_rotation_ut) {
  2055. rrdcontext_recalculate_retention(WORKER_JOB_RETENTION);
  2056. rrdcontext_garbage_collect();
  2057. rrdcontext_next_db_rotation_ut = 0;
  2058. }
  2059. rrd_rdlock();
  2060. RRDHOST *host;
  2061. rrdhost_foreach_read(host) {
  2062. worker_is_busy(WORKER_JOB_HOSTS);
  2063. // check if we have received a streaming command for this host
  2064. if(!rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS))
  2065. continue;
  2066. // check if there are queued items to send
  2067. if(!dictionary_stats_entries((DICTIONARY *)host->rrdctx_queue))
  2068. continue;
  2069. if(!host->node_id)
  2070. continue;
  2071. size_t messages_added = 0;
  2072. contexts_updated_t bundle = NULL;
  2073. RRDCONTEXT *rc;
  2074. dfe_start_write((DICTIONARY *)host->rrdctx_queue, rc) {
  2075. if(unlikely(messages_added >= MESSAGES_PER_BUNDLE_TO_SEND_TO_HUB_PER_HOST))
  2076. break;
  2077. worker_is_busy(WORKER_JOB_QUEUED);
  2078. usec_t dispatch_ut = rrdcontext_calculate_queued_dispatch_time_ut(rc, now_ut);
  2079. char *claim_id = get_agent_claimid();
  2080. if(unlikely(now_ut >= dispatch_ut) && claim_id) {
  2081. worker_is_busy(WORKER_JOB_CHECK);
  2082. rrdcontext_lock(rc);
  2083. if(check_if_cloud_version_changed_unsafe(rc, true)) {
  2084. worker_is_busy(WORKER_JOB_SEND);
  2085. #ifdef ENABLE_ACLK
  2086. if(!bundle) {
  2087. // prepare the bundle to send the messages
  2088. char uuid[UUID_STR_LEN];
  2089. uuid_unparse_lower(*host->node_id, uuid);
  2090. bundle = contexts_updated_new(claim_id, uuid, 0, now_ut);
  2091. }
  2092. #endif
  2093. // update the hub data of the context, give a new version, pack the message
  2094. // and save an update to SQL
  2095. rrdcontext_message_send_unsafe(rc, false, bundle);
  2096. messages_added++;
  2097. rc->queue.dequeued_ut = now_ut;
  2098. }
  2099. else
  2100. rc->version = rc->hub.version;
  2101. // remove the queued flag, so that it can be queued again
  2102. rc->flags &= ~RRD_FLAG_QUEUED;
  2103. // remove it from the queue
  2104. worker_is_busy(WORKER_JOB_DEQUEUE);
  2105. dictionary_del_having_write_lock((DICTIONARY *)host->rrdctx_queue, string2str(rc->id));
  2106. if(unlikely(rrdcontext_should_be_deleted(rc))) {
  2107. // this is a deleted context - delete it forever...
  2108. worker_is_busy(WORKER_JOB_CLEANUP_DELETE);
  2109. rrdcontext_delete_from_sql_unsafe(rc);
  2110. STRING *id = string_dup(rc->id);
  2111. rrdcontext_unlock(rc);
  2112. // delete it from the master dictionary
  2113. if(dictionary_del((DICTIONARY *)host->rrdctx, string2str(rc->id)) != 0)
  2114. error("RRDCONTEXT: '%s' of host '%s' failed to be deleted from rrdcontext dictionary.",
  2115. string2str(id), host->hostname);
  2116. string_freez(id);
  2117. }
  2118. else
  2119. rrdcontext_unlock(rc);
  2120. }
  2121. freez(claim_id);
  2122. }
  2123. dfe_done(rc);
  2124. #ifdef ENABLE_ACLK
  2125. if(bundle) {
  2126. // we have a bundle to send messages
  2127. // update the version hash
  2128. contexts_updated_update_version_hash(bundle, rrdcontext_version_hash(host));
  2129. // send it
  2130. aclk_send_contexts_updated(bundle);
  2131. }
  2132. #endif
  2133. }
  2134. rrd_unlock();
  2135. }
  2136. netdata_thread_cleanup_pop(1);
  2137. return NULL;
  2138. }