Test Failed
Pull Request — master (#90)
by Antonio
03:04
created

build.models.evc.EVCDeploy._install_uni_flows()   B

Complexity

Conditions 4

Size

Total Lines 62
Code Lines 40

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 24
CRAP Score 4.0218

Importance

Changes 0
Metric Value
cc 4
eloc 40
nop 2
dl 0
loc 62
ccs 24
cts 27
cp 0.8889
crap 4.0218
rs 8.92
c 0
b 0
f 0

How to fix   Long Method   

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

1
"""Classes used in the main application."""  # pylint: disable=too-many-lines
2 1
from datetime import datetime
3 1
from threading import Lock
4 1
from uuid import uuid4
5
6 1
import requests
7 1
from glom import glom
8
9 1
from kytos.core import log
10 1
from kytos.core.common import EntityStatus, GenericEntity
11 1
from kytos.core.exceptions import KytosNoTagAvailableError
12 1
from kytos.core.helpers import get_time, now
13 1
from kytos.core.interface import UNI
14 1
from napps.kytos.mef_eline import settings
15 1
from napps.kytos.mef_eline.exceptions import FlowModException, InvalidPath
16 1
from napps.kytos.mef_eline.storehouse import StoreHouse
17 1
from napps.kytos.mef_eline.utils import compare_endpoint_trace, emit_event
18 1
from .path import Path, DynamicPathManager
19
20
21 1
class EVCBase(GenericEntity):
22
    """Class to represent a circuit."""
23
24 1
    read_only_attributes = [
25
        "creation_time",
26
        "active",
27
        "current_path",
28 1
        "_id",
29
        "archived",
30
    ]
31
    attributes_requiring_redeploy = [
32 1
        "primary_path",
33
        "backup_path",
34 1
        "dynamic_backup_path",
35
        "queue_id",
36
        "priority",
37
    ]
38
    required_attributes = ["name", "uni_a", "uni_z"]
39
40
    def __init__(self, controller, **kwargs):
41
        """Create an EVC instance with the provided parameters.
42
43
        Args:
44
            id(str): EVC identifier. Whether it's None an ID will be genereted.
45
                     Only the first 14 bytes passed will be used.
46
            name: represents an EVC name.(Required)
47
            uni_a (UNI): Endpoint A for User Network Interface.(Required)
48
            uni_z (UNI): Endpoint Z for User Network Interface.(Required)
49
            start_date(datetime|str): Date when the EVC was registred.
50
                                      Default is now().
51
            end_date(datetime|str): Final date that the EVC will be fineshed.
52
                                    Default is None.
53
            bandwidth(int): Bandwidth used by EVC instance. Default is 0.
54
            primary_links(list): Primary links used by evc. Default is []
55
            backup_links(list): Backups links used by evc. Default is []
56
            current_path(list): Circuit being used at the moment if this is an
57
                                active circuit. Default is [].
58
            primary_path(list): primary circuit offered to user IF one or more
59
                                links were provided. Default is [].
60
            backup_path(list): backup circuit offered to the user IF one or
61
                               more links were provided. Default is [].
62
            dynamic_backup_path(bool): Enable computer backup path dynamically.
63
                                       Dafault is False.
64
            creation_time(datetime|str): datetime when the circuit should be
65
                                         activated. default is now().
66
            enabled(Boolean): attribute to indicate the administrative state;
67
                              default is False.
68
            active(Boolean): attribute to indicate the operational state;
69
                             default is False.
70
            archived(Boolean): indicate the EVC has been deleted and is
71
                               archived; default is False.
72
            owner(str): The EVC owner. Default is None.
73 1
            priority(int): Service level provided in the request. Default is 0.
74 1
75
        Raises:
76
            ValueError: raised when object attributes are invalid.
77 1
78 1
        """
79 1
        self._validate(**kwargs)
80 1
        super().__init__()
81
82
        # required attributes
83 1
        self._id = kwargs.get("id", uuid4().hex)[:14]
84 1
        self.uni_a = kwargs.get("uni_a")
85 1
        self.uni_z = kwargs.get("uni_z")
86
        self.name = kwargs.get("name")
87 1
88 1
        # optional attributes
89 1
        self.start_date = get_time(kwargs.get("start_date")) or now()
90 1
        self.end_date = get_time(kwargs.get("end_date")) or None
91 1
        self.queue_id = kwargs.get("queue_id", None)
92 1
93 1
        self.bandwidth = kwargs.get("bandwidth", 0)
94 1
        self.primary_links = Path(kwargs.get("primary_links", []))
95 1
        self.backup_links = Path(kwargs.get("backup_links", []))
96 1
        self.current_path = Path(kwargs.get("current_path", []))
97 1
        self.primary_path = Path(kwargs.get("primary_path", []))
98
        self.backup_path = Path(kwargs.get("backup_path", []))
99 1
        self.dynamic_backup_path = kwargs.get("dynamic_backup_path", False)
100 1
        self.creation_time = get_time(kwargs.get("creation_time")) or now()
101 1
        self.owner = kwargs.get("owner", None)
102
        self.priority = kwargs.get("priority", -1)
103 1
        self.circuit_scheduler = kwargs.get("circuit_scheduler", [])
104
105 1
        self.current_links_cache = set()
106
        self.primary_links_cache = set()
107 1
        self.backup_links_cache = set()
108
109 1
        self.lock = Lock()
110 1
111
        self.archived = kwargs.get("archived", False)
112 1
113 1
        self.metadata = kwargs.get("metadata", {})
114
115 1
        self._storehouse = StoreHouse(controller)
116
        self._controller = controller
117 1
118 1
        if kwargs.get("active", False):
119
            self.activate()
120 1
        else:
121
            self.deactivate()
122
123
        if kwargs.get("enabled", False):
124 1
            self.enable()
125
        else:
126 1
            self.disable()
127
128 1
        # datetime of user request for a EVC (or datetime when object was
129
        # created)
130 1
        self.request_time = kwargs.get("request_time", now())
131 1
        # dict with the user original request (input)
132
        self._requested = kwargs
133 1
134
    def sync(self):
135
        """Sync this EVC in the storehouse."""
136
        self._storehouse.save_evc(self)
137
        log.info(f"EVC {self.id} was synced to the storehouse.")
138
139
    def update(self, **kwargs):
140
        """Update evc attributes.
141
142
        This method will raises an error trying to change the following
143
        attributes: [name, uni_a and uni_z]
144
145
        Returns:
146 1
            the values for enable and a redeploy attribute, if exists and None
147 1
            otherwise
148 1
        Raises:
149 1
            ValueError: message with error detail.
150 1
151 1
        """
152 1
        enable, redeploy = (None, None)
153
        uni_a = kwargs.get("uni_a") or self.uni_a
154 1
        uni_z = kwargs.get("uni_z") or self.uni_z
155 1
        for attribute, value in kwargs.items():
156 1
            if attribute in self.read_only_attributes:
157
                raise ValueError(f"{attribute} can't be updated.")
158 1
            if not hasattr(self, attribute):
159 1
                raise ValueError(f'The attribute "{attribute}" is invalid.')
160
            if attribute in ("primary_path", "backup_path"):
161 1
                try:
162 1
                    value.is_valid(
163 1
                        uni_a.interface.switch, uni_z.interface.switch
164 1
                    )
165
                except InvalidPath as exception:
166 1
                    raise ValueError(
167 1
                        f"{attribute} is not a " f"valid path: {exception}"
168
                    )
169 1
        for attribute, value in kwargs.items():
170 1
            if attribute in ("enable", "enabled"):
171 1
                if value:
172 1
                    self.enable()
173 1
                else:
174
                    self.disable()
175 1
                enable = value
176
            else:
177 1
                setattr(self, attribute, value)
178
                if attribute in self.attributes_requiring_redeploy:
179 1
                    redeploy = value
180
        self.sync()
181
        return enable, redeploy
182
183
    def __repr__(self):
184
        """Repr method."""
185
        return f"EVC({self._id}, {self.name})"
186
187
    def _validate(self, **kwargs):
188
        """Do Basic validations.
189 1
190
        Verify required attributes: name, uni_a, uni_z
191 1
        Verify if the attributes uni_a and uni_z are valid.
192 1
193
        Raises:
194 1
            ValueError: message with error detail.
195 1
196 1
        """
197
        for attribute in self.required_attributes:
198
199 1
            if attribute not in kwargs:
200 1
                raise ValueError(f"{attribute} is required.")
201 1
202 1
            if "uni" in attribute:
203
                uni = kwargs.get(attribute)
204 1
                if not isinstance(uni, UNI):
205
                    raise ValueError(f"{attribute} is an invalid UNI.")
206 1
207
                if not uni.is_valid():
208
                    tag = uni.user_tag.value
209 1
                    message = f"VLAN tag {tag} is not available in {attribute}"
210 1
                    raise ValueError(message)
211 1
212 1
    def __eq__(self, other):
213 1
        """Override the default implementation."""
214
        if not isinstance(other, EVC):
215 1
            return False
216
217 1
        attrs_to_compare = ["name", "uni_a", "uni_z", "owner", "bandwidth"]
218
        for attribute in attrs_to_compare:
219 1
            if getattr(other, attribute) != getattr(self, attribute):
220
                return False
221
        return True
222 1
223
    def shares_uni(self, other):
224 1
        """Check if two EVCs share an UNI."""
225
        if other.uni_a in (self.uni_a, self.uni_z) or other.uni_z in (
226
            self.uni_a,
227
            self.uni_z,
228 1
        ):
229
            return True
230 1
        return False
231 1
232 1
    def as_dict(self):
233
        """Return a dictionary representing an EVC object."""
234 1
        evc_dict = {
235 1
            "id": self.id,
236 1
            "name": self.name,
237
            "uni_a": self.uni_a.as_dict(),
238 1
            "uni_z": self.uni_z.as_dict(),
239 1
        }
240 1
241 1
        time_fmt = "%Y-%m-%dT%H:%M:%S"
242 1
243 1
        evc_dict["start_date"] = self.start_date
244 1
        if isinstance(self.start_date, datetime):
245 1
            evc_dict["start_date"] = self.start_date.strftime(time_fmt)
246 1
247
        evc_dict["end_date"] = self.end_date
248
        if isinstance(self.end_date, datetime):
249
            evc_dict["end_date"] = self.end_date.strftime(time_fmt)
250
251
        evc_dict["queue_id"] = self.queue_id
252
        evc_dict["bandwidth"] = self.bandwidth
253
        evc_dict["primary_links"] = self.primary_links.as_dict()
254
        evc_dict["backup_links"] = self.backup_links.as_dict()
255 1
        evc_dict["current_path"] = self.current_path.as_dict()
256 1
        evc_dict["primary_path"] = self.primary_path.as_dict()
257 1
        evc_dict["backup_path"] = self.backup_path.as_dict()
258
        evc_dict["dynamic_backup_path"] = self.dynamic_backup_path
259 1
        evc_dict["metadata"] = self.metadata
260 1
261
        # if self._requested:
262 1
        #     request_dict = self._requested.copy()
263 1
        #     request_dict['uni_a'] = request_dict['uni_a'].as_dict()
264
        #     request_dict['uni_z'] = request_dict['uni_z'].as_dict()
265
        #     request_dict['circuit_scheduler'] = self.circuit_scheduler
266 1
        #     evc_dict['_requested'] = request_dict
267 1
268 1
        evc_dict["request_time"] = self.request_time
269 1
        if isinstance(self.request_time, datetime):
270
            evc_dict["request_time"] = self.request_time.strftime(time_fmt)
271 1
272
        time = self.creation_time.strftime(time_fmt)
273 1
        evc_dict["creation_time"] = time
274
275
        evc_dict["owner"] = self.owner
276 1
        evc_dict["circuit_scheduler"] = [
277
            sc.as_dict() for sc in self.circuit_scheduler
278 1
        ]
279
280 1
        evc_dict["active"] = self.is_active()
281
        evc_dict["enabled"] = self.is_enabled()
282
        evc_dict["archived"] = self.archived
283
        evc_dict["priority"] = self.priority
284 1
285
        return evc_dict
286
287 1
    @property
288
    def id(self):  # pylint: disable=invalid-name
289
        """Return this EVC's ID."""
290 1
        return self._id
291
292
    def archive(self):
293
        """Archive this EVC on deletion."""
294 1
        self.archived = True
295
296
297 1
# pylint: disable=fixme, too-many-public-methods
298
class EVCDeploy(EVCBase):
299
    """Class to handle the deploy procedures."""
300 1
301
    def create(self):
302
        """Create a EVC."""
303
304 1
    def discover_new_paths(self):
305
        """Discover new paths to satisfy this circuit and deploy it."""
306
        return DynamicPathManager.get_best_paths(self)
307
308 1
    def change_path(self):
309
        """Change EVC path."""
310
311
    def reprovision(self):
312
        """Force the EVC (re-)provisioning."""
313 1
314
    def is_affected_by_link(self, link):
315
        """Return True if this EVC has the given link on its current path."""
316
        return link in self.current_path
317 1
318
    def link_affected_by_interface(self, interface):
319
        """Return True if this EVC has the given link on its current path."""
320
        return self.current_path.link_affected_by_interface(interface)
321 1
322
    def is_backup_path_affected_by_link(self, link):
323
        """Return True if the backup path of this EVC uses the given link."""
324
        return link in self.backup_path
325 1
326
    # pylint: disable=invalid-name
327
    def is_primary_path_affected_by_link(self, link):
328
        """Return True if the primary path of this EVC uses the given link."""
329
        return link in self.primary_path
330
331
    def is_using_primary_path(self):
332
        """Verify if the current deployed path is self.primary_path."""
333
        return self.primary_path and (self.current_path == self.primary_path)
334 1
335
    def is_using_backup_path(self):
336
        """Verify if the current deployed path is self.backup_path."""
337
        return self.backup_path and (self.current_path == self.backup_path)
338
339
    def is_using_dynamic_path(self):
340
        """Verify if the current deployed path is a dynamic path."""
341
        if (
342
            self.current_path
343
            and not self.is_using_primary_path()
344 1
            and not self.is_using_backup_path()
345
            and self.current_path.status == EntityStatus.UP
346
        ):
347
            return True
348 1
        return False
349 1
350 1
    def deploy_to_backup_path(self):
351
        """Deploy the backup path into the datapaths of this circuit.
352 1
353 1
        If the backup_path attribute is valid and up, this method will try to
354
        deploy this backup_path.
355 1
356
        If everything fails and dynamic_backup_path is True, then tries to
357 1
        deploy a dynamic path.
358
        """
359
        # TODO: Remove flows from current (cookies)
360
        if self.is_using_backup_path():
361 1
            # TODO: Log to say that cannot move backup to backup
362
            return True
363
364
        success = False
365
        if self.backup_path.status is EntityStatus.UP:
366
            success = self.deploy_to_path(self.backup_path)
367
368 1
        if success:
369
            return True
370
371
        if (
372 1
            self.dynamic_backup_path
373 1
            or self.uni_a.interface.switch == self.uni_z.interface.switch
374
        ):
375
            return self.deploy_to_path()
376 1
377
        return False
378
379
    def deploy_to_primary_path(self):
380
        """Deploy the primary path into the datapaths of this circuit.
381
382
        If the primary_path attribute is valid and up, this method will try to
383
        deploy this primary_path.
384
        """
385
        # TODO: Remove flows from current (cookies)
386
        if self.is_using_primary_path():
387
            # TODO: Log to say that cannot move primary to primary
388
            return True
389
390
        if self.primary_path.status is EntityStatus.UP:
391
            return self.deploy_to_path(self.primary_path)
392
        return False
393 1
394
    def deploy(self):
395
        """Deploy EVC to best path.
396
397
        Best path can be the primary path, if available. If not, the backup
398
        path, and, if it is also not available, a dynamic path.
399
        """
400
        if self.archived:
401
            return False
402
        self.enable()
403
        success = self.deploy_to_primary_path()
404
        if not success:
405
            success = self.deploy_to_backup_path()
406
407
        if success:
408
            emit_event(self._controller, "deployed", evc_id=self.id)
409
        return success
410 1
411
    @staticmethod
412 1
    def get_path_status(path):
413 1
        """Check for the current status of a path.
414 1
415 1
        If any link in this path is down, the path is considered down.
416
        """
417 1
        if not path:
418
            return EntityStatus.DISABLED
419 1
420
        for link in path:
421 1
            if link.status is not EntityStatus.UP:
422 1
                return link.status
423 1
        return EntityStatus.UP
424 1
425 1
    #    def discover_new_path(self):
426 1
    #        # TODO: discover a new path to satisfy this circuit and deploy
427 1
428
    def remove(self):
429 1
        """Remove EVC path and disable it."""
430
        self.remove_current_flows()
431
        self.disable()
432 1
        self.sync()
433 1
        emit_event(self._controller, "undeployed", evc_id=self.id)
434 1
435
    def remove_current_flows(self, current_path=None, force=True):
436
        """Remove all flows from current path."""
437
        switches = set()
438
439 1
        switches.add(self.uni_a.interface.switch)
440 1
        switches.add(self.uni_z.interface.switch)
441 1
        if not current_path:
442 1
            current_path = self.current_path
443
        for link in current_path:
444 1
            switches.add(link.endpoint_a.switch)
445 1
            switches.add(link.endpoint_b.switch)
446
447 1
        match = {
448
            "cookie": self.get_cookie(),
449 1
            "cookie_mask": 18446744073709551615,
450
        }
451 1
452
        for switch in switches:
453 1
            try:
454 1
                self._send_flow_mods(switch, [match], 'delete', force=force)
455 1
            except FlowModException:
456
                log.error(
457 1
                    f"Error removing flows from switch {switch.id} for"
458 1
                    f"EVC {self}"
459 1
                )
460
461 1
        current_path.make_vlans_available()
462 1
        self.current_path = Path([])
463 1
        self.deactivate()
464
        self.sync()
465 1
466
    @staticmethod
467 1
    def links_zipped(path=None):
468
        """Return an iterator which yields pairs of links in order."""
469
        if not path:
470
            return []
471
        return zip(path[:-1], path[1:])
472
473
    def should_deploy(self, path=None):
474
        """Verify if the circuit should be deployed."""
475
        if not path:
476
            log.debug("Path is empty.")
477
            return False
478
479
        if not self.is_enabled():
480
            log.debug(f"{self} is disabled.")
481
            return False
482 1
483 1
        if not self.is_active():
484 1
            log.debug(f"{self} will be deployed.")
485 1
            return True
486 1
487
        return False
488
489
    def deploy_to_path(self, path=None):
490 1
        """Install the flows for this circuit.
491 1
492
        Procedures to deploy:
493 1
494 1
        0. Remove current flows installed
495 1
        1. Decide if will deploy "path" or discover a new path
496
        2. Choose vlan
497
        3. Install NNI flows
498
        4. Install UNI flows
499 1
        5. Activate
500
        6. Update current_path
501 1
        7. Update links caches(primary, current, backup)
502 1
503 1
        """
504 1
        self.remove_current_flows()
505 1
        use_path = path
506
        if self.should_deploy(use_path):
507
            try:
508
                use_path.choose_vlans()
509 1
            except KytosNoTagAvailableError:
510
                use_path = None
511 1
        else:
512 1
            for use_path in self.discover_new_paths():
513 1
                if use_path is None:
514 1
                    continue
515 1
                try:
516 1
                    use_path.choose_vlans()
517 1
                    break
518 1
                except KytosNoTagAvailableError:
519 1
                    pass
520 1
            else:
521
                use_path = None
522 1
523
        try:
524
            if use_path:
525
                self._install_nni_flows(use_path)
526
                self._install_uni_flows(use_path)
527
            elif self.uni_a.interface.switch == self.uni_z.interface.switch:
528
                use_path = Path()
529
                self._install_direct_uni_flows()
530
            else:
531
                log.warn(
532
                    f"{self} was not deployed. " "No available path was found."
533
                )
534
                return False
535
        except FlowModException:
536
            log.error(f"Error deploying EVC {self} when calling flow_manager.")
537
            self.remove_current_flows(use_path)
538
            return False
539
        self.activate()
540
        self.current_path = use_path
541
        self.sync()
542
        log.info(f"{self} was deployed.")
543
        return True
544
545
    def _install_direct_uni_flows(self):
546
        """Install flows connecting two UNIs.
547
548
        This case happens when the circuit is between UNIs in the
549
        same switch.
550
        """
551
        vlan_a = self.uni_a.user_tag.value if self.uni_a.user_tag else None
552
        vlan_z = self.uni_z.user_tag.value if self.uni_z.user_tag else None
553
554
        flow_mod_az = self._prepare_flow_mod(
555
            self.uni_a.interface, self.uni_z.interface, self.queue_id
556
        )
557
        flow_mod_za = self._prepare_flow_mod(
558 1
            self.uni_z.interface, self.uni_a.interface, self.queue_id
559
        )
560 1
561 1
        if vlan_a and vlan_z:
562 1
            flow_mod_az["match"]["dl_vlan"] = vlan_a
563
            flow_mod_za["match"]["dl_vlan"] = vlan_z
564 1
            flow_mod_az["actions"].insert(
565
                0, {"action_type": "set_vlan", "vlan_id": vlan_z}
566 1
            )
567
            flow_mod_za["actions"].insert(
568
                0, {"action_type": "set_vlan", "vlan_id": vlan_a}
569
            )
570
        elif vlan_a:
571
            flow_mod_az["match"]["dl_vlan"] = vlan_a
572 1
            flow_mod_az["actions"].insert(0, {"action_type": "pop_vlan"})
573
            flow_mod_za["actions"].insert(
574
                0, {"action_type": "set_vlan", "vlan_id": vlan_a}
575
            )
576 1
        elif vlan_z:
577
            flow_mod_za["match"]["dl_vlan"] = vlan_z
578 1
            flow_mod_za["actions"].insert(0, {"action_type": "pop_vlan"})
579
            flow_mod_az["actions"].insert(
580 1
                0, {"action_type": "set_vlan", "vlan_id": vlan_z}
581
            )
582
        self._send_flow_mods(
583
            self.uni_a.interface.switch, [flow_mod_az, flow_mod_za]
584
        )
585 1
586 1
    def _install_nni_flows(self, path=None):
587
        """Install NNI flows."""
588 1
        for incoming, outcoming in self.links_zipped(path):
589 1
            in_vlan = incoming.get_metadata("s_vlan").value
590
            out_vlan = outcoming.get_metadata("s_vlan").value
591
592 1
            flows = []
593
            # Flow for one direction
594
            flows.append(
595 1
                self._prepare_nni_flow(
596
                    incoming.endpoint_b,
597
                    outcoming.endpoint_a,
598
                    in_vlan,
599 1
                    out_vlan,
600
                    queue_id=self.queue_id,
601
                )
602 1
            )
603
604
            # Flow for the other direction
605
            flows.append(
606 1
                self._prepare_nni_flow(
607
                    outcoming.endpoint_a,
608 1
                    incoming.endpoint_b,
609
                    out_vlan,
610
                    in_vlan,
611 1
                    queue_id=self.queue_id,
612
                )
613
            )
614 1
            self._send_flow_mods(incoming.endpoint_b.switch, flows)
615
616
    def _install_uni_flows(self, path=None):
617
        """Install UNI flows."""
618 1
        if not path:
619
            log.info("install uni flows without path.")
620
            return
621 1
622
        # Determine VLANs
623
        in_vlan_a = self.uni_a.user_tag.value if self.uni_a.user_tag else None
624
        out_vlan_a = path[0].get_metadata("s_vlan").value
625 1
626
        in_vlan_z = self.uni_z.user_tag.value if self.uni_z.user_tag else None
627 1
        out_vlan_z = path[-1].get_metadata("s_vlan").value
628
629 1
        # Flows for the first UNI
630 1
        flows_a = []
631
632
        # Flow for one direction, pushing the service tag
633
        push_flow = self._prepare_push_flow(
634
            self.uni_a.interface,
635
            path[0].endpoint_a,
636
            in_vlan_a,
637
            out_vlan_a,
638
            queue_id=self.queue_id,
639
        )
640 1
        flows_a.append(push_flow)
641
642 1
        # Flow for the other direction, popping the service tag
643 1
        pop_flow = self._prepare_pop_flow(
644 1
            path[0].endpoint_a,
645
            self.uni_a.interface,
646
            in_vlan_a,
647 1
            out_vlan_a,
648
            queue_id=self.queue_id,
649 1
        )
650
        flows_a.append(pop_flow)
651 1
652
        self._send_flow_mods(self.uni_a.interface.switch, flows_a)
653
654 1
        # Flows for the second UNI
655 1
        flows_z = []
656
657 1
        # Flow for one direction, pushing the service tag
658
        push_flow = self._prepare_push_flow(
659 1
            self.uni_z.interface,
660
            path[-1].endpoint_b,
661 1
            in_vlan_z,
662
            out_vlan_z,
663
            queue_id=self.queue_id,
664
        )
665
        flows_z.append(push_flow)
666 1
667
        # Flow for the other direction, popping the service tag
668
        pop_flow = self._prepare_pop_flow(
669 1
            path[-1].endpoint_b,
670
            self.uni_z.interface,
671
            in_vlan_z,
672 1
            out_vlan_z,
673
            queue_id=self.queue_id,
674 1
        )
675
        flows_z.append(pop_flow)
676 1
677 1
        self._send_flow_mods(self.uni_z.interface.switch, flows_z)
678
679 1
    @staticmethod
680
    def _send_flow_mods(switch, flow_mods, command='flows', force=False):
681 1
        """Send a flow_mod list to a specific switch.
682
683 1
        Args:
684
            switch(Switch): The target of flows.
685 1
            flow_mods(dict): Python dictionary with flow_mods.
686
            command(str): By default is 'flows'. To remove a flow is 'remove'.
687 1
            force(bool): True to send via consistency check in case of conn errors
688
689
        """
690
        endpoint = f"{settings.MANAGER_URL}/{command}/{switch.id}"
691
692
        data = {"flows": flow_mods, "force": force}
693
        response = requests.post(endpoint, json=data)
694
        if response.status_code >= 400:
695
            raise FlowModException
696
697
    def get_cookie(self):
698
        """Return the cookie integer from evc id."""
699
        return int(self.id, 16) + (settings.COOKIE_PREFIX << 56)
700
701 1
    @staticmethod
702
    def get_id_from_cookie(cookie):
703 1
        """Return the evc id given a cookie value."""
704
        evc_id = cookie - (settings.COOKIE_PREFIX << 56)
705
        return f"{evc_id:x}"
706
707 1
    def _prepare_flow_mod(self, in_interface, out_interface, queue_id=None):
708 1
        """Prepare a common flow mod."""
709
        default_actions = [
710 1
            {"action_type": "output", "port": out_interface.port_number}
711 1
        ]
712
        if queue_id:
713 1
            default_actions.append(
714
                {"action_type": "set_queue", "queue_id": queue_id}
715 1
            )
716 1
717 1
        flow_mod = {
718 1
            "match": {"in_port": in_interface.port_number},
719
            "cookie": self.get_cookie(),
720 1
            "actions": default_actions,
721
        }
722
        if self.priority > -1:
723
            flow_mod["priority"] = self.priority
724 1
725
        return flow_mod
726 1
727 1
    def _prepare_nni_flow(self, *args, queue_id=None):
728 1
        """Create NNI flows."""
729 1
        in_interface, out_interface, in_vlan, out_vlan = args
730 1
        flow_mod = self._prepare_flow_mod(
731 1
            in_interface, out_interface, queue_id
732 1
        )
733 1
        flow_mod["match"]["dl_vlan"] = in_vlan
734 1
735
        new_action = {"action_type": "set_vlan", "vlan_id": out_vlan}
736 1
        flow_mod["actions"].insert(0, new_action)
737
738
        return flow_mod
739
740 View Code Duplication
    def _prepare_push_flow(self, *args, queue_id=None):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
741
        """Prepare push flow.
742
743
        Arguments:
744
            in_interface(str): Interface input.
745
            out_interface(str): Interface output.
746
            in_vlan(str): Vlan input.
747
            out_vlan(str): Vlan output.
748
749
        Return:
750
            dict: An python dictionary representing a FlowMod
751
752
        """
753
        # assign all arguments
754
        in_interface, out_interface, in_vlan, out_vlan = args
755
756
        flow_mod = self._prepare_flow_mod(
757
            in_interface, out_interface, queue_id
758
        )
759 1
760
        # the service tag must be always pushed
761
        new_action = {"action_type": "set_vlan", "vlan_id": out_vlan}
762
        flow_mod["actions"].insert(0, new_action)
763
764
        new_action = {"action_type": "push_vlan", "tag_type": "s"}
765
        flow_mod["actions"].insert(0, new_action)
766
767
        if in_vlan:
768
            # if in_vlan is set, it must be included in the match
769
            flow_mod["match"]["dl_vlan"] = in_vlan
770
            new_action = {"action_type": "pop_vlan"}
771
            flow_mod["actions"].insert(0, new_action)
772
        return flow_mod
773
774 View Code Duplication
    def _prepare_pop_flow(
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
775
        self, in_interface, out_interface, in_vlan, out_vlan, queue_id=None
776
    ):
777
        # pylint: disable=too-many-arguments
778
        """Prepare pop flow."""
779
        flow_mod = self._prepare_flow_mod(
780
            in_interface, out_interface, queue_id
781
        )
782
        flow_mod["match"]["dl_vlan"] = out_vlan
783
        if in_vlan:
784
            new_action = {"action_type": "set_vlan", "vlan_id": in_vlan}
785
            flow_mod["actions"].insert(0, new_action)
786
            new_action = {"action_type": "push_vlan", "tag_type": "c"}
787 1
            flow_mod["actions"].insert(0, new_action)
788
        new_action = {"action_type": "pop_vlan"}
789
        flow_mod["actions"].insert(0, new_action)
790 1
        return flow_mod
791
792
    @staticmethod
793
    def run_sdntrace(uni):
794 1
        """Run SDN trace on control plane starting from EVC UNIs."""
795
        endpoint = f"{settings.SDN_TRACE_CP_URL}/trace"
796 1
        data_uni = {
797
            "trace": {
798 1
                "switch": {
799
                    "dpid": uni.interface.switch.dpid,
800 1
                    "in_port": uni.interface.port_number,
801
                }
802 1
            }
803
        }
804 1
        if uni.user_tag:
805
            data_uni["trace"]["eth"] = {
806
                "dl_type": 0x8100,
807
                "dl_vlan": uni.user_tag.value,
808
            }
809 1
        response = requests.put(endpoint, json=data_uni)
810
        if response.status_code >= 400:
811 1
            log.error(f"Failed to run sdntrace-cp: {response.text}")
812
            return []
813 1
        return response.json().get('result', [])
814 1
815 1
    def check_traces(self):
816
        """Check if current_path is deployed comparing with SDN traces."""
817 1
        trace_a = self.run_sdntrace(self.uni_a)
818 1
        if len(trace_a) != len(self.current_path) + 1:
819
            log.warn(f"Invalid trace from uni_a: {trace_a}")
820 1
            return False
821
        trace_z = self.run_sdntrace(self.uni_z)
822 1
        if len(trace_z) != len(self.current_path) + 1:
823
            log.warn(f"Invalid trace from uni_z: {trace_z}")
824
            return False
825
826
        for link, trace1, trace2 in zip(self.current_path,
827
                                        trace_a[1:],
828
                                        trace_z[:0:-1]):
829 1
            if compare_endpoint_trace(
830 1
               link.endpoint_a,
831
               glom(link.metadata, 's_vlan.value'), trace2) is False:
832 1
                log.warn(f"Invalid trace from uni_a: {trace_a}")
833 1
                return False
834 1
            if compare_endpoint_trace(
835
               link.endpoint_b,
836 1
               glom(link.metadata, 's_vlan.value'), trace1) is False:
837 1
                log.warn(f"Invalid trace from uni_z: {trace_z}")
838
                return False
839
840
        return True
841 1
842
843
class LinkProtection(EVCDeploy):
844
    """Class to handle link protection."""
845
846 1
    def is_affected_by_link(self, link=None):
847 1
        """Verify if the current path is affected by link down event."""
848
        return self.current_path.is_affected_by_link(link)
849 1
850 1
    def is_using_primary_path(self):
851
        """Verify if the current deployed path is self.primary_path."""
852
        return self.current_path == self.primary_path
853
854
    def is_using_backup_path(self):
855
        """Verify if the current deployed path is self.backup_path."""
856
        return self.current_path == self.backup_path
857
858
    def is_using_dynamic_path(self):
859 1
        """Verify if the current deployed path is dynamic."""
860
        if (
861
            self.current_path
862
            and not self.is_using_primary_path()
863
            and not self.is_using_backup_path()
864
            and self.current_path.status is EntityStatus.UP
865
        ):
866 1
            return True
867 1
        return False
868 1
869 1
    def deploy_to(self, path_name=None, path=None):
870 1
        """Create a deploy to path."""
871
        if self.current_path == path:
872 1
            log.debug(f"{path_name} is equal to current_path.")
873 1
            return True
874
875 1
        if path.status is EntityStatus.UP:
876 1
            return self.deploy_to_path(path)
877
878 1
        return False
879 1
880 1
    def handle_link_up(self, link):
881 1
        """Handle circuit when link down.
882
883 1
        Args:
884
            link(Link): Link affected by link.down event.
885
886 1
        """
887
        if self.is_using_primary_path():
888
            return True
889
890
        success = False
891
        if self.primary_path.is_affected_by_link(link):
892
            success = self.deploy_to_primary_path()
893
894
        if success:
895
            return True
896
897
        # We tried to deploy(primary_path) without success.
898
        # And in this case is up by some how. Nothing to do.
899
        if self.is_using_backup_path() or self.is_using_dynamic_path():
900
            return True
901
902
        # In this case, probably the circuit is not being used and
903
        # we can move to backup
904
        if self.backup_path.is_affected_by_link(link):
905
            success = self.deploy_to_backup_path()
906
907
        if success:
908
            return True
909
910
        # In this case, the circuit is not being used and we should
911
        # try a dynamic path
912
        if self.dynamic_backup_path:
913
            return self.deploy_to_path()
914
915
        return True
916
917
    def handle_link_down(self):
918
        """Handle circuit when link down.
919
920
        Returns:
921
            bool: True if the re-deploy was successly otherwise False.
922
923
        """
924
        success = False
925
        if self.is_using_primary_path():
926
            success = self.deploy_to_backup_path()
927
        elif self.is_using_backup_path():
928
            success = self.deploy_to_primary_path()
929
930
        if not success and self.dynamic_backup_path:
931
            success = self.deploy_to_path()
932
933
        if success:
934
            log.debug(f"{self} deployed after link down.")
935
        else:
936
            self.deactivate()
937
            self.current_path = Path([])
938
            self.sync()
939
            log.debug(f"Failed to re-deploy {self} after link down.")
940
941
        return success
942
943
944
class EVC(LinkProtection):
945
    """Class that represents a E-Line Virtual Connection."""
946