build.automate.Automate.check_step() - Code Metrics - Inspection of "Merge pull request #42 from Alopalao/feature/of_v0..." - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( f5b80f...0cda6c )

by Vinicius

created 2022-09-21 18:49 UTC

build.automate.Automate.check_step() A

↳ Parent: build.automate

Complexity

Conditions

Size

Total Lines	5
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	2
dl	0
loc	5
rs	10
c	0
b	0
f	0
ccs	3
cts	3
cp	1
crap	1

"""Automate circuit traces."""

import time

import requests
from kytos.core import KytosEvent, log
from napps.amlight.sdntrace_cp import settings
from napps.amlight.sdntrace_cp.scheduler import Scheduler
from napps.amlight.sdntrace_cp.utils import clean_circuits, format_result
from pyof.v0x04.common.port import PortNo as Port13


class Automate:
    """Find all circuits and automate trace execution."""

    def __init__(self, tracer):
        self._tracer = tracer
        self._circuits = []
        self.find_circuits()
        self.ids = set()
        self.scheduler = Scheduler()

    # pylint: disable=too-many-nested-blocks, too-many-branches
    def find_circuits(self):
        """Discover all circuits in a topology.

        Using the list of flows per switch, run control plane
        traces to find a list of circuits."""
        all_flows = {}
        circuits = []

        for switch in self._tracer.controller.switches.copy().values():
            all_flows[switch] = []
            if switch.ofp_version == '0x04':
                controller_port = Port13.OFPP_CONTROLLER

            try:
                for flow in switch.generic_flows:
                    action_ok = False
                    in_port_ok = False
                    if 'in_port' in flow.match and flow.match['in_port'] != 0:
                        in_port_ok = True
                    if in_port_ok:
                        for action in flow.actions:
                            if action.action_type == 'output' \
                                    and action.port != controller_port:

                                action_ok = True
                    if action_ok:
                        all_flows[switch].append(flow)
            except AttributeError:
                pass

        for switch, flows in all_flows.items():
            for flow in flows:
                in_port = flow.match['in_port']
                vlan = None
                if 'vlan_vid' in flow.match:
                    vlan = flow.match['vlan_vid']
                if switch.ofp_version == '0x04':
                    in_port = in_port.value
                    if vlan:
                        vlan = vlan.value
                entries = {
                    'trace': {
                        'switch': {
                            'dpid': switch.dpid,
                            'in_port': in_port
                        },
                        'eth': {
                            'dl_vlan': vlan
                        }
                    }
                }
                result = self._tracer.tracepath(entries)
                circuits.append({'circuit': format_result(result),
                                 'entries': entries})

        self._circuits = clean_circuits(circuits, self._tracer.controller)

    def run_traces(self):
        """Run traces for all circuits."""

        results = []
        for circuit in self._circuits:
            entries = circuit['entries']
            result = self._tracer.tracepath(entries)
            try:
                result = format_result(result)
                if result != circuit['circuit']:
                    results.append(circuit)
            except KeyError:
                results.append(circuit)
        log.debug('Results %s, size %s', results, len(self._circuits))
        return results

    def get_circuit(self, circuit):
        """Find the given circuit in the list of circuits."""
        for steps in self._circuits:
            endpoint_a = steps['circuit'][0]
            endpoint_z = steps['circuit'][-1]
            # pylint: disable=too-many-boolean-expressions
            if (circuit['dpid_a'] == endpoint_a['dpid'] and
                    circuit['port_a'] == endpoint_a['in_port'] and
                    circuit['vlan_a'] == endpoint_a['in_vlan'] and
                    circuit['dpid_z'] == endpoint_z['dpid'] and
                    circuit['port_z'] == endpoint_z['out_port'] and
                    circuit['vlan_z'] == endpoint_z['out_vlan']):

                return steps['circuit']
        return None

    def _check_trace(self, circuit, trace):
        steps = self.get_circuit(circuit)
        if steps:
            if len(steps) != len(trace) - 1:
                return False
            for i, step in enumerate(steps):
                if not self.check_step(step, trace[i]):
                    return False
        else:
            return False
        return True

    def run_important_traces(self):
        """Run SDNTrace in data plane for important circuits as defined
            by user."""
        event = KytosEvent(name='amlight/kytos_courier.slack_send')
        content = {
            'channel': settings.SLACK_CHANNEL,
            'source': 'amlight/sdntrace_cp'
        }

        try:
            important_circuits = settings.IMPORTANT_CIRCUITS
        except AttributeError:
            return

        for circuit in important_circuits:
            entries = {
                'trace': {
                    'switch': {
                        'dpid': circuit['dpid_a'],
                        'in_port': circuit['port_a']
                    },
                    'eth': {
                        'dl_vlan': circuit['vlan_a']
                    }
                }
            }
            result = requests.put(settings.SDNTRACE_URL, json=entries)
            trace = result.json()
            trace_id = trace['result']['trace_id']
            step_type = None
            while step_type != 'last':
                time.sleep(5)
                result = requests.get(f'{settings.SDNTRACE_URL}/{trace_id}')
                trace = result.json()
                step_type = trace['result'][-1]['type']
            check = self._check_trace(circuit, trace['result'])
            if check is False:
                content['m_body'] = 'Trace in data plane different from ' \
                                    'trace in control plane for circuit ' \
                                    f'{circuit}'
                event.content['message'] = content
                self._tracer.controller.buffers.app.put(event)

    def schedule_id(self, id_):
        """Keep track of scheduled ids"""
        if not isinstance(id_, str):
            raise AttributeError("Invalid id type, for schedule.")
        self.ids.add(id_)
        return id_

    def schedule_traces(self, settings_=settings):
        """Check for invalid arguments from schedule"""
        if settings_.TRIGGER_SCHEDULE_TRACES:
            id_ = self.schedule_id('automatic_traces')
            trigger = settings_.SCHEDULE_TRIGGER
            kwargs = settings_.SCHEDULE_ARGS
            if (not isinstance(kwargs, dict) or
                    not isinstance(trigger, str)):
                raise AttributeError("Invalid attributes for "
                                     "job to be scheduled.")
            trigger_args = {}
            trigger_args['kwargs'] = {'trigger': trigger}
            trigger_args['kwargs'].update(kwargs)
            self.scheduler.add_callable(id_, self.run_traces,
                                        **trigger_args['kwargs'])
            return self.scheduler.get_job(id_)
        return None

    def schedule_important_traces(self, settings_=settings):
        """Check for invalid important arguments from schedule"""
        if settings_.TRIGGER_IMPORTANT_CIRCUITS:
            id_ = self.schedule_id('automatic_important_traces')
            trigger = settings_.IMPORTANT_CIRCUITS_TRIGGER
            kwargs = settings_.IMPORTANT_CIRCUITS_ARGS
            if (not isinstance(kwargs, dict) or
                    not isinstance(trigger, str)):
                raise AttributeError("Invalid attributes for "
                                     "job to be scheduled.")
            trigger_args = {}
            trigger_args['kwargs'] = {'trigger': trigger}
            trigger_args['kwargs'].update(kwargs)
            return self.scheduler.add_callable(id_,
                                               self.run_important_traces,
                                               **trigger_args['kwargs'])
        return None

    def unschedule_ids(self, id_set=None):
        """Remove ids to be unschedule"""
        if id_set is None:
            id_set = self.ids.copy()
        while id_set:
            id_ = id_set.pop()
            self.ids.discard(id_)
            self.scheduler.remove_job(id_)

    def sheduler_shutdown(self, wait):
        """Shutdown scheduler"""
        self.scheduler.shutdown(wait)

    @staticmethod
    def check_step(circuit_step, trace_step):
        """Check if a step in SDNTrace in data plane is what it should"""
        return (circuit_step['dpid'] == trace_step['dpid'] and
                circuit_step['in_port'] == trace_step['port'])


1		"""Automate circuit traces."""
2
3	1	import time
4
5	1	import requests
6	1	from kytos.core import KytosEvent, log
7	1	from napps.amlight.sdntrace_cp import settings
8	1	from napps.amlight.sdntrace_cp.scheduler import Scheduler
9	1	from napps.amlight.sdntrace_cp.utils import clean_circuits, format_result
10	1	from pyof.v0x04.common.port import PortNo as Port13
11
12
13	1	class Automate:
14		"""Find all circuits and automate trace execution."""
15
16	1	def __init__(self, tracer):
17	1	self._tracer = tracer
18	1	self._circuits = []
19	1	self.find_circuits()
20	1	self.ids = set()
21	1	self.scheduler = Scheduler()
22
23		# pylint: disable=too-many-nested-blocks, too-many-branches
24	1	def find_circuits(self):
25		"""Discover all circuits in a topology.
26
27		Using the list of flows per switch, run control plane
28		traces to find a list of circuits."""
29	1	all_flows = {}
30	1	circuits = []
31
32	1	for switch in self._tracer.controller.switches.copy().values():
33	1	all_flows[switch] = []
34	1	if switch.ofp_version == '0x04':
35	1	controller_port = Port13.OFPP_CONTROLLER
36
37	1	try:
38	1	for flow in switch.generic_flows:
39	1	action_ok = False
40	1	in_port_ok = False
41	1	if 'in_port' in flow.match and flow.match['in_port'] != 0:
42	1	in_port_ok = True
43	1	if in_port_ok:
44	1	for action in flow.actions:
45	1	if action.action_type == 'output' \
46		and action.port != controller_port:
		0 ignored issues – show introduced 2022-09-16 19:14 UTC by Report Bug Copy Issue Report The variable `controller_port` does not seem to be defined for all execution paths. Loading history...
47	1	action_ok = True
48	1	if action_ok:
49	1	all_flows[switch].append(flow)
50		except AttributeError:
51		pass
52
53	1	for switch, flows in all_flows.items():
54	1	for flow in flows:
55	1	in_port = flow.match['in_port']
56	1	vlan = None
57	1	if 'vlan_vid' in flow.match:
58		vlan = flow.match['vlan_vid']
59	1	if switch.ofp_version == '0x04':
60	1	in_port = in_port.value
61	1	if vlan:
62		vlan = vlan.value
63	1	entries = {
64		'trace': {
65		'switch': {
66		'dpid': switch.dpid,
67		'in_port': in_port
68		},
69		'eth': {
70		'dl_vlan': vlan
71		}
72		}
73		}
74	1	result = self._tracer.tracepath(entries)
75	1	circuits.append({'circuit': format_result(result),
76		'entries': entries})
77
78	1	self._circuits = clean_circuits(circuits, self._tracer.controller)
79
80	1	def run_traces(self):
81		"""Run traces for all circuits."""
82
83	1	results = []
84	1	for circuit in self._circuits:
85	1	entries = circuit['entries']
86	1	result = self._tracer.tracepath(entries)
87	1	try:
88	1	result = format_result(result)
89	1	if result != circuit['circuit']:
90	1	results.append(circuit)
91		except KeyError:
92		results.append(circuit)
93	1	log.debug('Results %s, size %s', results, len(self._circuits))
94	1	return results
95
96	1	def get_circuit(self, circuit):
97		"""Find the given circuit in the list of circuits."""
98	1	for steps in self._circuits:
99	1	endpoint_a = steps['circuit'][0]
100	1	endpoint_z = steps['circuit'][-1]
101		# pylint: disable=too-many-boolean-expressions
102	1	if (circuit['dpid_a'] == endpoint_a['dpid'] and
103		circuit['port_a'] == endpoint_a['in_port'] and
104		circuit['vlan_a'] == endpoint_a['in_vlan'] and
105		circuit['dpid_z'] == endpoint_z['dpid'] and
106		circuit['port_z'] == endpoint_z['out_port'] and
107		circuit['vlan_z'] == endpoint_z['out_vlan']):
108
109	1	return steps['circuit']
110	1	return None
111
112	1	def _check_trace(self, circuit, trace):
113	1	steps = self.get_circuit(circuit)
114	1	if steps:
115	1	if len(steps) != len(trace) - 1:
116	1	return False
117	1	for i, step in enumerate(steps):
118	1	if not self.check_step(step, trace[i]):
119	1	return False
120		else:
121	1	return False
122	1	return True
123
124	1	def run_important_traces(self):
125		"""Run SDNTrace in data plane for important circuits as defined
126		by user."""
127	1	event = KytosEvent(name='amlight/kytos_courier.slack_send')
128	1	content = {
129		'channel': settings.SLACK_CHANNEL,
130		'source': 'amlight/sdntrace_cp'
131		}
132
133	1	try:
134	1	important_circuits = settings.IMPORTANT_CIRCUITS
135		except AttributeError:
136		return
137
138	1	for circuit in important_circuits:
139	1	entries = {
140		'trace': {
141		'switch': {
142		'dpid': circuit['dpid_a'],
143		'in_port': circuit['port_a']
144		},
145		'eth': {
146		'dl_vlan': circuit['vlan_a']
147		}
148		}
149		}
150	1	result = requests.put(settings.SDNTRACE_URL, json=entries)
151	1	trace = result.json()
152	1	trace_id = trace['result']['trace_id']
153	1	step_type = None
154	1	while step_type != 'last':
155	1	time.sleep(5)
156	1	result = requests.get(f'{settings.SDNTRACE_URL}/{trace_id}')
157	1	trace = result.json()
158	1	step_type = trace['result'][-1]['type']
159	1	check = self._check_trace(circuit, trace['result'])
160	1	if check is False:
161	1	content['m_body'] = 'Trace in data plane different from ' \
162		'trace in control plane for circuit ' \
163		f'{circuit}'
164	1	event.content['message'] = content
165	1	self._tracer.controller.buffers.app.put(event)
166
167	1	def schedule_id(self, id_):
168		"""Keep track of scheduled ids"""
169	1	if not isinstance(id_, str):
170	1	raise AttributeError("Invalid id type, for schedule.")
171	1	self.ids.add(id_)
172	1	return id_
173
174	1	def schedule_traces(self, settings_=settings):
175		"""Check for invalid arguments from schedule"""
176	1	if settings_.TRIGGER_SCHEDULE_TRACES:
177	1	id_ = self.schedule_id('automatic_traces')
178	1	trigger = settings_.SCHEDULE_TRIGGER
179	1	kwargs = settings_.SCHEDULE_ARGS
180	1	if (not isinstance(kwargs, dict) or
181		not isinstance(trigger, str)):
182	1	raise AttributeError("Invalid attributes for "
183		"job to be scheduled.")
184	1	trigger_args = {}
185	1	trigger_args['kwargs'] = {'trigger': trigger}
186	1	trigger_args['kwargs'].update(kwargs)
187	1	self.scheduler.add_callable(id_, self.run_traces,
188		**trigger_args['kwargs'])
189	1	return self.scheduler.get_job(id_)
190	1	return None
191
192	1	def schedule_important_traces(self, settings_=settings):
193		"""Check for invalid important arguments from schedule"""
194	1	if settings_.TRIGGER_IMPORTANT_CIRCUITS:
195	1	id_ = self.schedule_id('automatic_important_traces')
196	1	trigger = settings_.IMPORTANT_CIRCUITS_TRIGGER
197	1	kwargs = settings_.IMPORTANT_CIRCUITS_ARGS
198	1	if (not isinstance(kwargs, dict) or
199		not isinstance(trigger, str)):
200	1	raise AttributeError("Invalid attributes for "
201		"job to be scheduled.")
202	1	trigger_args = {}
203	1	trigger_args['kwargs'] = {'trigger': trigger}
204	1	trigger_args['kwargs'].update(kwargs)
205	1	return self.scheduler.add_callable(id_,
206		self.run_important_traces,
207		**trigger_args['kwargs'])
208	1	return None
209
210	1	def unschedule_ids(self, id_set=None):
211		"""Remove ids to be unschedule"""
212	1	if id_set is None:
213	1	id_set = self.ids.copy()
214	1	while id_set:
215	1	id_ = id_set.pop()
216	1	self.ids.discard(id_)
217	1	self.scheduler.remove_job(id_)
218
219	1	def sheduler_shutdown(self, wait):
220		"""Shutdown scheduler"""
221		self.scheduler.shutdown(wait)
222
223	1	@staticmethod
224	1	def check_step(circuit_step, trace_step):
225		"""Check if a step in SDNTrace in data plane is what it should"""
226	1	return (circuit_step['dpid'] == trace_step['dpid'] and
227		circuit_step['in_port'] == trace_step['port'])
228

kytos-ng / sdntrace_cp

Push — master ( f5b80f...0cda6c )

build.automate.Automate.check_step() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like