1 # -*- coding: utf-8 -*-
3 # Copyright 2018 University of Bristol - High Performance Networks Research
7 # Contributors: Anderson Bravalheri, Dimitrios Gkounis, Abubakar Siddique
8 # Muqaddas, Navdeep Uniyal, Reza Nejabati and Dimitra Simeonidou
10 # Licensed under the Apache License, Version 2.0 (the "License"); you may
11 # not use this file except in compliance with the License. You may obtain
12 # a copy of the License at
14 # http://www.apache.org/licenses/LICENSE-2.0
16 # Unless required by applicable law or agreed to in writing, software
17 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
18 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
19 # License for the specific language governing permissions and limitations
22 # For those usages not covered by the Apache License, Version 2.0 please
23 # contact with: <highperformance-networks@bristol.ac.uk>
25 # Neither the name of the University of Bristol nor the names of its
26 # contributors may be used to endorse or promote products derived from
27 # this software without specific prior written permission.
29 # This work has been performed in the context of DCMS UK 5G Testbeds
30 # & Trials Programme and in the framework of the Metro-Haul project -
31 # funded by the European Commission under Grant number 761727 through the
32 # Horizon 2020 and 5G-PPP programmes.
36 Thread-based interaction with WIMs. Tasks are stored in the
37 database (vim_wim_actions table) and processed sequentially
39 Please check the Action class for information about the content of each action.
44 from contextlib
import contextmanager
45 from functools
import partial
46 from itertools
import islice
, chain
, takewhile
47 from operator
import itemgetter
, attrgetter
48 # from sys import exc_info
49 from time
import time
, sleep
53 from . import wan_link_actions
54 from ..utils
import ensure
, partition
, pipe
55 from .actions
import IGNORE
, PENDING
, REFRESH
59 InvalidParameters
as Invalid
,
62 from osm_ro_plugin
.sdn_failing
import SdnFailingConnector
63 from osm_ro_plugin
.sdnconn
import SdnConnectorError
64 from osm_ro_plugin
.sdn_dummy
import SdnDummyConnector
67 'instance_wim_nets': wan_link_actions
.ACTIONS
71 # "odl": wimconn_odl.OdlConnector,
72 "dummy": SdnDummyConnector
,
73 # Add extra connectors here not managed via plugins
77 class WimThread(threading
.Thread
):
78 """Specialized task queue implementation that runs in an isolated thread.
80 Objects of this class have a few methods that are intended to be used
81 outside of the thread:
88 All the other methods are used internally to manipulate/process the task
91 RETRY_SCHEDULED
= 10 # 10 seconds
92 REFRESH_BUILD
= 10 # 10 seconds
93 REFRESH_ACTIVE
= 60 # 1 minute
94 BATCH
= 10 # 10 actions per round
96 RECOVERY_TIME
= 5 # Sleep 5s to leave the system some time to recover
97 MAX_RECOVERY_TIME
= 180
98 WAITING_TIME
= 1 # Wait 1s for taks to arrive, when there are none
100 def __init__(self
, persistence
, plugins
, wim_account
, logger
=None, ovim
=None):
104 persistence: Database abstraction layer
105 plugins: dictionary with the vim/sdn plugins
106 wim_account: Record containing wim_account, tenant and wim
109 name
= '{}.{}.{}'.format(wim_account
['wim']['name'],
110 wim_account
['name'], wim_account
['uuid'])
111 super(WimThread
, self
).__init
__(name
=name
)
112 self
.plugins
= plugins
113 if "rosdn_dummy" not in self
.plugins
:
114 self
.plugins
["rosdn_dummy"] = SdnDummyConnector
117 self
.connector
= None
118 self
.wim_account
= wim_account
120 self
.logger
= logger
or logging
.getLogger('openmano.wim.'+self
.name
)
121 self
.persist
= persistence
124 self
.task_queue
= queue
.Queue(self
.QUEUE_SIZE
)
126 self
.refresh_tasks
= []
127 """Time ordered task list for refreshing the status of WIM nets"""
129 self
.pending_tasks
= []
130 """Time ordered task list for creation, deletion of WIM nets"""
132 self
.grouped_tasks
= {}
133 """ It contains all the creation/deletion pending tasks grouped by
134 its concrete vm, net, etc
137 - <task1> # e.g. CREATE task
138 <task2> # e.g. DELETE task
141 self
._insert
_task
= {
142 PENDING
: partial(self
.schedule
, list_name
='pending'),
143 REFRESH
: partial(self
.schedule
, list_name
='refresh'),
144 IGNORE
: lambda task
, *_
, **__
: task
.save(self
.persist
)}
145 """Send the task to the right processing queue"""
148 """Run a series of procedures every time the thread (re)starts"""
149 self
.connector
= self
.get_connector()
150 self
.reload_actions()
152 def get_connector(self
):
153 """Create an WimConnector instance according to the wim.type"""
155 account_id
= self
.wim_account
['uuid']
157 account
= self
.persist
.get_wim_account_by(
158 uuid
=account_id
, hide
=None) # Credentials need to be available
160 mapping
= self
.persist
.query('wim_port_mappings',
161 WHERE
={'wim_id': wim
['uuid']},
163 if wim
["type"] in CONNECTORS
:
164 return CONNECTORS
[wim
['type']](wim
, account
, {'service_endpoint_mapping': mapping
or []})
165 else: # load a plugin
166 return self
.plugins
["rosdn_" + wim
["type"]](
167 wim
, account
, {'service_endpoint_mapping': mapping
or []})
168 except DbBaseException
as ex
:
169 error_msg
= ('Error when retrieving WIM account ({})\n'
170 .format(account_id
)) + str(ex
)
171 self
.logger
.error(error_msg
, exc_info
=True)
172 except KeyError as ex
:
173 error_msg
= ('Unable to find the WIM connector for WIM ({})\n'
174 .format(wim
['type'])) + str(ex
)
175 self
.logger
.error(error_msg
)
176 except (SdnConnectorError
, Exception) as ex
:
177 # TODO: Remove the Exception class here when the connector class is
179 error_msg
= ('Error when loading WIM connector for WIM ({})\n'
180 .format(wim
['type'])) + str(ex
)
181 self
.logger
.error(error_msg
, exc_info
=True)
183 error_msg_extra
= ('Any task targeting WIM account {} ({}) will fail.'
184 .format(account_id
, self
.wim_account
.get('name')))
185 self
.logger
.warning(error_msg_extra
)
186 return SdnFailingConnector(error_msg
+ '\n' + error_msg_extra
)
189 def avoid_exceptions(self
):
190 """Make a real effort to keep the thread alive, by avoiding the
191 exceptions. They are instead logged as a critical errors.
195 except Exception as ex
:
196 self
.logger
.critical("Unexpected exception %s", ex
, exc_info
=True)
197 sleep(self
.RECOVERY_TIME
)
199 def reload_actions(self
, group_limit
=100):
200 """Read actions from database and reload them at memory.
202 This method will clean and reload the attributes ``refresh_tasks``,
203 ``pending_tasks`` and ``grouped_tasks``
206 group_limit (int): maximum number of action groups (those that
207 refer to the same ``<item, item_id>``) to be retrieved from the
208 database in each batch.
211 # First we clean the cache to let the garbage collector work
212 self
.refresh_tasks
= []
213 self
.pending_tasks
= []
214 self
.grouped_tasks
= {}
219 # Do things in batches
220 task_groups
= self
.persist
.get_actions_in_groups(
221 self
.wim_account
['uuid'], item_types
=('instance_wim_nets',),
222 group_offset
=offset
, group_limit
=group_limit
)
223 offset
+= (group_limit
- 1) # Update for the next batch
228 pending_groups
= (g
for _
, g
in task_groups
if is_pending_group(g
))
230 for task_list
in pending_groups
:
231 with self
.avoid_exceptions():
232 self
.insert_pending_tasks(filter_pending_tasks(task_list
))
235 'Reloaded wim actions pending: %d refresh: %d',
236 len(self
.pending_tasks
), len(self
.refresh_tasks
))
238 def insert_pending_tasks(self
, task_list
):
239 """Insert task in the list of actions being processed"""
240 task_list
= [action_from(task
, self
.logger
) for task
in task_list
]
242 for task
in task_list
:
243 group
= task
.group_key
244 self
.grouped_tasks
.setdefault(group
, [])
245 # Each task can try to supersede the other ones,
246 # but just DELETE actions will actually do
247 task
.supersede(self
.grouped_tasks
[group
])
248 self
.grouped_tasks
[group
].append(task
)
250 # We need a separate loop so each task can check all the other
251 # ones before deciding
252 for task
in task_list
:
253 self
._insert
_task
[task
.processing
](task
)
254 self
.logger
.debug('Insert WIM task: %s (%s): %s %s',
255 task
.id, task
.status
, task
.action
, task
.item
)
257 def schedule(self
, task
, when
=None, list_name
='pending'):
258 """Insert a task in the correct list, respecting the schedule.
259 The refreshing list is ordered by threshold_time (task.process_at)
260 It is assumed that this is called inside this thread
263 task (Action): object representing the task.
264 This object must implement the ``process`` method and inherit
265 from the ``Action`` class
266 list_name: either 'refresh' or 'pending'
267 when (float): unix time in seconds since as a float number
269 processing_list
= {'refresh': self
.refresh_tasks
,
270 'pending': self
.pending_tasks
}[list_name
]
272 when
= when
or time()
273 task
.process_at
= when
275 schedule
= (t
.process_at
for t
in processing_list
)
276 index
= len(list(takewhile(lambda moment
: moment
<= when
, schedule
)))
278 processing_list
.insert(index
, task
)
280 'Schedule of %s in "%s" - waiting position: %d (%f)',
281 task
.id, list_name
, index
, task
.process_at
)
285 def process_list(self
, list_name
='pending'):
286 """Process actions in batches and reschedule them if necessary"""
287 task_list
, handler
= {
288 'refresh': (self
.refresh_tasks
, self
._refresh
_single
),
289 'pending': (self
.pending_tasks
, self
._process
_single
)}[list_name
]
292 waiting
= ((i
, task
) for i
, task
in enumerate(task_list
)
293 if task
.process_at
is None or task
.process_at
<= now
)
295 is_superseded
= pipe(itemgetter(1), attrgetter('is_superseded'))
296 superseded
, active
= partition(is_superseded
, waiting
)
297 superseded
= [(i
, t
.save(self
.persist
)) for i
, t
in superseded
]
299 batch
= islice(active
, self
.BATCH
)
300 refreshed
= [(i
, handler(t
)) for i
, t
in batch
]
302 # Since pop changes the indexes in the list, we need to do it backwards
303 remove
= sorted([i
for i
, _
in chain(refreshed
, superseded
)])
304 return len([task_list
.pop(i
) for i
in reversed(remove
)])
306 def _refresh_single(self
, task
):
307 """Refresh just a single task, and reschedule it if necessary"""
310 result
= task
.refresh(self
.connector
, self
.persist
)
311 self
.logger
.debug('Refreshing WIM task: %s (%s): %s %s => %r',
312 task
.id, task
.status
, task
.action
, task
.item
, result
)
314 interval
= self
.REFRESH_BUILD
if task
.is_build
else self
.REFRESH_ACTIVE
315 self
.schedule(task
, now
+ interval
, 'refresh')
319 def _process_single(self
, task
):
320 """Process just a single task, and reschedule it if necessary"""
323 result
= task
.process(self
.connector
, self
.persist
, self
.ovim
)
324 self
.logger
.debug('Executing WIM task: %s (%s): %s %s => %r',
325 task
.id, task
.status
, task
.action
, task
.item
, result
)
327 if task
.action
== 'DELETE':
328 del self
.grouped_tasks
[task
.group_key
]
330 self
._insert
_task
[task
.processing
](task
, now
+ self
.RETRY_SCHEDULED
)
334 def insert_task(self
, task
):
335 """Send a message to the running thread
337 This function is supposed to be called outside of the WIM Thread.
340 task (str or dict): `"exit"`, `"reload"` or dict representing a
341 task. For more information about the fields in task, please
342 check the Action class.
345 self
.task_queue
.put(task
, False)
347 except queue
.Full
as e
:
348 ex
= QueueFull(self
.name
)
352 """Send a message to the running thread to reload itself"""
353 self
.insert_task('reload')
356 """Send a message to the running thread to kill itself"""
357 self
.insert_task('exit')
360 self
.logger
.debug('Starting: %s', self
.name
)
364 reload_thread
= False
365 self
.logger
.debug('Reloaded: %s', self
.name
)
368 with self
.avoid_exceptions():
369 while not self
.task_queue
.empty():
370 task
= self
.task_queue
.get()
371 if isinstance(task
, dict):
372 self
.insert_pending_tasks([task
])
373 elif isinstance(task
, list):
374 self
.insert_pending_tasks(task
)
375 elif isinstance(task
, str):
377 self
.logger
.debug('Finishing: %s', self
.name
)
379 elif task
== 'reload':
382 self
.task_queue
.task_done()
387 if not(self
.process_list('pending') +
388 self
.process_list('refresh')):
389 sleep(self
.WAITING_TIME
)
391 if isinstance(self
.connector
, SdnFailingConnector
):
392 # Wait sometime to try instantiating the connector
394 # Increase the recovery time if restarting is not
395 # working (up to a limit)
396 recovery_time
= min(self
.MAX_RECOVERY_TIME
,
397 recovery_time
+ self
.RECOVERY_TIME
)
403 self
.logger
.debug("Finishing")
406 def is_pending_group(group
):
407 return all(task
['action'] != 'DELETE' or
408 task
['status'] == 'SCHEDULED'
412 def filter_pending_tasks(group
):
413 return (t
for t
in group
414 if (t
['status'] == 'SCHEDULED' or
415 t
['action'] in ('CREATE', 'FIND')))
418 def action_from(record
, logger
=None, mapping
=ACTIONS
):
419 """Create an Action object from a action record (dict)
422 mapping (dict): Nested data structure that maps the relationship
423 between action properties and object constructors. This data
424 structure should be a dict with 2 levels of keys: item type and
425 action type. Example::
427 {'CREATE': WanLinkCreate}
430 record (dict): action information
433 (Action.Base): Object representing the action
435 ensure('item' in record
, Invalid('`record` should contain "item"'))
436 ensure('action' in record
, Invalid('`record` should contain "action"'))
439 factory
= mapping
[record
['item']][record
['action']]
440 return factory(record
, logger
=logger
)
441 except KeyError as e
:
442 ex
= UndefinedAction(record
['item'], record
['action'])