# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (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. from typing import Any, Optional from flask import redirect, render_template, Blueprint, flash, url_for from common.proto.context_pb2 import Empty from common.proto.load_generator_pb2 import Parameters, RequestTypeEnum from load_generator.client.LoadGeneratorClient import LoadGeneratorClient from load_generator.tools.ListScalarRange import ( list_scalar_range__grpc_to_str, list_scalar_range__list_to_grpc, parse_list_scalar_range) from .forms import LoadGenForm load_gen = Blueprint('load_gen', __name__, url_prefix='/load_gen') def set_properties(field, data : Any, readonly : Optional[bool] = None, disabled : Optional[bool] = None) -> None: if not hasattr(field, 'render_kw'): field.render_kw = dict() elif field.render_kw is None: field.render_kw = dict() if readonly is not None: field.render_kw['readonly'] = readonly if disabled is not None: field.render_kw['disabled'] = disabled if (readonly is not None and readonly) or (disabled is not None and disabled): field.data = data @load_gen.route('home', methods=['GET']) def home(): load_gen_client = LoadGeneratorClient() load_gen_client.connect() status = load_gen_client.GetStatus(Empty()) load_gen_client.close() request_types = status.parameters.request_types _request_type_service_l2nm = RequestTypeEnum.REQUESTTYPE_SERVICE_L2NM in request_types _request_type_service_l3nm = RequestTypeEnum.REQUESTTYPE_SERVICE_L3NM in request_types _request_type_service_mw = RequestTypeEnum.REQUESTTYPE_SERVICE_MW in request_types _request_type_service_tapi = RequestTypeEnum.REQUESTTYPE_SERVICE_TAPI in request_types _request_type_slice_l2nm = RequestTypeEnum.REQUESTTYPE_SLICE_L2NM in request_types _request_type_slice_l3nm = RequestTypeEnum.REQUESTTYPE_SLICE_L3NM in request_types _offered_load = round(status.parameters.offered_load , ndigits=4) _holding_time = round(status.parameters.holding_time , ndigits=4) _inter_arrival_time = round(status.parameters.inter_arrival_time , ndigits=4) _availability = list_scalar_range__grpc_to_str(status.parameters.availability ) _capacity_gbps = list_scalar_range__grpc_to_str(status.parameters.capacity_gbps ) _e2e_latency_ms = list_scalar_range__grpc_to_str(status.parameters.e2e_latency_ms) form = LoadGenForm() set_properties(form.num_requests , status.parameters.num_requests , readonly=status.running) set_properties(form.offered_load , _offered_load , readonly=status.running) set_properties(form.holding_time , _holding_time , readonly=status.running) set_properties(form.inter_arrival_time , _inter_arrival_time , readonly=status.running) set_properties(form.availability , _availability , readonly=status.running) set_properties(form.capacity_gbps , _capacity_gbps , readonly=status.running) set_properties(form.e2e_latency_ms , _e2e_latency_ms , readonly=status.running) set_properties(form.max_workers , status.parameters.max_workers , readonly=status.running) set_properties(form.do_teardown , status.parameters.do_teardown , disabled=status.running) set_properties(form.record_to_dlt , status.parameters.record_to_dlt, disabled=status.running) set_properties(form.dlt_domain_id , status.parameters.dlt_domain_id, readonly=status.running) set_properties(form.request_type_service_l2nm, _request_type_service_l2nm , disabled=status.running) set_properties(form.request_type_service_l3nm, _request_type_service_l3nm , disabled=status.running) set_properties(form.request_type_service_mw , _request_type_service_mw , disabled=status.running) set_properties(form.request_type_service_tapi, _request_type_service_tapi , disabled=status.running) set_properties(form.request_type_slice_l2nm , _request_type_slice_l2nm , disabled=status.running) set_properties(form.request_type_slice_l3nm , _request_type_slice_l3nm , disabled=status.running) set_properties(form.num_generated , status.num_generated , disabled=True) set_properties(form.infinite_loop , status.infinite_loop , disabled=True) set_properties(form.running , status.running , disabled=True) form.submit.label.text = 'Stop' if status.running else 'Start' form_action = url_for('load_gen.stop') if status.running else url_for('load_gen.start') return render_template('load_gen/home.html', form=form, form_action=form_action) @load_gen.route('start', methods=['POST']) def start(): form = LoadGenForm() if form.validate_on_submit(): try: _availability = parse_list_scalar_range(form.availability.data ) _capacity_gbps = parse_list_scalar_range(form.capacity_gbps.data ) _e2e_latency_ms = parse_list_scalar_range(form.e2e_latency_ms.data) load_gen_params = Parameters() load_gen_params.num_requests = form.num_requests.data load_gen_params.offered_load = form.offered_load.data load_gen_params.holding_time = form.holding_time.data load_gen_params.inter_arrival_time = form.inter_arrival_time.data load_gen_params.max_workers = form.max_workers.data load_gen_params.do_teardown = form.do_teardown.data load_gen_params.dry_mode = False load_gen_params.record_to_dlt = form.record_to_dlt.data load_gen_params.dlt_domain_id = form.dlt_domain_id.data list_scalar_range__list_to_grpc(_availability, load_gen_params.availability ) # pylint: disable=no-member list_scalar_range__list_to_grpc(_capacity_gbps, load_gen_params.capacity_gbps ) # pylint: disable=no-member list_scalar_range__list_to_grpc(_e2e_latency_ms, load_gen_params.e2e_latency_ms) # pylint: disable=no-member del load_gen_params.request_types[:] # pylint: disable=no-member request_types = list() if form.request_type_service_l2nm.data: request_types.append(RequestTypeEnum.REQUESTTYPE_SERVICE_L2NM) if form.request_type_service_l3nm.data: request_types.append(RequestTypeEnum.REQUESTTYPE_SERVICE_L3NM) if form.request_type_service_mw .data: request_types.append(RequestTypeEnum.REQUESTTYPE_SERVICE_MW ) if form.request_type_service_tapi.data: request_types.append(RequestTypeEnum.REQUESTTYPE_SERVICE_TAPI) if form.request_type_slice_l2nm .data: request_types.append(RequestTypeEnum.REQUESTTYPE_SLICE_L2NM ) if form.request_type_slice_l3nm .data: request_types.append(RequestTypeEnum.REQUESTTYPE_SLICE_L3NM ) load_gen_params.request_types.extend(request_types) # pylint: disable=no-member load_gen_client = LoadGeneratorClient() load_gen_client.connect() load_gen_client.Start(load_gen_params) load_gen_client.close() flash('Load Generator Started.', 'success') except Exception as e: # pylint: disable=broad-except flash('Problem starting Load Generator. {:s}'.format(str(e)), 'danger') return redirect(url_for('load_gen.home')) @load_gen.route('stop', methods=['POST']) def stop(): form = LoadGenForm() if form.validate_on_submit(): try: load_gen_client = LoadGeneratorClient() load_gen_client.connect() load_gen_client.Stop(Empty()) load_gen_client.close() flash('Load Generator Stopped.', 'success') except Exception as e: # pylint: disable=broad-except flash('Problem stopping Load Generator. {:s}'.format(str(e)), 'danger') return redirect(url_for('load_gen.home'))