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     def healthcheck(self):
  72         """checks if the sdl connection is healthy"""
  73         return self.sdl.is_active()
  74
  75
  76 SDL = SDLWrapper()
  77
  78 TYPE_PREFIX = "a1.policy_type."
  79 INSTANCE_PREFIX = "a1.policy_instance."
  80 METADATA_PREFIX = "a1.policy_inst_metadata."
  81 HANDLER_PREFIX = "a1.policy_handler."
  82
  83
  84 # Internal helpers
  85
  86
  87 def _generate_type_key(policy_type_id):
  88     """
  89     generate a key for a policy type
  90     """
  91     return "{0}{1}".format(TYPE_PREFIX, policy_type_id)
  92
  93
  94 def _generate_instance_key(policy_type_id, policy_instance_id):
  95     """
  96     generate a key for a policy instance
  97     """
  98     return "{0}{1}.{2}".format(INSTANCE_PREFIX, policy_type_id, policy_instance_id)
  99
 100
 101 def _generate_instance_metadata_key(policy_type_id, policy_instance_id):
 102     """
 103     generate a key for a policy instance metadata
 104     """
 105     return "{0}{1}.{2}".format(METADATA_PREFIX, policy_type_id, policy_instance_id)
 106
 107
 108 def _generate_handler_prefix(policy_type_id, policy_instance_id):
 109     """
 110     generate the prefix to a handler key
 111     """
 112     return "{0}{1}.{2}.".format(HANDLER_PREFIX, policy_type_id, policy_instance_id)
 113
 114
 115 def _generate_handler_key(policy_type_id, policy_instance_id, handler_id):
 116     """
 117     generate a key for a policy handler
 118     """
 119     return "{0}{1}".format(_generate_handler_prefix(policy_type_id, policy_instance_id), handler_id)
 120
 121
 122 def _type_is_valid(policy_type_id):
 123     """
 124     check that a type is valid
 125     """
 126     if SDL.get(_generate_type_key(policy_type_id)) is None:
 127         raise PolicyTypeNotFound()
 128
 129
 130 def _instance_is_valid(policy_type_id, policy_instance_id):
 131     """
 132     check that an instance is valid
 133     """
 134     _type_is_valid(policy_type_id)
 135     if SDL.get(_generate_instance_key(policy_type_id, policy_instance_id)) is None:
 136         raise PolicyInstanceNotFound
 137
 138
 139 def _get_statuses(policy_type_id, policy_instance_id):
 140     """
 141     shared helper to get statuses for an instance
 142     """
 143     _instance_is_valid(policy_type_id, policy_instance_id)
 144     prefixes_for_handler = "{0}{1}.{2}.".format(HANDLER_PREFIX, policy_type_id, policy_instance_id)
 145     return list(SDL.find_and_get(prefixes_for_handler).values())
 146
 147
 148 def _get_instance_list(policy_type_id):
 149     """
 150     shared helper to get instance list for a type
 151     """
 152     _type_is_valid(policy_type_id)
 153     prefixes_for_type = "{0}{1}.".format(INSTANCE_PREFIX, policy_type_id)
 154     instancekeys = SDL.find_and_get(prefixes_for_type).keys()
 155     return [k.split(prefixes_for_type)[1] for k in instancekeys]
 156
 157
 158 def _clear_handlers(policy_type_id, policy_instance_id):
 159     """
 160     delete all the handlers for a policy instance
 161     """
 162     all_handlers_pref = _generate_handler_prefix(policy_type_id, policy_instance_id)
 163     keys = SDL.find_and_get(all_handlers_pref)
 164     for k in keys:
 165         SDL.delete(k)
 166
 167
 168 def _get_metadata(policy_type_id, policy_instance_id):
 169     """
 170     get instance metadata
 171     """
 172     _instance_is_valid(policy_type_id, policy_instance_id)
 173     metadata_key = _generate_instance_metadata_key(policy_type_id, policy_instance_id)
 174     return SDL.get(metadata_key)
 175
 176
 177 def _delete_after(policy_type_id, policy_instance_id, ttl):
 178     """
 179     this is a blocking function, must call this in a thread to not block!
 180     waits ttl seconds, then deletes the instance
 181     """
 182     _instance_is_valid(policy_type_id, policy_instance_id)
 183
 184     time.sleep(ttl)
 185
 186     # ready to delete
 187     _clear_handlers(policy_type_id, policy_instance_id)  # delete all the handlers
 188     SDL.delete(_generate_instance_key(policy_type_id, policy_instance_id))  # delete instance
 189     SDL.delete(_generate_instance_metadata_key(policy_type_id, policy_instance_id))  # delete instance metadata
 190     mdc_logger.debug("type {0} instance {1} deleted".format(policy_type_id, policy_instance_id))
 191
 192
 193 # Types
 194
 195
 196 def get_type_list():
 197     """
 198     retrieve all type ids
 199     """
 200     typekeys = SDL.find_and_get(TYPE_PREFIX).keys()
 201     # policy types are ints but they get butchered to strings in the KV
 202     return [int(k.split(TYPE_PREFIX)[1]) for k in typekeys]
 203
 204
 205 def store_policy_type(policy_type_id, body):
 206     """
 207     store a policy type if it doesn't already exist
 208     """
 209     key = _generate_type_key(policy_type_id)
 210     if SDL.get(key) is not None:
 211         raise PolicyTypeAlreadyExists()
 212     SDL.set(key, body)
 213
 214
 215 def delete_policy_type(policy_type_id):
 216     """
 217     delete a policy type; can only be done if there are no instances (business logic)
 218     """
 219     pil = get_instance_list(policy_type_id)
 220     if pil == []:  # empty, can delete
 221         SDL.delete(_generate_type_key(policy_type_id))
 222     else:
 223         raise CantDeleteNonEmptyType()
 224
 225
 226 def get_policy_type(policy_type_id):
 227     """
 228     retrieve a type
 229     """
 230     _type_is_valid(policy_type_id)
 231     return SDL.get(_generate_type_key(policy_type_id))
 232
 233
 234 # Instances
 235
 236
 237 def store_policy_instance(policy_type_id, policy_instance_id, instance):
 238     """
 239     Store a policy instance
 240     """
 241     _type_is_valid(policy_type_id)
 242     creation_timestamp = time.time()
 243
 244     # store the instance
 245     key = _generate_instance_key(policy_type_id, policy_instance_id)
 246     if SDL.get(key) is not None:
 247         # Reset the statuses because this is a new policy instance, even if it was overwritten
 248         _clear_handlers(policy_type_id, policy_instance_id)  # delete all the handlers
 249     SDL.set(key, instance)
 250
 251     metadata_key = _generate_instance_metadata_key(policy_type_id, policy_instance_id)
 252     SDL.set(metadata_key, {"created_at": creation_timestamp, "has_been_deleted": False})
 253
 254
 255 def get_policy_instance(policy_type_id, policy_instance_id):
 256     """
 257     Retrieve a policy instance
 258     """
 259     _instance_is_valid(policy_type_id, policy_instance_id)
 260     return SDL.get(_generate_instance_key(policy_type_id, policy_instance_id))
 261
 262
 263 def get_instance_list(policy_type_id):
 264     """
 265     retrieve all instance ids for a type
 266     """
 267     return _get_instance_list(policy_type_id)
 268
 269
 270 def delete_policy_instance(policy_type_id, policy_instance_id):
 271     """
 272     initially sets has_been_deleted in the status
 273     then launches a thread that waits until the relevent timer expires, and finally deletes the instance
 274     """
 275     _instance_is_valid(policy_type_id, policy_instance_id)
 276
 277     # set the metadata first
 278     deleted_timestamp = time.time()
 279     metadata_key = _generate_instance_metadata_key(policy_type_id, policy_instance_id)
 280     existing_metadata = _get_metadata(policy_type_id, policy_instance_id)
 281     SDL.set(
 282         metadata_key,
 283         {"created_at": existing_metadata["created_at"], "has_been_deleted": True, "deleted_at": deleted_timestamp},
 284     )
 285
 286     # wait, then delete
 287     vector = _get_statuses(policy_type_id, policy_instance_id)
 288     if vector == []:
 289         # handler is empty; we wait for t1 to expire then goodnight
 290         clos = lambda: _delete_after(policy_type_id, policy_instance_id, INSTANCE_DELETE_NO_RESP_TTL)
 291     else:
 292         # handler is not empty, we wait max t1,t2 to expire then goodnight
 293         clos = lambda: _delete_after(
 294             policy_type_id, policy_instance_id, max(INSTANCE_DELETE_RESP_TTL, INSTANCE_DELETE_NO_RESP_TTL)
 295         )
 296     Thread(target=clos).start()
 297
 298
 299 # Statuses
 300
 301
 302 def set_policy_instance_status(policy_type_id, policy_instance_id, handler_id, status):
 303     """
 304     update the database status for a handler
 305     called from a1's rmr thread
 306     """
 307     _type_is_valid(policy_type_id)
 308     _instance_is_valid(policy_type_id, policy_instance_id)
 309     SDL.set(_generate_handler_key(policy_type_id, policy_instance_id, handler_id), status)
 310
 311
 312 def get_policy_instance_status(policy_type_id, policy_instance_id):
 313     """
 314     Gets the status of an instance
 315     """
 316     _instance_is_valid(policy_type_id, policy_instance_id)
 317     metadata = _get_metadata(policy_type_id, policy_instance_id)
 318     metadata["instance_status"] = "NOT IN EFFECT"
 319     for i in _get_statuses(policy_type_id, policy_instance_id):
 320         if i == "OK":
 321             metadata["instance_status"] = "IN EFFECT"
 322             break
 323     return metadata