p4_driver.py

# Copyright 2022-2024 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.

"""
P4 driver plugin for the TeraFlow SDN controller.
"""

import os
import json
import logging
import threading
from typing import Any, Iterator, List, Optional, Tuple, Union
from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method
from common.type_checkers.Checkers import chk_type, chk_length, chk_string
from device.service.driver_api._Driver import RESOURCE_ENDPOINTS, RESOURCE_RULES
from .p4_common import matches_ipv4, matches_ipv6, valid_port,\
    compose_resource_endpoints, parse_resource_string_from_json,\
    P4_ATTR_DEV_ID, P4_ATTR_DEV_NAME, P4_ATTR_DEV_ENDPOINTS,\
    P4_ATTR_DEV_VENDOR, P4_ATTR_DEV_HW_VER, P4_ATTR_DEV_SW_VER,\
    P4_ATTR_DEV_P4BIN, P4_ATTR_DEV_P4INFO, P4_ATTR_DEV_TIMEOUT,\
    P4_VAL_DEF_VENDOR, P4_VAL_DEF_HW_VER, P4_VAL_DEF_SW_VER,\
    P4_VAL_DEF_TIMEOUT
from .p4_manager import P4Manager, \
    KEY_TABLE, KEY_ACTION, KEY_ACTION_PROFILE, \
    KEY_COUNTER, KEY_DIR_COUNTER, KEY_METER, KEY_DIR_METER,\
    KEY_CTL_PKT_METADATA, KEY_DIGEST, KEY_CLONE_SESSION,\
    KEY_ENDPOINT
from .p4_client import WriteOperation

try:
    from _Driver import _Driver
except ImportError:
    from device.service.driver_api._Driver import _Driver

LOGGER = logging.getLogger(__name__)

DRIVER_NAME = 'p4'
METRICS_POOL = MetricsPool('Device', 'Driver', labels={'driver': DRIVER_NAME})

class P4Driver(_Driver):
    """
    P4Driver class inherits the abstract _Driver class to support P4 devices.

    Attributes
    ----------
    address : str
        IP address of the P4Runtime server running on the P4 device
    port : int
        transport port number of the P4Runtime server running on the P4 device
    **settings : map
        id : int
            P4 device datapath ID (Mandatory)
        name : str
            P4 device name (Optional)
        endpoints : list
            List of P4 device endpoints, i.e., ports (Optional)
        vendor : str
            P4 device vendor (Optional)
        hw_ver : str
            Hardware version of the P4 device (Optional)
        sw_ver : str
            Software version of the P4 device (Optional)
        p4bin : str
            Path to P4 binary file (Optional, but must be combined with p4info)
        p4info : str
            Path to P4 info file (Optional, but must be combined with p4bin)
        timeout : int
            P4 device timeout in seconds (Optional)
        rules : list
            List of rules to configure the P4 device's pipeline
    """

    def __init__(self, address: str, port: int, **settings) -> None:
        super().__init__(name=DRIVER_NAME, address=address, port=port, setting=settings)
        self.__manager = None
        self.__address = address
        self.__port = int(port)
        self.__grpc_endpoint = None
        self.__settings = settings
        self.__id = None
        self.__name = None
        self.__endpoints = []
        self.__rules = {}
        self.__vendor = P4_VAL_DEF_VENDOR
        self.__hw_version = P4_VAL_DEF_HW_VER
        self.__sw_version = P4_VAL_DEF_SW_VER
        self.__p4bin_path = None
        self.__p4info_path = None
        self.__timeout = P4_VAL_DEF_TIMEOUT
        self.__lock = threading.Lock()
        self.__started = threading.Event()
        self.__terminate = threading.Event()

        self.__parse_and_validate_settings()

        LOGGER.info("Initializing P4 device at %s:%d with settings:",
                    self.__address, self.__port)

        for key, value in settings.items():
            LOGGER.info("\t%9s = %s", key, value)

    def Connect(self) -> bool:
        """
        Establish a connection between the P4 device driver and a P4 device.

        :return: boolean connection status.
        """
        LOGGER.info("Connecting to P4 device %s ...", self.__grpc_endpoint)

        with self.__lock:
            # Skip if already connected
            if self.__started.is_set():
                return True

            # TODO: Dynamically devise an election ID
            election_id = (1, 0)

            # Spawn a P4 manager for this device
            self.__manager = P4Manager(
                device_id=self.__id,
                ip_address=self.__address,
                port=self.__port,
                election_id=election_id)
            assert self.__manager

            # Start the P4 manager
            try:
                self.__manager.start(self.__p4bin_path, self.__p4info_path)
            except Exception as ex:  # pylint: disable=broad-except
                raise Exception(ex) from ex

            LOGGER.info("\tConnected via P4Runtime")
            self.__started.set()

            return True

    def Disconnect(self) -> bool:
        """
        Terminate the connection between the P4 device driver and a P4 device.

        :return: boolean disconnection status.
        """
        LOGGER.info("Disconnecting from P4 device %s ...", self.__grpc_endpoint)

        # If not started, assume it is already disconnected
        if not self.__started.is_set():
            return True

        # P4 manager must already be instantiated
        assert self.__manager

        # Trigger termination of loops and processes
        self.__terminate.set()

        # Trigger connection tear down with the P4Runtime server
        self.__manager.stop()
        self.__manager = None

        LOGGER.info("\tDisconnected!")

        return True

    @metered_subclass_method(METRICS_POOL)
    def GetInitialConfig(self) -> List[Tuple[str, Any]]:
        """
        Retrieve the initial configuration of a P4 device.

        :return: list of initial configuration items.
        """

        resource_keys = [RESOURCE_ENDPOINTS] if self.__endpoints else []

        with self.__lock:
            if not resource_keys:
                LOGGER.warning("No initial configuration for P4 device {} ...".format(self.__grpc_endpoint))
                return []
            LOGGER.info("Initial configuration for P4 device {}:".format(self.__grpc_endpoint))
            return self.GetConfig(resource_keys)

    @metered_subclass_method(METRICS_POOL)
    def GetConfig(self, resource_keys: List[str] = [])\
            -> List[Tuple[str, Union[Any, None, Exception]]]:
        """
        Retrieve the current configuration of a P4 device.

        :param resource_keys: P4 resource keys to retrieve.
        :return: list of values associated with the requested resource keys or
        None/Exception.
        """
        LOGGER.info(
            "Getting configuration from P4 device %s ...", self.__grpc_endpoint)

        # No resource keys means fetch all configuration
        if len(resource_keys) == 0:
            LOGGER.warning(
                "GetConfig with no resource keys "
                "implies getting all resource keys!")
            resource_keys = [
                obj_name for obj_name, _ in self.__manager.p4_objects.items()
            ] + [RESOURCE_ENDPOINTS] + [RESOURCE_RULES]

        # Verify the input type
        chk_type("resources", resource_keys, list)

        with self.__lock:
            return self.__get_resources(resource_keys)

    @metered_subclass_method(METRICS_POOL)
    def SetConfig(self, resources: List[Tuple[str, Any]])\
            -> List[Union[bool, Exception]]:
        """
        Submit a new configuration to a P4 device.

        :param resources: P4 resources to set.
        :return: list of boolean results or Exceptions for resource key
        changes requested.
        """
        LOGGER.info(
            "Setting configuration to P4 device %s ...", self.__grpc_endpoint)

        if not resources or len(resources) == 0:
            LOGGER.warning(
                "SetConfig requires a list of resources to store "
                "into the device. Nothing is provided though.")
            return []

        assert isinstance(resources, list)

        with self.__lock:
            return self.__set_resources(resources)

    @metered_subclass_method(METRICS_POOL)
    def DeleteConfig(self, resources: List[Tuple[str, Any]])\
            -> List[Union[bool, Exception]]:
        """
        Revoke P4 device configuration.

        :param resources: list of tuples with resource keys to be deleted.
        :return: list of boolean results or Exceptions for resource key
        deletions requested.
        """
        LOGGER.info(
            "Deleting configuration from P4 device %s ...", self.__grpc_endpoint)

        if not resources or len(resources) == 0:
            LOGGER.warning(
                "DeleteConfig requires a list of resources to delete "
                "from the device. Nothing is provided though.")
            return []

        with self.__lock:
            return self.__delete_resources(resources)

    def GetResource(self, endpoint_uuid: str) -> Optional[str]:
        """
        Retrieve a certain resource from a P4 device.

        :param endpoint_uuid: target endpoint UUID.
        :return: The path of the endpoint or None if not found.
        """
        LOGGER.warning("GetResource() RPC not yet implemented by the P4 driver")
        return ""

    def GetState(self,
                 blocking=False,
                 terminate: Optional[threading.Event] = None) -> \
                 Iterator[Tuple[str, Any]]:
        """
        Retrieve the state of a P4 device.

        :param blocking: if non-blocking, the driver terminates the loop and
        returns.
        :param terminate: termination flag.
        :return: sequences of state sample.
        """
        LOGGER.warning("GetState() RPC not yet implemented by the P4 driver")
        return []

    @metered_subclass_method(METRICS_POOL)
    def SubscribeState(self, subscriptions: List[Tuple[str, float, float]])\
            -> List[Union[bool, Exception]]:
        """
        Subscribe to certain state information.

        :param subscriptions: list of tuples with resources to be subscribed.
        :return: list of results for resource subscriptions requested.
        """
        LOGGER.warning(
            "SubscribeState() RPC not yet implemented by the P4 driver")
        return [False for _ in subscriptions]

    @metered_subclass_method(METRICS_POOL)
    def UnsubscribeState(self, subscriptions: List[Tuple[str, float, float]])\
            -> List[Union[bool, Exception]]:
        """
        Unsubscribe from certain state information.

        :param subscriptions: list of tuples with resources to be unsubscribed.
        :return: list of results for resource un-subscriptions requested.
        """
        LOGGER.warning(
            "UnsubscribeState() RPC not yet implemented by the P4 driver")
        return [False for _ in subscriptions]

    def get_manager(self):
        """
        Get an instance of the P4 manager.

        :return: P4 manager instance
        """
        return self.__manager

    def is_started(self):
        """
        Check if an instance of the P4 manager is started.

        :return: boolean P4 manager instance status
        """
        return self.__started.is_set()

    def __parse_and_validate_settings(self):
        """
        Verify that the driver inputs comply to what is expected.

        :return: void or exception in case of validation error
        """
        # Device endpoint information
        assert matches_ipv4(self.__address) or (matches_ipv6(self.__address)),\
            f"{self.__address} not a valid IPv4 or IPv6 address"
        assert valid_port(self.__port), \
            f"{self.__port} not a valid transport port"
        self.__grpc_endpoint = f"{self.__address}:{self.__port}"

        # Device ID
        try:
            self.__id = self.__settings.get(P4_ATTR_DEV_ID)
        except Exception as ex:
            LOGGER.error("P4 device ID is a mandatory setting")
            raise Exception from ex

        # Device name
        if P4_ATTR_DEV_NAME in self.__settings:
            self.__name = self.__settings.get(P4_ATTR_DEV_NAME)
        else:
            self.__name = str(self.__id)
            LOGGER.warning(
                "No device name is provided. Setting default name: %s",
                self.__name)

        # Device endpoints
        if P4_ATTR_DEV_ENDPOINTS in self.__settings:
            endpoints = self.__settings.get(P4_ATTR_DEV_ENDPOINTS, [])
            endpoint_resources = compose_resource_endpoints(endpoints)
            if endpoint_resources:
                LOGGER.info("Setting endpoints: {}".format(endpoint_resources))
                self.SetConfig(endpoint_resources)
        else:
            LOGGER.warning("No device endpoints are provided.")

        # Device vendor
        if P4_ATTR_DEV_VENDOR in self.__settings:
            self.__vendor = self.__settings.get(P4_ATTR_DEV_VENDOR)
        else:
            LOGGER.warning(
                "No device vendor is provided. Setting default vendor: %s",
                self.__vendor)

        # Device hardware version
        if P4_ATTR_DEV_HW_VER in self.__settings:
            self.__hw_version = self.__settings.get(P4_ATTR_DEV_HW_VER)
        else:
            LOGGER.warning(
                "No HW version is provided. Setting default HW version: %s",
                self.__hw_version)

        # Device software version
        if P4_ATTR_DEV_SW_VER in self.__settings:
            self.__sw_version = self.__settings.get(P4_ATTR_DEV_SW_VER)
        else:
            LOGGER.warning(
                "No SW version is provided. Setting default SW version: %s",
                self.__sw_version)

        # Path to P4 binary file
        if P4_ATTR_DEV_P4BIN in self.__settings:
            self.__p4bin_path = self.__settings.get(P4_ATTR_DEV_P4BIN)
            assert os.path.exists(self.__p4bin_path),\
                "Invalid path to p4bin file: {}".format(self.__p4bin_path)
            assert P4_ATTR_DEV_P4INFO in self.__settings,\
                "p4info and p4bin settings must be provided together"

        # Path to P4 info file
        if P4_ATTR_DEV_P4INFO in self.__settings:
            self.__p4info_path = self.__settings.get(P4_ATTR_DEV_P4INFO)
            assert os.path.exists(self.__p4info_path),\
                "Invalid path to p4info file: {}".format(self.__p4info_path)
            assert P4_ATTR_DEV_P4BIN in self.__settings,\
                "p4info and p4bin settings must be provided together"

        if (not self.__p4bin_path) or (not self.__p4info_path):
            LOGGER.warning(
                "No P4 binary and info files are provided, hence "
                "no pipeline will be installed on the whitebox device.\n"
                "This driver will attempt to manage whatever pipeline "
                "is available on the target device.")

        # Device timeout
        if P4_ATTR_DEV_TIMEOUT in self.__settings:
            self.__timeout = self.__settings.get(P4_ATTR_DEV_TIMEOUT)
            assert self.__timeout > 0,\
                "Device timeout must be a positive integer"
        else:
            LOGGER.warning(
                "No device timeout is provided. Setting default timeout: %s",
                self.__timeout)

    def __get_resources(self, resource_keys):
        """
        Retrieve the current configuration of a P4 device.

        :param resource_keys: P4 resource keys to retrieve.
        :return: list of values associated with the requested resource keys or
        None/Exception.
        """
        resources = []

        LOGGER.info("GetConfig() -> Keys: {}".format(resource_keys))

        for resource_key in resource_keys:
            entries = []
            try:
                if KEY_TABLE == resource_key:
                    for table_name in self.__manager.get_table_names():
                        t_entries = self.__manager.table_entries_to_json(
                            table_name)
                        if t_entries:
                            entries.append(t_entries)
                elif KEY_COUNTER == resource_key:
                    for cnt_name in self.__manager.get_counter_names():
                        c_entries = self.__manager.counter_entries_to_json(
                            cnt_name)
                        if c_entries:
                            entries.append(c_entries)
                elif KEY_DIR_COUNTER == resource_key:
                    for d_cnt_name in self.__manager.get_direct_counter_names():
                        dc_entries = self.__manager.direct_counter_entries_to_json(
                                d_cnt_name)
                        if dc_entries:
                            entries.append(dc_entries)
                elif KEY_METER == resource_key:
                    for meter_name in self.__manager.get_meter_names():
                        m_entries = self.__manager.meter_entries_to_json(
                            meter_name)
                        if m_entries:
                            entries.append(m_entries)
                elif KEY_DIR_METER == resource_key:
                    for d_meter_name in self.__manager.get_direct_meter_names():
                        dm_entries = self.__manager.direct_meter_entries_to_json(
                                d_meter_name)
                        if dm_entries:
                            entries.append(dm_entries)
                elif KEY_ACTION_PROFILE == resource_key:
                    for ap_name in self.__manager.get_action_profile_names():
                        ap_entries = self.__manager.action_prof_member_entries_to_json(
                                ap_name)
                        if ap_entries:
                            entries.append(ap_entries)
                elif KEY_ACTION == resource_key:
                    # To be implemented or deprecated
                    pass
                elif KEY_CTL_PKT_METADATA == resource_key:
                    #TODO: Handle controller packet metadata
                    msg = f"{resource_key.capitalize()} is not a " \
                          f"retrievable resource"
                    LOGGER.warning("%s", msg)
                elif KEY_DIGEST == resource_key:
                    #TODO: Handle digests
                    msg = f"{resource_key.capitalize()} is not a " \
                          f"retrievable resource"
                    LOGGER.warning("%s", msg)
                elif RESOURCE_ENDPOINTS == resource_key:
                    resources += self.__endpoints
                    continue
                elif RESOURCE_RULES == resource_key:
                     resources = self.__rules_into_resources()
                     continue
                else:
                    msg = f"GetConfig failed due to invalid " \
                          f"resource key: {resource_key}"
                    raise Exception(msg)
                resources.append(
                    (resource_key, entries if entries else None)
                )
            except Exception as e:  # pylint: disable=broad-except
                resources.append((resource_key, e))

        LOGGER.info("GetConfig() -> Results: %s", resources)

        return resources

    def __set_resources(self, resources):
        """
        Submit a new configuration to a P4 device.

        :param resources: P4 resources to set.
        :return: list of boolean results or Exceptions for resource key
        changes requested.
        """
        results = []

        LOGGER.info("SetConfig -> Resources {}".format(resources))

        for i, resource in enumerate(resources):
            str_resource_name = f"resources[#{i}]"
            resource_key = ""
            try:
                chk_type(
                    str_resource_name, resource, (list, tuple))
                chk_length(
                    str_resource_name, resource, min_length=2, max_length=2)
                resource_key, resource_value = resource
                chk_string(
                    str_resource_name, resource_key, allow_empty=False)
            except Exception as e:  # pylint: disable=broad-except
                LOGGER.exception(
                    "Exception validating %s: %s",
                    str_resource_name, str(resource_key))
                results.append(e)  # store the exception if validation fails
                continue

            try:
                # Rules are JSON-based, endpoints are not
                if "endpoint" not in resource_key:
                    resource_value = json.loads(resource_value)
            except Exception as e:  # pylint: disable=broad-except
                LOGGER.exception("Exception validating resource value {}".format(resource_value))
                results.append(e)
                continue

            LOGGER.info(
                "SetConfig() -> Key: %s - Value: %s",
                resource_key, resource_value)

            # Default operation is insert.
            # P4 manager has internal logic to judge whether an entry
            # to be inserted already exists, thus simply needs an update.
            operation = WriteOperation.insert

            # Dataplane and cache rule insertion process
            try:
                r2, r3 = False, True
                r1 = self.__cache_rule_insert(resource_key, resource_value, operation)
                # Cache insertion succeeded, proceed to dataplane
                if r1:
                    r2 = self.__apply_operation(resource_key, resource_value, operation)
                # Dataplane insertion did not succeed --> Revert caching
                if not r2 and r1:
                    r3 = self.__cache_rule_remove(resource_key)
                results.append(r1 & r2 & r3)
            except Exception as e:  # pylint: disable=broad-except
                results.append(e)
                continue

        LOGGER.info("SetConfig() -> Results: {}".format(results))

        return results

    def __delete_resources(self, resources):
        """
        Revoke P4 device configuration.

        :param resources: list of tuples with resource keys to be deleted.
        :return: list of boolean results or Exceptions for resource key
        deletions requested.
        """
        results = []

        for i, resource in enumerate(resources):
            str_resource_name = f"resources[#{i}]"
            resource_key = ""
            try:
                chk_type(
                    str_resource_name, resource, (list, tuple))
                chk_length(
                    str_resource_name, resource, min_length=2, max_length=2)
                resource_key, resource_value = resource
                chk_string(
                    str_resource_name, resource_key, allow_empty=False)
            except Exception as e:  # pylint: disable=broad-except
                LOGGER.exception(
                    "Exception validating %s: %s",
                    str_resource_name, str(resource_key))
                results.append(e)  # store the exception if validation fails
                continue

            try:
                resource_value = json.loads(resource_value)
            except Exception as e:  # pylint: disable=broad-except
                results.append(e)
                continue

            LOGGER.info("DeleteConfig() -> Key: %s - Value: %s",
                         resource_key, resource_value)

            operation = WriteOperation.delete

            # Dataplane and cache rule removal process
            try:
                r2, r3 = False, True
                r1 = self.__cache_rule_remove(resource_key)
                # Cache removal succeeded, proceed to dataplane
                if r1:
                    r2 = self.__apply_operation(resource_key, resource_value, operation)
                # Dataplane removal did not succeed --> Revert caching
                if not r2 and r1:
                    r3 = self.__cache_rule_insert(resource_key, resource_value, WriteOperation.insert)
                results.append(r1 & r2 & r3)
            except Exception as e:  # pylint: disable=broad-except
                results.append(e)
                continue

        LOGGER.info("DeleteConfig() -> Results: {}".format(results))

        return results

    def __apply_operation(
            self, resource_key, resource_value, operation: WriteOperation):
        """
        Apply a write operation to a P4 resource.

        :param resource_key: P4 resource key
        :param resource_value: P4 resource value in JSON format
        :param operation: write operation (i.e., insert, update, delete)
        to apply
        :return: True if operation is successfully applied or raise Exception
        """

        # Apply settings to the various tables
        if KEY_TABLE in resource_key:
            self.__manager.table_entry_operation_from_json(
                resource_value, operation)
        elif KEY_COUNTER in resource_key:
            self.__manager.counter_entry_operation_from_json(
                resource_value, operation)
        elif KEY_DIR_COUNTER in resource_key:
            self.__manager.direct_counter_entry_operation_from_json(
                resource_value, operation)
        elif KEY_METER in resource_key:
            self.__manager.meter_entry_operation_from_json(
                resource_value, operation)
        elif KEY_DIR_METER in resource_key:
            self.__manager.direct_meter_entry_operation_from_json(
                resource_value, operation)
        elif KEY_ACTION_PROFILE in resource_key:
            self.__manager.action_prof_member_entry_operation_from_json(
                resource_value, operation)
            self.__manager.action_prof_group_entry_operation_from_json(
                resource_value, operation)
        elif KEY_CLONE_SESSION in resource_key:
            self.__manager.clone_session_entry_operation_from_json(
                resource_value, operation)
        elif KEY_CTL_PKT_METADATA in resource_key:
            msg = f"{resource_key.capitalize()} is not a " \
                  f"configurable resource"
            raise Exception(msg)
        elif KEY_DIGEST in resource_key:
            msg = f"{resource_key.capitalize()} is not a " \
                  f"configurable resource"
            raise Exception(msg)
        elif KEY_ENDPOINT in resource_key:
            self.__endpoints.append((resource_key, resource_value))
        else:
            msg = f"{operation} on invalid key {resource_key}"
            LOGGER.error(msg)
            raise Exception(msg)

        return True

    def __cache_rule_insert(self, resource_key, resource_value, operation):
        """
        Insert a new rule into the rule cache or update an existing one.

        :param resource_key: P4 resource key
        :param resource_value: P4 resource value in JSON format
        :param operation: write operation (i.e., insert, update) to apply
        :return: True if new rule is inserted or existing is updated, otherwise False
        """
        if (resource_key in self.__rules.keys()) and (operation == WriteOperation.insert):
            LOGGER.error("Attempting to insert an existing rule key: {}".format(resource_key))
            return False
        elif (resource_key not in self.__rules.keys()) and (operation == WriteOperation.update):
            LOGGER.error("Attempting to update a non-existing rule key: {}".format(resource_key))
            return False
        elif (resource_key in self.__rules.keys()) and (operation == WriteOperation.update):
            LOGGER.warning("Updating an existing rule key: {}".format(resource_key))
        self.__rules[resource_key] = resource_value
        return True

    def __cache_rule_remove(self, resource_key):
        """
        Remove an existing rule from the rule cache.

        :param resource_key: P4 resource key
        :return: True if existing rule is removed, otherwise False
        """
        if resource_key not in self.__rules.keys():
            LOGGER.error("Attempting to remove a non-existing rule key: {}".format(resource_key))
            return False
        self.__rules.pop(resource_key)
        return True

    def __rules_into_resources(self):
        """
        Transform rules from the driver's rule map into
        resources exposed through the SBI API.
        """
        resource_list = []
        for rule_key, rule_val in self.__rules.items():
            resource_list.append((rule_key, rule_val))
        return resource_list