build.models.MaintenanceEnd.__call__() - 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 2023-01-10 15:13 UTC

build.models.MaintenanceEnd.call() A

↳ Parent: build.models

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

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

"""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, Optional
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 apscheduler.triggers.date import DateTrigger
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)
    inserted_at: Optional[datetime] = Field(default = None)
    updated_at: Optional[datetime] = Field(default = None)

    @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: key not in values or 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)

    def end_mw(self, controller: Controller):
        """Actions taken when a maintenance window finishes."""
        self.maintenance_event('end', controller)

    class Config:
        json_encoders = {
            datetime: lambda v: v.strftime(TIME_FMT),
        }

class MaintenanceWindows(BaseModel):
    __root__: list[MaintenanceWindow]

    def __iter__(self):
        return iter(self.__root__)

    def __getitem__(self, item):
        return self.__root__[item]

    def __len__(self):
        return len(self.__root__)

    class Config:
        json_encoders = {
            datetime: lambda v: v.strftime(TIME_FMT),
        }

@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)

class OverlapError(Exception):
    """
    Exception for when a Maintenance Windows execution
    period overlaps with one or more windows.
    """
    new_window: MaintenanceWindow
    interferring: MaintenanceWindows

    def __init__(self, new_window: MaintenanceWindow, interferring: MaintenanceWindows):
        self.new_window = new_window
        self.interferring = interferring

    def __str__(self):
        return f"Maintenance Window '{self.new_window.id}'<{self.new_window.start} to {self.new_window.end}> " +\
            "interfers with the following windows: " +\
            '[' +\
            ', '.join([f"'{window.id}'<{window.start} to {window.end}>" for window in self.interferring]) +\
            ']'

@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.
        """
        self.db.prepare_start()

        # 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.
        """
        windows = self.db.get_windows()

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

        self.scheduler.remove_all_jobs()
        self.scheduler.shutdown()

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

        # Activate Running
        window.start_mw(self.controller)

        # Schedule next task
        self._schedule(window)

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

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

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

        # Unschedule tasks
        self._unschedule(window)

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

        if force is False:
            overlapping_windows = self.db.check_overlap(window)
            if overlapping_windows:
                raise OverlapError(window, overlapping_windows)

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

        # Schedule next task
        self._schedule(window)

    def update(self, window: MaintenanceWindow):
        """Update an existing Maintenance Window."""

        # Update window
        self.db.update_window(window)

        # Reschedule any pending tasks
        self._reschedule(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):
        log.info(f'Scheduling "{window.id}"')
        if window.status == Status.PENDING:
            self.scheduler.add_job(
                MaintenanceStart(self, window.id),
                'date',
                id=f'{window.id}-start',
                run_date=window.start
            )
            log.info(f'Scheduled "{window.id}" start at {window.start}')
        if window.status == 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
            )
            log.info(f'Scheduled "{window.id}" end at {window.end}')

    def _reschedule(self, window: MaintenanceWindow):
        log.info(f'Rescheduling "{window.id}"')
        try:
            self.scheduler.remove_job(
                f'{window.id}-start',
            )
            self.scheduler.add_job(
                MaintenanceStart(self, window.id),
                'date',
                id=f'{window.id}-start',
                run_date=window.start
            )
            log.info(f'Rescheduled "{window.id}" start to {window.start}')
        except JobLookupError:
            log.info(f'Could not reschedule "{window.id}" start, no start job')
        try:
            self.scheduler.remove_job(
                f'{window.id}-end',
            )
            self.scheduler.add_job(
                MaintenanceEnd(self, window.id),
                'date',
                id=f'{window.id}-end',
                run_date=window.end
            )
            log.info(f'Rescheduled "{window.id}" end to {window.end}')
        except JobLookupError:
            log.info(f'Could not reschedule "{window.id}" end, no end job')

    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.end_mw(self.controller)

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

    def list_maintenances(self) -> MaintenanceWindows:
        """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, Optional
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 apscheduler.triggers.date import DateTrigger
17		from pydantic import BaseModel, Field, validator, root_validator
18	1
19		from kytos.core import KytosEvent, log
20		from kytos.core.controller import Controller
21	1
22		TIME_FMT = "%Y-%m-%dT%H:%M:%S%z"
23
24	1	class Status(str, Enum):
25	1	"""Maintenance windows status."""
26	1
27		PENDING = 'pending'
28		RUNNING = 'running'
29	1	FINISHED = 'finished'
30
31
32	1	MaintenanceID = NewType('MaintenanceID', str)
33
34
35		class MaintenanceWindow(BaseModel):
36		"""Class for structure of maintenance windows.
37		"""
38		start: datetime
39		end: datetime
40		switches: list[str] = Field(default_factory = list)
41		interfaces: list[str] = Field(default_factory = list)
42	1	links: list[str] = Field(default_factory = list)
43	1	id: MaintenanceID = Field(
44	1	default_factory = lambda: MaintenanceID(uuid4().hex)
45		)
46	1	description: str = Field(default = '')
47	1	status: Status = Field(default=Status.PENDING)
48	1	inserted_at: Optional[datetime] = Field(default = None)
49	1	updated_at: Optional[datetime] = Field(default = None)
50	1
51	1	@validator('start', 'end', pre = True)
52	1	def convert_time(cls, time):
53	1	if isinstance(time, str):
54	1	time = datetime.strptime(time, TIME_FMT)
55	1	return time
56
57	1	@validator('start')
58	1	def check_start_in_past(cls, start_time):
59		if start_time < datetime.now(pytz.utc):
60	1	raise ValueError('Start in the past not allowed')
61		return start_time
62	1
63	1	@validator('end')
64		def check_end_before_start(cls, end_time, values):
65	1	if 'start' in values and end_time <= values['start']:
66	1	raise ValueError('End before start not allowed')
67	1	return end_time
68	1
69	1	@root_validator
70		def check_items_empty(cls, values):
71	1	if all(map(lambda key: key not in values or len(values[key]) == 0, ['switches', 'links', 'interfaces'])):
72		raise ValueError('At least one item must be provided')
73		return values
74	1
75		def maintenance_event(self, operation, controller: Controller):
76	1	"""Create events to start/end a maintenance."""
77		if self.switches:
78	1	event = KytosEvent(
79	1	name=f'kytos/maintenance.{operation}_switch',
80	1	content={'switches': self.switches}
81	1	)
82	1	controller.buffers.app.put(event)
83	1	if self.interfaces:
84	1	event = KytosEvent(
85	1	name=f'kytos/maintenance.{operation}_interface',
86	1	content={'unis': self.interfaces}
87	1	)
88	1	controller.buffers.app.put(event)
89	1	if self.links:
90	1	event = KytosEvent(
91		name=f'kytos/maintenance.{operation}_link',
92	1	content={'links': self.links}
93	1	)
94		controller.buffers.app.put(event)
95
96		def start_mw(self, controller: Controller):
97		"""Actions taken when a maintenance window starts."""
98		self.maintenance_event('start', controller)
99
100		def end_mw(self, controller: Controller):
101		"""Actions taken when a maintenance window finishes."""
102		self.maintenance_event('end', controller)
103
104		class Config:
105		json_encoders = {
106		datetime: lambda v: v.strftime(TIME_FMT),
107		}
108
109		class MaintenanceWindows(BaseModel):
110	1	__root__: list[MaintenanceWindow]
111
112	1	def __iter__(self):
113	1	return iter(self.__root__)
114	1
115	1	def __getitem__(self, item):
116	1	return self.__root__[item]
117	1
118	1	def __len__(self):
119	1	return len(self.__root__)
120	1
121	1	class Config:
122		json_encoders = {
123	1	datetime: lambda v: v.strftime(TIME_FMT),
124		}
125	1
126	1	@dataclass
127	1	class MaintenanceStart:
128	1	"""
129	1	Callable used for starting maintenance windows
130	1	"""
131	1	maintenance_scheduler: 'Scheduler'
132		mw_id: MaintenanceID
133
134	1	def __call__(self):
135	1	self.maintenance_scheduler.start_maintenance(self.mw_id)
136
137
138		@dataclass
139		class MaintenanceEnd:
140	1	"""
141	1	Callable used for ending maintenance windows
142		"""
143		maintenance_scheduler: 'Scheduler'
144		mw_id: MaintenanceID
145
146		def __call__(self):
147		self.maintenance_scheduler.end_maintenance(self.mw_id)
148
149		class OverlapError(Exception):
150	1	"""
151	1	Exception for when a Maintenance Windows execution
152		period overlaps with one or more windows.
153		"""
154		new_window: MaintenanceWindow
155		interferring: MaintenanceWindows
156
157		def __init__(self, new_window: MaintenanceWindow, interferring: MaintenanceWindows):
158		self.new_window = new_window
159		self.interferring = interferring
160
161		def __str__(self):
162		return f"Maintenance Window '{self.new_window.id}'<{self.new_window.start} to {self.new_window.end}> " +\
163		"interfers with the following windows: " +\
164		'[' +\
165		', '.join([f"'{window.id}'<{window.start} to {window.end}>" for window in self.interferring]) +\
166		']'
167	1
168	1	@dataclass
169		class Scheduler:
170	1	"""Scheduler for a maintenance window."""
171	1	controller: Controller
172		db: 'MaintenanceController'
173	1	scheduler: BaseScheduler
174
175	1	@classmethod
176	1	def new_scheduler(cls, controller: Controller):
177	1	"""
178	1	Creates a new scheduler from the given kytos controller
179	1	"""
180	1	scheduler = BackgroundScheduler(timezone=pytz.utc)
181	1	from napps.kytos.maintenance.controllers import MaintenanceController
182		db = MaintenanceController()
183	1	db.bootstrap_indexes()
184	1	instance = cls(controller, db, scheduler)
185	1	return instance
186
187	1	def start(self):
188	1	"""
189	1	Begin running the scheduler.
190		"""
191	1	self.db.prepare_start()
192
193	1	# Populate the scheduler with all pending tasks
194		windows = self.db.get_windows()
195	1	for window in windows:
196	1	self._schedule(window)
197
198	1	# Start the scheduler
199		self.scheduler.start()
200	1
201	1	def shutdown(self):
202		"""
203		Stop running the scheduler.
204	1	"""
205		windows = self.db.get_windows()
206
207	1	# Depopulate the scheduler
208		for window in windows:
209	1	self._unschedule(window)
210	1
211		self.scheduler.remove_all_jobs()
212	1	self.scheduler.shutdown()
213
214	1	def start_maintenance(self, mw_id: MaintenanceID):
215		"""Begins executing the maintenance window
216		"""
217	1	# Get Maintenance from DB and Update
218		window = self.db.start_window(mw_id)
219
220		# Activate Running
221	1	window.start_mw(self.controller)
222
223	1	# Schedule next task
224	1	self._schedule(window)
225	1
226	1	def end_maintenance(self, mw_id: MaintenanceID):
227	1	"""Ends execution of the maintenance window
228	1	"""
229	1	# Get Maintenance from DB
230	1	window = self.db.end_window(mw_id)
231
232		# Set to Ending
233		window.end_mw(self.controller)
234
235		def end_maintenance_early(self, mw_id: MaintenanceID):
236		"""Ends execution of the maintenance window early
237		"""
238		# Get Maintenance from DB
239		window = self.db.end_window(mw_id)
240
241		# Unschedule tasks
242		self._unschedule(window)
243
244		def add(self, window: MaintenanceWindow, force = False):
245		"""Add jobs to start and end a maintenance window."""
246
247		if force is False:
248		overlapping_windows = self.db.check_overlap(window)
249		if overlapping_windows:
250		raise OverlapError(window, overlapping_windows)
251
252		# Add window to DB
253		self.db.insert_window(window)
254
255		# Schedule next task
256		self._schedule(window)
257
258		def update(self, window: MaintenanceWindow):
259		"""Update an existing Maintenance Window."""
260
261		# Update window
262		self.db.update_window(window)
263
264		# Reschedule any pending tasks
265		self._reschedule(window)
266
267		def remove(self, mw_id: MaintenanceID):
268		"""Remove jobs that start and end a maintenance window."""
269		# Get Maintenance from DB
270		window = self.db.get_window(mw_id)
271
272		# Remove from schedule
273		self._unschedule(window)
274
275		# Remove from DB
276		self.db.remove_window(mw_id)
277
278		def _schedule(self, window: MaintenanceWindow):
279		log.info(f'Scheduling "{window.id}"')
280		if window.status == Status.PENDING:
281		self.scheduler.add_job(
282		MaintenanceStart(self, window.id),
283		'date',
284		id=f'{window.id}-start',
285		run_date=window.start
286		)
287		log.info(f'Scheduled "{window.id}" start at {window.start}')
288		if window.status == Status.RUNNING:
289		window.start_mw(self.controller)
290		self.scheduler.add_job(
291		MaintenanceEnd(self, window.id),
292		'date',
293		id=f'{window.id}-end',
294		run_date=window.end
295		)
296		log.info(f'Scheduled "{window.id}" end at {window.end}')
297
298		def _reschedule(self, window: MaintenanceWindow):
299		log.info(f'Rescheduling "{window.id}"')
300		try:
301		self.scheduler.remove_job(
302		f'{window.id}-start',
303		)
304		self.scheduler.add_job(
305		MaintenanceStart(self, window.id),
306		'date',
307		id=f'{window.id}-start',
308		run_date=window.start
309		)
310		log.info(f'Rescheduled "{window.id}" start to {window.start}')
311		except JobLookupError:
312		log.info(f'Could not reschedule "{window.id}" start, no start job')
313		try:
314		self.scheduler.remove_job(
315		f'{window.id}-end',
316		)
317		self.scheduler.add_job(
318		MaintenanceEnd(self, window.id),
319		'date',
320		id=f'{window.id}-end',
321		run_date=window.end
322		)
323		log.info(f'Rescheduled "{window.id}" end to {window.end}')
324		except JobLookupError:
325		log.info(f'Could not reschedule "{window.id}" end, no end job')
326
327		def _unschedule(self, window: MaintenanceWindow):
328		"""Remove maintenance events from scheduler.
329		Does not update DB, due to being
330		primarily for shutdown startup cases.
331		"""
332		started = False
333		ended = False
334		try:
335		self.scheduler.remove_job(f'{window.id}-start')
336		except JobLookupError:
337		started = True
338		log.info(f'Job to start "{window.id}" already removed.')
339		try:
340		self.scheduler.remove_job(f'{window.id}-end')
341		except JobLookupError:
342		ended = True
343		log.info(f'Job to end "{window.id}" already removed.')
344		if started and not ended:
345		window.end_mw(self.controller)
346
347		def get_maintenance(self, mw_id: MaintenanceID) -> MaintenanceWindow:
348		"""Get a single maintenance by id"""
349		return self.db.get_window(mw_id)
350
351		def list_maintenances(self) -> MaintenanceWindows:
352		"""Returns a list of all maintenances"""
353		return self.db.get_windows()
354

kytos-ng / maintenance

Pull Request — master (#64)

build.models.MaintenanceEnd.__call__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

build.models.MaintenanceEnd.call() A