PR-15: Prepare service for production environments

.github/workflows/tests.yml

@@ -53,7 +53,7 @@ jobs:
 
     - name: Run outliers
       run: |
-        python3.8 resources/src/__main__.py --test
+        ENVIRONMENT=test python3.8 resources/src/__main__.py
 
 
 

resources/src/__main__.py

@@ -17,18 +17,41 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-import sys
-from Logger.logger import logger
-from server.rest import APIServer
+import os
+from logger.logger import logger
+from server.rest import APIServer, config
+from server.production import GunicornApp
 
 class Outliers:
     def __init__(self) -> None:
-        logger.info("Starting Outliers API REST")
+        self.environment = os.environ.get("ENVIRONMENT", "development")
+        self.server = None
+        self.app = None
+        self.run()
+
+    def run(self):
+        if "production" in self.environment:
+            self.run_production_server()
+        if "development" in self.environment:
+            self.run_test_server(False)
+        if "test" in self.environment:
+            self.run_test_server(True)
+
+    def run_test_server(self, test_run_github_action):
         self.api = APIServer()
-        logger.info("Starting Outliers Server")
-        if len(sys.argv) > 1:
-            self.api.start_server(True)
-        else:
-            self.api.start_server(False)
+        self.api.start_test_server(test_run_github_action)
+
+    def run_production_server(self):
+        logger.info("Starting Outliers API REST")
+        __binding_host__ = config.get("OutliersServerProduction", "outliers_binding_address")
+        __binding_port__ = config.get("OutliersServerProduction", "outliers_server_port")
+        gunicorn_workers = config.get("OutliersServerProduction", "outliers_server_workers")
+        options = {
+            'bind': f"{__binding_host__}:{__binding_port__}",
+            'workers': gunicorn_workers
+        }
+        self.server = APIServer()
+        self.app = GunicornApp(self.server, options)
+        self.app.run()
 
-Outliers_ = Outliers()
+_Outliers = Outliers()

resources/src/IA/outliers.py → resources/src/ai/outliers.py

@@ -1,6 +1,7 @@
 # Copyright (C) 2023 Eneo Tecnologia S.L.
 #
 # Authors :
+# Miguel Álvarez Adsuara <malvarez@redborder.com>
 # Pablo Rodriguez Flores <prodriguez@redborder.com>
 #
 # This program is free software: you can redistribute it and/or modify

resources/src/config.ini

@@ -1,9 +1,21 @@
-[OutliersServer]
+[OutliersServerProduction]
+outliers_binding_address=0.0.0.0
 outliers_server_port=39091
-metric = bytes
+outliers_server_workers=4
+
+[OutliersServerTesting]
+outliers_binding_address=0.0.0.0
+outliers_server_port=39091
+
+[Outliers]
+metric=bytes
 
 [Druid]
-druid_endpoint = http://x.x.x.x:8080/druid/v2/
+druid_endpoint=http://10.0.209.10:8080/druid/v2/
 
 [Logger]
-log_file = ./outliers.log
+log_file=./outliers.log
+
+[redBorder]
+api_endpoint=http://x.x.x.x/api/v1/
+api_oauth_token=abcdabcdabcd

resources/src/Druid/query_builder.py → resources/src/druid/query_builder.py

@@ -23,6 +23,7 @@
 """
 
 import json
+
 class QueryBuilder:
     """
     Class used to modify the manager usual query to exctract data about

resources/src/Logger/logger.py → resources/src/logger/logger.py

@@ -36,12 +36,12 @@ def __init__(self, log_level=logging.INFO, log_file=None):  # Accept log_file as
         if log_file is None:
             log_file = './outliers.log'
         try:
-            from Config import configmanager
+            from config import configmanager
             config = configmanager.ConfigManager(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "config.ini"))
             log_file = config.get('Logger', 'log_file')
         except Exception as e:
             print("Could not resolve ConfigManager, default set to ./outliers.log")
-        self.logger = PyLogrus(name="OutliersLogger")
+        self.logger = PyLogrus(name="Outlierslogger")
         log_dir = os.path.dirname(log_file)
         if not os.path.isdir(log_dir):
             os.makedirs(log_dir)

resources/src/redborder/client.py

@@ -0,0 +1,34 @@
+# Copyright (C) 2023 Eneo Tecnologia S.L.
+#
+# Authors :
+# Miguel Álvarez Adsuara <malvarez@redborder.com>
+# Pablo Rodriguez Flores <prodriguez@redborder.com>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import requests
+
+class RedBorderAPI:
+    def __init__(self, endpoint, oauth_token) -> None:
+        self.api_endpoint = endpoint
+        self.oauth_token = oauth_token
+
+    def request_flow_sensors(self):
+        response = requests.get(f'{self.api_endpoint}/sensors/flow/?auth_token={self.oauth_token}')
+
+        if response.status_code == 200:
+            response_json = response.json()
+            return response_json['flow']
+
+        raise Exception(f"redBorder api request failed with status code {response.status_code}.")

resources/src/requirements.txt

@@ -50,4 +50,5 @@ typing_extensions==4.7.1
 urllib3==1.26.16
 Werkzeug==2.2.3
 wrapt<1.15,>=1.11.0
-zipp==3.15.0
+zipp==3.15.0
+gunicorn==21.2.0

resources/src/server/production.py

@@ -0,0 +1,35 @@
+# Copyright (C) 2023 Eneo Tecnologia S.L.
+#
+# Authors :
+# Miguel Álvarez Adsuara <malvarez@redborder.com>
+# Pablo Rodriguez Flores <prodriguez@redborder.com>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+from gunicorn.app.base import BaseApplication
+
+class GunicornApp(BaseApplication):
+    def __init__(self, app, options=None):
+        self.options = options or {}
+        self.application = app
+        super().__init__()
+
+    def load_config(self):
+        for key, value in self.options.items():
+            if key in self.cfg.settings and value is not None:
+                self.cfg.set(key, value)
+
+    def load(self):
+        return self.application.app

resources/src/server/rest.py

@@ -23,10 +23,10 @@
 import time
 import base64
 import threading
-from IA import outliers
-from Druid import client, query_builder
-from Logger import logger
-from Config import configmanager
+from ai import outliers
+from druid import client, query_builder
+from logger import logger
+from config import configmanager
 from flask import Flask, jsonify, request
 
 '''
@@ -36,8 +36,8 @@
 config = configmanager.ConfigManager(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "config.ini"))
 druid_client = client.DruidClient(config.get("Druid", "druid_endpoint"))
 query_modifier = query_builder.QueryBuilder(
-    os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "Druid", "aggregations.json"),
-    os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "Druid", "postAggregations.json")
+    os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "druid", "aggregations.json"),
+    os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "druid", "postAggregations.json")
 )
 
 class APIServer:
@@ -58,9 +58,9 @@ def calculate():
                 try:
                     return jsonify(outliers.Autoencoder.execute_prediction_model(
                         data,
-                        config.get("OutliersServer", "metric"),
-                        os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "IA", "traffic.keras"),
-                        os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "IA", "traffic.ini")
+                        config.get("Outliers", "metric"),
+                        os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "ai", "traffic.keras"),
+                        os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "ai", "traffic.ini")
                     ))
                 except Exception as e:
                     logger.logger.error("Error while calculating prediction model -> " + str(e))
@@ -69,19 +69,19 @@ def calculate():
                 logger.logger.error("Error while proccessing, Druid query is empty")
                 return jsonify(outliers.Autoencoder.return_error())
 
-    def run_app(self):
+    def run_test_app(self):
         try:
-            self.app.run(debug=False, host="0.0.0.0", port=config.get("OutliersServer", "outliers_server_port"))
+            self.app.run(debug=False, host=config.get("OutliersServerTesting", "outliers_binding_address"), port=config.get("OutliersServerTesting", "outliers_server_port"))
         except Exception as e:
             logger.logger.error(f"Exception in server thread: {e}")
             self.exit_code = 1
 
-    def start_server(self, test):
-        if test:
-            self.server_thread = threading.Thread(target=self.run_app)
+    def start_test_server(self, test_run_github_action):
+        if test_run_github_action:
+            self.server_thread = threading.Thread(target=self.run_test_app)
             self.server_thread.daemon = True
             self.server_thread.start()
             time.sleep(30)
             sys.exit(self.exit_code)
         else:
-            self.run_app()
+            self.run_test_app()

resources/systemd/rb-aioutliers.service

@@ -15,7 +15,7 @@ Group=root
 KillSignal=SIGTERM  # Use SIGTERM for graceful termination
 Type=simple
 
-ExecStart=/usr/bin/scl enable rh-python38 -- python3 /opt/rb-aioutliers/resources/src/__main__.py
+ExecStart=/usr/bin/scl enable rh-python38 -- ENVIRONMENT=production python3 /opt/rb-aioutliers/resources/src/__main__.py
 
 TimeoutStopSec=60
 

resources/tests/test_configmanager.py

@@ -24,7 +24,7 @@
 import configparser
 
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-from src.Config.configmanager import ConfigManager
+from src.config.configmanager import ConfigManager
 
 class TestConfigManager(unittest.TestCase):
     def setUp(self):

resources/tests/test_druid.py

@@ -17,13 +17,13 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import os
+import sys
 import unittest
 from unittest.mock import MagicMock, patch
-import sys
-import os
 
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-from src.Druid.client import DruidClient
+from src.druid.client import DruidClient
 
 class TestDruidClient(unittest.TestCase):
     def setUp(self):

resources/tests/test_logger.py

@@ -17,15 +17,14 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-import unittest
 import os
 import sys
 import logging
-from unittest.mock import patch
+import unittest
 from io import StringIO
 
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-from src.Logger.logger import Logger
+from src.logger.logger import Logger
 
 class TestLogger(unittest.TestCase):
     def setUp(self):

resources/tests/test_outliers.py

@@ -23,10 +23,8 @@
 import logging
 import json
 
-from unittest.mock import patch
-
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-from src.IA.outliers import Autoencoder
+from src.ai.outliers import Autoencoder
 
 class TestAutoencoder(unittest.TestCase):
     def setUp(self):
@@ -41,15 +39,15 @@ def test_model_execution_with_no_data(self):
             Autoencoder.execute_prediction_model(
                 {},
                 "bytes",
-                os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src", "IA", "traffic.keras"),
-                os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src", "IA", "traffic.ini")
+                os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src", "ai", "traffic.keras"),
+                os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src", "ai", "traffic.ini")
             )
     def test_model_execution_with_sample_data(self):
         Autoencoder.execute_prediction_model(
             self.sample_data,
             "bytes",
-            os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src", "IA", "traffic.keras"),
-            os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src", "IA", "traffic.ini")
+            os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src", "ai", "traffic.keras"),
+            os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src", "ai", "traffic.ini")
         )
 
 if __name__ == '__main__':

resources/tests/test_query_builder.py

@@ -18,17 +18,17 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import unittest
-from unittest.mock import MagicMock, patch
 import sys
 import os
 
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
-from src.Druid import query_builder
+from src.druid import query_builder
+
 class TestQueryBuilder(unittest.TestCase):
     def setUp(self) -> None:
-        aggregations_file = os.path.join(os.getcwd(),"resources", "src", "Druid", "aggregations.json")
-        post_aggregations_file = os.path.join(os.getcwd(),"resources",  "src", "Druid", "postAggregations.json")
+        aggregations_file = os.path.join(os.getcwd(),"resources", "src", "druid", "aggregations.json")
+        post_aggregations_file = os.path.join(os.getcwd(),"resources",  "src", "druid", "postAggregations.json")
         self.builder = query_builder.QueryBuilder(aggregations_file, post_aggregations_file)
 
     def test_known_granularities_granularities_to_seconds(self):