diff --git a/.vscode/settings.json b/.vscode/settings.json index a595a07..dc84355 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -8,5 +8,6 @@ }, "python.languageServer": "Pylance", "python.analysis.typeCheckingMode": "standard", - "editor.suggestSelection": "first" + "editor.suggestSelection": "first", + "windsurfPyright.disableLanguageServices": true } diff --git a/docker-compose.yaml b/docker-compose.yaml index 2216ee5..1dbc796 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -88,7 +88,7 @@ services: GIT_BRANCH: ${SIMULATOR_GIT_BRANCH} container_name: simulator ports: - - "4840:4840" + - "4841:4840" depends_on: - kafka - redis diff --git a/ingestor/app.py b/ingestor/app.py index e7c68f8..651d4a5 100644 --- a/ingestor/app.py +++ b/ingestor/app.py @@ -16,14 +16,21 @@ POD_ID = os.getenv("HOSTNAME", "localhost") def main(): start_prometheus_server() ingestor = Ingestor() - ingestor.prepare_ingestor() + try: + ingestor.prepare_ingestor() + except Exception as e: + metrics.APP_ERRORS_TOTAL.labels( + pod_id=POD_ID).inc() # Increment errors + print(f"Failed to prepare ingestor: {e}") + exit_signal.set() ingestor.logger.info("Ingestor prepared. Starting main loop.") while not exit_signal.is_set(): start_time = time() # Start loop timer try: ingestor.loop() - metrics.APP_LOOP_COUNT.labels(pod_id=POD_ID).inc() # Increment loop counter + metrics.APP_LOOP_COUNT.labels( + pod_id=POD_ID).inc() # Increment loop counter exit_signal.wait(ingestor.poll_interval) @@ -33,7 +40,8 @@ def main(): except Exception: print("Exception in main loop. Setting exit_signal flag.") traceback.print_exc() - metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc() # Increment errors + metrics.APP_ERRORS_TOTAL.labels( + pod_id=POD_ID).inc() # Increment errors exit_signal.set() finally: # Record loop duration diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 883af54..6540426 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -49,13 +49,14 @@ class Ingestor: self.notification_handler = NotificationHandler( servers=self.kafka_servers, logger=self.logger, - project_name="OPC_INGESTOR", - pipeline_name="-", - trigger_name="-", - model_name="-", - model="-", + project_name="OPC_INGESTOR" ) + self.notification_handler.base_notification.pipeline = 'OPC_INGESTOR' + self.notification_handler.base_notification.trigger = 'INGESTOR' + self.notification_handler.base_notification.model_name = '-' + self.notification_handler.base_notification.model_id = '-' + self.ingestor_manager = None def shutdown(self): @@ -79,7 +80,8 @@ class Ingestor: logger = getLogger(__name__) logger.setLevel(getenv("LOG_LEVEL", "INFO")) handler = StreamHandler() - formatter = Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") + formatter = Formatter( + "%(asctime)s - %(name)s - %(levelname)s - %(message)s") handler.setFormatter(formatter) logger.addHandler(handler) diff --git a/requirements.txt b/requirements.txt index 40a9a06..8d5d26c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ asyncua==1.1.5 redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.1.14 prometheus_client diff --git a/simulator/redis-feeder.py b/simulator/redis-feeder.py index d3dde93..d8e42c0 100644 --- a/simulator/redis-feeder.py +++ b/simulator/redis-feeder.py @@ -1,5 +1,6 @@ -import redis import json +import redis + # Redis connection settings REDIS_HOST = "localhost" @@ -7,19 +8,19 @@ REDIS_PORT = 6379 REDIS_USERNAME = None # "default" REDIS_PASSWORD = None # "bdnZOpcyiL" -OPC_URL = "opc.tcp://sientia-opc-simulator-service.sientia-opc.svc.cluster.local:4840" -OPC_URL = "opc.tcp://localhost:4840" +OPC_URL = "opc.tcp://sientia-opc-simulator-service.sientia-opc.svc.cluster.local:4841" +OPC_URL = "opc.tcp://localhost:4841" # Connect to Redis r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True, username=REDIS_USERNAME, password=REDIS_PASSWORD) # Define the key pattern to target -pattern = "slot:opc_tags:*" +PATTERN = "slot:opc_tags:*" # Step 1: Find and delete matching keys -print("🔍 Searching for keys matching:", pattern) -for key in r.scan_iter(match=pattern): +print("🔍 Searching for keys matching:", PATTERN) +for key in r.scan_iter(match=PATTERN): r.delete(key) print(f"❌ Deleted: {key}") diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 54ebdc7..cd39f23 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -47,11 +47,7 @@ def test___init__(notification_handler, init_logger, getenv): notification_handler.assert_called_once_with( servers=["localhost:9092", "localhost:35"], logger=ingestor.logger, - project_name="OPC_INGESTOR", - pipeline_name="-", - trigger_name="-", - model_name="-", - model="-" + project_name="OPC_INGESTOR" )