Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : */
4 :
5 : //
6 : // sandesh_uve.h
7 : //
8 : // This file has the interface for the generator-side UVE
9 : // cache.
10 : //
11 :
12 : #ifndef __SANDESH_UVE_H__
13 : #define __SANDESH_UVE_H__
14 :
15 : #include <map>
16 : #include <vector>
17 : #include <mutex>
18 :
19 : #include <boost/ptr_container/ptr_map.hpp>
20 : #include <boost/assign/ptr_map_inserter.hpp>
21 : #include <sandesh/sandesh_types.h>
22 : #include <sandesh/sandesh.h>
23 : #include <tbb/concurrent_hash_map.h>
24 : #include <boost/functional/hash.hpp>
25 :
26 : class SandeshUVEPerTypeMap;
27 :
28 : // This class holds a map of all per-SandeshUVE-type caches.
29 : // Each cache registers with this class during static initialization.
30 : //
31 : // The Sandesh Session state machine uses this class to
32 : // find the per-SandeshUVE-type caches and invoke operations on them.
33 : //
34 : class SandeshUVETypeMaps {
35 : public:
36 : typedef std::pair<int, SandeshUVEPerTypeMap *> uve_global_elem;
37 : typedef std::map<std::string, uve_global_elem> uve_global_map;
38 : typedef std::map<std::string, std::string> ds_conf_elem;
39 :
40 : static void RegisterType(const std::string &s, SandeshUVEPerTypeMap *tmap,
41 : int period) {
42 : assert(GetMap()->insert(std::make_pair(s,
43 : std::make_pair(period, tmap))).second);
44 : }
45 0 : static const uve_global_elem TypeMap(const std::string &s) {
46 0 : uve_global_map::const_iterator it = GetMap()->find(s);
47 0 : if (it == GetMap()->end())
48 0 : return std::make_pair(0, (SandeshUVEPerTypeMap *)(NULL));
49 : else
50 0 : return it->second;
51 : }
52 : static void SyncAllMaps(const std::map<std::string,uint32_t> &,
53 : bool periodic = false);
54 : static uint32_t Clear(const std::string& proxy, int partition);
55 : static bool InitDerivedStats(
56 : const std::map<std::string, ds_conf_elem> &);
57 : static void SyncIntrospect(std::string tname, std::string table, std::string key);
58 :
59 97 : static uve_global_map::const_iterator Begin() { return GetMap()->begin(); }
60 560 : static uve_global_map::const_iterator End() { return GetMap()->end(); }
61 : static const int kProxyPartitions = 30;
62 : private:
63 : static uve_global_map *map_;
64 :
65 38 : static uve_global_map * GetMap() {
66 38 : if (!map_) {
67 28 : map_ = new uve_global_map();
68 : }
69 38 : return map_;
70 : }
71 : };
72 :
73 :
74 : // This is the interface of the per-SandeshUVE-type caches.
75 : //
76 : class SandeshUVEPerTypeMap {
77 : public:
78 : virtual ~SandeshUVEPerTypeMap() { }
79 : virtual uint32_t TypeSeq() const = 0;
80 : virtual uint32_t SyncUVE(const std::string &table, SandeshUVE::SendType st,
81 : uint32_t seqno, uint32_t cycle, const std::string &ctx) = 0;
82 : virtual bool InitDerivedStats(
83 : const std::map<std::string,std::string> & dsconf) = 0;
84 : virtual bool SendUVE(const std::string& table, const std::string& name,
85 : const std::string& ctx) const = 0;
86 : virtual std::map<std::string, std::string> GetDSConf(void) const = 0;
87 : virtual int GetTimeout(void) const = 0;
88 : virtual uint32_t ClearUVEs(const std::string& proxy, int partition) = 0;
89 : };
90 :
91 :
92 : // This is the per-SandeshUVE-type cache.
93 : //
94 : //
95 : // It is parameterized on the Sandesh type, and the underlying UVE Type.
96 : //
97 : // Assumptions about Sandesh Type:
98 : // - Has a single element called "data", of the UVE type.
99 : // - Provides static "Send" function.
100 : // - Per-object sequence number is available as "lseqnum()"
101 : //
102 : // Assumptions about UVE Type
103 : // - Can be constructed with 0 arguments (default constructor)
104 : // - Has a boolean field called "deleted"
105 : // - The UVE key is in a field called "name"
106 : //
107 :
108 :
109 : template<typename T, typename U, int P, int TM>
110 : class SandeshUVEPerTypeMapImpl {
111 : public:
112 :
113 : struct HashCompare {
114 : static size_t hash( const std::string& key )
115 : { return boost::hash_value(key); }
116 : static bool equal( const std::string& key1, const std::string& key2 )
117 : { return ( key1 == key2 ); }
118 : };
119 :
120 : struct UVEMapEntry {
121 : UVEMapEntry(const std::string &table, uint32_t seqnum,
122 : SandeshLevel::type level):
123 : data(table), seqno(seqnum), level(level) {
124 : }
125 : U data;
126 : uint32_t seqno;
127 : SandeshLevel::type level;
128 : };
129 :
130 : // The key is the table name
131 : typedef boost::ptr_map<std::string, UVEMapEntry> uve_table_map;
132 :
133 : // The key is the UVE-Key
134 : typedef tbb::concurrent_hash_map<std::string, uve_table_map, HashCompare > uve_cmap;
135 :
136 : SandeshUVEPerTypeMapImpl() :
137 : dsconf_(T::_DSConf()) {}
138 :
139 : // This function is called whenever a SandeshUVE is sent from
140 : // the generator to the collector.
141 : // It updates the cache.
142 : bool UpdateUVE(U& data, uint32_t seqnum, uint64_t mono_usec,
143 : SandeshLevel::type level) {
144 : bool send = false;
145 : const std::string &table = data.table_;
146 : assert(!table.empty());
147 : const std::string &s = data.get_name();
148 : if (!mono_usec) mono_usec = ClockMonotonicUsec();
149 :
150 : // pickup DS Config
151 : std::unique_lock<std::mutex> lock(uve_mutex_);
152 : std::map<string,string> dsconf = dsconf_;
153 :
154 : // If we are going to erase, we need a global lock
155 : // to coordinate with iterators
156 : // To prevent deadlock, we always acquire global lock
157 : // before accessor
158 : if (!data.get_deleted()) {
159 : lock.unlock();
160 : }
161 : typename uve_cmap::accessor a;
162 :
163 : // Ensure that the entry exists, and we have an accessor to it
164 : while (true) {
165 : if (cmap_.find(a, s)) break;
166 : else {
167 : if (cmap_.insert(a, s)) break;
168 : }
169 : }
170 :
171 : typename uve_table_map::iterator imapentry = a->second.find(table);
172 : if (imapentry == a->second.end()) {
173 : std::auto_ptr<UVEMapEntry> ume(new
174 : UVEMapEntry(data.table_, seqnum, level));
175 : T::_InitDerivedStats(ume->data, dsconf);
176 : send = T::UpdateUVE(data, ume->data, mono_usec, level);
177 : imapentry = a->second.insert(table, ume).first;
178 : } else {
179 : if (TM != 0) {
180 : // If we get an update , mark this UVE so that it is not
181 : // deleted during the next round of periodic processing
182 : imapentry->second->data.set_deleted(false);
183 : }
184 : send = T::UpdateUVE(data, imapentry->second->data, mono_usec,
185 : level);
186 : imapentry->second->seqno = seqnum;
187 : }
188 : if (data.get_deleted()) {
189 : a->second.erase(imapentry);
190 : if (a->second.empty()) cmap_.erase(a);
191 : lock.unlock();
192 : }
193 :
194 : return send;
195 : }
196 :
197 : // Clear all UVEs in this cache
198 : // This is used ONLY with proxy groups
199 : uint32_t ClearUVEs(void) {
200 :
201 : // Global lock is needed for iterator
202 : std::scoped_lock lock(uve_mutex_);
203 : uint32_t count = 0;
204 : (const_cast<SandeshUVEPerTypeMapImpl<T,U,P,TM> *>(this))->cmap_.rehash();
205 : typename uve_cmap::iterator git = cmap_.begin();
206 : while (git != cmap_.end()) {
207 : typename uve_cmap::accessor a;
208 : if (!cmap_.find(a, git->first)) continue;
209 : typename uve_table_map::iterator uit = a->second.begin();
210 : while (uit != a->second.end()) {
211 : SANDESH_LOG(INFO, __func__ << " Clearing " << uit->first <<
212 : " val " << uit->second->data.log() << " proxy " <<
213 : SandeshStructProxyTrait<U>::get(uit->second->data) <<
214 : " seq " << uit->second->seqno);
215 : uit->second->data.set_deleted(true);
216 : T::Send(uit->second->data, uit->second->level,
217 : SandeshUVE::ST_SYNC, uit->second->seqno, 0, "");
218 : count++;
219 : ++uit;
220 : }
221 : ++git;
222 : }
223 : cmap_.clear();
224 : return count;
225 : }
226 :
227 : bool InitDerivedStats(const std::map<std::string,std::string> & dsconf) {
228 :
229 : // Global lock is needed for iterator
230 : std::scoped_lock lock(uve_mutex_);
231 :
232 : // Copy the existing configuration
233 : // We will be replacing elements in it.
234 : std::map<std::string,std::string> dsnew = dsconf_;
235 :
236 : bool failure = false;
237 : for (map<std::string,std::string>::const_iterator n_iter = dsconf.begin();
238 : n_iter != dsconf.end(); n_iter++) {
239 : if (dsnew.find(n_iter->first) != dsnew.end()) {
240 : SANDESH_LOG(INFO, __func__ << " Overide DSConf for " <<
241 : n_iter->first << " , " << dsnew[n_iter->first] <<
242 : " with " << n_iter->second);
243 : dsnew[n_iter->first] = n_iter->second;
244 : } else {
245 : SANDESH_LOG(INFO, __func__ << " Cannot find DSConf for " <<
246 : n_iter->first << " , " << n_iter->second);
247 : failure = true;
248 : }
249 : }
250 :
251 : if (failure) return false;
252 :
253 : // Copy the new conf into the old one if there we no errors
254 : dsconf_ = dsnew;
255 :
256 : (const_cast<SandeshUVEPerTypeMapImpl<T,U,P,TM> *>(this))->cmap_.rehash();
257 : for (typename uve_cmap::iterator git = cmap_.begin();
258 : git != cmap_.end(); git++) {
259 : typename uve_cmap::accessor a;
260 : if (!cmap_.find(a, git->first)) continue;
261 : for (typename uve_table_map::iterator uit = a->second.begin();
262 : uit != a->second.end(); uit++) {
263 : SANDESH_LOG(INFO, __func__ << " Reset Derived Stats for " <<
264 : git->first);
265 : T::_InitDerivedStats(uit->second->data, dsconf_);
266 : }
267 : }
268 : return true;
269 : }
270 :
271 : uint32_t SyncUVE(const std::string &table,
272 : SandeshUVE::SendType st,
273 : uint32_t seqno, uint32_t cycle,
274 : const std::string &ctx) {
275 : // Global lock is needed for iterator
276 : std::scoped_lock lock(uve_mutex_);
277 : uint32_t count = 0;
278 : (const_cast<SandeshUVEPerTypeMapImpl<T,U,P,TM> *>(this))->cmap_.rehash();
279 : typename uve_cmap::iterator git = cmap_.begin();
280 : while (git != cmap_.end()) {
281 : typename uve_cmap::accessor a;
282 : if (!cmap_.find(a, git->first)) continue;
283 : typename uve_table_map::iterator uit = a->second.begin();
284 : while (uit != a->second.end()) {
285 : typename uve_table_map::iterator dit = a->second.end();
286 : if (!table.empty() && uit->first != table) continue;
287 : if ((seqno < uit->second->seqno) || (seqno == 0)) {
288 : if (ctx.empty()) {
289 : SANDESH_LOG(INFO, __func__ << " Syncing " << uit->first <<
290 : " val " << uit->second->data.log() << " proxy " <<
291 : SandeshStructProxyTrait<U>::get(uit->second->data) <<
292 : " seq " << uit->second->seqno);
293 : }
294 : T::Send(uit->second->data, uit->second->level, st,
295 : uit->second->seqno, cycle, ctx);
296 : if ((TM != 0) && (st == SandeshUVE::ST_PERIODIC)) {
297 : if ((cycle % TM) == 0) {
298 : if (uit->second->data.get_deleted()) {
299 : // This UVE was marked for deletion during the
300 : // last periodic processing round, and there
301 : // have been no UVE updates
302 : dit = uit;
303 : } else {
304 : // Mark this UVE to be deleted during the next
305 : // periodic processing round, unless it is updated
306 : uit->second->data.set_deleted(true);
307 : }
308 : }
309 : }
310 : count++;
311 : }
312 : ++uit;
313 : if (dit != a->second.end()) a->second.erase(dit);
314 : }
315 : ++git;
316 : if (a->second.empty()) cmap_.erase(a);
317 : }
318 : return count;
319 : }
320 :
321 : bool SendUVE(const std::string& table, const std::string& name,
322 : const std::string& ctx) const {
323 : bool sent = false;
324 : typename uve_cmap::const_accessor a;
325 : if (cmap_.find(a, name)) {
326 : for (typename uve_table_map::const_iterator uve_entry = a->second.begin();
327 : uve_entry != a->second.end(); uve_entry++) {
328 : if (!table.empty() && uve_entry->first != table) continue;
329 : sent = true;
330 : T::Send(uve_entry->second->data, uve_entry->second->level,
331 : (ctx.empty() ? SandeshUVE::ST_INTROSPECT : SandeshUVE::ST_SYNC),
332 : uve_entry->second->seqno, 0, ctx);
333 : }
334 : }
335 : return sent;
336 : }
337 :
338 : std::map<std::string, std::string> GetDSConf(void) const {
339 : return dsconf_;
340 : }
341 :
342 : private:
343 :
344 : uve_cmap cmap_;
345 : std::map<std::string, std::string> dsconf_;
346 : mutable std::mutex uve_mutex_;
347 : };
348 :
349 : #define SANDESH_UVE_DEF(x,y,z,w) \
350 : static SandeshUVEPerTypeMapGroup<x, y, z, w> uvemap ## x(#y)
351 :
352 : template<typename T, typename U, int P, int TM>
353 : class SandeshUVEPerTypeMapGroup: public SandeshUVEPerTypeMap {
354 : public:
355 : typedef SandeshUVEPerTypeMapImpl<T,U,P,TM> uve_emap;
356 :
357 : // Native map
358 : typedef boost::ptr_map<string, uve_emap> uve_nmap;
359 :
360 : // One UVE Type Map per partition
361 : typedef boost::ptr_map<int, uve_emap> uve_pmap;
362 :
363 : // One set of per-partition UVE Type Maps for each proxy group
364 : typedef boost::ptr_map<string, uve_pmap> uve_gmap;
365 :
366 : SandeshUVEPerTypeMapGroup(char const * u_name) {
367 : std::string native_name(std::string(""));
368 : uve_emap* ne = new uve_emap;
369 : native_group_map_.insert(native_name, ne);
370 : nativep_ = ne;
371 : SandeshUVETypeMaps::RegisterType(u_name, this, P);
372 : }
373 :
374 : string GetNMapUVEname(U &data) {
375 : std::string proxy = SandeshStructProxyTrait<U>::get(data);
376 : return proxy;
377 : }
378 :
379 : // Get the native map for the given UVE proxy name
380 : uve_emap *GetNMap(const std::string& name) {
381 : if (name == std::string("")){
382 : return(nativep_);
383 : }
384 : std::scoped_lock lock(nmutex_);
385 : if (native_group_map_.find(name) == native_group_map_.end()) {
386 : uve_emap* ne = new uve_emap;
387 : std::string native_name(name);
388 : native_group_map_.insert(native_name, ne);
389 : }
390 : return &(native_group_map_.at(name));
391 : }
392 :
393 : // Get the Native UVE and Native proxy UVE maps for this Native type
394 : std::vector<uve_emap*> GetNMaps(void) {
395 : std::scoped_lock lock(nmutex_);
396 : std::vector<uve_emap *> nev;
397 : for (typename uve_nmap::iterator uni = native_group_map_.begin();
398 : uni != native_group_map_.end(); uni++) {
399 : nev.push_back(uni->second);
400 : }
401 : return nev;
402 : }
403 :
404 : // Get the partition UVE Type maps for the given proxy group
405 : uve_pmap *GetGMap(const std::string& proxy) {
406 : std::scoped_lock lock(gmutex_);
407 : if (group_map_.find(proxy) == group_map_.end()) {
408 : uve_pmap* up = new uve_pmap;
409 : std::string kstring(proxy);
410 : for (size_t idx=0; idx<SandeshUVETypeMaps::kProxyPartitions; idx++) {
411 : boost::assign::ptr_map_insert(*up)(idx);
412 : }
413 : group_map_.insert(kstring ,up);
414 : }
415 : return &(group_map_.at(proxy));
416 : }
417 :
418 : // Get the partition UVE Type maps for all proxy groups
419 : std::vector<uve_pmap *> GetGMaps(void) {
420 : std::scoped_lock lock(gmutex_);
421 : std::vector<uve_pmap *> pv;
422 : for (typename uve_gmap::iterator ugi = group_map_.begin();
423 : ugi != group_map_.end(); ugi++) {
424 : pv.push_back(ugi->second);
425 : }
426 : return pv;
427 : }
428 :
429 : // This function can be used by the Sandesh Session state machine
430 : // to get the seq num of the last message sent for this SandeshUVE type
431 : uint32_t TypeSeq(void) const {
432 : return T::lseqnum();
433 : }
434 :
435 : int GetTimeout(void) const {
436 : return TM;
437 : }
438 :
439 : // Sync UVEs for both the native UVE Map and the proxy groups
440 : uint32_t SyncUVE(const std::string &table,
441 : SandeshUVE::SendType st,
442 : uint32_t seqno, uint32_t cycle,
443 : const std::string &ctx) {
444 : uint32_t count=0;
445 : std::vector<uve_emap *> nev = GetNMaps();
446 : for (size_t idx=0; idx<nev.size(); idx++) {
447 : count += nev[idx]->SyncUVE(table, st, seqno, cycle,ctx);
448 : }
449 :
450 : std::vector<uve_pmap *> pv = GetGMaps();
451 : for (size_t jdx=0; jdx<pv.size(); jdx++) {
452 : for (size_t idx=0; idx<SandeshUVETypeMaps::kProxyPartitions; idx++) {
453 : count += pv[jdx]->at(idx).SyncUVE(table, st, seqno, cycle, ctx);
454 : }
455 : }
456 : return count;
457 : }
458 :
459 : // DerivedStats for proxy groups cannot be re-configured
460 : // at InitGenerator time or from Introspect
461 : // The Proxy group configuration should be changed instead
462 : bool InitDerivedStats(
463 : const std::map<std::string,std::string> & dsconf) {
464 : return (nativep_->InitDerivedStats(dsconf));
465 : }
466 :
467 : // Send the given UVE, for the native UVE Type map, as
468 : // well as proxy groups
469 : bool SendUVE(const std::string& table, const std::string& name,
470 : const std::string& ctx) const {
471 : bool sent = false;
472 : std::vector<uve_emap *> nev =
473 : const_cast<SandeshUVEPerTypeMapGroup<T,U,P,TM> * >(this)->GetNMaps();
474 : for (size_t idx=0; idx<nev.size(); idx++) {
475 : if (nev[idx]->SendUVE(table, name, ctx)) sent = true;
476 : }
477 :
478 : std::vector<uve_pmap *> pv =
479 : const_cast<SandeshUVEPerTypeMapGroup<T,U,P,TM> * >(this)->GetGMaps();
480 : for (size_t jdx=0; jdx<pv.size(); jdx++) {
481 : for (size_t idx=0; idx<SandeshUVETypeMaps::kProxyPartitions; idx++) {
482 : if (pv[jdx]->at(idx).SendUVE(table, name, ctx)) sent = true;
483 : }
484 : }
485 : return sent;
486 : }
487 :
488 : std::map<std::string, std::string> GetDSConf(void) const {
489 : return nativep_->GetDSConf();
490 : }
491 :
492 : bool UpdateUVE(U& data, uint32_t seqnum,
493 : uint64_t mono_usec, int partition, SandeshLevel::type level) {
494 : if (partition == -1) {
495 : uve_emap *ne = GetNMap(GetNMapUVEname(data));
496 : return ne->UpdateUVE(data, seqnum, mono_usec, level);
497 : } else {
498 : std::string proxy = SandeshStructProxyTrait<U>::get(data);
499 : assert(partition < SandeshUVETypeMaps::kProxyPartitions);
500 : uve_pmap * pp = GetGMap(proxy);
501 : return pp->at(partition).UpdateUVE(data, seqnum, mono_usec, level);
502 : }
503 : }
504 :
505 : // Delete all UVEs for the given partition for the given proxy group
506 : uint32_t ClearUVEs(const std::string& proxy, int partition) {
507 : if (partition != -1) {
508 : std::scoped_lock lock(gmutex_);
509 : typename uve_gmap::iterator gi = group_map_.find(proxy);
510 : if (gi != group_map_.end()) {
511 : assert(partition < SandeshUVETypeMaps::kProxyPartitions);
512 : return gi->second->at(partition).ClearUVEs();
513 : }
514 : }
515 : return 0;
516 : }
517 :
518 : private:
519 : mutable std::mutex gmutex_;
520 : uve_gmap group_map_;
521 : mutable std::mutex nmutex_;
522 : uve_nmap native_group_map_;
523 : uve_emap *nativep_;
524 : };
525 :
526 : #endif
|
