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 : #ifndef DB_HANDLER_H_
7 : #define DB_HANDLER_H_
8 :
9 : #include <mutex>
10 :
11 : #include <boost/scoped_ptr.hpp>
12 : #include <boost/shared_ptr.hpp>
13 : #include <boost/asio/ip/tcp.hpp>
14 : #include <boost/ptr_container/ptr_map.hpp>
15 : #include <boost/uuid/uuid.hpp>
16 : #include <boost/uuid/name_generator.hpp>
17 : #include <boost/bind/bind.hpp>
18 :
19 : #if __GNUC_PREREQ(4, 6)
20 : #pragma GCC diagnostic push
21 : #pragma GCC diagnostic ignored "-Wunused-result"
22 : #endif
23 : #include <boost/uuid/uuid_generators.hpp>
24 : #if __GNUC_PREREQ(4, 6)
25 : #pragma GCC diagnostic pop
26 : #endif
27 :
28 : #include <boost/tuple/tuple.hpp>
29 :
30 : #include "base/parse_object.h"
31 : #include "io/event_manager.h"
32 : #include "base/random_generator.h"
33 : #include <base/watermark.h>
34 : #include "gendb_if.h"
35 : #include "gendb_statistics.h"
36 : #include "viz_message.h"
37 : #include <analytics/uflow_types.h>
38 : #include <analytics/viz_constants.h>
39 : #include <database/cassandra/cql/cql_types.h>
40 : #include <analytics/collector_uve_types.h>
41 : #include "config_client_collector.h"
42 : #include "usrdef_counters.h"
43 : #include "options.h"
44 :
45 : using namespace boost::placeholders;
46 :
47 : class Options;
48 :
49 : /*
50 : * Stats for SessionTable
51 : */
52 : class SessionTableDbStats {
53 : public:
54 8 : SessionTableDbStats() :
55 8 : num_messages(0),
56 8 : num_writes(0),
57 8 : num_samples(0),
58 8 : curr_json_size(0) {}
59 : uint64_t num_messages;
60 : uint64_t num_writes;
61 : uint64_t num_samples;
62 : uint64_t curr_json_size;
63 : };
64 :
65 : class DbHandler {
66 : public:
67 : static const int DefaultDbTTL = 0;
68 : static boost::uuids::uuid seed_uuid;
69 :
70 : typedef enum {
71 : INVALID = 0,
72 : UINT64 = 1,
73 : STRING = 2,
74 : DOUBLE = 3,
75 : LIST = 4,
76 : MAP = 5,
77 : MAXVAL
78 : } VarType;
79 :
80 : struct Var {
81 65 : Var() : type(INVALID), str(""), num(0), dbl(0), vec(std::vector<std::string>()),
82 65 : map(std::map<std::string, std::string>()) {}
83 152 : Var(const std::string &s) : type(STRING), str(s), num(0), dbl(0),
84 152 : vec(std::vector<std::string>()),
85 152 : map(std::map<std::string, std::string>()) {}
86 21 : Var(uint64_t v) : type(UINT64), str(""), num(v), dbl(0),
87 21 : vec(std::vector<std::string>()),
88 21 : map(std::map<std::string, std::string>()) {}
89 0 : Var(double d) : type(DOUBLE), str(""), num(0), dbl(d),
90 0 : vec(std::vector<std::string>()),
91 0 : map(std::map<std::string, std::string>()) {}
92 0 : Var(const std::vector<std::string> &v) : type(LIST), str(""), num(0),
93 0 : dbl(0), vec(v),
94 0 : map(std::map<std::string, std::string>()) {}
95 0 : Var(const std::map<std::string, std::string> &m) : type(MAP), str(""), num(0),
96 0 : dbl(0), vec(std::vector<std::string>()),
97 0 : map(m) {}
98 : VarType type;
99 : std::string str;
100 : uint64_t num;
101 : double dbl;
102 : std::vector<std::string> vec;
103 : std::map<std::string, std::string> map;
104 21 : bool operator==(const Var &other) const {
105 21 : if (type!=other.type) return false;
106 21 : switch (type) {
107 18 : case STRING:
108 18 : if (str!=other.str) return false;
109 18 : break;
110 3 : case UINT64:
111 3 : if (num!=other.num) return false;
112 3 : break;
113 0 : case DOUBLE:
114 0 : if (dbl!=other.dbl) return false;
115 0 : break;
116 0 : case LIST:
117 0 : if (vec!=other.vec) return false;
118 0 : break;
119 0 : case MAP:
120 0 : if (map!=other.map) return false;
121 0 : break;
122 0 : default:
123 0 : break;
124 : }
125 21 : return true;
126 : }
127 : friend inline std::ostream& operator<<(std::ostream& out,
128 : const Var& value);
129 : };
130 :
131 : typedef std::map<std::string, std::string> RuleMap;
132 :
133 : typedef std::map<std::string, Var > AttribMap;
134 : typedef std::multimap<std::string, std::pair<Var, AttribMap> > TagMap;
135 : typedef std::vector<std::string> ObjectNamesVec;
136 :
137 : DbHandler(EventManager *evm, GenDb::GenDbIf::DbErrorHandler err_handler,
138 : std::string name,
139 : const Options::Cassandra &cassandra_options,
140 : bool use_db_write_options,
141 : const DbWriteOptions &db_write_options,
142 : ConfigClientCollector *config_client);
143 : DbHandler(GenDb::GenDbIf *dbif, const TtlMap& ttl_map);
144 : virtual ~DbHandler();
145 :
146 : static uint64_t GetTtlInHourFromMap(const TtlMap& ttl_map,
147 : TtlType::type type);
148 : static uint64_t GetTtlFromMap(const TtlMap& ttl_map,
149 : TtlType::type type);
150 : bool DropMessage(const SandeshHeader &header, const VizMsg *vmsg);
151 : bool Init(bool initial);
152 : void UnInit();
153 : void GetRuleMap(RuleMap& rulemap);
154 :
155 : virtual void MessageTableInsert(const VizMsg *vmsgp,
156 : const ObjectNamesVec &object_names,
157 : GenDb::GenDbIf::DbAddColumnCb db_cb);
158 : void ObjectTableInsert(const std::string &table, const std::string &rowkey,
159 : uint64_t ×tamp, const boost::uuids::uuid& unm,
160 : const VizMsg *vmsgp, GenDb::GenDbIf::DbAddColumnCb db_cb);
161 : void StatTableInsert(uint64_t ts,
162 : const std::string& statName,
163 : const std::string& statAttr,
164 : const TagMap & attribs_tag,
165 : const AttribMap & attribs_all,
166 : GenDb::GenDbIf::DbAddColumnCb db_cb);
167 : bool SessionTableInsert(const pugi::xml_node& parent,
168 : const SandeshHeader &header, GenDb::GenDbIf::DbAddColumnCb db_cb);
169 : bool UnderlayFlowSampleInsert(const UFlowData& flow_data,
170 : uint64_t timestamp, GenDb::GenDbIf::DbAddColumnCb db_cb);
171 :
172 : bool GetStats(uint64_t *queue_count, uint64_t *enqueues) const;
173 :
174 : bool GetStats(std::vector<GenDb::DbTableInfo> *vdbti,
175 : GenDb::DbErrors *dbe, std::vector<GenDb::DbTableInfo> *vstats_dbti);
176 : bool GetCumulativeStats(std::vector<GenDb::DbTableInfo> *vdbti,
177 : GenDb::DbErrors *dbe, std::vector<GenDb::DbTableInfo> *vstats_dbti)
178 : const;
179 : void GetSandeshStats(std::string *drop_level,
180 : std::vector<SandeshStats> *vdropmstats) const;
181 : bool GetSessionTableDbInfo(SessionTableDbInfo *session_table_info);
182 : bool GetCqlMetrics(cass::cql::Metrics *metrics) const;
183 : bool GetCqlStats(cass::cql::DbStats *stats) const;
184 : void SetDbQueueWaterMarkInfo(Sandesh::QueueWaterMarkInfo &wm,
185 : boost::function<void (void)> defer_undefer_cb);
186 : void ResetDbQueueWaterMarkInfo();
187 : std::vector<boost::asio::ip::tcp::endpoint> GetEndpoints() const;
188 : std::string GetName() const;
189 : ConfigClientCollector *GetConfigClient() {
190 : return config_client_;
191 : }
192 : bool IsAllWritesDisabled() const;
193 : bool IsStatisticsWritesDisabled() const;
194 : bool IsMessagesWritesDisabled() const;
195 : void DisableAllWrites(bool disable);
196 : void DisableStatisticsWrites(bool disable);
197 : void DisableMessagesWrites(bool disable);
198 :
199 : // Disk Usage Percentage
200 : void SetDiskUsagePercentageDropLevel(size_t count,
201 : SandeshLevel::type drop_level);
202 0 : SandeshLevel::type GetDiskUsagePercentageDropLevel() const {
203 0 : return disk_usage_percentage_drop_level_;
204 : }
205 : void SetDiskUsagePercentage(size_t disk_usage_percentage);
206 : uint32_t GetDiskUsagePercentage() const { return disk_usage_percentage_; }
207 : void SetDiskUsagePercentageHighWaterMark(uint32_t disk_usage_percentage,
208 : SandeshLevel::type level);
209 : void SetDiskUsagePercentageLowWaterMark(uint32_t disk_usage_percentage,
210 : SandeshLevel::type level);
211 : void ProcessDiskUsagePercentage(uint32_t disk_usage_percentage);
212 :
213 : // Pending Compaction Tasks
214 : void SetPendingCompactionTasksDropLevel(size_t count,
215 : SandeshLevel::type drop_level);
216 0 : SandeshLevel::type GetPendingCompactionTasksDropLevel() const {
217 0 : return pending_compaction_tasks_drop_level_;
218 : }
219 : void SetPendingCompactionTasks(size_t pending_compaction_tasks);
220 : uint32_t GetPendingCompactionTasks() const {
221 : return pending_compaction_tasks_;
222 : }
223 : void SetPendingCompactionTasksHighWaterMark(
224 : uint32_t pending_compaction_tasks,
225 : SandeshLevel::type level);
226 : void SetPendingCompactionTasksLowWaterMark(
227 : uint32_t pending_compaction_tasks,
228 : SandeshLevel::type level);
229 : void ProcessPendingCompactionTasks(uint32_t pending_compaction_tasks);
230 : void GetUDCConfig(std::vector<LogStatisticConfigInfo> *config_info) {
231 : udc_->GetUDCConfig(config_info);
232 : }
233 0 : void ReceiveConfig(const contrail_rapidjson::Document &jdoc, bool add_change) {
234 0 : udc_->UDCHandler(jdoc, add_change);
235 0 : }
236 : private:
237 : void StatTableInsertTtl(uint64_t ts,
238 : const std::string& statName,
239 : const std::string& statAttr,
240 : const TagMap & attribs_tag,
241 : const AttribMap & attribs_all, int ttl,
242 : GenDb::GenDbIf::DbAddColumnCb db_cb);
243 : void FieldNamesTableInsert(uint64_t timestamp,
244 : const std::string& table_name, const std::string& field_name,
245 : const std::string& field_val, int ttl,
246 : GenDb::GenDbIf::DbAddColumnCb db_cb);
247 : void MessageTableOnlyInsert(const VizMsg *vmsgp,
248 : const ObjectNamesVec &object_names,
249 : GenDb::GenDbIf::DbAddColumnCb db_cb);
250 : bool AllowMessageTableInsert(const SandeshHeader &header);
251 : bool CreateTables();
252 : void SetDropLevel(size_t queue_count, SandeshLevel::type level,
253 : boost::function<void (void)> cb);
254 : bool Setup();
255 : bool Initialize();
256 : bool StatTableWrite(uint32_t t2, const std::string& statName,
257 : const std::string& statAttr, const std::string& source,
258 : const std::string& name, const std::string& key, const std::string& proxy,
259 : const std::vector<std::vector<std::string> >& tags,
260 : uint32_t t1, const boost::uuids::uuid& unm, const std::string& jsonline,
261 : int ttl, GenDb::GenDbIf::DbAddColumnCb db_cb);
262 : bool StatTableWrite(uint32_t t2,
263 : const std::string& statName, const std::string& statAttr,
264 : const std::pair<std::string,DbHandler::Var>& ptag,
265 : const std::pair<std::string,DbHandler::Var>& stag,
266 : uint32_t t1, const boost::uuids::uuid& unm,
267 : const std::string& jsonline, int ttl,
268 : GenDb::GenDbIf::DbAddColumnCb db_cb);
269 : bool SessionSampleAdd(const pugi::xml_node& sessiondata,
270 : const SandeshHeader& header,
271 : GenDb::GenDbIf::DbAddColumnCb db_cb);
272 5 : uint64_t GetTtl(TtlType::type type) {
273 5 : return GetTtlFromMap(ttl_map_, type);
274 : }
275 : bool CanRecordDataForT2(uint32_t, std::string);
276 : bool InsertIntoDb(std::auto_ptr<GenDb::ColList> col_list,
277 : GenDb::DbConsistency::type dconsistency,
278 : GenDb::GenDbIf::DbAddColumnCb db_cb);
279 :
280 : boost::scoped_ptr<GenDb::GenDbIf> dbif_;
281 : // Random generator for UUIDs
282 : ThreadSafeUuidGenerator umn_gen_;
283 : std::string name_;
284 : std::string col_name_;
285 : SandeshLevel::type drop_level_;
286 : VizMsgStatistics dropped_msg_stats_;
287 : GenDb::DbTableStatistics stable_stats_;
288 : mutable std::mutex smutex_;
289 : TtlMap ttl_map_;
290 : static uint32_t field_cache_index_;
291 : static std::set<std::string> field_cache_set_;
292 : static std::mutex fmutex_;
293 : std::string tablespace_;
294 : std::string compaction_strategy_;
295 : std::string replication_factor_;
296 : std::string flow_tables_compaction_strategy_;
297 : UniformInt8RandomGenerator gen_partition_no_;
298 : bool disable_all_writes_;
299 : bool disable_statistics_writes_;
300 : bool disable_messages_writes_;
301 : ConfigClientCollector *config_client_;
302 : boost::scoped_ptr<UserDefinedCounters> udc_;
303 : static const int kUDCPollInterval = 120 * 1000; // in ms
304 : bool use_db_write_options_;
305 : uint32_t disk_usage_percentage_;
306 : SandeshLevel::type disk_usage_percentage_drop_level_;
307 : uint32_t pending_compaction_tasks_;
308 : SandeshLevel::type pending_compaction_tasks_drop_level_;
309 : mutable std::mutex disk_usage_percentage_water_mutex_;
310 : mutable std::mutex pending_compaction_tasks_water_mutex_;
311 : WaterMarkTuple disk_usage_percentage_watermark_tuple_;
312 : WaterMarkTuple pending_compaction_tasks_watermark_tuple_;
313 : SessionTableDbStats session_table_db_stats_;
314 :
315 : friend class DbHandlerTest;
316 :
317 : DISALLOW_COPY_AND_ASSIGN(DbHandler);
318 : };
319 :
320 : typedef boost::shared_ptr<DbHandler> DbHandlerPtr;
321 :
322 75 : inline std::ostream& operator<<(std::ostream& out, const DbHandler::Var& value) {
323 75 : switch (value.type) {
324 60 : case DbHandler::STRING:
325 60 : out << value.str;
326 60 : break;
327 15 : case DbHandler::UINT64:
328 15 : out << value.num;
329 15 : break;
330 0 : case DbHandler::DOUBLE:
331 0 : out << value.dbl;
332 0 : break;
333 0 : case DbHandler::LIST:
334 0 : out << boost::algorithm::join(value.vec, "; ");
335 0 : break;
336 0 : case DbHandler::MAP:
337 0 : for (std::map<std::string, std::string>::const_iterator itr = value.map.begin();
338 0 : itr != value.map.end(); itr++) {
339 0 : out << "{" << itr->first << ":" << itr->second << "}, ";
340 : }
341 0 : break;
342 0 : default:
343 0 : out << "Invalid type: " << value.type;
344 0 : break;
345 : }
346 75 : return out;
347 : }
348 :
349 : namespace zookeeper {
350 : namespace client {
351 : class ZookeeperClient;
352 : class ZookeeperLock;
353 : } // namespace client
354 : } // namespace zookeeper
355 :
356 : //
357 : // DbHandlerInitializer - Wrapper to perform DbHandler initialization
358 : //
359 : class DbHandlerInitializer {
360 : public:
361 : typedef boost::function<void(void)> InitializeDoneCb;
362 : DbHandlerInitializer(EventManager *evm,
363 : const std::string &db_name,
364 : const std::string &timer_task_name, InitializeDoneCb callback,
365 : const Options::Cassandra &cassandra_options,
366 : const std::string &zookeeper_server_list,
367 : bool use_zookeeper,
368 : const DbWriteOptions &db_write_options,
369 : ConfigClientCollector *config_client);
370 : DbHandlerInitializer(EventManager *evm,
371 : const std::string &db_name,
372 : const std::string &timer_task_name, InitializeDoneCb callback,
373 : DbHandlerPtr db_handler);
374 : virtual ~DbHandlerInitializer();
375 : bool Initialize();
376 : void Shutdown();
377 : DbHandlerPtr GetDbHandler() const;
378 :
379 : private:
380 : bool InitTimerExpired();
381 : void InitTimerErrorHandler(std::string error_name,
382 : std::string error_message);
383 : void StartInitTimer();
384 : void ScheduleInit();
385 :
386 : static const int kInitRetryInterval = 10 * 1000; // in ms
387 : const std::string db_name_;
388 : DbHandlerPtr db_handler_;
389 : InitializeDoneCb callback_;
390 : Timer *db_init_timer_;
391 : std::string zookeeper_server_list_;
392 : bool use_zookeeper_;
393 : bool zoo_locked_;
394 : boost::scoped_ptr<zookeeper::client::ZookeeperClient> zoo_client_;
395 : boost::scoped_ptr<zookeeper::client::ZookeeperLock> zoo_mutex_;
396 : };
397 :
398 : std::string PrependT2(uint32_t T2, const std::string &str);
399 :
400 : #endif /* DB_HANDLER_H_ */
|
