create_iptrunk.py

import ipaddress
from uuid import uuid4

from orchestrator.forms import FormPage
from orchestrator.targets import Target
from orchestrator.workflow import inputstep
from orchestrator.forms.validators import Accept
from orchestrator.types import FormGenerator, State
from orchestrator.types import SubscriptionLifecycle, UUIDstr
from orchestrator.workflow import done, init, step, workflow
from orchestrator.workflows.steps import resync, set_status
from orchestrator.workflows.steps import store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form
from gso.products.product_types import iptrunk
from gso.services import ipam, provisioning_proxy

from gso.products.product_types import device
from orchestrator.db.models import ProductTable, SubscriptionTable
from orchestrator.forms.validators import Choice, choice_list
from typing import List, Optional


def device_selector(choice_value: str) -> list:
    device_subscriptions = {}
    for device_id, device_description in (
        SubscriptionTable.query.join(ProductTable)
        .filter(
            ProductTable.product_type == "Device",
            SubscriptionTable.status == "active",
        )
        .with_entities(SubscriptionTable.subscription_id, SubscriptionTable.description)
        .all()
    ):
        device_subscriptions[str(device_id)] = device_description

    return choice_list(
        Choice(choice_value, zip(device_subscriptions.keys(), device_subscriptions.items())),  # type:ignore
        min_items=1,
        max_items=1,
    )


def initial_input_form_generator(product_name: str) -> FormGenerator:
    class CreateIptrunkForm(FormPage):
        class Config:
            title = product_name

        geant_s_sid: str
        iptrunk_speed: str  # This should be an enum: 1/10/100/400
        iptrunk_minimum_links: int

        iptrunk_sideA_fqdn: device_selector(choice_value='DeviceEnumA')
        iptrunk_sideA_ae_iface: str
        iptrunk_sideA_ae_geant_a_sid: str
        iptrunk_sideA_ae_members: list[str]

        iptrunk_sideB_fqdn: device_selector(choice_value='DeviceEnumB')
        iptrunk_sideB_ae_iface: str
        iptrunk_sideB_ae_geant_a_sid: str
        iptrunk_sideB_ae_members: list[str]
        # TODO: we need additional validation: 
        # sideA fqdn must be different from sideB fqdn
        # the lenght of iptrunk_sideA_ae_members should be the same as iptrunk_sideA_ae_members
        # interface names must be validated
        # We also need additional fields like a "name" or a description for trunk itself and for all the members

    user_input = yield CreateIptrunkForm

    return user_input.dict()


@step("Create subscription")
def create_subscription(product: UUIDstr) -> State:
    subscription = iptrunk.IptrunkInactive.from_product_id(product, uuid4())

    return {
        "subscription": subscription,
        "subscription_id": subscription.subscription_id,
    }


@step("Get information from IPAM ")
def get_info_from_ipam(subscription: iptrunk.IptrunkInactive) -> State:
    # TODO: get info about how these should be generated
    subscription.iptrunk.iptrunk_ipv4_network = "192.168.1.0/31"
    subscription.iptrunk.iptrunk_ipv6_network = "fc00:798:1::150/126"
    return {"subscription": subscription}




@step("Initialize subscription")
def initialize_subscription(
    subscription: iptrunk.IptrunkInactive,
    geant_s_sid: str,
    iptrunk_speed: str,
    iptrunk_minimum_links: int,
    iptrunk_sideA_fqdn: str,
    iptrunk_sideA_ae_iface: str,
    iptrunk_sideA_ae_geant_a_sid: str,
    iptrunk_sideA_ae_members: list[str],
    iptrunk_sideB_fqdn: str, 
    iptrunk_sideB_ae_iface: str,
    iptrunk_sideB_ae_geant_a_sid: str,
    iptrunk_sideB_ae_members: list[str]
) -> State:
   # subscription.iptrunk_type = iptrunk_type

    subscription.iptrunk.geant_s_sid = geant_s_sid
    subscription.iptrunk.iptrunk_speed = iptrunk_speed
    subscription.iptrunk.iptrunk_minimum_links = iptrunk_minimum_links

    subscription.iptrunk.iptrunk_sideA_fqdn = iptrunk_sideA_fqdn
    subscription.iptrunk.iptrunk_sideA_ae_iface = iptrunk_sideA_ae_iface
    subscription.iptrunk.iptrunk_sideA_ae_geant_a_sid = iptrunk_sideA_ae_geant_a_sid
    subscription.iptrunk.iptrunk_sideA_ae_members = iptrunk_sideA_ae_members

    subscription.iptrunk.iptrunk_sideB_fqdn = iptrunk_sideB_fqdn
    subscription.iptrunk.iptrunk_sideB_ae_iface = iptrunk_sideB_ae_iface
    subscription.iptrunk.iptrunk_sideB_ae_geant_a_sid = iptrunk_sideB_ae_geant_a_sid
    subscription.iptrunk.iptrunk_sideB_ae_members = iptrunk_sideB_ae_members
    subscription.description = f"Iptrunk {geant_s_sid}:  \
                                ({subscription.iptrunk_type})"
    subscription = iptrunk.IptrunkProvisioning.from_other_lifecycle(
        subscription, SubscriptionLifecycle.PROVISIONING
    )

    return {"subscription": subscription}


@step("Provision iptrunk [DRY RUN]")
def provision_iptrunk_dry(
    subscription: iptrunk.IptrunkProvisioning,
    fqdn: str,
    ts_address: str,
    ts_port: str
) -> State:
    # import ansible_runner
    #
    # r = ansible_runner.run(
    #     private_data_dir="/opt/geant-gap-ansible",
    #     playbook="base_config.yaml",
    #     inventory=subscription.iptrunk.fqdn,
    #     extravars={
    #         "lo_ipv4_address": str(subscription.iptrunk.lo_ipv4_address),
    #         "lo_ipv6_address": str(subscription.iptrunk.lo_ipv6_address),
    #         "lo_iso_address": subscription.iptrunk.lo_iso_address,
    #         "snmp_location": subscription.iptrunk.snmp_location,
    #         "si_ipv4_network": str(subscription.iptrunk.si_ipv4_network),
    #         "lt_ipv4_network": str(subscription.iptrunk.ias_lt_ipv4_network),
    #         "lt_ipv6_network": str(subscription.iptrunk.ias_lt_ipv6_network),
    #         "site_country_code": subscription.iptrunk.site_country_code,
    #         "verb": "deploy",
    #     },
    # )
    # out = r.stdout.read()
    # out_splitted = out.splitlines()
    # # if r.rc != 0:
    # #    raise ValueError("Ansible has failed")
    # return {"dry_run_output": out_splitted, "return_code": r.rc}
    provisioning_proxy.provision_node(
        node_subscription_params=subscription,
        dry_run=True)
    # TODO: figure out what to return when we are suspending & waiting
    #       for the provisioning-proxy to call back
    return {"return_code": 0}


@inputstep("Confirm step", assignee="CHANGES")
def confirm_step() -> FormGenerator:
    class ConfirmForm(FormPage):
        confirm: Accept

    user_input = yield ConfirmForm

    return {"confirm": user_input.confirm}


@step("Provision iptrunk [FOR REAL]")
def provision_iptrunk_real(
    subscription: iptrunk.IptrunkProvisioning,
    fqdn: str,
    ts_address: str,
    ts_port: str
) -> State:
    # import ansible_runner
    #
    # r = ansible_runner.run(
    #     private_data_dir="/opt/geant-gap-ansible",
    #     playbook="base_config.yaml",
    #     inventory=subscription.iptrunk.fqdn,
    #     extravars={
    #         "lo_ipv4_address": str(subscription.iptrunk.lo_ipv4_address),
    #         "lo_ipv6_address": str(subscription.iptrunk.lo_ipv6_address),
    #         "lo_iso_address": subscription.iptrunk.lo_iso_address,
    #         "snmp_location": subscription.iptrunk.snmp_location,
    #         "si_ipv4_network": str(subscription.iptrunk.si_ipv4_network),
    #         "lt_ipv4_network": str(subscription.iptrunk.ias_lt_ipv4_network),
    #         "lt_ipv6_network": str(subscription.iptrunk.ias_lt_ipv6_network),
    #         "site_country_code": subscription.iptrunk.site_country_code,
    #         "verb": "deploy",
    #     },
    # )
    # out = r.stdout.read()
    # out_splitted = out.splitlines()
    #
    # return {"real_run_output": out_splitted, "return_code": r.rc}
    provisioning_proxy.provision_node(
        node_subscription_params=subscription)
    # TODO: figure out what to return when we are suspending & waiting
    #       for the provisioning-proxy to call back
    return {"return_code": 0}


@workflow(
    "Create Iptrunk",
    initial_input_form=wrap_create_initial_input_form(
                       initial_input_form_generator),
    target=Target.CREATE,
)
def create_iptrunk():
    return (
        init
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        #>> get_info_from_ipam
        #>> get_snmp_info
        #>> initialize_subscription
        #>> provision_iptrunk_dry
        #>> confirm_step
        #>> provision_iptrunk_real
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )