Skip to content

Validate iptrunk

Router validation workflow. Used in a nightly schedule.

validate_router_config(subscription)

Run an Ansible playbook that validates the configuration that is present on an active IP trunk.

Source code in gso/workflows/iptrunk/validate_iptrunk.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@step("Validate IP trunk configuration")
def validate_router_config(subscription: Iptrunk) -> LSOState:
    """Run an Ansible playbook that validates the configuration that is present on an active IP trunk."""
    extra_vars = {"wfo_trunk_json": json.loads(json_dumps(subscription)), "verb": "validate"}

    return {
        "playbook_name": "gap_ansible/playbooks/base_config.yaml",
        "inventory": {
            "all": {
                "hosts": {
                    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None,
                    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn: None,
                }
            }
        },
        "extra_vars": extra_vars,
    }

verify_ipam_records(subscription)

Validate the IPAM resources for the LAG interfaces.

Raises:

Type Description
ProcessFailureError

if IPAM is configured incorrectly.
Source code in gso/workflows/iptrunk/validate_iptrunk.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 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
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@step("Verify IPAM resources for LAG interfaces")
def verify_ipam_records(subscription: Iptrunk) -> None:
    """Validate the IPAM resources for the LAG interfaces.

    Raises:
        ProcessFailureError: if IPAM is configured incorrectly.
    """
    ipam_errors = []
    ipam_v4_network = infoblox.find_network_by_cidr(subscription.iptrunk.iptrunk_ipv4_network)
    ipam_v6_network = infoblox.find_network_by_cidr(subscription.iptrunk.iptrunk_ipv6_network)

    if not ipam_v4_network or not ipam_v6_network:
        ipam_errors.append(
            "Missing IP trunk IPAM records, found the following instead.\n"
            f"IPv4 expected '{subscription.iptrunk.iptrunk_ipv4_network}', actual: '{ipam_v4_network}'\n"
            f"IPv6 expected '{subscription.iptrunk.iptrunk_ipv6_network}', actual: '{ipam_v6_network}'"
        )

    for index, side in enumerate(subscription.iptrunk.iptrunk_sides):
        lag_fqdn = f"{side.iptrunk_side_ae_iface}-0.{side.iptrunk_side_node.router_fqdn}"
        side_v4 = subscription.iptrunk.iptrunk_ipv4_network[index]
        side_v6 = subscription.iptrunk.iptrunk_ipv6_network[index + 1]
        #  Validate IPv4 address allocation
        record = infoblox.find_host_by_fqdn(lag_fqdn)
        if not record:
            ipam_errors.append(f"No IPv4 host record found with FQDN '{lag_fqdn}'")
        else:
            #  Allocation inside IPv4 network must be correct
            if str(side_v4) != record.ipv4addr:
                ipam_errors.append(
                    f"Incorrectly allocated host record for FQDN '{lag_fqdn}'.\n"
                    f"Expected '{side_v4}', actual: '{record.ipv4addr}'"
                )

            #  Allocated host record needs to be set correctly
            if record.comment != str(subscription.subscription_id):
                ipam_errors.append(
                    f"Incorrect host record found for '{lag_fqdn}' at '{side_v4}'. Comment should have been equal "
                    f"to subscription ID '{subscription.subscription_id}'."
                )

        #  Validate IPv6 address allocation
        record = infoblox.find_v6_host_by_fqdn(lag_fqdn)
        if not record:
            ipam_errors.append(f"No IPv6 host record found with FQDN '{lag_fqdn}'")
        else:
            #  Allocation inside IPv6 network must be correct
            if str(side_v6) != record.ipv6addr:
                ipam_errors.append(
                    f"Incorrectly allocated host record for FQDN '{lag_fqdn}'.\n"
                    f"Expected '{side_v6}', actual: '{record.ipv6addr}'"
                )

            #  Allocated host record needs to be set correctly
            if record.comment != str(subscription.subscription_id):
                ipam_errors.append(
                    f"Incorrect host record found for '{lag_fqdn}' at '{side_v6}'. Comment should have been equal "
                    f"to subscription ID '{subscription.subscription_id}'."
                )

    if ipam_errors:
        raise ProcessFailureError(message="IPAM misconfiguration(s) found", details=str(ipam_errors))

verify_netbox_entries(subscription)

Validate required entries for an IP trunk in NetBox.

Source code in gso/workflows/iptrunk/validate_iptrunk.py 
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
@step("Verify NetBox entries")
def verify_netbox_entries(subscription: Iptrunk) -> None:
    """Validate required entries for an IP trunk in NetBox."""
    nbclient = NetboxClient()
    netbox_errors = []
    for side in subscription.iptrunk.iptrunk_sides:
        if get_router_vendor(side.iptrunk_side_node.owner_subscription_id) == Vendor.NOKIA:
            #  Raises en exception when not found.
            interface = nbclient.get_interface_by_name_and_device(
                side.iptrunk_side_ae_iface, side.iptrunk_side_node.router_fqdn
            )
            if interface.description != str(subscription.subscription_id):
                netbox_errors.append(
                    f"Incorrect description for '{side.iptrunk_side_ae_iface}', expected "
                    f"'{subscription.subscription_id}' but got '{interface.description}'"
                )
            if not interface.enabled:
                netbox_errors.append(f"NetBox interface '{side.iptrunk_side_ae_iface}' is not enabled.")
            for member in side.iptrunk_side_ae_members:
                interface = nbclient.get_interface_by_name_and_device(
                    member.interface_name, side.iptrunk_side_node.router_fqdn
                )
                if interface.description != str(subscription.subscription_id):
                    netbox_errors.append(
                        f"Incorrect description for '{member.interface_name}', expected "
                        f"'{subscription.subscription_id}' but got '{interface.description}'"
                    )
                if not interface.enabled:
                    netbox_errors.append(f"NetBox interface '{side.iptrunk_side_ae_iface}' is not enabled.")

    if netbox_errors:
        raise ProcessFailureError(message="NetBox misconfiguration(s) found", details=str(netbox_errors))

verify_iptrunk_config(subscription)

Check for configuration drift on the relevant routers.

Source code in gso/workflows/iptrunk/validate_iptrunk.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@step("Verify configuration of IPtrunk")
def verify_iptrunk_config(subscription: Iptrunk) -> LSOState:
    """Check for configuration drift on the relevant routers."""
    return {
        "playbook_name": "gap_ansible/playbooks/iptrunks.yaml",
        "inventory": {
            "all": {
                "hosts": {
                    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None,
                    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn: None,
                }
            }
        },
        "extra_vars": {
            "wfo_trunk_json": json.loads(json_dumps(subscription)),
            "verb": "deploy",
            "dry_run": "true",
            "config_object": "trunk_interface",
            "is_verification_workflow": "true",
        },
    }

check_ip_trunk_isis(subscription)

Run an Ansible playbook to check for any ISIS configuration drift.

Source code in gso/workflows/iptrunk/validate_iptrunk.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
@step("Check ISIS configuration")
def check_ip_trunk_isis(subscription: Iptrunk) -> LSOState:
    """Run an Ansible playbook to check for any ISIS configuration drift."""
    return {
        "playbook_name": "gap_ansible/playbooks/iptrunks.yaml",
        "inventory": {
            "all": {
                "hosts": {
                    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None,
                    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn: None,
                }
            }
        },
        "extra_vars": {
            "wfo_trunk_json": json.loads(json_dumps(subscription)),
            "verb": "deploy",
            "dry_run": "true",
            "config_object": "isis_interface",
            "is_verification_workflow": "true",
        },
    }

verify_twamp_config(subscription)

Check for configuration drift of TWAMP.

Source code in gso/workflows/iptrunk/validate_iptrunk.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
@step("Verify TWAMP configuration")
def verify_twamp_config(subscription: Iptrunk) -> LSOState:
    """Check for configuration drift of TWAMP."""
    return {
        "playbook_name": "gap_ansible/playbooks/deploy_twamp.yaml",
        "inventory": {
            "all": {
                "hosts": {
                    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None,
                    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn: None,
                }
            }
        },
        "extra_vars": {
            "subscription": json.loads(json_dumps(subscription)),
            "verb": "deploy",
            "dry_run": "true",
            "is_verification_workflow": "true",
        },
    }

validate_iptrunk()

Validate an existing, active IP Trunk subscription.

  • Verify that the LAG interfaces are correctly configured in IPAM.
  • Check correct configuration of interfaces in NetBox.
  • Verify the configuration on both sides of the trunk is intact.
  • Check the ISIS metric of the trunk.
  • Verify that TWAMP configuration is correct.

If a trunk has a Juniper router on both sides, it is considered legacy and does not require validation.

Source code in gso/workflows/iptrunk/validate_iptrunk.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@workflow(
    "Validate IP trunk configuration", target=Target.SYSTEM, initial_input_form=wrap_modify_initial_input_form(None)
)
def validate_iptrunk() -> StepList:
    """Validate an existing, active IP Trunk subscription.

    * Verify that the LAG interfaces are correctly configured in IPAM.
    * Check correct configuration of interfaces in NetBox.
    * Verify the configuration on both sides of the trunk is intact.
    * Check the ISIS metric of the trunk.
    * Verify that TWAMP configuration is correct.

    If a trunk has a Juniper router on both sides, it is considered legacy and does not require validation.
    """
    skip_legacy_trunks = conditional(
        lambda state: all(
            side.iptrunk_side_node.vendor == Vendor.JUNIPER
            for side in Iptrunk.from_subscription(state["subscription_id"]).iptrunk.iptrunk_sides
        )
    )

    return (
        begin
        >> store_process_subscription(Target.SYSTEM)
        >> skip_legacy_trunks(done)
        >> unsync
        >> verify_ipam_records
        >> verify_netbox_entries
        >> anonymous_lso_interaction(verify_iptrunk_config)
        >> anonymous_lso_interaction(check_ip_trunk_isis)
        >> anonymous_lso_interaction(verify_twamp_config)
        >> resync
        >> done
    )