9140549691

Committed 18 May 2024 02:26PM UTC coverage: 84.164% (+3.4%) from 80.789%

Build # 9140549691

Build Type

Pull #184

github

Committed by

web-flow

Commit Message

Merge e662ba12c into cf507ef45

Pull Request Pull Request #184: 183 about user specified vlan

Run Details

381 of 477 branches covered (79.87%)

Branch coverage included in aggregate %.

91 of 111 new or added lines in 5 files covered. (81.98%)

18 existing lines in 1 file now uncovered.

1038 of 1209 relevant lines covered (85.86%)

3.43 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.05

/src/sdx_pce/topology/manager.py

import copy
import datetime
import logging
from typing import Mapping

import networkx as nx
from sdx_datamodel.models.topology import (
    TOPOLOGY_INITIAL_VERSION,
    SDX_TOPOLOGY_ID_prefix,
)
from sdx_datamodel.parsing.topologyhandler import TopologyHandler

from .grenmlconverter import GrenmlConverter


class TopologyManager:
    """
    Manager for topology operations.

    Operations are:

        - Merge multiple topologies.

        - Convert to grenml (XML).
    """

    def __init__(self):
        # The merged "super" topology of topologies of different
        # domains, with inter-domain links between them computed.
        self._topology = None

        # Mapping from topology ID to topology.
        self._topology_map = {}

        # Mapping from port ID to link.
        self._port_map = {}

        # Mapping from port ID to node.
        self._port_node_map = {}

        # Number of interdomain links we computed.
        self._num_interdomain_link = 0

        self._logger = logging.getLogger(__name__)

    def get_handler(self):
        return self.topology_handler

    def topology_id(self, id):
        self._topology._id(id)

    def set_topology(self, topology):
        self._topology = topology

    def get_topology(self):
        return self._topology

    def get_topology_map(self) -> dict:
        return self._topology_map

    def get_port_map(self) -> Mapping[str, dict]:
        """
        Return a mapping between port IDs and links.
        """
        return self._port_map

    def get_port_node_map(self) -> Mapping[str, dict]:
        """
        Return a mapping between port IDs and nodes.
        """
        return self._port_node_map

    def clear_topology(self):
        self._topology = None
        self._topology_map = {}
        self._port_map = {}

    def add_topology(self, data):
        topology = TopologyHandler().import_topology_data(data)
        self._topology_map[topology.id] = topology

        if self._topology is None:
            self._topology = copy.deepcopy(topology)

            # Generate a new topology id
            self.generate_id()

            # Addding to the port list
            # links = topology.links
            # for link in links:
            #    for port in link.ports:
            #        self._port_map[port["id"]] = link
        else:
            # check the inter-domain links first.
            self._num_interdomain_link += self.inter_domain_check(topology)
            if self._num_interdomain_link == 0:
                self._logger.debug(
                    f"Warning: no interdomain links detected in {topology.id}!"
                )

            # Nodes
            nodes = topology.nodes
            self._topology.add_nodes(nodes)

            # links
            links = topology.links
            self._topology.add_links(links)

            # version
            self.update_version(False)

        # Addding to the port list
        links = topology.links
        for link in links:
            for port in link.ports:
                self._port_map[port["id"]] = link

        # Addding to the port node
        nodes = topology.nodes
        for node in nodes:
            for port in node.ports:
                self._port_node_map[port.id] = node

        self.update_timestamp()

    def get_domain_name(self, node_id):
        """
        Find the topology ID associated with the given node ID.

        A topology ID is expected to be of the format
        "urn:sdx:topology:amlight.net", and from this, we
        can find the domain name associated with the topology.

        TODO: This function name may be a misnomer?
        """
        domain_id = None
        # print(f"len of topology_list: {len(self._topology_map)}")
        for topology_id, topology in self._topology_map.items():
            if topology.has_node_by_id(node_id):
                domain_id = topology_id
                break

        return domain_id

    def generate_id(self):
        self._topology.id = SDX_TOPOLOGY_ID_prefix
        self._topology.version = TOPOLOGY_INITIAL_VERSION
        return id

    def remove_topology(self, topology_id):
        self._topology_map.pop(topology_id, None)
        self.update_version(False)
        self.update_timestamp()

    def is_link_interdomain(self, link, topology):
        """
        Check if a link is an interdomain link.
        """
        for port in link.ports:
            if port["id"] not in self._port_map:
                return True
        return False

    def update_topology(self, data):
        # likely adding new inter-domain links
        update_handler = TopologyHandler()
        topology = update_handler.import_topology_data(data)
        self._topology_map[topology.id] = topology

        # Nodes.
        nodes = topology.nodes
        for node in nodes:
            self._topology.remove_node(node.id)

        # Links.
        links = topology.links
        for link in links:
            if not self.is_link_interdomain(link, topology):
                # print(link.id+";......."+str(link.nni))
                self._topology.remove_link(link.id)
                for port in link.ports:
                    self._port_map.pop(port["id"])

        # Check the inter-domain links first.
        num_interdomain_link = self.inter_domain_check(topology)
        if num_interdomain_link == 0:
            self._logger.warning("Warning: no interdomain links detected!")

        # Nodes.
        nodes = topology.nodes
        self._topology.add_nodes(nodes)

        # Links.
        links = topology.links
        self._topology.add_links(links)

        self.update_version(True)
        self.update_timestamp()

    def update_version(self, sub: bool):
        try:
            [ver, sub_ver] = self._topology.version.split(".")
        except ValueError:
            ver = self._topology.version
            sub_ver = "0"

        self._topology.version = self.new_version(ver, sub_ver, sub)

        return self._topology.version

    def new_version(self, ver, sub_ver, sub: bool):
        if not sub:
            ver = str(int(ver) + 1)
            sub_ver = "0"
        else:
            sub_ver = str(int(sub_ver) + 1)

        return ver + "." + sub_ver

    def update_timestamp(self):
        ct = datetime.datetime.now().isoformat()
        self._topology.timestamp = ct

        return ct

    def inter_domain_check(self, topology):
        interdomain_port_dict = {}
        num_interdomain_link = 0
        links = topology.links
        link_dict = {}
        for link in links:
            link_dict[link.id] = link
            for port in link.ports:
                interdomain_port_dict[port["id"]] = link

        # ToDo: raise an warning or exception
        if len(interdomain_port_dict) == 0:
            self._logger.info("interdomain_port_dict==0")
            return False

        # match any ports in the existing topology
        for port_id in interdomain_port_dict:
            # print("interdomain_port:")
            # print(port_id)
            for existing_port, existing_link in self._port_map.items():
                # print(existing_port)
                if port_id == existing_port:
                    # print("Interdomain port:" + port_id)
                    # remove redundant link between two domains
                    self._topology.remove_link(existing_link.id)
                    num_interdomain_link = +1
            self._port_map[port_id] = interdomain_port_dict[port_id]

        return num_interdomain_link

    # adjacent matrix of the graph, in jason?
    def generate_graph(self):
        graph = nx.Graph()

        if self._topology is None:
            self._logger.warning("We do not have a topology yet")
            return None

        links = self._topology.links
        for link in links:
            inter_domain_link = False
            ports = link.ports
            end_nodes = []
            for port in ports:
                node = self._topology.get_node_by_port(port["id"])
                if node is None:
                    self._logger.warning(
                        f"This port (id: {port['id']}) does not belong to "
                        f"any node in the topology, likely a Non-SDX port!"
                    )
                    inter_domain_link = True
                    break
                else:
                    end_nodes.append(node)
                    # print("graph node:"+node.id)
            if not inter_domain_link:
                graph.add_edge(end_nodes[0].id, end_nodes[1].id)
                edge = graph.edges[end_nodes[0].id, end_nodes[1].id]
                edge["id"] = link.id
                edge["latency"] = link.latency
                edge["bandwidth"] = link.bandwidth
                edge["residual_bandwidth"] = link.residual_bandwidth
                edge["weight"] = 1000.0 * (1.0 / link.residual_bandwidth)
                edge["packet_loss"] = link.packet_loss
                edge["availability"] = link.availability

        return graph

    def generate_grenml(self):
        self.converter = GrenmlConverter(self._topology)

        return self.converter.read_topology()

    def add_domain_service(self):
        pass

    # may need to read from a configuration file.
    def update_private_properties(self):
        pass

    # on performance properties for now
    def update_link_property(self, link_id, property, value):
        # 1. update the individual topology
        for id, topology in self._topology_map.items():
            links = topology.links
            for link in links:
                self._logger.info(f"link.id={link.id}; id={id}")
                if link.id == link_id:
                    setattr(link, property, value)
                    self._logger.info("updated the link.")
                    # 1.2 need to change the sub_ver of the topology?

        # 2. check on the inter-domain link?
        # 3. update the interodamin topology
        links = self._topology.links
        for link in links:
            if link.id == link_id:
                setattr(link, property, value)
                self._logger.info("updated the link.")
                # 2.2 need to change the sub_ver of the topology?

        self.update_version(True)
        self.update_timestamp()
        # 4. Signal update the (networkx) graph

        # 5. signal Reoptimization of TE?

    def update_element_property_json(self, data, element, element_id, property, value):
        elements = data[element]
        for element in elements:
            if element["id"] == element_id:
                element[property] = value

        try:
            [ver, sub_ver] = data["version"].split(".")
        except ValueError:
            ver = "0"
            sub_ver = "0"

        data["version"] = self.new_version(ver, sub_ver, True)
        data["timestamp"] = datetime.datetime.now().isoformat()

    def get_port_by_id(self, port_id: str):
        """
        Given port id, returns a Port.
        """
        for node in self.get_topology().nodes:
            for port in node.ports:
                if port.id == port_id:
                    return port.to_dict()
        return None

    def are_two_ports_same_domain(self, port1_id: str, port2_id: str):
        """
        Check if two ports are in the same domain.
        """
        node1 = self.get_topology().get_node_by_port(port1_id)
        node2 = self.get_topology().get_node_by_port(port2_id)
        if node1 is None or node2 is None:
            return False

        domain1 = self.get_domain_name(node1.id)
        domain2 = self.get_domain_name(node2.id)
        return domain1 == domain2

    def update_node_property(self):
        pass

    def update_port_property(self):
        pass

1	import copy	4✔
2	import datetime	4✔
3	import logging	4✔
4	from typing import Mapping	4✔
5
6	import networkx as nx	4✔
7	from sdx_datamodel.models.topology import (	4✔
8	TOPOLOGY_INITIAL_VERSION,
9	SDX_TOPOLOGY_ID_prefix,
10	)
11	from sdx_datamodel.parsing.topologyhandler import TopologyHandler	4✔
12
13	from .grenmlconverter import GrenmlConverter	4✔
14
15
16	class TopologyManager:	4✔
17	"""	4✔
18	Manager for topology operations.
19
20	Operations are:
21
22	- Merge multiple topologies.
23
24	- Convert to grenml (XML).
25	"""
26
27	def __init__(self):	4✔
28	# The merged "super" topology of topologies of different
29	# domains, with inter-domain links between them computed.
30	self._topology = None	4✔
31
32	# Mapping from topology ID to topology.
33	self._topology_map = {}	4✔
34
35	# Mapping from port ID to link.
36	self._port_map = {}	4✔
37
38	# Mapping from port ID to node.
39	self._port_node_map = {}	4✔
40
41	# Number of interdomain links we computed.
42	self._num_interdomain_link = 0	4✔
43
44	self._logger = logging.getLogger(__name__)	4✔
45
46	def get_handler(self):	4✔
47	return self.topology_handler	×
48
49	def topology_id(self, id):	4✔
50	self._topology._id(id)	×
51
52	def set_topology(self, topology):	4✔
53	self._topology = topology	4✔
54
55	def get_topology(self):	4✔
56	return self._topology	4✔
57
58	def get_topology_map(self) -> dict:	4✔
59	return self._topology_map	4✔
60
61	def get_port_map(self) -> Mapping[str, dict]:	4✔
62	"""
63	Return a mapping between port IDs and links.
64	"""
65	return self._port_map	4✔
66
67	def get_port_node_map(self) -> Mapping[str, dict]:	4✔
68	"""
69	Return a mapping between port IDs and nodes.
70	"""
71	return self._port_node_map	4✔
72
73	def clear_topology(self):	4✔
74	self._topology = None	×
75	self._topology_map = {}	×
76	self._port_map = {}	×
77
78	def add_topology(self, data):	4✔
79	topology = TopologyHandler().import_topology_data(data)	4✔
80	self._topology_map[topology.id] = topology	4✔
81
82	if self._topology is None:	4✔
83	self._topology = copy.deepcopy(topology)	4✔
84
85	# Generate a new topology id
86	self.generate_id()	4✔
87
88	# Addding to the port list
89	# links = topology.links
90	# for link in links:
91	# for port in link.ports:
92	# self._port_map[port["id"]] = link
93	else:
94	# check the inter-domain links first.
95	self._num_interdomain_link += self.inter_domain_check(topology)	4✔
96	if self._num_interdomain_link == 0:	4✔
97	self._logger.debug(	4✔
98	f"Warning: no interdomain links detected in {topology.id}!"
99	)
100
101	# Nodes
102	nodes = topology.nodes	4✔
103	self._topology.add_nodes(nodes)	4✔
104
105	# links
106	links = topology.links	4✔
107	self._topology.add_links(links)	4✔
108
109	# version
110	self.update_version(False)	4✔
111
112	# Addding to the port list
113	links = topology.links	4✔
114	for link in links:	4✔
115	for port in link.ports:	4✔
116	self._port_map[port["id"]] = link	4✔
117
118	# Addding to the port node
119	nodes = topology.nodes	4✔
120	for node in nodes:	4✔
121	for port in node.ports:	4✔
122	self._port_node_map[port.id] = node	4✔
123
124	self.update_timestamp()	4✔
125
126	def get_domain_name(self, node_id):	4✔
127	"""
128	Find the topology ID associated with the given node ID.
129
130	A topology ID is expected to be of the format
131	"urn:sdx:topology:amlight.net", and from this, we
132	can find the domain name associated with the topology.
133
134	TODO: This function name may be a misnomer?
135	"""
136	domain_id = None	4✔
137	# print(f"len of topology_list: {len(self._topology_map)}")
138	for topology_id, topology in self._topology_map.items():	4✔
139	if topology.has_node_by_id(node_id):	4✔
140	domain_id = topology_id	4✔
141	break	4✔
142
143	return domain_id	4!
144
145	def generate_id(self):	4✔
146	self._topology.id = SDX_TOPOLOGY_ID_prefix	4✔
147	self._topology.version = TOPOLOGY_INITIAL_VERSION	4✔
148	return id	4✔
149
150	def remove_topology(self, topology_id):	4✔
151	self._topology_map.pop(topology_id, None)	×
152	self.update_version(False)	×
153	self.update_timestamp()	×
154
155	def is_link_interdomain(self, link, topology):	4✔
156	"""
157	Check if a link is an interdomain link.
158	"""
159	for port in link.ports:	4✔
160	if port["id"] not in self._port_map:	4✔
NEW 161	return True	×
162	return False	4✔
163
164	def update_topology(self, data):	4✔
165	# likely adding new inter-domain links
166	update_handler = TopologyHandler()	4✔
167	topology = update_handler.import_topology_data(data)	4✔
168	self._topology_map[topology.id] = topology	4✔
169
170	# Nodes.
171	nodes = topology.nodes	4✔
172	for node in nodes:	4✔
173	self._topology.remove_node(node.id)	4✔
174
175	# Links.
176	links = topology.links	4✔
177	for link in links:	4!
178	if not self.is_link_interdomain(link, topology):	4✔
179	# print(link.id+";......."+str(link.nni))
180	self._topology.remove_link(link.id)	4✔
181	for port in link.ports:	4✔
182	self._port_map.pop(port["id"])	4✔
183
184	# Check the inter-domain links first.
185	num_interdomain_link = self.inter_domain_check(topology)	4✔
186	if num_interdomain_link == 0:	4✔
187	self._logger.warning("Warning: no interdomain links detected!")	4✔
188
189	# Nodes.
190	nodes = topology.nodes	4!
191	self._topology.add_nodes(nodes)	4✔
192
193	# Links.
194	links = topology.links	4✔
195	self._topology.add_links(links)	4✔
196
197	self.update_version(True)	4✔
198	self.update_timestamp()	4✔
199
200	def update_version(self, sub: bool):	4✔
201	try:	4✔
202	[ver, sub_ver] = self._topology.version.split(".")	4✔
203	except ValueError:	×
204	ver = self._topology.version	×
205	sub_ver = "0"	×
206
207	self._topology.version = self.new_version(ver, sub_ver, sub)	4✔
208
209	return self._topology.version	4✔
210
211	def new_version(self, ver, sub_ver, sub: bool):	4✔
212	if not sub:	4✔
213	ver = str(int(ver) + 1)	4✔
214	sub_ver = "0"	4✔
215	else:
216	sub_ver = str(int(sub_ver) + 1)	4✔
217
218	return ver + "." + sub_ver	4✔
219
220	def update_timestamp(self):	4✔
221	ct = datetime.datetime.now().isoformat()	4✔
222	self._topology.timestamp = ct	4✔
223
224	return ct	4✔
225
226	def inter_domain_check(self, topology):	4✔
227	interdomain_port_dict = {}	4✔
228	num_interdomain_link = 0	4✔
229	links = topology.links	4✔
230	link_dict = {}	4✔
231	for link in links:	4✔
232	link_dict[link.id] = link	4✔
233	for port in link.ports:	4✔
234	interdomain_port_dict[port["id"]] = link	4✔
235
236	# ToDo: raise an warning or exception
237	if len(interdomain_port_dict) == 0:	4✔
238	self._logger.info("interdomain_port_dict==0")	×
239	return False	×
240
241	# match any ports in the existing topology
242	for port_id in interdomain_port_dict:	4✔
243	# print("interdomain_port:")
244	# print(port_id)
245	for existing_port, existing_link in self._port_map.items():	4✔
246	# print(existing_port)
247	if port_id == existing_port:	4✔
248	# print("Interdomain port:" + port_id)
249	# remove redundant link between two domains
250	self._topology.remove_link(existing_link.id)	4✔
251	num_interdomain_link = +1	4✔
252	self._port_map[port_id] = interdomain_port_dict[port_id]	4✔
253
254	return num_interdomain_link	4✔
255
256	# adjacent matrix of the graph, in jason?
257	def generate_graph(self):	4✔
258	graph = nx.Graph()	4✔
259
260	if self._topology is None:	4✔
261	self._logger.warning("We do not have a topology yet")	×
262	return None	×
263
264	links = self._topology.links	4✔
265	for link in links:	4✔
266	inter_domain_link = False	4✔
267	ports = link.ports	4✔
268	end_nodes = []	4✔
269	for port in ports:	4✔
270	node = self._topology.get_node_by_port(port["id"])	4✔
271	if node is None:	4✔
272	self._logger.warning(	4✔
273	f"This port (id: {port['id']}) does not belong to "
274	f"any node in the topology, likely a Non-SDX port!"
275	)
276	inter_domain_link = True	4✔
277	break	4✔
278	else:
279	end_nodes.append(node)	4✔
280	# print("graph node:"+node.id)
281	if not inter_domain_link:	4✔
282	graph.add_edge(end_nodes[0].id, end_nodes[1].id)	4✔
283	edge = graph.edges[end_nodes[0].id, end_nodes[1].id]	4✔
284	edge["id"] = link.id	4✔
285	edge["latency"] = link.latency	4✔
286	edge["bandwidth"] = link.bandwidth	4✔
287	edge["residual_bandwidth"] = link.residual_bandwidth	4✔
288	edge["weight"] = 1000.0 * (1.0 / link.residual_bandwidth)	4✔
289	edge["packet_loss"] = link.packet_loss	4✔
290	edge["availability"] = link.availability	4✔
291
292	return graph	4✔
293
294	def generate_grenml(self):	4✔
295	self.converter = GrenmlConverter(self._topology)	×
296
297	return self.converter.read_topology()	×
298
299	def add_domain_service(self):	4✔
300	pass	×
301
302	# may need to read from a configuration file.
303	def update_private_properties(self):	4✔
304	pass	×
305
306	# on performance properties for now
307	def update_link_property(self, link_id, property, value):	4✔
308	# 1. update the individual topology
309	for id, topology in self._topology_map.items():	4✔
310	links = topology.links	4✔
311	for link in links:	4✔
312	self._logger.info(f"link.id={link.id}; id={id}")	4✔
313	if link.id == link_id:	4✔
314	setattr(link, property, value)	4✔
315	self._logger.info("updated the link.")	4✔
316	# 1.2 need to change the sub_ver of the topology?
317
318	# 2. check on the inter-domain link?
319	# 3. update the interodamin topology
320	links = self._topology.links	4✔
321	for link in links:	4✔
322	if link.id == link_id:	4✔
323	setattr(link, property, value)	4✔
324	self._logger.info("updated the link.")	4✔
325	# 2.2 need to change the sub_ver of the topology?
326
327	self.update_version(True)	4✔
328	self.update_timestamp()	4✔
329	# 4. Signal update the (networkx) graph
330
331	# 5. signal Reoptimization of TE?
332
333	def update_element_property_json(self, data, element, element_id, property, value):	4✔
334	elements = data[element]	4✔
335	for element in elements:	4✔
336	if element["id"] == element_id:	4✔
337	element[property] = value	4✔
338
339	try:	4✔
340	[ver, sub_ver] = data["version"].split(".")	4✔
341	except ValueError:	×
342	ver = "0"	×
343	sub_ver = "0"	×
344
345	data["version"] = self.new_version(ver, sub_ver, True)	4✔
346	data["timestamp"] = datetime.datetime.now().isoformat()	4✔
347
348	def get_port_by_id(self, port_id: str):	4✔
349	"""
350	Given port id, returns a Port.
351	"""
352	for node in self.get_topology().nodes:	4✔
353	for port in node.ports:	4✔
354	if port.id == port_id:	4✔
355	return port.to_dict()	4✔
356	return None	×
357
358	def are_two_ports_same_domain(self, port1_id: str, port2_id: str):	4✔
359	"""
360	Check if two ports are in the same domain.
361	"""
362	node1 = self.get_topology().get_node_by_port(port1_id)	4✔
363	node2 = self.get_topology().get_node_by_port(port2_id)	4✔
364	if node1 is None or node2 is None:	4✔
365	return False	×
366
367	domain1 = self.get_domain_name(node1.id)	4✔
368	domain2 = self.get_domain_name(node2.id)	4✔
369	return domain1 == domain2	4✔
370
371	def update_node_property(self):	4✔
372	pass	×
373
374	def update_port_property(self):	4✔
375	pass	×

atlanticwave-sdx / pce / 9140549691

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous