-
Karel van Klink authoredKarel van Klink authored
create_device.py 6.74 KiB
import ipaddress
import re
from uuid import uuid4
from orchestrator.db.models import ProductTable, SubscriptionTable
# noinspection PyProtectedMember
from orchestrator.forms import FormPage
from orchestrator.forms.validators import Choice, choice_list
from orchestrator.targets import Target
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_blocks import device as device_pb
from gso.products.product_types import device
from gso.products.product_types.device import DeviceInactive, \
DeviceProvisioning
from gso.products.product_types.site import Site
from gso.services import _ipam
from gso.services import provisioning_proxy
from gso.services.provisioning_proxy import await_pp_results, \
confirm_pp_results
def site_selector() -> list:
site_subscriptions = {}
for site_id, site_description in (
SubscriptionTable.query
.join(ProductTable)
.filter(
ProductTable.product_type == 'Site',
SubscriptionTable.status == 'active')
.with_entities(SubscriptionTable.subscription_id,
SubscriptionTable.description)
.all()
):
site_subscriptions[str(site_id)] = site_description
# noinspection PyTypeChecker
return choice_list(
Choice('site_selection',
zip(site_subscriptions.keys(),
site_subscriptions.items())), # type:ignore
min_items=1,
max_items=1,
)
def initial_input_form_generator(product_name: str) -> FormGenerator:
class CreateDeviceForm(FormPage):
class Config:
title = product_name
device_site: site_selector()
hostname: str
ts_address: ipaddress.IPv4Address
ts_port: int
device_vendor: device_pb.DeviceVendor
device_role: device_pb.DeviceRole
user_input = yield CreateDeviceForm
return user_input.dict()
@step('Create subscription')
def create_subscription(product: UUIDstr) -> State:
subscription = DeviceInactive.from_product_id(product, uuid4())
return {
'subscription': subscription,
'subscription_id': subscription.subscription_id,
}
def iso_from_ipv4(ipv4_address):
padded_octets = [f'{x:>03}' for x in ipv4_address.split('.')]
joined_octets = ''.join(padded_octets)
re_split = '.'.join(re.findall('....', joined_octets))
result = '.'.join(['49.51e5.0001', re_split, '00'])
return result
@step('Get information from IPAM')
def get_info_from_ipam(subscription: DeviceProvisioning) -> State:
lo0_alias = re.sub('.geant.net', '', subscription.device.device_fqdn)
lo0_name = f'lo0.{lo0_alias}'
lo0_addr = _ipam.allocate_service_host(hostname=lo0_name,
service_type='LO',
cname_aliases=[lo0_alias])
subscription.device.device_lo_ipv4_address = lo0_addr.v4
subscription.device.device_lo_ipv6_address = lo0_addr.v6
subscription.device.device_lo_iso_address \
= iso_from_ipv4(str(subscription.device.device_lo_ipv4_address))
subscription.device.device_si_ipv4_network = '192.168.0.0/31'
subscription.device.device_ias_lt_ipv4_network = '192.168.1.0/31'
subscription.device.device_ias_lt_ipv6_network = 'fc00:798:1::150/126'
return {'subscription': subscription}
@step('Initialize subscription')
def initialize_subscription(
subscription: device.DeviceInactive,
hostname: str,
ts_address: ipaddress.IPv4Address,
ts_port: str,
device_vendor: device_pb.DeviceVendor,
device_site: str,
device_role: device_pb.DeviceRole
) -> State:
subscription.device.device_ts_address = str(ts_address)
subscription.device.device_ts_port = str(ts_port)
subscription.device.device_vendor = device_vendor
subscription.device.device_site \
= Site.from_subscription(device_site[0]).site
fqdn = f'{hostname}.{subscription.device.device_site.site_name.lower()}.' \
f'{subscription.device.device_site.site_country_code.lower()}' \
f'.geant.net'
subscription.device.device_fqdn = fqdn
subscription.device.device_role = device_role
subscription.description = f'Device {fqdn} ' \
f'({subscription.device_type})'
subscription = device.DeviceProvisioning.from_other_lifecycle(
subscription, SubscriptionLifecycle.PROVISIONING
)
return {'subscription': subscription}
@step('Provision device [DRY RUN]')
def provision_device_dry(subscription: DeviceProvisioning,
process_id: UUIDstr) -> State:
provisioning_proxy.provision_device(subscription, process_id)
return {'subscription': subscription,
'label_text': f'This is a dry run for the deployment of a new '
f'{subscription.device_type}. Deployment is being '
f'taken care of by the provisioning proxy, please '
f'wait for the results to come back before '
f'continuing.'
}
@step('Provision device [FOR REAL]')
def provision_device_real(subscription: DeviceProvisioning,
process_id: UUIDstr) -> State:
provisioning_proxy.provision_device(subscription, process_id, False)
return {'subscription': subscription,
'label_text': f'This is a live deployment of a new '
f'{subscription.device_type}. Deployment is being '
f'taken care of by the provisioning proxy, please '
f'wait for the results to come back before '
f'continuing.'
}
@workflow(
'Create device',
initial_input_form=wrap_create_initial_input_form(
initial_input_form_generator),
target=Target.CREATE,
)
def create_device():
return (
init
>> create_subscription
>> store_process_subscription(Target.CREATE)
>> initialize_subscription
>> get_info_from_ipam
>> provision_device_dry
>> await_pp_results
>> confirm_pp_results
>> provision_device_real
>> await_pp_results
>> confirm_pp_results
>> set_status(SubscriptionLifecycle.ACTIVE)
>> resync
>> done
)