Finished feature service-placeholders.

geant_service_orchestrator/main.py

@@ -2,8 +2,8 @@ from orchestrator import OrchestratorCore
 from orchestrator.cli.main import app as core_cli
 from orchestrator.settings import AppSettings
 
-import products
-import workflows
+# import products
+# import workflows
 
 app = OrchestratorCore(base_settings=AppSettings())
 

geant_service_orchestrator/oss-params-example.json

+{
+  "RESOURCE_MANAGER_API_PREFIX": "http://localhost:44444",
+  "IPAM": {
+    "INFOBLOX": {
+      "scheme": "https",
+      "wapi_version": "v2.12",
+      "host": "10.0.0.1",
+      "username": "robot-user",
+      "password": "robot-user-password"
+    },
+    "TRUNK": {
+      "V4": {"container": "1.1.0.0/19", "mask": 31},
+      "V6": {"container": "dead:beef::/29", "mask": 96}
+    },
+    "GEANT_IP": {
+      "V4": {"container": "1.1.8.0/19", "mask": 31},
+      "V6": {"container": "dead:beef::/29", "mask": 96}
+    }
+  }
+}
\ No newline at end of file

geant_service_orchestrator/services/ipam.py

+import ipaddress
+from pydantic import BaseSettings
+from geant_service_orchestrator import settings
+
+
+class ServiceNetworks(BaseSettings):
+    v4: ipaddress.IPv4Network
+    v6: ipaddress.IPv6Network
+
+
+class HostAddresses(BaseSettings):
+    v4: ipaddress.IPv4Address
+    v6: ipaddress.IPv6Address
+
+
+def new_service_networks(
+        ipam: settings.IPAMParams,
+        service_params: settings.ServiceNetworkParams) -> ServiceNetworks:
+    # TODO: load from ipam
+    # cf. https://gitlab.geant.org/goat/gap-jenkins/-/blob/development/service-editor/gap_service_editor/ipam.py#L35-66  # noqa: E501
+    return ServiceNetworks(
+        v4=ipaddress.IPv4Network('10.0.0.0/24'),
+        v6=ipaddress.IPv6Network('dead:beef::/120'))
+
+
+def new_host_address(fqdn: str, networks: ServiceNetworks) -> HostAddresses:
+    # TODO: load from ipam
+    return HostAddresses(
+        v4=ipaddress.IPv4Address('10.0.0.1'),
+        v6=ipaddress.IPv6Address('dead:beef::1'))

geant_service_orchestrator/services/resource_manager.py

+from geant_service_orchestrator import settings
+import requests
+
+
+def import_new_router(router_name, oss_params=settings.OSSParams):
+    r = requests.post(
+        f'{oss_params.RESOURCE_MANAGER_API_PREFIX}'
+        f'/api/interfaces/initialize-router/{router_name}')
+    r.raise_for_status()
+
+
+def next_lag(router_name, oss_params=settings.OSSParams):
+    r = requests.post(
+        f'{oss_params.RESOURCE_MANAGER_API_PREFIX}'
+        f'/api/interfaces/next-lag/{router_name}')
+    r.raise_for_status()
+    response = r.json()
+    return response['name']
+
+
+def next_physical(router_name, lag_name, oss_params=settings.OSSParams):
+    # TODO: speed needed (if first interface)
+    r = requests.post(
+        f'{oss_params.RESOURCE_MANAGER_API_PREFIX}'
+        f'/api/interfaces/next-physical/{router_name}/{lag_name}')
+    r.raise_for_status()
+    response = r.json()
+    return response['name']

geant_service_orchestrator/settings.py

+import ipaddress
+import json
+import os
+from pydantic import BaseSettings
+
+
+class InfoBloxParams(BaseSettings):
+    scheme: str
+    wapi_version: str
+    host: str
+    username: str
+    password: str
+
+
+class V4NetworkParams(BaseSettings):
+    container: ipaddress.IPv4Network
+    mask: int  # TODO: validation on mask?
+
+
+class V6NetworkParams(BaseSettings):
+    container: ipaddress.IPv6Network
+    mask: int  # TODO: validation on mask?
+
+
+class ServiceNetworkParams(BaseSettings):
+    V4: V4NetworkParams
+    V6: V6NetworkParams
+
+
+class IPAMParams(BaseSettings):
+    INFOBLOX: InfoBloxParams
+    TRUNK: ServiceNetworkParams
+    GEANT_IP: ServiceNetworkParams
+
+
+class OSSParams(BaseSettings):
+    IPAM: IPAMParams
+    RESOURCE_MANAGER_API_PREFIX: str  # api prefix
+
+
+def load_oss_params() -> OSSParams:
+    """
+    look for OSS_PARAMS_FILENAME in the environment and load the
+    parameters from that file
+    """
+    with open(os.environ['OSS_PARAMS_FILENAME']) as f:
+        return OSSParams(**json.loads(f.read()))
+
+
+if __name__ == '__main__':
+    print(load_oss_params())

requirements.txt

 orchestrator-core
+requests
 
 pytest