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