# Copyright 2022-2024 ETSI OSG/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.
import json
import logging, requests, threading
from requests.auth import HTTPBasicAuth
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_string, chk_type
from device.service.driver_api._Driver import _Driver,RESOURCE_ENDPOINTS
from device.service.drivers.OpenFlow.TfsApiClient import TfsApiClient
from device.service.drivers.OpenFlow.Tools import find_key, get_switches, get_flows , add_flow , delete_flow , get_desc,get_port_desc, get_links_information,get_switches_information,del_flow_entry
LOGGER = logging.getLogger(__name__)
DRIVER_NAME = 'ryu'
METRICS_POOL = MetricsPool('Device', 'Driver', labels={'driver': DRIVER_NAME})
ALL_RESOURCE_KEYS = [
RESOURCE_ENDPOINTS,
]
class OpenFlowDriver(_Driver):
def __init__(self, address: str, port: int, **settings) -> None:
super().__init__(DRIVER_NAME, address, port, **settings)
self.__lock = threading.Lock()
self.__started = threading.Event()
self.__terminate = threading.Event()
username = self.settings.get('username')
password = self.settings.get('password')
self.__auth = HTTPBasicAuth(username, password) if username is not None and password is not None else None
scheme = self.settings.get('scheme', 'http')
self.__base_url = '{:s}://{:s}:{:d}'.format(scheme, self.address, int(self.port))
self.__timeout = int(self.settings.get('timeout', 120))
self.tac = TfsApiClient(self.address, int(self.port), scheme=scheme, username=username, password=password)
def Connect(self) -> bool:
url = f"{self.__base_url}"
with self.__lock:
try:
response = requests.get(url, timeout=self.__timeout, verify=False, auth=self.__auth)
response.raise_for_status()
except requests.exceptions.Timeout:
LOGGER.exception(f"Timeout connecting to {self.__base_url}")
return False
except requests.exceptions.RequestException as e:
LOGGER.exception(f"Exception connecting to {self.__base_url}: {e}")
return False
else:
self.__started.set()
return True
def Disconnect(self) -> bool:
with self.__lock:
self.__terminate.set()
return True
@metered_subclass_method(METRICS_POOL)
def GetInitialConfig(self) -> List[Tuple[str, Any]]:
with self.__lock:
return []
@metered_subclass_method(METRICS_POOL)
def GetConfig(self, resource_keys: List[str] = []) -> List[Tuple[str, Union[Any, None, Exception]]]:
chk_type('resources', resource_keys, list)
results = []
with self.__lock:
if len(resource_keys) == 0:resource_keys = ALL_RESOURCE_KEYS
LOGGER.info(f'resource_key:{ALL_RESOURCE_KEYS}')
for i, resource_key in enumerate(resource_keys):
str_resource_name = 'resource_key[#{:d}]'.format(i)
try:
chk_string(str_resource_name, resource_key, allow_empty=False)
if resource_key == RESOURCE_ENDPOINTS:
LOGGER.info(f'resource_key:{RESOURCE_ENDPOINTS}')
results.extend(self.tac.get_devices_endpoints())
except Exception as e:
LOGGER.exception('Unhandled error processing resource_key({:s})'.format(str(resource_key)))
results.append((resource_key, e))
return results
# @metered_subclass_method(METRICS_POOL)
# def GetConfig(self, resource_keys: List[str] = []) -> List[Tuple[str, Union[Any, None, Exception]]]:
# chk_type('resources', resource_keys, list)
# results = []
# with self.__lock:
# for key in resource_keys:
# try:
# if key.startswith('flows:'):
# dpid = key.split(':', 1)[1]
# flows = get_flows(self.__base_url, dpid, auth=self.__auth, timeout=self.__timeout)
# results.append((key, flows))
# elif key.startswith('description:'):
# dpid = key.split(':', 1)[1]
# desc = get_desc(self.__base_url, dpid, auth=self.__auth, timeout=self.__timeout)
# results.append((key, desc))
# elif key.startswith('switches'):
# switches = get_switches(self.__base_url, auth=self.__auth, timeout=self.__timeout)
# results.append((key, switches))
# elif key.startswith('port_description:'):
# dpid = key.split(':', 1)[1]
# desc = get_port_desc(self.__base_url,dpid, auth=self.__auth, timeout=self.__timeout)
# results.append((key, desc))
# elif key.startswith('switch_info'):
# sin = get_switches_information(self.__base_url, auth=self.__auth, timeout=self.__timeout)
# results.append((key, sin))
# elif key.startswith('links_info'):
# lin = get_links_information(self.__base_url, auth=self.__auth, timeout=self.__timeout)
# results.append((key, lin))
# else:
# results.append((key, None)) # If key not handled, append None
# except Exception as e:
# results.append((key, e))
# return results
#
# @metered_subclass_method(METRICS_POOL)
# def DeleteConfig(self, resource_keys: List[str] = []) -> List[Tuple[str, Union[Any, None, Exception]]]:
# chk_type('resources', resource_keys, list)
# results = []
# with self.__lock:
# for item in resource_keys:
# try:
# if isinstance(item, tuple):
# key, data = item
# else:
# key, data = item, None
# if key.startswith('flowentry_delete:'):
# dpid = key.split(':', 1)[1]
# flows = del_flow_entry(self.__base_url, dpid, auth=self.__auth, timeout=self.__timeout)
# results.append((key, flows))
# elif key=='flow_data' and data:
# flow_del = delete_flow (self.__base_url,data,auth=self.__auth, timeout=self.__timeout)
# results.append((key, flow_del))
# else:
# results.append((key, None))
# except Exception as e:
# results.append((key, e))
# return results
#
@metered_subclass_method(METRICS_POOL)
def SetConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
results = []
LOGGER.info(f'SetConfig_resources:{resources}')
if not resources:
return results
return results
#
#
#
# @metered_subclass_method(METRICS_POOL)
# def SubscribeState(self, subscriptions : List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]:
# # TODO: TAPI does not support monitoring by now
# return [False for _ in subscriptions]
#
# @metered_subclass_method(METRICS_POOL)
# def UnsubscribeState(self, subscriptions : List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]:
# # TODO: TAPI does not support monitoring by now
# return [False for _ in subscriptions]
#
# def GetState(
# self, blocking=False, terminate : Optional[threading.Event] = None
# ) -> Iterator[Tuple[float, str, Any]]:
# # TODO: TAPI does not support monitoring by now
# return []