Test Failed
Pull Request — master (#411)
by
unknown
03:40
created

build.models.evc.EVCBase.shares_uni()   A

Complexity

Conditions 3

Size

Total Lines 8
Code Lines 6

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 1
CRAP Score 6.7968

Importance

Changes 0
Metric Value
cc 3
eloc 6
nop 2
dl 0
loc 8
ccs 1
cts 4
cp 0.25
crap 6.7968
rs 10
c 0
b 0
f 0
1
"""Classes used in the main application."""  # pylint: disable=too-many-lines
2 1
import traceback
3 1
from collections import OrderedDict
4 1
from copy import deepcopy
5 1
from datetime import datetime
6 1
from operator import eq, ne
7 1
from threading import Lock
8 1
from typing import Union
9 1
from uuid import uuid4
10
11 1
import requests
12 1
from glom import glom
13 1
from requests.exceptions import Timeout
14
15 1
from kytos.core import log
16 1
from kytos.core.common import EntityStatus, GenericEntity
17 1
from kytos.core.exceptions import KytosNoTagAvailableError, KytosTagError
18 1
from kytos.core.helpers import get_time, now
19 1
from kytos.core.interface import UNI, Interface, TAGRange
20 1
from kytos.core.link import Link
21 1
from kytos.core.tag_ranges import range_difference
22 1
from napps.kytos.mef_eline import controllers, settings
23 1
from napps.kytos.mef_eline.exceptions import FlowModException, InvalidPath, DuplicatedNoTagUNI
24 1
from napps.kytos.mef_eline.utils import (check_disabled_component,
25
                                         compare_endpoint_trace,
26
                                         compare_uni_out_trace, emit_event,
27
                                         make_uni_list, map_dl_vlan,
28
                                         map_evc_event_content)
29
30 1
from .path import DynamicPathManager, Path
31
32
33 1
class EVCBase(GenericEntity):
34
    """Class to represent a circuit."""
35
36 1
    read_only_attributes = [
37
        "creation_time",
38
        "active",
39
        "current_path",
40
        "failover_path",
41
        "_id",
42
        "archived",
43
    ]
44 1
    attributes_requiring_redeploy = [
45
        "primary_path",
46
        "backup_path",
47
        "dynamic_backup_path",
48
        "queue_id",
49
        "sb_priority",
50
        "primary_constraints",
51
        "secondary_constraints",
52
        "uni_a",
53
        "uni_z",
54
    ]
55 1
    required_attributes = ["name", "uni_a", "uni_z"]
56
57 1
    def __init__(self, controller, **kwargs):
58
        """Create an EVC instance with the provided parameters.
59
60
        Args:
61
            id(str): EVC identifier. Whether it's None an ID will be genereted.
62
                     Only the first 14 bytes passed will be used.
63
            name: represents an EVC name.(Required)
64
            uni_a (UNI): Endpoint A for User Network Interface.(Required)
65
            uni_z (UNI): Endpoint Z for User Network Interface.(Required)
66
            start_date(datetime|str): Date when the EVC was registred.
67
                                      Default is now().
68
            end_date(datetime|str): Final date that the EVC will be fineshed.
69
                                    Default is None.
70
            bandwidth(int): Bandwidth used by EVC instance. Default is 0.
71
            primary_links(list): Primary links used by evc. Default is []
72
            backup_links(list): Backups links used by evc. Default is []
73
            current_path(list): Circuit being used at the moment if this is an
74
                                active circuit. Default is [].
75
            failover_path(list): Path being used to provide EVC protection via
76
                                failover during link failures. Default is [].
77
            primary_path(list): primary circuit offered to user IF one or more
78
                                links were provided. Default is [].
79
            backup_path(list): backup circuit offered to the user IF one or
80
                               more links were provided. Default is [].
81
            dynamic_backup_path(bool): Enable computer backup path dynamically.
82
                                       Dafault is False.
83
            creation_time(datetime|str): datetime when the circuit should be
84
                                         activated. default is now().
85
            enabled(Boolean): attribute to indicate the administrative state;
86
                              default is False.
87
            active(Boolean): attribute to indicate the operational state;
88
                             default is False.
89
            archived(Boolean): indicate the EVC has been deleted and is
90
                               archived; default is False.
91
            owner(str): The EVC owner. Default is None.
92
            sb_priority(int): Service level provided in the request.
93
                              Default is None.
94
            service_level(int): Service level provided. The higher the better.
95
                                Default is 0.
96
97
        Raises:
98
            ValueError: raised when object attributes are invalid.
99
100
        """
101 1
        self._controller = controller
102 1
        self._validate(**kwargs)
103 1
        super().__init__()
104
105
        # required attributes
106 1
        self._id = kwargs.get("id", uuid4().hex)[:14]
107 1
        self.uni_a: UNI = kwargs.get("uni_a")
108 1
        self.uni_z: UNI = kwargs.get("uni_z")
109 1
        self.name = kwargs.get("name")
110
111
        # optional attributes
112 1
        self.start_date = get_time(kwargs.get("start_date")) or now()
113 1
        self.end_date = get_time(kwargs.get("end_date")) or None
114 1
        self.queue_id = kwargs.get("queue_id", -1)
115
116 1
        self.bandwidth = kwargs.get("bandwidth", 0)
117 1
        self.primary_links = Path(kwargs.get("primary_links", []))
118 1
        self.backup_links = Path(kwargs.get("backup_links", []))
119 1
        self.current_path = Path(kwargs.get("current_path", []))
120 1
        self.failover_path = Path(kwargs.get("failover_path", []))
121 1
        self.primary_path = Path(kwargs.get("primary_path", []))
122 1
        self.backup_path = Path(kwargs.get("backup_path", []))
123 1
        self.dynamic_backup_path = kwargs.get("dynamic_backup_path", False)
124 1
        self.primary_constraints = kwargs.get("primary_constraints", {})
125 1
        self.secondary_constraints = kwargs.get("secondary_constraints", {})
126 1
        self.creation_time = get_time(kwargs.get("creation_time")) or now()
127 1
        self.owner = kwargs.get("owner", None)
128 1
        self.sb_priority = kwargs.get("sb_priority", None) or kwargs.get(
129
            "priority", None
130
        )
131 1
        self.service_level = kwargs.get("service_level", 0)
132 1
        self.circuit_scheduler = kwargs.get("circuit_scheduler", [])
133 1
        self.flow_removed_at = get_time(kwargs.get("flow_removed_at")) or None
134 1
        self.updated_at = get_time(kwargs.get("updated_at")) or now()
135 1
        self.execution_rounds = kwargs.get("execution_rounds", 0)
136
137 1
        self.current_links_cache = set()
138 1
        self.primary_links_cache = set()
139 1
        self.backup_links_cache = set()
140
141 1
        self.lock = Lock()
142
143 1
        self.archived = kwargs.get("archived", False)
144
145 1
        self.metadata = kwargs.get("metadata", {})
146
147 1
        self._mongo_controller = controllers.ELineController()
148
149 1
        if kwargs.get("active", False):
150 1
            self.activate()
151
        else:
152 1
            self.deactivate()
153
154 1
        if kwargs.get("enabled", False):
155 1
            self.enable()
156
        else:
157 1
            self.disable()
158
159
        # datetime of user request for a EVC (or datetime when object was
160
        # created)
161 1
        self.request_time = kwargs.get("request_time", now())
162
        # dict with the user original request (input)
163 1
        self._requested = kwargs
164
165
        # Special cases: No tag, any, untagged
166 1
        self.special_cases = {None, "4096/4096", 0}
167 1
        self.table_group = kwargs.get("table_group")
168
169 1
    def sync(self, keys: set = None):
170
        """Sync this EVC in the MongoDB."""
171 1
        self.updated_at = now()
172 1
        if keys:
173 1
            self._mongo_controller.update_evc(self.as_dict(keys))
174 1
            return
175 1
        self._mongo_controller.upsert_evc(self.as_dict())
176
177 1
    def _get_unis_use_tags(self, **kwargs) -> tuple[UNI, UNI]:
178
        """Obtain both UNIs (uni_a, uni_z).
179
        If a UNI is changing, verify tags"""
180 1
        uni_a = kwargs.get("uni_a", None)
181 1
        uni_a_flag = False
182 1
        if uni_a and uni_a != self.uni_a:
183 1
            uni_a_flag = True
184 1
            self._use_uni_vlan(uni_a, uni_dif=self.uni_a)
185
186 1
        uni_z = kwargs.get("uni_z", None)
187 1
        if uni_z and uni_z != self.uni_z:
188 1
            try:
189 1
                self._use_uni_vlan(uni_z, uni_dif=self.uni_z)
190 1
                self.make_uni_vlan_available(self.uni_z, uni_dif=uni_z)
191 1
            except KytosTagError as err:
192 1
                if uni_a_flag:
193 1
                    self.make_uni_vlan_available(uni_a, uni_dif=self.uni_a)
194 1
                raise err
195
        else:
196 1
            uni_z = self.uni_z
197
198 1
        if uni_a_flag:
199 1
            self.make_uni_vlan_available(self.uni_a, uni_dif=uni_a)
200
        else:
201 1
            uni_a = self.uni_a
202 1
        return uni_a, uni_z
203
204 1
    def update(self, **kwargs):
205
        """Update evc attributes.
206
207
        This method will raises an error trying to change the following
208
        attributes: [creation_time, active, current_path, failover_path,
209
        _id, archived]
210
        [name, uni_a and uni_z]
211
212
        Returns:
213
            the values for enable and a redeploy attribute, if exists and None
214
            otherwise
215
        Raises:
216
            ValueError: message with error detail.
217
218
        """
219 1
        enable, redeploy = (None, None)
220 1
        if not self._tag_lists_equal(**kwargs):
221 1
            raise ValueError(
222
                "UNI_A and UNI_Z tag lists should be the same."
223
            )
224 1
        uni_a, uni_z = self._get_unis_use_tags(**kwargs)
225 1
        check_disabled_component(uni_a, uni_z)
226 1
        self._validate_has_primary_or_dynamic(
227
            primary_path=kwargs.get("primary_path"),
228
            dynamic_backup_path=kwargs.get("dynamic_backup_path"),
229
            uni_a=uni_a,
230
            uni_z=uni_z,
231
        )
232 1
        for attribute, value in kwargs.items():
233 1
            if attribute in self.read_only_attributes:
234 1
                raise ValueError(f"{attribute} can't be updated.")
235 1
            if not hasattr(self, attribute):
236 1
                raise ValueError(f'The attribute "{attribute}" is invalid.')
237 1
            if attribute in ("primary_path", "backup_path"):
238 1
                try:
239 1
                    value.is_valid(
240
                        uni_a.interface.switch, uni_z.interface.switch
241
                    )
242 1
                except InvalidPath as exception:
243 1
                    raise ValueError(  # pylint: disable=raise-missing-from
244
                        f"{attribute} is not a " f"valid path: {exception}"
245
                    )
246 1
        for attribute, value in kwargs.items():
247 1
            if attribute in ("enable", "enabled"):
248 1
                if value:
249 1
                    self.enable()
250
                else:
251 1
                    self.disable()
252 1
                enable = value
253
            else:
254 1
                setattr(self, attribute, value)
255 1
                if attribute in self.attributes_requiring_redeploy:
256 1
                    redeploy = True
257 1
        self.sync(set(kwargs.keys()))
258 1
        return enable, redeploy
259
260 1
    def set_flow_removed_at(self):
261
        """Update flow_removed_at attribute."""
262
        self.flow_removed_at = now()
263
264 1
    def has_recent_removed_flow(self, setting=settings):
265
        """Check if any flow has been removed from the evc"""
266
        if self.flow_removed_at is None:
267
            return False
268
        res_seconds = (now() - self.flow_removed_at).seconds
269
        return res_seconds < setting.TIME_RECENT_DELETED_FLOWS
270
271 1
    def is_recent_updated(self, setting=settings):
272
        """Check if the evc has been updated recently"""
273
        res_seconds = (now() - self.updated_at).seconds
274
        return res_seconds < setting.TIME_RECENT_UPDATED
275
276 1
    def __repr__(self):
277
        """Repr method."""
278 1
        return f"EVC({self._id}, {self.name})"
279
280 1
    def _validate(self, **kwargs):
281
        """Do Basic validations.
282
283
        Verify required attributes: name, uni_a, uni_z
284
285
        Raises:
286
            ValueError: message with error detail.
287
288
        """
289 1
        for attribute in self.required_attributes:
290
291 1
            if attribute not in kwargs:
292 1
                raise ValueError(f"{attribute} is required.")
293
294 1
            if "uni" in attribute:
295 1
                uni = kwargs.get(attribute)
296 1
                if not isinstance(uni, UNI):
297
                    raise ValueError(f"{attribute} is an invalid UNI.")
298
299 1
    def _tag_lists_equal(self, **kwargs):
300
        """Verify that tag lists are the same."""
301 1
        uni_a = kwargs.get("uni_a") or self.uni_a
302 1
        uni_z = kwargs.get("uni_z") or self.uni_z
303 1
        uni_a_list = uni_z_list = False
304 1
        if (uni_a.user_tag and isinstance(uni_a.user_tag, TAGRange)):
305 1
            uni_a_list = True
306 1
        if (uni_z.user_tag and isinstance(uni_z.user_tag, TAGRange)):
307 1
            uni_z_list = True
308 1
        if uni_a_list and uni_z_list:
309 1
            return uni_a.user_tag.value == uni_z.user_tag.value
310 1
        return uni_a_list == uni_z_list
311
312 1
    def _validate_has_primary_or_dynamic(
313
        self,
314
        primary_path=None,
315
        dynamic_backup_path=None,
316
        uni_a=None,
317
        uni_z=None,
318
    ) -> None:
319
        """Validate that it must have a primary path or allow dynamic paths."""
320 1
        primary_path = (
321
            primary_path
322
            if primary_path is not None
323
            else self.primary_path
324
        )
325 1
        dynamic_backup_path = (
326
            dynamic_backup_path
327
            if dynamic_backup_path is not None
328
            else self.dynamic_backup_path
329
        )
330 1
        uni_a = uni_a if uni_a is not None else self.uni_a
331 1
        uni_z = uni_z if uni_z is not None else self.uni_z
332 1
        if (
333
            not primary_path
334
            and not dynamic_backup_path
335
            and uni_a and uni_z
336
            and uni_a.interface.switch != uni_z.interface.switch
337
        ):
338 1
            msg = "The EVC must have a primary path or allow dynamic paths."
339 1
            raise ValueError(msg)
340
341 1
    def __eq__(self, other):
342
        """Override the default implementation."""
343 1
        if not isinstance(other, EVC):
344
            return False
345
346 1
        attrs_to_compare = ["name", "uni_a", "uni_z", "owner", "bandwidth"]
347 1
        for attribute in attrs_to_compare:
348 1
            if getattr(other, attribute) != getattr(self, attribute):
349 1
                return False
350 1
        return True
351
352 1
    def is_intra_switch(self):
353
        """Check if the UNIs are in the same switch."""
354 1
        return self.uni_a.interface.switch == self.uni_z.interface.switch
355
356 1
    def check_no_tag_duplicate(self, other_uni: UNI):
357
        """Check if a no tag UNI is duplicated."""
358
        if other_uni in (self.uni_a, self.uni_z):
359
            msg = f"UNI with interface {other_uni.interface.id} is"\
360
                  f" duplicated with EVC {self.id}."
361
            raise DuplicatedNoTagUNI(msg)
362
363
    def as_dict(self, keys: set = None):
364
        """Return a dictionary representing an EVC object.
365 1
            keys: Only fields on this variable will be
366
                  returned in the dictionary"""
367
        evc_dict = {
368
            "id": self.id,
369 1
            "name": self.name,
370
            "uni_a": self.uni_a.as_dict(),
371
            "uni_z": self.uni_z.as_dict(),
372
        }
373
374
        time_fmt = "%Y-%m-%dT%H:%M:%S"
375
376 1
        evc_dict["start_date"] = self.start_date
377
        if isinstance(self.start_date, datetime):
378 1
            evc_dict["start_date"] = self.start_date.strftime(time_fmt)
379 1
380 1
        evc_dict["end_date"] = self.end_date
381
        if isinstance(self.end_date, datetime):
382 1
            evc_dict["end_date"] = self.end_date.strftime(time_fmt)
383 1
384 1
        evc_dict["queue_id"] = self.queue_id
385
        evc_dict["bandwidth"] = self.bandwidth
386 1
        evc_dict["primary_links"] = self.primary_links.as_dict()
387 1
        evc_dict["backup_links"] = self.backup_links.as_dict()
388 1
        evc_dict["current_path"] = self.current_path.as_dict()
389 1
        evc_dict["failover_path"] = self.failover_path.as_dict()
390 1
        evc_dict["primary_path"] = self.primary_path.as_dict()
391 1
        evc_dict["backup_path"] = self.backup_path.as_dict()
392 1
        evc_dict["dynamic_backup_path"] = self.dynamic_backup_path
393 1
        evc_dict["metadata"] = self.metadata
394 1
395 1
        evc_dict["request_time"] = self.request_time
396
        if isinstance(self.request_time, datetime):
397 1
            evc_dict["request_time"] = self.request_time.strftime(time_fmt)
398 1
399 1
        time = self.creation_time.strftime(time_fmt)
400
        evc_dict["creation_time"] = time
401 1
402 1
        evc_dict["owner"] = self.owner
403
        evc_dict["circuit_scheduler"] = [
404 1
            sc.as_dict() for sc in self.circuit_scheduler
405 1
        ]
406
407
        evc_dict["active"] = self.is_active()
408
        evc_dict["enabled"] = self.is_enabled()
409 1
        evc_dict["archived"] = self.archived
410 1
        evc_dict["sb_priority"] = self.sb_priority
411 1
        evc_dict["service_level"] = self.service_level
412 1
        evc_dict["primary_constraints"] = self.primary_constraints
413 1
        evc_dict["secondary_constraints"] = self.secondary_constraints
414 1
        evc_dict["flow_removed_at"] = self.flow_removed_at
415 1
        evc_dict["updated_at"] = self.updated_at
416 1
417 1
        if keys:
418
            selected = {}
419 1
            for key in keys:
420 1
                if key == "enable":
421 1
                    selected["enabled"] = evc_dict["enabled"]
422 1
                    continue
423
                selected[key] = evc_dict[key]
424
            selected["id"] = evc_dict["id"]
425 1
            return selected
426 1
        return evc_dict
427 1
428 1
    @property
429
    def id(self):  # pylint: disable=invalid-name
430 1
        """Return this EVC's ID."""
431 1
        return self._id
432
433 1
    def archive(self):
434
        """Archive this EVC on deletion."""
435 1
        self.archived = True
436
437 1
    def _use_uni_vlan(
438
        self,
439 1
        uni: UNI,
440
        uni_dif: Union[None, UNI] = None
441
    ):
442
        """Use tags from UNI"""
443
        if uni.user_tag is None:
444
            return
445 1
        tag = uni.user_tag.value
446 1
        if not tag:
447 1
            return
448 1
        tag_type = uni.user_tag.tag_type
449
        if (uni_dif and isinstance(tag, list) and
450 1
                isinstance(uni_dif.user_tag.value, list)):
451 1
            tag = range_difference(tag, uni_dif.user_tag.value)
452
            if not tag:
453 1
                return
454 1
        uni.interface.use_tags(
455 1
            self._controller, tag, tag_type, use_lock=True, check_order=False
456 1
        )
457
458
    def make_uni_vlan_available(
459
        self,
460 1
        uni: UNI,
461
        uni_dif: Union[None, UNI] = None,
462
    ):
463
        """Make available tag from UNI"""
464
        if uni.user_tag is None:
465
            return
466 1
        tag = uni.user_tag.value
467 1
        if not tag:
468 1
            return
469 1
        tag_type = uni.user_tag.tag_type
470 1
        if (uni_dif and isinstance(tag, list) and
471 1
                isinstance(uni_dif.user_tag.value, list)):
472 1
            tag = range_difference(tag, uni_dif.user_tag.value)
473
            if not tag:
474 1
                return
475 1
        try:
476
            conflict = uni.interface.make_tags_available(
477 1
                self._controller, tag, tag_type, use_lock=True,
478 1
                check_order=False
479
            )
480
        except KytosTagError as err:
481
            log.error(f"Error in circuit {self._id}: {err}")
482 1
            return
483 1
        if conflict:
484 1
            intf = uni.interface.id
485 1
            log.warning(f"Tags {conflict} was already available in {intf}")
486 1
487 1
    def remove_uni_tags(self):
488
        """Remove both UNI usage of a tag"""
489 1
        self.make_uni_vlan_available(self.uni_a)
490
        self.make_uni_vlan_available(self.uni_z)
491 1
492 1
493
# pylint: disable=fixme, too-many-public-methods
494
class EVCDeploy(EVCBase):
495
    """Class to handle the deploy procedures."""
496 1
497
    def create(self):
498
        """Create a EVC."""
499 1
500
    def discover_new_paths(self):
501
        """Discover new paths to satisfy this circuit and deploy it."""
502 1
        return DynamicPathManager.get_best_paths(self,
503
                                                 **self.primary_constraints)
504
505
    def get_failover_path_candidates(self):
506
        """Get failover paths to satisfy this EVC."""
507 1
        # in the future we can return primary/backup paths as well
508
        # we just have to properly handle link_up and failover paths
509
        # if (
510
        #     self.is_using_primary_path() and
511
        #     self.backup_path.status is EntityStatus.UP
512
        # ):
513
        #     yield self.backup_path
514
        return DynamicPathManager.get_disjoint_paths(self, self.current_path)
515
516 1
    def change_path(self):
517
        """Change EVC path."""
518 1
519
    def reprovision(self):
520
        """Force the EVC (re-)provisioning."""
521 1
522
    def is_affected_by_link(self, link):
523
        """Return True if this EVC has the given link on its current path."""
524 1
        return link in self.current_path
525
526 1
    def link_affected_by_interface(self, interface):
527
        """Return True if this EVC has the given link on its current path."""
528 1
        return self.current_path.link_affected_by_interface(interface)
529
530
    def is_backup_path_affected_by_link(self, link):
531
        """Return True if the backup path of this EVC uses the given link."""
532 1
        return link in self.backup_path
533
534 1
    # pylint: disable=invalid-name
535
    def is_primary_path_affected_by_link(self, link):
536
        """Return True if the primary path of this EVC uses the given link."""
537 1
        return link in self.primary_path
538
539 1
    def is_failover_path_affected_by_link(self, link):
540
        """Return True if this EVC has the given link on its failover path."""
541 1
        return link in self.failover_path
542
543 1
    def is_eligible_for_failover_path(self):
544
        """Verify if this EVC is eligible for failover path (EP029)"""
545 1
        # In the future this function can be augmented to consider
546
        # primary/backup, primary/dynamic, and other path combinations
547
        return (
548
            self.dynamic_backup_path and
549 1
            not self.primary_path and not self.backup_path
550
        )
551
552
    def is_using_primary_path(self):
553
        """Verify if the current deployed path is self.primary_path."""
554 1
        return self.primary_path and (self.current_path == self.primary_path)
555
556 1
    def is_using_backup_path(self):
557
        """Verify if the current deployed path is self.backup_path."""
558 1
        return self.backup_path and (self.current_path == self.backup_path)
559
560 1
    def is_using_dynamic_path(self):
561
        """Verify if the current deployed path is a dynamic path."""
562 1
        if (
563
            self.current_path
564 1
            and not self.is_using_primary_path()
565
            and not self.is_using_backup_path()
566
            and self.current_path.status == EntityStatus.UP
567
        ):
568
            return True
569
        return False
570
571 1
    def deploy_to_backup_path(self):
572
        """Deploy the backup path into the datapaths of this circuit.
573 1
574
        If the backup_path attribute is valid and up, this method will try to
575
        deploy this backup_path.
576
577
        If everything fails and dynamic_backup_path is True, then tries to
578
        deploy a dynamic path.
579
        """
580
        # TODO: Remove flows from current (cookies)
581
        if self.is_using_backup_path():
582
            # TODO: Log to say that cannot move backup to backup
583 1
            return True
584
585
        success = False
586
        if self.backup_path.status is EntityStatus.UP:
587 1
            success = self.deploy_to_path(self.backup_path)
588 1
589 1
        if success:
590
            return True
591 1
592 1
        if self.dynamic_backup_path or self.is_intra_switch():
593
            return self.deploy_to_path()
594 1
595 1
        return False
596
597
    def deploy_to_primary_path(self):
598
        """Deploy the primary path into the datapaths of this circuit.
599 1
600
        If the primary_path attribute is valid and up, this method will try to
601
        deploy this primary_path.
602
        """
603
        # TODO: Remove flows from current (cookies)
604
        if self.is_using_primary_path():
605
            # TODO: Log to say that cannot move primary to primary
606 1
            return True
607
608
        if self.primary_path.status is EntityStatus.UP:
609
            return self.deploy_to_path(self.primary_path)
610 1
        return False
611 1
612 1
    def deploy(self):
613
        """Deploy EVC to best path.
614 1
615
        Best path can be the primary path, if available. If not, the backup
616
        path, and, if it is also not available, a dynamic path.
617
        """
618
        if self.archived:
619
            return False
620 1
        self.enable()
621 1
        success = self.deploy_to_primary_path()
622 1
        if not success:
623 1
            success = self.deploy_to_backup_path()
624 1
625 1
        if success:
626
            emit_event(self._controller, "deployed",
627 1
                       content=map_evc_event_content(self))
628 1
        return success
629
630 1
    @staticmethod
631
    def get_path_status(path):
632 1
        """Check for the current status of a path.
633 1
634
        If any link in this path is down, the path is considered down.
635
        """
636
        if not path:
637
            return EntityStatus.DISABLED
638 1
639 1
        for link in path:
640
            if link.status is not EntityStatus.UP:
641 1
                return link.status
642 1
        return EntityStatus.UP
643 1
644 1
    #    def discover_new_path(self):
645
    #        # TODO: discover a new path to satisfy this circuit and deploy
646
647
    def remove(self):
648
        """Remove EVC path and disable it."""
649 1
        self.remove_current_flows()
650
        self.remove_failover_flows()
651 1
        self.disable()
652 1
        self.sync()
653 1
        emit_event(self._controller, "undeployed",
654 1
                   content=map_evc_event_content(self))
655 1
656
    def remove_failover_flows(self, exclude_uni_switches=True,
657
                              force=True, sync=True) -> None:
658 1
        """Remove failover_flows.
659
660
        By default, it'll exclude UNI switches, if mef_eline has already
661
        called remove_current_flows before then this minimizes the number
662
        of FlowMods and IO.
663
        """
664
        if not self.failover_path:
665
            return
666 1
        switches, cookie, excluded = OrderedDict(), self.get_cookie(), set()
667 1
        links = set()
668 1
        if exclude_uni_switches:
669 1
            excluded.add(self.uni_a.interface.switch.id)
670 1
            excluded.add(self.uni_z.interface.switch.id)
671 1
        for link in self.failover_path:
672 1
            if link.endpoint_a.switch.id not in excluded:
673 1
                switches[link.endpoint_a.switch.id] = link.endpoint_a.switch
674 1
                links.add(link)
675 1
            if link.endpoint_b.switch.id not in excluded:
676 1
                switches[link.endpoint_b.switch.id] = link.endpoint_b.switch
677 1
                links.add(link)
678 1
        for switch in switches.values():
679 1
            try:
680 1
                self._send_flow_mods(
681 1
                    switch.id,
682 1
                    [
683
                        {
684
                            "cookie": cookie,
685
                            "cookie_mask": int(0xffffffffffffffff),
686
                        }
687
                    ],
688
                    "delete",
689
                    force=force,
690
                )
691
            except FlowModException as err:
692
                log.error(
693
                    f"Error removing flows from switch {switch.id} for"
694
                    f"EVC {self}: {err}"
695
                )
696
        try:
697
            self.failover_path.make_vlans_available(self._controller)
698 1
        except KytosTagError as err:
699 1
            log.error(f"Error when removing failover flows: {err}")
700
        self.failover_path = Path([])
701
        if sync:
702 1
            self.sync()
703 1
704 1
    def remove_current_flows(self, current_path=None, force=True):
705
        """Remove all flows from current path."""
706 1
        switches = set()
707
708 1
        switches.add(self.uni_a.interface.switch)
709
        switches.add(self.uni_z.interface.switch)
710 1
        if not current_path:
711 1
            current_path = self.current_path
712 1
        for link in current_path:
713 1
            switches.add(link.endpoint_a.switch)
714 1
            switches.add(link.endpoint_b.switch)
715 1
716 1
        match = {
717
            "cookie": self.get_cookie(),
718 1
            "cookie_mask": int(0xffffffffffffffff)
719
        }
720
721
        for switch in switches:
722
            try:
723 1
                self._send_flow_mods(switch.id, [match], 'delete', force=force)
724 1
            except FlowModException as err:
725 1
                log.error(
726 1
                    f"Error removing flows from switch {switch.id} for"
727 1
                    f"EVC {self}: {err}"
728
                )
729
        try:
730
            current_path.make_vlans_available(self._controller)
731 1
        except KytosTagError as err:
732 1
            log.error(f"Error when removing current path flows: {err}")
733
        self.current_path = Path([])
734
        self.deactivate()
735 1
        self.sync()
736 1
737 1
    def remove_path_flows(self, path=None, force=True):
738
        """Remove all flows from path."""
739 1
        if not path:
740
            return
741 1
742 1
        dpid_flows_match = {}
743
744 1
        try:
745
            nni_flows = self._prepare_nni_flows(path)
746 1
        # pylint: disable=broad-except
747 1
        except Exception:
748
            err = traceback.format_exc().replace("\n", ", ")
749
            log.error(f"Fail to remove NNI failover flows for {self}: {err}")
750
            nni_flows = {}
751
752
        for dpid, flows in nni_flows.items():
753
            dpid_flows_match.setdefault(dpid, [])
754 1
            for flow in flows:
755 1
                dpid_flows_match[dpid].append({
756 1
                    "cookie": flow["cookie"],
757 1
                    "match": flow["match"],
758
                    "cookie_mask": int(0xffffffffffffffff)
759
                })
760
761
        try:
762
            uni_flows = self._prepare_uni_flows(path, skip_in=True)
763 1
        # pylint: disable=broad-except
764 1
        except Exception:
765
            err = traceback.format_exc().replace("\n", ", ")
766
            log.error(f"Fail to remove UNI failover flows for {self}: {err}")
767
            uni_flows = {}
768
769
        for dpid, flows in uni_flows.items():
770
            dpid_flows_match.setdefault(dpid, [])
771 1
            for flow in flows:
772 1
                dpid_flows_match[dpid].append({
773 1
                    "cookie": flow["cookie"],
774 1
                    "match": flow["match"],
775
                    "cookie_mask": int(0xffffffffffffffff)
776
                })
777
778
        for dpid, flows in dpid_flows_match.items():
779
            try:
780 1
                self._send_flow_mods(dpid, flows, 'delete', force=force)
781 1
            except FlowModException as err:
782 1
                log.error(
783 1
                    "Error removing failover flows: "
784 1
                    f"dpid={dpid} evc={self} error={err}"
785
                )
786
        try:
787
            path.make_vlans_available(self._controller)
788 1
        except KytosTagError as err:
789 1
            log.error(f"Error when removing path flows: {err}")
790
791
    @staticmethod
792
    def links_zipped(path=None):
793 1
        """Return an iterator which yields pairs of links in order."""
794 1
        if not path:
795
            return []
796 1
        return zip(path[:-1], path[1:])
797
798 1
    def should_deploy(self, path=None):
799
        """Verify if the circuit should be deployed."""
800 1
        if not path:
801
            log.debug("Path is empty.")
802 1
            return False
803 1
804 1
        if not self.is_enabled():
805
            log.debug(f"{self} is disabled.")
806 1
            return False
807 1
808 1
        if not self.is_active():
809
            log.debug(f"{self} will be deployed.")
810 1
            return True
811 1
812 1
        return False
813
814 1
    def deploy_to_path(self, path=None):  # pylint: disable=too-many-branches
815
        """Install the flows for this circuit.
816 1
817
        Procedures to deploy:
818
819
        0. Remove current flows installed
820
        1. Decide if will deploy "path" or discover a new path
821
        2. Choose vlan
822
        3. Install NNI flows
823
        4. Install UNI flows
824
        5. Activate
825
        6. Update current_path
826
        7. Update links caches(primary, current, backup)
827
828
        """
829
        self.remove_current_flows()
830
        use_path = path
831 1
        if self.should_deploy(use_path):
832 1
            try:
833 1
                use_path.choose_vlans(self._controller)
834 1
            except KytosNoTagAvailableError:
835 1
                use_path = None
836 1
        else:
837 1
            for use_path in self.discover_new_paths():
838
                if use_path is None:
839 1
                    continue
840 1
                try:
841
                    use_path.choose_vlans(self._controller)
842 1
                    break
843 1
                except KytosNoTagAvailableError:
844 1
                    pass
845 1
            else:
846 1
                use_path = None
847
848 1
        try:
849
            if use_path:
850 1
                self._install_nni_flows(use_path)
851 1
                self._install_uni_flows(use_path)
852 1
            elif self.is_intra_switch():
853 1
                use_path = Path()
854 1
                self._install_direct_uni_flows()
855 1
            else:
856 1
                log.warning(
857
                    f"{self} was not deployed. No available path was found."
858 1
                )
859
                return False
860
        except FlowModException as err:
861 1
            log.error(
862 1
                f"Error deploying EVC {self} when calling flow_manager: {err}"
863 1
            )
864
            self.remove_current_flows(use_path)
865
            return False
866 1
        self.activate()
867 1
        self.current_path = use_path
868 1
        self.sync()
869 1
        log.info(f"{self} was deployed.")
870 1
        return True
871 1
872 1
    def setup_failover_path(self):
873
        """Install flows for the failover path of this EVC.
874 1
875
        Procedures to deploy:
876
877
        0. Remove flows currently installed for failover_path (if any)
878
        1. Discover a disjoint path from current_path
879
        2. Choose vlans
880
        3. Install NNI flows
881
        4. Install UNI egress flows
882
        5. Update failover_path
883
        """
884
        # Intra-switch EVCs have no failover_path
885
        if self.is_intra_switch():
886
            return False
887 1
888 1
        # For not only setup failover path for totally dynamic EVCs
889
        if not self.is_eligible_for_failover_path():
890
            return False
891 1
892 1
        reason = ""
893
        self.remove_path_flows(self.failover_path)
894 1
        self.failover_path = Path([])
895 1
        for use_path in self.get_failover_path_candidates():
896 1
            if not use_path:
897 1
                continue
898 1
            try:
899 1
                use_path.choose_vlans(self._controller)
900 1
                break
901 1
            except KytosNoTagAvailableError:
902 1
                pass
903 1
        else:
904 1
            use_path = Path([])
905
            reason = "No available path was found"
906 1
907 1
        try:
908
            if use_path:
909 1
                self._install_nni_flows(use_path)
910 1
                self._install_uni_flows(use_path, skip_in=True)
911 1
        except FlowModException as err:
912 1
            reason = "Error deploying failover path"
913 1
            log.error(
914 1
                f"{reason} for {self}. FlowManager error: {err}"
915 1
            )
916
            self.remove_path_flows(use_path)
917
            use_path = Path([])
918 1
919 1
        self.failover_path = use_path
920
        self.sync()
921 1
922 1
        if not use_path:
923
            log.warning(
924 1
                f"Failover path for {self} was not deployed: {reason}"
925 1
            )
926
            return False
927
        log.info(f"Failover path for {self} was deployed.")
928 1
        return True
929 1
930 1
    def get_failover_flows(self):
931
        """Return the flows needed to make the failover path active, i.e. the
932 1
        flows for ingress forwarding.
933
934
        Return:
935
            dict: A dict of flows indexed by the switch_id will be returned, or
936
                an empty dict if no failover_path is available.
937
        """
938
        if not self.failover_path:
939
            return {}
940 1
        return self._prepare_uni_flows(self.failover_path, skip_out=True)
941 1
942 1
    # pylint: disable=too-many-branches
943
    def _prepare_direct_uni_flows(self):
944
        """Prepare flows connecting two UNIs for intra-switch EVC."""
945 1
        vlan_a = self._get_value_from_uni_tag(self.uni_a)
946
        vlan_z = self._get_value_from_uni_tag(self.uni_z)
947 1
948 1
        flow_mod_az = self._prepare_flow_mod(
949
            self.uni_a.interface, self.uni_z.interface,
950 1
            self.queue_id, vlan_a
951
        )
952
        flow_mod_za = self._prepare_flow_mod(
953
            self.uni_z.interface, self.uni_a.interface,
954 1
            self.queue_id, vlan_z
955
        )
956
957 View Code Duplication
        if not isinstance(vlan_z, list) and vlan_z not in self.special_cases:
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
958
            flow_mod_az["actions"].insert(
959 1
                0, {"action_type": "set_vlan", "vlan_id": vlan_z}
960 1
            )
961
            if not vlan_a:
962
                flow_mod_az["actions"].insert(
963 1
                    0, {"action_type": "push_vlan", "tag_type": "c"}
964 1
                )
965
            if vlan_a == 0:
966
                flow_mod_za["actions"].insert(0, {"action_type": "pop_vlan"})
967 1
        elif vlan_a == 0 and vlan_z == "4096/4096":
968 1
            flow_mod_za["actions"].insert(0, {"action_type": "pop_vlan"})
969 1
970 1 View Code Duplication
        if not isinstance(vlan_a, list) and vlan_a not in self.special_cases:
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
971
            flow_mod_za["actions"].insert(
972 1
                    0, {"action_type": "set_vlan", "vlan_id": vlan_a}
973 1
                )
974
            if not vlan_z:
975
                flow_mod_za["actions"].insert(
976 1
                    0, {"action_type": "push_vlan", "tag_type": "c"}
977 1
                )
978
            if vlan_z == 0:
979
                flow_mod_az["actions"].insert(0, {"action_type": "pop_vlan"})
980 1
        elif vlan_a == "4096/4096" and vlan_z == 0:
981 1
            flow_mod_az["actions"].insert(0, {"action_type": "pop_vlan"})
982 1
983 1
        flows = []
984
        if isinstance(vlan_a, list):
985 1
            for mask_a in vlan_a:
986 1
                flow_aux = deepcopy(flow_mod_az)
987 1
                flow_aux["match"]["dl_vlan"] = mask_a
988 1
                flows.append(flow_aux)
989 1
        else:
990 1
            if vlan_a is not None:
991
                flow_mod_az["match"]["dl_vlan"] = vlan_a
992 1
            flows.append(flow_mod_az)
993 1
994 1
        if isinstance(vlan_z, list):
995
            for mask_z in vlan_z:
996 1
                flow_aux = deepcopy(flow_mod_za)
997 1
                flow_aux["match"]["dl_vlan"] = mask_z
998 1
                flows.append(flow_aux)
999 1
        else:
1000 1
            if vlan_z is not None:
1001
                flow_mod_za["match"]["dl_vlan"] = vlan_z
1002 1
            flows.append(flow_mod_za)
1003 1
        return (
1004 1
            self.uni_a.interface.switch.id, flows
1005 1
        )
1006
1007
    def _install_direct_uni_flows(self):
1008
        """Install flows connecting two UNIs.
1009 1
1010
        This case happens when the circuit is between UNIs in the
1011
        same switch.
1012
        """
1013
        (dpid, flows) = self._prepare_direct_uni_flows()
1014
        self._send_flow_mods(dpid, flows)
1015 1
1016 1
    def _prepare_nni_flows(self, path=None):
1017
        """Prepare NNI flows."""
1018 1
        nni_flows = OrderedDict()
1019
        previous = self.uni_a.interface.switch.dpid
1020 1
        for incoming, outcoming in self.links_zipped(path):
1021 1
            in_vlan = incoming.get_metadata("s_vlan").value
1022 1
            out_vlan = outcoming.get_metadata("s_vlan").value
1023 1
            in_endpoint = self.get_endpoint_by_id(incoming, previous, ne)
1024 1
            out_endpoint = self.get_endpoint_by_id(
1025 1
                outcoming, in_endpoint.switch.id, eq
1026 1
            )
1027
1028
            flows = []
1029
            # Flow for one direction
1030 1
            flows.append(
1031
                self._prepare_nni_flow(
1032 1
                    in_endpoint,
1033
                    out_endpoint,
1034
                    in_vlan,
1035
                    out_vlan,
1036
                    queue_id=self.queue_id,
1037
                )
1038
            )
1039
1040
            # Flow for the other direction
1041
            flows.append(
1042
                self._prepare_nni_flow(
1043 1
                    out_endpoint,
1044
                    in_endpoint,
1045
                    out_vlan,
1046
                    in_vlan,
1047
                    queue_id=self.queue_id,
1048
                )
1049
            )
1050
            previous = in_endpoint.switch.id
1051
            nni_flows[in_endpoint.switch.id] = flows
1052 1
        return nni_flows
1053 1
1054 1
    def _install_nni_flows(self, path=None):
1055
        """Install NNI flows."""
1056 1
        for dpid, flows in self._prepare_nni_flows(path).items():
1057
            self._send_flow_mods(dpid, flows)
1058 1
1059 1
    @staticmethod
1060
    def _get_value_from_uni_tag(uni: UNI):
1061 1
        """Returns the value from tag. In case of any and untagged
1062 1
        it should return 4096/4096 and 0 respectively"""
1063
        special = {"any": "4096/4096", "untagged": 0}
1064
        if uni.user_tag:
1065 1
            value = uni.user_tag.value
1066 1
            if isinstance(value, list):
1067 1
                return uni.user_tag.mask_list
1068 1
            return special.get(value, value)
1069 1
        return None
1070 1
1071 1
    # pylint: disable=too-many-locals
1072
    def _prepare_uni_flows(self, path=None, skip_in=False, skip_out=False):
1073
        """Prepare flows to install UNIs."""
1074 1
        uni_flows = {}
1075
        if not path:
1076 1
            log.info("install uni flows without path.")
1077 1
            return uni_flows
1078 1
1079 1
        # Determine VLANs
1080
        in_vlan_a = self._get_value_from_uni_tag(self.uni_a)
1081
        out_vlan_a = path[0].get_metadata("s_vlan").value
1082 1
1083 1
        in_vlan_z = self._get_value_from_uni_tag(self.uni_z)
1084
        out_vlan_z = path[-1].get_metadata("s_vlan").value
1085 1
1086 1
        # Get endpoints from path
1087
        endpoint_a = self.get_endpoint_by_id(
1088
            path[0], self.uni_a.interface.switch.id, eq
1089 1
        )
1090
        endpoint_z = self.get_endpoint_by_id(
1091
            path[-1], self.uni_z.interface.switch.id, eq
1092 1
        )
1093
1094
        # Flows for the first UNI
1095
        flows_a = []
1096
1097 1
        # Flow for one direction, pushing the service tag
1098
        if not skip_in:
1099
            if isinstance(in_vlan_a, list):
1100 1
                for in_mask_a in in_vlan_a:
1101 1
                    push_flow = self._prepare_push_flow(
1102 1
                        self.uni_a.interface,
1103 1
                        endpoint_a,
1104
                        in_mask_a,
1105
                        out_vlan_a,
1106
                        in_vlan_z,
1107
                        queue_id=self.queue_id,
1108
                    )
1109
                    flows_a.append(push_flow)
1110
            else:
1111 1
                push_flow = self._prepare_push_flow(
1112
                    self.uni_a.interface,
1113 1
                    endpoint_a,
1114
                    in_vlan_a,
1115
                    out_vlan_a,
1116
                    in_vlan_z,
1117
                    queue_id=self.queue_id,
1118
                )
1119
                flows_a.append(push_flow)
1120
1121 1
        # Flow for the other direction, popping the service tag
1122
        if not skip_out:
1123
            pop_flow = self._prepare_pop_flow(
1124 1
                endpoint_a,
1125 1
                self.uni_a.interface,
1126
                out_vlan_a,
1127
                queue_id=self.queue_id,
1128
            )
1129
            flows_a.append(pop_flow)
1130
1131 1
        uni_flows[self.uni_a.interface.switch.id] = flows_a
1132
1133 1
        # Flows for the second UNI
1134
        flows_z = []
1135
1136 1
        # Flow for one direction, pushing the service tag
1137
        if not skip_in:
1138
            if isinstance(in_vlan_z, list):
1139 1
                for in_mask_z in in_vlan_z:
1140 1
                    push_flow = self._prepare_push_flow(
1141 1
                        self.uni_z.interface,
1142 1
                        endpoint_z,
1143
                        in_mask_z,
1144
                        out_vlan_z,
1145
                        in_vlan_a,
1146
                        queue_id=self.queue_id,
1147
                    )
1148
                    flows_z.append(push_flow)
1149
            else:
1150 1
                push_flow = self._prepare_push_flow(
1151
                    self.uni_z.interface,
1152 1
                    endpoint_z,
1153
                    in_vlan_z,
1154
                    out_vlan_z,
1155
                    in_vlan_a,
1156
                    queue_id=self.queue_id,
1157
                )
1158
                flows_z.append(push_flow)
1159
1160 1
        # Flow for the other direction, popping the service tag
1161
        if not skip_out:
1162
            pop_flow = self._prepare_pop_flow(
1163 1
                endpoint_z,
1164 1
                self.uni_z.interface,
1165
                out_vlan_z,
1166
                queue_id=self.queue_id,
1167
            )
1168
            flows_z.append(pop_flow)
1169
1170 1
        uni_flows[self.uni_z.interface.switch.id] = flows_z
1171
1172 1
        return uni_flows
1173
1174 1
    def _install_uni_flows(self, path=None, skip_in=False, skip_out=False):
1175
        """Install UNI flows."""
1176 1
        uni_flows = self._prepare_uni_flows(path, skip_in, skip_out)
1177
1178 1
        for (dpid, flows) in uni_flows.items():
1179
            self._send_flow_mods(dpid, flows)
1180 1
1181 1
    @staticmethod
1182
    def _send_flow_mods(dpid, flow_mods, command='flows', force=False):
1183 1
        """Send a flow_mod list to a specific switch.
1184 1
1185
        Args:
1186
            dpid(str): The target of flows (i.e. Switch.id).
1187
            flow_mods(dict): Python dictionary with flow_mods.
1188
            command(str): By default is 'flows'. To remove a flow is 'remove'.
1189
            force(bool): True to send via consistency check in case of errors
1190
1191
        """
1192
1193
        endpoint = f"{settings.MANAGER_URL}/{command}/{dpid}"
1194
1195 1
        data = {"flows": flow_mods, "force": force}
1196
        response = requests.post(endpoint, json=data)
1197 1
        if response.status_code >= 400:
1198 1
            raise FlowModException(str(response.text))
1199 1
1200 1
    def get_cookie(self):
1201
        """Return the cookie integer from evc id."""
1202 1
        return int(self.id, 16) + (settings.COOKIE_PREFIX << 56)
1203
1204 1
    @staticmethod
1205
    def get_id_from_cookie(cookie):
1206 1
        """Return the evc id given a cookie value."""
1207 1
        evc_id = cookie - (settings.COOKIE_PREFIX << 56)
1208
        return f"{evc_id:x}".zfill(14)
1209 1
1210 1
    def set_flow_table_group_id(self, flow_mod: dict, vlan) -> dict:
1211
        """Set table_group and table_id"""
1212 1
        table_group = "epl" if vlan is None else "evpl"
1213
        flow_mod["table_group"] = table_group
1214 1
        flow_mod["table_id"] = self.table_group[table_group]
1215 1
        return flow_mod
1216 1
1217 1
    @staticmethod
1218
    def get_priority(vlan):
1219 1
        """Return priority value depending on vlan value"""
1220 1
        if isinstance(vlan, list):
1221
            return settings.EVPL_SB_PRIORITY
1222 1
        if vlan not in {None, "4096/4096", 0}:
1223 1
            return settings.EVPL_SB_PRIORITY
1224 1
        if vlan == 0:
1225 1
            return settings.UNTAGGED_SB_PRIORITY
1226 1
        if vlan == "4096/4096":
1227 1
            return settings.ANY_SB_PRIORITY
1228 1
        return settings.EPL_SB_PRIORITY
1229 1
1230 1
    def _prepare_flow_mod(self, in_interface, out_interface,
1231
                          queue_id=None, vlan=True):
1232 1
        """Prepare a common flow mod."""
1233
        default_actions = [
1234
            {"action_type": "output", "port": out_interface.port_number}
1235 1
        ]
1236
        queue_id = settings.QUEUE_ID if queue_id == -1 else queue_id
1237
        if queue_id is not None:
1238 1
            default_actions.append(
1239 1
                {"action_type": "set_queue", "queue_id": queue_id}
1240 1
            )
1241
1242
        flow_mod = {
1243
            "match": {"in_port": in_interface.port_number},
1244 1
            "cookie": self.get_cookie(),
1245
            "actions": default_actions,
1246
            "owner": "mef_eline",
1247
        }
1248
1249
        self.set_flow_table_group_id(flow_mod, vlan)
1250
        if self.sb_priority:
1251 1
            flow_mod["priority"] = self.sb_priority
1252 1
        else:
1253 1
            flow_mod["priority"] = self.get_priority(vlan)
1254
        return flow_mod
1255 1
1256 1
    def _prepare_nni_flow(self, *args, queue_id=None):
1257
        """Create NNI flows."""
1258 1
        in_interface, out_interface, in_vlan, out_vlan = args
1259
        flow_mod = self._prepare_flow_mod(
1260 1
            in_interface, out_interface, queue_id
1261 1
        )
1262
        flow_mod["match"]["dl_vlan"] = in_vlan
1263
        new_action = {"action_type": "set_vlan", "vlan_id": out_vlan}
1264 1
        flow_mod["actions"].insert(0, new_action)
1265 1
1266 1
        return flow_mod
1267
1268 1
    def _prepare_push_flow(self, *args, queue_id=None):
1269
        """Prepare push flow.
1270 1
1271
        Arguments:
1272
            in_interface(str): Interface input.
1273
            out_interface(str): Interface output.
1274
            in_vlan(int,str,None): Vlan input.
1275
            out_vlan(str): Vlan output.
1276
            new_c_vlan(int,str,list,None): New client vlan.
1277
1278
        Return:
1279
            dict: An python dictionary representing a FlowMod
1280
1281
        """
1282
        # assign all arguments
1283
        in_interface, out_interface, in_vlan, out_vlan, new_c_vlan = args
1284
        vlan_pri = in_vlan if not isinstance(new_c_vlan, list) else new_c_vlan
1285 1
        flow_mod = self._prepare_flow_mod(
1286 1
            in_interface, out_interface, queue_id, vlan_pri
1287 1
        )
1288
        # the service tag must be always pushed
1289
        new_action = {"action_type": "set_vlan", "vlan_id": out_vlan}
1290
        flow_mod["actions"].insert(0, new_action)
1291 1
1292 1
        new_action = {"action_type": "push_vlan", "tag_type": "s"}
1293
        flow_mod["actions"].insert(0, new_action)
1294 1
1295 1
        if in_vlan is not None:
1296
            # if in_vlan is set, it must be included in the match
1297 1
            flow_mod["match"]["dl_vlan"] = in_vlan
1298
1299 1
        if (not isinstance(new_c_vlan, list) and in_vlan != new_c_vlan and
1300
                new_c_vlan not in self.special_cases):
1301 1
            # new_in_vlan is an integer but zero, action to set is required
1302
            new_action = {"action_type": "set_vlan", "vlan_id": new_c_vlan}
1303
            flow_mod["actions"].insert(0, new_action)
1304 1
1305 1
        if in_vlan not in self.special_cases and new_c_vlan == 0:
1306
            # # new_in_vlan is an integer but zero and new_c_vlan does not
1307 1
            # a pop action is required
1308
            new_action = {"action_type": "pop_vlan"}
1309
            flow_mod["actions"].insert(0, new_action)
1310 1
1311 1
        elif in_vlan == "4096/4096" and new_c_vlan == 0:
1312
            # if in_vlan match with any tags and new_c_vlan does not
1313 1
            # a pop action is required
1314
            new_action = {"action_type": "pop_vlan"}
1315
            flow_mod["actions"].insert(0, new_action)
1316 1
1317 1
        elif (not in_vlan and
1318
                (not isinstance(new_c_vlan, list) and
1319 1
                 new_c_vlan not in self.special_cases)):
1320
            # new_in_vlan is an integer but zero and in_vlan is not set
1321
            # then it is set now
1322
            new_action = {"action_type": "push_vlan", "tag_type": "c"}
1323
            flow_mod["actions"].insert(0, new_action)
1324 1
1325 1
        return flow_mod
1326
1327 1
    def _prepare_pop_flow(
1328
        self, in_interface, out_interface, out_vlan, queue_id=None
1329 1
    ):
1330
        # pylint: disable=too-many-arguments
1331
        """Prepare pop flow."""
1332
        flow_mod = self._prepare_flow_mod(
1333
            in_interface, out_interface, queue_id
1334 1
        )
1335
        flow_mod["match"]["dl_vlan"] = out_vlan
1336
        new_action = {"action_type": "pop_vlan"}
1337 1
        flow_mod["actions"].insert(0, new_action)
1338 1
        return flow_mod
1339 1
1340 1
    @staticmethod
1341
    def run_bulk_sdntraces(
1342 1
        uni_list: list[tuple[Interface, Union[str, int, None]]]
1343 1
    ) -> dict:
1344
        """Run SDN traces on control plane starting from EVC UNIs."""
1345
        endpoint = f"{settings.SDN_TRACE_CP_URL}/traces"
1346
        data = []
1347 1
        for interface, tag_value in uni_list:
1348 1
            data_uni = {
1349 1
                "trace": {
1350 1
                            "switch": {
1351
                                "dpid": interface.switch.dpid,
1352
                                "in_port": interface.port_number,
1353
                            }
1354
                        }
1355
                }
1356
            if tag_value:
1357
                uni_dl_vlan = map_dl_vlan(tag_value)
1358 1
                if uni_dl_vlan:
1359 1
                    data_uni["trace"]["eth"] = {
1360 1
                                            "dl_type": 0x8100,
1361 1
                                            "dl_vlan": uni_dl_vlan,
1362
                                            }
1363
            data.append(data_uni)
1364
        try:
1365 1
            response = requests.put(endpoint, json=data, timeout=30)
1366 1
        except Timeout as exception:
1367 1
            log.error(f"Request has timed out: {exception}")
1368 1
            return {"result": []}
1369 1
        if response.status_code >= 400:
1370 1
            log.error(f"Failed to run sdntrace-cp: {response.text}")
1371 1
            return {"result": []}
1372 1
        return response.json()
1373 1
1374 1
    # pylint: disable=too-many-return-statements, too-many-arguments
1375
    @staticmethod
1376
    def check_trace(
1377 1
        tag_a: Union[None, int, str],
1378 1
        tag_z: Union[None, int, str],
1379
        interface_a: Interface,
1380
        interface_z: Interface,
1381
        current_path: list,
1382
        trace_a: list,
1383
        trace_z: list
1384
    ) -> bool:
1385
        """Auxiliar function to check an individual trace"""
1386
        if (
1387
            len(trace_a) != len(current_path) + 1
1388 1
            or not compare_uni_out_trace(tag_z, interface_z, trace_a[-1])
1389
        ):
1390
            log.warning(f"Invalid trace from uni_a: {trace_a}")
1391
            return False
1392 1
        if (
1393 1
            len(trace_z) != len(current_path) + 1
1394 1
            or not compare_uni_out_trace(tag_a, interface_a, trace_z[-1])
1395
        ):
1396
            log.warning(f"Invalid trace from uni_z: {trace_z}")
1397
            return False
1398 1
1399 1
        for link, trace1, trace2 in zip(current_path,
1400
                                        trace_a[1:],
1401 1
                                        trace_z[:0:-1]):
1402
            metadata_vlan = None
1403
            if link.metadata:
1404 1
                metadata_vlan = glom(link.metadata, 's_vlan.value')
1405 1
            if compare_endpoint_trace(
1406 1
                                        link.endpoint_a,
1407 1
                                        metadata_vlan,
1408
                                        trace2
1409
                                    ) is False:
1410
                log.warning(f"Invalid trace from uni_a: {trace_a}")
1411
                return False
1412 1
            if compare_endpoint_trace(
1413 1
                                        link.endpoint_b,
1414 1
                                        metadata_vlan,
1415
                                        trace1
1416
                                    ) is False:
1417
                log.warning(f"Invalid trace from uni_z: {trace_z}")
1418
                return False
1419 1
1420 1
        return True
1421
1422 1
    @staticmethod
1423
    def check_range(circuit, traces: list) -> bool:
1424 1
        """Check traces when for UNI with TAGRange"""
1425 1
        check = True
1426
        for i, mask in enumerate(circuit.uni_a.user_tag.mask_list):
1427 1
            trace_a = traces[i*2]
1428 1
            trace_z = traces[i*2+1]
1429 1
            check &= EVCDeploy.check_trace(
1430 1
                mask, mask,
1431 1
                circuit.uni_a.interface,
1432
                circuit.uni_z.interface,
1433
                circuit.current_path,
1434
                trace_a, trace_z,
1435
            )
1436
        return check
1437
1438 1
    @staticmethod
1439
    def check_list_traces(list_circuits: list) -> dict:
1440 1
        """Check if current_path is deployed comparing with SDN traces."""
1441 1
        if not list_circuits:
1442
            return {}
1443 1
        uni_list = make_uni_list(list_circuits)
1444 1
        traces = EVCDeploy.run_bulk_sdntraces(uni_list)["result"]
1445 1
1446 1
        if not traces:
1447
            return {}
1448 1
1449 1
        try:
1450
            circuits_checked = {}
1451 1
            i = 0
1452 1
            for circuit in list_circuits:
1453 1
                if isinstance(circuit.uni_a.user_tag, TAGRange):
1454 1
                    length = len(circuit.uni_a.user_tag.mask_list)
1455 1
                    circuits_checked[circuit.id] = EVCDeploy.check_range(
1456 1
                        circuit, traces[i:i+length*2]
1457 1
                    )
1458
                    i += length*2
1459
                else:
1460 1
                    trace_a = traces[i]
1461
                    trace_z = traces[i+1]
1462 1
                    tag_a = None
1463 1
                    if circuit.uni_a.user_tag:
1464 1
                        tag_a = circuit.uni_a.user_tag.value
1465 1
                    tag_z = None
1466 1
                    if circuit.uni_z.user_tag:
1467 1
                        tag_z = circuit.uni_z.user_tag.value
1468 1
                    circuits_checked[circuit.id] = EVCDeploy.check_trace(
1469 1
                        tag_a,
1470 1
                        tag_z,
1471
                        circuit.uni_a.interface,
1472
                        circuit.uni_z.interface,
1473
                        circuit.current_path,
1474
                        trace_a, trace_z
1475
                    )
1476
                    i += 2
1477
        except IndexError as err:
1478 1
            log.error(
1479 1
                f"Bulk sdntraces returned fewer items than expected."
1480 1
                f"Error = {err}"
1481
            )
1482
            return {}
1483
1484 1
        return circuits_checked
1485
1486 1
    @staticmethod
1487
    def get_endpoint_by_id(
1488 1
        link: Link,
1489 1
        id_: str,
1490
        operator: Union[eq, ne]
1491
    ) -> Interface:
1492
        """Return endpoint from link
1493
        either equal(eq) or not equal(ne) to id"""
1494
        if operator(link.endpoint_a.switch.id, id_):
1495
            return link.endpoint_a
1496 1
        return link.endpoint_b
1497 1
1498 1
1499
class LinkProtection(EVCDeploy):
1500
    """Class to handle link protection."""
1501 1
1502
    def is_affected_by_link(self, link=None):
1503
        """Verify if the current path is affected by link down event."""
1504 1
        return self.current_path.is_affected_by_link(link)
1505
1506
    def is_using_primary_path(self):
1507
        """Verify if the current deployed path is self.primary_path."""
1508 1
        return self.current_path == self.primary_path
1509
1510 1
    def is_using_backup_path(self):
1511
        """Verify if the current deployed path is self.backup_path."""
1512 1
        return self.current_path == self.backup_path
1513
1514 1
    def is_using_dynamic_path(self):
1515
        """Verify if the current deployed path is dynamic."""
1516 1
        if (
1517
            self.current_path
1518 1
            and not self.is_using_primary_path()
1519
            and not self.is_using_backup_path()
1520
            and self.current_path.status is EntityStatus.UP
1521
        ):
1522
            return True
1523
        return False
1524
1525 1
    def deploy_to(self, path_name=None, path=None):
1526
        """Create a deploy to path."""
1527 1
        if self.current_path == path:
1528
            log.debug(f"{path_name} is equal to current_path.")
1529 1
            return True
1530 1
1531 1
        if path.status is EntityStatus.UP:
1532
            return self.deploy_to_path(path)
1533 1
1534 1
        return False
1535
1536 1
    def handle_link_up(self, link):
1537
        """Handle circuit when link up.
1538 1
1539
        Args:
1540
            link(Link): Link affected by link.up event.
1541
1542
        """
1543
        condition_pairs = [
1544
            (
1545 1
                lambda me: me.is_using_primary_path(),
1546
                lambda _: (True, 'nothing')
1547
            ),
1548
            (
1549
                lambda me: me.is_intra_switch(),
1550
                lambda _: (True, 'nothing')
1551
            ),
1552
            (
1553
                lambda me: me.primary_path.is_affected_by_link(link),
1554
                lambda me: (me.deploy_to_primary_path(), 'redeploy')
1555
            ),
1556
            # We tried to deploy(primary_path) without success.
1557
            # And in this case is up by some how. Nothing to do.
1558
            (
1559
                lambda me: me.is_using_backup_path(),
1560
                lambda _: (True, 'nothing')
1561
            ),
1562
            (
1563
                lambda me:  me.is_using_dynamic_path(),
1564
                lambda _: (True, 'nothing')
1565
            ),
1566
            # In this case, probably the circuit is not being used and
1567
            # we can move to backup
1568
            (
1569
                lambda me: me.backup_path.is_affected_by_link(link),
1570
                lambda me: (me.deploy_to_backup_path(), 'redeploy')
1571
            ),
1572
            # In this case, the circuit is not being used and we should
1573
            # try a dynamic path
1574
            (
1575
                lambda me: me.dynamic_backup_path,
1576
                lambda me: (me.deploy_to_path(), 'redeploy')
1577
            )
1578
        ]
1579
        for predicate, action in condition_pairs:
1580
            if not predicate(self):
1581 1
                continue
1582 1
            success, succcess_type = action(self)
1583 1
            if success:
1584 1
                if succcess_type == 'redeploy':
1585 1
                    emit_event(
1586 1
                        self._controller,
1587 1
                        "redeployed_link_up",
1588
                        content=map_evc_event_content(self)
1589
                    )
1590
                return True
1591
        return False
1592 1
1593 1
    def handle_link_down(self):
1594
        """Handle circuit when link down.
1595 1
1596
        Returns:
1597
            bool: True if the re-deploy was successly otherwise False.
1598
1599
        """
1600
        success = False
1601
        if self.is_using_primary_path():
1602 1
            success = self.deploy_to_backup_path()
1603 1
        elif self.is_using_backup_path():
1604 1
            success = self.deploy_to_primary_path()
1605 1
1606 1
        if not success and self.dynamic_backup_path:
1607
            success = self.deploy_to_path()
1608 1
1609 1
        if success:
1610
            log.debug(f"{self} deployed after link down.")
1611 1
        else:
1612 1
            self.deactivate()
1613
            self.current_path = Path([])
1614 1
            self.sync()
1615 1
            log.debug(f"Failed to re-deploy {self} after link down.")
1616 1
1617 1
        return success
1618
1619 1
    @staticmethod
1620
    def get_interface_from_switch(uni: UNI, switches: dict) -> Interface:
1621 1
        """Get interface from switch by uni"""
1622 1
        switch = switches[uni.interface.switch.dpid]
1623
        interface = switch.interfaces[uni.interface.port_number]
1624 1
        return interface
1625 1
1626 1
    def are_unis_active(self, switches: dict) -> bool:
1627
        """Determine whether this EVC should be active"""
1628 1
        interface_a = self.get_interface_from_switch(self.uni_a, switches)
1629
        interface_z = self.get_interface_from_switch(self.uni_z, switches)
1630 1
        active, _ = self.is_uni_interface_active(interface_a, interface_z)
1631 1
        return active
1632 1
1633 1
    @staticmethod
1634
    def is_uni_interface_active(
1635 1
        *interfaces: Interface
1636 1
    ) -> tuple[bool, dict]:
1637
        """Determine whether a UNI should be active"""
1638
        active = True
1639
        bad_interfaces = [
1640 1
            interface
1641 1
            for interface in interfaces
1642
            if interface.status != EntityStatus.UP
1643
        ]
1644
        if bad_interfaces:
1645
            active = False
1646 1
            interfaces = bad_interfaces
1647 1
        return active, {
1648 1
            interface.id: {
1649 1
                'status': interface.status.value,
1650
                'status_reason': interface.status_reason,
1651
            }
1652
            for interface in interfaces
1653
        }
1654
1655
    def handle_interface_link_up(self, interface: Interface):
1656
        """
1657 1
        Handler for interface link_up events
1658
        """
1659
        if self.archived:  # TODO: Remove when addressing issue #369
1660
            return
1661 1
        if self.is_active():
1662
            return
1663 1
        interfaces = (self.uni_a.interface, self.uni_z.interface)
1664 1
        if interface not in interfaces:
1665 1
            return
1666 1
        down_interfaces = [
1667
            interface
1668 1
            for interface in interfaces
1669
            if interface.status != EntityStatus.UP
1670
        ]
1671
        if down_interfaces:
1672
            return
1673 1
        interface_dicts = {
1674
            interface.id: {
1675 1
                'status': interface.status.value,
1676
                'status_reason': interface.status_reason,
1677
            }
1678
            for interface in interfaces
1679
        }
1680
        self.activate()
1681
        log.info(
1682 1
            f"Activating EVC {self.id}. Interfaces: "
1683 1
            f"{interface_dicts}."
1684
        )
1685
        self.sync()
1686
1687 1
    def handle_interface_link_down(self, interface):
1688
        """
1689 1
        Handler for interface link_down events
1690
        """
1691
        if self.archived:
1692
            return
1693 1
        if not self.is_active():
1694
            return
1695 1
        interfaces = (self.uni_a.interface, self.uni_z.interface)
1696 1
        if interface not in interfaces:
1697 1
            return
1698 1
        down_interfaces = [
1699
            interface
1700 1
            for interface in interfaces
1701
            if interface.status != EntityStatus.UP
1702
        ]
1703
        if not down_interfaces:
1704
            return
1705 1
        interface_dicts = {
1706
            interface.id: {
1707 1
                'status': interface.status.value,
1708
                'status_reason': interface.status_reason,
1709
            }
1710
            for interface in down_interfaces
1711
        }
1712
        self.deactivate()
1713
        log.info(
1714 1
            f"Deactivating EVC {self.id}. Interfaces: "
1715 1
            f"{interface_dicts}."
1716
        )
1717
        self.sync()
1718
1719 1
1720
class EVC(LinkProtection):
1721
    """Class that represents a E-Line Virtual Connection."""
1722