diff --git a/gso/services/_ipam.py b/gso/services/_ipam.py
index 7e9ad241a91a0b86b25e8a7c75b44cb64981976b..0688dcad3be2f321f5aafc31826a0310d0ce9b8d 100644
--- a/gso/services/_ipam.py
+++ b/gso/services/_ipam.py
@@ -18,8 +18,8 @@ class V6ServiceNetwork(BaseSettings):
class ServiceNetworks(BaseSettings):
- v4: V4ServiceNetwork
- v6: V6ServiceNetwork
+ v4: ipaddress.IPv4Network
+ v6: ipaddress.IPv6Network
class V4HostAddress(BaseSettings):
@@ -27,12 +27,12 @@ class V4HostAddress(BaseSettings):
class V6HostAddress(BaseSettings):
- v6: ipaddress.IPv6Address
+ v6: ipaddress.IPv6Address
class HostAddresses(BaseSettings):
- v4: V4HostAddress
- v6: V6HostAddress
+ v4: ipaddress.IPv4Address
+ v6: ipaddress.IPv6Address
# TODO: remove this!
@@ -65,7 +65,7 @@ def _ip_network_version(network):
assert ip_version in [4, 6]
return ip_version
-def find_containers(network=None, ip_version=4):
+def _find_containers(network=None, ip_version=4):
"""
If network is not None, find the container that contains specified network.
Otherwise find all containers.
@@ -84,7 +84,7 @@ def find_containers(network=None, ip_version=4):
assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
return r.json()
-def find_networks(network_container=None, network=None, ip_version=4):
+def _find_networks(network_container=None, network=None, ip_version=4):
"""
If network_container is not None, find all networks within the specified container.
Otherwise if network is not None, find the specified network.
@@ -110,7 +110,7 @@ def find_networks(network_container=None, network=None, ip_version=4):
assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
return r.json()
-def get_network_capacity(network=None):
+def _get_network_capacity(network=None):
"""
Get utilization of a IPv4 network in a fraction of 1000.
"""
@@ -128,6 +128,7 @@ def get_network_capacity(network=None):
auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
verify=False
)
+ # Utilization info takes several minutes to converge. The IPAM utilization bar in the GUI as well. Why?
assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
capacity_info = r.json()
assert len(capacity_info) == 1, "Requested IPv4 network doesn't exist."
@@ -208,11 +209,11 @@ def allocate_service_ipv6_network(service_type) -> V6ServiceNetwork:
assert hasattr(ipam_params, service_type) and service_type != 'INFOBLOX', "Invalid service type."
return _allocate_network(ipam_params.INFOBLOX, getattr(ipam_params, service_type).V6, 6)
-def delete_network(network) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
+def _delete_network(network) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
"""
Delete IPv4 or IPv6 network by CIDR.
"""
- # TODO: should we check that there are no hosts in this network?
+ # TODO: should we check that there are no hosts in this network before deleting?
# Deleting a network deletes the hosts in it, but not the associated DNS records.
oss = settings.load_oss_params()
assert oss.IPAM.INFOBLOX
@@ -220,7 +221,7 @@ def delete_network(network) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
ip_version = _ip_network_version(network)
- network_info = find_networks(network=network, ip_version=ip_version)
+ network_info = _find_networks(network=network, ip_version=ip_version)
assert len(network_info) == 1, "Network does not exist."
assert '_ref' in network_info[0]
@@ -245,20 +246,20 @@ def _find_next_available_ip(infoblox_params, network_ref):
auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
verify=False
)
- # TODO: handle no more available IPs in the network
+ # TODO: propagate no more available IPs in the network
assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
assert 'ips' in r.json()
received_ip = r.json()['ips']
assert len(received_ip) == 1
return received_ip[0]
-def allocate_host(hostname=None, addr=None, network=None) -> Union[V4HostAddress, V6HostAddress]:
+def _allocate_host(hostname=None, addr=None, network=None) -> Union[V4HostAddress, V6HostAddress]:
"""
If network is not None, allocate host in that network.
Otherwise if addr is not None, allocate host with that address.
hostname parameter must be full name including domain name.
"""
- # TODO: should hostnames be unique (i.e. fail if hostname already exists in this domain)?
+ # TODO: should hostnames be unique (i.e. fail if hostname already exists in this domain/service)?
assert addr or network, "You must specify either the host address or the network CIDR."
oss = settings.load_oss_params()
assert oss.IPAM.INFOBLOX
@@ -267,7 +268,7 @@ def allocate_host(hostname=None, addr=None, network=None) -> Union[V4HostAddress
if network:
ip_version = _ip_network_version(network)
# Find the next available IP address in the network
- network_info = find_networks(network=network, ip_version=ip_version)
+ network_info = _find_networks(network=network, ip_version=ip_version)
assert len(network_info) == 1, "Network does not exist. Create it first."
assert '_ref' in network_info[0]
addr = _find_next_available_ip(infoblox_params, network_info[0]["_ref"])
@@ -339,49 +340,58 @@ def allocate_host(hostname=None, addr=None, network=None) -> Union[V4HostAddress
else:
return V6HostAddress(v6=addr)
-def allocate_service_host(hostname=None, service_type=None) -> HostAddresses:
+def allocate_service_host(hostname=None, service_type=None, service_networks: ServiceNetworks = None) -> HostAddresses:
"""
Allocate host with both IPv4 and IPv6 address (and respective DNS records).
- The first network with available space for this service type is used.
The domain name is also taken from the service type and appended to specified hostname.
+ If service_networks is not provided, the first network with available space for this service type is used.
+ If service_networks is provided, that one is used.
"""
oss = settings.load_oss_params()
assert oss.IPAM
ipam_params = oss.IPAM
- assert ipam_params.INFOBLOX
- infoblox_params = ipam_params.INFOBLOX
assert hasattr(ipam_params, service_type) and service_type != 'INFOBLOX', "Invalid service type."
ipv4_container = getattr(ipam_params, service_type).V4.container
ipv6_container = getattr(ipam_params, service_type).V6.container
domain_name = getattr(ipam_params, service_type).domain_name
- ipv4_networks_info = find_networks(network_container=str(ipv4_container), ip_version=4)
- assert len(ipv4_networks_info) >= 1, "No IPv4 network exists in the container for this service type."
- first_nonfull_ipv4_network = None
- for ipv4_network_info in ipv4_networks_info:
- assert 'network' in ipv4_network_info
- capacity = get_network_capacity(ipv4_network_info["network"])
- if capacity < 1000:
- first_nonfull_ipv4_network = ipv4_network_info["network"]
- break
- # Create a new network if the existing networks in the container for the service type are all full.
- if not first_nonfull_ipv4_network:
- first_nonfull_ipv4_network = str(allocate_service_ipv4_network(service_type=service_type).v4)
- assert first_nonfull_ipv4_network, "No available IPv4 addresses for this service type."
- v4_host = allocate_host(hostname=hostname+domain_name, network=first_nonfull_ipv4_network)
-
- # ipv6 does not support capacity fetching (not even the GUI displays it).
- # Maybe it's assumed that there is always available space?
- ipv6_networks_info = find_networks(network_container=str(ipv6_container), ip_version=6)
- assert len(ipv6_networks_info) >= 1, "No IPv6 network exists in the container for this service type."
- assert 'network' in ipv6_networks_info[0]
- # TODO: if "no available IP" error, create a new network?
- v6_host = allocate_host(hostname=hostname+domain_name, network=ipv6_networks_info[0]['network'])
-
- return HostAddresses(v4=v4_host,v6=v6_host)
-
-def delete_host_by_ip(addr):
+ # IPv4
+ if not service_networks:
+ ipv4_networks_info = _find_networks(network_container=str(ipv4_container), ip_version=4)
+ assert len(ipv4_networks_info) >= 1, "No IPv4 network exists in the container for this service type."
+ first_nonfull_ipv4_network = None
+ for ipv4_network_info in ipv4_networks_info:
+ assert 'network' in ipv4_network_info
+ capacity = _get_network_capacity(ipv4_network_info["network"])
+ if capacity < 1000:
+ first_nonfull_ipv4_network = ipv4_network_info["network"]
+ break
+ # Create a new network if the existing networks in the container for the service type are all full.
+ if not first_nonfull_ipv4_network:
+ first_nonfull_ipv4_network = str(allocate_service_ipv4_network(service_type=service_type).v4)
+ assert first_nonfull_ipv4_network, "No available IPv4 addresses for this service type."
+ v4_host = _allocate_host(hostname=hostname+domain_name, network=first_nonfull_ipv4_network)
+ else:
+ network = str(service_networks.v4)
+ v4_host = _allocate_host(hostname=hostname+domain_name, network=network)
+
+ # IPv6
+ if not service_networks:
+ # ipv6 does not support capacity fetching (not even the GUI displays it).
+ # Maybe it's assumed that there is always available space?
+ ipv6_networks_info = _find_networks(network_container=str(ipv6_container), ip_version=6)
+ assert len(ipv6_networks_info) >= 1, "No IPv6 network exists in the container for this service type."
+ assert 'network' in ipv6_networks_info[0]
+ # TODO: if "no available IP" error, create a new network?
+ v6_host = _allocate_host(hostname=hostname+domain_name, network=ipv6_networks_info[0]['network'])
+ else:
+ network = str(service_networks.v6)
+ v6_host = _allocate_host(hostname=hostname+domain_name, network=network)
+
+ return HostAddresses(v4=v4_host.v4,v6=v6_host.v6)
+
+def _delete_host_by_ip(addr) -> Union[V4HostAddress, V6HostAddress]:
"""
Delete IPv4 or IPv6 host by its address.
"""
@@ -455,17 +465,17 @@ if __name__ == '__main__':
if choice == '1':
ip_version = int(input("Enter IP version (4 or 6): "))
- containers = find_containers(ip_version=ip_version)
+ containers = _find_containers(ip_version=ip_version)
print(json.dumps(containers, indent=2))
elif choice == '2':
ip_version = int(input("Enter IP version (4 or 6): "))
- networks = find_networks(ip_version=ip_version)
+ networks = _find_networks(ip_version=ip_version)
print(json.dumps(networks, indent=2))
elif choice == '3':
network = input("Enter network (in CIDR notation): ")
- network_capacity = get_network_capacity(network=network)
+ network_capacity = _get_network_capacity(network=network)
print(json.dumps(network_capacity, indent=2))
elif choice == '4':
@@ -482,19 +492,19 @@ if __name__ == '__main__':
elif choice == '5':
network = input("Enter network to delete (in CIDR notation): ")
- deleted_network = delete_network(network=network)
+ deleted_network = _delete_network(network=network)
print(json.dumps(str(deleted_network), indent=2))
elif choice == '6':
hostname = input("Enter host name (full name including domain name): ")
addr = input("Enter IP address to allocate: ")
- alloc_ip = allocate_host(hostname=hostname, addr=addr)
+ alloc_ip = _allocate_host(hostname=hostname, addr=addr)
print(json.dumps(str(alloc_ip), indent=2))
elif choice == '7':
hostname = input("Enter host name (full name including domain name): ")
network = input("Enter an existing network to allocate from (in CIDR notation): ")
- alloc_ip = allocate_host(hostname=hostname, network=network)
+ alloc_ip = _allocate_host(hostname=hostname, network=network)
print(json.dumps(str(alloc_ip), indent=2))
elif choice == '8':
@@ -505,7 +515,7 @@ if __name__ == '__main__':
elif choice == '9':
addr = input("Enter IP address of host to delete: ")
- deleted_host = delete_host_by_ip(addr=addr)
+ deleted_host = _delete_host_by_ip(addr=addr)
print(json.dumps(str(deleted_host), indent=2))
elif choice == '10':
diff --git a/gso/services/ipam.py b/gso/services/ipam.py
index fc646e433c0532050d3e82e2466c5cd74d7449a9..723e112ed77f4546bef1b52417cf2cb3b7ae640a 100644
--- a/gso/services/ipam.py
+++ b/gso/services/ipam.py
@@ -14,8 +14,8 @@ class V6ServiceNetwork(BaseSettings):
class ServiceNetworks(BaseSettings):
- v4: V4ServiceNetwork
- v6: V6ServiceNetwork
+ v4: ipaddress.IPv4Network
+ v6: ipaddress.IPv6Network
class V4HostAddress(BaseSettings):
@@ -23,26 +23,33 @@ class V4HostAddress(BaseSettings):
class V6HostAddress(BaseSettings):
- v6: ipaddress.IPv6Address
+ v6: ipaddress.IPv6Address
class HostAddresses(BaseSettings):
- v4: V4HostAddress
- v6: V6HostAddress
+ v4: ipaddress.IPv4Address
+ v6: ipaddress.IPv6Address
def new_service_networks(service_type) -> ServiceNetworks:
v4_service_network = _ipam.allocate_service_ipv4_network(service_type=service_type)
v6_service_network = _ipam.allocate_service_ipv6_network(service_type=service_type)
return ServiceNetworks(
- v4=v4_service_network,
- v6=v6_service_network)
+ v4=v4_service_network.v4,
+ v6=v6_service_network.v6)
-def new_service_host(hostname, service_type) -> HostAddresses:
- return _ipam.allocate_service_host(hostname=hostname, service_type=service_type)
+def new_service_host(hostname, service_type, service_networks: ServiceNetworks) -> HostAddresses:
+ return _ipam.allocate_service_host(hostname=hostname, service_type=service_type, service_networks=service_networks)
if __name__ == '__main__':
- new_service_networks('TRUNK')
- #new_service_host('newhost12', 'TRUNK')
\ No newline at end of file
+ hostname_A = 'newhost1'
+ hostname_B = 'newhost2'
+ lo1_service_networks = new_service_networks('LO')
+ new_service_host(hostname_A, 'LO', lo1_service_networks)
+ lo2_service_networks = new_service_networks('LO')
+ new_service_host(hostname_B, 'LO', lo2_service_networks)
+ trunk12_service_networks = new_service_networks('TRUNK')
+ new_service_host(hostname_A, 'TRUNK', trunk12_service_networks)
+ new_service_host(hostname_B, 'TRUNK', trunk12_service_networks)
\ No newline at end of file