Fixed Reflection bug

Author: BuxtonCalvin

classes/transports/timescaledb.py

@@ -273,6 +273,13 @@ class timescaledb(transport_base):
     schema_needs_refresh: bool = True  # flag to indicate if ORM schema refresh is needed after reconnect or column changes
     current_metric_count: int = 0
 
+    # pushover notification settings for stale data alerts.
+    # These are optional and can be configured by the user if they want
+    # to receive pushover notifications when stale data is detected.
+    enable_pushover: bool = False
+    pushover_token: str = ""
+    pushover_user: str = ""
+
     def __init__(self, settings: SectionProxy) -> None:
         """
         Initialize the TimescaleDB transport bridge.
@@ -1085,11 +1092,25 @@ def _validate_wide_row(self, row: dict) -> bool:
 
             # metadata keys like m_time and device_info_id are excluded in _cache_wide_table_columns
             extra_keys: set = set(row) - self._wide_columns
+            fewer_keys: set = self._wide_columns - set(row)
 
             if extra_keys:
-                self._log.error( f"Wide-table schema mismatch; unknown columns: {sorted(extra_keys)}" )
-                msg: str= f"Unknown columns: {sorted(extra_keys)}"
-                raise ValueError(msg)
+                self._log.info(f"New metrics detected: {extra_keys}. Triggering resync...")
+
+                # 1. Trigger the sync logic to update the table schema and internal column cache.
+                self._sync_single_table_schema()
+
+                # 2. Re-check: Did the resync actually add the keys?
+                # (In case the DB hasn't been updated yet)
+                still_extra = set(row) - self._wide_columns - {"m_time", "device_info_id"}
+                if still_extra:
+                    msg = f"Database schema is still missing columns after resync: {sorted(still_extra)}"
+                    self._log.error(msg)
+                    raise ValueError(msg)
+            elif fewer_keys:
+                self._log.warning( f"Wide-table schema mismatch; missing keys in scrape data: {sorted(fewer_keys)}" )
+                msg: str= f"Missing columns: {sorted(fewer_keys)}"
+
             else:
                 return True
 
@@ -1111,7 +1132,7 @@ def _sync_single_table_schema(self) -> None:
             else:
                 self._log.warning(f"No existing table found for {table_name} during resync. Continuing with new reflection.")
 
-            # 2. Reflect the NEW structure from the database
+            # 2. Reflect the new structure from the database
             new_table = Table(
                 table_name,
                 Base.metadata,
@@ -1263,14 +1284,15 @@ def _flush_worker(self) -> None:
                             self._flush_batch_narrow(narrow_data, device_info_id, timestamp, session)
                             if self.wide_table_flag:
                                 if valid_row:
-                                    stmt: Insert = pg_insert(DeviceMetricsWide).values(**wide_data)
+                                    target_table: Table = Base.metadata.tables[DeviceMetricsWide.__tablename__]
+                                    stmt: Insert = pg_insert(target_table).values(**wide_data)
                                     session.execute(stmt)
 
                         self._commit_transport_state(transport_name, metrics_only, timestamp, is_stale=False)
                         self._log.debug(f"data write complete from [{transport_name}] to timescaledb bridge")
                 except (SQLAlchemyError, ValueError) as e1:
                     session.rollback()
-                    self._log.warning(f"metrics data write failed.{e1}")
+                    self._log.error(f"metrics data write failed.{e1}")
 
                     # Only backlog if setting enabled and DB is down
                     with self._reconnect_lock:
@@ -2618,6 +2640,7 @@ def _get_dynamic_settings(self) -> dict:
 
         return self.performance_tiers["tier_low"] # Fallback default
 
+
     def _view_exists(self, session: Session, view_name: str) -> bool:
         """Check to see if a continuous aggregate exists in the catalog.
         Parameters:
@@ -2626,9 +2649,20 @@ def _view_exists(self, session: Session, view_name: str) -> bool:
         Returns:
             bool: True if the view exists, False otherwise.
         """
-        check_sql: TextClause = text("SELECT 1 FROM timescaledb_information.continuous_aggregates WHERE view_name = :name")
-        return session.execute(check_sql, {"name": view_name}).fetchone() is not None
+        check_sql = text("SELECT 1 FROM timescaledb_information.continuous_aggregates WHERE view_name = :name")
 
+        try:
+            # Keep ONLY the risky operation in the try block
+            result = session.execute(check_sql, {"name": view_name}).fetchone()
+        except Exception as e:
+            # Handle the error and clean up
+            self._log.error(f"Error checking view existence for {view_name}: {e}")
+            session.rollback()
+            return False
+        else:
+            # The 'else' block runs ONLY if the try block succeeded
+            session.rollback()  # Resets state for your next AUTOCOMMIT call
+            return result is not None
 
     def _drop_all_continuous_aggregates(self, session: Session) -> None:
         """
@@ -2776,7 +2810,7 @@ def _refresh_single_rollup(self, session: Session, view_name: str, start_offset:
         self._log.info(f"Starting {mode} refresh for {view_name}...")
 
         # AUTOCOMMIT is mandatory for CALL refresh_continuous_aggregate
-        with session.connection().execution_options(isolation_level="AUTOCOMMIT") as conn:
+        with self.engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
             conn.execute(text(f"SET LOCAL work_mem = '{r_settings['work_mem']}';"))
             try:
                 if force_full:

classes/transports/transport_base.py

@@ -80,7 +80,7 @@ class transport_base:
     on_message : Callable[["transport_base", registry_map_entry, str], None] = None
     ''' callback, on message received '''
 
-    request_upstream_reconnect: Callable[[], None] | None = None # callback for reconnect.
+    request_upstream_reconnect: Callable[[str], None] | None = None # callback for reconnect.
 
     _log : logging.Logger = None
 

protocol_gateway.py

@@ -126,7 +126,7 @@ class Protocol_Gateway:
     _logging_initialized = False
 
     @classmethod
-    def _setup_logging(cls, cfg) -> None:
+    def _setup_logging(cls, cfg: ConfigParser) -> None:
         """
         created to eliminate multi gig log file sizes in docker as logging.StreamHandler(sys.stdout) results in a
         json file that can quickly grow quiet large.
@@ -138,7 +138,7 @@ def _setup_logging(cls, cfg) -> None:
             return
 
         # Read logging config
-        level_name: str = cfg.get("logging", "level", fallback="INFO").upper()
+        level_name: str = cfg.get("logging", "level", fallback="INFO").strip().upper()
         level: int = getattr(logging, level_name, logging.INFO)
 
         log_dir = Path(cfg.get("logging", "log_dir", fallback="logs"))
@@ -190,7 +190,7 @@ def _setup_logging(cls, cfg) -> None:
         handler.setFormatter(formatter)
 
         # ---- Root logger wiring ----
-        root = logging.getLogger()
+        root: logging.Logger = logging.getLogger()
         root.setLevel(level)
         root.handlers.clear()
         root.addHandler(handler)
@@ -203,7 +203,7 @@ def _setup_logging(cls, cfg) -> None:
 
         cls._logging_initialized = True
 
-    __log = None
+    __log : logging.Logger
     # log level, available log levels are CRITICAL, FATAL, ERROR, WARNING, INFO, DEBUG, EXCEPTION
     __log_level = "DEBUG"
 
@@ -213,12 +213,12 @@ def _setup_logging(cls, cfg) -> None:
     __transports : list[transport_base] = []
     ''' transport_base is for type hinting. this can be any transport'''
 
-    config_file : str
+    config_file : Path
 
     # Simple read completion tracking
     __read_completion_tracker : dict[str, bool] = {}
     ''' Track which transports have completed their current read cycle '''
-    __read_tracker_lock : threading.Lock = None
+    __read_tracker_lock : threading.Lock
 
     # Concurrency control
     __enable_concurrency : bool = False
@@ -245,14 +245,15 @@ def __init__(self, config_file : str):
         if alternate_cfg.is_file():
             self.config_file = alternate_cfg
         else:
+            self.__log.warning(f"Config file not found {alternate_cfg}, using default: {default_cfg}")
             self.config_file = default_cfg
 
         #pymodbus_log = logging.getLogger('pymodbus')
         #pymodbus_log.setLevel(logging.DEBUG)
         #pymodbus_log.addHandler(handler)
 
         self.__settings = CustomConfigParser()
-        self.__settings.read(self.config_file)
+        self.__settings.read(self.config_file.as_posix())
 
         self._setup_logging(self.__settings)
         self.__log: logging.Logger = logging.getLogger(__name__)
@@ -261,15 +262,17 @@ def __init__(self, config_file : str):
         self.__log_level = self.__settings.get("general","log_level", fallback="INFO")
 
         # Read concurrency setting - default to sequential (disabled) for better stability
-        self.__enable_concurrency = self.__settings.getboolean("general", "enable_concurrency", fallback=False)
+        self.__enable_concurrency = bool(self.__settings.getboolean("general", "enable_concurrency", fallback=False))
         self.__log.info(f"Concurrency mode: {'Concurrent' if self.__enable_concurrency else 'Sequential'}")
 
         # Read sequential delay setting
-        self.__sequential_delay = self.__settings.getfloat("general", "sequential_delay", fallback=1.0)
+        self.__sequential_delay = float(
+            self.__settings.getfloat("general", "sequential_delay", fallback=1.0) or 1.0
+        )
         if not self.__enable_concurrency:
             self.__log.info(f"Sequential delay between transports: {self.__sequential_delay} seconds")
 
-        log_level = getattr(logging, self.__log_level, logging.INFO)
+        log_level = getattr(logging, str(self.__log_level), logging.INFO)
         self.__log.setLevel(log_level)
         self.__log.info("Loading...")
 
@@ -362,6 +365,13 @@ def _process_transport_read(self, transport) -> None:
         except Exception as err:
             self.__log.exception(f"Error processing transport {transport.transport_name} and {err}")
             # traceback.print_exc()
+            # Errno 104 - Connection reset by peer (common for MQTT disconnects)
+            # Errno 32 - Broken pipe (common for MQTT disconnects)
+            # Errno 110 - Connection timed out (common for network issues)
+            if err == 'Errno 104' or err == 'Errno 32' or err == 'Errno 110':
+                # traceback.print_exc()
+                transport.connect()
+            self.__log.warning(f"Attempting reconnect for {transport.transport_name}")
             self._mark_read_complete(transport)
 
     def _mark_read_complete(self, transport) -> None: