Dev

config/integrations/elastic.properties

@@ -1,4 +1,5 @@
 [MAIN]
 ELK_SERVER = access_url
 ELK_TOKEN = access_token
+ELK_SSL_VERIFY = True
 BATCH_SIZE = 20000

config/model_quality/model_quality.properties

@@ -5,4 +5,6 @@ ELK_STATISTICS_INDEX = emfos-model-statistics
 BORDER_LIMIT = 250
 LINE_LIMIT_TEMPERATURE = 25 C
 IGM_RULE_SET = impedance,line_rating
-CGM_RULE_SET = kruonis,rtec,outage,lt_pl_xborder
+CGM_RULE_SET = kruonis,rtec,outage,lt_pl_xborder
+#either a BA or EU
+CGM_TYPE = BA

config/task_generator/manual_task_generator.properties

@@ -9,6 +9,7 @@ RUN_TYPE =
 INCLUDED_TSO =
 EXCLUDED_TSO =
 LOCAL_IMPORT =
+REPLACE_TSO =
 PROCESS_TIME_SHIFT =
 TASK_REFERENCE_TIME =
 TASK_VERSION =

config/task_generator/process_conf.json

@@ -28,6 +28,7 @@
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning",
         "included": [],
         "excluded": [],
+        "replace_tso": [],
         "time_horizon": "ID",
         "version": "001",
         "replacement": "False",
@@ -63,6 +64,7 @@
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning",
         "included": [],
         "excluded": [],
+        "replace_tso": [],
         "time_horizon": "ID",
         "version": "001",
         "replacement": "False",
@@ -98,6 +100,7 @@
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning",
         "included": [],
         "excluded": [],
+        "replace_tso": [],
         "time_horizon": "ID",
         "version": "001",
         "replacement": "False",
@@ -133,6 +136,7 @@
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning",
         "included": [],
         "excluded": [],
+        "replace_tso": [],
         "time_horizon": "1D",
         "version": "001",
         "replacement": "False",
@@ -168,6 +172,7 @@
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning",
         "included": [],
         "excluded": [],
+        "replace_tso": [],
         "time_horizon": "2D",
         "version": "001",
         "replacement": "False",
@@ -220,6 +225,7 @@
         "local_import": [
          "LITGRID"
         ],
+        "replace_tso": [],
         "time_horizon": "ID",
         "version": "001",
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM",
@@ -262,6 +268,7 @@
         "local_import": [
          "LITGRID"
         ],
+        "replace_tso": [],
         "time_horizon": "ID",
         "version": "001",
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM",
@@ -304,6 +311,7 @@
         "local_import": [
          "LITGRID"
         ],
+        "replace_tso": [],
         "time_horizon": "ID",
         "version": "001",
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM",
@@ -345,6 +353,7 @@
         ],
         "excluded": [],
         "local_import": [],
+        "replace_tso": [],
         "time_horizon": "ID",
         "version": "001",
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM",
@@ -387,6 +396,7 @@
         "local_import": [
          "LITGRID"
         ],
+        "replace_tso": [],
         "time_horizon": "1D",
         "version": "001",
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM",
@@ -429,6 +439,7 @@
         "local_import": [
          "LITGRID"
         ],
+        "replace_tso": [],
         "time_horizon": "2D",
         "version": "001",
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM",
@@ -471,6 +482,7 @@
         "local_import": [
          "LITGRID"
         ],
+        "replace_tso": [],
         "time_horizon": "YR",
         "version": "001",
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM",
@@ -517,6 +529,7 @@
          "AST",
          "ELERING"
         ],
+        "replace_tso": [],
         "time_horizon": "WK",
         "version": "001",
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM",
@@ -560,6 +573,7 @@
          "AST",
          "ELERING"
         ],
+        "replace_tso": [],
         "time_horizon": "MO",
         "version": "001",
         "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM",

emf/common/helpers/tasks.py

@@ -26,6 +26,10 @@ def update_task_status(task: dict, status_text: str, publish: bool = True):
         "timestamp": utc_now
     })
 
+    # Fix boolean values
+    for key, value in task['task_properties'].items():
+        task['task_properties'][key] = value.strip().lower() == "true" if isinstance(value, str) and value.strip().lower() in {"true", "false"} else value
+
     # TODO - better handling if elk is not available, possibly set elk connection timeout really small or refactor the sending to happen via rabbit
     if publish:
         try:

emf/common/integrations/elastic.py

@@ -5,34 +5,51 @@
 import pandas as pd
 import json
 import uuid
+import os
 from typing import List, Dict
 from elasticsearch import Elasticsearch
 import config
 from emf.common.config_parser import parse_app_properties
 
-import warnings
-from elasticsearch.exceptions import ElasticsearchWarning
-warnings.simplefilter('ignore', ElasticsearchWarning)
-
 logger = logging.getLogger(__name__)
 
 parse_app_properties(caller_globals=globals(), path=config.paths.integrations.elastic)
 
 
 class Elastic:
 
-    def __init__(self, server: str = ELK_SERVER, api_key: str = ELK_TOKEN, debug: bool = False):
+    def __init__(self,
+                 server: str = ELK_SERVER,
+                 api_key: str = ELK_TOKEN,
+                 ssl_verify: bool = json.loads(ELK_SSL_VERIFY.lower()),
+                 debug: bool = False):
+
         self.server = server
+        self.api_key = api_key
+        self.ssl_verify = ssl_verify
         self.debug = debug
-        self.client = Elasticsearch(self.server, api_key=api_key)
+
+        # Override SSL CERT FILE from ENV variables if defined
+        ssl_cert_file = None
+        if self.ssl_verify:
+            ssl_cert_file = os.getenv("SSL_CERT_FILE", None)
+            if not ssl_cert_file:
+                raise Exception("SSL verification is enabled but SSL_CERT_FILE environment variable is not set.")
+            else:
+                logger.debug(f"Using SSL certificate file: {ssl_cert_file}")
+
+        # Create client
+        self.client = Elasticsearch(self.server, api_key=self.api_key, verify_certs=self.ssl_verify, ca_certs=ssl_cert_file)
 
     @staticmethod
     def send_to_elastic(index: str,
                         json_message: dict,
                         id: str = None,
                         server: str = ELK_SERVER,
                         api_key: str = ELK_TOKEN,
+                        ssl_verify: bool = json.loads(ELK_SSL_VERIFY.lower()),
                         iso_timestamp: str = None,
+                        index_rollover: bool = True,
                         debug: bool = False):
         """
         Method to send single message to ELK
@@ -41,6 +58,7 @@
         :param id:
         :param server: url of ELK server
         :param iso_timestamp: message timestamp
+        :param index_rollover: if true, adds monthly indication to index name
         :param debug: flag for debug mode
         :return:
         """
@@ -53,8 +71,9 @@
         json_message["@timestamp"] = iso_timestamp
 
         # Create server url with relevant index pattern
-        _index = f"{index}-{datetime.datetime.today():%Y%m}"
-        url = f"{server}/{_index}/_doc"
+        if index_rollover:
+            index = f"{index}-{datetime.datetime.today():%Y%m}"
+        url = f"{server}/{index}/_doc"
 
         if id:
             url = url + f"/{id}"
@@ -66,7 +85,7 @@
             json_message.pop('args')
         json_data = json.dumps(json_message, default=str, ensure_ascii=True, skipkeys=True)
         headers = {"Authorization": f"ApiKey {api_key}", "Content-Type": "application/json"}
-        response = requests.post(url=url, data=json_data.encode(), headers=headers, verify=False)
+        response = requests.post(url=url, data=json_data.encode(), headers=headers, verify=ssl_verify)
         if json.loads(response.content).get('error'):
             logger.error(f"Send to Elasticsearch responded with error: {response.text}")
         if debug:
@@ -82,8 +101,10 @@
                              hashing: bool = False,
                              server: str = ELK_SERVER,
                              api_key: str = ELK_TOKEN,
+                             ssl_verify: bool = json.loads(ELK_SSL_VERIFY.lower()),
                              batch_size: int = int(BATCH_SIZE),
                              iso_timestamp: str | None = None,
+                             index_rollover: bool = True,
                              debug: bool = False):
         """
         Method to send bulk message to ELK
@@ -95,6 +116,7 @@
         :param server: url of ELK server
         :param batch_size: maximum size of batch
         :param iso_timestamp: timestamp to be included in documents
+        :param index_rollover: if true, adds monthly indication to index name
         :param debug: flag for debug mode
         :return:
         """
@@ -116,7 +138,8 @@
         json_message_list = [{**element, '@timestamp': iso_timestamp} for element in json_message_list]
 
         # Define server url with relevant index pattern (monthly indication is added)
-        index = f"{index}-{datetime.datetime.today():%Y%m}"
+        if index_rollover:
+            index = f"{index}-{datetime.datetime.today():%Y%m}"
         url = f"{server}/{index}/_bulk"
 
         if id_from_metadata:
@@ -135,7 +158,7 @@
                                      data=(ndjson.dumps(json_message_list[batch:batch + batch_size])+"\n").encode(),
                                      timeout=None,
                                      headers=headers,
-                                     verify=False)
+                                     verify=ssl_verify)
             if json.loads(response.content).get('errors'):
                 logger.error(f"Send to Elasticsearch responded with errors: {response.text}")
             if debug:
@@ -226,21 +249,22 @@
                  hashing: bool = False,
                  headers: Dict | None = None,
                  auth: object | None = None,
-                 verify: bool = False,
+                 ssl_verify: bool = json.loads(ELK_SSL_VERIFY.lower()),
                  debug: bool = False):
 
         self.index = index
         self.server = server
         self.id_from_metadata = id_from_metadata
         self.id_metadata_list = id_metadata_list
         self.hashing = hashing
+        self.ssl_verify = ssl_verify
         self.debug = debug
 
         if not headers:
             headers = {'Content-Type': 'text/json'}
 
         self.session = requests.Session()
-        self.session.verify = verify
+        self.session.verify = self.ssl_verify
         self.session.headers.update(headers)
         self.session.auth = auth
 

emf/common/loadflow_tool/settings_manager.py

@@ -45,16 +45,14 @@ class LoadflowSettingsManager:
 
     def __init__(self,
                  elastic_server: str = ELK_SERVER,
-                 elastic_username: str | None = None,
-                 elastic_password: str | None = None,
+                 elastic_api_key: str = ELK_TOKEN,
                  elastic_index: str = 'config-lf-parameters',
                  settings_keyword: str = 'EU_DEFAULT',
                  override_path: str | None = None,
                  ):
 
         self.elastic_server = elastic_server
-        self.elastic_username = elastic_username
-        self.elastic_password = elastic_password
+        self.elastic_api_key = elastic_api_key
         self.elastic_index = elastic_index
         self.settings_keyword = settings_keyword
 
@@ -82,7 +80,7 @@ def __init__(self,
 
     # ----------------- I/O -----------------
     def _get_defaults_from_elastic(self) -> dict:
-        client = Elasticsearch(self.elastic_server)
+        client = Elasticsearch(self.elastic_server, api_key=self.elastic_api_key)
         logger.info(f"Retrieving base loadflow settings fromm Elasticsearch with key: {self.settings_keyword}")
         response = client.get(index=self.elastic_index, id=self.settings_keyword)
 

emf/common/logging/custom_logger.py

@@ -1,4 +1,5 @@
 import sys
+import os
 import logging
 import requests
 from emf.common.integrations import elastic
@@ -107,7 +108,7 @@ def __init__(self,
     def elk_connection(self):
         try:
             headers = {"Authorization": f"ApiKey {self.api_key}"}
-            response = requests.get(self.server, timeout=5, headers=headers, verify=False)
+            response = requests.get(self.server, timeout=5, headers=headers, verify=os.getenv("ELK_SSL_VERIFY", "false").lower() == "true")
             if response.status_code == 200:
                 logger.info(f"Connection to {self.server} successful")
                 return True
@@ -133,7 +134,10 @@ def emit(self, record):
             elk_record.update(self.extra)
 
         # Send to Elk
-        elastic.Elastic.send_to_elastic(index=self.index, json_message=elk_record, server=self.server)
+        elastic.Elastic.send_to_elastic(index=self.index,
+                                        json_message=elk_record,
+                                        server=self.server,
+                                        index_rollover=False)
 
     # TODO - Move tracing to seperate class, that on destroy will stop tracing?
     def start_trace(self, trace_parameters: dict):