Passed
Pull Request — master (#142)
by Vinicius
02:53
created

build.controllers   A

Complexity

Total Complexity 36

Size/Duplication

Total Lines 316
Duplicated Lines 0 %

Test Coverage

Coverage 97.3%

Importance

Changes 0
Metric Value
eloc 203
dl 0
loc 316
ccs 108
cts 111
cp 0.973
rs 9.52
c 0
b 0
f 0
wmc 36

27 Methods

Rating   Name   Duplication   Size   Complexity  
A TopoController.add_interface_metadata() 0 8 1
A TopoController._update_interface() 0 14 2
A TopoController.enable_link() 0 3 1
A TopoController._update_link() 0 4 1
A TopoController.disable_link() 0 3 1
A TopoController.delete_interface_metadata_key() 0 6 1
A TopoController.upsert_link() 0 46 1
A TopoController.disable_interface() 0 4 1
A TopoController.bootstrap_indexes() 0 10 3
A TopoController.enable_switch() 0 3 1
A TopoController.get_topology() 0 5 1
A TopoController.get_links() 0 9 1
A TopoController.add_switch_metadata() 0 6 1
A TopoController._update_switch() 0 4 1
A TopoController.add_link_metadata() 0 8 1
A TopoController.disable_switch() 0 4 1
A TopoController.delete_link_metadata_key() 0 5 1
A TopoController.delete_switch_metadata_key() 0 5 1
A TopoController.get_switches() 0 9 1
A TopoController._set_updated_at() 0 7 2
A TopoController.enable_interface() 0 4 1
A TopoController.__init__() 0 6 2
B TopoController.bulk_upsert_interface_details() 0 31 5
A TopoController.upsert_switch() 0 16 1
A TopoController.get_interfaces() 0 11 1
A TopoController.bulk_delete_link_metadata_key() 0 8 1
A TopoController.get_interfaces_details() 0 7 1
1
"""TopoController."""
2
3
# pylint: disable=invalid-name
4 1
import os
5 1
from datetime import datetime
6 1
from threading import Lock
7 1
from typing import List, Optional, Tuple
8
9 1
import pymongo
10 1
from pymongo.collection import ReturnDocument
11 1
from pymongo.errors import AutoReconnect
12 1
from pymongo.operations import UpdateOne
13 1
from tenacity import retry_if_exception_type, stop_after_attempt, wait_random
14
15 1
from kytos.core import log
16 1
from kytos.core.db import Mongo
17 1
from kytos.core.retry import before_sleep, for_all_methods, retries
18 1
from napps.kytos.topology.db.models import (InterfaceDetailDoc, LinkDoc,
19
                                            SwitchDoc)
20
21
22 1
@for_all_methods(
23
    retries,
24
    stop=stop_after_attempt(
25
        int(os.environ.get("MONGO_AUTO_RETRY_STOP_AFTER_ATTEMPT", 3))
26
    ),
27
    wait=wait_random(
28
        min=int(os.environ.get("MONGO_AUTO_RETRY_WAIT_RANDOM_MIN", 0.1)),
29
        max=int(os.environ.get("MONGO_AUTO_RETRY_WAIT_RANDOM_MAX", 1)),
30
    ),
31
    before_sleep=before_sleep,
32
    retry=retry_if_exception_type((AutoReconnect,)),
33
)
34 1
class TopoController:
35
    """TopoController."""
36
37 1
    def __init__(self, get_mongo=lambda: Mongo()) -> None:
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable Mongo does not seem to be defined.
Loading history...
38
        """Constructor of TopoController."""
39 1
        self.mongo = get_mongo()
40 1
        self.db_client = self.mongo.client
41 1
        self.db = self.db_client[self.mongo.db_name]
42 1
        self.interface_details_lock = Lock()
43
44 1
    def bootstrap_indexes(self) -> None:
45
        """Bootstrap all topology related indexes."""
46 1
        index_tuples = [
47
            ("switches", [("interfaces.id", pymongo.ASCENDING)]),
48
            ("links", [("endpoints.id", pymongo.ASCENDING)]),
49
        ]
50 1
        for collection, keys in index_tuples:
51 1
            if self.mongo.bootstrap_index(collection, keys):
52 1
                log.info(
53
                    f"Created DB index {keys}, collection: {collection})"
54
                )
55
56 1
    def get_topology(self) -> dict:
57
        """Get topology from DB."""
58 1
        switches = self.get_switches()
59 1
        links = self.get_links()
60 1
        return {"topology": {**links, **switches}}
61
62 1
    def get_switches(self) -> dict:
63
        """Get switches from DB."""
64 1
        switches = self.db.switches.aggregate(
65
            [
66
                {"$sort": {"_id": 1}},
67
                {"$project": SwitchDoc.projection()},
68
            ]
69
        )
70 1
        return {"switches": {value["id"]: value for value in switches}}
71
72 1
    def get_links(self) -> dict:
73
        """Get links from DB."""
74 1
        links = self.db.links.aggregate(
75
            [
76
                {"$sort": {"_id": 1}},
77
                {"$project": LinkDoc.projection()},
78
            ]
79
        )
80 1
        return {"links": {value["id"]: value for value in links}}
81
82 1
    def get_interfaces(self) -> dict:
83
        """Get interfaces from DB."""
84 1
        interfaces = self.db.switches.aggregate(
85
            [
86
                {"$sort": {"_id": 1}},
87
                {"$project": {"interfaces": 1, "_id": 0}},
88
                {"$unwind": "$interfaces"},
89
                {"$replaceRoot": {"newRoot": "$interfaces"}},
90
            ]
91
        )
92 1
        return {"interfaces": {value["id"]: value for value in interfaces}}
93
94 1
    @staticmethod
95 1
    def _set_updated_at(update_expr: dict) -> None:
96
        """Set updated_at on $set expression."""
97 1
        if "$set" in update_expr:
98 1
            update_expr["$set"].update({"updated_at": datetime.utcnow()})
99
        else:
100 1
            update_expr.update({"$set": {"updated_at": datetime.utcnow()}})
101
102 1
    def _update_switch(self, dpid: str, update_expr: dict) -> Optional[dict]:
103
        """Try to find one switch and update it given an update expression."""
104 1
        self._set_updated_at(update_expr)
105 1
        return self.db.switches.find_one_and_update({"_id": dpid}, update_expr)
106
107 1
    def upsert_switch(self, dpid: str, switch_dict: dict) -> Optional[dict]:
108
        """Update or insert switch."""
109 1
        utc_now = datetime.utcnow()
110 1
        model = SwitchDoc(
111
            **{**switch_dict, **{"_id": dpid, "updated_at": utc_now}}
112
        )
113 1
        updated = self.db.switches.find_one_and_update(
114
            {"_id": dpid},
115
            {
116
                "$set": model.dict(exclude={"inserted_at"}),
117
                "$setOnInsert": {"inserted_at": utc_now},
118
            },
119
            return_document=ReturnDocument.AFTER,
120
            upsert=True,
121
        )
122 1
        return updated
123
124 1
    def enable_switch(self, dpid: str) -> Optional[dict]:
125
        """Try to find one switch and enable it."""
126 1
        return self._update_switch(dpid, {"$set": {"enabled": True}})
127
128 1
    def disable_switch(self, dpid: str) -> Optional[dict]:
129
        """Try to find one switch and disable it."""
130 1
        return self._update_switch(
131
            dpid, {"$set": {"enabled": False, "interfaces.$[].enabled": False}}
132
        )
133
134 1
    def add_switch_metadata(self, dpid: str, metadata: dict) -> Optional[dict]:
135
        """Try to find a switch and add to its metadata."""
136 1
        update_expr = {
137
            "$set": {f"metadata.{k}": v for k, v in metadata.items()}
138
        }
139 1
        return self._update_switch(dpid, update_expr)
140
141 1
    def delete_switch_metadata_key(
142
        self, dpid: str, key: str
143
    ) -> Optional[dict]:
144
        """Try to find a switch and delete a metadata key."""
145 1
        return self._update_switch(dpid, {"$unset": {f"metadata.{key}": ""}})
146
147 1
    def enable_interface(self, interface_id: str) -> Optional[dict]:
148
        """Try to enable one interface and its embedded object on links."""
149 1
        return self._update_interface(
150
            interface_id, {"$set": {"enabled": True}}
151
        )
152
153 1
    def disable_interface(self, interface_id: str) -> Optional[dict]:
154
        """Try to disable one interface and its embedded object on links."""
155 1
        return self._update_interface(
156
            interface_id, {"$set": {"enabled": False}}
157
        )
158
159 1
    def add_interface_metadata(
160
        self, interface_id: str, metadata: dict
161
    ) -> Optional[dict]:
162
        """Try to find an interface and add to its metadata."""
163 1
        update_expr = {
164
            "$set": {f"metadata.{k}": v for k, v in metadata.items()}
165
        }
166 1
        return self._update_interface(interface_id, update_expr)
167
168 1
    def delete_interface_metadata_key(
169
        self, interface_id: str, key: str
170
    ) -> Optional[dict]:
171
        """Try to find an interface and delete a metadata key."""
172 1
        return self._update_interface(
173
            interface_id, {"$unset": {f"metadata.{key}": ""}}
174
        )
175
176 1
    def _update_interface(
177
        self, interface_id: str, update_expr: dict
178
    ) -> Optional[dict]:
179
        """Try to update one interface and its embedded object on links."""
180 1
        self._set_updated_at(update_expr)
181 1
        interfaces_expression = {}
182 1
        for operator, values in update_expr.items():
183 1
            interfaces_expression[operator] = {
184
                f"interfaces.$.{k}": v for k, v in values.items()
185
            }
186 1
        return self.db.switches.find_one_and_update(
187
            {"interfaces.id": interface_id},
188
            interfaces_expression,
189
            return_document=ReturnDocument.AFTER,
190
        )
191
192 1
    def upsert_link(self, link_id: str, link_dict: dict) -> dict:
193
        """Update or insert a Link."""
194 1
        utc_now = datetime.utcnow()
195
196 1
        endpoint_a = link_dict.get("endpoint_a")
197 1
        endpoint_b = link_dict.get("endpoint_b")
198 1
        model = LinkDoc(
199
            **{
200
                **link_dict,
201
                **{
202
                    "updated_at": utc_now,
203
                    "_id": link_id,
204
                    "endpoints": [endpoint_a, endpoint_b],
205
                },
206
            }
207
        )
208 1
        updated = self.db.links.find_one_and_update(
209
            {"_id": link_id},
210
            {
211
                "$set": model.dict(exclude={"inserted_at"}),
212
                "$setOnInsert": {"inserted_at": utc_now},
213
            },
214
            return_document=ReturnDocument.AFTER,
215
            upsert=True,
216
        )
217 1
        self.db.switches.find_one_and_update(
218
            {"interfaces.id": endpoint_a},
219
            {
220
                "$set": {
221
                    "interfaces.$.link_id": link_id,
222
                    "interfaces.$.link_side": "endpoint_a",
223
                    "updated_at": utc_now,
224
                }
225
            },
226
        )
227 1
        self.db.switches.find_one_and_update(
228
            {"interfaces.id": endpoint_b},
229
            {
230
                "$set": {
231
                    "interfaces.$.link_id": link_id,
232
                    "interfaces.$.link_side": "endpoint_b",
233
                    "updated_at": utc_now,
234
                }
235
            },
236
        )
237 1
        return updated
238
239 1
    def _update_link(self, link_id: str, update_expr: dict) -> Optional[dict]:
240
        """Try to find one link and update it given an update expression."""
241 1
        self._set_updated_at(update_expr)
242 1
        return self.db.links.find_one_and_update({"_id": link_id}, update_expr)
243
244 1
    def enable_link(self, link_id: str) -> Optional[dict]:
245
        """Try to find one link and enable it."""
246 1
        return self._update_link(link_id, {"$set": {"enabled": True}})
247
248 1
    def disable_link(self, link_id: str) -> Optional[dict]:
249
        """Try to find one link and disable it."""
250 1
        return self._update_link(link_id, {"$set": {"enabled": False}})
251
252 1
    def add_link_metadata(
253
        self, link_id: str, metadata: dict
254
    ) -> Optional[dict]:
255
        """Try to find link and add to its metadata."""
256 1
        update_expr = {
257
            "$set": {f"metadata.{k}": v for k, v in metadata.items()}
258
        }
259 1
        return self._update_link(link_id, update_expr)
260
261 1
    def delete_link_metadata_key(
262
        self, link_id: str, key: str
263
    ) -> Optional[dict]:
264
        """Try to find a link and delete a metadata key."""
265 1
        return self._update_link(link_id, {"$unset": {f"metadata.{key}": ""}})
266
267 1
    def bulk_delete_link_metadata_key(
268
        self, link_ids: List[str], key: str
269
    ) -> Optional[dict]:
270
        """Bulk delelete link metadata key."""
271
        update_expr = {"$unset": {f"metadata.{key}": 1}}
272
        self._set_updated_at(update_expr)
273
        return self.db.links.update_many({"_id": {"$in": link_ids}},
274
                                         update_expr)
275
276 1
    def bulk_upsert_interface_details(
277
        self, ids_details: List[Tuple[str, dict]]
278
    ) -> Optional[dict]:
279
        """Update or insert interfaces details."""
280 1
        utc_now = datetime.utcnow()
281 1
        ops = []
282 1
        for _id, detail_dict in ids_details:
283 1
            ops.append(
284
                UpdateOne(
285
                    {"_id": _id},
286
                    {
287
                        "$set": InterfaceDetailDoc(
288
                            **{
289
                                **detail_dict,
290
                                **{
291
                                    "updated_at": utc_now,
292
                                    "_id": _id,
293
                                },
294
                            }
295
                        ).dict(exclude={"inserted_at"}),
296
                        "$setOnInsert": {"inserted_at": utc_now},
297
                    },
298
                    upsert=True,
299
                ),
300
            )
301
302 1
        with self.interface_details_lock:
303 1
            with self.db_client.start_session() as session:
304 1
                with session.start_transaction():
305 1
                    return self.db.interface_details.bulk_write(
306
                        ops, ordered=False, session=session
307
                    )
308
309 1
    def get_interfaces_details(
310
        self, interface_ids: List[str]
311
    ) -> Optional[dict]:
312
        """Try to get interfaces details given a list of interface ids."""
313 1
        return self.db.interface_details.aggregate(
314
            [
315
                {"$match": {"_id": {"$in": interface_ids}}},
316
            ]
317
        )
318