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