a1/data.py

   1 """
   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
   5
   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
   8 """
   9 # ==================================================================================
  10 #       Copyright (c) 2019-2020 Nokia
  11 #       Copyright (c) 2018-2020 AT&T Intellectual Property.
  12 #
  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
  16 #
  17 #          http://www.apache.org/licenses/LICENSE-2.0
  18 #
  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 # ==================================================================================
  25 import os
  26 import time
  27 from threading import Thread
  28 import msgpack
  29 from mdclogpy import Logger
  30
  31 from ricsdl.syncstorage import SyncStorage
  32
  33 from a1.exceptions import PolicyTypeNotFound, PolicyInstanceNotFound, PolicyTypeAlreadyExists, CantDeleteNonEmptyType
  34
  35 mdc_logger = Logger(name=__name__)
  36
  37
  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))
  40
  41 A1NS = "A1m_ns"
  42
  43
  44 class SDLWrapper:
  45     """
  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.
  49
  50     We use msgpack for binary (de)serialization: https://msgpack.org/index.html
  51     """
  52
  53     def __init__(self):
  54         self.sdl = SyncStorage()
  55
  56     def set(self, key, value):
  57         """set a key"""
  58         self.sdl.set(A1NS, {key: msgpack.packb(value, use_bin_type=True)})
  59
  60     def get(self, key):
  61         """get a key"""
  62         ret_dict = self.sdl.get(A1NS, {key})
  63         if key in ret_dict:
  64             return msgpack.unpackb(ret_dict[key], raw=False)
  65
  66         return None
  67
  68     def find_and_get(self, prefix):
  69         """get all k v pairs that start with prefix"""
  70         ret_dict = self.sdl.find_and_get(A1NS, "{0}".format(prefix), atomic=True)
  71         found = {k: msgpack.unpackb(v, raw=False) for k, v in ret_dict.items()}
  72         # TODO: upgrade to sdl 2.0.0 which does the sorting for us
  73         return {k: found[k] for k in sorted(found)}
  74
  75     def delete(self, key):
  76         """ delete a key"""
  77         self.sdl.remove(A1NS, {key})
  78
  79
  80 SDL = SDLWrapper()
  81
  82 TYPE_PREFIX = "a1.policy_type."
  83 INSTANCE_PREFIX = "a1.policy_instance."
  84 METADATA_PREFIX = "a1.policy_inst_metadata."
  85 HANDLER_PREFIX = "a1.policy_handler."
  86
  87
  88 # Internal helpers
  89
  90
  91 def _generate_type_key(policy_type_id):
  92     """
  93     generate a key for a policy type
  94     """
  95     return "{0}{1}".format(TYPE_PREFIX, policy_type_id)
  96
  97
  98 def _generate_instance_key(policy_type_id, policy_instance_id):
  99     """
 100     generate a key for a policy instance
 101     """
 102     return "{0}{1}.{2}".format(INSTANCE_PREFIX, policy_type_id, policy_instance_id)
 103
 104
 105 def _generate_instance_metadata_key(policy_type_id, policy_instance_id):
 106     """
 107     generate a key for a policy instance metadata
 108     """
 109     return "{0}{1}.{2}".format(METADATA_PREFIX, policy_type_id, policy_instance_id)
 110
 111
 112 def _generate_handler_prefix(policy_type_id, policy_instance_id):
 113     """
 114     generate the prefix to a handler key
 115     """
 116     return "{0}{1}.{2}.".format(HANDLER_PREFIX, policy_type_id, policy_instance_id)
 117
 118
 119 def _generate_handler_key(policy_type_id, policy_instance_id, handler_id):
 120     """
 121     generate a key for a policy handler
 122     """
 123     return "{0}{1}".format(_generate_handler_prefix(policy_type_id, policy_instance_id), handler_id)
 124
 125
 126 def _type_is_valid(policy_type_id):
 127     """
 128     check that a type is valid
 129     """
 130     if SDL.get(_generate_type_key(policy_type_id)) is None:
 131         raise PolicyTypeNotFound()
 132
 133
 134 def _instance_is_valid(policy_type_id, policy_instance_id):
 135     """
 136     check that an instance is valid
 137     """
 138     _type_is_valid(policy_type_id)
 139     if SDL.get(_generate_instance_key(policy_type_id, policy_instance_id)) is None:
 140         raise PolicyInstanceNotFound
 141
 142
 143 def _get_statuses(policy_type_id, policy_instance_id):
 144     """
 145     shared helper to get statuses for an instance
 146     """
 147     _instance_is_valid(policy_type_id, policy_instance_id)
 148     prefixes_for_handler = "{0}{1}.{2}.".format(HANDLER_PREFIX, policy_type_id, policy_instance_id)
 149     return list(SDL.find_and_get(prefixes_for_handler).values())
 150
 151
 152 def _get_instance_list(policy_type_id):
 153     """
 154     shared helper to get instance list for a type
 155     """
 156     _type_is_valid(policy_type_id)
 157     prefixes_for_type = "{0}{1}.".format(INSTANCE_PREFIX, policy_type_id)
 158     instancekeys = SDL.find_and_get(prefixes_for_type).keys()
 159     return [k.split(prefixes_for_type)[1] for k in instancekeys]
 160
 161
 162 def _clear_handlers(policy_type_id, policy_instance_id):
 163     """
 164     delete all the handlers for a policy instance
 165     """
 166     all_handlers_pref = _generate_handler_prefix(policy_type_id, policy_instance_id)
 167     keys = SDL.find_and_get(all_handlers_pref)
 168     for k in keys:
 169         SDL.delete(k)
 170
 171
 172 def _get_metadata(policy_type_id, policy_instance_id):
 173     """
 174     get instance metadata
 175     """
 176     _instance_is_valid(policy_type_id, policy_instance_id)
 177     metadata_key = _generate_instance_metadata_key(policy_type_id, policy_instance_id)
 178     return SDL.get(metadata_key)
 179
 180
 181 def _delete_after(policy_type_id, policy_instance_id, ttl):
 182     """
 183     this is a blocking function, must call this in a thread to not block!
 184     waits ttl seconds, then deletes the instance
 185     """
 186     _instance_is_valid(policy_type_id, policy_instance_id)
 187
 188     time.sleep(ttl)
 189
 190     # ready to delete
 191     _clear_handlers(policy_type_id, policy_instance_id)  # delete all the handlers
 192     SDL.delete(_generate_instance_key(policy_type_id, policy_instance_id))  # delete instance
 193     SDL.delete(_generate_instance_metadata_key(policy_type_id, policy_instance_id))  # delete instance metadata
 194     mdc_logger.debug("type {0} instance {1} deleted".format(policy_type_id, policy_instance_id))
 195
 196
 197 # Types
 198
 199
 200 def get_type_list():
 201     """
 202     retrieve all type ids
 203     """
 204     typekeys = SDL.find_and_get(TYPE_PREFIX).keys()
 205     # policy types are ints but they get butchered to strings in the KV
 206     return [int(k.split(TYPE_PREFIX)[1]) for k in typekeys]
 207
 208
 209 def store_policy_type(policy_type_id, body):
 210     """
 211     store a policy type if it doesn't already exist
 212     """
 213     key = _generate_type_key(policy_type_id)
 214     if SDL.get(key) is not None:
 215         raise PolicyTypeAlreadyExists()
 216     SDL.set(key, body)
 217
 218
 219 def delete_policy_type(policy_type_id):
 220     """
 221     delete a policy type; can only be done if there are no instances (business logic)
 222     """
 223     pil = get_instance_list(policy_type_id)
 224     if pil == []:  # empty, can delete
 225         SDL.delete(_generate_type_key(policy_type_id))
 226     else:
 227         raise CantDeleteNonEmptyType()
 228
 229
 230 def get_policy_type(policy_type_id):
 231     """
 232     retrieve a type
 233     """
 234     _type_is_valid(policy_type_id)
 235     return SDL.get(_generate_type_key(policy_type_id))
 236
 237
 238 # Instances
 239
 240
 241 def store_policy_instance(policy_type_id, policy_instance_id, instance):
 242     """
 243     Store a policy instance
 244     """
 245     _type_is_valid(policy_type_id)
 246     creation_timestamp = time.time()
 247
 248     # store the instance
 249     key = _generate_instance_key(policy_type_id, policy_instance_id)
 250     if SDL.get(key) is not None:
 251         # Reset the statuses because this is a new policy instance, even if it was overwritten
 252         _clear_handlers(policy_type_id, policy_instance_id)  # delete all the handlers
 253     SDL.set(key, instance)
 254
 255     metadata_key = _generate_instance_metadata_key(policy_type_id, policy_instance_id)
 256     SDL.set(metadata_key, {"created_at": creation_timestamp, "has_been_deleted": False})
 257
 258
 259 def get_policy_instance(policy_type_id, policy_instance_id):
 260     """
 261     Retrieve a policy instance
 262     """
 263     _instance_is_valid(policy_type_id, policy_instance_id)
 264     return SDL.get(_generate_instance_key(policy_type_id, policy_instance_id))
 265
 266
 267 def get_instance_list(policy_type_id):
 268     """
 269     retrieve all instance ids for a type
 270     """
 271     return _get_instance_list(policy_type_id)
 272
 273
 274 def delete_policy_instance(policy_type_id, policy_instance_id):
 275     """
 276     initially sets has_been_deleted
 277     then launches a thread that waits until the relevent timer expires, and finally deletes the instance
 278     """
 279     _instance_is_valid(policy_type_id, policy_instance_id)
 280
 281     # set the metadata first
 282     deleted_timestamp = time.time()
 283     metadata_key = _generate_instance_metadata_key(policy_type_id, policy_instance_id)
 284     existing_metadata = _get_metadata(policy_type_id, policy_instance_id)
 285     SDL.set(
 286         metadata_key,
 287         {"created_at": existing_metadata["created_at"], "has_been_deleted": True, "deleted_at": deleted_timestamp},
 288     )
 289
 290     # wait, then delete
 291     vector = _get_statuses(policy_type_id, policy_instance_id)
 292     if vector == []:
 293         # handler is empty; we wait for t1 to expire then goodnight
 294         clos = lambda: _delete_after(policy_type_id, policy_instance_id, INSTANCE_DELETE_NO_RESP_TTL)
 295     else:
 296         # handler is not empty, we wait max t1,t2 to expire then goodnight
 297         clos = lambda: _delete_after(
 298             policy_type_id, policy_instance_id, max(INSTANCE_DELETE_RESP_TTL, INSTANCE_DELETE_NO_RESP_TTL)
 299         )
 300     Thread(target=clos).start()
 301
 302
 303 # Statuses
 304
 305
 306 def set_policy_instance_status(policy_type_id, policy_instance_id, handler_id, status):
 307     """
 308     update the database status for a handler
 309     called from a1's rmr thread
 310     """
 311     _type_is_valid(policy_type_id)
 312     _instance_is_valid(policy_type_id, policy_instance_id)
 313     SDL.set(_generate_handler_key(policy_type_id, policy_instance_id, handler_id), status)
 314
 315
 316 def get_policy_instance_status(policy_type_id, policy_instance_id):
 317     """
 318     Gets the status of an instance
 319     """
 320     _instance_is_valid(policy_type_id, policy_instance_id)
 321     metadata = _get_metadata(policy_type_id, policy_instance_id)
 322     metadata["instance_status"] = "NOT IN EFFECT"
 323     for i in _get_statuses(policy_type_id, policy_instance_id):
 324         if i == "OK":
 325             metadata["instance_status"] = "IN EFFECT"
 326             break
 327     return metadata