Passed
Push — master ( 7cf1bc...b1dc2a )
by Vinicius
03:59 queued 13s
created

build.main   F

Complexity

Total Complexity 164

Size/Duplication

Total Lines 976
Duplicated Lines 0 %

Test Coverage

Coverage 93.99%

Importance

Changes 0
Metric Value
wmc 164
eloc 624
dl 0
loc 976
ccs 532
cts 566
cp 0.9399
rs 1.976
c 0
b 0
f 0

42 Methods

Rating   Name   Duplication   Size   Complexity  
A Main.setup() 0 26 1
A Main.list_circuits() 0 14 1
A Main.on_flow_delete() 0 4 1
A Main.get_circuit() 0 12 2
A Main.shutdown() 0 2 1
A Main.execute() 0 8 3
A Main.get_evcs_by_svc_level() 0 7 2
A Main.get_eline_controller() 0 4 1
C Main.create_circuit() 0 97 10
D Main.execute_consistency() 0 32 13
A Main.handle_flow_delete() 0 7 2
B Main.add_metadata() 0 30 6
C Main._evc_dict_with_instances() 0 41 9
A Main.on_link_down() 0 4 1
A Main.handle_evc_deployed() 0 7 3
A Main.update_schedule() 0 50 3
A Main._evc_from_dict() 0 3 1
A Main._find_evc_by_schedule_id() 0 21 5
A Main.load_all_evcs() 0 6 3
A Main.get_metadata() 0 10 2
A Main.delete_metadata() 0 11 2
A Main.handle_link_up() 0 7 5
A Main.on_link_up() 0 4 1
A Main.on_evc_affected_by_link_down() 0 4 1
A Main.delete_schedule() 0 35 3
A Main.handle_evc_affected_by_link_down() 0 14 4
F Main.handle_link_down() 0 75 15
A Main.on_evc_deployed() 0 4 1
A Main._json_from_request() 0 23 4
B Main.list_schedules() 0 31 5
B Main._link_from_dict() 0 26 6
A Main._uni_from_dict() 0 22 4
A Main.redeploy() 0 20 4
A Main.handle_flow_mod_error() 0 9 3
A Main._is_duplicated_evc() 0 14 4
D Main.update() 0 60 13
A Main.on_flow_mod_error() 0 4 1
A Main.on_topology_loaded() 0 4 1
A Main._load_evc() 0 17 3
B Main.create_schedule() 0 76 7
A Main._get_circuits_buffer() 0 13 3
A Main.delete_circuit() 0 38 4

How to fix   Complexity   

Complexity

Complex classes like build.main often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
# pylint: disable=protected-access
2 1
"""Main module of kytos/mef_eline Kytos Network Application.
3
4
NApp to provision circuits from user request.
5
"""
6 1
import time
7 1
from threading import Lock
8
9 1
from flask import jsonify, request
10 1
from werkzeug.exceptions import (BadRequest, Conflict, Forbidden,
11
                                 MethodNotAllowed, NotFound,
12
                                 UnsupportedMediaType)
13
14 1
from kytos.core import KytosNApp, log, rest
15 1
from kytos.core.helpers import listen_to
16 1
from kytos.core.interface import TAG, UNI
17 1
from kytos.core.link import Link
18 1
from napps.kytos.mef_eline import controllers, settings
19 1
from napps.kytos.mef_eline.exceptions import InvalidPath
20 1
from napps.kytos.mef_eline.models import (EVC, DynamicPathManager, EVCDeploy,
21
                                          Path)
22 1
from napps.kytos.mef_eline.scheduler import CircuitSchedule, Scheduler
23 1
from napps.kytos.mef_eline.utils import emit_event, load_spec, validate
24
25
26
# pylint: disable=too-many-public-methods
27 1
class Main(KytosNApp):
28
    """Main class of amlight/mef_eline NApp.
29
30
    This class is the entry point for this napp.
31
    """
32
33 1
    spec = load_spec()
34
35 1
    def setup(self):
36
        """Replace the '__init__' method for the KytosNApp subclass.
37
38
        The setup method is automatically called by the controller when your
39
        application is loaded.
40
41
        So, if you have any setup routine, insert it here.
42
        """
43
        # object used to scheduler circuit events
44 1
        self.sched = Scheduler()
45
46
        # object to save and load circuits
47 1
        self.mongo_controller = self.get_eline_controller()
48 1
        self.mongo_controller.bootstrap_indexes()
49
50
        # set the controller that will manager the dynamic paths
51 1
        DynamicPathManager.set_controller(self.controller)
52
53
        # dictionary of EVCs created. It acts as a circuit buffer.
54
        # Every create/update/delete must be synced to mongodb.
55 1
        self.circuits = {}
56
57 1
        self._lock = Lock()
58 1
        self.execute_as_loop(settings.DEPLOY_EVCS_INTERVAL)
59
60 1
        self.load_all_evcs()
61
62 1
    def get_evcs_by_svc_level(self) -> list:
63
        """Get circuits sorted by desc service level and asc creation_time.
64
65
        In the future, as more ops are offloaded it should be get from the DB.
66
        """
67 1
        return sorted(self.circuits.values(),
68
                      key=lambda x: (-x.service_level, x.creation_time))
69
70 1
    @staticmethod
71 1
    def get_eline_controller():
72
        """Return the ELineController instance."""
73
        return controllers.ELineController()
74
75 1
    def execute(self):
76
        """Execute once when the napp is running."""
77 1
        if self._lock.locked():
78 1
            return
79 1
        log.debug("Starting consistency routine")
80 1
        with self._lock:
81 1
            self.execute_consistency()
82 1
        log.debug("Finished consistency routine")
83
84 1
    def execute_consistency(self):
85
        """Execute consistency routine."""
86 1
        circuits_to_check = {}
87 1
        stored_circuits = self.mongo_controller.get_circuits()['circuits']
88 1
        for circuit in self.get_evcs_by_svc_level():
89 1
            stored_circuits.pop(circuit.id, None)
90 1
            if (
91
                circuit.is_enabled()
92
                and not circuit.is_active()
93
                and not circuit.lock.locked()
94
                and not circuit.has_recent_removed_flow()
95
                and not circuit.is_recent_updated()
96
            ):
97 1
                circuits_to_check[circuit.id] = circuit
98 1
        circuits_checked = EVCDeploy.check_list_traces(circuits_to_check)
99 1
        for circuit_id, circuit in circuits_to_check.items():
100 1
            is_checked = circuits_checked.get(circuit_id)
101 1
            if is_checked:
102 1
                circuit.execution_rounds = 0
103 1
                log.info(f"{circuit} enabled but inactive - activating")
104 1
                with circuit.lock:
105 1
                    circuit.activate()
106 1
                    circuit.sync()
107
            else:
108 1
                circuit.execution_rounds += 1
109 1
                if circuit.execution_rounds > settings.WAIT_FOR_OLD_PATH:
110 1
                    log.info(f"{circuit} enabled but inactive - redeploy")
111 1
                    with circuit.lock:
112 1
                        circuit.deploy()
113 1
        for circuit_id in stored_circuits:
114 1
            log.info(f"EVC found in mongodb but unloaded {circuit_id}")
115 1
            self._load_evc(stored_circuits[circuit_id])
116
117 1
    def shutdown(self):
118
        """Execute when your napp is unloaded.
119
120
        If you have some cleanup procedure, insert it here.
121
        """
122
123 1
    @rest("/v2/evc/", methods=["GET"])
124 1
    def list_circuits(self):
125
        """Endpoint to return circuits stored.
126
127
        archive query arg if defined (not null) will be filtered
128
        accordingly, by default only non archived evcs will be listed
129
        """
130 1
        log.debug("list_circuits /v2/evc")
131 1
        archived = request.args.get("archived", "false").lower()
132 1
        archived_to_optional = {"null": None, "true": True, "false": False}
133 1
        archived = archived_to_optional.get(archived, False)
134 1
        circuits = self.mongo_controller.get_circuits(archived=archived)
135 1
        circuits = circuits['circuits']
136 1
        return jsonify(circuits), 200
137
138 1
    @rest("/v2/evc/<circuit_id>", methods=["GET"])
139 1
    def get_circuit(self, circuit_id):
140
        """Endpoint to return a circuit based on id."""
141 1
        log.debug("get_circuit /v2/evc/%s", circuit_id)
142 1
        circuit = self.mongo_controller.get_circuit(circuit_id)
143 1
        if not circuit:
144 1
            result = f"circuit_id {circuit_id} not found"
145 1
            log.debug("get_circuit result %s %s", result, 404)
146 1
            raise NotFound(result)
147 1
        status = 200
148 1
        log.debug("get_circuit result %s %s", circuit, status)
149 1
        return jsonify(circuit), status
150
151 1
    @rest("/v2/evc/", methods=["POST"])
152 1
    @validate(spec)
153 1
    def create_circuit(self, data):
154
        """Try to create a new circuit.
155
156
        Firstly, for EVPL: E-Line NApp verifies if UNI_A's requested C-VID and
157
        UNI_Z's requested C-VID are available from the interfaces' pools. This
158
        is checked when creating the UNI object.
159
160
        Then, E-Line NApp requests a primary and a backup path to the
161
        Pathfinder NApp using the attributes primary_links and backup_links
162
        submitted via REST
163
164
        # For each link composing paths in #3:
165
        #  - E-Line NApp requests a S-VID available from the link VLAN pool.
166
        #  - Using the S-VID obtained, generate abstract flow entries to be
167
        #    sent to FlowManager
168
169
        Push abstract flow entries to FlowManager and FlowManager pushes
170
        OpenFlow entries to datapaths
171
172
        E-Line NApp generates an event to notify all Kytos NApps of a new EVC
173
        creation
174
175
        Finnaly, notify user of the status of its request.
176
        """
177
        # Try to create the circuit object
178 1
        log.debug("create_circuit /v2/evc/")
179
180 1
        try:
181 1
            evc = self._evc_from_dict(data)
182 1
        except ValueError as exception:
183 1
            log.debug("create_circuit result %s %s", exception, 400)
184 1
            raise BadRequest(str(exception)) from BadRequest
185
186 1
        if evc.primary_path:
187
            try:
188
                evc.primary_path.is_valid(
189
                    evc.uni_a.interface.switch,
190
                    evc.uni_z.interface.switch,
191
                    bool(evc.circuit_scheduler),
192
                )
193
            except InvalidPath as exception:
194
                raise BadRequest(
195
                    f"primary_path is not valid: {exception}"
196
                ) from exception
197 1
        if evc.backup_path:
198
            try:
199
                evc.backup_path.is_valid(
200
                    evc.uni_a.interface.switch,
201
                    evc.uni_z.interface.switch,
202
                    bool(evc.circuit_scheduler),
203
                )
204
            except InvalidPath as exception:
205
                raise BadRequest(
206
                    f"backup_path is not valid: {exception}"
207
                ) from exception
208
209
        # verify duplicated evc
210 1
        if self._is_duplicated_evc(evc):
211 1
            result = "The EVC already exists."
212 1
            log.debug("create_circuit result %s %s", result, 409)
213 1
            raise Conflict(result)
214
215 1
        try:
216 1
            evc._validate_has_primary_or_dynamic()
217 1
        except ValueError as exception:
218 1
            raise BadRequest(str(exception)) from exception
219
220
        # store circuit in dictionary
221 1
        self.circuits[evc.id] = evc
222
223
        # save circuit
224 1
        evc.sync()
225
226
        # Schedule the circuit deploy
227 1
        self.sched.add(evc)
228
229
        # Circuit has no schedule, deploy now
230 1
        if not evc.circuit_scheduler:
231 1
            with evc.lock:
232 1
                evc.deploy()
233
234
        # Notify users
235 1
        result = {"circuit_id": evc.id}
236 1
        status = 201
237 1
        log.debug("create_circuit result %s %s", result, status)
238 1
        emit_event(self.controller, name="created", content={
239
            "id": evc.id,
240
            "name": evc.name,
241
            "metadata": evc.metadata,
242
            "active": evc._active,
243
            "enabled": evc._enabled,
244
            "uni_a": evc.uni_a,
245
            "uni_z": evc.uni_z
246
        })
247 1
        return jsonify(result), status
248
249 1
    @listen_to('kytos/flow_manager.flow.removed')
250 1
    def on_flow_delete(self, event):
251
        """Capture delete messages to keep track when flows got removed."""
252
        self.handle_flow_delete(event)
253
254 1
    def handle_flow_delete(self, event):
255
        """Keep track when the EVC got flows removed by deriving its cookie."""
256 1
        flow = event.content["flow"]
257 1
        evc = self.circuits.get(EVC.get_id_from_cookie(flow.cookie))
258 1
        if evc:
259 1
            log.debug("Flow removed in EVC %s", evc.id)
260 1
            evc.set_flow_removed_at()
261
262 1
    @rest("/v2/evc/<circuit_id>", methods=["PATCH"])
263 1
    def update(self, circuit_id):
264
        """Update a circuit based on payload.
265
266
        The EVC required attributes (name, uni_a, uni_z) can't be updated.
267
        """
268 1
        log.debug("update /v2/evc/%s", circuit_id)
269 1
        try:
270 1
            evc = self.circuits[circuit_id]
271 1
        except KeyError:
272 1
            result = f"circuit_id {circuit_id} not found"
273 1
            log.debug("update result %s %s", result, 404)
274 1
            raise NotFound(result) from NotFound
275
276 1
        if evc.archived:
277 1
            result = "Can't update archived EVC"
278 1
            log.debug("update result %s %s", result, 405)
279 1
            raise MethodNotAllowed(["GET"], result)
280
281 1
        try:
282 1
            data = request.get_json()
283 1
        except BadRequest:
284 1
            result = "The request body is not a well-formed JSON."
285 1
            log.debug("update result %s %s", result, 400)
286 1
            raise BadRequest(result) from BadRequest
287 1
        if data is None:
288 1
            result = "The request body mimetype is not application/json."
289 1
            log.debug("update result %s %s", result, 415)
290 1
            raise UnsupportedMediaType(result) from UnsupportedMediaType
291
292 1
        try:
293 1
            enable, redeploy = evc.update(
294
                **self._evc_dict_with_instances(data)
295
            )
296 1
        except ValueError as exception:
297 1
            log.error(exception)
298 1
            log.debug("update result %s %s", exception, 400)
299 1
            raise BadRequest(str(exception)) from BadRequest
300
301 1
        if evc.is_active():
302
            if enable is False:  # disable if active
303
                with evc.lock:
304
                    evc.remove()
305
            elif redeploy is not None:  # redeploy if active
306
                with evc.lock:
307
                    evc.remove()
308
                    evc.deploy()
309
        else:
310 1
            if enable is True:  # enable if inactive
311 1
                with evc.lock:
312 1
                    evc.deploy()
313 1
        result = {evc.id: evc.as_dict()}
314 1
        status = 200
315
316 1
        log.debug("update result %s %s", result, status)
317 1
        emit_event(self.controller, "updated", content={
318
            "evc_id": evc.id,
319
            "data": data
320
        })
321 1
        return jsonify(result), status
322
323 1
    @rest("/v2/evc/<circuit_id>", methods=["DELETE"])
324 1
    def delete_circuit(self, circuit_id):
325
        """Remove a circuit.
326
327
        First, the flows are removed from the switches, and then the EVC is
328
        disabled.
329
        """
330 1
        log.debug("delete_circuit /v2/evc/%s", circuit_id)
331 1
        try:
332 1
            evc = self.circuits[circuit_id]
333 1
        except KeyError:
334 1
            result = f"circuit_id {circuit_id} not found"
335 1
            log.debug("delete_circuit result %s %s", result, 404)
336 1
            raise NotFound(result) from NotFound
337
338 1
        if evc.archived:
339 1
            result = f"Circuit {circuit_id} already removed"
340 1
            log.debug("delete_circuit result %s %s", result, 404)
341 1
            raise NotFound(result) from NotFound
342
343 1
        log.info("Removing %s", evc)
344 1
        with evc.lock:
345 1
            evc.remove_current_flows()
346 1
            evc.remove_failover_flows(sync=False)
347 1
            evc.deactivate()
348 1
            evc.disable()
349 1
            self.sched.remove(evc)
350 1
            evc.archive()
351 1
            evc.sync()
352 1
        log.info("EVC removed. %s", evc)
353 1
        result = {"response": f"Circuit {circuit_id} removed"}
354 1
        status = 200
355
356 1
        log.debug("delete_circuit result %s %s", result, status)
357 1
        emit_event(self.controller, "deleted", content={
358
            "evc_id": evc.id
359
        })
360 1
        return jsonify(result), status
361
362 1
    @rest("v2/evc/<circuit_id>/metadata", methods=["GET"])
363 1
    def get_metadata(self, circuit_id):
364
        """Get metadata from an EVC."""
365 1
        try:
366 1
            return (
367
                jsonify({"metadata": self.circuits[circuit_id].metadata}),
368
                200,
369
            )
370
        except KeyError as error:
371
            raise NotFound(f"circuit_id {circuit_id} not found.") from error
372
373 1
    @rest("v2/evc/<circuit_id>/metadata", methods=["POST"])
374 1
    def add_metadata(self, circuit_id):
375
        """Add metadata to an EVC."""
376 1
        try:
377 1
            metadata = request.get_json()
378 1
            content_type = request.content_type
379 1
        except BadRequest as error:
380 1
            result = "The request body is not a well-formed JSON."
381 1
            raise BadRequest(result) from error
382 1
        if content_type is None:
383 1
            result = "The request body is empty."
384 1
            raise BadRequest(result)
385 1
        if metadata is None:
386 1
            if content_type != "application/json":
387 1
                result = (
388
                    "The content type must be application/json "
389
                    f"(received {content_type})."
390
                )
391
            else:
392
                result = "Metadata is empty."
393 1
            raise UnsupportedMediaType(result)
394
395 1
        try:
396 1
            evc = self.circuits[circuit_id]
397 1
        except KeyError as error:
398 1
            raise NotFound(f"circuit_id {circuit_id} not found.") from error
399
400 1
        evc.extend_metadata(metadata)
401 1
        evc.sync()
402 1
        return jsonify("Operation successful"), 201
403
404 1
    @rest("v2/evc/<circuit_id>/metadata/<key>", methods=["DELETE"])
405 1
    def delete_metadata(self, circuit_id, key):
406
        """Delete metadata from an EVC."""
407 1
        try:
408 1
            evc = self.circuits[circuit_id]
409 1
        except KeyError as error:
410 1
            raise NotFound(f"circuit_id {circuit_id} not found.") from error
411
412 1
        evc.remove_metadata(key)
413 1
        evc.sync()
414 1
        return jsonify("Operation successful"), 200
415
416 1
    @rest("/v2/evc/<circuit_id>/redeploy", methods=["PATCH"])
417 1
    def redeploy(self, circuit_id):
418
        """Endpoint to force the redeployment of an EVC."""
419 1
        log.debug("redeploy /v2/evc/%s/redeploy", circuit_id)
420 1
        try:
421 1
            evc = self.circuits[circuit_id]
422 1
        except KeyError:
423 1
            result = f"circuit_id {circuit_id} not found"
424 1
            raise NotFound(result) from NotFound
425 1
        if evc.is_enabled():
426 1
            with evc.lock:
427 1
                evc.remove_current_flows()
428 1
                evc.deploy()
429 1
            result = {"response": f"Circuit {circuit_id} redeploy received."}
430 1
            status = 202
431
        else:
432 1
            result = {"response": f"Circuit {circuit_id} is disabled."}
433 1
            status = 409
434
435 1
        return jsonify(result), status
436
437 1
    @rest("/v2/evc/schedule", methods=["GET"])
438 1
    def list_schedules(self):
439
        """Endpoint to return all schedules stored for all circuits.
440
441
        Return a JSON with the following template:
442
        [{"schedule_id": <schedule_id>,
443
         "circuit_id": <circuit_id>,
444
         "schedule": <schedule object>}]
445
        """
446 1
        log.debug("list_schedules /v2/evc/schedule")
447 1
        circuits = self.mongo_controller.get_circuits()['circuits'].values()
448 1
        if not circuits:
449 1
            result = {}
450 1
            status = 200
451 1
            return jsonify(result), status
452
453 1
        result = []
454 1
        status = 200
455 1
        for circuit in circuits:
456 1
            circuit_scheduler = circuit.get("circuit_scheduler")
457 1
            if circuit_scheduler:
458 1
                for scheduler in circuit_scheduler:
459 1
                    value = {
460
                        "schedule_id": scheduler.get("id"),
461
                        "circuit_id": circuit.get("id"),
462
                        "schedule": scheduler,
463
                    }
464 1
                    result.append(value)
465
466 1
        log.debug("list_schedules result %s %s", result, status)
467 1
        return jsonify(result), status
468
469 1
    @rest("/v2/evc/schedule/", methods=["POST"])
470 1
    def create_schedule(self):
471
        """
472
        Create a new schedule for a given circuit.
473
474
        This service do no check if there are conflicts with another schedule.
475
        Payload example:
476
            {
477
              "circuit_id":"aa:bb:cc",
478
              "schedule": {
479
                "date": "2019-08-07T14:52:10.967Z",
480
                "interval": "string",
481
                "frequency": "1 * * * *",
482
                "action": "create"
483
              }
484
            }
485
        """
486 1
        log.debug("create_schedule /v2/evc/schedule/")
487
488 1
        json_data = self._json_from_request("create_schedule")
489 1
        try:
490 1
            circuit_id = json_data["circuit_id"]
491 1
        except TypeError:
492 1
            result = "The payload should have a dictionary."
493 1
            log.debug("create_schedule result %s %s", result, 400)
494 1
            raise BadRequest(result) from BadRequest
495 1
        except KeyError:
496 1
            result = "Missing circuit_id."
497 1
            log.debug("create_schedule result %s %s", result, 400)
498 1
            raise BadRequest(result) from BadRequest
499
500 1
        try:
501 1
            schedule_data = json_data["schedule"]
502 1
        except KeyError:
503 1
            result = "Missing schedule data."
504 1
            log.debug("create_schedule result %s %s", result, 400)
505 1
            raise BadRequest(result) from BadRequest
506
507
        # Get EVC from circuits buffer
508 1
        circuits = self._get_circuits_buffer()
509
510
        # get the circuit
511 1
        evc = circuits.get(circuit_id)
512
513
        # get the circuit
514 1
        if not evc:
515 1
            result = f"circuit_id {circuit_id} not found"
516 1
            log.debug("create_schedule result %s %s", result, 404)
517 1
            raise NotFound(result) from NotFound
518
        # Can not modify circuits deleted and archived
519 1
        if evc.archived:
520 1
            result = f"Circuit {circuit_id} is archived. Update is forbidden."
521 1
            log.debug("create_schedule result %s %s", result, 403)
522 1
            raise Forbidden(result) from Forbidden
523
524
        # new schedule from dict
525 1
        new_schedule = CircuitSchedule.from_dict(schedule_data)
526
527
        # If there is no schedule, create the list
528 1
        if not evc.circuit_scheduler:
529 1
            evc.circuit_scheduler = []
530
531
        # Add the new schedule
532 1
        evc.circuit_scheduler.append(new_schedule)
533
534
        # Add schedule job
535 1
        self.sched.add_circuit_job(evc, new_schedule)
536
537
        # save circuit to mongodb
538 1
        evc.sync()
539
540 1
        result = new_schedule.as_dict()
541 1
        status = 201
542
543 1
        log.debug("create_schedule result %s %s", result, status)
544 1
        return jsonify(result), status
545
546 1
    @rest("/v2/evc/schedule/<schedule_id>", methods=["PATCH"])
547 1
    def update_schedule(self, schedule_id):
548
        """Update a schedule.
549
550
        Change all attributes from the given schedule from a EVC circuit.
551
        The schedule ID is preserved as default.
552
        Payload example:
553
            {
554
              "date": "2019-08-07T14:52:10.967Z",
555
              "interval": "string",
556
              "frequency": "1 * * *",
557
              "action": "create"
558
            }
559
        """
560 1
        log.debug("update_schedule /v2/evc/schedule/%s", schedule_id)
561
562
        # Try to find a circuit schedule
563 1
        evc, found_schedule = self._find_evc_by_schedule_id(schedule_id)
564
565
        # Can not modify circuits deleted and archived
566 1
        if not found_schedule:
567 1
            result = f"schedule_id {schedule_id} not found"
568 1
            log.debug("update_schedule result %s %s", result, 404)
569 1
            raise NotFound(result) from NotFound
570 1
        if evc.archived:
571 1
            result = f"Circuit {evc.id} is archived. Update is forbidden."
572 1
            log.debug("update_schedule result %s %s", result, 403)
573 1
            raise Forbidden(result) from Forbidden
574
575 1
        data = self._json_from_request("update_schedule")
576
577 1
        new_schedule = CircuitSchedule.from_dict(data)
578 1
        new_schedule.id = found_schedule.id
579
        # Remove the old schedule
580 1
        evc.circuit_scheduler.remove(found_schedule)
581
        # Append the modified schedule
582 1
        evc.circuit_scheduler.append(new_schedule)
583
584
        # Cancel all schedule jobs
585 1
        self.sched.cancel_job(found_schedule.id)
586
        # Add the new circuit schedule
587 1
        self.sched.add_circuit_job(evc, new_schedule)
588
        # Save EVC to mongodb
589 1
        evc.sync()
590
591 1
        result = new_schedule.as_dict()
592 1
        status = 200
593
594 1
        log.debug("update_schedule result %s %s", result, status)
595 1
        return jsonify(result), status
596
597 1
    @rest("/v2/evc/schedule/<schedule_id>", methods=["DELETE"])
598 1
    def delete_schedule(self, schedule_id):
599
        """Remove a circuit schedule.
600
601
        Remove the Schedule from EVC.
602
        Remove the Schedule from cron job.
603
        Save the EVC to the Storehouse.
604
        """
605 1
        log.debug("delete_schedule /v2/evc/schedule/%s", schedule_id)
606 1
        evc, found_schedule = self._find_evc_by_schedule_id(schedule_id)
607
608
        # Can not modify circuits deleted and archived
609 1
        if not found_schedule:
610 1
            result = f"schedule_id {schedule_id} not found"
611 1
            log.debug("delete_schedule result %s %s", result, 404)
612 1
            raise NotFound(result)
613
614 1
        if evc.archived:
615 1
            result = f"Circuit {evc.id} is archived. Update is forbidden."
616 1
            log.debug("delete_schedule result %s %s", result, 403)
617 1
            raise Forbidden(result)
618
619
        # Remove the old schedule
620 1
        evc.circuit_scheduler.remove(found_schedule)
621
622
        # Cancel all schedule jobs
623 1
        self.sched.cancel_job(found_schedule.id)
624
        # Save EVC to mongodb
625 1
        evc.sync()
626
627 1
        result = "Schedule removed"
628 1
        status = 200
629
630 1
        log.debug("delete_schedule result %s %s", result, status)
631 1
        return jsonify(result), status
632
633 1
    def _is_duplicated_evc(self, evc):
634
        """Verify if the circuit given is duplicated with the stored evcs.
635
636
        Args:
637
            evc (EVC): circuit to be analysed.
638
639
        Returns:
640
            boolean: True if the circuit is duplicated, otherwise False.
641
642
        """
643 1
        for circuit in tuple(self.circuits.values()):
644 1
            if not circuit.archived and circuit.shares_uni(evc):
645 1
                return True
646 1
        return False
647
648 1
    @listen_to("kytos/topology.link_up")
649 1
    def on_link_up(self, event):
650
        """Change circuit when link is up or end_maintenance."""
651
        self.handle_link_up(event)
652
653 1
    def handle_link_up(self, event):
654
        """Change circuit when link is up or end_maintenance."""
655 1
        log.info("Event handle_link_up %s", event.content["link"])
656 1
        for evc in self.get_evcs_by_svc_level():
657 1
            if evc.is_enabled() and not evc.archived:
658 1
                with evc.lock:
659 1
                    evc.handle_link_up(event.content["link"])
660
661 1
    @listen_to("kytos/topology.link_down")
662 1
    def on_link_down(self, event):
663
        """Change circuit when link is down or under_mantenance."""
664
        self.handle_link_down(event)
665
666 1
    def handle_link_down(self, event):
667
        """Change circuit when link is down or under_mantenance."""
668 1
        link = event.content["link"]
669 1
        log.info("Event handle_link_down %s", link)
670 1
        switch_flows = {}
671 1
        evcs_with_failover = []
672 1
        evcs_normal = []
673 1
        check_failover = []
674 1
        for evc in self.get_evcs_by_svc_level():
675 1
            if evc.is_affected_by_link(link):
676
                # if there is no failover path, handles link down the
677
                # tradditional way
678 1
                if (
679
                    not getattr(evc, 'failover_path', None) or
680
                    evc.is_failover_path_affected_by_link(link)
681
                ):
682 1
                    evcs_normal.append(evc)
683 1
                    continue
684 1
                for dpid, flows in evc.get_failover_flows().items():
685 1
                    switch_flows.setdefault(dpid, [])
686 1
                    switch_flows[dpid].extend(flows)
687 1
                evcs_with_failover.append(evc)
688
            else:
689 1
                check_failover.append(evc)
690
691 1
        offset = 0
692 1
        while switch_flows:
693 1
            offset = (offset + settings.BATCH_SIZE) or None
694 1
            switches = list(switch_flows.keys())
695 1
            for dpid in switches:
696 1
                emit_event(
697
                    self.controller,
698
                    context="kytos.flow_manager",
699
                    name="flows.install",
700
                    content={
701
                        "dpid": dpid,
702
                        "flow_dict": {"flows": switch_flows[dpid][:offset]},
703
                    }
704
                )
705 1
                if offset is None or offset >= len(switch_flows[dpid]):
706 1
                    del switch_flows[dpid]
707 1
                    continue
708 1
                switch_flows[dpid] = switch_flows[dpid][offset:]
709 1
            time.sleep(settings.BATCH_INTERVAL)
710
711 1
        for evc in evcs_with_failover:
712 1
            with evc.lock:
713 1
                old_path = evc.current_path
714 1
                evc.current_path = evc.failover_path
715 1
                evc.failover_path = old_path
716 1
                evc.sync()
717 1
            emit_event(self.controller, "redeployed_link_down", content={
718
                "evc_id": evc.id
719
            })
720 1
            log.info(
721
                f"{evc} redeployed with failover due to link down {link.id}"
722
            )
723
724 1
        for evc in evcs_normal:
725 1
            emit_event(
726
                self.controller,
727
                "evc_affected_by_link_down",
728
                content={
729
                    "evc_id": evc.id,
730
                    "link_id": link.id,
731
                }
732
            )
733
734
        # After handling the hot path, check if new failover paths are needed.
735
        # Note that EVCs affected by link down will generate a KytosEvent for
736
        # deployed|redeployed, which will trigger the failover path setup.
737
        # Thus, we just need to further check the check_failover list
738 1
        for evc in check_failover:
739 1
            if evc.is_failover_path_affected_by_link(link):
740 1
                evc.setup_failover_path()
741
742 1
    @listen_to("kytos/mef_eline.evc_affected_by_link_down")
743 1
    def on_evc_affected_by_link_down(self, event):
744
        """Change circuit when link is down or under_mantenance."""
745
        self.handle_evc_affected_by_link_down(event)
746
747 1
    def handle_evc_affected_by_link_down(self, event):
748
        """Change circuit when link is down or under_mantenance."""
749 1
        evc = self.circuits.get(event.content["evc_id"])
750 1
        link_id = event.content['link_id']
751 1
        if not evc:
752 1
            return
753 1
        with evc.lock:
754 1
            result = evc.handle_link_down()
755 1
        event_name = "error_redeploy_link_down"
756 1
        if result:
757 1
            log.info(f"{evc} redeployed due to link down {link_id}")
758 1
            event_name = "redeployed_link_down"
759 1
        emit_event(self.controller, event_name, content={
760
            "evc_id": evc.id
761
        })
762
763 1
    @listen_to("kytos/mef_eline.(redeployed_link_(up|down)|deployed)")
764 1
    def on_evc_deployed(self, event):
765
        """Handle EVC deployed|redeployed_link_down."""
766
        self.handle_evc_deployed(event)
767
768 1
    def handle_evc_deployed(self, event):
769
        """Setup failover path on evc deployed."""
770 1
        evc = self.circuits.get(event.content["evc_id"])
771 1
        if not evc:
772 1
            return
773 1
        with evc.lock:
774 1
            evc.setup_failover_path()
775
776 1
    @listen_to("kytos/topology.topology_loaded")
777 1
    def on_topology_loaded(self, event):  # pylint: disable=unused-argument
778
        """Load EVCs once the topology is available."""
779
        self.load_all_evcs()
780
781 1
    def load_all_evcs(self):
782
        """Try to load all EVCs on startup."""
783 1
        circuits = self.mongo_controller.get_circuits()['circuits'].items()
784 1
        for circuit_id, circuit in circuits:
785 1
            if circuit_id not in self.circuits:
786 1
                self._load_evc(circuit)
787
788 1
    def _load_evc(self, circuit_dict):
789
        """Load one EVC from mongodb to memory."""
790 1
        try:
791 1
            evc = self._evc_from_dict(circuit_dict)
792 1
        except ValueError as exception:
793 1
            log.error(
794
                f"Could not load EVC: dict={circuit_dict} error={exception}"
795
            )
796 1
            return None
797
798 1
        if evc.archived:
799 1
            return None
800 1
        evc.deactivate()
801 1
        evc.sync()
802 1
        self.circuits.setdefault(evc.id, evc)
803 1
        self.sched.add(evc)
804 1
        return evc
805
806 1
    @listen_to("kytos/flow_manager.flow.error")
807 1
    def on_flow_mod_error(self, event):
808
        """Handle flow mod errors related to an EVC."""
809
        self.handle_flow_mod_error(event)
810
811 1
    def handle_flow_mod_error(self, event):
812
        """Handle flow mod errors related to an EVC."""
813 1
        flow = event.content["flow"]
814 1
        command = event.content.get("error_command")
815 1
        if command != "add":
816
            return
817 1
        evc = self.circuits.get(EVC.get_id_from_cookie(flow.cookie))
818 1
        if evc:
819 1
            evc.remove_current_flows()
820
821 1
    def _evc_dict_with_instances(self, evc_dict):
822
        """Convert some dict values to instance of EVC classes.
823
824
        This method will convert: [UNI, Link]
825
        """
826 1
        data = evc_dict.copy()  # Do not modify the original dict
827 1
        for attribute, value in data.items():
828
            # Get multiple attributes.
829
            # Ex: uni_a, uni_z
830 1
            if "uni" in attribute:
831 1
                try:
832 1
                    data[attribute] = self._uni_from_dict(value)
833 1
                except ValueError as exception:
834 1
                    result = "Error creating UNI: Invalid value"
835 1
                    raise ValueError(result) from exception
836
837 1
            if attribute == "circuit_scheduler":
838 1
                data[attribute] = []
839 1
                for schedule in value:
840 1
                    data[attribute].append(CircuitSchedule.from_dict(schedule))
841
842
            # Get multiple attributes.
843
            # Ex: primary_links,
844
            #     backup_links,
845
            #     current_links_cache,
846
            #     primary_links_cache,
847
            #     backup_links_cache
848 1
            if "links" in attribute:
849 1
                data[attribute] = [
850
                    self._link_from_dict(link) for link in value
851
                ]
852
853
            # Ex: current_path,
854
            #     primary_path,
855
            #     backup_path
856 1
            if "path" in attribute and attribute != "dynamic_backup_path":
857 1
                data[attribute] = Path(
858
                    [self._link_from_dict(link) for link in value]
859
                )
860
861 1
        return data
862
863 1
    def _evc_from_dict(self, evc_dict):
864 1
        data = self._evc_dict_with_instances(evc_dict)
865 1
        return EVC(self.controller, **data)
866
867 1
    def _uni_from_dict(self, uni_dict):
868
        """Return a UNI object from python dict."""
869 1
        if uni_dict is None:
870 1
            return False
871
872 1
        interface_id = uni_dict.get("interface_id")
873 1
        interface = self.controller.get_interface_by_id(interface_id)
874 1
        if interface is None:
875 1
            result = (
876
                "Error creating UNI:"
877
                + f"Could not instantiate interface {interface_id}"
878
            )
879 1
            raise ValueError(result) from ValueError
880
881 1
        tag_dict = uni_dict.get("tag", None)
882 1
        if tag_dict:
883 1
            tag = TAG.from_dict(tag_dict)
884
        else:
885 1
            tag = None
886 1
        uni = UNI(interface, tag)
887
888 1
        return uni
889
890 1
    def _link_from_dict(self, link_dict):
891
        """Return a Link object from python dict."""
892 1
        id_a = link_dict.get("endpoint_a").get("id")
893 1
        id_b = link_dict.get("endpoint_b").get("id")
894
895 1
        endpoint_a = self.controller.get_interface_by_id(id_a)
896 1
        endpoint_b = self.controller.get_interface_by_id(id_b)
897 1
        if not endpoint_a:
898 1
            error_msg = f"Could not get interface endpoint_a id {id_a}"
899 1
            raise ValueError(error_msg)
900 1
        if not endpoint_b:
901
            error_msg = f"Could not get interface endpoint_b id {id_b}"
902
            raise ValueError(error_msg)
903
904 1
        link = Link(endpoint_a, endpoint_b)
905 1
        if "metadata" in link_dict:
906 1
            link.extend_metadata(link_dict.get("metadata"))
907
908 1
        s_vlan = link.get_metadata("s_vlan")
909 1
        if s_vlan:
910 1
            tag = TAG.from_dict(s_vlan)
911 1
            if tag is False:
912
                error_msg = f"Could not instantiate tag from dict {s_vlan}"
913
                raise ValueError(error_msg)
914 1
            link.update_metadata("s_vlan", tag)
915 1
        return link
916
917 1
    def _find_evc_by_schedule_id(self, schedule_id):
918
        """
919
        Find an EVC and CircuitSchedule based on schedule_id.
920
921
        :param schedule_id: Schedule ID
922
        :return: EVC and Schedule
923
        """
924 1
        circuits = self._get_circuits_buffer()
925 1
        found_schedule = None
926 1
        evc = None
927
928
        # pylint: disable=unused-variable
929 1
        for c_id, circuit in circuits.items():
930 1
            for schedule in circuit.circuit_scheduler:
931 1
                if schedule.id == schedule_id:
932 1
                    found_schedule = schedule
933 1
                    evc = circuit
934 1
                    break
935 1
            if found_schedule:
936 1
                break
937 1
        return evc, found_schedule
938
939 1
    def _get_circuits_buffer(self):
940
        """
941
        Return the circuit buffer.
942
943
        If the buffer is empty, try to load data from mongodb.
944
        """
945 1
        if not self.circuits:
946
            # Load circuits from mongodb to buffer
947 1
            circuits = self.mongo_controller.get_circuits()['circuits']
948 1
            for c_id, circuit in circuits.items():
949 1
                evc = self._evc_from_dict(circuit)
950 1
                self.circuits[c_id] = evc
951 1
        return self.circuits
952
953 1
    @staticmethod
954 1
    def _json_from_request(caller):
955
        """Return a json from request.
956
957
        If it was not possible to get a json from the request, log, for debug,
958
        who was the caller and the error that ocurred, and raise an
959
        Exception.
960
        """
961 1
        try:
962 1
            json_data = request.get_json()
963 1
        except ValueError as exception:
964
            log.error(exception)
965
            log.debug(f"{caller} result {exception} 400")
966
            raise BadRequest(str(exception)) from BadRequest
967 1
        except BadRequest:
968 1
            result = "The request is not a valid JSON."
969 1
            log.debug(f"{caller} result {result} 400")
970 1
            raise BadRequest(result) from BadRequest
971 1
        if json_data is None:
972 1
            result = "Content-Type must be application/json"
973 1
            log.debug(f"{caller} result {result} 415")
974 1
            raise UnsupportedMediaType(result)
975
        return json_data
976