build.models.Scheduler.end_maintenance() - 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-16 19:42 UTC

build.models.Scheduler.end_maintenance() A

↳ Parent: build.models

Complexity

Conditions

Size

Total Lines	8
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	6
CRAP Score	1

Importance

Changes

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

"""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 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: 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)


@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 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):
        """Add jobs to start and end a maintenance window."""

        # 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):
        if window.status is Status.PENDING:
            self.scheduler.add_job(
                MaintenanceStart(self, window.id),
                'date',
                id=f'{window.id}-start',
                run_date=window.start
            )
        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 _reschedule(self, window: MaintenanceWindow):
        try:
            self.scheduler.modify_job(
                f'{window.id}-start',
                run_date = window.start,
            )
        except JobLookupError:
            log.info(f'Could not reschedule start, already started')
        try:
            self.scheduler.modify_job(
                f'{window.id}-end',
                run_date = window.end,
            )
        except JobLookupError:
            log.info(f'Could not reschedule end, already ended')

    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):
        """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, 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 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	inserted_at: Optional[datetime] = Field(default = None)
48	1	updated_at: Optional[datetime] = Field(default = None)
49	1
50	1	@validator('start', 'end', pre = True)
51	1	def convert_time(cls, time):
52	1	if isinstance(time, str):
53	1	time = datetime.strptime(time, TIME_FMT)
54	1	return time
55	1
56		@validator('start')
57	1	def check_start_in_past(cls, start_time):
58	1	if start_time < datetime.now(pytz.utc):
59		raise ValueError('Start in the past not allowed')
60	1	return start_time
61
62	1	@validator('end')
63	1	def check_end_before_start(cls, end_time, values):
64		if 'start' in values and end_time <= values['start']:
65	1	raise ValueError('End before start not allowed')
66	1	return end_time
67	1
68	1	@root_validator
69	1	def check_items_empty(cls, values):
70		if all(map(lambda key: len(values[key]) == 0, ['switches', 'links', 'interfaces'])):
71	1	raise ValueError('At least one item must be provided')
72		return values
73
74	1	def maintenance_event(self, operation, controller: Controller):
75		"""Create events to start/end a maintenance."""
76	1	if self.switches:
77		event = KytosEvent(
78	1	name=f'kytos/maintenance.{operation}_switch',
79	1	content={'switches': self.switches}
80	1	)
81	1	controller.buffers.app.put(event)
82	1	if self.interfaces:
83	1	event = KytosEvent(
84	1	name=f'kytos/maintenance.{operation}_interface',
85	1	content={'unis': self.interfaces}
86	1	)
87	1	controller.buffers.app.put(event)
88	1	if self.links:
89	1	event = KytosEvent(
90	1	name=f'kytos/maintenance.{operation}_link',
91		content={'links': self.links}
92	1	)
93	1	controller.buffers.app.put(event)
94
95		def start_mw(self, controller: Controller):
96		"""Actions taken when a maintenance window starts."""
97		self.maintenance_event('start', controller)
98
99		def end_mw(self, controller: Controller):
100		"""Actions taken when a maintenance window finishes."""
101		self.maintenance_event('end', controller)
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 and Update
175	1	window = self.db.start_window(mw_id)
176	1
177	1	# Activate Running
178	1	window.start_mw(self.controller)
179	1
180	1	# Schedule next task
181	1	self._schedule(window)
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.end_window(mw_id)
188	1
189	1	# Set to Ending
190		window.end_mw(self.controller)
191	1
192		def end_maintenance_early(self, mw_id: MaintenanceID):
193	1	"""Ends execution of the maintenance window early
194		"""
195	1	# Get Maintenance from DB
196	1	window = self.db.end_window(mw_id)
197
198	1	# Unschedule tasks
199		self._unschedule(window)
200	1
201	1	def add(self, window: MaintenanceWindow):
202		"""Add jobs to start and end a maintenance window."""
203
204	1	# Add window to DB
205		self.db.insert_window(window)
206
207	1	# Schedule next task
208		self._schedule(window)
209	1
210	1	def update(self, window: MaintenanceWindow):
211		"""Update an existing Maintenance Window."""
212	1
213		# Update window
214	1	self.db.update_window(window)
215
216		# Reschedule any pending tasks
217	1	self._reschedule(window)
218
219		def remove(self, mw_id: MaintenanceID):
220		"""Remove jobs that start and end a maintenance window."""
221	1	# Get Maintenance from DB
222		window = self.db.get_window(mw_id)
223	1
224	1	# Remove from schedule
225	1	self._unschedule(window)
226	1
227	1	# Remove from DB
228	1	self.db.remove_window(mw_id)
229	1
230	1	def _schedule(self, window: MaintenanceWindow):
231		if window.status is Status.PENDING:
232		self.scheduler.add_job(
233		MaintenanceStart(self, window.id),
234		'date',
235		id=f'{window.id}-start',
236		run_date=window.start
237		)
238		if window.status is Status.RUNNING:
239		window.start_mw(self.controller)
240		self.scheduler.add_job(
241		MaintenanceEnd(self, window.id),
242		'date',
243		id=f'{window.id}-end',
244		run_date=window.end
245		)
246
247		def _reschedule(self, window: MaintenanceWindow):
248		try:
249		self.scheduler.modify_job(
250		f'{window.id}-start',
251		run_date = window.start,
252		)
253		except JobLookupError:
254		log.info(f'Could not reschedule start, already started')
255		try:
256		self.scheduler.modify_job(
257		f'{window.id}-end',
258		run_date = window.end,
259		)
260		except JobLookupError:
261		log.info(f'Could not reschedule end, already ended')
262
263		def _unschedule(self, window: MaintenanceWindow):
264		"""Remove maintenance events from scheduler.
265		Does not update DB, due to being
266		primarily for shutdown startup cases.
267		"""
268		started = False
269		ended = False
270		try:
271		self.scheduler.remove_job(f'{window.id}-start')
272		except JobLookupError:
273		started = True
274		log.info(f'Job to start {window.id} already removed.')
275		try:
276		self.scheduler.remove_job(f'{window.id}-end')
277		except JobLookupError:
278		ended = True
279		log.info(f'Job to end {window.id} already removed.')
280		if started and not ended:
281		window.end_mw(self.controller)
282
283		def get_maintenance(self, mw_id: MaintenanceID):
284		"""Get a single maintenance by id"""
285		return self.db.get_window(mw_id)
286
287		def list_maintenances(self):
288		"""Returns a list of all maintenances"""
289		return self.db.get_windows()
290

kytos-ng / maintenance

Pull Request — master (#64)

build.models.Scheduler.end_maintenance() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like