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-08 15:04 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	16
CRAP Score	4

Importance

Changes

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

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

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


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)

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

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