a1/data.py

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