Test Failed
Pull Request — master (#396)
by
unknown
06:24
created

LinkProtection.get_interface_from_switch()   A

Complexity

Conditions 1

Size

Total Lines 6
Code Lines 5

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 0
CRAP Score 2

Importance

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