Skip to content
Snippets Groups Projects
Tools.py 12.3 KiB
Newer Older
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)
#
# 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
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
from typing import Any, Dict, List, Tuple, Union
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
from common.Constants import DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME
from common.method_wrappers.ServiceExceptions import InvalidArgumentException
from common.proto.context_pb2 import ConfigActionEnum, Device, DeviceConfig
from common.proto.device_pb2 import MonitoringSettings
from common.proto.kpi_sample_types_pb2 import KpiSampleType
from common.tools.grpc.Tools import grpc_message_to_json
from .driver_api._Driver import _Driver, RESOURCE_ENDPOINTS
from .monitoring.MonitoringLoops import MonitoringLoops
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
from .Errors import (
    ERROR_BAD_ENDPOINT, ERROR_DELETE, ERROR_GET, ERROR_GET_INIT, ERROR_MISSING_KPI, ERROR_SAMPLETYPE, ERROR_SET,
    ERROR_SUBSCRIBE, ERROR_UNSUBSCRIBE)
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

def check_connect_rules(device_config : DeviceConfig) -> Dict[str, Any]:
    connection_config_rules = dict()
    unexpected_config_rules = list()
    for config_rule in device_config.config_rules:
        is_action_set = (config_rule.action == ConfigActionEnum.CONFIGACTION_SET)
        is_custom_rule = (config_rule.WhichOneof('config_rule') == 'custom')
        if is_action_set and is_custom_rule and (config_rule.custom.resource_key.startswith('_connect/')):
            connect_attribute = config_rule.custom.resource_key.replace('_connect/', '')
            connection_config_rules[connect_attribute] = config_rule.custom.resource_value
        else:
            unexpected_config_rules.append(config_rule)

    if len(unexpected_config_rules) > 0:
        unexpected_config_rules = grpc_message_to_json(device_config)
        unexpected_config_rules = unexpected_config_rules['config_rules']
        unexpected_config_rules = list(filter(
            lambda cr: cr.get('custom', {})['resource_key'].replace('_connect/', '') not in connection_config_rules,
            unexpected_config_rules))
        str_unexpected_config_rules = json.dumps(unexpected_config_rules, sort_keys=True)
        raise InvalidArgumentException(
            'device.device_config.config_rules', str_unexpected_config_rules,
            extra_details='RPC method AddDevice only accepts connection Config Rules that should start '\
                            'with "_connect/" tag. Others should be configured after adding the device.')

    return connection_config_rules

def get_connect_rules(device_config : DeviceConfig) -> Dict[str, Any]:
    connect_rules = dict()
    for config_rule in device_config.config_rules:
        if config_rule.action != ConfigActionEnum.CONFIGACTION_SET: continue
        if config_rule.WhichOneof('config_rule') != 'custom': continue
        if not config_rule.custom.resource_key.startswith('_connect/'): continue
        connect_attribute = config_rule.custom.resource_key.replace('_connect/', '')
        connect_rules[connect_attribute] = config_rule.custom.resource_value
    return connect_rules

def check_no_endpoints(device_endpoints) -> None:
    if len(device_endpoints) == 0: return
    unexpected_endpoints = []
    for device_endpoint in device_endpoints:
        unexpected_endpoints.append(grpc_message_to_json(device_endpoint))
    str_unexpected_endpoints = json.dumps(unexpected_endpoints, sort_keys=True)
    raise InvalidArgumentException(
        'device.device_endpoints', str_unexpected_endpoints,
        extra_details='RPC method AddDevice does not accept Endpoints. Endpoints are discovered through '\
                        'interrogation of the physical device.')

def populate_endpoints(device : Device, driver : _Driver, monitoring_loops : MonitoringLoops) -> List[str]:
    device_uuid = device.device_id.device_uuid.uuid

    resources_to_get = [RESOURCE_ENDPOINTS]
    results_getconfig = driver.GetConfig(resources_to_get)

    errors : List[str] = list()
    for endpoint in results_getconfig:
        if len(endpoint) != 2:
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
            errors.append(ERROR_BAD_ENDPOINT.format(device_uuid, str(endpoint)))
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
            continue

        resource_key, resource_value = endpoint
        if isinstance(resource_value, Exception):
            errors.append(ERROR_GET.format(device_uuid, str(resource_key), str(resource_value)))
            continue

        endpoint_uuid = resource_value.get('uuid')

        device_endpoint = device.device_endpoints.add()
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
        device_endpoint.endpoint_id.topology_id.context_id.context_uuid.uuid = DEFAULT_CONTEXT_NAME
        device_endpoint.endpoint_id.topology_id.topology_uuid.uuid = DEFAULT_TOPOLOGY_NAME
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
        device_endpoint.endpoint_id.device_id.device_uuid.uuid = device_uuid
        device_endpoint.endpoint_id.endpoint_uuid.uuid = endpoint_uuid
        device_endpoint.endpoint_type = resource_value.get('type')

        sample_types : Dict[int, str] = resource_value.get('sample_types', {})
        for kpi_sample_type, monitor_resource_key in sample_types.items():
            device_endpoint.kpi_sample_types.append(kpi_sample_type)
            monitoring_loops.add_resource_key(device_uuid, endpoint_uuid, kpi_sample_type, monitor_resource_key)

    return errors

Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
def _raw_config_rules_to_grpc(
    device_uuid : str, device_config : DeviceConfig, error_template : str, default_config_action : ConfigActionEnum,
    raw_config_rules : List[Tuple[str, Union[Any, Exception, None]]]
) -> List[str]:
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    errors : List[str] = list()
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    for resource_key, resource_value in raw_config_rules:
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
        if isinstance(resource_value, Exception):
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
            errors.append(error_template.format(device_uuid, str(resource_key), str(resource_value)))
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
            continue

Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
        config_rule = device_config.config_rules.add()
        config_rule.action = default_config_action
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
        config_rule.custom.resource_key = resource_key
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
        config_rule.custom.resource_value = \
            resource_value if isinstance(resource_value, str) else json.dumps(resource_value, sort_keys=True)
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    return errors

Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
def populate_config_rules(device : Device, driver : _Driver) -> List[str]:
    device_uuid = device.device_id.device_uuid.uuid
    results_getconfig = driver.GetConfig()
    return _raw_config_rules_to_grpc(
        device_uuid, device.device_config, ERROR_GET, ConfigActionEnum.CONFIGACTION_SET, results_getconfig)

Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
def populate_initial_config_rules(device_uuid : str, device_config : DeviceConfig, driver : _Driver) -> List[str]:
    results_getinitconfig = driver.GetInitialConfig()
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    return _raw_config_rules_to_grpc(
        device_uuid, device_config, ERROR_GET_INIT, ConfigActionEnum.CONFIGACTION_SET, results_getinitconfig)
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

def compute_rules_to_add_delete(
    device : Device, request : Device
) -> Tuple[List[Tuple[str, Any]], List[Tuple[str, Any]]]:
    # convert config rules from context into a dictionary
    # TODO: add support for non-custom config rules
    context_config_rules = {
        config_rule.custom.resource_key: config_rule.custom.resource_value
        for config_rule in device.device_config.config_rules
        if config_rule.WhichOneof('config_rule') == 'custom'
    }

    # convert config rules from request into a list
    # TODO: add support for non-custom config rules
    request_config_rules = [
        (config_rule.action, config_rule.custom.resource_key, config_rule.custom.resource_value)
        for config_rule in request.device_config.config_rules
        if config_rule.WhichOneof('config_rule') == 'custom'
    ]

    resources_to_set    : List[Tuple[str, Any]] = [] # key, value
    resources_to_delete : List[Tuple[str, Any]] = [] # key, value

    for action, key, value in request_config_rules:
        if action == ConfigActionEnum.CONFIGACTION_SET:
            if (key in context_config_rules) and (context_config_rules[key][0] == value): continue
            resources_to_set.append((key, value))
        elif action == ConfigActionEnum.CONFIGACTION_DELETE:
            if key not in context_config_rules: continue
            resources_to_delete.append((key, value))

    return resources_to_set, resources_to_delete

def configure_rules(device : Device, driver : _Driver, resources_to_set : List[Tuple[str, Any]]) -> List[str]:
    device_uuid = device.device_id.device_uuid.uuid

    results_setconfig = driver.SetConfig(resources_to_set)
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    results_setconfig = [
        (resource_key, result if isinstance(result, Exception) else resource_value)
        for (resource_key, resource_value), result in zip(resources_to_set, results_setconfig)
    ]
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    device_config = DeviceConfig() # ignored; added at the end of ConfigureDevice
    return _raw_config_rules_to_grpc(
        device_uuid, device_config, ERROR_SET, ConfigActionEnum.CONFIGACTION_SET, results_setconfig)
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

def deconfigure_rules(device : Device, driver : _Driver, resources_to_delete : List[Tuple[str, Any]]) -> List[str]:
    device_uuid = device.device_id.device_uuid.uuid

    results_deleteconfig = driver.DeleteConfig(resources_to_delete)
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    results_deleteconfig = [
        (resource_key, result if isinstance(result, Exception) else resource_value)
        for (resource_key, resource_value), result in zip(resources_to_delete, results_deleteconfig)
    ]
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    device_config = DeviceConfig() # ignored; added at the end of ConfigureDevice
    return _raw_config_rules_to_grpc(
        device_uuid, device_config, ERROR_DELETE, ConfigActionEnum.CONFIGACTION_DELETE, results_deleteconfig)
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

def subscribe_kpi(request : MonitoringSettings, driver : _Driver, monitoring_loops : MonitoringLoops) -> List[str]:
    kpi_uuid = request.kpi_id.kpi_id.uuid
    device_uuid = request.kpi_descriptor.device_id.device_uuid.uuid
    endpoint_uuid = request.kpi_descriptor.endpoint_id.endpoint_uuid.uuid
    kpi_sample_type = request.kpi_descriptor.kpi_sample_type

    resource_key = monitoring_loops.get_resource_key(device_uuid, endpoint_uuid, kpi_sample_type)
    if resource_key is None:
        kpi_sample_type_name = KpiSampleType.Name(kpi_sample_type).upper().replace('KPISAMPLETYPE_', '')
        return [
            ERROR_SAMPLETYPE.format(
                str(device_uuid), str(endpoint_uuid), str(kpi_sample_type), str(kpi_sample_type_name)
            )
        ]

    sampling_duration = request.sampling_duration_s # seconds
    sampling_interval = request.sampling_interval_s # seconds

    resources_to_subscribe = [(resource_key, sampling_duration, sampling_interval)]
    results_subscribestate = driver.SubscribeState(resources_to_subscribe)

    errors : List[str] = list()
    for (resource_key, duration, interval), result in zip(resources_to_subscribe, results_subscribestate):
        if isinstance(result, Exception):
            errors.append(ERROR_SUBSCRIBE.format(
                str(device_uuid), str(resource_key), str(duration), str(interval), str(result)
            ))
            continue

    monitoring_loops.add_kpi(device_uuid, resource_key, kpi_uuid, sampling_duration, sampling_interval)
    monitoring_loops.add_device(device_uuid, driver)

    return errors

def unsubscribe_kpi(request : MonitoringSettings, driver : _Driver, monitoring_loops : MonitoringLoops) -> List[str]:
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    kpi_uuid = request.kpi_id.kpi_id.uuid    
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    kpi_details = monitoring_loops.get_kpi_by_uuid(kpi_uuid)
    if kpi_details is None:
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
        return [ERROR_MISSING_KPI.format(str(kpi_uuid))]
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    device_uuid, resource_key, sampling_duration, sampling_interval = kpi_details

    resources_to_unsubscribe = [(resource_key, sampling_duration, sampling_interval)]
    results_unsubscribestate = driver.UnsubscribeState(resources_to_unsubscribe)

    errors : List[str] = list()
    for (resource_key, duration, interval), result in zip(resources_to_unsubscribe, results_unsubscribestate):
        if isinstance(result, Exception):
            errors.append(ERROR_UNSUBSCRIBE.format(
                device_uuid, str(resource_key), str(duration), str(interval), str(result)))
            continue

    monitoring_loops.remove_kpi(kpi_uuid)
    #monitoring_loops.remove_device(device_uuid) # Do not remove; one monitoring_loop/device used by multiple requests

    return errors