build.main.Main.remove_int_flows() - Code Metrics - Inspection of "refactor: minimized IO, bulk updates, general enh..." - kytos-ng/telemetry_int - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#35)

by Vinicius

created 2023-08-25 19:01 UTC

build.main.Main.remove_int_flows() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	27
Code Lines	20

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes

Metric	Value
cc	3
eloc	20
nop	2
dl	0
loc	27
ccs	0
cts	9
cp	0
crap	12
rs	9.4
c	0
b	0
f	0

"""Main module of kytos/telemetry Network Application.

Napp to deploy In-band Network Telemetry over Ethernet Virtual Circuits

"""

from napps.kytos.telemetry_int import utils
from tenacity import RetryError

from kytos.core import KytosNApp, log, rest
from kytos.core.rest_api import HTTPException, JSONResponse, Request, aget_json_or_400

from .exceptions import (
    EVCHasINT,
    EVCHasNoINT,
    EVCNotFound,
    FlowsNotFound,
    ProxyPortNotFound,
    ProxyPortStatusNotUP,
    UnrecoverableError,
)
from .kytos_api_helper import get_evc, get_evcs
from .managers.int import INTManager

# pylint: disable=fixme


class Main(KytosNApp):
    """Main class of kytos/telemetry NApp.

    This class is the entry point for this NApp.
    """

    def setup(self):
        """Replace the '__init__' method for the KytosNApp subclass.

        The setup method is automatically called by the controller when your
        application is loaded.

        So, if you have any setup routine, insert it here.
        """

        self.int_manager = INTManager(self.controller)

    def execute(self):
        """Run after the setup method execution.

        You can also use this method in loop mode if you add to the above setup
        method a line like the following example:

            self.execute_as_loop(30)  # 30-second interval.
        """

    def shutdown(self):
        """Run when your NApp is unloaded.

        If you have some cleanup procedure, insert it here.
        """

    @rest("v1/evc/enable", methods=["POST"])

    async def enable_telemetry(self, request: Request) -> JSONResponse:
        """REST to enable INT flows on EVCs.

        If a list of evc_ids is empty, it'll enable on non-INT EVCs.
        """

        try:
            content = await aget_json_or_400(request)
            evc_ids = content["evc_ids"]
            force = bool(content.get("force", False))
        except (TypeError, KeyError):
            raise HTTPException(400, detail=f"Invalid payload: {content}")

        try:
            evcs = await get_evcs() if len(evc_ids) != 1 else await get_evc(evc_ids[0])
        except RetryError as exc:
            exc_error = str(exc.last_attempt.exception())
            log.error(exc_error)
            raise HTTPException(503, detail=exc_error)

        if evc_ids:
            evcs = {evc_id: evcs.get(evc_id, {}) for evc_id in evc_ids}
        else:
            evcs = {k: v for k, v in evcs.items() if not utils.has_int_enabled(v)}
            if not evcs:
                # There's no non-INT EVCs to get enabled.
                return JSONResponse({})

        try:
            await self.int_manager.enable_int(evcs, force)
        except (EVCNotFound, FlowsNotFound, ProxyPortNotFound) as exc:
            raise HTTPException(404, detail=str(exc))
        except (EVCHasINT, ProxyPortStatusNotUP) as exc:
            raise HTTPException(400, detail=str(exc))
        except RetryError as exc:
            exc_error = str(exc.last_attempt.exception())
            log.error(exc_error)
            raise HTTPException(503, detail=exc_error)
        except UnrecoverableError as exc:
            exc_error = str(exc)
            log.error(exc_error)
            raise HTTPException(500, detail=exc_error)

        return JSONResponse({}, status_code=201)

    @rest("v1/evc/disable", methods=["POST"])

    async def disable_telemetry(self, request: Request) -> JSONResponse:
        """REST to disable/remove INT flows for an EVC_ID

        If a list of evc_ids is empty, it'll disable on all INT EVCs.
        """
        try:
            content = await aget_json_or_400(request)
            evc_ids = content["evc_ids"]
            force = bool(content.get("force", False))
        except (TypeError, KeyError):
            raise HTTPException(400, detail=f"Invalid payload: {content}")

        try:
            evcs = await get_evcs() if len(evc_ids) != 1 else await get_evc(evc_ids[0])
        except RetryError as exc:
            exc_error = str(exc.last_attempt.exception())
            log.error(exc_error)
            raise HTTPException(503, detail=exc_error)

        if evc_ids:
            evcs = {evc_id: evcs.get(evc_id, {}) for evc_id in evc_ids}
        else:
            evcs = {k: v for k, v in evcs.items() if utils.has_int_enabled(v)}
            if not evcs:
                # There's no INT EVCs to get disabled.
                return JSONResponse({})

        try:
            await self.int_manager.disable_int(evcs, force)
        except EVCNotFound as exc:
            raise HTTPException(404, detail=str(exc))
        except EVCHasNoINT as exc:
            raise HTTPException(400, detail=str(exc))
        except RetryError as exc:
            exc_error = str(exc.last_attempt.exception())
            log.error(exc_error)
            raise HTTPException(503, detail=exc_error)
        except UnrecoverableError as exc:
            exc_error = str(exc)
            log.error(exc_error)
            raise HTTPException(500, detail=exc_error)

        return JSONResponse({})

    @rest("v1/evc")
    def get_evcs(self, _request: Request) -> JSONResponse:
        """REST to return the list of EVCs with INT enabled"""
        return JSONResponse(utils.get_evc_with_telemetry())

    @rest("v1/sync")
    def sync_flows(self, _request: Request) -> JSONResponse:
        """Endpoint to force the telemetry napp to search for INT flows and delete them
        accordingly to the evc metadata."""

        # TODO
        # for evc_id in get_evcs_ids():
        return JSONResponse("TBD")

    @rest("v1/evc/update")
    def update_evc(self, _request: Request) -> JSONResponse:
        """If an EVC changed from unidirectional to bidirectional telemetry,
        make the change."""
        return JSONResponse({})

    # Event-driven methods: future
    def listen_for_new_evcs(self):
        """Change newly created EVC to INT-enabled EVC based on the metadata field
        (future)"""
        pass

    def listen_for_evc_change(self):
        """Change newly created EVC to INT-enabled EVC based on the
        metadata field (future)"""
        pass

    def listen_for_path_changes(self):
        """Change EVC's new path to INT-enabled EVC based on the metadata field
        when there is a path change. (future)"""
        pass

    def listen_for_evcs_removed(self):
        """Remove all INT flows belonging the just removed EVC (future)"""
        pass

    def listen_for_topology_changes(self):
        """If the topology changes, make sure it is not the loop ports.
        If so, update proxy ports"""
        # TODO:
        # self.proxy_ports = create_proxy_ports(self.proxy_ports)
        pass


1		"""Main module of kytos/telemetry Network Application.
2
3		Napp to deploy In-band Network Telemetry over Ethernet Virtual Circuits
4
5		"""
6
7		from napps.kytos.telemetry_int import utils
8		from tenacity import RetryError
9
10		from kytos.core import KytosNApp, log, rest
11		from kytos.core.rest_api import HTTPException, JSONResponse, Request, aget_json_or_400
12
13		from .exceptions import (
14		EVCHasINT,
15		EVCHasNoINT,
16		EVCNotFound,
17		FlowsNotFound,
18		ProxyPortNotFound,
19		ProxyPortStatusNotUP,
20		UnrecoverableError,
21		)
22		from .kytos_api_helper import get_evc, get_evcs
23		from .managers.int import INTManager
24
25		# pylint: disable=fixme
26
27
28		class Main(KytosNApp):
29		"""Main class of kytos/telemetry NApp.
30
31		This class is the entry point for this NApp.
32		"""
33
34		def setup(self):
35		"""Replace the '__init__' method for the KytosNApp subclass.
36
37		The setup method is automatically called by the controller when your
38		application is loaded.
39
40		So, if you have any setup routine, insert it here.
41		"""
42
43		self.int_manager = INTManager(self.controller)
44
45		def execute(self):
46		"""Run after the setup method execution.
47
48		You can also use this method in loop mode if you add to the above setup
49		method a line like the following example:
50
51		self.execute_as_loop(30) # 30-second interval.
52		"""
53
54		def shutdown(self):
55		"""Run when your NApp is unloaded.
56
57		If you have some cleanup procedure, insert it here.
58		"""
59
60	View Code Duplication	@rest("v1/evc/enable", methods=["POST"])
		0 ignored issues – show Duplication introduced 2023-08-02 20:05 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
61		async def enable_telemetry(self, request: Request) -> JSONResponse:
62		"""REST to enable INT flows on EVCs.
63
64		If a list of evc_ids is empty, it'll enable on non-INT EVCs.
65		"""
66
67		try:
68		content = await aget_json_or_400(request)
69		evc_ids = content["evc_ids"]
70		force = bool(content.get("force", False))
71		except (TypeError, KeyError):
72		raise HTTPException(400, detail=f"Invalid payload: {content}")
73
74		try:
75		evcs = await get_evcs() if len(evc_ids) != 1 else await get_evc(evc_ids[0])
76		except RetryError as exc:
77		exc_error = str(exc.last_attempt.exception())
78		log.error(exc_error)
79		raise HTTPException(503, detail=exc_error)
80
81		if evc_ids:
82		evcs = {evc_id: evcs.get(evc_id, {}) for evc_id in evc_ids}
83		else:
84		evcs = {k: v for k, v in evcs.items() if not utils.has_int_enabled(v)}
85		if not evcs:
86		# There's no non-INT EVCs to get enabled.
87		return JSONResponse({})
88
89		try:
90		await self.int_manager.enable_int(evcs, force)
91		except (EVCNotFound, FlowsNotFound, ProxyPortNotFound) as exc:
92		raise HTTPException(404, detail=str(exc))
93		except (EVCHasINT, ProxyPortStatusNotUP) as exc:
94		raise HTTPException(400, detail=str(exc))
95		except RetryError as exc:
96		exc_error = str(exc.last_attempt.exception())
97		log.error(exc_error)
98		raise HTTPException(503, detail=exc_error)
99		except UnrecoverableError as exc:
100		exc_error = str(exc)
101		log.error(exc_error)
102		raise HTTPException(500, detail=exc_error)
103
104		return JSONResponse({}, status_code=201)
105
106	View Code Duplication	@rest("v1/evc/disable", methods=["POST"])
		0 ignored issues – show Duplication introduced 2023-08-02 20:05 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
107		async def disable_telemetry(self, request: Request) -> JSONResponse:
108		"""REST to disable/remove INT flows for an EVC_ID
109
110		If a list of evc_ids is empty, it'll disable on all INT EVCs.
111		"""
112		try:
113		content = await aget_json_or_400(request)
114		evc_ids = content["evc_ids"]
115		force = bool(content.get("force", False))
116		except (TypeError, KeyError):
117		raise HTTPException(400, detail=f"Invalid payload: {content}")
118
119		try:
120		evcs = await get_evcs() if len(evc_ids) != 1 else await get_evc(evc_ids[0])
121		except RetryError as exc:
122		exc_error = str(exc.last_attempt.exception())
123		log.error(exc_error)
124		raise HTTPException(503, detail=exc_error)
125
126		if evc_ids:
127		evcs = {evc_id: evcs.get(evc_id, {}) for evc_id in evc_ids}
128		else:
129		evcs = {k: v for k, v in evcs.items() if utils.has_int_enabled(v)}
130		if not evcs:
131		# There's no INT EVCs to get disabled.
132		return JSONResponse({})
133
134		try:
135		await self.int_manager.disable_int(evcs, force)
136		except EVCNotFound as exc:
137		raise HTTPException(404, detail=str(exc))
138		except EVCHasNoINT as exc:
139		raise HTTPException(400, detail=str(exc))
140		except RetryError as exc:
141		exc_error = str(exc.last_attempt.exception())
142		log.error(exc_error)
143		raise HTTPException(503, detail=exc_error)
144		except UnrecoverableError as exc:
145		exc_error = str(exc)
146		log.error(exc_error)
147		raise HTTPException(500, detail=exc_error)
148
149		return JSONResponse({})
150
151		@rest("v1/evc")
152		def get_evcs(self, _request: Request) -> JSONResponse:
153		"""REST to return the list of EVCs with INT enabled"""
154		return JSONResponse(utils.get_evc_with_telemetry())
155
156		@rest("v1/sync")
157		def sync_flows(self, _request: Request) -> JSONResponse:
158		"""Endpoint to force the telemetry napp to search for INT flows and delete them
159		accordingly to the evc metadata."""
160
161		# TODO
162		# for evc_id in get_evcs_ids():
163		return JSONResponse("TBD")
164
165		@rest("v1/evc/update")
166		def update_evc(self, _request: Request) -> JSONResponse:
167		"""If an EVC changed from unidirectional to bidirectional telemetry,
168		make the change."""
169		return JSONResponse({})
170
171		# Event-driven methods: future
172		def listen_for_new_evcs(self):
173		"""Change newly created EVC to INT-enabled EVC based on the metadata field
174		(future)"""
175		pass
176
177		def listen_for_evc_change(self):
178		"""Change newly created EVC to INT-enabled EVC based on the
179		metadata field (future)"""
180		pass
181
182		def listen_for_path_changes(self):
183		"""Change EVC's new path to INT-enabled EVC based on the metadata field
184		when there is a path change. (future)"""
185		pass
186
187		def listen_for_evcs_removed(self):
188		"""Remove all INT flows belonging the just removed EVC (future)"""
189		pass
190
191		def listen_for_topology_changes(self):
192		"""If the topology changes, make sure it is not the loop ports.
193		If so, update proxy ports"""
194		# TODO:
195		# self.proxy_ports = create_proxy_ports(self.proxy_ports)
196		pass
197

kytos-ng / telemetry_int

Pull Request — master (#35)

build.main.Main.remove_int_flows() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like