diff --git a/manifests/nginx_ingress_http.yaml b/manifests/nginx_ingress_http.yaml
index ed713bf29ad8228ab3f5b051af24519c2fb9ef09..619d85f7a82af48c71464038bf14a833903d1a58 100644
--- a/manifests/nginx_ingress_http.yaml
+++ b/manifests/nginx_ingress_http.yaml
@@ -66,6 +66,6 @@ spec:
pathType: Prefix
backend:
service:
- name: qkd-appservice
+ name: nbiservice
port:
- number: 8005
+ number: 8080
diff --git a/manifests/qkd_appservice.yaml b/manifests/qkd_appservice.yaml
index 4f89d6c6f8400b509dc595f551e8f181e70b2f51..ba02e2e1da34ac670f68e17c28e9847603401d3a 100644
--- a/manifests/qkd_appservice.yaml
+++ b/manifests/qkd_appservice.yaml
@@ -28,36 +28,35 @@ spec:
spec:
terminationGracePeriodSeconds: 5
containers:
- - name: server
- image: labs.etsi.org:5050/tfs/controller/qkd_app:latest
- imagePullPolicy: Always
- ports:
- - containerPort: 10060
- - containerPort: 9192
- - containerPort: 8005
- env:
- - name: LOG_LEVEL
- value: "DEBUG"
- - name: CRDB_DATABASE_APP
- value: "qkd_app"
- envFrom:
- - secretRef:
- name: crdb-data
- - secretRef:
- name: nats-data
- readinessProbe:
- exec:
- command: ["/bin/grpc_health_probe", "-addr=:10060"]
- livenessProbe:
- exec:
- command: ["/bin/grpc_health_probe", "-addr=:10060"]
- resources:
- requests:
- cpu: 150m
- memory: 128Mi
- limits:
- cpu: 500m
- memory: 512Mi
+ - name: server
+ image: labs.etsi.org:5050/tfs/controller/qkd_app:latest
+ imagePullPolicy: Always
+ ports:
+ - containerPort: 10060
+ - containerPort: 9192
+ env:
+ - name: LOG_LEVEL
+ value: "INFO"
+ - name: CRDB_DATABASE
+ value: "qkd_app"
+ envFrom:
+ - secretRef:
+ name: crdb-data
+ - secretRef:
+ name: nats-data
+ readinessProbe:
+ exec:
+ command: ["/bin/grpc_health_probe", "-addr=:10060"]
+ livenessProbe:
+ exec:
+ command: ["/bin/grpc_health_probe", "-addr=:10060"]
+ resources:
+ requests:
+ cpu: 150m
+ memory: 128Mi
+ limits:
+ cpu: 500m
+ memory: 512Mi
---
apiVersion: v1
kind: Service
@@ -70,14 +69,11 @@ spec:
selector:
app: qkd-appservice
ports:
- - name: grpc
- protocol: TCP
- port: 10060
- targetPort: 10060
- - name: metrics
- protocol: TCP
- port: 9192
- targetPort: 9192
- - name: http
- port: 8005
- targetPort: 8005
+ - name: grpc
+ protocol: TCP
+ port: 10060
+ targetPort: 10060
+ - name: metrics
+ protocol: TCP
+ port: 9192
+ targetPort: 9192
diff --git a/proto/kpi_value_api.proto b/proto/kpi_value_api.proto
index 4d3a1f216406841344d40712ea04ec82cedf04d0..a97b0ae2b023dfe0a535aa3cb1ba63b00b418371 100644
--- a/proto/kpi_value_api.proto
+++ b/proto/kpi_value_api.proto
@@ -19,19 +19,19 @@ import "context.proto";
import "kpi_manager.proto";
service KpiValueAPIService {
- rpc StoreKpiValues (KpiValueList) returns (context.Empty) {}
- rpc SelectKpiValues (KpiValueFilter) returns (KpiValueList) {}
+ rpc StoreKpiValues (KpiValueList ) returns (context.Empty ) {}
+ rpc SelectKpiValues (KpiValueFilter ) returns (KpiValueList ) {}
rpc GetKpiAlarms (kpi_manager.KpiId) returns (stream KpiAlarms) {}
}
message KpiValue {
- kpi_manager.KpiId kpi_id = 1;
- context.Timestamp timestamp = 2;
- KpiValueType kpi_value_type = 3;
+ kpi_manager.KpiId kpi_id = 1;
+ context.Timestamp timestamp = 2;
+ KpiValueType kpi_value_type = 3;
}
message KpiValueList {
- repeated KpiValue kpi_value_list = 1;
+ repeated KpiValue kpi_value_list = 1;
}
message KpiValueType {
@@ -47,9 +47,9 @@ message KpiValueType {
}
message KpiValueFilter {
- repeated kpi_manager.KpiId kpi_id = 1;
- repeated context.Timestamp start_timestamp = 2;
- repeated context.Timestamp end_timestamp = 3;
+ repeated kpi_manager.KpiId kpi_id = 1;
+ repeated context.Timestamp start_timestamp = 2;
+ repeated context.Timestamp end_timestamp = 3;
}
message KpiAlarms {
diff --git a/proto/qkd_app.proto b/proto/qkd_app.proto
index 7b6c47330833849b889e770aac43844ec6e6072c..b22948d6cd3040146677d378a080283786d44f96 100644
--- a/proto/qkd_app.proto
+++ b/proto/qkd_app.proto
@@ -1,11 +1,26 @@
+// 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.
+
syntax = "proto3";
package qkd_app;
import "context.proto";
-// Optare: Change this if you want to change App's structure or enums.
-// Optare: If a message (structure) is changed it must be changed in src/app/service/database
+// Define Empty message if you don't want to use google.protobuf.Empty.
+message Empty {}
+// Enum representing possible states of a QKD application.
enum QKDAppStatusEnum {
QKDAPPSTATUS_ON = 0;
QKDAPPSTATUS_DISCONNECTED = 1;
@@ -13,16 +28,26 @@ enum QKDAppStatusEnum {
QKDAPPSTATUS_ZOMBIE = 3;
}
+// Enum representing QKD application types.
enum QKDAppTypesEnum {
QKDAPPTYPES_INTERNAL = 0;
QKDAPPTYPES_CLIENT = 1;
}
+// Message representing a QKDL (Quantum Key Distribution Link) identifier.
message QKDLId {
context.Uuid qkdl_uuid = 1;
}
+// Define QoS parameters for QKD applications
+message QoS {
+ uint32 max_bandwidth = 1; // Maximum bandwidth (in bits per second)
+ uint32 min_bandwidth = 2; // Minimum bandwidth (optional)
+ uint32 jitter = 3; // Maximum jitter (in milliseconds)
+ uint32 ttl = 4; // Time-to-live (in seconds)
+}
+// Main message representing a QKD application with all required fields.
message App {
AppId app_id = 1;
QKDAppStatusEnum app_status = 2;
@@ -32,22 +57,24 @@ message App {
repeated QKDLId backing_qkdl_id = 6;
context.DeviceId local_device_id = 7;
context.DeviceId remote_device_id = 8;
+ QoS qos = 9; // Include QoS in the App message
}
-
+// Message representing an identifier for an app.
message AppId {
context.ContextId context_id = 1;
context.Uuid app_uuid = 2;
}
-
+// Service definition for AppService, including app registration and listing.
service AppService {
rpc RegisterApp(App) returns (context.Empty) {}
- rpc ListApps (context.ContextId ) returns ( AppList ) {}
- }
-
-
+ rpc ListApps(context.ContextId) returns (AppList) {}
+ rpc GetApp(AppId) returns (App) {}
+ rpc DeleteApp (AppId) returns (Empty) {} // Use locally defined Empty
+}
- message AppList {
+// Message representing a list of apps.
+message AppList {
repeated App apps = 1;
}
diff --git a/scripts/run_tests_locally-telemetry-DB.sh b/scripts/run_tests_locally-telemetry-DB.sh
index f7c80dd86aa7503a7f42fe52db34167377ba0a37..b9f86a208aab3fddbf243ba986de153775f30c2c 100755
--- a/scripts/run_tests_locally-telemetry-DB.sh
+++ b/scripts/run_tests_locally-telemetry-DB.sh
@@ -21,7 +21,7 @@ cd $PROJECTDIR/src
# coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
# kpi_manager/tests/test_unitary.py
CRDB_SQL_ADDRESS=$(kubectl get service --namespace ${CRDB_NAMESPACE} cockroachdb-public -o 'jsonpath={.spec.clusterIP}')
-export CRDB_URI="cockroachdb://tfs:tfs123@${CRDB_SQL_ADDRESS}:26257/tfs-telemetry?sslmode=require"
+export CRDB_URI="cockroachdb://tfs:tfs123@${CRDB_SQL_ADDRESS}:26257/tfs_telemetry?sslmode=require"
RCFILE=$PROJECTDIR/coverage/.coveragerc
python3 -m pytest --log-level=DEBUG --log-cli-level=debug --verbose \
telemetry/tests/test_telemetryDB.py
diff --git a/scripts/run_tests_locally-telemetry-backend.sh b/scripts/run_tests_locally-telemetry-backend.sh
index 97a06a0d6c16daf94e3e6b30bfc70eca3e7ce3a3..f711282401d8fff9b6c98cea7c9240482f1ba24b 100755
--- a/scripts/run_tests_locally-telemetry-backend.sh
+++ b/scripts/run_tests_locally-telemetry-backend.sh
@@ -19,7 +19,12 @@ PROJECTDIR=`pwd`
cd $PROJECTDIR/src
# RCFILE=$PROJECTDIR/coverage/.coveragerc
# coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
+# kpi_manager/tests/test_unitary.py
+# python3 kpi_manager/tests/test_unitary.py
+export KFK_SERVER_ADDRESS='127.0.0.1:9092'
+CRDB_SQL_ADDRESS=$(kubectl get service cockroachdb-public --namespace crdb -o jsonpath='{.spec.clusterIP}')
+export CRDB_URI="cockroachdb://tfs:tfs123@${CRDB_SQL_ADDRESS}:26257/tfs_telemetry?sslmode=require"
RCFILE=$PROJECTDIR/coverage/.coveragerc
diff --git a/scripts/run_tests_locally-telemetry-frontend.sh b/scripts/run_tests_locally-telemetry-frontend.sh
index 7506be5e0750b44e37368e86dbbfd00131c0d270..4bb7bc3fa8e198ead54224148f60044e079ccdf3 100755
--- a/scripts/run_tests_locally-telemetry-frontend.sh
+++ b/scripts/run_tests_locally-telemetry-frontend.sh
@@ -18,9 +18,10 @@ PROJECTDIR=`pwd`
cd $PROJECTDIR/src
-CRDB_SQL_ADDRESS=$(kubectl get service --namespace ${CRDB_NAMESPACE} cockroachdb-public -o 'jsonpath={.spec.clusterIP}')
-export CRDB_URI="cockroachdb://tfs:tfs123@${CRDB_SQL_ADDRESS}:26257/tfs_kpi_mgmt?sslmode=require"
-
+# python3 kpi_manager/tests/test_unitary.py
+export KFK_SERVER_ADDRESS='127.0.0.1:9092'
+CRDB_SQL_ADDRESS=$(kubectl get service cockroachdb-public --namespace crdb -o jsonpath='{.spec.clusterIP}')
+export CRDB_URI="cockroachdb://tfs:tfs123@${CRDB_SQL_ADDRESS}:26257/tfs_telemetry?sslmode=require"
RCFILE=$PROJECTDIR/coverage/.coveragerc
python3 -m pytest --log-level=DEBUG --log-cli-level=DEBUG --verbose \
telemetry/frontend/tests/test_frontend.py
diff --git a/src/analytics/backend/service/AnalyticsBackendService.py b/src/analytics/backend/service/AnalyticsBackendService.py
index fa78a5106ca8b4a4a183fe0e028488e55c3659d2..2e57e1e8a70d141474b395b5e13f21bd4840abc5 100755
--- a/src/analytics/backend/service/AnalyticsBackendService.py
+++ b/src/analytics/backend/service/AnalyticsBackendService.py
@@ -48,7 +48,7 @@ class AnalyticsBackendService(GenericGrpcService):
listener for requests on Kafka topic.
"""
LOGGER.info("Request Listener is initiated ...")
- print ("Request Listener is initiated ...")
+ # print ("Request Listener is initiated ...")
consumer = self.kafka_consumer
consumer.subscribe([KafkaTopic.ANALYTICS_REQUEST.value])
while True:
@@ -60,13 +60,13 @@ class AnalyticsBackendService(GenericGrpcService):
continue
else:
LOGGER.error("Consumer error: {:}".format(receive_msg.error()))
- print ("Consumer error: {:}".format(receive_msg.error()))
+ # print ("Consumer error: {:}".format(receive_msg.error()))
break
try:
analyzer = json.loads(receive_msg.value().decode('utf-8'))
analyzer_uuid = receive_msg.key().decode('utf-8')
LOGGER.debug('Recevied Analyzer: {:} - {:}'.format(analyzer_uuid, analyzer))
- print ('Recevied Analyzer: {:} - {:}'.format(analyzer_uuid, analyzer))
+ # print ('Recevied Analyzer: {:} - {:}'.format(analyzer_uuid, analyzer))
if analyzer["algo_name"] is None and analyzer["oper_mode"] is None:
self.StopDaskListener(analyzer_uuid)
@@ -74,7 +74,7 @@ class AnalyticsBackendService(GenericGrpcService):
self.StartDaskListener(analyzer_uuid, analyzer)
except Exception as e:
LOGGER.warning("Unable to consume message from topic: {:}. ERROR: {:}".format(KafkaTopic.ANALYTICS_REQUEST.value, e))
- print ("Unable to consume message from topic: {:}. ERROR: {:}".format(KafkaTopic.ANALYTICS_REQUEST.value, e))
+ # print ("Unable to consume message from topic: {:}. ERROR: {:}".format(KafkaTopic.ANALYTICS_REQUEST.value, e))
def StartDaskListener(self, analyzer_uuid, analyzer):
kpi_list = analyzer[ 'input_kpis' ]
@@ -84,8 +84,8 @@ class AnalyticsBackendService(GenericGrpcService):
LOGGER.debug ("Received parameters: {:} - {:} - {:} - {:}".format(
kpi_list, thresholds, window_size, window_slider))
- print ("Received parameters: {:} - {:} - {:} - {:}".format(
- kpi_list, thresholds, window_size, window_slider))
+ # print ("Received parameters: {:} - {:} - {:} - {:}".format(
+ # kpi_list, thresholds, window_size, window_slider))
try:
stop_event = Event()
thread = Thread(
@@ -98,11 +98,11 @@ class AnalyticsBackendService(GenericGrpcService):
)
thread.start()
self.running_threads[analyzer_uuid] = (thread, stop_event)
- print ("Initiated Analyzer backend: {:}".format(analyzer_uuid))
+ # print ("Initiated Analyzer backend: {:}".format(analyzer_uuid))
LOGGER.info("Initiated Analyzer backend: {:}".format(analyzer_uuid))
return True
except Exception as e:
- print ("Failed to initiate Analyzer backend: {:}".format(e))
+ # print ("Failed to initiate Analyzer backend: {:}".format(e))
LOGGER.error("Failed to initiate Analyzer backend: {:}".format(e))
return False
@@ -113,12 +113,12 @@ class AnalyticsBackendService(GenericGrpcService):
stop_event.set()
thread.join()
del self.running_threads[analyzer_uuid]
- print ("Terminating backend (by TerminateBackend): Analyzer Id: {:}".format(analyzer_uuid))
+ # print ("Terminating backend (by TerminateBackend): Analyzer Id: {:}".format(analyzer_uuid))
LOGGER.info("Terminating backend (by TerminateBackend): Analyzer Id: {:}".format(analyzer_uuid))
return True
except Exception as e:
LOGGER.error("Failed to terminate. Analyzer Id: {:} - ERROR: {:}".format(analyzer_uuid, e))
return False
else:
- print ("Analyzer not found in active collectors. Analyzer Id: {:}".format(analyzer_uuid))
+ # print ("Analyzer not found in active collectors. Analyzer Id: {:}".format(analyzer_uuid))
LOGGER.warning("Analyzer not found in active collectors: Analyzer Id: {:}".format(analyzer_uuid))
diff --git a/src/analytics/backend/tests/test_backend.py b/src/analytics/backend/tests/test_backend.py
index 470729160c75fd7491e58191f534db9f4da61806..86de220a21b4c2c1c38d518c01ae13f33ee200d5 100644
--- a/src/analytics/backend/tests/test_backend.py
+++ b/src/analytics/backend/tests/test_backend.py
@@ -15,7 +15,7 @@
import time, json
from typing import Dict
import logging
-import threading
+from threading import Event, Thread
from common.tools.kafka.Variables import KafkaTopic
from analytics.backend.service.AnalyticsBackendService import AnalyticsBackendService
from analytics.backend.tests.messages import get_kpi_id_list, get_operation_list, get_threshold_dict
@@ -31,10 +31,10 @@ LOGGER = logging.getLogger(__name__)
###########################
# --- "test_validate_kafka_topics" should be run before the functionality tests ---
-# def test_validate_kafka_topics():
-# LOGGER.debug(" >>> test_validate_kafka_topics: START <<< ")
-# response = KafkaTopic.create_all_topics()
-# assert isinstance(response, bool)
+def test_validate_kafka_topics():
+ LOGGER.debug(" >>> test_validate_kafka_topics: START <<< ")
+ response = KafkaTopic.create_all_topics()
+ assert isinstance(response, bool)
# --- To test Dask Streamer functionality ---
@@ -119,20 +119,20 @@ LOGGER = logging.getLogger(__name__)
# assert isinstance(response, bool)
# --- To TEST StartRequestListenerFunctionality
-def test_StartRequestListener():
- LOGGER.info('test_RunRequestListener')
- AnalyticsBackendServiceObj = AnalyticsBackendService()
- AnalyticsBackendServiceObj.stop_event = Event()
- listener_thread = Thread(target=AnalyticsBackendServiceObj.RequestListener, args=())
- listener_thread.start()
+# def test_StartRequestListener():
+# LOGGER.info('test_RunRequestListener')
+# AnalyticsBackendServiceObj = AnalyticsBackendService()
+# AnalyticsBackendServiceObj.stop_event = Event()
+# listener_thread = Thread(target=AnalyticsBackendServiceObj.RequestListener, args=())
+# listener_thread.start()
- time.sleep(100)
+# time.sleep(100)
# AnalyticsBackendServiceObj.stop_event.set()
# LOGGER.info('Backend termination initiated. waiting for termination... 10 seconds')
# listener_thread.join(timeout=10)
# assert not listener_thread.is_alive(), "RequestListener thread did not terminate as expected."
- LOGGER.info('Completed test_RunRequestListener')
+ # LOGGER.info('Completed test_RunRequestListener')
# To test START and STOP communication together
# def test_StopRequestListener():
diff --git a/src/analytics/frontend/service/AnalyticsFrontendServiceServicerImpl.py b/src/analytics/frontend/service/AnalyticsFrontendServiceServicerImpl.py
index d48132de5d4b507411014b2e194c5fcd71b9bafa..219821ac9bf9bf05e7c599480795d988d1015605 100644
--- a/src/analytics/frontend/service/AnalyticsFrontendServiceServicerImpl.py
+++ b/src/analytics/frontend/service/AnalyticsFrontendServiceServicerImpl.py
@@ -12,7 +12,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-
import logging, grpc, json, queue
from typing import Dict
@@ -83,7 +82,7 @@ class AnalyticsFrontendServiceServicerImpl(AnalyticsFrontendServiceServicer):
)
LOGGER.info("Analyzer Start Request Generated: Analyzer Id: {:}, Value: {:}".format(analyzer_uuid, analyzer_to_generate))
self.kafka_producer.flush()
-
+
def StartResponseListener(self, filter_key=None):
"""
@@ -102,10 +101,8 @@ class AnalyticsFrontendServiceServicerImpl(AnalyticsFrontendServiceServicer):
LOGGER.info(f"Started Kafka listener for topic {self.listener_topic}...")
try:
while True:
- LOGGER.info("entering while...")
- key, value = self.result_queue.get() # Wait until a result is available
- LOGGER.info("In while true ...")
- yield key, value # Yield the result to the calling function
+ key, value = self.result_queue.get()
+ yield key, value
except KeyboardInterrupt:
LOGGER.warning("Listener stopped manually.")
finally:
@@ -135,8 +132,6 @@ class AnalyticsFrontendServiceServicerImpl(AnalyticsFrontendServiceServicer):
self.result_queue.put((key, value))
else:
LOGGER.info(f"Skipping message with unmatched key: {key}")
- # value = json.loads(msg.value().decode('utf-8')) # Added for debugging
- # self.result_queue.put((filter_key, value)) # Added for debugging
except Exception as e:
LOGGER.error(f"Error processing Kafka message: {e}")
@@ -207,7 +202,7 @@ class AnalyticsFrontendServiceServicerImpl(AnalyticsFrontendServiceServicer):
def delivery_callback(self, err, msg):
if err:
LOGGER.debug('Message delivery failed: {:}'.format(err))
- print ('Message delivery failed: {:}'.format(err))
+ # print ('Message delivery failed: {:}'.format(err))
else:
LOGGER.debug('Message delivered to topic {:}'.format(msg.topic()))
- print('Message delivered to topic {:}'.format(msg.topic()))
+ # print('Message delivered to topic {:}'.format(msg.topic()))
diff --git a/src/analytics/frontend/tests/test_frontend.py b/src/analytics/frontend/tests/test_frontend.py
index 8903c7bf8cafd7f4f6e6bd4ebdf8f585fd8dc320..bdab55198575addaa07e1e16f0883ba558c033da 100644
--- a/src/analytics/frontend/tests/test_frontend.py
+++ b/src/analytics/frontend/tests/test_frontend.py
@@ -84,10 +84,10 @@ def analyticsFrontend_client(analyticsFrontend_service : AnalyticsFrontendServic
###########################
# --- "test_validate_kafka_topics" should be executed before the functionality tests ---
-# def test_validate_kafka_topics():
-# LOGGER.debug(" >>> test_validate_kafka_topics: START <<< ")
-# response = KafkaTopic.create_all_topics()
-# assert isinstance(response, bool)
+def test_validate_kafka_topics():
+ LOGGER.debug(" >>> test_validate_kafka_topics: START <<< ")
+ response = KafkaTopic.create_all_topics()
+ assert isinstance(response, bool)
# ----- core funtionality test -----
# def test_StartAnalytics(analyticsFrontend_client):
@@ -97,9 +97,8 @@ def analyticsFrontend_client(analyticsFrontend_service : AnalyticsFrontendServic
# assert isinstance(response, AnalyzerId)
# To test start and stop listener together
-def test_StartStopAnalyzers(analyticsFrontend_client):
- LOGGER.info(' >>> test_StartStopAnalyzers START: <<< ')
- LOGGER.info('--> StartAnalyzer')
+def test_StartAnalyzers(analyticsFrontend_client):
+ LOGGER.info(' >>> test_StartAnalyzers START: <<< ')
added_analyzer_id = analyticsFrontend_client.StartAnalyzer(create_analyzer())
LOGGER.debug(str(added_analyzer_id))
LOGGER.info(' --> Calling StartResponseListener... ')
diff --git a/src/analytics/tests/test_analytics_db.py b/src/analytics/tests/test_analytics_db.py
index 7979b778c576fbf83c25989781b1676a76e9fe06..2794edb4a051b38d4cef902fd09aaad5db966179 100644
--- a/src/analytics/tests/test_analytics_db.py
+++ b/src/analytics/tests/test_analytics_db.py
@@ -15,13 +15,13 @@
import logging
from analytics.database.Analyzer_DB import AnalyzerDB
-from analytics.database.AnalyzerModel import Analyzer as AnalyzerModel
+from analytics.database.AnalyzerModel import Analyzer
LOGGER = logging.getLogger(__name__)
def test_verify_databases_and_tables():
LOGGER.info('>>> test_verify_databases_and_tables : START <<< ')
- AnalyzerDBobj = AnalyzerDB(AnalyzerModel)
+ AnalyzerDBobj = AnalyzerDB(Analyzer)
# AnalyzerDBobj.drop_database()
# AnalyzerDBobj.verify_tables()
AnalyzerDBobj.create_database()
diff --git a/src/common/Constants.py b/src/common/Constants.py
index ae00f0b113ee586f150c59165349a8a5ded8d74e..7b83baa814cc09a140dbbcf52b06976c354d3bcb 100644
--- a/src/common/Constants.py
+++ b/src/common/Constants.py
@@ -116,15 +116,12 @@ DEFAULT_SERVICE_GRPC_PORTS = {
# Default HTTP/REST-API service ports
DEFAULT_SERVICE_HTTP_PORTS = {
- ServiceNameEnum.CONTEXT .value : 8080,
- ServiceNameEnum.NBI .value : 8080,
- ServiceNameEnum.WEBUI .value : 8004,
- ServiceNameEnum.QKD_APP .value : 8005,
+ ServiceNameEnum.NBI .value : 8080,
+ ServiceNameEnum.WEBUI.value : 8004,
}
# Default HTTP/REST-API service base URLs
DEFAULT_SERVICE_HTTP_BASEURLS = {
- ServiceNameEnum.NBI .value : None,
- ServiceNameEnum.WEBUI .value : None,
- ServiceNameEnum.QKD_APP .value : None,
+ ServiceNameEnum.NBI .value : None,
+ ServiceNameEnum.WEBUI.value : None,
}
diff --git a/src/common/Settings.py b/src/common/Settings.py
index 13fcfc76966301599b0f5f39f2b188aea4e4d52a..5845ee5221be69e5d5cf00b60d9376a7fd6fcec3 100644
--- a/src/common/Settings.py
+++ b/src/common/Settings.py
@@ -15,9 +15,11 @@
import logging, os, re, time
from typing import Dict, List
from common.Constants import (
- DEFAULT_GRPC_BIND_ADDRESS, DEFAULT_GRPC_GRACE_PERIOD, DEFAULT_GRPC_MAX_WORKERS, DEFAULT_HTTP_BIND_ADDRESS,
- DEFAULT_LOG_LEVEL, DEFAULT_METRICS_PORT, DEFAULT_SERVICE_GRPC_PORTS, DEFAULT_SERVICE_HTTP_BASEURLS,
- DEFAULT_SERVICE_HTTP_PORTS, ServiceNameEnum
+ DEFAULT_GRPC_BIND_ADDRESS, DEFAULT_GRPC_GRACE_PERIOD,
+ DEFAULT_GRPC_MAX_WORKERS, DEFAULT_HTTP_BIND_ADDRESS,
+ DEFAULT_LOG_LEVEL, DEFAULT_METRICS_PORT, DEFAULT_SERVICE_GRPC_PORTS,
+ DEFAULT_SERVICE_HTTP_BASEURLS, DEFAULT_SERVICE_HTTP_PORTS,
+ ServiceNameEnum
)
LOGGER = logging.getLogger(__name__)
diff --git a/src/context/tests/conftest.py b/src/context/tests/conftest.py
index 905c87c3179aec400e35d12c4d0c5ddbcc21529f..108814c58c7d1f641c3fbcd5dc13a7c7f7159f48 100644
--- a/src/context/tests/conftest.py
+++ b/src/context/tests/conftest.py
@@ -18,8 +18,9 @@ from _pytest.terminal import TerminalReporter
from typing import Tuple
from common.Constants import ServiceNameEnum
from common.Settings import (
- ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, ENVVAR_SUFIX_SERVICE_PORT_HTTP, get_env_var_name,
- get_service_port_grpc, get_service_port_http)
+ ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC,
+ get_env_var_name, get_service_port_grpc
+)
from common.message_broker.Factory import get_messagebroker_backend
from common.message_broker.MessageBroker import MessageBroker
from common.method_wrappers.Decorator import MetricsPool
@@ -30,11 +31,9 @@ from context.service.database.models._Base import rebuild_database
LOCAL_HOST = '127.0.0.1'
GRPC_PORT = 10000 + int(get_service_port_grpc(ServiceNameEnum.CONTEXT)) # avoid privileged ports
-HTTP_PORT = 10000 + int(get_service_port_http(ServiceNameEnum.CONTEXT)) # avoid privileged ports
os.environ[get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_HOST )] = str(LOCAL_HOST)
os.environ[get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_PORT_GRPC)] = str(GRPC_PORT)
-os.environ[get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_PORT_HTTP)] = str(HTTP_PORT)
@pytest.fixture(scope='session')
def context_db_mb(request) -> Tuple[sqlalchemy.engine.Engine, MessageBroker]: # pylint: disable=unused-argument
diff --git a/src/kpi_value_api/tests/test_kpi_value_api.py b/src/kpi_value_api/tests/test_kpi_value_api.py
index c7eb8fef1c79d0280c951d89971695d2c39ab599..c245bb9ef64eaa29dc4d51955ff94adeeeeb8dda 100644
--- a/src/kpi_value_api/tests/test_kpi_value_api.py
+++ b/src/kpi_value_api/tests/test_kpi_value_api.py
@@ -85,7 +85,7 @@ def test_validate_kafka_topics():
# LOGGER.debug(str(response))
# assert isinstance(response, KpiAlarms)
-def test_store_kpi_values(kpi_value_api_client):
- LOGGER.debug(" >>> test_set_list_of_KPIs: START <<< ")
- response = kpi_value_api_client.StoreKpiValues(create_kpi_value_list())
- assert isinstance(response, Empty)
+# def test_store_kpi_values(kpi_value_api_client):
+# LOGGER.debug(" >>> test_set_list_of_KPIs: START <<< ")
+# response = kpi_value_api_client.StoreKpiValues(create_kpi_value_list())
+# assert isinstance(response, Empty)
diff --git a/src/nbi/Dockerfile b/src/nbi/Dockerfile
index 1435e9757226e290e92f208ef48a7182eb106c55..c5fc8d32491d7576e93534c75c264e79c37a36c6 100644
--- a/src/nbi/Dockerfile
+++ b/src/nbi/Dockerfile
@@ -89,6 +89,8 @@ COPY src/service/__init__.py service/__init__.py
COPY src/service/client/. service/client/
COPY src/slice/__init__.py slice/__init__.py
COPY src/slice/client/. slice/client/
+COPY src/qkd_app/__init__.py qkd_app/__init__.py
+COPY src/qkd_app/client/. qkd_app/client/
RUN mkdir -p /var/teraflow/tests/tools
COPY src/tests/tools/mock_osm/. tests/tools/mock_osm/
diff --git a/src/nbi/service/__main__.py b/src/nbi/service/__main__.py
index 58fbb9625addc43c6b62d06d7a9caa3f648203d5..8f4ef87e03a3954227d777ab06c220d373b70c08 100644
--- a/src/nbi/service/__main__.py
+++ b/src/nbi/service/__main__.py
@@ -16,9 +16,10 @@ import logging, signal, sys, threading
from prometheus_client import start_http_server
from common.Constants import ServiceNameEnum
from common.Settings import (
- ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, get_env_var_name, get_log_level, get_metrics_port,
- wait_for_environment_variables)
-
+ ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC,
+ get_env_var_name, get_log_level, get_metrics_port,
+ wait_for_environment_variables
+)
from .NbiService import NbiService
from .rest_server.RestServer import RestServer
from .rest_server.nbi_plugins.etsi_bwm import register_etsi_bwm_api
@@ -28,6 +29,7 @@ from .rest_server.nbi_plugins.ietf_l3vpn import register_ietf_l3vpn
from .rest_server.nbi_plugins.ietf_network import register_ietf_network
from .rest_server.nbi_plugins.ietf_network_slice import register_ietf_nss
from .rest_server.nbi_plugins.ietf_acl import register_ietf_acl
+from .rest_server.nbi_plugins.qkd_app import register_qkd_app
from .rest_server.nbi_plugins.tfs_api import register_tfs_api
terminate = threading.Event()
@@ -72,6 +74,7 @@ def main():
register_ietf_network(rest_server)
register_ietf_nss(rest_server) # Registering NSS entrypoint
register_ietf_acl(rest_server)
+ register_qkd_app(rest_server)
register_tfs_api(rest_server)
rest_server.start()
diff --git a/src/qkd_app/service/rest_server/qkd_app/Resources.py b/src/nbi/service/rest_server/nbi_plugins/qkd_app/Resources.py
similarity index 57%
rename from src/qkd_app/service/rest_server/qkd_app/Resources.py
rename to src/nbi/service/rest_server/nbi_plugins/qkd_app/Resources.py
index 6ba79d3940da91dfebc1a1c666893548caccbe6c..d14fe9575b21319a0fa597a7746510a11e102903 100644
--- a/src/qkd_app/service/rest_server/qkd_app/Resources.py
+++ b/src/nbi/service/rest_server/nbi_plugins/qkd_app/Resources.py
@@ -12,7 +12,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import uuid, json
+import uuid
+import json
from flask import request
from flask_restful import Resource
from common.proto.context_pb2 import Empty
@@ -21,7 +22,6 @@ from common.Constants import DEFAULT_CONTEXT_NAME
from context.client.ContextClient import ContextClient
from qkd_app.client.QKDAppClient import QKDAppClient
-
class _Resource(Resource):
def __init__(self) -> None:
super().__init__()
@@ -32,18 +32,56 @@ class Index(_Resource):
def get(self):
return {'hello': 'world'}
+class ListDevices(_Resource):
+ def get(self):
+ """
+ List devices and associate the apps with them.
+ """
+ devices = self.context_client.ListDevices(Empty()).devices
+ for device in devices:
+ # Fetch apps associated with this device
+ device.apps = self.get_apps_for_device(device.device_id.device_uuid.uuid)
+ return {'devices': [self.format_device(device) for device in devices]}
+
+ def get_apps_for_device(self, device_uuid):
+ """
+ Fetch the apps associated with a given device UUID.
+ """
+ try:
+ # Call the AppService to get the list of apps
+ apps_list = self.qkd_app_client.ListApps(Empty())
+
+ # Filter apps for this specific device
+ device_apps = []
+ for app in apps_list.apps:
+ if app.local_device_id.device_uuid.uuid == device_uuid or \
+ app.remote_device_id.device_uuid.uuid == device_uuid:
+ device_apps.append(app)
+ return device_apps
+
+ except Exception as e:
+ print(f"Error fetching apps for device {device_uuid}: {e}")
+ return []
+
+ def format_device(self, device):
+ """
+ Formats a device object to include the associated apps in the response.
+ """
+ return {
+ 'device_uuid': device.device_id.device_uuid.uuid,
+ 'name': device.name,
+ 'type': device.device_type,
+ 'status': device.device_operational_status,
+ 'apps': [{'app_id': app.app_id.app_uuid.uuid, 'app_status': app.app_status, 'app_type': app.app_type} for app in device.apps]
+ }
+
class CreateQKDApp(_Resource):
- # Optare: Post request for the QKD Node to call the TeraflowSDN. Example of requests below
def post(self):
app = request.get_json()['app']
-
- devices = self.context_client.ListDevices(Empty())
- devices = devices.devices
-
+ devices = self.context_client.ListDevices(Empty()).devices
local_device = None
-
- # This for-loop won't be necessary if we can garantee Device ID is the same as QKDN Id
+ # This for-loop won't be necessary if Device ID is guaranteed to be the same as QKDN Id
for device in devices:
for config_rule in device.device_config.config_rules:
if config_rule.custom.resource_key == '__node__':
@@ -53,15 +91,6 @@ class CreateQKDApp(_Resource):
local_device = device
break
- # Optare: Todo: Verify that a service is present for this app
- '''
- requests.post('http://10.211.36.220/app/create_qkd_app', json={'app': {'server_app_id':'1', 'client_app_id':[], 'app_status':'ON', 'local_qkdn_id':'00000001-0000-0000-0000-000000000000', 'backing_qkdl_id':['00000003-0002-0000-0000-000000000000']}})
-
-
- requests.post('http://10.211.36.220/app/create_qkd_app', json={'app': {'server_app_id':'1', 'client_app_id':[], 'app_status':'ON', 'local_qkdn_id':'00000003-0000-0000-0000-000000000000', 'backing_qkdl_id':['00000003-0002-0000-0000-000000000000']}})
- '''
-
-
if local_device is None:
return {"status": "fail"}
@@ -76,11 +105,7 @@ class CreateQKDApp(_Resource):
'remote_device_id': {'device_uuid': {'uuid': ''}},
}
-
- # Optare: This will call our internal RegisterApp which supports the creation of both internal and external app.
- # Optare the verification for knowing if two parties are requesting the same app is done inside RegisterApp's function
self.qkd_app_client.RegisterApp(App(**external_app_src_dst))
- # Optare: Todo: Communicate by SBI with both Nodes of the new App
-
return {"status": "success"}
+
diff --git a/src/qkd_app/service/rest_server/qkd_app/__init__.py b/src/nbi/service/rest_server/nbi_plugins/qkd_app/__init__.py
similarity index 67%
rename from src/qkd_app/service/rest_server/qkd_app/__init__.py
rename to src/nbi/service/rest_server/nbi_plugins/qkd_app/__init__.py
index 6fc23b371414dcb2bac4afde63524febf71e5337..30982f104e189eb4af5b3f49b1abe2b0e5fb2d85 100644
--- a/src/qkd_app/service/rest_server/qkd_app/__init__.py
+++ b/src/nbi/service/rest_server/nbi_plugins/qkd_app/__init__.py
@@ -4,7 +4,7 @@
# 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
+# 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,
@@ -12,19 +12,18 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from qkd_app.service.rest_server.RestServer import RestServer
-from .Resources import (
- CreateQKDApp, Index)
+from nbi.service.rest_server.RestServer import RestServer
+from .Resources import CreateQKDApp, Index
URL_PREFIX = '/qkd_app'
# Use 'path' type since some identifiers might contain char '/' and Flask is unable to recognize them in 'string' type.
RESOURCES = [
# (endpoint_name, resource_class, resource_url)
- ('api.index', Index, '/'),
- ('api.register_qkd_app', CreateQKDApp, '/create_qkd_app'),
+ ('api.index', Index, '/'),
+ ('api.register_qkd_app', CreateQKDApp, '/create_qkd_app'),
]
-def register_qkd_app(app_server : RestServer):
+def register_qkd_app(rest_server : RestServer):
for endpoint_name, resource_class, resource_url in RESOURCES:
- app_server.add_resource(resource_class, URL_PREFIX + resource_url, endpoint=endpoint_name)
+ rest_server.add_resource(resource_class, URL_PREFIX + resource_url, endpoint=endpoint_name)
diff --git a/src/qkd_app/client/QKDAppClient.py b/src/qkd_app/client/QKDAppClient.py
index 1a174df6adc69ab9ce88b0d8878c92b9b9e7820e..a35b18f1d9f380ae4f95f1a643716075642e31d6 100644
--- a/src/qkd_app/client/QKDAppClient.py
+++ b/src/qkd_app/client/QKDAppClient.py
@@ -12,53 +12,92 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import grpc, logging
+import grpc
+import logging
from common.Constants import ServiceNameEnum
from common.Settings import get_service_host, get_service_port_grpc
-from common.proto.context_pb2 import Empty, ContextId
from common.proto.qkd_app_pb2 import App, AppId, AppList
from common.proto.qkd_app_pb2_grpc import AppServiceStub
from common.tools.client.RetryDecorator import retry, delay_exponential
from common.tools.grpc.Tools import grpc_message_to_json_string
LOGGER = logging.getLogger(__name__)
+
+# Define retry mechanism
MAX_RETRIES = 15
DELAY_FUNCTION = delay_exponential(initial=0.01, increment=2.0, maximum=5.0)
-RETRY_DECORATOR = retry(max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect')
class QKDAppClient:
def __init__(self, host=None, port=None):
- if not host: host = get_service_host(ServiceNameEnum.QKD_APP)
- if not port: port = get_service_port_grpc(ServiceNameEnum.QKD_APP)
- self.endpoint = '{:s}:{:s}'.format(str(host), str(port))
- LOGGER.debug('Creating channel to {:s}...'.format(self.endpoint))
+ self.host = host or get_service_host(ServiceNameEnum.QKD_APP)
+ self.port = port or get_service_port_grpc(ServiceNameEnum.QKD_APP)
+ self.endpoint = f'{self.host}:{self.port}'
+ LOGGER.debug(f'Initializing gRPC client to {self.endpoint}...')
self.channel = None
self.stub = None
self.connect()
- LOGGER.debug('Channel created')
def connect(self):
- self.channel = grpc.insecure_channel(self.endpoint)
- self.stub = AppServiceStub(self.channel)
+ try:
+ self.channel = grpc.insecure_channel(self.endpoint)
+ self.stub = AppServiceStub(self.channel)
+ LOGGER.debug(f'gRPC channel to {self.endpoint} established successfully')
+ except Exception as e:
+ LOGGER.error(f"Failed to establish gRPC connection: {e}")
+ self.stub = None
def close(self):
- if self.channel is not None: self.channel.close()
+ if self.channel:
+ self.channel.close()
+ LOGGER.debug(f'gRPC channel to {self.endpoint} closed')
self.channel = None
self.stub = None
+ def check_connection(self):
+ if self.stub is None:
+ LOGGER.error("gRPC connection is not established. Retrying...")
+ self.connect()
+ if self.stub is None:
+ raise ConnectionError("gRPC connection could not be established.")
+
+ @retry(max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION)
+ def RegisterApp(self, app_request: App) -> None:
+ """Register a new QKD app."""
+ self.check_connection()
+ LOGGER.debug(f'RegisterApp request: {grpc_message_to_json_string(app_request)}')
+ self.stub.RegisterApp(app_request)
+ LOGGER.debug('App registered successfully')
+ @retry(max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION)
+ def UpdateApp(self, app_request: App) -> None:
+ """Update an existing QKD app."""
+ self.check_connection()
+ LOGGER.debug(f'UpdateApp request: {grpc_message_to_json_string(app_request)}')
+ self.stub.UpdateApp(app_request)
+ LOGGER.debug('App updated successfully')
- @RETRY_DECORATOR
- def RegisterApp(self, request : App) -> Empty:
- LOGGER.debug('RegisterApp request: {:s}'.format(grpc_message_to_json_string(request)))
- response = self.stub.RegisterApp(request)
- LOGGER.debug('RegisterApp result: {:s}'.format(grpc_message_to_json_string(response)))
+ @retry(max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION)
+ def ListApps(self, context_id) -> AppList:
+ """List all apps for a given context."""
+ self.check_connection()
+ LOGGER.debug(f'ListApps request for context_id: {grpc_message_to_json_string(context_id)}')
+ response = self.stub.ListApps(context_id)
+ LOGGER.debug(f'ListApps result: {grpc_message_to_json_string(response)}')
return response
-
- @RETRY_DECORATOR
- def ListApps(self, request: ContextId) -> AppList:
- LOGGER.debug('ListApps request: {:s}'.format(grpc_message_to_json_string(request)))
- response = self.stub.ListApps(request)
- LOGGER.debug('ListApps result: {:s}'.format(grpc_message_to_json_string(response)))
+ @retry(max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION)
+ def GetApp(self, app_id: AppId) -> App:
+ """Fetch details of a specific app by its ID."""
+ self.check_connection()
+ LOGGER.debug(f'GetApp request for app_id: {grpc_message_to_json_string(app_id)}')
+ response = self.stub.GetApp(app_id)
+ LOGGER.debug(f'GetApp result: {grpc_message_to_json_string(response)}')
return response
+
+ @retry(max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION)
+ def DeleteApp(self, app_id: AppId) -> None:
+ """Delete an app by its ID."""
+ self.check_connection() # Ensures connection is established
+ LOGGER.debug(f'DeleteApp request for app_id: {grpc_message_to_json_string(app_id)}')
+ self.stub.DeleteApp(app_id) # Calls the gRPC service
+ LOGGER.debug('App deleted successfully')
diff --git a/src/qkd_app/service/QKDAppService.py b/src/qkd_app/service/QKDAppService.py
index a6c93cd811a72594804fe8e8e86a9586533a1317..b9b34ed9b8239ea119c5af1364bbc53914b42c51 100644
--- a/src/qkd_app/service/QKDAppService.py
+++ b/src/qkd_app/service/QKDAppService.py
@@ -12,7 +12,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import logging, sqlalchemy
+import logging
+import sqlalchemy
from common.Constants import ServiceNameEnum
from common.Settings import get_service_port_grpc
from common.message_broker.MessageBroker import MessageBroker
@@ -20,18 +21,38 @@ from common.proto.qkd_app_pb2_grpc import add_AppServiceServicer_to_server
from common.tools.service.GenericGrpcService import GenericGrpcService
from qkd_app.service.QKDAppServiceServicerImpl import AppServiceServicerImpl
-# Custom gRPC settings
-GRPC_MAX_WORKERS = 200 # multiple clients might keep connections alive for Get*Events() RPC methods
+# Configure maximum number of workers for gRPC
+GRPC_MAX_WORKERS = 200 # Adjusted for high concurrency
LOGGER = logging.getLogger(__name__)
-
class AppService(GenericGrpcService):
+ """
+ gRPC Service for handling QKD App-related operations.
+ This class initializes the gRPC server and installs the servicers.
+ """
def __init__(
- self, db_engine : sqlalchemy.engine.Engine, messagebroker : MessageBroker, cls_name: str = __name__
+ self, db_engine: sqlalchemy.engine.Engine, messagebroker: MessageBroker, cls_name: str = __name__
) -> None:
+ """
+ Initializes the AppService with the provided database engine and message broker.
+ Sets up the gRPC server to handle app-related requests.
+
+ Args:
+ db_engine (sqlalchemy.engine.Engine): Database engine for handling app data.
+ messagebroker (MessageBroker): Message broker for inter-service communication.
+ cls_name (str): Class name for logging purposes (default is __name__).
+ """
+ # Get the port for the gRPC AppService
port = get_service_port_grpc(ServiceNameEnum.QKD_APP)
+ # Initialize the base class with port and max worker configuration
super().__init__(port, max_workers=GRPC_MAX_WORKERS, cls_name=cls_name)
+ # Initialize the AppServiceServicer with the database and message broker
self.app_servicer = AppServiceServicerImpl(db_engine, messagebroker)
def install_servicers(self):
+ """
+ Installs the AppService servicers to the gRPC server.
+ This allows the server to handle requests for QKD app operations.
+ """
add_AppServiceServicer_to_server(self.app_servicer, self.server)
+ LOGGER.debug("AppService servicer installed")
diff --git a/src/qkd_app/service/QKDAppServiceServicerImpl.py b/src/qkd_app/service/QKDAppServiceServicerImpl.py
index df7a885c47eda9d7a6137c9905388da49c698e7e..3ef13b75e5510a95b91a9a67fa65ed06c32a3526 100644
--- a/src/qkd_app/service/QKDAppServiceServicerImpl.py
+++ b/src/qkd_app/service/QKDAppServiceServicerImpl.py
@@ -12,62 +12,175 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import grpc, logging, sqlalchemy
-#from typing import Iterator, Optional
+import grpc
+import logging
+import sqlalchemy
+import uuid
from common.message_broker.MessageBroker import MessageBroker
-import grpc, json, logging #, deepdiff
-from common.proto.context_pb2 import (
- Empty, Service, ServiceId, ServiceStatusEnum, ServiceTypeEnum, ContextId)
-from common.proto.qkd_app_pb2 import (App, AppId, AppList, QKDAppTypesEnum)
+from common.proto.context_pb2 import Empty, ContextId
+from common.proto.qkd_app_pb2 import App, AppId, AppList, QKDAppTypesEnum, QoS
+from common.method_wrappers.ServiceExceptions import InvalidArgumentException, NotFoundException
from common.proto.qkd_app_pb2_grpc import AppServiceServicer
+from common.tools.grpc.Tools import grpc_message_to_json_string
from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
-#from common.tools.context_queries.InterDomain import is_inter_domain #, is_multi_domain
-#from common.tools.grpc.ConfigRules import copy_config_rules
-#from common.tools.grpc.Constraints import copy_constraints
-#from common.tools.grpc.EndPointIds import copy_endpoint_ids
-#from common.tools.grpc.ServiceIds import update_service_ids
-#from common.tools.grpc.Tools import grpc_message_to_json_string
-#from context.client.ContextClient import ContextClient
-#from qkd_app.client.QKDAppClient import QKDAppClient
-from .database.QKDApp import app_set, app_list_objs, app_get, app_get_by_server
-from common.method_wrappers.ServiceExceptions import NotFoundException
+from .database.QKDApp import app_set, app_list_objs, app_get, app_get_by_server, app_delete
LOGGER = logging.getLogger(__name__)
METRICS_POOL = MetricsPool('QkdApp', 'RPC')
-# Optare: This file must be edited based on app's logic
-
class AppServiceServicerImpl(AppServiceServicer):
- def __init__(self, db_engine : sqlalchemy.engine.Engine, messagebroker : MessageBroker):
- LOGGER.debug('Creating Servicer...')
+ def __init__(self, db_engine: sqlalchemy.engine.Engine, messagebroker: MessageBroker):
+ LOGGER.debug('Initializing AppServiceServicer...')
self.db_engine = db_engine
self.messagebroker = messagebroker
- LOGGER.debug('Servicer Created')
+ LOGGER.debug('AppServiceServicer initialized')
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
- def RegisterApp(self, request : App, context : grpc.ServicerContext) -> Empty:
- # Optare: This is the main function required for the project.
- # Optare: If it's an internal it will save it directly. If it's an external one it will save it as pending by not providing the remote until the other party requests it too
- # Optare: Ideally, the only thing needed to change is the code inside the try block. Currently it just searches by a pending app with the same server_id but you can put more restrictions or different search and raise the NotFoundException
+ def RegisterApp(self, request: App, context: grpc.ServicerContext) -> Empty:
+ """
+ Registers an app in the system, handling both internal and external applications
+ with ETSI GS QKD 015 compliance.
+ """
+ LOGGER.debug(f"Received RegisterApp request: {grpc_message_to_json_string(request)}")
- if request.app_type == QKDAppTypesEnum.QKDAPPTYPES_INTERNAL:
- app_set(self.db_engine, self.messagebroker, request)
+ try:
+ # Validate QoS parameters as per ETSI 015 requirements
+ self._validate_qos(request.qos)
- else:
- try:
- app = app_get_by_server(self.db_engine, request.server_app_id)
- except NotFoundException:
- app = request
- app_set(self.db_engine, self.messagebroker, app)
+ # Check if an app with the same server_app_id and local_device_id already exists
+ existing_app = self._check_existing_app(request.server_app_id, request.local_device_id.device_uuid.uuid)
+
+ if existing_app:
+ if request.app_type == QKDAppTypesEnum.QKDAPPTYPES_CLIENT:
+ LOGGER.debug(f"Handling external app registration for server_app_id: {request.server_app_id}")
+ # Handle second-party registration for external apps
+ if not existing_app.remote_device_id.device_uuid.uuid:
+ existing_app.remote_device_id.device_uuid.uuid = request.local_device_id.device_uuid.uuid
+ app_set(self.db_engine, self.messagebroker, existing_app)
+ LOGGER.debug(f"Updated external app with server_app_id: {request.server_app_id}, remote_device_id: {request.local_device_id.device_uuid.uuid}")
+ else:
+ context.set_code(grpc.StatusCode.ALREADY_EXISTS)
+ context.set_details(f"App with server_app_id {request.server_app_id} already has both parties registered.")
+ return Empty()
+ else:
+ context.set_code(grpc.StatusCode.ALREADY_EXISTS)
+ context.set_details(f"App with server_app_id {request.server_app_id} already exists.")
+ return Empty()
else:
- app.remote_device_id.device_uuid.uuid = request.local_device_id.device_uuid.uuid
- app_set(self.db_engine, self.messagebroker, app)
-
-
- return Empty()
-
+ # Assign application IDs as required
+ self._validate_and_assign_app_ids(request)
+
+ # Register the app
+ if request.app_type == QKDAppTypesEnum.QKDAPPTYPES_INTERNAL:
+ LOGGER.debug(f"Registering internal app with app_uuid: {request.app_id.app_uuid.uuid}")
+ app_set(self.db_engine, self.messagebroker, request)
+ else:
+ self._register_external_app(request)
+
+ LOGGER.debug(f"RegisterApp completed successfully for app: {request.server_app_id}")
+ return Empty()
+
+ except InvalidArgumentException as e:
+ context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
+ context.set_details(str(e))
+ raise e
+ except Exception as e:
+ context.set_code(grpc.StatusCode.INTERNAL)
+ context.set_details("An internal error occurred during app registration.")
+ raise e
+
+ def _validate_qos(self, qos: QoS) -> None:
+ """
+ Validates the QoS parameters for the application, ensuring ETSI 015 compliance.
+ """
+ if qos.max_bandwidth and qos.min_bandwidth and qos.max_bandwidth < qos.min_bandwidth:
+ raise InvalidArgumentException("QoS max_bandwidth cannot be less than min_bandwidth.")
+
+ if qos.ttl and qos.ttl <= 0:
+ raise InvalidArgumentException("QoS TTL must be a positive value.")
+
+ LOGGER.debug(f"QoS validated: {qos}")
+
+ def _check_existing_app(self, server_app_id: str, local_device_id: str):
+ try:
+ return app_get_by_server(self.db_engine, server_app_id)
+ except NotFoundException:
+ return None
+
+ def _validate_and_assign_app_ids(self, request: App) -> None:
+ """
+ Validates and assigns app IDs (app_uuid, server_app_id, client_app_id) if not provided.
+ """
+ if not request.app_id.app_uuid.uuid:
+ request.app_id.app_uuid.uuid = str(uuid.uuid4())
+ LOGGER.debug(f"Assigned new app_uuid: {request.app_id.app_uuid.uuid}")
+
+ if not request.server_app_id:
+ request.server_app_id = str(uuid.uuid4())
+ LOGGER.debug(f"Assigned new server_app_id: {request.server_app_id}")
+
+ del request.client_app_id[:] # Clear the repeated field for clients
+
+ def _register_external_app(self, request: App) -> None:
+ try:
+ existing_app = app_get_by_server(self.db_engine, request.server_app_id)
+
+ if not existing_app.remote_device_id.device_uuid.uuid:
+ existing_app.remote_device_id.device_uuid.uuid = request.local_device_id.device_uuid.uuid
+ app_set(self.db_engine, self.messagebroker, existing_app)
+ else:
+ LOGGER.debug(f"App with server_app_id: {request.server_app_id} already has both parties registered.")
+ except NotFoundException:
+ app_set(self.db_engine, self.messagebroker, request)
+
+ @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
+ def ListApps(self, request: ContextId, context: grpc.ServicerContext) -> AppList:
+ """
+ Lists all apps in the system, including their statistics and QoS attributes.
+ """
+ LOGGER.debug(f"Received ListApps request: {grpc_message_to_json_string(request)}")
+
+ try:
+ apps = app_list_objs(self.db_engine, request.context_uuid.uuid)
+ for app in apps.apps:
+ LOGGER.debug(f"App retrieved: {grpc_message_to_json_string(app)}")
+
+ LOGGER.debug(f"ListApps returned {len(apps.apps)} apps for context_id: {request.context_uuid.uuid}")
+ return apps
+ except Exception as e:
+ context.set_code(grpc.StatusCode.INTERNAL)
+ context.set_details("An internal error occurred while listing apps.")
+ raise e
+
+ @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
+ def GetApp(self, request: AppId, context: grpc.ServicerContext) -> App:
+ """
+ Fetches details of a specific app based on its AppId, including QoS and performance stats.
+ """
+ LOGGER.debug(f"Received GetApp request: {grpc_message_to_json_string(request)}")
+ try:
+ app = app_get(self.db_engine, request)
+ LOGGER.debug(f"GetApp found app with app_uuid: {request.app_uuid.uuid}")
+ return app
+ except NotFoundException as e:
+ context.set_code(grpc.StatusCode.NOT_FOUND)
+ context.set_details(f"App not found: {e}")
+ raise e
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
- def ListApps(self, request: ContextId, context : grpc.ServicerContext) -> AppList:
- return app_list_objs(self.db_engine)
+ def DeleteApp(self, request: AppId, context: grpc.ServicerContext) -> Empty:
+ """
+ Deletes an app from the system by its AppId, following ETSI compliance.
+ """
+ LOGGER.debug(f"Received DeleteApp request for app_uuid: {request.app_uuid.uuid}")
+ try:
+ app_delete(self.db_engine, request.app_uuid.uuid)
+ LOGGER.debug(f"App with UUID {request.app_uuid.uuid} deleted successfully.")
+ return Empty()
+ except NotFoundException as e:
+ context.set_code(grpc.StatusCode.NOT_FOUND)
+ context.set_details(f"App not found: {e}")
+ raise e
+
+
diff --git a/src/qkd_app/service/__main__.py b/src/qkd_app/service/__main__.py
index ed7e554728eb2de6240dd4facb7f084337a026a4..17f3ac240c33b28454e2df43862dd66a50eb5c45 100644
--- a/src/qkd_app/service/__main__.py
+++ b/src/qkd_app/service/__main__.py
@@ -4,7 +4,7 @@
# 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
+# 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,
@@ -12,82 +12,82 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import logging, signal, sys, threading
+import logging
+import signal
+import sys
+import threading
from prometheus_client import start_http_server
-#from common.Constants import ServiceNameEnum
-from common.Settings import (
- #ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, get_env_var_name,
- get_log_level, get_metrics_port, wait_for_environment_variables)
+from common.Settings import get_log_level, get_metrics_port
from qkd_app.service.QKDAppService import AppService
-from qkd_app.service.rest_server.RestServer import RestServer
-from qkd_app.service.rest_server.qkd_app import register_qkd_app
-#from common.message_broker.Factory import get_messagebroker_backend
-#from common.message_broker.MessageBroker import MessageBroker
from qkd_app.service.database.Engine import Engine
from qkd_app.service.database.models._Base import rebuild_database
+# Set up logging
+LOG_LEVEL = get_log_level()
+logging.basicConfig(level=LOG_LEVEL, format="[%(asctime)s] %(levelname)s:%(name)s:%(message)s")
+LOGGER = logging.getLogger(__name__)
+
+#LOGGER.addHandler(logging.StreamHandler(stream=sys.stderr))
+#LOGGER.setLevel(logging.WARNING)
+
+# Event for terminating the service gracefully
terminate = threading.Event()
-LOGGER : logging.Logger = None
-def signal_handler(signal, frame): # pylint: disable=redefined-outer-name
- LOGGER.warning('Terminate signal received')
+def signal_handler(signum, frame):
+ """
+ Handle termination signals like SIGINT and SIGTERM to ensure graceful shutdown.
+ """
+ LOGGER.warning('Termination signal received')
terminate.set()
def main():
- global LOGGER # pylint: disable=global-statement
-
- log_level = get_log_level()
- logging.basicConfig(level=log_level, format="[%(asctime)s] %(levelname)s:%(name)s:%(message)s")
- LOGGER = logging.getLogger(__name__)
-
- wait_for_environment_variables([
- #get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_HOST ),
- #get_env_var_name(ServiceNameEnum.CONTEXT, ENVVAR_SUFIX_SERVICE_PORT_GRPC),
- ])
-
+ LOGGER.info('Starting...')
+ # Register signal handlers for graceful shutdown
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
- LOGGER.info('Starting...')
-
- # Start metrics server
+ # Start Prometheus metrics server
metrics_port = get_metrics_port()
start_http_server(metrics_port)
+ LOGGER.info(f'Metrics server started on port {metrics_port}')
- # Get Database Engine instance and initialize database, if needed
+ # Initialize the SQLAlchemy database engine
LOGGER.info('Getting SQLAlchemy DB Engine...')
db_engine = Engine.get_engine()
if db_engine is None:
- LOGGER.error('Unable to get SQLAlchemy DB Engine...')
+ LOGGER.error('Unable to get SQLAlchemy DB Engine. Exiting...')
return -1
+ # Try creating the database or log any issues
try:
Engine.create_database(db_engine)
- except: # pylint: disable=bare-except # pragma: no cover
- LOGGER.exception('Failed to check/create the database: {:s}'.format(str(db_engine.url)))
+ except Exception as e: # More specific exception handling
+ LOGGER.exception(f'Failed to check/create the database: {db_engine.url}. Error: {str(e)}')
+ return -1
+ # Rebuild the database schema if necessary
rebuild_database(db_engine)
- # Get message broker instance
- messagebroker = None #MessageBroker(get_messagebroker_backend())
+ # Initialize the message broker (if needed)
+ messagebroker = None # Disabled until further notice, can be re-enabled when necessary
+ # messagebroker = MessageBroker(get_messagebroker_backend())
- # Starting context service
+ # Start the gRPC App Service
grpc_service = AppService(db_engine, messagebroker)
grpc_service.start()
- rest_server = RestServer()
- register_qkd_app(rest_server)
- rest_server.start()
+ LOGGER.info('Services started. Waiting for termination signal...')
# Wait for Ctrl+C or termination signal
- while not terminate.wait(timeout=1.0): pass
+ # Keep the process running until a termination signal is received
+ while not terminate.wait(timeout=1.0):
+ pass
- LOGGER.info('Terminating...')
+ # Shutdown services gracefully on termination
+ LOGGER.info('Terminating services...')
grpc_service.stop()
- rest_server.shutdown()
- rest_server.join()
- LOGGER.info('Bye')
+ LOGGER.info('Shutdown complete. Exiting...')
return 0
if __name__ == '__main__':
diff --git a/src/qkd_app/service/database/Engine.py b/src/qkd_app/service/database/Engine.py
index 8f528f9a1b3cacca2ea260901ab808461dd3183d..ec86618218cc396dea31404b88d33e46d320ed58 100644
--- a/src/qkd_app/service/database/Engine.py
+++ b/src/qkd_app/service/database/Engine.py
@@ -28,7 +28,7 @@ class Engine:
if crdb_uri is None:
CRDB_NAMESPACE = get_setting('CRDB_NAMESPACE')
CRDB_SQL_PORT = get_setting('CRDB_SQL_PORT')
- CRDB_DATABASE = get_setting('CRDB_DATABASE_APP')
+ CRDB_DATABASE = get_setting('CRDB_DATABASE')
CRDB_USERNAME = get_setting('CRDB_USERNAME')
CRDB_PASSWORD = get_setting('CRDB_PASSWORD')
CRDB_SSLMODE = get_setting('CRDB_SSLMODE')
diff --git a/src/qkd_app/service/database/QKDApp.py b/src/qkd_app/service/database/QKDApp.py
index b1fb90d4efcd0770bcc4c48c1f00deb0e95687ad..539555a4dac055d56158298ed7db6dceec39f446 100644
--- a/src/qkd_app/service/database/QKDApp.py
+++ b/src/qkd_app/service/database/QKDApp.py
@@ -12,174 +12,162 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import datetime, logging, uuid
+import datetime
+import logging
+from typing import Dict, List, Optional
+
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.engine import Engine
-from sqlalchemy.orm import Session, selectinload, sessionmaker
+from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy_cockroachdb import run_transaction
-from typing import Dict, List, Optional, Set, Tuple
-from common.method_wrappers.ServiceExceptions import InvalidArgumentException, NotFoundException
+
+from common.method_wrappers.ServiceExceptions import NotFoundException
from common.message_broker.MessageBroker import MessageBroker
-from common.proto.context_pb2 import Empty
-from common.proto.qkd_app_pb2 import (
- AppList, App, AppId)
-from common.tools.grpc.Tools import grpc_message_to_json_string
+from common.proto.qkd_app_pb2 import AppList, App, AppId
+from qkd_app.service.database.uuids._Builder import get_uuid_from_string, get_uuid_random
+from common.method_wrappers.ServiceExceptions import InvalidArgumentsException
+from common.tools.object_factory.QKDApp import json_app_id
+from common.tools.object_factory.Context import json_context_id
+
from .models.QKDAppModel import AppModel
-from .models.enums.QKDAppStatus import grpc_to_enum__qkd_app_status
-from .models.enums.QKDAppTypes import grpc_to_enum__qkd_app_types
from .uuids.QKDApp import app_get_uuid
-from common.tools.object_factory.Context import json_context_id
-from common.tools.object_factory.QKDApp import json_app_id
from context.service.database.uuids.Context import context_get_uuid
+from .models.enums.QKDAppStatus import grpc_to_enum__qkd_app_status
+from .models.enums.QKDAppTypes import grpc_to_enum__qkd_app_types
+LOGGER = logging.getLogger(__name__)
-#from .Events import notify_event_context, notify_event_device, notify_event_topology
+def app_list_objs(db_engine: Engine, context_uuid: str = None) -> AppList:
+ """
+ Fetches a list of all QKD applications from the database. Optionally filters by context UUID.
-LOGGER = logging.getLogger(__name__)
+ :param db_engine: SQLAlchemy Engine for DB connection
+ :param context_uuid: UUID of the context to filter by (optional)
+ :return: AppList containing all apps
+ """
+ def callback(session: Session) -> List[Dict]:
+ query = session.query(AppModel)
+
+ if context_uuid:
+ query = query.filter_by(context_uuid=context_uuid)
+ return [obj.dump() for obj in query.all()]
-def app_list_objs(db_engine : Engine) -> AppList:
- def callback(session : Session) -> List[Dict]:
- obj_list : List[AppModel] = session.query(AppModel)\
- .all()
- return [obj.dump() for obj in obj_list]
apps = run_transaction(sessionmaker(bind=db_engine), callback)
return AppList(apps=apps)
-def app_get(db_engine : Engine, request : AppId) -> App:
+
+def app_get(db_engine: Engine, request: AppId) -> App:
+ """
+ Fetches a specific app by its UUID.
+
+ :param db_engine: SQLAlchemy Engine for DB connection
+ :param request: AppId protobuf containing app ID and context ID
+ :return: App protobuf object
+ :raises NotFoundException: If the app is not found in the database
+ """
app_uuid = app_get_uuid(request, allow_random=False)
- def callback(session : Session) -> Optional[Dict]:
- obj : Optional[AppModel] = session.query(AppModel)\
- .filter_by(app_uuid=app_uuid).one_or_none()
- return None if obj is None else obj.dump()
+
+ def callback(session: Session) -> Optional[Dict]:
+ obj = session.query(AppModel).filter_by(app_uuid=app_uuid).one_or_none()
+ return obj.dump() if obj else None
+
obj = run_transaction(sessionmaker(bind=db_engine), callback)
- if obj is None:
- raw_app_uuid = request.app_uuid.uuid
- raise NotFoundException('App', raw_app_uuid, extra_details=[
- 'app_uuid generated was: {:s}'.format(app_uuid)
+
+ if not obj:
+ raise NotFoundException('App', request.app_uuid.uuid, extra_details=[
+ f'app_uuid generated was: {app_uuid}'
])
+
return App(**obj)
-def app_set(db_engine : Engine, messagebroker : MessageBroker, request : App) -> AppId:
- context_uuid = context_get_uuid(request.app_id.context_id, allow_random=False)
- raw_app_uuid = request.app_id.app_uuid.uuid
- app_uuid = app_get_uuid(request.app_id, allow_random=True)
- app_type = request.app_type
- app_status = grpc_to_enum__qkd_app_status(request.app_status)
- app_type = grpc_to_enum__qkd_app_types(request.app_type)
+def app_set(db_engine: Engine, messagebroker: MessageBroker, request: App) -> AppId:
+ """
+ Creates or updates an app in the database. If the app already exists, updates the app.
+ Otherwise, inserts a new entry.
- now = datetime.datetime.utcnow()
+ :param db_engine: SQLAlchemy Engine for DB connection
+ :param messagebroker: MessageBroker instance for notifications
+ :param request: App protobuf object containing app data
+ :return: AppId protobuf object representing the newly created or updated app
+ """
+ context_uuid = context_get_uuid(request.app_id.context_id, allow_random=False)
+ app_uuid = app_get_uuid(request.app_id, allow_random=True)
-
- app_data = [{
- 'context_uuid' : context_uuid,
- 'app_uuid' : app_uuid,
- 'app_status' : app_status,
- 'app_type' : app_type,
- 'server_app_id' : request.server_app_id,
- 'client_app_id' : request.client_app_id,
- 'backing_qkdl_uuid' : [qkdl_id.qkdl_uuid.uuid for qkdl_id in request.backing_qkdl_id],
- 'local_device_uuid' : request.local_device_id.device_uuid.uuid,
- 'remote_device_uuid' : request.remote_device_id.device_uuid.uuid or None,
- 'created_at' : now,
- 'updated_at' : now,
- }]
-
-
- def callback(session : Session) -> Tuple[bool, List[Dict]]:
+ # Prepare app data for insertion/update
+ app_data = {
+ 'context_uuid': context_uuid,
+ 'app_uuid': app_uuid,
+ 'app_status': grpc_to_enum__qkd_app_status(request.app_status),
+ 'app_type': grpc_to_enum__qkd_app_types(request.app_type),
+ 'server_app_id': request.server_app_id,
+ 'client_app_id': request.client_app_id,
+ 'backing_qkdl_uuid': [qkdl.qkdl_uuid.uuid for qkdl in request.backing_qkdl_id],
+ 'local_device_uuid': request.local_device_id.device_uuid.uuid,
+ 'remote_device_uuid': request.remote_device_id.device_uuid.uuid if request.remote_device_id.device_uuid.uuid else None,
+ 'created_at': datetime.datetime.utcnow(),
+ 'updated_at': datetime.datetime.utcnow(),
+ }
+
+ def callback(session: Session) -> bool:
+ # Create the insert statement
stmt = insert(AppModel).values(app_data)
+
+ # Apply the conflict resolution
stmt = stmt.on_conflict_do_update(
index_elements=[AppModel.app_uuid],
set_=dict(
- app_status = stmt.excluded.app_status,
- app_type = stmt.excluded.app_type,
- server_app_id = stmt.excluded.server_app_id,
- client_app_id = stmt.excluded.client_app_id,
- backing_qkdl_uuid = stmt.excluded.backing_qkdl_uuid,
- local_device_uuid = stmt.excluded.local_device_uuid,
- remote_device_uuid = stmt.excluded.remote_device_uuid,
- updated_at = stmt.excluded.updated_at,
+ app_status=stmt.excluded.app_status,
+ app_type=stmt.excluded.app_type,
+ server_app_id=stmt.excluded.server_app_id,
+ client_app_id=stmt.excluded.client_app_id,
+ backing_qkdl_uuid=stmt.excluded.backing_qkdl_uuid,
+ local_device_uuid=stmt.excluded.local_device_uuid,
+ remote_device_uuid=stmt.excluded.remote_device_uuid,
+ updated_at=stmt.excluded.updated_at
)
)
- stmt = stmt.returning(AppModel.created_at, AppModel.updated_at)
- created_at,updated_at = session.execute(stmt).fetchone()
- updated = updated_at > created_at
-
- return updated
-
- updated = run_transaction(sessionmaker(bind=db_engine), callback)
- context_id = json_context_id(context_uuid)
- app_id = json_app_id(app_uuid, context_id=context_id)
- #event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE
- #notify_event_app(messagebroker, event_type, app_id)
- #notify_event_context(messagebroker, EventTypeEnum.EVENTTYPE_UPDATE, context_id)
+ session.execute(stmt)
+ return True
+
+ run_transaction(sessionmaker(bind=db_engine), callback)
+ app_id = json_app_id(app_uuid, context_id=json_context_id(context_uuid))
+
return AppId(**app_id)
+def app_get_by_server(db_engine: Engine, server_app_id: str) -> App:
+ """
+ Fetches an app by its server_app_id.
+ """
+ def callback(session: Session) -> Optional[Dict]:
+ obj = session.query(AppModel).filter_by(server_app_id=server_app_id).one_or_none()
+ return obj.dump() if obj else None
-def app_get_by_server(db_engine : Engine, request : str) -> App:
- def callback(session : Session) -> Optional[Dict]:
- obj : Optional[AppModel] = session.query(AppModel)\
- .filter_by(server_app_id=request).one_or_none()
- return None if obj is None else obj.dump()
obj = run_transaction(sessionmaker(bind=db_engine), callback)
- if obj is None:
- raise NotFoundException('No app match found for', request)
+
+ if not obj:
+ raise NotFoundException('App', server_app_id)
+
return App(**obj)
+def app_delete(db_engine: Engine, app_uuid: str) -> None:
+ """
+ Deletes an app by its UUID from the database.
+
+ :param db_engine: SQLAlchemy Engine for DB connection
+ :param app_uuid: The UUID of the app to be deleted
+ """
+ def callback(session: Session) -> bool:
+ app_obj = session.query(AppModel).filter_by(app_uuid=app_uuid).one_or_none()
+
+ if app_obj is None:
+ raise NotFoundException('App', app_uuid)
+
+ session.delete(app_obj)
+ return True
-"""
-def device_delete(db_engine : Engine, messagebroker : MessageBroker, request : DeviceId) -> Empty:
- device_uuid = device_get_uuid(request, allow_random=False)
- def callback(session : Session) -> Tuple[bool, List[Dict]]:
- query = session.query(TopologyDeviceModel)
- query = query.filter_by(device_uuid=device_uuid)
- topology_device_list : List[TopologyDeviceModel] = query.all()
- topology_ids = [obj.topology.dump_id() for obj in topology_device_list]
- num_deleted = session.query(DeviceModel).filter_by(device_uuid=device_uuid).delete()
- return num_deleted > 0, topology_ids
- deleted, updated_topology_ids = run_transaction(sessionmaker(bind=db_engine), callback)
- device_id = json_device_id(device_uuid)
- if deleted:
- notify_event_device(messagebroker, EventTypeEnum.EVENTTYPE_REMOVE, device_id)
-
- context_ids : Dict[str, Dict] = dict()
- topology_ids : Dict[str, Dict] = dict()
- for topology_id in updated_topology_ids:
- topology_uuid = topology_id['topology_uuid']['uuid']
- topology_ids[topology_uuid] = topology_id
- context_id = topology_id['context_id']
- context_uuid = context_id['context_uuid']['uuid']
- context_ids[context_uuid] = context_id
-
- for topology_id in topology_ids.values():
- notify_event_topology(messagebroker, EventTypeEnum.EVENTTYPE_UPDATE, topology_id)
-
- for context_id in context_ids.values():
- notify_event_context(messagebroker, EventTypeEnum.EVENTTYPE_UPDATE, context_id)
-
- return Empty()
-
-def device_select(db_engine : Engine, request : DeviceFilter) -> DeviceList:
- device_uuids = [
- device_get_uuid(device_id, allow_random=False)
- for device_id in request.device_ids.device_ids
- ]
- dump_params = dict(
- include_endpoints =request.include_endpoints,
- include_config_rules=request.include_config_rules,
- include_components =request.include_components,
- )
- def callback(session : Session) -> List[Dict]:
- query = session.query(DeviceModel)
- if request.include_endpoints : query = query.options(selectinload(DeviceModel.endpoints))
- if request.include_config_rules: query = query.options(selectinload(DeviceModel.config_rules))
- #if request.include_components : query = query.options(selectinload(DeviceModel.components))
- obj_list : List[DeviceModel] = query.filter(DeviceModel.device_uuid.in_(device_uuids)).all()
- return [obj.dump(**dump_params) for obj in obj_list]
- devices = run_transaction(sessionmaker(bind=db_engine), callback)
- return DeviceList(devices=devices)
-"""
+ run_transaction(sessionmaker(bind=db_engine), callback)
diff --git a/src/qkd_app/service/database/models/QKDAppModel.py b/src/qkd_app/service/database/models/QKDAppModel.py
index c32b4e28c95105d8659cb52790f51b330764c2cf..d9797d73c9c168efee5810014a70e0425f67e051 100644
--- a/src/qkd_app/service/database/models/QKDAppModel.py
+++ b/src/qkd_app/service/database/models/QKDAppModel.py
@@ -12,52 +12,70 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import operator
-from sqlalchemy import CheckConstraint, Column, DateTime, Float, Enum, ForeignKey, Integer, String
-from sqlalchemy.dialects.postgresql import UUID, ARRAY
-from sqlalchemy.orm import relationship
+from sqlalchemy import Column, DateTime, String, Enum, ARRAY
+from sqlalchemy.dialects.postgresql import UUID
from typing import Dict
from ._Base import _Base
from .enums.QKDAppStatus import ORM_QKDAppStatusEnum
from .enums.QKDAppTypes import ORM_QKDAppTypesEnum
+
class AppModel(_Base):
+ """
+ ORM model representing a QKD (Quantum Key Distribution) Application.
+ This model stores information about the QKD app status, type, and related device and
+ backing QKD links. It is stored in the 'qkd_app' table.
+ """
__tablename__ = 'qkd_app'
- app_uuid = Column(UUID(as_uuid=False), primary_key=True)
- context_uuid = Column(UUID(as_uuid=False), nullable=False) # Supposed to be Foreign Key
- app_status = Column(Enum(ORM_QKDAppStatusEnum), nullable=False)
- app_type = Column(Enum(ORM_QKDAppTypesEnum), nullable=False)
- server_app_id = Column(String, nullable=False)
- client_app_id = Column(ARRAY(String), nullable=False)
- backing_qkdl_uuid = Column(ARRAY(UUID(as_uuid=False)), nullable=False)
- local_device_uuid = Column(UUID(as_uuid=False), nullable=False)
- remote_device_uuid = Column(UUID(as_uuid=False), nullable=True)
+ # Primary Key
+ app_uuid = Column(UUID(as_uuid=False), primary_key=True, nullable=False, doc="Unique identifier for the QKD app.")
+
+ # Foreign Key-like field (context)
+ context_uuid = Column(UUID(as_uuid=False), nullable=False, doc="Foreign key linking to the application's context.")
+
+ # Status and type
+ app_status = Column(Enum(ORM_QKDAppStatusEnum), nullable=False, doc="Current status of the QKD app.")
+ app_type = Column(Enum(ORM_QKDAppTypesEnum), nullable=False, doc="Type of the QKD app (internal or client).")
- # Optare: Created_at and Updated_at are only used to know if an app was updated later on the code. Don't change it
+ # Application IDs
+ server_app_id = Column(String, nullable=False, doc="ID of the server-side QKD application.")
+ client_app_id = Column(ARRAY(String), nullable=False, doc="List of client-side QKD application IDs.")
- created_at = Column(DateTime, nullable=False)
- updated_at = Column(DateTime, nullable=False)
+ # Backing QKD links and devices
+ backing_qkdl_uuid = Column(ARRAY(UUID(as_uuid=False)), nullable=False, doc="List of UUIDs of the backing QKD links.")
+ local_device_uuid = Column(UUID(as_uuid=False), nullable=False, doc="UUID of the local QKD device.")
+ remote_device_uuid = Column(UUID(as_uuid=False), nullable=True, doc="UUID of the remote QKD device (nullable).")
- #__table_args__ = (
- # CheckConstraint(... >= 0, name='name_value_...'),
- #)
+ # Timestamps
+ created_at = Column(DateTime, nullable=False, doc="Timestamp when the QKD app record was created.")
+ updated_at = Column(DateTime, nullable=False, doc="Timestamp when the QKD app record was last updated.")
def dump_id(self) -> Dict:
+ """
+ Serializes the primary key fields (context and app UUID) into a dictionary.
+
+ :return: A dictionary with 'context_id' and 'app_uuid' keys.
+ """
return {
'context_id': {'context_uuid': {'uuid': self.context_uuid}},
'app_uuid': {'uuid': self.app_uuid}
}
def dump(self) -> Dict:
- result = {
- 'app_id' : self.dump_id(),
- 'app_status' : self.app_status.value,
- 'app_type' : self.app_type.value,
- 'server_app_id' : self.server_app_id,
- 'client_app_id' : self.client_app_id,
- 'backing_qkdl_id' : [{'qkdl_uuid': {'uuid': qkdl_id}} for qkdl_id in self.backing_qkdl_uuid],
- 'local_device_id' : {'device_uuid': {'uuid': self.local_device_uuid}},
- 'remote_device_id' : {'device_uuid': {'uuid': self.remote_device_uuid}},
+ """
+ Serializes the entire QKD app model into a dictionary, including app status, type, IDs,
+ device info, and backing QKD links.
+
+ :return: A dictionary representation of the QKD app.
+ """
+ return {
+ 'app_id': self.dump_id(),
+ 'app_status': self.app_status.value,
+ 'app_type': self.app_type.value,
+ 'server_app_id': self.server_app_id,
+ 'client_app_id': self.client_app_id,
+ 'backing_qkdl_id': [{'qkdl_uuid': {'uuid': qkdl_id}} for qkdl_id in self.backing_qkdl_uuid],
+ 'local_device_id': {'device_uuid': {'uuid': self.local_device_uuid}},
+ 'remote_device_id': {'device_uuid': {'uuid': self.remote_device_uuid}} if self.remote_device_uuid else None,
}
- return result
diff --git a/src/qkd_app/service/database/models/_Base.py b/src/qkd_app/service/database/models/_Base.py
index 51863e1d5c06a875c298eab726cfdc3b7fcb75ca..f17fb9a56dcd120ad1dc95ceee720aa942f6ae6c 100644
--- a/src/qkd_app/service/database/models/_Base.py
+++ b/src/qkd_app/service/database/models/_Base.py
@@ -13,32 +13,17 @@
# limitations under the License.
import sqlalchemy
-from typing import Any, List
-from sqlalchemy.orm import Session, sessionmaker, declarative_base
-from sqlalchemy.sql import text
-from sqlalchemy_cockroachdb import run_transaction
+from sqlalchemy.orm import declarative_base
_Base = declarative_base()
-'''
-def create_performance_enhancers(db_engine : sqlalchemy.engine.Engine) -> None:
- def index_storing(
- index_name : str, table_name : str, index_fields : List[str], storing_fields : List[str]
- ) -> Any:
- str_index_fields = ','.join(['"{:s}"'.format(index_field) for index_field in index_fields])
- str_storing_fields = ','.join(['"{:s}"'.format(storing_field) for storing_field in storing_fields])
- INDEX_STORING = 'CREATE INDEX IF NOT EXISTS {:s} ON "{:s}" ({:s}) STORING ({:s});'
- return text(INDEX_STORING.format(index_name, table_name, str_index_fields, str_storing_fields))
-
- statements = [
- # In case of relations
- ]
- def callback(session : Session) -> bool:
- for stmt in statements: session.execute(stmt)
- run_transaction(sessionmaker(bind=db_engine), callback)
-'''
-
-def rebuild_database(db_engine : sqlalchemy.engine.Engine, drop_if_exists : bool = False):
- if drop_if_exists: _Base.metadata.drop_all(db_engine)
+def rebuild_database(db_engine: sqlalchemy.engine.Engine, drop_if_exists: bool = False):
+ """
+ Rebuild the database schema for the QKD application. Optionally drop the existing schema if specified.
+
+ :param db_engine: SQLAlchemy engine instance.
+ :param drop_if_exists: Boolean indicating if the schema should be dropped before rebuilding.
+ """
+ if drop_if_exists:
+ _Base.metadata.drop_all(db_engine)
_Base.metadata.create_all(db_engine)
- #create_performance_enhancers(db_engine)
diff --git a/src/qkd_app/service/database/models/enums/QKDAppStatus.py b/src/qkd_app/service/database/models/enums/QKDAppStatus.py
index d3063ef56704ce1bdd48d15ea8c6486ed7c8cfae..980a8b14adca8c69eaf567037dc09872f33622f9 100644
--- a/src/qkd_app/service/database/models/enums/QKDAppStatus.py
+++ b/src/qkd_app/service/database/models/enums/QKDAppStatus.py
@@ -12,16 +12,19 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import enum, functools
+import enum
+import functools
from common.proto.qkd_app_pb2 import QKDAppStatusEnum
from ._GrpcToEnum import grpc_to_enum
+# Enum mapping for ORM-based app statuses.
class ORM_QKDAppStatusEnum(enum.Enum):
- ON = QKDAppStatusEnum.QKDAPPSTATUS_ON
+ ON = QKDAppStatusEnum.QKDAPPSTATUS_ON
DISCONNECTED = QKDAppStatusEnum.QKDAPPSTATUS_DISCONNECTED
- OUT_OF_TIME = QKDAppStatusEnum.QKDAPPSTATUS_OUT_OF_TIME
- ZOMBIE = QKDAppStatusEnum.QKDAPPSTATUS_ZOMBIE
-
+ OUT_OF_TIME = QKDAppStatusEnum.QKDAPPSTATUS_OUT_OF_TIME
+ ZOMBIE = QKDAppStatusEnum.QKDAPPSTATUS_ZOMBIE
+# Function to map between gRPC and ORM enums.
grpc_to_enum__qkd_app_status = functools.partial(
- grpc_to_enum, QKDAppStatusEnum, ORM_QKDAppStatusEnum)
+ grpc_to_enum, QKDAppStatusEnum, ORM_QKDAppStatusEnum
+)
diff --git a/src/qkd_app/service/database/models/enums/QKDAppTypes.py b/src/qkd_app/service/database/models/enums/QKDAppTypes.py
index f50b8982d80c0af97c2cbd96d336f450afc50f9b..fdd318cec49b7b7ec8e1a0347c8b0f8c1aeb3a5c 100644
--- a/src/qkd_app/service/database/models/enums/QKDAppTypes.py
+++ b/src/qkd_app/service/database/models/enums/QKDAppTypes.py
@@ -12,14 +12,17 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import enum, functools
+import enum
+import functools
from common.proto.qkd_app_pb2 import QKDAppTypesEnum
from ._GrpcToEnum import grpc_to_enum
+# Enum mapping for ORM-based app types.
class ORM_QKDAppTypesEnum(enum.Enum):
INTERNAL = QKDAppTypesEnum.QKDAPPTYPES_INTERNAL
- CLIENT = QKDAppTypesEnum.QKDAPPTYPES_CLIENT
-
+ CLIENT = QKDAppTypesEnum.QKDAPPTYPES_CLIENT
+# Function to map between gRPC and ORM enums.
grpc_to_enum__qkd_app_types = functools.partial(
- grpc_to_enum, QKDAppTypesEnum, ORM_QKDAppTypesEnum)
+ grpc_to_enum, QKDAppTypesEnum, ORM_QKDAppTypesEnum
+)
diff --git a/src/qkd_app/service/database/uuids/QKDApp.py b/src/qkd_app/service/database/uuids/QKDApp.py
index 175f1d5f3cf4ceda12a022b4afadb376e11ae5a5..f6c0e1e12d591e747adf9497c77177e0fdd4dc51 100644
--- a/src/qkd_app/service/database/uuids/QKDApp.py
+++ b/src/qkd_app/service/database/uuids/QKDApp.py
@@ -16,15 +16,22 @@ from common.proto.qkd_app_pb2 import AppId
from common.method_wrappers.ServiceExceptions import InvalidArgumentsException
from ._Builder import get_uuid_from_string, get_uuid_random
-def app_get_uuid(
- app_id : AppId, allow_random : bool = False
-) -> str:
+def app_get_uuid(app_id: AppId, allow_random: bool = False) -> str:
+ """
+ Retrieves or generates the UUID for an app.
+
+ :param app_id: AppId object that contains the app UUID
+ :param allow_random: If True, generates a random UUID if app_uuid is not set
+ :return: App UUID as a string
+ """
app_uuid = app_id.app_uuid.uuid
- if len(app_uuid) > 0:
+ if app_uuid:
return get_uuid_from_string(app_uuid)
- if allow_random: return get_uuid_random()
+
+ if allow_random:
+ return get_uuid_random()
raise InvalidArgumentsException([
('app_id.app_uuid.uuid', app_uuid),
- ], extra_details=['At least one is required to produce a App UUID'])
+ ], extra_details=['At least one UUID is required to identify the app.'])
diff --git a/src/qkd_app/service/database/uuids/_Builder.py b/src/qkd_app/service/database/uuids/_Builder.py
index 39c98de69d577ce2722693e57c4ee678124f9e30..c5996b0f9d2cced27fd05e3966c3a60fa9bae24d 100644
--- a/src/qkd_app/service/database/uuids/_Builder.py
+++ b/src/qkd_app/service/database/uuids/_Builder.py
@@ -15,30 +15,37 @@
from typing import Optional, Union
from uuid import UUID, uuid4, uuid5
-# Generate a UUIDv5-like from the SHA-1 of "TFS" and no namespace to be used as the NAMESPACE for all
-# the context UUIDs generated. For efficiency purposes, the UUID is hardcoded; however, it is produced
-# using the following code:
-# from hashlib import sha1
-# from uuid import UUID
-# hash = sha1(bytes('TFS', 'utf-8')).digest()
-# NAMESPACE_TFS = UUID(bytes=hash[:16], version=5)
+# Hardcoded namespace for generating UUIDs.
NAMESPACE_TFS = UUID('200e3a1f-2223-534f-a100-758e29c37f40')
-def get_uuid_from_string(str_uuid_or_name : Union[str, UUID], prefix_for_name : Optional[str] = None) -> str:
- # if UUID given, assume it is already a valid UUID
- if isinstance(str_uuid_or_name, UUID): return str_uuid_or_name
+def get_uuid_from_string(str_uuid_or_name: Union[str, UUID], prefix_for_name: Optional[str] = None) -> str:
+ """
+ Convert a string or UUID object into a UUID string. If input is a name, generate a UUID using the TFS namespace.
+
+ :param str_uuid_or_name: Input string or UUID to be converted into UUID format
+ :param prefix_for_name: Optional prefix to add before the name when generating a name-based UUID
+ :return: A valid UUID string
+ :raises ValueError: If the input is invalid and cannot be converted to a UUID
+ """
+ if isinstance(str_uuid_or_name, UUID):
+ return str(str_uuid_or_name) # Ensure returning a string representation
+
if not isinstance(str_uuid_or_name, str):
- MSG = 'Parameter({:s}) cannot be used to produce a UUID'
- raise Exception(MSG.format(str(repr(str_uuid_or_name))))
+ raise ValueError(f"Invalid parameter ({repr(str_uuid_or_name)}). Expected a string or UUID to produce a valid UUID.")
+
try:
- # try to parse as UUID
+ # Try to interpret the input as a UUID
return str(UUID(str_uuid_or_name))
- except: # pylint: disable=bare-except
- # produce a UUID within TFS namespace from parameter
- if prefix_for_name is not None:
- str_uuid_or_name = '{:s}/{:s}'.format(prefix_for_name, str_uuid_or_name)
+ except ValueError:
+ # If the input isn't a valid UUID, generate one using the name-based approach
+ if prefix_for_name:
+ str_uuid_or_name = f"{prefix_for_name}/{str_uuid_or_name}"
return str(uuid5(NAMESPACE_TFS, str_uuid_or_name))
def get_uuid_random() -> str:
- # Generate random UUID. No need to use namespace since "namespace + random = random".
+ """
+ Generate and return a new random UUID as a string.
+
+ :return: A randomly generated UUID string
+ """
return str(uuid4())
diff --git a/src/qkd_app/service/rest_server/RestServer.py b/src/qkd_app/service/rest_server/RestServer.py
deleted file mode 100644
index e21531c5bcf0e1cf15a8f08952d6325a8349f398..0000000000000000000000000000000000000000
--- a/src/qkd_app/service/rest_server/RestServer.py
+++ /dev/null
@@ -1,23 +0,0 @@
-# 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.
-
-from common.Constants import ServiceNameEnum
-from common.Settings import get_service_baseurl_http, get_service_port_http
-from common.tools.service.GenericRestServer import GenericRestServer
-
-class RestServer(GenericRestServer):
- def __init__(self, cls_name: str = __name__) -> None:
- bind_port = get_service_port_http(ServiceNameEnum.QKD_APP)
- base_url = get_service_baseurl_http(ServiceNameEnum.QKD_APP)
- super().__init__(bind_port, base_url, cls_name=cls_name)
diff --git a/src/qkd_app/service/rest_server/__init__.py b/src/qkd_app/service/rest_server/__init__.py
deleted file mode 100644
index 07d08814021ef82220611ee21c01ba01806682e9..0000000000000000000000000000000000000000
--- a/src/qkd_app/service/rest_server/__init__.py
+++ /dev/null
@@ -1,13 +0,0 @@
-# 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.
diff --git a/src/service/service/service_handlers/qkd/qkd_service_handler.py b/src/service/service/service_handlers/qkd/qkd_service_handler.py
index 0977388005ef72fe036de93de2dc73438f0c6163..2bfbcb59dd045d10b9267dc15119e1d17e1929d4 100644
--- a/src/service/service/service_handlers/qkd/qkd_service_handler.py
+++ b/src/service/service/service_handlers/qkd/qkd_service_handler.py
@@ -17,7 +17,8 @@ import json, logging, uuid
from typing import Any, Dict, List, Optional, Tuple, Union
from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method
from common.proto.context_pb2 import ConfigRule, DeviceId, Service
-from common.proto.qkd_app_pb2 import App, QKDAppStatusEnum, QKDAppTypesEnum
+from common.proto.qkd_app_pb2 import App, AppId, QKDAppStatusEnum, QKDAppTypesEnum
+from common.proto.context_pb2 import ContextId, Uuid
from common.tools.object_factory.ConfigRule import json_config_rule_delete, json_config_rule_set
from common.tools.object_factory.Device import json_device_id
from common.type_checkers.Checkers import chk_type
@@ -41,6 +42,7 @@ class QKDServiceHandler(_ServiceHandler):
self.__service = service
self.__task_executor = task_executor
self.__settings_handler = SettingsHandler(service.service_config, **settings)
+ self.qkd_app_client = task_executor._qkd_app_client # Initialize qkd_app_client
# Optare: This function is where the service is created
@@ -88,14 +90,9 @@ class QKDServiceHandler(_ServiceHandler):
interfaces.append([0,0])
links.append([])
-
-
-
-
endpoint_left = get_endpoint_name_by_uuid(device, endpoint_left_uuid) if idx > 0 else None
endpoint_right = get_endpoint_name_by_uuid(device, endpoint_right_uuid) if 2 * idx + 2 < len(endpoints) else None
-
for config_rule in device.device_config.config_rules:
resource_key = config_rule.custom.resource_key
@@ -271,7 +268,7 @@ class QKDServiceHandler(_ServiceHandler):
'remote_device_id': dst_device.device_id,
}
- self.__task_executor.register_app(App(**internal_app_src_dst))
+ self.__task_executor.register_qkd_app(App(**internal_app_src_dst))
# Register App
@@ -286,7 +283,7 @@ class QKDServiceHandler(_ServiceHandler):
'remote_device_id': src_device.device_id,
}
- self.__task_executor.register_app(App(**internal_app_dst_src))
+ self.__task_executor.register_qkd_app(App(**internal_app_dst_src))
results.append(True)
except Exception as e: # pylint: disable=broad-except
@@ -297,31 +294,127 @@ class QKDServiceHandler(_ServiceHandler):
# Optare: This will be to delete a service
def DeleteEndpoint(
- self, endpoints : List[Tuple[str, str, Optional[str]]],
- connection_uuid : Optional[str] = None
+ self, endpoints: List[Tuple[str, str, Optional[str]]], connection_uuid: Optional[str] = None
) -> List[Union[bool, Exception]]:
- """ Delete service endpoints form a list.
- Parameters:
- endpoints: List[Tuple[str, str, Optional[str]]]
- List of tuples, each containing a device_uuid,
- endpoint_uuid, and the topology_uuid of the endpoint
- to be removed.
- connection_uuid : Optional[str]
- If specified, is the UUID of the connection this endpoint is associated to.
- Returns:
- results: List[Union[bool, Exception]]
- List of results for endpoint deletions requested.
- Return values must be in the same order as the requested
- endpoints. If an endpoint is properly deleted, True must be
- returned; otherwise, the Exception that is raised during
- the processing must be returned.
- """
- raise NotImplementedError()
+ chk_type('endpoints', endpoints, list)
+ if len(endpoints) == 0:
+ return []
+
+ LOGGER.info(f'Deleting Endpoints: {endpoints}')
+ LOGGER.info(f'Connection UUID: {connection_uuid}')
+
+ service_uuid = self.__service.service_id.service_uuid.uuid
+ context_uuid = self.__service.service_id.context_id.context_uuid.uuid
+ LOGGER.info(f'Service UUID: {service_uuid}, Context UUID: {context_uuid}')
+
+ results = []
+ apps = list() # Initialize apps as an empty list, in case fetching fails
+ try:
+ # Initialize device lists and QKDN IDs
+ devices = []
+ qkdn_ids = []
+ interfaces = []
+ links = []
+
+ # Populate devices and QKDN ids from endpoints
+ for idx, endpoint in enumerate(endpoints[::2]):
+ device_uuid, endpoint_left_uuid = get_device_endpoint_uuids(endpoint)
+ _, endpoint_right_uuid = get_device_endpoint_uuids(endpoints[2 * idx + 1])
+
+ device = self.__task_executor.get_device(DeviceId(**json_device_id(device_uuid)))
+ LOGGER.info(f'Device: {device}, Endpoint Left: {endpoint_left_uuid}, Endpoint Right: {endpoint_right_uuid}')
+
+ devices.append(device)
+ interfaces.append([0, 0])
+ links.append([])
+
+ for config_rule in device.device_config.config_rules:
+ resource_key = config_rule.custom.resource_key
+
+ if resource_key == '__node__':
+ value = json.loads(config_rule.custom.resource_value)
+ qkdn_ids.append(value['qkdn_id'])
+
+ elif resource_key.startswith('/interface'):
+ value = json.loads(config_rule.custom.resource_value)
+ try:
+ endpoint_str = value['qkdi_att_point']['uuid']
+ if endpoint_str == endpoint_left_uuid:
+ interfaces[idx][0] = value['qkdi_id']
+ elif endpoint_str == endpoint_right_uuid:
+ interfaces[idx][1] = value['qkdi_id']
+ except KeyError:
+ pass
+
+ elif resource_key.startswith('/link'):
+ value = json.loads(config_rule.custom.resource_value)
+ links[idx].append((
+ value['uuid'],
+ (value['src_qkdn_id'], value['src_interface_id']),
+ (value['dst_qkdn_id'], value['dst_interface_id'])
+ ))
+
+ LOGGER.info(f'Interfaces: {interfaces}, Links: {links}, QKDN IDs: {qkdn_ids}')
+
+ # Fetch the related apps for the service using the same pattern as in routes.py
+ try:
+ context_id = ContextId(context_uuid=Uuid(uuid=context_uuid))
+ apps_response = self.__task_executor._qkd_app_client.ListApps(context_id)
+ apps = apps_response.apps # Assign the apps to the list, if successful
+ LOGGER.info(f"Apps retrieved: {apps}")
+ except grpc.RpcError as e:
+ LOGGER.error(f"gRPC error while fetching apps: {e.details()}")
+ if e.code() != grpc.StatusCode.NOT_FOUND:
+ raise
+ apps = list() # If an error occurs, ensure `apps` is still an empty list
+
+ # Filter related internal apps
+ related_apps = [
+ app for app in apps
+ if app.server_app_id == service_uuid and app.app_type == QKDAppTypesEnum.QKDAPPTYPES_INTERNAL
+ ]
+
+ # Log each app's details
+ for app in related_apps:
+ LOGGER.info(f"App ID: {app.app_id.app_uuid.uuid}, Status: {app.app_status}")
+
+ # Update each app status to DISCONNECTED before deletion
+ for app in related_apps:
+ self.__task_executor.update_qkd_app_status(app, QKDAppStatusEnum.QKDAPPSTATUS_DISCONNECTED)
+
+ results.append(True)
+
+ except Exception as e: # pylint: disable=broad-except
+ LOGGER.error(f"Failed to delete QKD service: {str(e)}")
+ results.append(e)
+
+ return results
+
+ def fetch_related_internal_apps(self, context_uuid: str, service_uuid: str) -> List[App]:
+ try:
+ context_id = ContextId(context_uuid=Uuid(uuid=context_uuid))
+ apps_response = self.qkd_app_client.ListApps(context_id)
+
+ # Log the apps retrieved to ensure they exist and have a status
+ LOGGER.info(f"Apps retrieved: {apps_response.apps}")
+
+ internal_apps = [
+ app for app in apps_response.apps
+ if app.app_type == QKDAppTypesEnum.QKDAPPTYPES_INTERNAL
+ and app.server_app_id == service_uuid
+ and app.app_status == QKDAppStatusEnum.ACTIVE # Ensure you are checking status
+ ]
+
+ LOGGER.info(f"Filtered internal apps: {internal_apps}")
+ return internal_apps
+
+ except Exception as e:
+ LOGGER.error(f"Error fetching related internal apps: {e}")
+ return []
# Optare: Can be ingored. It's in case if a service is later updated. Not required to proper functioning
- def SetConstraint(self, constraints: List[Tuple[str, Any]]) \
- -> List[Union[bool, Exception]]:
+ def SetConstraint(self, constraints: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
""" Create/Update service constraints.
Parameters:
constraints: List[Tuple[str, Any]]
@@ -335,10 +428,30 @@ class QKDServiceHandler(_ServiceHandler):
returned; otherwise, the Exception that is raised during
the processing must be returned.
"""
- raise NotImplementedError()
+ results = []
+ try:
+ for constraint_type, constraint_value in constraints:
+ LOGGER.info(f"Setting constraint: {constraint_type} with value: {constraint_value}")
+
+ # Assuming you store constraints as part of service config rules
+ constraint_key = f"/constraints/{constraint_type}"
+ json_config_rule = json_config_rule_set(constraint_key, constraint_value)
- def DeleteConstraint(self, constraints: List[Tuple[str, Any]]) \
- -> List[Union[bool, Exception]]:
+ # Apply the configuration rule to the service
+ self.__service.service_config.config_rules.append(ConfigRule(**json_config_rule))
+
+ # Reconfigure the service with new constraints
+ self.__task_executor.configure_service(self.__service)
+
+ results.append(True)
+
+ except Exception as e:
+ LOGGER.error(f"Failed to set constraints: {str(e)}")
+ results.append(e)
+
+ return results
+
+ def DeleteConstraint(self, constraints: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
""" Delete service constraints.
Parameters:
constraints: List[Tuple[str, Any]]
@@ -354,16 +467,37 @@ class QKDServiceHandler(_ServiceHandler):
be returned; otherwise, the Exception that is raised during
the processing must be returned.
"""
- raise NotImplementedError()
+ results = []
+ try:
+ for constraint_type, _ in constraints:
+ LOGGER.info(f"Deleting constraint: {constraint_type}")
+
+ # Remove the constraint from the service config rules
+ constraint_key = f"/constraints/{constraint_type}"
+ json_config_rule = json_config_rule_delete(constraint_key)
+
+ for rule in self.__service.service_config.config_rules:
+ if rule.custom.resource_key == constraint_key:
+ self.__service.service_config.config_rules.remove(rule)
+
+ # Reconfigure the service after removing constraints
+ self.__task_executor.configure_service(self.__service)
+
+ results.append(True)
+
+ except Exception as e:
+ LOGGER.error(f"Failed to delete constraints: {str(e)}")
+ results.append(e)
- def SetConfig(self, resources: List[Tuple[str, Any]]) \
- -> List[Union[bool, Exception]]:
+ return results
+
+ def SetConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
""" Create/Update configuration for a list of service resources.
Parameters:
resources: List[Tuple[str, Any]]
List of tuples, each containing a resource_key pointing to
- the resource to be modified, and a resource_value
- containing the new value to be set.
+ the resource to be modified, and a resource_value containing
+ the new value to be set.
Returns:
results: List[Union[bool, Exception]]
List of results for resource key changes requested.
@@ -372,10 +506,28 @@ class QKDServiceHandler(_ServiceHandler):
returned; otherwise, the Exception that is raised during
the processing must be returned.
"""
- raise NotImplementedError()
+ results = []
+ try:
+ for resource_key, resource_value in resources:
+ LOGGER.info(f"Setting config: {resource_key} with value: {resource_value}")
+
+ json_config_rule = json_config_rule_set(resource_key, resource_value)
+
+ # Apply the configuration rule to the service
+ self.__service.service_config.config_rules.append(ConfigRule(**json_config_rule))
+
+ # Reconfigure the service with new configurations
+ self.__task_executor.configure_service(self.__service)
+
+ results.append(True)
+
+ except Exception as e:
+ LOGGER.error(f"Failed to set config: {str(e)}")
+ results.append(e)
+
+ return results
- def DeleteConfig(self, resources: List[Tuple[str, Any]]) \
- -> List[Union[bool, Exception]]:
+ def DeleteConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
""" Delete configuration for a list of service resources.
Parameters:
resources: List[Tuple[str, Any]]
@@ -391,4 +543,25 @@ class QKDServiceHandler(_ServiceHandler):
be returned; otherwise, the Exception that is raised during
the processing must be returned.
"""
- raise NotImplementedError()
+ results = []
+ try:
+ for resource_key, _ in resources:
+ LOGGER.info(f"Deleting config: {resource_key}")
+
+ json_config_rule = json_config_rule_delete(resource_key)
+
+ # Remove the matching configuration rule
+ for rule in self.__service.service_config.config_rules:
+ if rule.custom.resource_key == resource_key:
+ self.__service.service_config.config_rules.remove(rule)
+
+ # Reconfigure the service after deleting configurations
+ self.__task_executor.configure_service(self.__service)
+
+ results.append(True)
+
+ except Exception as e:
+ LOGGER.error(f"Failed to delete config: {str(e)}")
+ results.append(e)
+
+ return results
diff --git a/src/service/service/task_scheduler/TaskExecutor.py b/src/service/service/task_scheduler/TaskExecutor.py
index cb27993702963b4aac88ec04eca2a1c796d0c364..67f6a516c9bc3031b5cdd2aed80cd6fdd7e1c9c2 100644
--- a/src/service/service/task_scheduler/TaskExecutor.py
+++ b/src/service/service/task_scheduler/TaskExecutor.py
@@ -16,11 +16,14 @@ import json, logging
from enum import Enum
from typing import TYPE_CHECKING, Any, Dict, Optional, Union
from common.method_wrappers.ServiceExceptions import NotFoundException
+from typing import List
+from common.proto.qkd_app_pb2 import QKDAppStatusEnum
from common.proto.context_pb2 import (
Connection, ConnectionId, Device, DeviceDriverEnum, DeviceId, Service, ServiceId,
OpticalConfig, OpticalConfigId
)
-from common.proto.qkd_app_pb2 import App
+from common.proto.qkd_app_pb2 import App, AppId
+from common.proto.context_pb2 import ContextId
from common.tools.context_queries.Connection import get_connection_by_id
from common.tools.context_queries.Device import get_device
from common.tools.context_queries.Service import get_service_by_id
@@ -33,7 +36,7 @@ from service.service.service_handler_api.Exceptions import (
UnsatisfiedFilterException, UnsupportedFilterFieldException, UnsupportedFilterFieldValueException
)
from service.service.service_handler_api.ServiceHandlerFactory import ServiceHandlerFactory, get_service_handler_class
-from service.service.tools.ObjectKeys import get_connection_key, get_device_key, get_service_key, get_app_key
+from service.service.tools.ObjectKeys import get_connection_key, get_device_key, get_service_key, get_qkd_app_key
if TYPE_CHECKING:
from service.service.service_handler_api._ServiceHandler import _ServiceHandler
@@ -229,8 +232,56 @@ class TaskExecutor:
# ----- QkdApp-related methods -------------------------------------------------------------------------------------
- def register_app(self, app: App) -> None:
- app_key = get_app_key(app.app_id)
+ def register_qkd_app(self, app: App) -> None:
+ """
+ Registers a QKD App and stores it in the cache.
+ """
+ qkd_app_key = get_qkd_app_key(app.app_id)
self._qkd_app_client.RegisterApp(app)
- LOGGER.info("reg registered")
- self._store_grpc_object(CacheableObjectType.QKD_APP, app_key, app)
+ LOGGER.info("QKD app registered with key: %s", qkd_app_key)
+ self._store_grpc_object(CacheableObjectType.QKD_APP, qkd_app_key, app)
+
+ def update_qkd_app_status(self, app: App, new_status: QKDAppStatusEnum) -> None:
+ """
+ Updates the status of a QKD app and persists it to the database.
+ """
+ try:
+ app.app_status = new_status
+ LOGGER.info(f"Attempting to update app {app.app_id.app_uuid.uuid} to status {new_status}")
+ self._qkd_app_client.UpdateApp(app)
+ LOGGER.info(f"Successfully updated app {app.app_id.app_uuid.uuid} to status {new_status}")
+ except Exception as e:
+ LOGGER.error(f"Failed to update QKD app {app.app_id.app_uuid.uuid}: {str(e)}")
+ raise e
+
+ def list_qkd_apps(self, context_id: ContextId) -> List[App]:
+ """
+ Retrieves a list of QKD apps from the QKD App service.
+ """
+ try:
+ apps_response = self._qkd_app_client.ListApps(context_id)
+ LOGGER.info(f"ListApps retrieved: {len(apps_response.apps)} apps with status")
+
+ # Ensure that the status is logged and used
+ for app in apps_response.apps:
+ LOGGER.info(f"App ID: {app.app_id.app_uuid.uuid}, Status: {app.app_status}")
+
+ return apps_response.apps
+ except Exception as e:
+ LOGGER.error(f"Failed to list QKD apps: {str(e)}")
+ return []
+
+ def delete_qkd_app(self, app_id: AppId) -> None:
+ """
+ Deletes a QKD App by its AppId and removes it from the cache.
+ """
+ qkd_app_key = get_qkd_app_key(app_id)
+ try:
+ LOGGER.info(f"Attempting to delete QKD app with AppId: {app_id}")
+ self._qkd_app_client.DeleteApp(app_id)
+ LOGGER.info(f"QKD app deleted with key: {qkd_app_key}")
+ self._delete_grpc_object(CacheableObjectType.QKD_APP, qkd_app_key)
+ except Exception as e:
+ LOGGER.error(f"Failed to delete QKD app with AppId {app_id}: {str(e)}")
+ raise e
+
diff --git a/src/service/service/tools/ObjectKeys.py b/src/service/service/tools/ObjectKeys.py
index cfc719bba736a4ea0789b028a97ca267b2d04089..f67cb02e143b78127484d6644a7fdd8c9c71e29c 100644
--- a/src/service/service/tools/ObjectKeys.py
+++ b/src/service/service/tools/ObjectKeys.py
@@ -26,6 +26,6 @@ def get_service_key(service_id : ServiceId) -> str:
service_uuid = service_id.service_uuid.uuid
return '{:s}/{:s}'.format(context_uuid, service_uuid)
-def get_app_key(app_id : AppId) -> str:
+def get_qkd_app_key(app_id: AppId) -> str:
return app_id.app_uuid.uuid
diff --git a/src/telemetry/backend/service/TelemetryBackendService.py b/src/telemetry/backend/service/TelemetryBackendService.py
index f3cf18d65eac0fda6c56a366ce8d806a4ea2562a..79a35d343860d19992518c0e8b29e427e5cbbef4 100755
--- a/src/telemetry/backend/service/TelemetryBackendService.py
+++ b/src/telemetry/backend/service/TelemetryBackendService.py
@@ -55,7 +55,7 @@ class TelemetryBackendService(GenericGrpcService):
listener for requests on Kafka topic.
"""
LOGGER.info('Telemetry backend request listener is running ...')
- print ('Telemetry backend request listener is running ...')
+ # print ('Telemetry backend request listener is running ...')
consumer = self.kafka_consumer
consumer.subscribe([KafkaTopic.REQUEST.value])
while True:
@@ -66,13 +66,13 @@ class TelemetryBackendService(GenericGrpcService):
if receive_msg.error().code() == KafkaError._PARTITION_EOF:
continue
else:
- print("Consumer error: {}".format(receive_msg.error()))
+ # print("Consumer error: {}".format(receive_msg.error()))
break
try:
collector = json.loads(receive_msg.value().decode('utf-8'))
collector_id = receive_msg.key().decode('utf-8')
LOGGER.debug('Recevied Collector: {:} - {:}'.format(collector_id, collector))
- print('Recevied Collector: {:} - {:}'.format(collector_id, collector))
+ # print('Recevied Collector: {:} - {:}'.format(collector_id, collector))
if collector['duration'] == -1 and collector['interval'] == -1:
self.TerminateCollectorBackend(collector_id)
@@ -80,18 +80,19 @@ class TelemetryBackendService(GenericGrpcService):
self.RunInitiateCollectorBackend(collector_id, collector)
except Exception as e:
LOGGER.warning("Unable to consumer message from topic: {:}. ERROR: {:}".format(KafkaTopic.REQUEST.value, e))
- print ("Unable to consumer message from topic: {:}. ERROR: {:}".format(KafkaTopic.REQUEST.value, e))
+ # print ("Unable to consumer message from topic: {:}. ERROR: {:}".format(KafkaTopic.REQUEST.value, e))
def TerminateCollectorBackend(self, collector_id):
if collector_id in self.running_threads:
thread, stop_event = self.running_threads[collector_id]
stop_event.set()
thread.join()
- print ("Terminating backend (by StopCollector): Collector Id: ", collector_id)
+ # print ("Terminating backend (by StopCollector): Collector Id: ", collector_id)
del self.running_threads[collector_id]
self.GenerateCollectorTerminationSignal(collector_id, "-1", -1) # Termination confirmation to frontend.
else:
- print ('Backend collector {:} not found'.format(collector_id))
+ # print ('Backend collector {:} not found'.format(collector_id))
+ LOGGER.warning('Backend collector {:} not found'.format(collector_id))
def RunInitiateCollectorBackend(self, collector_id: str, collector: str):
stop_event = threading.Event()
@@ -104,7 +105,8 @@ class TelemetryBackendService(GenericGrpcService):
"""
Method receives collector request and initiates collecter backend.
"""
- print("Initiating backend for collector: ", collector_id)
+ # print("Initiating backend for collector: ", collector_id)
+ LOGGER.info("Initiating backend for collector: ", collector_id)
start_time = time.time()
while not stop_event.is_set():
if int(collector['duration']) != -1 and time.time() - start_time >= collector['duration']: # condition to terminate backend
@@ -136,8 +138,7 @@ class TelemetryBackendService(GenericGrpcService):
Method to extract kpi value.
"""
measured_kpi_value = random.randint(1,100) # TODO: To be extracted from a device
- print ("Measured Kpi value: {:}".format(measured_kpi_value))
- # measured_kpi_value = self.fetch_node_exporter_metrics() # exporter extracted metric value against default KPI
+ # print ("Measured Kpi value: {:}".format(measured_kpi_value))
self.GenerateCollectorResponse(collector_id, kpi_id , measured_kpi_value)
def GenerateCollectorResponse(self, collector_id: str, kpi_id: str, measured_kpi_value: Any):
@@ -166,7 +167,7 @@ class TelemetryBackendService(GenericGrpcService):
"""
if err:
LOGGER.debug('Message delivery failed: {:}'.format(err))
- print(f'Message delivery failed: {err}')
+ # print(f'Message delivery failed: {err}')
else:
LOGGER.info('Message delivered to topic {:}'.format(msg.topic()))
- print(f'Message delivered to topic {msg.topic()}')
+ # print(f'Message delivered to topic {msg.topic()}')
diff --git a/src/telemetry/backend/tests/test_backend.py b/src/telemetry/backend/tests/test_backend.py
index 4764d7f5f10aefe211bae840f06eed9c82386bf8..8bbde9769ae1dfb16a33ef528f74031d2ba94c01 100644
--- a/src/telemetry/backend/tests/test_backend.py
+++ b/src/telemetry/backend/tests/test_backend.py
@@ -26,12 +26,12 @@ LOGGER = logging.getLogger(__name__)
###########################
# --- "test_validate_kafka_topics" should be run before the functionality tests ---
-# def test_validate_kafka_topics():
-# LOGGER.debug(" >>> test_validate_kafka_topics: START <<< ")
-# response = KafkaTopic.create_all_topics()
-# assert isinstance(response, bool)
+def test_validate_kafka_topics():
+ LOGGER.debug(" >>> test_validate_kafka_topics: START <<< ")
+ response = KafkaTopic.create_all_topics()
+ assert isinstance(response, bool)
-def test_RunRequestListener():
- LOGGER.info('test_RunRequestListener')
- TelemetryBackendServiceObj = TelemetryBackendService()
- threading.Thread(target=TelemetryBackendServiceObj.RequestListener).start()
\ No newline at end of file
+# def test_RunRequestListener():
+# LOGGER.info('test_RunRequestListener')
+# TelemetryBackendServiceObj = TelemetryBackendService()
+# threading.Thread(target=TelemetryBackendServiceObj.RequestListener).start()
\ No newline at end of file
diff --git a/src/telemetry/frontend/service/TelemetryFrontendServiceServicerImpl.py b/src/telemetry/frontend/service/TelemetryFrontendServiceServicerImpl.py
index ad99dff12dc641232972f8cff8226878caefd71b..5c569e2ddd1d75dd89f88fe9ae08517330470254 100644
--- a/src/telemetry/frontend/service/TelemetryFrontendServiceServicerImpl.py
+++ b/src/telemetry/frontend/service/TelemetryFrontendServiceServicerImpl.py
@@ -153,7 +153,7 @@ class TelemetryFrontendServiceServicerImpl(TelemetryFrontendServiceServicer):
"""
if err:
LOGGER.debug('Message delivery failed: {:}'.format(err))
- print('Message delivery failed: {:}'.format(err))
+ # print('Message delivery failed: {:}'.format(err))
# else:
# LOGGER.debug('Message delivered to topic {:}'.format(msg.topic()))
# print('Message delivered to topic {:}'.format(msg.topic()))
@@ -177,7 +177,8 @@ class TelemetryFrontendServiceServicerImpl(TelemetryFrontendServiceServicer):
if receive_msg.error().code() == KafkaError._PARTITION_EOF:
continue
else:
- print("Consumer error: {}".format(receive_msg.error()))
+ # print("Consumer error: {:}".format(receive_msg.error()))
+ LOGGER.error("Consumer error: {:}".format(receive_msg.error()))
break
try:
collector_id = receive_msg.key().decode('utf-8')
@@ -185,13 +186,17 @@ class TelemetryFrontendServiceServicerImpl(TelemetryFrontendServiceServicer):
kpi_value = json.loads(receive_msg.value().decode('utf-8'))
self.process_response(collector_id, kpi_value['kpi_id'], kpi_value['kpi_value'])
else:
- print(f"collector id does not match.\nRespone ID: '{collector_id}' --- Active IDs: '{ACTIVE_COLLECTORS}' ")
+ # print(f"collector id does not match.\nRespone ID: '{collector_id}' --- Active IDs: '{ACTIVE_COLLECTORS}' ")
+ LOGGER.info("collector id does not match.\nRespone ID: {:} --- Active IDs: {:}".format(collector_id, ACTIVE_COLLECTORS))
except Exception as e:
- print(f"Error extarcting msg key or value: {str(e)}")
+ # print(f"Error extarcting msg key or value: {str(e)}")
+ LOGGER.info("Error extarcting msg key or value: {:}".format(e))
continue
def process_response(self, collector_id: str, kpi_id: str, kpi_value: Any):
if kpi_id == "-1" and kpi_value == -1:
- print ("Backend termination confirmation for collector id: ", collector_id)
+ # print ("Backend termination confirmation for collector id: ", collector_id)
+ LOGGER.info("Backend termination confirmation for collector id: ", collector_id)
else:
- print ("KPI Value: Collector Id:", collector_id, ", Kpi Id:", kpi_id, ", Value:", kpi_value)
+ LOGGER.info("Backend termination confirmation for collector id: ", collector_id)
+ # print ("KPI Value: Collector Id:", collector_id, ", Kpi Id:", kpi_id, ", Value:", kpi_value)
diff --git a/src/telemetry/requirements.in b/src/telemetry/requirements.in
index 4e9981afbcb7460344f160dcecf329f44d14e792..503468a662599f0225b293d0ef4c4e4313fa3e0f 100644
--- a/src/telemetry/requirements.in
+++ b/src/telemetry/requirements.in
@@ -17,4 +17,4 @@ psycopg2-binary==2.9.3
python-dateutil==2.8.2
python-json-logger==2.0.2
pytz==2024.1
-requests==2.27.1
\ No newline at end of file
+requests==2.27.1
diff --git a/src/tests/tools/mock_qkd_nodes/start.sh b/src/tests/tools/mock_qkd_nodes/start.sh
index faf2f84baf61f16565b497b53bf5f41f45007c00..89797b9c9496cdf58061c406ad2886be0d9c47f6 100755
--- a/src/tests/tools/mock_qkd_nodes/start.sh
+++ b/src/tests/tools/mock_qkd_nodes/start.sh
@@ -13,18 +13,30 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+#!/bin/bash
cd "$(dirname "$0")"
+# Function to kill all background processes
killbg() {
- for p in "${pids[@]}" ; do
- kill "$p";
- done
+ for p in "${pids[@]}" ; do
+ kill "$p";
+ done
}
trap killbg EXIT
pids=()
+
+# Set FLASK_APP and run the Flask instances on different ports
+export FLASK_APP=wsgi
flask run --host 0.0.0.0 --port 11111 &
pids+=($!)
+
flask run --host 0.0.0.0 --port 22222 &
pids+=($!)
-flask run --host 0.0.0.0 --port 33333
+
+flask run --host 0.0.0.0 --port 33333 &
+pids+=($!)
+
+# Wait for all background processes to finish
+wait
+
diff --git a/src/webui/service/qkd_app/routes.py b/src/webui/service/qkd_app/routes.py
index 71243fb75e552ec5568eedacdcadabbc39516b4e..200d6ebdd17918dbb3d026980c4286212f519b82 100644
--- a/src/webui/service/qkd_app/routes.py
+++ b/src/webui/service/qkd_app/routes.py
@@ -12,21 +12,23 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import grpc, json, logging
-
+import grpc
+import logging
from flask import current_app, render_template, Blueprint, flash, session, redirect, url_for
-from common.proto.context_pb2 import Empty, Link, LinkId, LinkList
-from common.proto.qkd_app_pb2 import App, QKDAppStatusEnum, QKDAppTypesEnum
+from common.proto.qkd_app_pb2 import App, AppId, QKDAppStatusEnum, QKDAppTypesEnum
+from common.proto.context_pb2 import Uuid, ContextId
from common.tools.context_queries.Context import get_context
from common.tools.context_queries.Device import get_device
-from common.tools.context_queries.Topology import get_topology
from context.client.ContextClient import ContextClient
from qkd_app.client.QKDAppClient import QKDAppClient
-
+# Set up logging
LOGGER = logging.getLogger(__name__)
+
+# Blueprint for QKDApp routes
qkd_app = Blueprint('qkd_app', __name__, url_prefix='/qkd_app')
+# Initialize clients
qkd_app_client = QKDAppClient()
context_client = ContextClient()
@@ -35,79 +37,131 @@ def home():
if 'context_uuid' not in session or 'topology_uuid' not in session:
flash("Please select a context!", "warning")
return redirect(url_for("main.home"))
+
context_uuid = session['context_uuid']
topology_uuid = session['topology_uuid']
+ # Connect to context client
context_client.connect()
device_names = dict()
- context_obj = get_context(context_client, context_uuid, rw_copy=False)
- if context_obj is None:
- flash('Context({:s}) not found'.format(str(context_uuid)), 'danger')
- apps = list()
- else:
- try:
- apps = qkd_app_client.ListApps(context_obj.context_id)
- apps = apps.apps
- except grpc.RpcError as e:
- if e.code() != grpc.StatusCode.NOT_FOUND: raise
- if e.details() != 'Context({:s}) not found'.format(context_uuid): raise
+ try:
+
+ # Fetch context object
+ context_obj = get_context(context_client, context_uuid, rw_copy=False)
+ if context_obj is None:
+ flash('Context({:s}) not found'.format(str(context_uuid)), 'danger')
apps = list()
else:
- # Too many requests to context_client if it has too many apps (update in the future)
- for app in apps:
- if app.local_device_id.device_uuid.uuid not in device_names:
- device = get_device(context_client, app.local_device_id.device_uuid.uuid)
- if device is not None:
- device_names[app.local_device_id.device_uuid.uuid] = device.name
-
- if app.remote_device_id.device_uuid.uuid and app.remote_device_id.device_uuid.uuid not in device_names:
- device = get_device(context_client, app.remote_device_id.device_uuid.uuid)
- if device is not None:
- device_names[app.remote_device_id.device_uuid.uuid] = device.name
-
- context_client.close()
- return render_template(
- 'qkd_app/home.html', apps=apps, device_names=device_names, ate=QKDAppTypesEnum, ase=QKDAppStatusEnum)
+ try:
+ # Call ListApps using the context_id
+ apps_response = qkd_app_client.ListApps(context_obj.context_id)
+ apps = apps_response.apps
+ except grpc.RpcError as e:
+ LOGGER.error(f"gRPC error while fetching apps: {e.details()}")
+ if e.code() != grpc.StatusCode.NOT_FOUND: raise
+ if e.details() != 'Context({:s}) not found'.format(context_uuid): raise
+ apps = list()
+ else:
+ # Map local and remote device names
+ for app in apps:
+ if app.local_device_id.device_uuid.uuid not in device_names:
+ device = get_device(context_client, app.local_device_id.device_uuid.uuid)
+ if device:
+ device_names[app.local_device_id.device_uuid.uuid] = device.name
+ if app.remote_device_id.device_uuid.uuid and app.remote_device_id.device_uuid.uuid not in device_names:
+ device = get_device(context_client, app.remote_device_id.device_uuid.uuid)
+ if device:
+ device_names[app.remote_device_id.device_uuid.uuid] = device.name
+ finally:
+ context_client.close()
+
+ # Render the template with app list and device names
+ return render_template(
+ 'qkd_app/home.html',
+ apps=apps,
+ device_names=device_names,
+ ate=QKDAppTypesEnum,
+ ase=QKDAppStatusEnum
+ )
-@qkd_app.route('detail/<path:app_uuid>', methods=('GET', 'POST'))
+@qkd_app.route('detail/<path:app_uuid>', methods=['GET', 'POST'])
def detail(app_uuid: str):
- '''
- context_client.connect()
- link_obj = get_link(context_client, link_uuid, rw_copy=False)
- if link_obj is None:
- flash('Link({:s}) not found'.format(str(link_uuid)), 'danger')
- link_obj = Link()
- device_names, endpoints_data = dict(), dict()
- else:
- device_names, endpoints_data = get_endpoint_names(context_client, link_obj.link_endpoint_ids)
- context_client.close()
- return render_template('link/detail.html',link=link_obj, device_names=device_names, endpoints_data=endpoints_data)
- '''
- pass
+ """
+ Displays details for a specific QKD app identified by its UUID.
+ """
+ try:
+ qkd_app_client.connect()
+
+ # Wrap the app_uuid in a Uuid object and fetch details
+ uuid_message = Uuid(uuid=app_uuid)
+ app_id = AppId(app_uuid=uuid_message)
+ app_detail = qkd_app_client.GetApp(app_id)
+
+ if not app_detail:
+ flash(f"App with UUID {app_uuid} not found", "danger")
+ return redirect(url_for("qkd_app.home"))
+
+ # Fetch device details
+ context_client.connect()
+ device_names = {}
+
+ try:
+ if app_detail.local_device_id.device_uuid.uuid:
+ local_device = get_device(context_client, app_detail.local_device_id.device_uuid.uuid)
+ if local_device:
+ device_names[app_detail.local_device_id.device_uuid.uuid] = local_device.name
+
+ if app_detail.remote_device_id.device_uuid.uuid:
+ remote_device = get_device(context_client, app_detail.remote_device_id.device_uuid.uuid)
+ if remote_device:
+ device_names[app_detail.remote_device_id.device_uuid.uuid] = remote_device.name
+
+ except grpc.RpcError as e:
+ LOGGER.error(f"Failed to retrieve device details for app {app_uuid}: {e}")
+ flash(f"Error retrieving device details: {e.details()}", "danger")
+ return redirect(url_for("qkd_app.home"))
+
+ finally:
+ context_client.close()
+
+ return render_template(
+ 'qkd_app/detail.html',
+ app=app_detail,
+ ase=QKDAppStatusEnum,
+ ate=QKDAppTypesEnum,
+ device_names=device_names
+ )
+
+ except grpc.RpcError as e:
+ LOGGER.error(f"Failed to retrieve app details for {app_uuid}: {e}")
+ flash(f"Error retrieving app details: {e.details()}", "danger")
+ return redirect(url_for("qkd_app.home"))
+
+ finally:
+ qkd_app_client.close()
@qkd_app.get('<path:app_uuid>/delete')
-def delete(app_uuid):
- '''
+def delete(app_uuid: str):
+ """
+ Deletes a specific QKD app identified by its UUID.
+ """
try:
+ request = AppId(app_uuid=Uuid(uuid=app_uuid))
- # first, check if link exists!
- # request: LinkId = LinkId()
- # request.link_uuid.uuid = link_uuid
- # response: Link = client.GetLink(request)
- # TODO: finalize implementation
+ qkd_app_client.connect()
+ qkd_app_client.DeleteApp(request) # Call the DeleteApp method
+ qkd_app_client.close()
- request = LinkId()
- request.link_uuid.uuid = link_uuid # pylint: disable=no-member
- context_client.connect()
- context_client.RemoveLink(request)
- context_client.close()
+ flash(f'App "{app_uuid}" deleted successfully!', 'success')
+
+ except grpc.RpcError as e:
+ LOGGER.error(f"Problem deleting app {app_uuid}: {e}")
+ flash(f"Problem deleting app {app_uuid}: {e.details()}", 'danger')
+
+ except Exception as e:
+ LOGGER.exception(f"Unexpected error while deleting app {app_uuid}: {e}")
+ flash(f"Unexpected error: {str(e)}", 'danger')
- flash(f'Link "{link_uuid}" deleted successfully!', 'success')
- except Exception as e: # pylint: disable=broad-except
- flash(f'Problem deleting link "{link_uuid}": {e.details()}', 'danger')
- current_app.logger.exception(e)
- return redirect(url_for('link.home'))
- '''
- pass
+ return redirect(url_for('qkd_app.home'))
diff --git a/src/webui/service/templates/qkd_app/detail.html b/src/webui/service/templates/qkd_app/detail.html
new file mode 100644
index 0000000000000000000000000000000000000000..078895cec81b4680715f9d027beac3158afdc4a4
--- /dev/null
+++ b/src/webui/service/templates/qkd_app/detail.html
@@ -0,0 +1,126 @@
+{% extends 'base.html' %}
+
+{% block content %}
+<h1>App {{ app.app_id.app_uuid.uuid }}</h1>
+
+<!-- Action Buttons -->
+<div class="row mb-3">
+ <div class="col-sm-3">
+ <button type="button" class="btn btn-success" onclick="window.location.href='{{ url_for('qkd_app.home') }}';">
+ <i class="bi bi-box-arrow-in-left"></i> Back to app list
+ </button>
+ </div>
+ <div class="col-sm-3">
+ <button type="button" class="btn btn-danger" data-bs-toggle="modal" data-bs-target="#deleteModal">
+ <i class="bi bi-x-square"></i> Delete app
+ </button>
+ </div>
+</div>
+
+<!-- App Basic Info -->
+<div class="row mb-3">
+ <div class="col-sm-4">
+ <b>UUID: </b>{{ app.app_id.app_uuid.uuid }}<br>
+ <b>Status: </b>{{ ase.Name(app.app_status).replace('QKDAPPSTATUS_', '') }}<br>
+ <b>Type: </b>{{ ate.Name(app.app_type).replace('QKDAPPTYPES_', '').replace('CLIENT', 'EXTERNAL') }}<br>
+ </div>
+
+ <!-- Associated Devices -->
+ <div class="col-sm-8">
+ <h5>Associated Devices</h5>
+ <table class="table table-striped table-hover">
+ <thead>
+ <tr>
+ <th scope="col">Device</th>
+ <th scope="col">Endpoint Type</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>
+ <a href="{{ url_for('device.detail', device_uuid=app.local_device_id.device_uuid.uuid) }}">
+ {{ device_names.get(app.local_device_id.device_uuid.uuid, app.local_device_id.device_uuid.uuid) }}
+ <svg xmlns="http://www.w3.org/2000/svg" width="16" height="16" fill="currentColor" class="bi bi-eye" viewBox="0 0 16 16">
+ <path d="M16 8s-3-5.5-8-5.5S0 8 0 8s3 5.5 8 5.5S16 8 16 8z"/>
+ <path d="M8 5.5a2.5 2.5 0 1 0 0 5 2.5 2.5 0 0 0 0-5z"/>
+ <path d="M4.5 8a3.5 3.5 0 1 1 7 0 3.5 3.5 0 0 1-7 0z"/>
+ </svg>
+ </a>
+ </td>
+ <td>Local Device</td>
+ </tr>
+ {% if app.remote_device_id.device_uuid.uuid %}
+ <tr>
+ <td>
+ <a href="{{ url_for('device.detail', device_uuid=app.remote_device_id.device_uuid.uuid) }}">
+ {{ device_names.get(app.remote_device_id.device_uuid.uuid, app.remote_device_id.device_uuid.uuid) }}
+ <svg xmlns="http://www.w3.org/2000/svg" width="16" height="16" fill="currentColor" class="bi bi-eye" viewBox="0 0 16 16">
+ <path d="M16 8s-3-5.5-8-5.5S0 8 0 8s3 5.5 8 5.5S16 8 16 8z"/>
+ <path d="M8 5.5a2.5 2.5 0 1 0 0 5 2.5 2.5 0 0 0 0-5z"/>
+ <path d="M4.5 8a3.5 3.5 0 1 1 7 0 3.5 3.5 0 0 1-7 0z"/>
+ </svg>
+ </a>
+ </td>
+ <td>Remote Device</td>
+ </tr>
+ {% endif %}
+ </tbody>
+ </table>
+ </div>
+</div>
+
+<!-- QoS -->
+<div class="row mb-3">
+ <div class="col-sm-12">
+ <h5>App QoS</h5>
+ <table class="table table-striped">
+ <thead>
+ <tr>
+ <th scope="col">QoS Parameter</th>
+ <th scope="col">Value</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>Max Bandwidth</td>
+ <td>{{ app.qos.max_bandwidth }} bps</td>
+ </tr>
+ <tr>
+ <td>Min Bandwidth</td>
+ <td>{{ app.qos.min_bandwidth or 'N/A' }} bps</td>
+ </tr>
+ <tr>
+ <td>Jitter</td>
+ <td>{{ app.qos.jitter or 'N/A' }} ms</td>
+ </tr>
+ <tr>
+ <td>TTL</td>
+ <td>{{ app.qos.ttl or 'N/A' }} seconds</td>
+ </tr>
+ </tbody>
+ </table>
+ </div>
+</div>
+
+<!-- Modal for Deleting App -->
+<div class="modal fade" id="deleteModal" data-bs-backdrop="static" data-bs-keyboard="false" tabindex="-1" aria-labelledby="staticBackdropLabel" aria-hidden="true">
+ <div class="modal-dialog">
+ <div class="modal-content">
+ <div class="modal-header">
+ <h5 class="modal-title" id="staticBackdropLabel">Delete app?</h5>
+ <button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="Close"></button>
+ </div>
+ <div class="modal-body">
+ Are you sure you want to delete the app "{{ app.app_id.app_uuid.uuid }}"?
+ </div>
+ <div class="modal-footer">
+ <button type="button" class="btn btn-secondary" data-bs-dismiss="modal">No</button>
+ <a type="button" class="btn btn-danger" href="{{ url_for('qkd_app.delete', app_uuid=app.app_id.app_uuid.uuid) }}">
+ <i class="bi bi-exclamation-diamond"></i>Yes
+ </a>
+ </div>
+ </div>
+ </div>
+</div>
+
+{% endblock %}
diff --git a/src/webui/service/templates/qkd_app/home.html b/src/webui/service/templates/qkd_app/home.html
index 9573013f41410a5d8560e71c174ce6a85237089f..39b43ecc4b9b71e03d27bf080ae6c7be2bf1a90d 100644
--- a/src/webui/service/templates/qkd_app/home.html
+++ b/src/webui/service/templates/qkd_app/home.html
@@ -74,14 +74,12 @@
{% endif %}
</td>
<td>
- <!--
- <a href="{{ url_for('qkd_app.detail', app_uuid=app.app_id.app_uuid.uuid) }}">
- <svg xmlns="http://www.w3.org/2000/svg" width="16" height="16" fill="currentColor" class="bi bi-eye" viewBox="0 0 16 16">
- <path d="M16 8s-3-5.5-8-5.5S0 8 0 8s3 5.5 8 5.5S16 8 16 8zM1.173 8a13.133 13.133 0 0 1 1.66-2.043C4.12 4.668 5.88 3.5 8 3.5c2.12 0 3.879 1.168 5.168 2.457A13.133 13.133 0 0 1 14.828 8c-.058.087-.122.183-.195.288-.335.48-.83 1.12-1.465 1.755C11.879 11.332 10.119 12.5 8 12.5c-2.12 0-3.879-1.168-5.168-2.457A13.134 13.134 0 0 1 1.172 8z"/>
- <path d="M8 5.5a2.5 2.5 0 1 0 0 5 2.5 2.5 0 0 0 0-5zM4.5 8a3.5 3.5 0 1 1 7 0 3.5 3.5 0 0 1-7 0z"/>
- </svg>
- </a>
- -->
+ <a href="{{ url_for('qkd_app.detail', app_uuid=app.app_id.app_uuid.uuid) }}">
+ <svg xmlns="http://www.w3.org/2000/svg" width="16" height="16" fill="currentColor" class="bi bi-eye" viewBox="0 0 16 16">
+ <path d="M16 8s-3-5.5-8-5.5S0 8 0 8s3 5.5 8 5.5S16 8 16 8zM1.173 8a13.133 13.133 0 0 1 1.66-2.043C4.12 4.668 5.88 3.5 8 3.5c2.12 0 3.879 1.168 5.168 2.457A13.133 13.133 0 0 1 14.828 8c-.058.087-.122.183-.195.288-.335.48-.83 1.12-1.465 1.755C11.879 11.332 10.119 12.5 8 12.5c-2.12 0-3.879-1.168-5.168-2.457A13.134 13.134 0 0 1 1.172 8z"/>
+ <path d="M8 5.5a2.5 2.5 0 1 0 0 5 2.5 2.5 0 0 0 0-5zM4.5 8a3.5 3.5 0 1 1 7 0 3.5 3.5 0 0 1-7 0z"/>
+ </svg>
+ </a>
</td>
</tr>
{% endfor %}