2 Represents A1s database and database access functions.
3 In the future, this may change to use a different backend, possibly dramatically.
4 Hopefully, the access functions are a good api so nothing else has to change when this happens
6 For now, the database is in memory.
7 We use dict data structures (KV) with the expectation of having to move this into Redis
9 # ==================================================================================
10 # Copyright (c) 2019-2020 Nokia
11 # Copyright (c) 2018-2020 AT&T Intellectual Property.
13 # Licensed under the Apache License, Version 2.0 (the "License");
14 # you may not use this file except in compliance with the License.
15 # You may obtain a copy of the License at
17 # http://www.apache.org/licenses/LICENSE-2.0
19 # Unless required by applicable law or agreed to in writing, software
20 # distributed under the License is distributed on an "AS IS" BASIS,
21 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
22 # See the License for the specific language governing permissions and
23 # limitations under the License.
24 # ==================================================================================
27 from threading import Thread
29 from mdclogpy import Logger
31 from ricsdl.syncstorage import SyncStorage
33 from a1.exceptions import PolicyTypeNotFound, PolicyInstanceNotFound, PolicyTypeAlreadyExists, CantDeleteNonEmptyType
35 mdc_logger = Logger(name=__name__)
38 INSTANCE_DELETE_NO_RESP_TTL = int(os.environ.get("INSTANCE_DELETE_NO_RESP_TTL", 5))
39 INSTANCE_DELETE_RESP_TTL = int(os.environ.get("INSTANCE_DELETE_RESP_TTL", 5))
46 This is a wrapper around the expected SDL Python interface.
47 The usage of POLICY_DATA will be replaced with SDL when SDL for python is available.
48 The eventual SDL API is expected to be very close to what is here.
50 We use msgpack for binary (de)serialization: https://msgpack.org/index.html
54 self.sdl = SyncStorage()
56 def set(self, key, value):
58 self.sdl.set(A1NS, {key: msgpack.packb(value, use_bin_type=True)})
62 ret_dict = self.sdl.get(A1NS, {key})
64 return msgpack.unpackb(ret_dict[key], raw=False)
68 def find_and_get(self, prefix):
69 """get all k v pairs that start with prefix"""
70 # note: SDL "*" usage is inconsistent with real python regex, where it would be ".*"
71 ret_dict = self.sdl.find_and_get(A1NS, "{0}*".format(prefix))
72 return {k: msgpack.unpackb(v, raw=False) for k, v in ret_dict.items()}
74 def delete(self, key):
76 self.sdl.remove(A1NS, {key})
81 TYPE_PREFIX = "a1.policy_type."
82 INSTANCE_PREFIX = "a1.policy_instance."
83 METADATA_PREFIX = "a1.policy_inst_metadata."
84 HANDLER_PREFIX = "a1.policy_handler."
90 def _generate_type_key(policy_type_id):
92 generate a key for a policy type
94 return "{0}{1}".format(TYPE_PREFIX, policy_type_id)
97 def _generate_instance_key(policy_type_id, policy_instance_id):
99 generate a key for a policy instance
101 return "{0}{1}.{2}".format(INSTANCE_PREFIX, policy_type_id, policy_instance_id)
104 def _generate_instance_metadata_key(policy_type_id, policy_instance_id):
106 generate a key for a policy instance metadata
108 return "{0}{1}.{2}".format(METADATA_PREFIX, policy_type_id, policy_instance_id)
111 def _generate_handler_prefix(policy_type_id, policy_instance_id):
113 generate the prefix to a handler key
115 return "{0}{1}.{2}.".format(HANDLER_PREFIX, policy_type_id, policy_instance_id)
118 def _generate_handler_key(policy_type_id, policy_instance_id, handler_id):
120 generate a key for a policy handler
122 return "{0}{1}".format(_generate_handler_prefix(policy_type_id, policy_instance_id), handler_id)
125 def _type_is_valid(policy_type_id):
127 check that a type is valid
129 if SDL.get(_generate_type_key(policy_type_id)) is None:
130 raise PolicyTypeNotFound()
133 def _instance_is_valid(policy_type_id, policy_instance_id):
135 check that an instance is valid
137 _type_is_valid(policy_type_id)
138 if SDL.get(_generate_instance_key(policy_type_id, policy_instance_id)) is None:
139 raise PolicyInstanceNotFound
142 def _get_statuses(policy_type_id, policy_instance_id):
144 shared helper to get statuses for an instance
146 _instance_is_valid(policy_type_id, policy_instance_id)
147 prefixes_for_handler = "{0}{1}.{2}.".format(HANDLER_PREFIX, policy_type_id, policy_instance_id)
148 return list(SDL.find_and_get(prefixes_for_handler).values())
151 def _get_instance_list(policy_type_id):
153 shared helper to get instance list for a type
155 _type_is_valid(policy_type_id)
156 prefixes_for_type = "{0}{1}.".format(INSTANCE_PREFIX, policy_type_id)
157 instancekeys = SDL.find_and_get(prefixes_for_type).keys()
158 return [k.split(prefixes_for_type)[1] for k in instancekeys]
161 def _clear_handlers(policy_type_id, policy_instance_id):
163 delete all the handlers for a policy instance
165 all_handlers_pref = _generate_handler_prefix(policy_type_id, policy_instance_id)
166 keys = SDL.find_and_get(all_handlers_pref)
171 def _get_metadata(policy_type_id, policy_instance_id):
173 get instance metadata
175 _instance_is_valid(policy_type_id, policy_instance_id)
176 metadata_key = _generate_instance_metadata_key(policy_type_id, policy_instance_id)
177 return SDL.get(metadata_key)
180 def _delete_after(policy_type_id, policy_instance_id, ttl):
182 this is a blocking function, must call this in a thread to not block!
183 waits ttl seconds, then deletes the instance
185 _instance_is_valid(policy_type_id, policy_instance_id)
190 _clear_handlers(policy_type_id, policy_instance_id) # delete all the handlers
191 SDL.delete(_generate_instance_key(policy_type_id, policy_instance_id)) # delete instance
192 SDL.delete(_generate_instance_metadata_key(policy_type_id, policy_instance_id)) # delete instance metadata
193 mdc_logger.debug("type {0} instance {1} deleted".format(policy_type_id, policy_instance_id))
201 retrieve all type ids
203 typekeys = SDL.find_and_get(TYPE_PREFIX).keys()
204 # policy types are ints but they get butchered to strings in the KV
205 return [int(k.split(TYPE_PREFIX)[1]) for k in typekeys]
208 def store_policy_type(policy_type_id, body):
210 store a policy type if it doesn't already exist
212 key = _generate_type_key(policy_type_id)
213 if SDL.get(key) is not None:
214 raise PolicyTypeAlreadyExists()
218 def delete_policy_type(policy_type_id):
220 delete a policy type; can only be done if there are no instances (business logic)
222 pil = get_instance_list(policy_type_id)
223 if pil == []: # empty, can delete
224 SDL.delete(_generate_type_key(policy_type_id))
226 raise CantDeleteNonEmptyType()
229 def get_policy_type(policy_type_id):
233 _type_is_valid(policy_type_id)
234 return SDL.get(_generate_type_key(policy_type_id))
240 def store_policy_instance(policy_type_id, policy_instance_id, instance):
242 Store a policy instance
244 _type_is_valid(policy_type_id)
245 creation_timestamp = time.time()
248 key = _generate_instance_key(policy_type_id, policy_instance_id)
249 if SDL.get(key) is not None:
250 # Reset the statuses because this is a new policy instance, even if it was overwritten
251 _clear_handlers(policy_type_id, policy_instance_id) # delete all the handlers
252 SDL.set(key, instance)
254 metadata_key = _generate_instance_metadata_key(policy_type_id, policy_instance_id)
255 SDL.set(metadata_key, {"created_at": creation_timestamp, "has_been_deleted": False})
258 def get_policy_instance(policy_type_id, policy_instance_id):
260 Retrieve a policy instance
262 _instance_is_valid(policy_type_id, policy_instance_id)
263 return SDL.get(_generate_instance_key(policy_type_id, policy_instance_id))
266 def get_instance_list(policy_type_id):
268 retrieve all instance ids for a type
270 return _get_instance_list(policy_type_id)
273 def delete_policy_instance(policy_type_id, policy_instance_id):
275 initially sets has_been_deleted
276 then launches a thread that waits until the relevent timer expires, and finally deletes the instance
278 _instance_is_valid(policy_type_id, policy_instance_id)
280 # set the metadata first
281 deleted_timestamp = time.time()
282 metadata_key = _generate_instance_metadata_key(policy_type_id, policy_instance_id)
283 existing_metadata = _get_metadata(policy_type_id, policy_instance_id)
286 {"created_at": existing_metadata["created_at"], "has_been_deleted": True, "deleted_at": deleted_timestamp},
290 vector = _get_statuses(policy_type_id, policy_instance_id)
292 # handler is empty; we wait for t1 to expire then goodnight
293 clos = lambda: _delete_after(policy_type_id, policy_instance_id, INSTANCE_DELETE_NO_RESP_TTL)
295 # handler is not empty, we wait max t1,t2 to expire then goodnight
296 clos = lambda: _delete_after(
297 policy_type_id, policy_instance_id, max(INSTANCE_DELETE_RESP_TTL, INSTANCE_DELETE_NO_RESP_TTL)
299 Thread(target=clos).start()
305 def set_policy_instance_status(policy_type_id, policy_instance_id, handler_id, status):
307 update the database status for a handler
308 called from a1's rmr thread
310 _type_is_valid(policy_type_id)
311 _instance_is_valid(policy_type_id, policy_instance_id)
312 SDL.set(_generate_handler_key(policy_type_id, policy_instance_id, handler_id), status)
315 def get_policy_instance_status(policy_type_id, policy_instance_id):
317 Gets the status of an instance
319 _instance_is_valid(policy_type_id, policy_instance_id)
320 metadata = _get_metadata(policy_type_id, policy_instance_id)
321 metadata["instance_status"] = "NOT IN EFFECT"
322 for i in _get_statuses(policy_type_id, policy_instance_id):
324 metadata["instance_status"] = "IN EFFECT"