Line data Source code
1 : /*
2 : * Copyright (c) 2016 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : #ifndef config_cass_client_h
7 : #define config_cass_client_h
8 :
9 : #include <boost/ptr_container/ptr_map.hpp>
10 : #include <boost/shared_ptr.hpp>
11 :
12 : #include <atomic>
13 : #include <list>
14 : #include <map>
15 : #include <set>
16 : #include <string>
17 : #include <utility>
18 : #include <vector>
19 :
20 : #include "base/queue_task.h"
21 : #include "base/timer.h"
22 :
23 : #include "config_db_client.h"
24 : #include "config_json_parser_base.h"
25 : #include "config_cassandra_client.h"
26 : #include "database/gendb_if.h"
27 : #include "json_adapter_data.h"
28 :
29 : class EventManager;
30 : class ConfigClientManager;
31 : struct ConfigDBConnInfo;
32 : class TaskTrigger;
33 : class ConfigCassandraClient;
34 : struct ConfigCassandraParseContext;
35 : class ConfigDBFQNameCacheEntry;
36 : class ConfigDBUUIDCacheEntry;
37 :
38 : class ConfigCassandraPartition {
39 : public:
40 : ConfigCassandraPartition(ConfigCassandraClient *client, size_t idx);
41 : virtual ~ConfigCassandraPartition();
42 :
43 : typedef boost::shared_ptr<WorkQueue<ObjectProcessReq *> >
44 : ObjProcessWorkQType;
45 :
46 : struct FieldTimeStampInfo {
47 : uint64_t time_stamp;
48 : bool refreshed;
49 : };
50 :
51 : struct cmp_json_key {
52 627933 : bool operator()(const JsonAdapterDataType &k1,
53 : const JsonAdapterDataType &k2) const {
54 627933 : return k1.key < k2.key;
55 : }
56 : };
57 : typedef std::map<JsonAdapterDataType, FieldTimeStampInfo, cmp_json_key>
58 : FieldDetailMap;
59 : class ObjCacheEntry : public ObjectCacheEntry {
60 : public:
61 23520 : ObjCacheEntry(ConfigCassandraPartition *parent,
62 : uint64_t last_read_tstamp)
63 23520 : : ObjectCacheEntry(last_read_tstamp),
64 23520 : retry_count_(0), retry_timer_(NULL),
65 23520 : parent_(parent) {
66 23520 : }
67 :
68 : virtual ~ObjCacheEntry();
69 :
70 : void EnableCassandraReadRetry(const std::string uuid);
71 : void DisableCassandraReadRetry(const std::string uuid);
72 832063 : FieldDetailMap &GetFieldDetailMap() { return field_detail_map_; }
73 194 : const FieldDetailMap &GetFieldDetailMap() const {
74 194 : return field_detail_map_;
75 : }
76 161 : uint32_t GetRetryCount() const { return retry_count_; }
77 38 : bool IsRetryTimerCreated() const { return (retry_timer_ != NULL); }
78 : bool IsRetryTimerRunning() const;
79 4 : Timer *GetRetryTimer() { return retry_timer_; }
80 :
81 : private:
82 : friend class ConfigCassandraPartitionTest;
83 : friend class ConfigCassandraPartitionTest2;
84 : friend class ConfigCassandraClientPartitionTest;
85 :
86 : bool CassReadRetryTimerExpired(const std::string uuid);
87 : void CassReadRetryTimerErrorHandler();
88 : uint32_t retry_count_;
89 : Timer *retry_timer_;
90 : FieldDetailMap field_detail_map_;
91 : ConfigCassandraPartition *parent_;
92 : };
93 :
94 : static const uint32_t kMaxUUIDRetryTimePowOfTwo = 20;
95 : static const uint32_t kMinUUIDRetryTimeMSec = 100;
96 : typedef boost::ptr_map<std::string, ObjCacheEntry> ObjectCacheMap;
97 :
98 33696 : ObjProcessWorkQType obj_process_queue() {
99 33696 : return obj_process_queue_;
100 : }
101 :
102 : virtual int UUIDRetryTimeInMSec(const ObjCacheEntry *obj) const;
103 : ObjCacheEntry *GetObjCacheEntry(const std::string &uuid);
104 : const ObjCacheEntry *GetObjCacheEntry(const std::string &uuid) const;
105 : bool StoreKeyIfUpdated(const std::string &uuid,
106 : JsonAdapterDataType *adapter,
107 : uint64_t timestamp,
108 : ConfigCassandraParseContext &context);
109 : void ListMapPropReviseUpdateList(const std::string &uuid,
110 : ConfigCassandraParseContext &context);
111 : ObjCacheEntry *MarkCacheDirty(const std::string &uuid);
112 4 : void DeleteCacheMap(const std::string &uuid) {
113 4 : object_cache_map_.erase(uuid);
114 4 : }
115 : void Enqueue(ObjectProcessReq *req);
116 :
117 :
118 : bool UUIDToObjCacheShow(
119 : const std::string &search_string, const std::string &last_uuid,
120 : uint32_t num_entries,
121 : std::vector<ConfigDBUUIDCacheEntry> *entries) const;
122 2 : int GetInstanceId() const { return worker_id_; }
123 :
124 : boost::asio::io_context *ioservice();
125 :
126 : bool IsListOrMapPropEmpty(const string &uuid_key, const string &lookup_key);
127 : bool IsTaskTriggered() const;
128 : protected:
129 : virtual bool ReadObjUUIDTable(const std::set<std::string> &uuid_list);
130 : bool ProcessObjUUIDTableEntry(const std::string &uuid_key,
131 : const GenDb::ColList &col_list);
132 : virtual void ParseObjUUIDTableEntry(const std::string &uuid,
133 : const GenDb::ColList &col_list, CassColumnKVVec *cass_data_vec,
134 : ConfigCassandraParseContext &context);
135 : void ParseObjUUIDTableEachColumnBuildContext(const std::string &uuid,
136 : const std::string &key, const std::string &value,
137 : uint64_t timestamp, CassColumnKVVec *cass_data_vec,
138 : ConfigCassandraParseContext &context);
139 : virtual void HandleObjectDelete(const string &uuid, bool add_change);
140 821654 : ConfigCassandraClient *client() { return config_client_; }
141 :
142 : private:
143 : friend class ConfigCassandraClient;
144 :
145 : struct ObjectProcessRequestType {
146 129644 : ObjectProcessRequestType(const std::string &in_oper,
147 : const std::string &in_obj_type,
148 : const std::string &in_uuid)
149 129644 : : oper(in_oper), obj_type(in_obj_type), uuid(in_uuid) {
150 129641 : }
151 : std::string oper;
152 : std::string obj_type;
153 : std::string uuid;
154 : };
155 :
156 : typedef std::map<std::string, ObjectProcessRequestType *> UUIDProcessSet;
157 :
158 : bool RequestHandler(ObjectProcessReq *req);
159 : void AddUUIDToRequestList(const std::string &oper,
160 : const std::string &obj_type, const std::string &uuid_str);
161 : bool ConfigReader();
162 : void RemoveObjReqEntries(std::set<std::string> &req_list);
163 : void RemoveObjReqEntry(std::string &uuid);
164 : virtual void GenerateAndPushJson(
165 : const string &uuid_key, const string &obj_type,
166 : const CassColumnKVVec &cass_data_vec, bool add_change);
167 :
168 : void FillUUIDToObjCacheInfo(const std::string &uuid,
169 : ObjectCacheMap::const_iterator uuid_iter,
170 : ConfigDBUUIDCacheEntry *entry) const;
171 :
172 : ObjProcessWorkQType obj_process_queue_;
173 : UUIDProcessSet uuid_read_set_;
174 : ObjectCacheMap object_cache_map_;
175 : boost::shared_ptr<TaskTrigger> config_reader_;
176 : ConfigCassandraClient *config_client_;
177 : int worker_id_;
178 : };
179 :
180 : /*
181 : * This class has the functionality to interact with the cassandra servers that
182 : * store the user configuration.
183 : */
184 : class ConfigCassandraClient : public ConfigDbClient {
185 : public:
186 : // Cassandra table names
187 : static const std::string kUuidTableName;
188 : static const std::string kFqnTableName;
189 :
190 : // Task names
191 : static const std::string kCassClientTaskId;
192 : static const std::string kObjectProcessTaskId;
193 :
194 : // Number of UUIDs to read in one read request
195 : static const int kMaxNumUUIDToRead = 64;
196 :
197 : // Number of FQName entries to read in one read request
198 : static const int kNumFQNameEntriesToRead = 4096;
199 :
200 : typedef boost::scoped_ptr<GenDb::GenDbIf> GenDbIfPtr;
201 : typedef std::vector<ConfigCassandraPartition *> PartitionList;
202 :
203 : ConfigCassandraClient(ConfigClientManager *mgr, EventManager *evm,
204 : const ConfigClientOptions &options,
205 : int num_workers);
206 : virtual ~ConfigCassandraClient();
207 :
208 : virtual void InitDatabase();
209 : void BulkSyncDone();
210 : ConfigCassandraPartition *GetPartition(const std::string &uuid);
211 : const ConfigCassandraPartition *GetPartition(const std::string &uuid) const;
212 : const ConfigCassandraPartition *GetPartition(int worker_id) const;
213 :
214 : void EnqueueUUIDRequest(std::string oper, std::string obj_type,
215 : std::string uuid_str);
216 :
217 : virtual bool UUIDToObjCacheShow(
218 : const std::string &search_string, int inst_num,
219 : const std::string &last_uuid, uint32_t num_entries,
220 : std::vector<ConfigDBUUIDCacheEntry> *entries) const;
221 : virtual bool IsListOrMapPropEmpty(const string &uuid_key,
222 : const string &lookup_key);
223 : virtual bool IsTaskTriggered() const;
224 : protected:
225 : typedef std::pair<std::string, std::string> ObjTypeUUIDType;
226 : typedef std::list<ObjTypeUUIDType> ObjTypeUUIDList;
227 :
228 : void UpdateFQNameCache(const std::string &key, const std::string &obj_type,
229 : ObjTypeUUIDList &uuid_list);
230 : virtual bool BulkDataSync();
231 : bool EnqueueDBSyncRequest(const ObjTypeUUIDList &uuid_list);
232 : virtual std::string FetchUUIDFromFQNameEntry(const std::string &key) const;
233 :
234 : virtual int HashUUID(const std::string &uuid_str) const;
235 632 : virtual bool SkipTimeStampCheckForTypeAndFQName() const { return true; }
236 0 : virtual uint32_t GetFQNameEntriesToRead() const {
237 0 : return kNumFQNameEntriesToRead;
238 : }
239 : int num_workers() const { return num_workers_; }
240 4212 : PartitionList &partitions() { return partitions_; }
241 : virtual void PostShutdown();
242 :
243 : private:
244 : friend class ConfigCassandraPartition;
245 :
246 : bool InitRetry();
247 :
248 : bool FQNameReader();
249 : bool ParseFQNameRowGetUUIDList(const std::string &obj_type,
250 : const GenDb::ColList &col_list, ObjTypeUUIDList &uuid_list,
251 : std::string *last_column);
252 :
253 : void HandleCassandraConnectionStatus(bool success,
254 : bool force_update = false);
255 :
256 : GenDbIfPtr dbif_;
257 : int num_workers_;
258 : PartitionList partitions_;
259 : boost::scoped_ptr<TaskTrigger> fq_name_reader_;
260 : std::atomic<long> bulk_sync_status_;
261 : };
262 :
263 : #endif // config_cass_client_h
|
