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
51 from six
import reraise
52 from six
.moves
import queue
54 from . import wan_link_actions
55 from ..utils
import ensure
, partition
, pipe
56 from .actions
import IGNORE
, PENDING
, REFRESH
60 InvalidParameters
as Invalid
,
63 from .failing_connector
import FailingConnector
64 from .wimconn
import WimConnectorError
65 from .wimconn_dynpac
import DynpacConnector
66 from .wimconn_fake
import FakeConnector
67 from .wimconn_ietfl2vpn
import WimconnectorIETFL2VPN
70 'instance_wim_nets': wan_link_actions
.ACTIONS
74 # "odl": wimconn_odl.OdlConnector,
75 "dynpac": DynpacConnector
,
76 "fake": FakeConnector
,
77 "tapi": WimconnectorIETFL2VPN
,
78 # Add extra connectors here
82 class WimThread(threading
.Thread
):
83 """Specialized task queue implementation that runs in an isolated thread.
85 Objects of this class have a few methods that are intended to be used
86 outside of the thread:
93 All the other methods are used internally to manipulate/process the task
96 RETRY_SCHEDULED
= 10 # 10 seconds
97 REFRESH_BUILD
= 10 # 10 seconds
98 REFRESH_ACTIVE
= 60 # 1 minute
99 BATCH
= 10 # 10 actions per round
101 RECOVERY_TIME
= 5 # Sleep 5s to leave the system some time to recover
102 MAX_RECOVERY_TIME
= 180
103 WAITING_TIME
= 1 # Wait 1s for taks to arrive, when there are none
105 def __init__(self
, persistence
, wim_account
, logger
=None, ovim
=None):
109 persistence: Database abstraction layer
110 wim_account: Record containing wim_account, tenant and wim
113 name
= '{}.{}.{}'.format(wim_account
['wim']['name'],
114 wim_account
['name'], wim_account
['uuid'])
115 super(WimThread
, self
).__init
__(name
=name
)
118 self
.connector
= None
119 self
.wim_account
= wim_account
121 self
.logger
= logger
or logging
.getLogger('openmano.wim.'+self
.name
)
122 self
.persist
= persistence
125 self
.task_queue
= queue
.Queue(self
.QUEUE_SIZE
)
127 self
.refresh_tasks
= []
128 """Time ordered task list for refreshing the status of WIM nets"""
130 self
.pending_tasks
= []
131 """Time ordered task list for creation, deletion of WIM nets"""
133 self
.grouped_tasks
= {}
134 """ It contains all the creation/deletion pending tasks grouped by
135 its concrete vm, net, etc
138 - <task1> # e.g. CREATE task
139 <task2> # e.g. DELETE task
142 self
._insert
_task
= {
143 PENDING
: partial(self
.schedule
, list_name
='pending'),
144 REFRESH
: partial(self
.schedule
, list_name
='refresh'),
145 IGNORE
: lambda task
, *_
, **__
: task
.save(self
.persist
)}
146 """Send the task to the right processing queue"""
149 """Run a series of procedures every time the thread (re)starts"""
150 self
.connector
= self
.get_connector()
151 self
.reload_actions()
153 def get_connector(self
):
154 """Create an WimConnector instance according to the wim.type"""
156 account_id
= self
.wim_account
['uuid']
158 account
= self
.persist
.get_wim_account_by(
159 uuid
=account_id
, hide
=None) # Credentials need to be available
161 mapping
= self
.persist
.query('wim_port_mappings',
162 WHERE
={'wim_id': wim
['uuid']},
164 return CONNECTORS
[wim
['type']](wim
, account
, {
165 'service_endpoint_mapping': mapping
or []
167 except DbBaseException
as ex
:
168 error_msg
= ('Error when retrieving WIM account ({})\n'
169 .format(account_id
)) + str(ex
)
170 self
.logger
.error(error_msg
, exc_info
=True)
171 except KeyError as ex
:
172 error_msg
= ('Unable to find the WIM connector for WIM ({})\n'
173 .format(wim
['type'])) + str(ex
)
174 self
.logger
.error(error_msg
, exc_info
=True)
175 except (WimConnectorError
, Exception) as ex
:
176 # TODO: Remove the Exception class here when the connector class is
178 error_msg
= ('Error when loading WIM connector for WIM ({})\n'
179 .format(wim
['type'])) + str(ex
)
180 self
.logger
.error(error_msg
, exc_info
=True)
182 error_msg_extra
= ('Any task targeting WIM account {} ({}) will fail.'
183 .format(account_id
, self
.wim_account
.get('name')))
184 self
.logger
.warning(error_msg_extra
)
185 return FailingConnector(error_msg
+ '\n' + error_msg_extra
)
188 def avoid_exceptions(self
):
189 """Make a real effort to keep the thread alive, by avoiding the
190 exceptions. They are instead logged as a critical errors.
194 except Exception as ex
:
195 self
.logger
.critical("Unexpected exception %s", ex
, exc_info
=True)
196 sleep(self
.RECOVERY_TIME
)
198 def reload_actions(self
, group_limit
=100):
199 """Read actions from database and reload them at memory.
201 This method will clean and reload the attributes ``refresh_tasks``,
202 ``pending_tasks`` and ``grouped_tasks``
205 group_limit (int): maximum number of action groups (those that
206 refer to the same ``<item, item_id>``) to be retrieved from the
207 database in each batch.
210 # First we clean the cache to let the garbage collector work
211 self
.refresh_tasks
= []
212 self
.pending_tasks
= []
213 self
.grouped_tasks
= {}
218 # Do things in batches
219 task_groups
= self
.persist
.get_actions_in_groups(
220 self
.wim_account
['uuid'], item_types
=('instance_wim_nets',),
221 group_offset
=offset
, group_limit
=group_limit
)
222 offset
+= (group_limit
- 1) # Update for the next batch
227 pending_groups
= (g
for _
, g
in task_groups
if is_pending_group(g
))
229 for task_list
in pending_groups
:
230 with self
.avoid_exceptions():
231 self
.insert_pending_tasks(filter_pending_tasks(task_list
))
234 'Reloaded wim actions pending: %d refresh: %d',
235 len(self
.pending_tasks
), len(self
.refresh_tasks
))
237 def insert_pending_tasks(self
, task_list
):
238 """Insert task in the list of actions being processed"""
239 task_list
= [action_from(task
, self
.logger
) for task
in task_list
]
241 for task
in task_list
:
242 group
= task
.group_key
243 self
.grouped_tasks
.setdefault(group
, [])
244 # Each task can try to supersede the other ones,
245 # but just DELETE actions will actually do
246 task
.supersede(self
.grouped_tasks
[group
])
247 self
.grouped_tasks
[group
].append(task
)
249 # We need a separate loop so each task can check all the other
250 # ones before deciding
251 for task
in task_list
:
252 self
._insert
_task
[task
.processing
](task
)
253 self
.logger
.debug('Insert WIM task: %s (%s): %s %s',
254 task
.id, task
.status
, task
.action
, task
.item
)
256 def schedule(self
, task
, when
=None, list_name
='pending'):
257 """Insert a task in the correct list, respecting the schedule.
258 The refreshing list is ordered by threshold_time (task.process_at)
259 It is assumed that this is called inside this thread
262 task (Action): object representing the task.
263 This object must implement the ``process`` method and inherit
264 from the ``Action`` class
265 list_name: either 'refresh' or 'pending'
266 when (float): unix time in seconds since as a float number
268 processing_list
= {'refresh': self
.refresh_tasks
,
269 'pending': self
.pending_tasks
}[list_name
]
271 when
= when
or time()
272 task
.process_at
= when
274 schedule
= (t
.process_at
for t
in processing_list
)
275 index
= len(list(takewhile(lambda moment
: moment
<= when
, schedule
)))
277 processing_list
.insert(index
, task
)
279 'Schedule of %s in "%s" - waiting position: %d (%f)',
280 task
.id, list_name
, index
, task
.process_at
)
284 def process_list(self
, list_name
='pending'):
285 """Process actions in batches and reschedule them if necessary"""
286 task_list
, handler
= {
287 'refresh': (self
.refresh_tasks
, self
._refresh
_single
),
288 'pending': (self
.pending_tasks
, self
._process
_single
)}[list_name
]
291 waiting
= ((i
, task
) for i
, task
in enumerate(task_list
)
292 if task
.process_at
is None or task
.process_at
<= now
)
294 is_superseded
= pipe(itemgetter(1), attrgetter('is_superseded'))
295 superseded
, active
= partition(is_superseded
, waiting
)
296 superseded
= [(i
, t
.save(self
.persist
)) for i
, t
in superseded
]
298 batch
= islice(active
, self
.BATCH
)
299 refreshed
= [(i
, handler(t
)) for i
, t
in batch
]
301 # Since pop changes the indexes in the list, we need to do it backwards
302 remove
= sorted([i
for i
, _
in chain(refreshed
, superseded
)])
303 return len([task_list
.pop(i
) for i
in reversed(remove
)])
305 def _refresh_single(self
, task
):
306 """Refresh just a single task, and reschedule it if necessary"""
309 result
= task
.refresh(self
.connector
, self
.persist
)
310 self
.logger
.debug('Refreshing WIM task: %s (%s): %s %s => %r',
311 task
.id, task
.status
, task
.action
, task
.item
, result
)
313 interval
= self
.REFRESH_BUILD
if task
.is_build
else self
.REFRESH_ACTIVE
314 self
.schedule(task
, now
+ interval
, 'refresh')
318 def _process_single(self
, task
):
319 """Process just a single task, and reschedule it if necessary"""
322 result
= task
.process(self
.connector
, self
.persist
, self
.ovim
)
323 self
.logger
.debug('Executing WIM task: %s (%s): %s %s => %r',
324 task
.id, task
.status
, task
.action
, task
.item
, result
)
326 if task
.action
== 'DELETE':
327 del self
.grouped_tasks
[task
.group_key
]
329 self
._insert
_task
[task
.processing
](task
, now
+ self
.RETRY_SCHEDULED
)
333 def insert_task(self
, task
):
334 """Send a message to the running thread
336 This function is supposed to be called outside of the WIM Thread.
339 task (str or dict): `"exit"`, `"reload"` or dict representing a
340 task. For more information about the fields in task, please
341 check the Action class.
344 self
.task_queue
.put(task
, False)
347 ex
= QueueFull(self
.name
)
348 reraise(ex
.__class
__, ex
, exc_info()[2])
351 """Send a message to the running thread to reload itself"""
352 self
.insert_task('reload')
355 """Send a message to the running thread to kill itself"""
356 self
.insert_task('exit')
359 self
.logger
.debug('Starting: %s', self
.name
)
363 reload_thread
= False
364 self
.logger
.debug('Reloaded: %s', self
.name
)
367 with self
.avoid_exceptions():
368 while not self
.task_queue
.empty():
369 task
= self
.task_queue
.get()
370 if isinstance(task
, dict):
371 self
.insert_pending_tasks([task
])
372 elif isinstance(task
, list):
373 self
.insert_pending_tasks(task
)
374 elif isinstance(task
, str):
376 self
.logger
.debug('Finishing: %s', self
.name
)
378 elif task
== 'reload':
381 self
.task_queue
.task_done()
386 if not(self
.process_list('pending') +
387 self
.process_list('refresh')):
388 sleep(self
.WAITING_TIME
)
390 if isinstance(self
.connector
, FailingConnector
):
391 # Wait sometime to try instantiating the connector
393 # Increase the recovery time if restarting is not
394 # working (up to a limit)
395 recovery_time
= min(self
.MAX_RECOVERY_TIME
,
396 recovery_time
+ self
.RECOVERY_TIME
)
402 self
.logger
.debug("Finishing")
405 def is_pending_group(group
):
406 return all(task
['action'] != 'DELETE' or
407 task
['status'] == 'SCHEDULED'
411 def filter_pending_tasks(group
):
412 return (t
for t
in group
413 if (t
['status'] == 'SCHEDULED' or
414 t
['action'] in ('CREATE', 'FIND')))
417 def action_from(record
, logger
=None, mapping
=ACTIONS
):
418 """Create an Action object from a action record (dict)
421 mapping (dict): Nested data structure that maps the relationship
422 between action properties and object constructors. This data
423 structure should be a dict with 2 levels of keys: item type and
424 action type. Example::
426 {'CREATE': WanLinkCreate}
429 record (dict): action information
432 (Action.Base): Object representing the action
434 ensure('item' in record
, Invalid('`record` should contain "item"'))
435 ensure('action' in record
, Invalid('`record` should contain "action"'))
438 factory
= mapping
[record
['item']][record
['action']]
439 return factory(record
, logger
=logger
)
441 ex
= UndefinedAction(record
['item'], record
['action'])
442 reraise(ex
.__class
__, ex
, exc_info()[2])