Skip to content

Circuit Maintenance Jobs

nautobot_circuit_maintenance.jobs

Circuit Maintenance plugin jobs.

Site searching Job definition.

FindSitesWithMaintenanceOverlap

Bases: Job

Nautobot Job definition for finding sites without redundant circuit for impactful maintenance.

Current iteration of this job assumes that many sites will be only dual carrier connected. The searches are going to search for a single overlap in this first iteration.

Future iterations may include the ability to search for multiple circuit overlaps that would allow for just a single circuit to be available.

Source code in nautobot_circuit_maintenance/jobs/site_search.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class FindSitesWithMaintenanceOverlap(Job):
    """Nautobot Job definition for finding sites without redundant circuit for impactful maintenance.

    Current iteration of this job assumes that many sites will be only dual carrier connected. The searches are going
    to search for a single overlap in this first iteration.

    Future iterations may include the ability to search for multiple circuit overlaps that would allow for just a single
    circuit to be available.
    """

    job_debug = BooleanVar(description="Enable for more verbose debug logging")

    class Meta:
        """Meta definition for the Job."""

        name = "Find Sites With Circuit Maintenance Overlap"
        description = "Search for sites with overlapping circuit maintenances, 1 or more circuit impacts, assuming only TWO circuits per site."

    def run(self, data=None, commit=None):
        """Executes the Job.

        Args:
            data (dict, optional): Data from Nautobot from. Defaults to None.
            commit (bool, optional): Commit boolean. Defaults to None.
        """
        # Query for all of the circuits maintenances that are on going in the future
        today = date.today()
        circuit_maintenances = CircuitMaintenance.objects.filter(start_time__gte=today).order_by("start_time")

        # Build a circuit mapper
        circuit_maintenance_mapper = build_sites_to_maintenance_mapper(circuit_maintenances)

        # Loop over each of the circuit maintenance records
        # pylint: disable=too-many-nested-blocks
        processed_maintenance_list = []  # List of the processed maintenance records
        for circuit_maint in circuit_maintenances:
            # Get the set of sites
            site_set: set = set()
            for circuit in circuit_maint.circuits:
                site_set.update(get_sites_from_circuit(circuit))

            processed_maintenance_list.append(circuit_maint)
            overlapping_maintenance = False  # Flag for overlapping maintenance present
            # Check to see if there are any circuit maintenances that are overlapping at some moment in time
            for site in site_set:
                for other_circuit_maint in circuit_maintenance_mapper[site.name]:
                    # Report failures for any time where a circuit will take an outage
                    # Check to see if the other circuit maintenance has already had overlap checked.
                    # If this is the same circuit maintenance record, that will also show up in the processed list
                    # since the list is being added to before the loop.
                    if other_circuit_maint in processed_maintenance_list:
                        continue
                    if check_for_overlap(circuit_maint, other_circuit_maint):
                        overlapping_maintenance = True
                        if PLUGIN_SETTINGS.get("overlap_job_exclude_no_impact"):
                            for circuit_impact in circuit_maint.circuitimpact_set:
                                if circuit_impact.impact != "NO-IMPACT":
                                    break
                            else:
                                continue

                        self.log_warning(
                            obj=site,
                            message=f"There is an overlapping maintenance for site: {site.name} on {circuit_maint.start_time}. Other maintenances: {other_circuit_maint}|{circuit_maint}",
                        )

            # Log success for when there is not an overlapping maintenance at a site
            if overlapping_maintenance is False and data["job_debug"]:
                self.log_info(obj=circuit_maint, message="Checked maintenance for overlap, no overlap was found.")

        self.log_success(
            obj=None,
            message=f"Successfully checked through {circuit_maintenances.count()} maintenance notification{'s'[:circuit_maintenances.count()^1]}.",
        )
Meta

Meta definition for the Job.

Source code in nautobot_circuit_maintenance/jobs/site_search.py 
101
102
103
104
105
class Meta:
    """Meta definition for the Job."""

    name = "Find Sites With Circuit Maintenance Overlap"
    description = "Search for sites with overlapping circuit maintenances, 1 or more circuit impacts, assuming only TWO circuits per site."
run(data=None, commit=None)

Executes the Job.

Parameters:

Name Type Description Default
data dict

Data from Nautobot from. Defaults to None.

 None
commit bool

Commit boolean. Defaults to None.

 None
Source code in nautobot_circuit_maintenance/jobs/site_search.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def run(self, data=None, commit=None):
    """Executes the Job.

    Args:
        data (dict, optional): Data from Nautobot from. Defaults to None.
        commit (bool, optional): Commit boolean. Defaults to None.
    """
    # Query for all of the circuits maintenances that are on going in the future
    today = date.today()
    circuit_maintenances = CircuitMaintenance.objects.filter(start_time__gte=today).order_by("start_time")

    # Build a circuit mapper
    circuit_maintenance_mapper = build_sites_to_maintenance_mapper(circuit_maintenances)

    # Loop over each of the circuit maintenance records
    # pylint: disable=too-many-nested-blocks
    processed_maintenance_list = []  # List of the processed maintenance records
    for circuit_maint in circuit_maintenances:
        # Get the set of sites
        site_set: set = set()
        for circuit in circuit_maint.circuits:
            site_set.update(get_sites_from_circuit(circuit))

        processed_maintenance_list.append(circuit_maint)
        overlapping_maintenance = False  # Flag for overlapping maintenance present
        # Check to see if there are any circuit maintenances that are overlapping at some moment in time
        for site in site_set:
            for other_circuit_maint in circuit_maintenance_mapper[site.name]:
                # Report failures for any time where a circuit will take an outage
                # Check to see if the other circuit maintenance has already had overlap checked.
                # If this is the same circuit maintenance record, that will also show up in the processed list
                # since the list is being added to before the loop.
                if other_circuit_maint in processed_maintenance_list:
                    continue
                if check_for_overlap(circuit_maint, other_circuit_maint):
                    overlapping_maintenance = True
                    if PLUGIN_SETTINGS.get("overlap_job_exclude_no_impact"):
                        for circuit_impact in circuit_maint.circuitimpact_set:
                            if circuit_impact.impact != "NO-IMPACT":
                                break
                        else:
                            continue

                    self.log_warning(
                        obj=site,
                        message=f"There is an overlapping maintenance for site: {site.name} on {circuit_maint.start_time}. Other maintenances: {other_circuit_maint}|{circuit_maint}",
                    )

        # Log success for when there is not an overlapping maintenance at a site
        if overlapping_maintenance is False and data["job_debug"]:
            self.log_info(obj=circuit_maint, message="Checked maintenance for overlap, no overlap was found.")

    self.log_success(
        obj=None,
        message=f"Successfully checked through {circuit_maintenances.count()} maintenance notification{'s'[:circuit_maintenances.count()^1]}.",
    )

build_sites_to_maintenance_mapper(maintenance_queryset)

Build a site to circuit maintenance mapper so the data can be quickly accessed of what possible overlaps.

Leverages defaultdict to provide the default value of an empty set to each key that will be added. Then adds each particular circuit maintenance object to the dictionary to be used as a map of maintenances going on at the site. Build a data dictionary that maps sites to circuit maintenances:

data_dict = { "msp": { CM1, CM2 }, "nyc": { CM3, CM4 } }

Parameters:

Name Type Description Default
maintenance_queryset Queryset

Queryset of all Circuit Maintenance objects

 required

Returns:

Name Type Description
dict dict

Dictionary of a set of maintenance records
Source code in nautobot_circuit_maintenance/jobs/site_search.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def build_sites_to_maintenance_mapper(maintenance_queryset) -> dict:
    """Build a site to circuit maintenance mapper so the data can be quickly accessed of what possible overlaps.

    Leverages defaultdict to provide the default value of an empty set to each key that will be added. Then adds each
    particular circuit maintenance object to the dictionary to be used as a map of maintenances going on at the site.
    Build a data dictionary that maps sites to circuit maintenances:

    data_dict = {
        "msp": {
            CM1,
            CM2
        },
        "nyc": {
            CM3,
            CM4
        }
    }

    Args:
        maintenance_queryset (Queryset): Queryset of all Circuit Maintenance objects

    Returns:
        dict: Dictionary of a set of maintenance records
    """
    return_dictionary = collections.defaultdict(set)
    for record in maintenance_queryset:
        # Get the site overall
        # Start with looping over the circuits that are defined
        for circuit in record.circuits:
            # Check both termination A and Z for values of None, add that record to the set for each site
            for term in [circuit.termination_a, circuit.termination_z]:
                if term is not None and term.site is not None:
                    return_dictionary[term.site.name].add(record)

    return dict(return_dictionary)

check_for_overlap(record1, record2)

Checks for the overlap of two circuit maintenance records.

Parameters:

Name Type Description Default
record1 CircuitMaintenance

First maintenance record

 required
record2 CircuitMaintenance

Second maintenance record

 required

Returns:

Name Type Description
bool bool

True if there is overlap, otherwise False.
Source code in nautobot_circuit_maintenance/jobs/site_search.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def check_for_overlap(record1: CircuitMaintenance, record2: CircuitMaintenance) -> bool:
    """Checks for the overlap of two circuit maintenance records.

    Args:
        record1 (CircuitMaintenance): First maintenance record
        record2 (CircuitMaintenance): Second maintenance record

    Returns:
        bool: True if there is overlap, otherwise False.
    """
    if record1.end_time < record2.start_time or record2.end_time < record1.start_time:
        return False

    return True

get_sites_from_circuit(circuit)

Get a list of sites from a circuit object in Nautobot.

Parameters:

Name Type Description Default
circuit Circuit

Nautobot Circuit Object to find the sites

 required

Returns:

Name Type Description
list set

List of Nautobot Site objects
Source code in nautobot_circuit_maintenance/jobs/site_search.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def get_sites_from_circuit(circuit: Circuit) -> set:
    """Get a list of sites from a circuit object in Nautobot.

    Args:
        circuit (Circuit): Nautobot Circuit Object to find the sites


    Returns:
        list: List of Nautobot Site objects
    """
    site_set: set = set()
    for term in [circuit.termination_a, circuit.termination_z]:
        if term is not None and getattr(term, "provider_network") is None:
            site_set.add(term.site)

    return site_set