-SpineLeaf:

export TFS_REGISTRY_IMAGES="http://localhost:32000/tfs/"

# Set the list of components, separated by spaces, you want to build images for, and deploy.

export TFS_COMPONENTS="context device pathcomp service nbi webui"

export TFS_COMPONENTS="context device pathcomp service nbi webui spine_leaf"

# Uncomment to activate Monitoring (old)

#export TFS_COMPONENTS="${TFS_COMPONENTS} monitoring"

#!/bin/bash

# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

########################################################################################################################

# Define your deployment settings here

########################################################################################################################

# If not already set, set the name of the Kubernetes namespace to deploy to.

export TFS_K8S_NAMESPACE=${TFS_K8S_NAMESPACE:-"tfs"}

########################################################################################################################

# Automated steps start here

########################################################################################################################

kubectl --namespace $TFS_K8S_NAMESPACE logs deployment/spine-leafservice -c server

#    # do test ...

#    descriptor_loader.unload()

import concurrent.futures, copy, json, logging, operator

import concurrent.futures, copy, ipaddress, json, logging, operator

from typing import Any, Dict, List, Optional, Tuple, Union

from common.proto.context_pb2 import (

    Connection, Context, ContextId, Device, DeviceId, Empty,

        return device_map.get(device_identifier)

    @staticmethod

    def _derive_peer_ip(local_cidr: str) -> str:

        if not local_cidr:

            return ''

        try:

            iface = ipaddress.ip_interface(local_cidr)

            if iface.network.num_addresses != 2:

                return ''

            for host in iface.network.hosts():

                host_s = str(host)

                if host_s != str(iface.ip):

                    return host_s

        except ValueError:

            return ''

        return ''

    def _build_spine_leaf_hints(self, fabric_name: str, device_uuid: str) -> Dict[str, Any]:

        logical_fabric = None

        for fabric in self.__logical_resources:

            lf_name = str(fabric.get('fabric_name', fabric.get('fabric_id', '')))

            if lf_name == fabric_name:

                logical_fabric = fabric

                break

        if logical_fabric is None:

            return {}

        # Build mappings: device_uuid -> {asn, ip, router_id, device_name}

        asn_allocs = logical_fabric.get('asn', {}).get('allocations', {})

        asn_by_device = {}

        name_by_device = {}

        for alloc_device, alloc in asn_allocs.items():

            asn_by_device[str(alloc_device)] = str(alloc.get('value', ''))

            name_by_device[str(alloc_device)] = str(alloc.get('device_name', ''))

        ip_allocs = logical_fabric.get('ip', {}).get('allocations', {})

        ip_by_device = {}

        for alloc_device, alloc in ip_allocs.items():

            ip_by_device[str(alloc_device)] = str(alloc.get('value', ''))

        router_id_allocs = logical_fabric.get('router_id', {}).get('allocations', {})

        router_id_by_device = {}

        for alloc_device, alloc in router_id_allocs.items():

            router_id_by_device[str(alloc_device)] = str(alloc.get('value', ''))

        interfaces_alloc = logical_fabric.get('interface_ip_address', {}).get('allocations', {})

        device_interfaces = interfaces_alloc.get(device_uuid, {})

        interfaces = device_interfaces.get('interfaces', device_interfaces.get('endpoints', {}))

        hint_interfaces: List[Dict[str, str]] = []

        underlay_ip = ''

        remote_address = ''

        remote_as = ''

        session_name = ''

        for interface_name, interface_data in interfaces.items():

            if isinstance(interface_data, dict):

                ip_value = str(interface_data.get('ip', ''))

                comment = str(interface_data.get('comment', ''))

                hint_interfaces.append({

                    'interface': str(interface_name),

                    'address': ip_value,

                    'comment': comment,

                })

                is_loopback = (

                    str(interface_name).lower() == 'lo'

                    or ip_value.endswith('/32')

                    or 'loopback' in comment.lower()

                )

                if not is_loopback and not underlay_ip:

                    underlay_ip = ip_value

                    target_device = str(interface_data.get('link_to_device', ''))

                    # Resolve remote_as and remote_address from link_to_device lookup

                    if target_device:

                        # Resolve remote AS: explicit in interface entry takes precedence,

                        # otherwise look up the ASN allocated to the target device.

                        remote_as = str(interface_data.get('remote_as', '')) or asn_by_device.get(target_device, '')

                        # Resolve remote address:

                        # 1) use explicit remote_address in interface data if present

                        # 2) else, use the target device's allocated ip (strip CIDR to host)

                        # 3) else, derive peer from local ip_value

                        remote_address = str(interface_data.get('remote_address', ''))

                        if not remote_address:

                            remote_ip_cidr = ip_by_device.get(target_device, '')

                            if remote_ip_cidr:

                                try:

                                    remote_address = str(ipaddress.ip_interface(remote_ip_cidr).ip)

                                except Exception:

                                    # fallback: strip mask if simple CIDR string

                                    remote_address = remote_ip_cidr.split('/')[0] if '/' in remote_ip_cidr else remote_ip_cidr

                        if not remote_address:

                            remote_address = self._derive_peer_ip(ip_value)

                        target_name = name_by_device.get(target_device, '')

                        if target_name:

                            slug = ''.join(c.lower() if c.isalnum() else '-' for c in target_name).strip('-')

                            if slug:

                                session_name = f'to-{slug}'

                    else:

                        # No link_to_device specified; use explicit interface data or derive

                        remote_address = str(interface_data.get('remote_address', '')) or self._derive_peer_ip(ip_value)

                        remote_as = str(interface_data.get('remote_as', ''))

            else:

                hint_interfaces.append({

                    'interface': str(interface_name),

                    'address': str(interface_data),

                    'comment': '',

                })

        hints: Dict[str, Any] = {}

        if hint_interfaces:

            hints['interfaces'] = hint_interfaces

        if underlay_ip:

            hints['underlay_ip'] = underlay_ip

        if remote_address:

            hints['remote_address'] = remote_address

        if remote_as:

            hints['remote_as'] = remote_as

        if session_name:

            hints['bgp_session_name'] = session_name

        return hints

    def _process_spine_leaf(self) -> None:

        if len(self.__spine_leaf) == 0:

            return

                    continue

                try:

                    method(SetDeviceRoleRequest(device_uuid=device_uuid, fabric_id=fabric_id))

                    request_kwargs = {

                        'device_uuid': device_uuid,

                        'fabric_id': fabric_id,

                    }

                    hints = self._build_spine_leaf_hints(fabric_name, device_uuid)

                    if hints:

                        request_kwargs['role'] = json.dumps(hints)

                    method(SetDeviceRoleRequest(**request_kwargs))

                    num_ok += 1

                except Exception as e:  # pylint: disable=broad-except

                    error_list.append(f'{fabric_id}/{device_identifier}: {str(e)}')

        self.__results.append(('spine_leaf', 'config', num_ok, error_list))

    @staticmethod

    def worker(grpc_method, grpc_class, entity) -> Any:

        return grpc_method(grpc_class(**entity))

import json

import ipaddress

from typing import List, Tuple

class FabricConfigBuilder:

    def __init__(self, state_store):

        self.store = state_store

    @staticmethod

    def _strip_prefix(address: str) -> str:

        if not address:

            return ''

        return address.split('/')[0]

    def _derive_p2p_peer(self, local_underlay: str) -> str:

        if not local_underlay:

            return ''

        try:

            iface = ipaddress.ip_interface(local_underlay)

            net = iface.network

            if net.num_addresses != 2:

                return ''

            for host in net.hosts():

                host_s = str(host)

                if host_s != str(iface.ip):

                    return host_s

        except ValueError:

            return ''

        return ''

    def build(self, fabric_id: str, device_id: str) -> List[Tuple[str, str]]:

        rules = []

        role = device["role"]

        asn  = device["asn"]

        lo   = device["loopback_ip"]

        try:

            asn_value = int(asn)

        except (TypeError, ValueError):

            asn_value = asn

        interface_rules = device.get('interface_ips', [])

        if interface_rules:

            for entry in interface_rules:

                address = str(entry.get('address', '') or '')

                if not address:

                    continue

                interface = str(entry.get('interface', '') or '')

                comment = str(entry.get('comment', '') or '')

                rules.append((

                    "/interfaces/ip",

                    json.dumps({

                        "address": address,

                        "interface": interface,

                        "comment": comment,

                    })

                ))

                if interface.lower() == 'lo' or address.endswith('/32'):

                    lo = self._strip_prefix(address)

        # ---- Loopback IP ----

        if not interface_rules:

            rules.append((

                "/interfaces/ip",

                json.dumps({

            "/network_instances/bgp_instance",

            json.dumps({

                "name": "default",

                "as": asn,

                "as": asn_value,

                "router_id": lo

            })

        ))

                ))

        # ---- BGP sessions ----

        # Prefer per-link underlay IP lists when available. If both local and remote expose

        # underlay_ips lists, pick the first element of each as the session addresses. Fall

        # back to single 'underlay_ip' or loopback as necessary.

        if role == "spine":

            leaves = inventory.get("leaves", [])

            for leaf in leaves:

                # prefer lists, then single value, then loopback

                local_underlays = device.get('underlay_ips', [])

                remote_underlays = spine.get('underlay_ips', [])

                if local_underlays and remote_underlays:

                    local_ip = local_underlays[0]

                    remote_ip = remote_underlays[0]

                    local_ip = self._strip_prefix(local_underlays[0])

                    remote_ip = self._strip_prefix(remote_underlays[0])

                else:

                    local_ip = device.get('underlay_ip') or (local_underlays[0] if local_underlays else None) or lo

                    remote_ip = spine.get('underlay_ip') or (remote_underlays[0] if remote_underlays else None) or spine.get('loopback_ip')

                    local_ip = self._strip_prefix(

                        device.get('underlay_ip') or (local_underlays[0] if local_underlays else None) or lo

                    )

                    remote_ip = self._strip_prefix(

                        spine.get('underlay_ip') or (remote_underlays[0] if remote_underlays else None) or spine.get('loopback_ip')

                    )

                rules.append(self._bgp_session(

                    name=spine["device_id"],

                    remote_as=spine["asn"]

                ))

            if not spines:

                local_underlay = device.get('underlay_ip') or (device.get('underlay_ips', [None])[0] or '')

                local_ip = self._strip_prefix(local_underlay) or lo

                remote_ip = device.get('bgp_remote_address') or self._derive_p2p_peer(local_underlay)

                remote_as = device.get('remote_as')

                session_name = device.get('bgp_session_name') or 'to-spine1'

                if remote_ip and remote_as:

                    rules.append(self._bgp_session(

                        name=session_name,

                        local_ip=local_ip,

                        remote_ip=remote_ip,

                        remote_as=remote_as,

                    ))

        # ---- EVPN (leaf only) ----

        if role in ("leaf", "gateway") and vni is not None:

            rules.append((

        reply = self.logical_resource_client.GetResourcesByType(ResourceTypeQuery(resource_type=resource_type))

        return self._pick_first_available(reply)

    def _infer_remote_as(self, local_asn: str, fabric_id: str) -> Optional[str]:

        if self.logical_resource_client is None:

            return None

        reply = self.logical_resource_client.GetResourcesByType(ResourceTypeQuery(resource_type='asn'))

        for entry in reply.resources:

            if entry.value == local_asn:

                continue

            if entry.allocated and entry.fabric_id == fabric_id:

                return entry.value

        for entry in reply.resources:

            if entry.value != local_asn:

                return entry.value

        return None

    @staticmethod

    def _normalize_loopback(loopback_ip: str) -> str:

        return loopback_ip.split('/')[0] if loopback_ip else ''

            profile['include_overlay'] = True

            profile['afi'] = 'ip,l2vpn'

        # Remove legacy IS-IS/OSPF automatic local_role mapping: use BGP underlay by default

        return profile

                'loopback_ip': config.ip_address,

                'underlay_ip': config.remote_address,  # Primary underlay IP for BGP sessions

                'underlay_ips': [config.remote_address] if config.remote_address else [],

                'remote_as': self._infer_remote_as(str(config.asn), fabric_id),

                'vlans': [str(config.vlan_tag)] if config.vlan_tag else [],

                'vnis': [str(config.vni)] if config.vni else [],

            }