Added GetScalabilityConfig rpc to CAD:

- Allows DAD to start async csv generation
- Allows DAD to send scalability parameters to CAD

proto/l3_centralizedattackdetector.proto

@@ -25,6 +25,9 @@ service L3Centralizedattackdetector {
 
   // Get the list of features used by the ML model in the CAD component
   rpc GetFeaturesIds (Empty) returns (AutoFeatures) {}
+
+  // Get Scalability Config from the DAD component
+  rpc GetScalabilityConfig (L3ScalabilityConfig) returns (Empty) {}
 }
 
 message Feature {
@@ -63,6 +66,12 @@ message L3CentralizedattackdetectorBatchInput {
 	repeated L3CentralizedattackdetectorMetrics metrics = 1;
 }
 
+// Exp2 - Scalability Config
+message L3ScalabilityConfig {
+	float time_to_stabilize = 1;
+	float max_connection_time = 2;
+}
+
 message Empty {
 	string message = 1;
 }

scalability_accuracy.csv

 TIME_CONS OVERALL_ACCURACY CRYPTO_ACCURACY TOTAL_PREDICTIONS TOTAL_POSITIVES F_POSITIVES T_NEGATIVES F_NEGATIVES CONFIDENCE TIMESTAMP TIME_TO_STABILIZE
 60 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 |18/04/2023 09:11:32| 2
+60 0.998019814491272 1.0 3536.0 3.0 0.0 3533.0 7.0 1.0 |18/04/2023 10:20:22| 5
+60.0 0.997032642364502 1.0 5056.0 5.0 0.0 5051.0 15.0 1.0 |18/04/2023 13:29:22| 5.0

src/l3_centralizedattackdetector/client/l3_centralizedattackdetectorClient.py

@@ -22,6 +22,7 @@ from common.proto.l3_centralizedattackdetector_pb2 import (
     Empty,
     L3CentralizedattackdetectorBatchInput,
     L3CentralizedattackdetectorMetrics,
+    L3ScalabilityConfig,
     ModelInput,
     ModelOutput
 )
@@ -70,5 +71,12 @@ class l3_centralizedattackdetectorClient:
         response = self.stub.GetOutput(request)
         LOGGER.debug('GetFeaturesIds result: {}'.format(response))
         return response
+    
+    @RETRY_DECORATOR
+    def GetScalabilityConfig(self, request: L3ScalabilityConfig) -> Empty:
+        LOGGER.debug('GetScalabilityConfig request: {}'.format(request))
+        response = self.stub.GetOutput(request)
+        LOGGER.debug('GetScalabilityConfig result: {}'.format(response))
+        return response
 
 

src/l3_centralizedattackdetector/service/l3_centralizedattackdetectorServiceServicerImpl.py

@@ -52,11 +52,7 @@ current_dir = os.path.dirname(os.path.abspath(__file__))
 # Demo constants
 DEMO_MODE = True
 ATTACK_IPS = ["37.187.95.110", "91.121.140.167", "94.23.23.52", "94.23.247.226", "149.202.83.171"]
-
-TIME_TO_STABILIZE = 2  # minutes
 TIME_START = time.time()
-MAX_CONNECTION_TIME = 60
-
 METRICS_POOL = MetricsPool('l3_centralizedattackdetector', 'RPC')
 
 
@@ -197,9 +193,8 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
         self.confidence = Value('f', 0)
         self.calculated_csv = False
         
-        # Start process to generate accuracy scalability csv asynchronically
-        p = Process(target=self.generate_accuracy_scalability_csv)
-        p.start()
+        self.max_connection_time = 60
+        self.time_to_stabilize = 5
 
     """
     Create a monitored KPI for a specific service and add it to the Monitoring Client
@@ -662,11 +657,11 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
             f.close()        
             
     def generate_accuracy_scalability_csv(self):
-        LOGGER.debug("Starting async prediction accuracy analysis 2")
+        LOGGER.debug("Starting async prediction accuracy analysis")
         LOGGER.debug("Correct csv load: {}".format(os.path.exists("/var/teraflow/scalability_accuracy.csv")))
         
         # Wait for the system to stabilize
-        time.sleep(TIME_TO_STABILIZE * 60)
+        time.sleep(self.time_to_stabilize * 60)
         LOGGER.debug("Scalability csv started")
         with open("/var/teraflow/scalability_accuracy.csv", 'a', newline='') as f:
             spamwriter = csv.writer(f, delimiter=' ', quotechar='|', quoting=csv.QUOTE_MINIMAL)
@@ -674,12 +669,25 @@ class l3_centralizedattackdetectorServiceServicerImpl(L3Centralizedattackdetecto
                                 'TOTAL_PREDICTIONS', 'TOTAL_POSITIVES', 'F_POSITIVES', 
                                 'T_NEGATIVES', 'F_NEGATIVES', 'CONFIDENCE', 'TIMESTAMP', 'TIME_TO_STABILIZE'])'''
             
-            spamwriter.writerow([MAX_CONNECTION_TIME, self.overall_detection_acc.value, self.cryptomining_attack_detection_acc.value,
+            spamwriter.writerow([self.max_connection_time, self.overall_detection_acc.value, self.cryptomining_attack_detection_acc.value,
                                 self.total_predictions.value, self.attack_connections_len.value, self.false_positives.value,
                                 self.total_predictions.value - self.attack_connections_len.value, self.false_negatives.value,
-                                self.confidence.value, datetime.now().strftime("%d/%m/%Y %H:%M:%S"), TIME_TO_STABILIZE])
+                                self.confidence.value, datetime.now().strftime("%d/%m/%Y %H:%M:%S"), self.time_to_stabilize])
             
             f.close()
+        
+    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
+    def GetScalabilityConfig(self, request, context):
+        LOGGER.info("Received scalability config request")
+        
+        self.max_connection_time = request.max_connection_time
+        self.time_to_stabilize = request.time_to_stabilize
+        
+        # Start process to generate accuracy scalability csv asynchronically
+        p = Process(target=self.generate_accuracy_scalability_csv)
+        p.start()
+        
+        return Empty(message="CSV generated")
 
     def AnalyzeBatchConnectionStatistics(self, request, context):
         batch_time_start = time.time()