Initial implementation of the gRPC calls.

manifests/cachingservice.yaml

-# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#      http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
----
-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  name: cachingservice
-spec:
-  selector:
-    matchLabels:
-      app: cachingservice
-  template:
-    metadata:
-      labels:
-        app: cachingservice
-    spec:
-      containers:
-      - name: redis
-        image: redis:7.0-alpine
-        env:
-        - name: REDIS_PASSWORD
-          valueFrom:
-            secretKeyRef:
-              name: redis-secrets
-              key: REDIS_PASSWORD
-        ports:
-        - containerPort: 6379
-          name: client
-        command: ["redis-server"]
-        args:
-        - --requirepass
-        - $(REDIS_PASSWORD)
----
-apiVersion: v1
-kind: Service
-metadata:
-  name: cachingservice
-spec:
-  type: ClusterIP
-  selector:
-    app: cachingservice
-  ports:
-  - name: redis
-    port: 6379
-    targetPort: 6379
\ No newline at end of file

proto/optical_attack_detector.proto

@@ -17,15 +17,10 @@ syntax = "proto3";
 package optical_attack_detector;
 
 import "context.proto";
-import "monitoring.proto";
 
 service OpticalAttackDetectorService {
   
   // rpc that triggers the attack detection loop
-  rpc DetectAttack        (context.Empty     ) returns (context.Empty) {}
+  rpc DetectAttack        (context.ServiceId     ) returns (context.Empty) {}
 
-  // rpc called by the distributed component to report KPIs
-  rpc ReportSummarizedKpi (monitoring.KpiList) returns (context.Empty) {}
-
-  rpc ReportKpi           (monitoring.KpiList) returns (context.Empty) {}
 }

src/dbscanserving/client/DbscanServingClient.py

@@ -53,5 +53,5 @@ class DbscanServingClient:
             request.num_features
         ))
         response = self.stub.Detect(request)
-        LOGGER.debug('Detect result: {:s}'.format(str(response)))
+        LOGGER.debug('Detect result with {:s} cluster indices'.format(len(response.cluster_indices)))
         return response

src/opticalattackdetector/requirements.in

-celery[redis]

src/opticalattackdetector/service/OpticalAttackDetectorServiceServicerImpl.py

@@ -33,7 +33,7 @@ from common.proto.optical_attack_detector_pb2_grpc import (
 LOGGER = logging.getLogger(__name__)
 
 SERVICE_NAME = 'OpticalAttackDetector'
-METHOD_NAMES = ['NotifyServiceUpdate', 'DetectAttack', 'ReportSummarizedKpi', 'ReportKpi']
+METHOD_NAMES = ['DetectAttack']
 METRICS = create_metrics(SERVICE_NAME, METHOD_NAMES)
 
 context_client: ContextClient = ContextClient()
@@ -50,65 +50,37 @@ class OpticalAttackDetectorServiceServicerImpl(OpticalAttackDetectorServiceServi
         LOGGER.debug('Servicer Created')
 
     @safe_and_metered_rpc_method(METRICS, LOGGER)
-    def DetectAttack(self, request : Empty, context : grpc.ServicerContext) -> Empty:
-        
-        # retrieve list with current contexts
-        # import pdb; pdb.set_trace()
-        context_ids: ContextIdList = context_client.ListContextIds(Empty())
-
-        # for each context, retrieve list of current services
-        services = []
-        for context_id in context_ids.context_ids:
-
-            context_services: ServiceIdList = context_client.ListServices(context_id)
-            for service in context_services.services:
-                services.append(service)
-
-        for service in services:
-            for endpoint in service.service_endpoint_ids:
-                # get instant KPI for this endpoint
-                LOGGER.warning(f'service: {service.service_id.service_uuid.uuid}\t endpoint: {endpoint.endpoint_uuid.uuid}\tdevice: {endpoint.device_id.device_uuid.uuid}')
-
-                # run attack detection for every service
-                request: DetectionRequest = DetectionRequest()
-
-                request.num_samples = 310
-                request.num_features = 100
-                request.eps = 100.5
-                request.min_samples = 50
-
-                for _ in range(200):
-                    grpc_sample = Sample()
-                    for __ in range(100):
-                        grpc_sample.features.append(random.uniform(0., 10.))
-                    request.samples.append(grpc_sample)
-                    
-                for _ in range(100):
-                    grpc_sample = Sample()
-                    for __ in range(100):
-                        grpc_sample.features.append(random.uniform(50., 60.))
-                    request.samples.append(grpc_sample)
-                    
-                for _ in range(10):
-                    grpc_sample = Sample()
-                    for __ in range(100):
-                        grpc_sample.features.append(random.uniform(5000., 6000.))
-                    request.samples.append(grpc_sample)
-
-                response: DetectionResponse = dbscanserving_client.Detect(request)
-
-                if -1 in response.cluster_indices:  # attack detected
-                    attack = AttackDescription()
-                    attack.cs_id.uuid = service.service_id.service_uuid.uuid
-                    response: AttackResponse = attack_mitigator_client.NotifyAttack(attack)
+    def DetectAttack(self, service_id : ServiceId, context : grpc.ServicerContext) -> Empty:
+        LOGGER.debug('Received request for {}/{}...'.format(
+            service_id.context_id.context_uuid.uuid,
+            service_id.service_uuid.uuid
+        ))
+        # run attack detection for every service
+        request: DetectionRequest = DetectionRequest()
+        request.num_samples = 310
+        request.num_features = 100
+        request.eps = 100.5
+        request.min_samples = 5
+        for _ in range(200):
+            grpc_sample = Sample()
+            for __ in range(100):
+                grpc_sample.features.append(random.uniform(0., 10.))
+            request.samples.append(grpc_sample)
+        for _ in range(100):
+            grpc_sample = Sample()
+            for __ in range(100):
+                grpc_sample.features.append(random.uniform(50., 60.))
+            request.samples.append(grpc_sample)
+        for _ in range(10):
+            grpc_sample = Sample()
+            for __ in range(100):
+                grpc_sample.features.append(random.uniform(5000., 6000.))
+            request.samples.append(grpc_sample)
+        response: DetectionResponse = dbscanserving_client.Detect(request)
+        if -1 in response.cluster_indices:  # attack detected
+            attack = AttackDescription()
+            attack.cs_id.uuid = service_id.service_uuid.uuid
+            response: AttackResponse = attack_mitigator_client.NotifyAttack(attack)
 
         # if attack is detected, run the attack mitigator
         return Empty()
-
-    @safe_and_metered_rpc_method(METRICS, LOGGER)
-    def ReportSummarizedKpi(self, request : KpiList, context : grpc.ServicerContext) -> Empty:
-        return Empty()
-
-    @safe_and_metered_rpc_method(METRICS, LOGGER)
-    def ReportKpi(self, request : KpiList, context : grpc.ServicerContext) -> Empty:
-        return Empty()

src/opticalattackdetector/service/__main__.py

@@ -15,7 +15,6 @@
 import logging, signal, sys, time, threading, random
 from multiprocessing import Manager, Process
 from prometheus_client import start_http_server
-from celery import Celery
 import asyncio
 
 from common.Constants import DEFAULT_GRPC_MAX_WORKERS, DEFAULT_GRPC_GRACE_PERIOD
@@ -47,28 +46,6 @@ def signal_handler(signal, frame): # pylint: disable=redefined-outer-name
     terminate.set()
 
 
-def detect_attack_old(monitoring_interval):
-    time.sleep(10)  # wait for the service to start
-    LOGGER.info("Starting the attack detection loop")
-    client = OpticalAttackDetectorClient(address='localhost', port=GRPC_SERVICE_PORT)
-    client.connect()
-    while True:  # infinite loop that runs until the terminate is set
-        if terminate.is_set():  # if terminate is set
-            LOGGER.warning("Stopping execution...")
-            client.close()
-            break  # break the while and stop execution
-        client.DetectAttack(Empty())
-        # sleep
-        LOGGER.debug("Sleeping for {} seconds...".format(monitoring_interval))
-        time.sleep(monitoring_interval)
-
-
-async def call_detection_grpc(request):
-    dbscanserving_client: DbscanServingClient = DbscanServingClient()
-    response: DetectionResponse = dbscanserving_client.Detect(request)
-    dbscanserving_client.close()
-    return result
-
 def main():
     global LOGGER # pylint: disable=global-statement
 
@@ -87,87 +64,13 @@ def main():
 
     # Start metrics server
     metrics_port = get_metrics_port()
-    start_http_server(metrics_port)
-
-    attack_mitigator_client: OpticalAttackMitigatorClient = OpticalAttackMitigatorClient()
-    monitoring_client: MonitoringClient = MonitoringClient()
+    # start_http_server(metrics_port)  # TODO: remove this comment
 
     # Starting CentralizedCybersecurity service
     grpc_service = OpticalAttackDetectorService(
         port=service_port, max_workers=max_workers, grace_period=grace_period)
     grpc_service.start()
 
-    # p = multiprocessing.Process(target=detect_attack, args=(monitoring_interval, ))
-    # p.start()
-    # detect_attack(monitoring_interval)
-
-    LOGGER.info('Connecting with REDIS...')
-    REDIS_PASSWORD = get_setting('REDIS_PASSWORD')
-    REDIS_HOST = get_setting('CACHINGSERVICE_SERVICE_HOST')
-    REDIS_PORT = get_setting('CACHINGSERVICE_SERVICE_PORT')
-    BROKER_URL = f'redis://:{REDIS_PASSWORD}@{REDIS_HOST}:{REDIS_PORT}/0'
-    BACKEND_URL = f'redis://:{REDIS_PASSWORD}@{REDIS_HOST}:{REDIS_PORT}/1'
-    app = Celery(
-        'cybersecurity',
-        broker=BROKER_URL,
-        backend=BACKEND_URL
-    )
-    LOGGER.info('Connected to REDIS...')
-
-    @app.task(name='detect_attack')
-    def detect_attack(context_id, service_id, kpi_id):
-        LOGGER.info('Detecting attack for {}/{}'.format(context_id, service_id))
-        alien_samples = random.randint(2, 10)
-        # run attack detection for every service
-        request: DetectionRequest = DetectionRequest()
-        request.num_samples = 200 + alien_samples
-        request.num_features = 20
-        request.eps = 100.5
-        request.min_samples = 5
-        for _ in range(200):
-            grpc_sample = Sample()
-            for __ in range(request.num_features):
-                grpc_sample.features.append(random.uniform(0., 10.))
-            request.samples.append(grpc_sample)
-        # for _ in range(100):
-        #     grpc_sample = Sample()
-        #     for __ in range(20):
-        #         grpc_sample.features.append(random.uniform(50., 60.))
-        #     request.samples.append(grpc_sample)                  
-        for _ in range(alien_samples):
-            grpc_sample = Sample()
-            for __ in range(request.num_features):
-                grpc_sample.features.append(random.uniform(5000., 6000.))
-            request.samples.append(grpc_sample)
-
-        # call the grpc
-        dbscanserving_client: DbscanServingClient = DbscanServingClient()
-        # response: DetectionResponse = dbscanserving_client.Detect(request)
-        dbscanserving_client.connect()
-        dbscanserving_client.close()
-
-        # including KPI
-        kpi = Kpi()
-        kpi.kpi_id.kpi_id.uuid = kpi_id
-        kpi.timestamp.timestamp = timestamp_utcnow_to_float()
-        # kpi.kpi_value.int32Val = response.cluster_indices[-1]
-        kpi.kpi_value.int32Val = 1
-        # monitoring_client.IncludeKpi(kpi)
-
-        # if -1 in response.cluster_indices:  # attack detected
-        #     attack = AttackDescription()
-        #     # attack.cs_id.uuid = service.service_id.service_uuid.uuid
-        #     response: AttackResponse = attack_mitigator_client.NotifyAttack(attack)
-        return "0"
-    
-    app.worker_main([
-        'worker',
-        '--loglevel={}'.format(log_level),
-        '--autoscale',
-        '1',
-        '--pool=gevent'
-    ])
-
     # Wait for Ctrl+C or termination signal
     while not terminate.wait(timeout=0.1): pass
 

src/opticalattackmanager/requirements.in

-celery[redis]
\ No newline at end of file

src/opticalattackmanager/send_task.py

-from celery import Celery
+import asyncio
+import logging
+
+import grpc
 import random
-from common.Settings import get_log_level, get_metrics_port, get_setting
+from common.Settings import get_log_level, get_setting
 from common.proto.dbscanserving_pb2 import DetectionRequest, DetectionResponse, Sample
-from dbscanserving.client.DbscanServingClient import DbscanServingClient
-
-
-dbscanserving_client: DbscanServingClient = DbscanServingClient()
-
-alien_samples = random.randint(2, 20)
-
-request: DetectionRequest = DetectionRequest()
-request.num_samples = 300 + alien_samples
-request.num_features = 20
-request.eps = 100.5
-request.min_samples = 5
-for _ in range(200):
-    grpc_sample = Sample()
-    for __ in range(20):
-        grpc_sample.features.append(random.uniform(0., 10.))
-    request.samples.append(grpc_sample)
-for _ in range(100):
-    grpc_sample = Sample()
-    for __ in range(20):
-        grpc_sample.features.append(random.uniform(50., 60.))
-    request.samples.append(grpc_sample)
-for _ in range(alien_samples):
-    grpc_sample = Sample()
-    for __ in range(20):
-        grpc_sample.features.append(random.uniform(5000., 6000.))
-    request.samples.append(grpc_sample)
-response: DetectionResponse = dbscanserving_client.Detect(request)
-
-REDIS_PASSWORD = get_setting('REDIS_PASSWORD')
-REDIS_HOST = get_setting('CACHINGSERVICE_SERVICE_HOST')
-REDIS_PORT = get_setting('CACHINGSERVICE_SERVICE_PORT')
-BROKER_URL = f'redis://:{REDIS_PASSWORD}@{REDIS_HOST}:{REDIS_PORT}/0'
-app = Celery(
-    'cybersecurity',
-    broker=BROKER_URL,
-    backend=BROKER_URL
-)
-
-service = {'context': 'admin', 'service': '23bb5c96-e377-4943-a47a-4db9c54104cc', 'kpi': '1'}
-result = app.send_task('detect_attack', (service['context'], service['service'], service['kpi']))
-print('waiting for result...')
-print('Result:', result.get())
+from common.proto.dbscanserving_pb2_grpc import DetectorStub
+
+# For more channel options, please see https://grpc.io/grpc/core/group__grpc__arg__keys.html
+CHANNEL_OPTIONS = [('grpc.lb_policy_name', 'pick_first'),
+                   ('grpc.enable_retries', True),
+                   ('grpc.keepalive_timeout_ms', 10000)]
+
+# based on https://github.com/grpc/grpc/blob/master/examples/python/helloworld/async_greeter_client_with_options.py
+
+async def run(endpoint, service_id) -> None:
+    
+    async with grpc.aio.insecure_channel(target=endpoint,
+                                         options=CHANNEL_OPTIONS) as channel:
+        stub = DetectorStub(channel)
+
+        # generate data
+        alien_samples = random.randint(2, 20)
+
+        request: DetectionRequest = DetectionRequest()
+        request.num_samples = 300 + alien_samples
+        request.num_features = 20
+        request.eps = 100.5
+        request.min_samples = 5
+        for _ in range(200):
+            grpc_sample = Sample()
+            for __ in range(20):
+                grpc_sample.features.append(random.uniform(0., 10.))
+            request.samples.append(grpc_sample)
+        for _ in range(100):
+            grpc_sample = Sample()
+            for __ in range(20):
+                grpc_sample.features.append(random.uniform(50., 60.))
+            request.samples.append(grpc_sample)
+        for _ in range(alien_samples):
+            grpc_sample = Sample()
+            for __ in range(20):
+                grpc_sample.features.append(random.uniform(5000., 6000.))
+            request.samples.append(grpc_sample)
+        # Timeout in seconds.
+        # Please refer gRPC Python documents for more detail. https://grpc.io/grpc/python/grpc.html
+        response: DetectionResponse = await stub.Detect(request,
+                                       timeout=10)
+    print("Greeter client received:", service_id)
\ No newline at end of file
+    return service_id * 2
+
+
+async def main_loop():
+    host = get_setting('DBSCANSERVINGSERVICE_SERVICE_HOST')
+    port = get_setting('DBSCANSERVINGSERVICE_SERVICE_PORT_GRPC')
+    endpoint = '{:s}:{:s}'.format(str(host), str(port))
+
+    ret = await asyncio.gather(
+        run(endpoint, 1),
+        run(endpoint, 2)
+    )
+
+    print(ret)
+
+
+if __name__ == '__main__':
+    
+    logging.basicConfig()
+    asyncio.run(main_loop())
\ No newline at end of file

src/opticalattackmanager/service/__main__.py

+import asyncio, grpc, random
+from common.proto.optical_attack_detector_pb2_grpc import OpticalAttackDetectorServiceStub
 import logging, signal, sys, time, threading
 from multiprocessing import Manager, Process
 from typing import List
 from prometheus_client import start_http_server
-from celery import Celery
 
 from common.Settings import get_log_level, get_metrics_port, get_setting
 from common.proto.context_pb2 import ContextIdList, Empty, EventTypeEnum, ServiceId, ServiceIdList
@@ -15,14 +16,41 @@ from monitoring.client.MonitoringClient import MonitoringClient
 terminate = threading.Event()
 LOGGER = None
 
+# For more channel options, please see https://grpc.io/grpc/core/group__grpc__arg__keys.html
+CHANNEL_OPTIONS = [('grpc.lb_policy_name', 'pick_first'),
+                   ('grpc.enable_retries', True),
+                   ('grpc.keepalive_timeout_ms', 10000)]
+# TODO: configure retries
+
 def signal_handler(signal, frame): # pylint: disable=redefined-outer-name
     LOGGER.warning('Terminate signal received')
     terminate.set()
 
 
-def monitor_services(app: Celery, service_list: List[ServiceId]):
+async def detect_attack(endpoint, context_id, service_id):
+    async with grpc.aio.insecure_channel(target=endpoint,
+                                         options=CHANNEL_OPTIONS) as channel:
+        stub = OpticalAttackDetectorServiceStub(channel)
+
+        service_id = ServiceId()
+        service_id.context_id.context_uuid.uuid = context_id
+        service_id.service_uuid.uuid = str(service_id)
+        # Timeout in seconds.
+        # Please refer gRPC Python documents for more detail. https://grpc.io/grpc/python/grpc.html
+        await stub.DetectAttack(service_id,
+                                       timeout=10)
+    print("Greeter client received:", service_id)
+
+
+async def monitor_services(service_list: List[ServiceId]):
+
+    monitoring_interval = int(get_setting('MONITORING_INTERVAL', default=MONITORING_INTERVAL))
+
+    host = get_setting('OPTICALATTACKDETECTORSERVICE_SERVICE_HOST')
+    port = get_setting('OPTICALATTACKDETECTORSERVICE_SERVICE_PORT_GRPC')
+    endpoint = '{:s}:{:s}'.format(str(host), str(port))
 
-    monitoring_interval = get_setting('MONITORING_INTERVAL', default=MONITORING_INTERVAL)
+    LOGGER.info('Starting execution of the async loop')
 
     while not terminate.is_set():
         
@@ -32,26 +60,16 @@ def monitor_services(app: Celery, service_list: List[ServiceId]):
 
         start_time = time.time()
 
-        try:
-            tasks = []
-
-            for service in service_list:
-                LOGGER.debug('Scheduling service: {}'.format(service))
-                tasks.append(
-                    app.send_task('detect_attack', (service['context'], service['service'], service['kpi']))
-                )
-            
-            for task in tasks:
-                LOGGER.debug('Waiting for task {}...'.format(task))
-                result = task.get()
-                LOGGER.debug('Result for task {} is {}...'.format(task, result))
-        except Exception as e:
-            LOGGER.exception(e)
+        tasks = []
+        for service in service_list:
+            aw = detect_attack(endpoint, service['context'], service['service'])
+            tasks.append(aw)
+        [await aw for aw in tasks]
         
         end_time = time.time()
 
         diff = end_time - start_time
-        LOGGER.info('Monitoring loop with {} services took {} seconds...'.format(len(service_list), diff))
+        LOGGER.info('Monitoring loop with {} services took {} seconds ({:.2f}%)... Waiting for {:.2f} seconds...'.format(len(service_list), diff, (diff / monitoring_interval) * 100, monitoring_interval - diff))
 
         if diff / monitoring_interval > 0.9:
             LOGGER.warning('Monitoring loop is taking {} % of the desired time ({} seconds)'.format((diff / monitoring_interval) * 100, monitoring_interval))
@@ -117,28 +135,17 @@ def main():
 
     # Start metrics server
     metrics_port = get_metrics_port()
-    start_http_server(metrics_port)
+    # start_http_server(metrics_port)  # TODO: uncomment this line
 
     LOGGER.info('Connecting with context component...')
     context_client: ContextClient = ContextClient()
     context_client.connect()
     LOGGER.info('Connected successfully...')
 
-    LOGGER.info('Connecting with REDIS...')
-    REDIS_PASSWORD = get_setting('REDIS_PASSWORD')
-    REDIS_HOST = get_setting('CACHINGSERVICE_SERVICE_HOST')
-    REDIS_PORT = get_setting('CACHINGSERVICE_SERVICE_PORT')
-    BROKER_URL = f'redis://:{REDIS_PASSWORD}@{REDIS_HOST}:{REDIS_PORT}/0'
-    BACKEND_URL = f'redis://:{REDIS_PASSWORD}@{REDIS_HOST}:{REDIS_PORT}/1'
-    app = Celery(
-        'cybersecurity',
-        broker=BROKER_URL,
-        backend=BACKEND_URL
-    )
-    LOGGER.info('Connected to REDIS...')
-
     # creating a thread-safe list to be shared among threads
     service_list = Manager().list()
+    service_list.append({'context': 'admin', "service": "1213"})
+    service_list.append({'context': 'admin', "service": "1456"})
 
     context_ids: ContextIdList = context_client.ListContextIds(Empty())
 
@@ -155,10 +162,10 @@ def main():
     process_context = Process(target=get_context_updates, args=(service_list,))
     process_context.start()
 
-    monitor_services(app, service_list)
-
-    # process_security_loop = Process(target=monitor_services, args=(app, service_list))
-    # process_security_loop.start()
+    # runs the async loop in the background
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(monitor_services(service_list))
+    # asyncio.create_task(monitor_services(service_list))
 
     # Wait for Ctrl+C or termination signal
     while not terminate.wait(timeout=0.1): pass