build.models.MaintenanceWindow.maintenance_event() - Code Metrics - Inspection of "Add in persistence to maintenance windows" - kytos-ng/maintenance - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#64)

unknown

created 2022-12-12 18:40 UTC

build.models.MaintenanceWindow.maintenance_event() A

↳ Parent: build.models

Complexity

Conditions

Size

Total Lines	20
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	15
CRAP Score	4.0039

Importance

Changes

Metric	Value
cc	4
eloc	16
nop	3
dl	0
loc	20
ccs	15
cts	16
cp	0.9375
crap	4.0039
rs	9.6
c	0
b	0
f	0

"""Models used by the maintenance NApp.

This module define models for the maintenance window itself and the
scheduler.
"""
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import NewType
from uuid import uuid4

import pytz
from apscheduler.jobstores.base import JobLookupError
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.base import BaseScheduler
from pydantic import BaseModel, Field, validator, root_validator

from kytos.core import KytosEvent, log
from kytos.core.controller import Controller

TIME_FMT = "%Y-%m-%dT%H:%M:%S%z"

class Status(str, Enum):
    """Maintenance windows status."""

    PENDING = 'pending'
    RUNNING = 'running'
    FINISHED = 'finished'


MaintenanceID = NewType('MaintenanceID', str)


class MaintenanceWindow(BaseModel):
    """Class for structure of maintenance windows.
    """
    start: datetime
    end: datetime
    switches: list[str] = Field(default_factory = list)
    interfaces: list[str] = Field(default_factory = list)
    links: list[str] = Field(default_factory = list)
    id: MaintenanceID = Field(
        default_factory = lambda: MaintenanceID(uuid4().hex)
    )
    description: str = Field(default = '')
    status: Status = Field(default=Status.PENDING)

    @validator('start', 'end', pre = True)
    def convert_time(cls, time):
        if isinstance(time, str):
            time = datetime.strptime(time, TIME_FMT)
        return time

    @validator('start')
    def check_start_in_past(cls, start_time):
        if start_time < datetime.now(pytz.utc):
            raise ValueError('Start in the past not allowed')
        return start_time

    @validator('end')
    def check_end_before_start(cls, end_time, values):
        if 'start' in values and end_time <= values['start']:
            raise ValueError('End before start not allowed')
        return end_time
    
    @root_validator
    def check_items_empty(cls, values):
        if all(map(lambda key: len(values[key]) == 0, ['switches', 'links', 'interfaces'])):
            raise ValueError('At least one item must be provided')
        return values

    def maintenance_event(self, operation, controller: Controller):
        """Create events to start/end a maintenance."""
        if self.switches:
            event = KytosEvent(
                name=f'kytos/maintenance.{operation}_switch',
                content={'switches': self.switches}
            )
            controller.buffers.app.put(event)
        if self.interfaces:
            event = KytosEvent(
                name=f'kytos/maintenance.{operation}_interface',
                content={'unis': self.interfaces}
            )
            controller.buffers.app.put(event)
        if self.links:
            event = KytosEvent(
                name=f'kytos/maintenance.{operation}_link',
                content={'links': self.links}
            )
            controller.buffers.app.put(event)

    def start_mw(self, controller: Controller):
        """Actions taken when a maintenance window starts."""
        self.maintenance_event('start', controller)
        return self.copy(update = {'status': Status.RUNNING})

    def end_mw(self, controller: Controller):
        """Actions taken when a maintenance window finishes."""
        self.maintenance_event('end', controller)
        return self.copy(update = {'status': Status.FINISHED})


@dataclass
class MaintenanceStart:
    """
    Callable used for starting maintenance windows
    """
    maintenance_scheduler: 'Scheduler'
    mw_id: MaintenanceID

    def __call__(self):
        self.maintenance_scheduler.start_maintenance(self.mw_id)


@dataclass
class MaintenanceEnd:
    """
    Callable used for ending maintenance windows
    """
    maintenance_scheduler: 'Scheduler'
    mw_id: MaintenanceID

    def __call__(self):
        self.maintenance_scheduler.end_maintenance(self.mw_id)


@dataclass
class Scheduler:
    """Scheduler for a maintenance window."""
    controller: Controller
    db: 'MaintenanceController'
    scheduler: BaseScheduler

    @classmethod
    def new_scheduler(cls, controller: Controller):
        """
        Creates a new scheduler from the given kytos controller
        """
        scheduler = BackgroundScheduler(timezone=pytz.utc)
        from napps.kytos.maintenance.controllers import MaintenanceController
        db = MaintenanceController()
        db.bootstrap_indexes()
        instance = cls(controller, db, scheduler)
        return instance

    def start(self):
        """
        Begin running the scheduler.
        """
        # Populate the scheduler with all pending tasks
        windows = self.db.get_windows()
        for window in windows:
            self._schedule(window)

        # Start the scheduler
        self.scheduler.start()

    def shutdown(self):
        """
        Stop running the scheduler.
        """
        self.scheduler.remove_all_jobs()
        self.scheduler.shutdown()
        windows = self.db.get_windows()

        # Depopulate the scheduler
        for window in windows:
            self._unschedule(window)

    def start_maintenance(self, mw_id: MaintenanceID):
        """Begins executing the maintenance window
        """
        # Get Maintenance from DB
        window = self.db.get_window(mw_id)

        # Set to Running
        next_win = window.start_mw(self.controller)

        # Update DB
        self.db.update_window(next_win)

    def end_maintenance(self, mw_id: MaintenanceID):
        """Ends execution of the maintenance window
        """
        # Get Maintenance from DB
        window = self.db.get_window(mw_id)

        # Set to Ending
        next_win = window.end_mw(self.controller)

        # Update DB
        self.db.update_window(next_win)

    def end_maintenance_early(self, mw_id: MaintenanceID):
        """Ends execution of the maintenance window early
        """
        # Get Maintenance from DB
        window = self.db.get_window(mw_id)

        # Set to Ending
        next_win = self._unschedule(window)

        # Update DB
        self.db.update_window(next_win)

    def add(self, window: MaintenanceWindow):
        """Add jobs to start and end a maintenance window."""

        # Add window to DB
        self.db.add_window(window)

        # Schedule window
        self._schedule(window)

    def remove(self, mw_id: MaintenanceID):
        """Remove jobs that start and end a maintenance window."""
        # Get Maintenance from DB
        window = self.db.get_window(mw_id)

        # Remove from schedule
        self._unschedule(window)

        # Remove from DB
        self.db.remove_window(mw_id)

    def _schedule(self, window: MaintenanceWindow):
        if window.status is Status.PENDING:
            self.scheduler.add_job(
                MaintenanceStart(self, window.id),
                'date',
                id=f'{window.id}-start',
                run_date=window.start
            )
            self.scheduler.add_job(
                MaintenanceEnd(self, window.id),
                'date',
                id=f'{window.id}-end',
                run_date=window.end
            )
        if window.status is Status.RUNNING:
            window.start_mw(self.controller)
            self.scheduler.add_job(
                MaintenanceEnd(self, window.id),
                'date',
                id=f'{window.id}-end',
                run_date=window.end
            )

    def _unschedule(self, window: MaintenanceWindow):
        """Remove maintenance events from scheduler.
        Does not update DB, due to being
        primarily for shutdown startup cases.
        """
        started = False
        ended = False
        try:
            self.scheduler.remove_job(f'{window.id}-start')
        except JobLookupError:
            started = True
            log.info(f'Job to start {window.id} already removed.')
        try:
            self.scheduler.remove_job(f'{window.id}-end')
        except JobLookupError:
            ended = True
            log.info(f'Job to end {window.id} already removed.')
        if started and not ended:
            window = window.end_mw(self.controller)
        return window

    def get_maintenance(self, mw_id: MaintenanceID):
        """Get a single maintenance by id"""
        return self.db.get_window(mw_id)

    def list_maintenances(self):
        """Returns a list of all maintenances"""
        return self.db.get_windows()


1		"""Models used by the maintenance NApp.
2
3		This module define models for the maintenance window itself and the
4		scheduler.
5		"""
6	1	from dataclasses import dataclass
7	1	from datetime import datetime
8	1	from enum import Enum
9		from typing import NewType
10	1	from uuid import uuid4
11	1
12	1	import pytz
13		from apscheduler.jobstores.base import JobLookupError
14	1	from apscheduler.schedulers.background import BackgroundScheduler
15	1	from apscheduler.schedulers.base import BaseScheduler
16	1	from pydantic import BaseModel, Field, validator, root_validator
17
18	1	from kytos.core import KytosEvent, log
19		from kytos.core.controller import Controller
20
21	1	TIME_FMT = "%Y-%m-%dT%H:%M:%S%z"
22
23		class Status(str, Enum):
24	1	"""Maintenance windows status."""
25	1
26	1	PENDING = 'pending'
27		RUNNING = 'running'
28		FINISHED = 'finished'
29	1
30
31		MaintenanceID = NewType('MaintenanceID', str)
32	1
33
34		class MaintenanceWindow(BaseModel):
35		"""Class for structure of maintenance windows.
36		"""
37		start: datetime
38		end: datetime
39		switches: list[str] = Field(default_factory = list)
40		interfaces: list[str] = Field(default_factory = list)
41		links: list[str] = Field(default_factory = list)
42	1	id: MaintenanceID = Field(
43	1	default_factory = lambda: MaintenanceID(uuid4().hex)
44	1	)
45		description: str = Field(default = '')
46	1	status: Status = Field(default=Status.PENDING)
47	1
48	1	@validator('start', 'end', pre = True)
49	1	def convert_time(cls, time):
50	1	if isinstance(time, str):
51	1	time = datetime.strptime(time, TIME_FMT)
52	1	return time
53	1
54	1	@validator('start')
55	1	def check_start_in_past(cls, start_time):
56		if start_time < datetime.now(pytz.utc):
57	1	raise ValueError('Start in the past not allowed')
58	1	return start_time
59
60	1	@validator('end')
61		def check_end_before_start(cls, end_time, values):
62	1	if 'start' in values and end_time <= values['start']:
63	1	raise ValueError('End before start not allowed')
64		return end_time
65	1
66	1	@root_validator
67	1	def check_items_empty(cls, values):
68	1	if all(map(lambda key: len(values[key]) == 0, ['switches', 'links', 'interfaces'])):
69	1	raise ValueError('At least one item must be provided')
70		return values
71	1
72		def maintenance_event(self, operation, controller: Controller):
73		"""Create events to start/end a maintenance."""
74	1	if self.switches:
75		event = KytosEvent(
76	1	name=f'kytos/maintenance.{operation}_switch',
77		content={'switches': self.switches}
78	1	)
79	1	controller.buffers.app.put(event)
80	1	if self.interfaces:
81	1	event = KytosEvent(
82	1	name=f'kytos/maintenance.{operation}_interface',
83	1	content={'unis': self.interfaces}
84	1	)
85	1	controller.buffers.app.put(event)
86	1	if self.links:
87	1	event = KytosEvent(
88	1	name=f'kytos/maintenance.{operation}_link',
89	1	content={'links': self.links}
90	1	)
91		controller.buffers.app.put(event)
92	1
93	1	def start_mw(self, controller: Controller):
94		"""Actions taken when a maintenance window starts."""
95		self.maintenance_event('start', controller)
96		return self.copy(update = {'status': Status.RUNNING})
97
98		def end_mw(self, controller: Controller):
99		"""Actions taken when a maintenance window finishes."""
100		self.maintenance_event('end', controller)
101		return self.copy(update = {'status': Status.FINISHED})
102
103
104		@dataclass
105		class MaintenanceStart:
106		"""
107		Callable used for starting maintenance windows
108		"""
109		maintenance_scheduler: 'Scheduler'
110	1	mw_id: MaintenanceID
111
112	1	def __call__(self):
113	1	self.maintenance_scheduler.start_maintenance(self.mw_id)
114	1
115	1
116	1	@dataclass
117	1	class MaintenanceEnd:
118	1	"""
119	1	Callable used for ending maintenance windows
120	1	"""
121	1	maintenance_scheduler: 'Scheduler'
122		mw_id: MaintenanceID
123	1
124		def __call__(self):
125	1	self.maintenance_scheduler.end_maintenance(self.mw_id)
126	1
127	1
128	1	@dataclass
129	1	class Scheduler:
130	1	"""Scheduler for a maintenance window."""
131	1	controller: Controller
132		db: 'MaintenanceController'
133		scheduler: BaseScheduler
134	1
135	1	@classmethod
136		def new_scheduler(cls, controller: Controller):
137		"""
138		Creates a new scheduler from the given kytos controller
139		"""
140	1	scheduler = BackgroundScheduler(timezone=pytz.utc)
141	1	from napps.kytos.maintenance.controllers import MaintenanceController
142		db = MaintenanceController()
143		db.bootstrap_indexes()
144		instance = cls(controller, db, scheduler)
145		return instance
146
147		def start(self):
148		"""
149		Begin running the scheduler.
150	1	"""
151	1	# Populate the scheduler with all pending tasks
152		windows = self.db.get_windows()
153		for window in windows:
154		self._schedule(window)
155
156		# Start the scheduler
157		self.scheduler.start()
158
159		def shutdown(self):
160		"""
161		Stop running the scheduler.
162		"""
163		self.scheduler.remove_all_jobs()
164		self.scheduler.shutdown()
165		windows = self.db.get_windows()
166
167	1	# Depopulate the scheduler
168	1	for window in windows:
169		self._unschedule(window)
170	1
171	1	def start_maintenance(self, mw_id: MaintenanceID):
172		"""Begins executing the maintenance window
173	1	"""
174		# Get Maintenance from DB
175	1	window = self.db.get_window(mw_id)
176	1
177	1	# Set to Running
178	1	next_win = window.start_mw(self.controller)
179	1
180	1	# Update DB
181	1	self.db.update_window(next_win)
182
183	1	def end_maintenance(self, mw_id: MaintenanceID):
184	1	"""Ends execution of the maintenance window
185	1	"""
186		# Get Maintenance from DB
187	1	window = self.db.get_window(mw_id)
188	1
189	1	# Set to Ending
190		next_win = window.end_mw(self.controller)
191	1
192		# Update DB
193	1	self.db.update_window(next_win)
194
195	1	def end_maintenance_early(self, mw_id: MaintenanceID):
196	1	"""Ends execution of the maintenance window early
197		"""
198	1	# Get Maintenance from DB
199		window = self.db.get_window(mw_id)
200	1
201	1	# Set to Ending
202		next_win = self._unschedule(window)
203
204	1	# Update DB
205		self.db.update_window(next_win)
206
207	1	def add(self, window: MaintenanceWindow):
208		"""Add jobs to start and end a maintenance window."""
209	1
210	1	# Add window to DB
211		self.db.add_window(window)
212	1
213		# Schedule window
214	1	self._schedule(window)
215
216		def remove(self, mw_id: MaintenanceID):
217	1	"""Remove jobs that start and end a maintenance window."""
218		# Get Maintenance from DB
219		window = self.db.get_window(mw_id)
220
221	1	# Remove from schedule
222		self._unschedule(window)
223	1
224	1	# Remove from DB
225	1	self.db.remove_window(mw_id)
226	1
227	1	def _schedule(self, window: MaintenanceWindow):
228	1	if window.status is Status.PENDING:
229	1	self.scheduler.add_job(
230	1	MaintenanceStart(self, window.id),
231		'date',
232		id=f'{window.id}-start',
233		run_date=window.start
234		)
235		self.scheduler.add_job(
236		MaintenanceEnd(self, window.id),
237		'date',
238		id=f'{window.id}-end',
239		run_date=window.end
240		)
241		if window.status is Status.RUNNING:
242		window.start_mw(self.controller)
243		self.scheduler.add_job(
244		MaintenanceEnd(self, window.id),
245		'date',
246		id=f'{window.id}-end',
247		run_date=window.end
248		)
249
250		def _unschedule(self, window: MaintenanceWindow):
251		"""Remove maintenance events from scheduler.
252		Does not update DB, due to being
253		primarily for shutdown startup cases.
254		"""
255		started = False
256		ended = False
257		try:
258		self.scheduler.remove_job(f'{window.id}-start')
259		except JobLookupError:
260		started = True
261		log.info(f'Job to start {window.id} already removed.')
262		try:
263		self.scheduler.remove_job(f'{window.id}-end')
264		except JobLookupError:
265		ended = True
266		log.info(f'Job to end {window.id} already removed.')
267		if started and not ended:
268		window = window.end_mw(self.controller)
269		return window
270
271		def get_maintenance(self, mw_id: MaintenanceID):
272		"""Get a single maintenance by id"""
273		return self.db.get_window(mw_id)
274
275		def list_maintenances(self):
276		"""Returns a list of all maintenances"""
277		return self.db.get_windows()
278

kytos-ng / maintenance

Pull Request — master (#64)

build.models.MaintenanceWindow.maintenance_event() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like