# 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 pytz import queue import logging import uuid import json from telemetry.backend.drivers.emulated.EmulatedHelper import EmulatedDriverHelper from telemetry.backend.driver_api._Driver import _Driver from anytree import Node, Resolver from apscheduler.events import EVENT_JOB_ADDED, EVENT_JOB_REMOVED from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.jobstores.memory import MemoryJobStore from apscheduler.executors.pool import ThreadPoolExecutor from datetime import datetime, timedelta from typing import Any, Iterator, List, Tuple, Union, Optional from .SyntheticMetricsGenerator import SyntheticMetricsGenerator logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class EmulatedDriver(_Driver): """ EmulatedDriver is a class that simulates a network driver for testing purposes. It provides functionalities to manage configurations, state subscriptions, and synthetic data generation. """ def __init__(self, address: str, port: int, **settings): super().__init__('emulated_driver', address, port, **settings) self._initial_config = Node('root') # Tree structure for initial config self._running_config = Node('root') # Tree structure for running config self._subscriptions = Node('subscriptions') # Tree for state subscriptions self._resolver = Resolver() # For path resolution in tree structures self._out_samples = queue.Queue() # Queue to hold synthetic state samples self._synthetic_data = SyntheticMetricsGenerator(metric_queue=self._out_samples) # Placeholder for synthetic data generator self._scheduler = BackgroundScheduler(daemon=True) self._scheduler.configure( jobstores = {'default': MemoryJobStore()}, executors = {'default': ThreadPoolExecutor(max_workers=1)}, timezone = pytz.utc ) self._scheduler.add_listener(self._listener_job_added_to_subscription_tree, EVENT_JOB_ADDED) self._scheduler.add_listener(self._listener_job_removed_from_subscription_tree, EVENT_JOB_REMOVED) self._helper_methods = EmulatedDriverHelper() self.logger = logging.getLogger(__name__) self.connected = False # To track connection state self.logger.info("EmulatedDriver initialized") def Connect(self) -> bool: self.logger.info(f"Connecting to {self.address}:{self.port}") self.connected = True self._scheduler.start() self.logger.info(f"Successfully connected to {self.address}:{self.port}") return True def Disconnect(self) -> bool: self.logger.info(f"Disconnecting from {self.address}:{self.port}") if not self.connected: self.logger.warning("Driver is not connected. Nothing to disconnect.") return False self._scheduler.shutdown() self.connected = False self.logger.info(f"Successfully disconnected from {self.address}:{self.port}") return True def _require_connection(self): if not self.connected: raise RuntimeError("Driver is not connected. Please connect before performing operations.") def SubscribeState(self, subscriptions: List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]: self._require_connection() results = [] for resource_key, duration, interval in subscriptions: resource_key = self._helper_methods.validate_resource_key(resource_key) # Validate the endpoint name self.logger.info(f"1. Subscribing to {resource_key} with duration {duration}s and interval {interval}s") try: self._resolver.get(self._running_config, resource_key) # Verify if the resource key exists in the running configuration self.logger.info(f"Resource key {resource_key} exists in the configuration.") resource_value = json.loads(self._resolver.get(self._running_config, resource_key).value) if resource_value is not None: sample_type_ids = resource_value['sample_types'] self.logger.info(f"Sample type IDs for {resource_key}: {sample_type_ids}") if len(sample_type_ids) == 0: self.logger.warning(f"No sample types found for {resource_key}. Skipping subscription.") results.append(False) continue else: self.logger.warning(f"No sample types found for {resource_key}. Skipping subscription.") results.append(False) continue # Add the job to the scheduler job_id = f"{resource_key}-{uuid.uuid4()}" self._scheduler.add_job( self._generate_sample, 'interval', seconds=interval, args=[resource_key, sample_type_ids], id=job_id, replace_existing=True, end_date=datetime.now(pytz.utc) + timedelta(seconds=duration) ) self.logger.info(f"Job added to scheduler for resource key {resource_key} with duration {duration}s and interval {interval}s") results.append(True) except Exception as e: self.logger.error(f"Failed to verify resource key or add job: {e}") results.append(e) return results def UnsubscribeState(self, subscriptions: List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]: self._require_connection() results = [] for resource_key, _, _ in subscriptions: resource_key = self._helper_methods.validate_resource_key(resource_key) try: # Check if job exists job_ids = [job.id for job in self._scheduler.get_jobs() if resource_key in job.id] if not job_ids: self.logger.warning(f"No active jobs found for {resource_key}. It might have already terminated.") results.append(False) continue # Remove jobs for job_id in job_ids: self._scheduler.remove_job(job_id) self.logger.info(f"Unsubscribed from {resource_key} with job IDs: {job_ids}") results.append(True) except Exception as e: self.logger.exception(f"Failed to unsubscribe from {resource_key}") results.append(e) return results def GetState(self, blocking: bool = False, terminate: Optional[queue.Queue] = None) -> Iterator[Tuple[float, str, Any]]: self._require_connection() start_time = datetime.now(pytz.utc) duration = 10 # Duration of the subscription in seconds (as an example) while True: try: if terminate and not terminate.empty(): self.logger.info("Termination signal received, stopping GetState") break elapsed_time = (datetime.now(pytz.utc) - start_time).total_seconds() if elapsed_time >= duration: self.logger.info("Duration expired, stopping GetState") break sample = self._out_samples.get(block=blocking, timeout=1 if blocking else 0.1) self.logger.info(f"Retrieved state sample: {sample}") yield sample except queue.Empty: if not blocking: self.logger.info("No more samples in queue, exiting GetState") return None def _generate_sample(self, resource_key: str, sample_type_ids : List[int]): # Simulate generating a sample for the resource key self.logger.debug(f"Executing _generate_sample for resource: {resource_key}") sample = self._synthetic_data.generate_synthetic_data_point(resource_key, sample_type_ids) self._out_samples.put(sample) # ------------- Event Listeners (START)----------------- def _listener_job_removed_from_subscription_tree(self, event): if event.job_id: # Extract the resource key from the job ID resource_key = event.job_id.split('-')[0] resource_key = self._helper_methods.validate_resource_key(resource_key) # Remove the subscription from the tree try: subscription_path = resource_key.split('/') parent = self._subscriptions for part in subscription_path: parent = next((child for child in parent.children if child.name == part), None) if not parent: raise ValueError(f"Subscription path '{resource_key}' not found in tree.") if parent: parent.parent.children = tuple(child for child in parent.parent.children if child != parent) self.logger.warning(f"Automatically removed subscription from subscription_tree for {resource_key} after job termination by listener. Maybe due to timeout.") except Exception as e: self.logger.warning(f"Failed to remove subscription for {resource_key}: {e}") def _listener_job_added_to_subscription_tree(self, event): try: job_id = event.job_id if job_id: resource_key = job_id.split('-')[0] # Extract resource key from job ID resource_key = self._helper_methods.validate_resource_key(resource_key) subscription_path = resource_key.split('/') parent = self._subscriptions for part in subscription_path: node = next((child for child in parent.children if child.name == part), None) if not node: node = Node(part, parent=parent) parent = node parent.value = { "job_id": job_id } self.logger.info(f"Automatically added subscription for {resource_key} to the subscription_tree by listener.") except Exception as e: self.logger.exception("Failed to add subscription to the tree") # ------------- Event Listeners (END)----------------- #------------------------------------------------------------------------------------- # ------- The below methods are kept for debugging purposes (test-case) only --------- #------------------------------------------------------------------------------------- # This method can be commented but this will arise an error in the test-case (@pytest.fixture --> connected_configured_driver()). def SetConfig(self, resources: dict) -> List[Union[bool, Exception]]: # For debugging purposes. self._require_connection() results = [] # if not isinstance(resources, dict): # self.logger.error("Invalid configuration format: resources must be a dictionary.") # raise ValueError("Invalid configuration format. Must be a dictionary.") if 'config_rules' not in resources or not isinstance(resources['config_rules'], list): self.logger.error("Invalid configuration format: 'config_rules' key missing or not a list.") raise ValueError("Invalid configuration format. Must contain a 'config_rules' key with a list of rules.") for rule in resources['config_rules']: try: if 'action' not in rule or 'custom' not in rule: raise ValueError(f"Invalid rule format: {rule}") action = rule['action'] custom = rule['custom'] resource_key = custom.get('resource_key') resource_value = custom.get('resource_value') if not resource_key: raise ValueError(f"Resource key is missing in rule: {rule}") if resource_value is None: raise ValueError(f"Resource value is None for key: {resource_key}") if not resource_key: raise ValueError(f"Resource key is missing in rule: {rule}") if action == 1: # Set action resource_path = self._helper_methods._parse_resource_key(resource_key) # self.logger.info(f"1. Setting configuration for resource key {resource_key} and resource_path: {resource_path}") parent = self._running_config for part in resource_path[:-1]: if '[' in part and ']' in part: base, index = part.split('[', 1) index = index.rstrip(']') parent = self._helper_methods._find_or_create_node(index, self._helper_methods._find_or_create_node(base, parent)) # self.logger.info(f"2a. Creating node: {base}, {index}, {parent}") elif resource_path[-1] != 'settings': # self.logger.info(f"2b. Creating node: {part}") parent = self._helper_methods._find_or_create_node(part, parent) final_part = resource_path[-1] if final_part in ['address', 'port']: self._helper_methods._create_or_update_node(final_part, parent, resource_value) self.logger.info(f"Configured: {resource_key} = {resource_value}") if resource_key.startswith("_connect/settings"): parent = self._helper_methods._find_or_create_node("_connect", self._running_config) settings_node = self._helper_methods._find_or_create_node("settings", parent) settings_node.value = None # Ensure settings node has None value endpoints_node = self._helper_methods._find_or_create_node("endpoints", settings_node) for endpoint in resource_value.get("endpoints", []): uuid = endpoint.get("uuid") uuid = uuid.replace('/', '_') if uuid else None if uuid: # self.logger.info(f"3. Creating endpoint: {uuid}, {endpoint}, {endpoints_node}") self._helper_methods._create_or_update_node(uuid, endpoints_node, endpoint) self.logger.info(f"Configured endpoint: {uuid} : {endpoint}") elif resource_key.startswith("/interface"): interface_parent = self._helper_methods._find_or_create_node("interface", self._running_config) name = resource_value.get("name") name = name.replace('/', '_') if name else None if name: self._helper_methods._create_or_update_node(name, interface_parent, resource_value) self.logger.info(f"Configured interface: {name} : {resource_value}") # self.logger.info(f"4. Configured interface: {name}") results.append(True) else: raise ValueError(f"Unsupported action '{action}' in rule: {rule}") if resource_value is None: raise ValueError(f"Resource value is None for key: {resource_key}") except Exception as e: self.logger.exception(f"Failed to apply rule: {rule}") results.append(e) return results #----------------------------------- # ------- EXTRA Methods ------------ #----------------------------------- # def log_active_jobs(self): # For debugging purposes. # """ # Logs the IDs of all active jobs. # This method retrieves the list of active jobs from the scheduler and logs their IDs using the logger. # """ # self._require_connection() # jobs = self._scheduler.get_jobs() # self.logger.info(f"Active jobs: {[job.id for job in jobs]}") # def print_config_tree(self): # For debugging purposes. # """ # Reads the configuration using GetConfig and prints it as a hierarchical tree structure. # """ # self._require_connection() # def print_tree(node, indent=""): # """ # Recursively prints the configuration tree. # Args: # node (Node): The current node to print. # indent (str): The current indentation level. # """ # if node.name != "root": # Skip the root node's name # value = getattr(node, "value", None) # print(f"{indent}- {node.name}: {json.loads(value) if value else ''}") # for child in node.children: # print_tree(child, indent + " ") # print("Configuration Tree:") # print_tree(self._running_config) # def GetInitialConfig(self) -> List[Tuple[str, Any]]: # comment # self._require_connection() # results = [] # for node in self._initial_config.descendants: # value = getattr(node, "value", None) # results.append((node.name, json.loads(value) if value else None)) # self.logger.info("Retrieved initial configurations") # return results # def GetConfig(self, resource_keys: List[str] = []) -> List[Tuple[str, Union[Any, dict, Exception]]]: # comment # """ # Retrieves the configuration for the specified resource keys. # If no keys are provided, returns the full configuration tree. # Args: # resource_keys (List[str]): A list of keys specifying the configuration to retrieve. # Returns: # List[Tuple[str, Union[Any, dict, Exception]]]: A list of tuples with the resource key and its value, # subtree, or an exception. # """ # self._require_connection() # results = [] # try: # if not resource_keys: # # If no specific keys are provided, return the full configuration tree # full_tree = self._helper_methods._generate_subtree(self._running_config) # # full_tree = self._generate_subtree(self._running_config) # return [("full_configuration", full_tree)] # for key in resource_keys: # try: # # Parse the resource key # resource_path = self._helper_methods.(key) # self.logger.info(f"1. Retrieving configuration for resource path : {resource_path}") # # Navigate to the node corresponding to the key # parent = self._running_config # for part in resource_path: # parent = self._find_or_raise_node(part, parent) # # Check if the node has a value # value = getattr(parent, "value", None) # if value: # # If a value exists, return it # results.append((key, json.loads(value))) # else: # # If no value, return the subtree of this node # subtree = self._helper_methods._generate_subtree(parent) # # subtree = self._generate_subtree(parent) # results.append((key, subtree)) # except Exception as e: # self.logger.exception(f"Failed to retrieve configuration for key: {key}") # results.append((key, e)) # except Exception as e: # self.logger.exception("Failed to retrieve configurations") # results.append(("Error", e)) # return results # def DeleteConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]: # comment # self._require_connection() # results = [] # for key in resources: # try: # # Parse resource key into parts, handling brackets correctly # resource_path = self._helper_methods.(key) # parent = self._running_config # for part in resource_path: # parent = self._find_or_raise_node(part, parent) # # Delete the final node # node_to_delete = parent # parent = node_to_delete.parent # parent.children = tuple(child for child in parent.children if child != node_to_delete) # self.logger.info(f"Deleted configuration for key: {key}") # # Handle endpoints structure # if "interface" in key and "settings" in key: # interface_name = key.split('[')[-1].split(']')[0] # endpoints_parent = self._find_or_raise_node("_connect", self._running_config) # endpoints_node = self._find_or_raise_node("endpoints", endpoints_parent) # endpoint_to_delete = next((child for child in endpoints_node.children if child.name == interface_name), None) # if endpoint_to_delete: # endpoints_node.children = tuple(child for child in endpoints_node.children if child != endpoint_to_delete) # self.logger.info(f"Removed endpoint entry for interface '{interface_name}'") # # Check if parent has no more children and is not the root # while parent and parent.name != "root" and not parent.children: # node_to_delete = parent # parent = node_to_delete.parent # parent.children = tuple(child for child in parent.children if child != node_to_delete) # self.logger.info(f"Deleted empty parent node: {node_to_delete.name}") # results.append(True) # except Exception as e: # self.logger.exception(f"Failed to delete configuration for key: {key}") # results.append(e) # return results