Refactor - oeg architecture

---

kind: PersistentVolume

apiVersion: v1

metadata:

  name: mongodb-oeg-pv-volume  # Sets PV's name

  labels:

    type: local  # Sets PV's type to local

    app: oegmongo

spec:

  storageClassName: manual

  capacity:

    storage: 50Mi # Sets PV Volume

  accessModes:

    - ReadWriteOnce

  hostPath:

    path: "/mnt/data/mongodb_oeg"

---

apiVersion: v1

kind: PersistentVolumeClaim

metadata:

  creationTimestamp: null

  labels:

    io.kompose.service: mongo-oeg

  name: mongo-oeg

spec:

  storageClassName: manual

  accessModes:

    - ReadWriteOnce

  resources:

    requests:

      storage: 50Mi

status: {}

---

apiVersion: apps/v1

kind: Deployment

metadata:

  annotations:

    kompose.cmd: kompose convert

    kompose.version: 1.26.0 (40646f47)

  creationTimestamp: null

  labels:

    io.kompose.service: oegmongo

  name: oegmongo

spec:

  replicas: 1

  selector:

    matchLabels:

      io.kompose.service: oegmongo

  strategy:

    type: Recreate

  template:

    metadata:

      annotations:

        kompose.cmd: kompose convert

        kompose.version: 1.26.0 (40646f47)

      creationTimestamp: null

      labels:

        #io.kompose.network/netEMPkub: "true"

        io.kompose.service: oegmongo

    spec:

      containers:

        - image: mongo

          name: oegmongo

          ports:

            - containerPort: 27017

          resources: {}

          volumeMounts:

            - mountPath: /data/db

              name: mongo-db

      restartPolicy: Always

      volumes:

        - name: mongo-db

          persistentVolumeClaim:

            claimName: mongo-oeg

status: {}

---

apiVersion: v1

kind: Service

metadata:

  annotations:

    kompose.cmd: kompose convert

    kompose.version: 1.26.0 (40646f47)

  creationTimestamp: null

  labels:

    io.kompose.service: oegmongo

  name: oegmongo

spec:

  type: ClusterIP

  ports:

    - name: "27017"

      port: 27017

      targetPort: 27017

  selector:

    io.kompose.service: oegmongo

status:

  loadBalancer: {}

---

apiVersion: apps/v1

kind: Deployment

metadata:

  annotations:

    kompose.cmd: kompose convert

    kompose.version: 1.26.0 (40646f47)

  creationTimestamp: null

  labels:

    io.kompose.service: oegcontroller

  name: oegcontroller

spec:

  replicas: 1

  selector:

    matchLabels:

      io.kompose.service: oegcontroller

  strategy: {}

  template:

    metadata:

      annotations:

        kompose.cmd: kompose convert

        kompose.version: 1.26.0 (40646f47)

      creationTimestamp: null

      labels:

        io.kompose.service: oegcontroller

    spec:

      containers:

        - env:

            - name: MONGO_URI

              value: mongodb://oegmongo:27017

            - name: SRM_HOST

              value: http://srm:8080/srm/1.0.0

            - name: FEDERATION_MANAGER_HOST

              value: http://federation-manager.federation-manager.svc.cluster.local:8989/operatorplatform/federation/v1

            - name: PARTNER_API_ROOT

              value: http://10.8.0.1:31002

            - name: TOKEN_ENDPOINT

              value: http://federation-manager.federation-manager.svc.cluster.local:8080/realms/federation/protocol/openid-connect/token

          image: ghcr.io/sunriseopenoperatorplatform/oeg/oeg:1.0.1

          name: oegcontroller

          ports:

            - containerPort: 8080

          resources: {}

          imagePullPolicy: Always

      restartPolicy: Always

status: {}

---

apiVersion: v1

kind: Service

metadata:

  name: oeg

  namespace: sunrise6g

spec:

  type: ClusterIP

  selector:

    io.kompose.service: oegcontroller

  ports:

    - name: http

      port: 80

      targetPort: 8080

---

apiVersion: networking.k8s.io/v1

kind: Ingress

metadata:

  name: oegcontroller-ingress

  namespace: sunrise6g

  annotations:

    traefik.ingress.kubernetes.io/router.entrypoints: web

    # traefik.ingress.kubernetes.io/rewrite-target: /$2

spec:

  ingressClassName: traefik

  rules:

    - host: isiath.duckdns.org

      http:

        paths:

          - path: /oeg

            pathType: Prefix

            backend:

              service:

                name: oeg

                port:

                  number: 80

 No newline at end of file

import os

from pydantic.v1 import BaseSettings

from dotenv import load_dotenv

load_dotenv()

class Configuration(BaseSettings):

    MONGO_URI: str = os.getenv("MONGO_URI")

    SRM_HOST: str = os.getenv("SRM_HOST")

    SRM_USERNAME: str = os.getenv("SRM_USERNAME")

    SRM_PASSWORD: str = os.getenv("SRM_PASSWORD")

    HTTP_PROXY: str = os.getenv("HTTP_PROXY")

    FEDERATION_MANAGER_HOST=os.getenv("FEDERATION_MANAGER_HOST")

    TOKEN_ENDPOINT = os.getenv('TOKEN_ENDPOINT')

    PARTNER_API_ROOT = os.getenv('PARTNER_API_ROOT')

    AVAIL_ZONE_NOTIF_LINK = os.getenv('AVAIL_ZONE_NOTIF_LINK')

config = Configuration()

import re

import uuid

def ensure_gsma_id(value, pattern, prefix, min_len, max_len, fallback_source):

    if value and re.match(pattern, value):

        return value

    base = re.sub(r"[^A-Za-z0-9_]", "", str(fallback_source or ""))

    candidate = f"{prefix}{base}"

    if not re.match(r"^[A-Za-z]", candidate):

        candidate = f"{prefix}{candidate}"

    candidate = candidate[:max_len]

    if len(candidate) < min_len:

        candidate = (candidate + uuid.uuid4().hex)[:max_len]

    if not re.match(r"^[A-Za-z]", candidate):

        candidate = f"{prefix}{candidate}"

        candidate = candidate[:max_len]

    return candidate

def ensure_service_name(value, prefix, fallback_source):

    pattern = r"^[A-Za-z0-9][A-Za-z0-9_]{6,62}[A-Za-z0-9]$"

    if value and re.match(pattern, value):

        return value

    base = re.sub(r"[^A-Za-z0-9_]", "", str(fallback_source or ""))

    candidate = f"{prefix}{base}"

    candidate = re.sub(r"^_+", "", candidate)

    candidate = candidate[:64]

    if len(candidate) < 8:

        candidate = (candidate + uuid.uuid4().hex)[:64]

    if not re.match(r"^[A-Za-z0-9]", candidate):

        candidate = f"s{candidate}"

    if not re.match(r"[A-Za-z0-9]$", candidate):

        candidate = f"{candidate}0"

    return candidate[:64]

def ensure_res_pool(value, fallback_source):

    pattern = r"^[A-Za-z0-9][A-Za-z0-9_]{6,30}[A-Za-z0-9]$"

    if value and re.match(pattern, value):

        return value

    base = re.sub(r"[^A-Za-z0-9_]", "", str(fallback_source or ""))

    candidate = f"respool{base}"[:32]

    if len(candidate) < 8:

        candidate = (candidate + uuid.uuid4().hex)[:32]

    if not re.match(r"^[A-Za-z0-9]", candidate):

        candidate = f"r{candidate}"

    if not re.match(r"[A-Za-z0-9]$", candidate):

        candidate = f"{candidate}0"

    return candidate[:32]

def normalize_federated_app_id(app_id):

    pattern = (

        r"^(?:[A-Za-z][A-Za-z0-9_]{7,63}|"

        r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-"

        r"[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12})$"

    )

    if app_id and re.match(pattern, str(app_id)):

        return app_id

    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"federated-app:{app_id}"))

def normalize_federated_app_provider_id(app_provider_id, fallback_source):

    return ensure_gsma_id(

        app_provider_id,

        r"^[A-Za-z][A-Za-z0-9_]{7,63}$",

        "provider",

        8,

        64,

        fallback_source,

    )

def resolve_federated_app_provider_id(app_id_value, app_provider_id):

    if not app_provider_id:

        return None

    return normalize_federated_app_provider_id(app_provider_id, app_id_value)

def resolve_federated_app_identity(app_id_value, app_provider_id):

    if not app_id_value or not app_provider_id:

        return None, None

    return (

        normalize_federated_app_id(app_id_value),

        resolve_federated_app_provider_id(app_id_value, app_provider_id),

    )

def normalize_federated_artefact_id(federation_context_id, artefact_id, fallback_source):

    if artefact_id and re.match(

        r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-"

        r"[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$",

        str(artefact_id),

    ):

        return str(uuid.uuid5(uuid.NAMESPACE_URL, f"{federation_context_id}:{artefact_id}"))

    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"{federation_context_id}:{fallback_source}"))

def resolve_app_provider(srm_client, app_id_value, app_payload=None):

    if isinstance(app_payload, dict):

        provider = app_payload.get("appProvider") or app_payload.get("appProviderId")

        if provider:

            return provider

        manifest = app_payload.get("appManifest")

        if isinstance(manifest, dict):

            provider = manifest.get("appProvider") or manifest.get("appProviderId")

            if provider:

                return provider

    app_response = srm_client.get_app(app_id_value)

    if isinstance(app_response, dict):

        manifest = app_response.get("appManifest")

        if isinstance(manifest, dict):

            provider = manifest.get("appProvider") or manifest.get("appProviderId")

            if provider:

                return provider

        provider = app_response.get("appProvider") or app_response.get("appProviderId")

        if provider:

            return provider

    return None

def get_catalog_app_provider_map(srm_client):

    app_provider_map = {}

    apps = srm_client.get_service_functions_catalogue()

    if not isinstance(apps, list):

        return app_provider_map

    for app in apps:

        if not isinstance(app, dict):

            continue

        app_id_value = app.get("appId") or app.get("id")

        if not app_id_value:

            continue

        provider = resolve_app_provider(srm_client, app_id_value, app_payload=app)

        if provider:

            app_provider_map[app_id_value] = provider

    return app_provider_map

def iter_federated_instances(federation_client, feds, app_id_provider_pairs):

    for fed in feds:

        fed_token = fed.get("token")

        federation_context_id = fed.get("_id")

        if not federation_context_id or not fed_token:

            continue

        zone_provider = fed.get("partnerOPFederationId") or "unknown"

        for app_id_value, app_provider_id in app_id_provider_pairs:

            federated_app_id, federated_app_provider_id = resolve_federated_app_identity(

                app_id_value,

                app_provider_id,

            )

            if not federated_app_id or not federated_app_provider_id:

                continue

            fed_instances, fed_code = federation_client.get_all_app_instances(

                federation_context_id=federation_context_id,

                app_id=federated_app_id,

                app_provider_id=federated_app_provider_id,

                token=fed_token,

            )

            if fed_code != 200 or not isinstance(fed_instances, list):

                continue

            yield app_id_value, federated_app_id, federation_context_id, fed_token, zone_provider, fed_instances

from edge_cloud_management_api.services.storage_service import get_zone

from edge_cloud_management_api.controllers.edge_cloud_controller import get_local_zones

def _find_local_zone(zone_id, zone_provider=None):

    for zone in get_local_zones():

        if not isinstance(zone, dict):

            continue

        if zone.get("edgeCloudZoneId") != zone_id:

            continue

        if zone_provider and zone.get("edgeCloudProvider") != zone_provider:

            continue

        return zone

    return None

def normalize_local_app_instance(instance):

    if not isinstance(instance, dict):

        return None

    app_instance_id = instance.get("appInstanceId") or instance.get("appInstIdentifier")

    if not app_instance_id:

        return None

    status = instance.get("status") or instance.get("appInstanceState") or "unknown"

    edge_cloud_zone = instance.get("edgeCloudZone")

    if not isinstance(edge_cloud_zone, dict):

        zone_id = instance.get("edgeCloudZoneId") or instance.get("zoneId")

        zone_name = instance.get("edgeCloudZoneName") or instance.get("zoneName") or "unknown"

        edge_cloud_provider = instance.get("edgeCloudProvider") or "unknown"

        if zone_id:

            edge_cloud_zone = {

                "edgeCloudZoneId": zone_id,

                "edgeCloudZoneName": zone_name,

                "edgeCloudProvider": edge_cloud_provider,

                "edgeCloudZoneStatus": instance.get("edgeCloudZoneStatus") or "unknown",

                "edgeCloudRegion": instance.get("edgeCloudRegion") or "unknown",

            }

    normalized = {

        "appInstanceId": app_instance_id,

        "status": status,

    }

    if instance.get("appId"):

        normalized["appId"] = instance.get("appId")

    if edge_cloud_zone:

        normalized["edgeCloudZone"] = edge_cloud_zone

    if instance.get("componentEndpointInfo"):

        normalized["componentEndpointInfo"] = instance.get("componentEndpointInfo")

    elif instance.get("accesspointInfo"):

        normalized["componentEndpointInfo"] = instance.get("accesspointInfo")

    if instance.get("kubernetesClusterRef"):

        normalized["kubernetesClusterRef"] = instance.get("kubernetesClusterRef")

    return normalized

def enrich_instance_zone_from_catalog(instance):

    if not isinstance(instance, dict):

        return instance

    zone = instance.get("edgeCloudZone")

    if not isinstance(zone, dict):

        return instance

    zone_id = zone.get("edgeCloudZoneId")

    if not zone_id:

        return instance

    zone_provider = zone.get("edgeCloudProvider")

    stored_zone = None

    if zone_provider and zone_provider != "unknown":

        stored_zone = get_zone(zone_id, zone_provider)

    else:

        stored_zone = _find_local_zone(zone_id)

        if not isinstance(stored_zone, dict):

            stored_zone = get_zone(zone_id)

    if not isinstance(stored_zone, dict):

        return instance

    enriched = dict(instance)

    enriched_zone = dict(zone)

    for key in (

        "edgeCloudZoneId",

        "edgeCloudZoneName",

        "edgeCloudProvider",

        "edgeCloudZoneStatus",

        "edgeCloudRegion",

    ):

        if enriched_zone.get(key) in (None, "unknown") and stored_zone.get(key) is not None:

            enriched_zone[key] = stored_zone.get(key)

    enriched["edgeCloudZone"] = enriched_zone

    return enriched

def normalize_federated_app_instances(fed_instances, zone_provider=None, region=None):

    normalized_instances = []

    if not isinstance(fed_instances, list):

        return normalized_instances

    for zone_info in fed_instances:

        if not isinstance(zone_info, dict):

            continue

        zone_id = zone_info.get("zoneId")

        for instance in zone_info.get("appInstanceInfo", []) or []:

            if not isinstance(instance, dict):

                continue

            app_instance_id = instance.get("appInstIdentifier") or instance.get("appInstanceId")

            if not app_instance_id:

                continue

            status = instance.get("appInstanceState") or "unknown"

            if isinstance(status, str) and status.startswith("Error 404"):

                continue

            normalized = {

                "appInstanceId": app_instance_id,

                "status": status,

            }

            if instance.get("appId"):

                normalized["appId"] = instance.get("appId")

            if zone_id:

                normalized["edgeCloudZone"] = {

                    "edgeCloudZoneId": zone_id,

                    "edgeCloudZoneName": "unknown",

                    "edgeCloudProvider": zone_provider or "unknown",

                    "edgeCloudZoneStatus": "unknown",

                    "edgeCloudRegion": region or "unknown",

                }

            normalized_instances.append(normalized)

    return normalized_instances

def dedupe_app_instances(instances):

    deduped = []

    by_instance_id = {}

    def zone_quality(instance):

        zone = instance.get("edgeCloudZone")

        if not isinstance(zone, dict):

            return 0

        score = 0

        if zone.get("edgeCloudProvider") and zone.get("edgeCloudProvider") != "unknown":

            score += 4

        if zone.get("edgeCloudZoneName") and zone.get("edgeCloudZoneName") != "unknown":

            score += 2

        if zone.get("edgeCloudZoneStatus") and zone.get("edgeCloudZoneStatus") != "unknown":

            score += 1

        return score

    def merge_instances(existing, incoming):

        merged = dict(existing)

        for key, value in incoming.items():

            if key == "edgeCloudZone" and isinstance(value, dict):

                existing_zone = merged.get("edgeCloudZone")

                if not isinstance(existing_zone, dict) or zone_quality(incoming) > zone_quality(existing):

                    merged["edgeCloudZone"] = dict(value)

                else:

                    zone = dict(existing_zone)

                    for zone_key, zone_value in value.items():

                        if zone_key not in zone or zone.get(zone_key) in (None, "unknown"):

                            zone[zone_key] = zone_value

                    merged["edgeCloudZone"] = zone

                continue

            if key not in merged or merged.get(key) in (None, "unknown", []):

                merged[key] = value

        return merged

    for instance in instances:

        if not isinstance(instance, dict):

            continue

        app_instance_id = instance.get("appInstanceId")

        if app_instance_id:

            existing = by_instance_id.get(app_instance_id)

            if existing is not None:

                merged = merge_instances(existing, instance)

                by_instance_id[app_instance_id] = merged

                for index, deduped_instance in enumerate(deduped):

                    if deduped_instance.get("appInstanceId") == app_instance_id:

                        deduped[index] = merged

                        break

                continue

            by_instance_id[app_instance_id] = instance

        deduped.append(instance)

    return deduped