Bug 615 - LCM loses kafka messages
[osm/LCM.git] / osm_lcm / lcm_utils.py
1 # -*- coding: utf-8 -*-
2
3 ##
4 # Copyright 2018 Telefonica S.A.
5 #
6 # Licensed under the Apache License, Version 2.0 (the "License"); you may
7 # not use this file except in compliance with the License. You may obtain
8 # a copy of the License at
9 #
10 # http://www.apache.org/licenses/LICENSE-2.0
11 #
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15 # License for the specific language governing permissions and limitations
16 # under the License.
17 ##
18
19 from collections import OrderedDict
20 from osm_common.dbbase import DbException
21
22 __author__ = "Alfonso Tierno"
23
24
25 class LcmException(Exception):
26 pass
27
28
29 class LcmExceptionNoMgmtIP(LcmException):
30 pass
31
32
33 class LcmExceptionExit(LcmException):
34 pass
35
36
37 def versiontuple(v):
38 """utility for compare dot separate versions. Fills with zeros to proper number comparison
39 package version will be something like 4.0.1.post11+gb3f024d.dirty-1. Where 4.0.1 is the git tag, postXX is the
40 number of commits from this tag, and +XXXXXXX is the git commit short id. Total length is 16 with until 999 commits
41 """
42 filled = []
43 for point in v.split("."):
44 filled.append(point.zfill(16))
45 return tuple(filled)
46
47
48 class TaskRegistry:
49 """
50 Implements a registry of task needed for later cancelation, look for related tasks that must be completed before
51 etc. It stores a four level dict
52 First level is the topic, ns, vim_account, sdn
53 Second level is the _id
54 Third level is the operation id
55 Fourth level is a descriptive name, the value is the task class
56 """
57
58 def __init__(self):
59 self.task_registry = {
60 "ns": {},
61 "nsi": {},
62 "vim_account": {},
63 "wim_account": {},
64 "sdn": {},
65 }
66
67 def register(self, topic, _id, op_id, task_name, task):
68 """
69 Register a new task
70 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
71 :param _id: _id of the related item
72 :param op_id: id of the operation of the related item
73 :param task_name: Task descriptive name, as create, instantiate, terminate. Must be unique in this op_id
74 :param task: Task class
75 :return: none
76 """
77 if _id not in self.task_registry[topic]:
78 self.task_registry[topic][_id] = OrderedDict()
79 if op_id not in self.task_registry[topic][_id]:
80 self.task_registry[topic][_id][op_id] = {task_name: task}
81 else:
82 self.task_registry[topic][_id][op_id][task_name] = task
83 # print("registering task", topic, _id, op_id, task_name, task)
84
85 def remove(self, topic, _id, op_id, task_name=None):
86 """
87 When task is ended, it should removed. It ignores missing tasks
88 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
89 :param _id: _id of the related item
90 :param op_id: id of the operation of the related item
91 :param task_name: Task descriptive name. If note it deletes all
92 :return:
93 """
94 if not self.task_registry[topic].get(_id) or not self.task_registry[topic][_id].get(op_id):
95 return
96 if not task_name:
97 # print("deleting tasks", topic, _id, op_id, self.task_registry[topic][_id][op_id])
98 del self.task_registry[topic][_id][op_id]
99 elif task_name in self.task_registry[topic][_id][op_id]:
100 # print("deleting tasks", topic, _id, op_id, task_name, self.task_registry[topic][_id][op_id][task_name])
101 del self.task_registry[topic][_id][op_id][task_name]
102 if not self.task_registry[topic][_id][op_id]:
103 del self.task_registry[topic][_id][op_id]
104 if not self.task_registry[topic][_id]:
105 del self.task_registry[topic][_id]
106
107 def lookfor_related(self, topic, _id, my_op_id=None):
108 task_list = []
109 task_name_list = []
110 if _id not in self.task_registry[topic]:
111 return "", task_name_list
112 for op_id in reversed(self.task_registry[topic][_id]):
113 if my_op_id:
114 if my_op_id == op_id:
115 my_op_id = None # so that the next task is taken
116 continue
117
118 for task_name, task in self.task_registry[topic][_id][op_id].items():
119 task_list.append(task)
120 task_name_list.append(task_name)
121 break
122 return ", ".join(task_name_list), task_list
123
124 def cancel(self, topic, _id, target_op_id=None, target_task_name=None):
125 """
126 Cancel all active tasks of a concrete ns, nsi, vim_account, sdn identified for _id. If op_id is supplied only
127 this is cancelled, and the same with task_name
128 """
129 if not self.task_registry[topic].get(_id):
130 return
131 for op_id in reversed(self.task_registry[topic][_id]):
132 if target_op_id and target_op_id != op_id:
133 continue
134 for task_name, task in self.task_registry[topic][_id][op_id].items():
135 if target_task_name and target_task_name != task_name:
136 continue
137 # result =
138 task.cancel()
139 # if result:
140 # self.logger.debug("{} _id={} order_id={} task={} cancelled".format(topic, _id, op_id, task_name))
141
142
143 class LcmBase:
144
145 def __init__(self, db, msg, fs, logger):
146 """
147
148 :param db: database connection
149 """
150 self.db = db
151 self.msg = msg
152 self.fs = fs
153 self.logger = logger
154
155 def update_db_2(self, item, _id, _desc):
156 """
157 Updates database with _desc information. Upon success _desc is cleared
158 :param item:
159 :param _id:
160 :param _desc:
161 :return:
162 """
163 if not _desc:
164 return
165 try:
166 self.db.set_one(item, {"_id": _id}, _desc)
167 _desc.clear()
168 except DbException as e:
169 self.logger.error("Updating {} _id={} with '{}'. Error: {}".format(item, _id, _desc, e))