dbIngest.py

Go to the documentation of this file.

import glob
from purepyindi2 import device, properties, constants, messages
from purepyindi2.messages import DefNumber, DefSwitch, DefText
import sys
import logging
import xconf
import psycopg
from magaox.indi.device import XDevice
from magaox.constants import DEFAULT_PREFIX
from magaox.db.config import BaseDbDeviceConfig
from magaox.db import Telem, FileOrigin, UserLog
from magaox.db import ingest
from magaox.utils import parse_iso_datetime_as_utc, creation_time_from_filename
 
import json
import orjson
import xconf
import subprocess
import queue
import socket
import threading
import pathlib
import time
import os.path
import os
import sys
import datetime
from datetime import timezone
from watchdog.observers import Observer, BaseObserverSubclassCallable
from watchdog.events import FileSystemEventHandler
 
class NewXFilesHandler(FileSystemEventHandler):
    def __init__(self, host, events_queue, log_name):
        self.host = host
        self.events_queue = events_queue
        self.log = logging.getLogger(log_name)
 
    def construct_message(self, stat_result, event, is_new_file=False):
        return FileOrigin(
            origin_host=self.host,
            origin_path=event.src_path,
            creation_time=creation_time_from_filename(event.src_path, stat_result=stat_result),
            modification_time=datetime.datetime.fromtimestamp(stat_result.st_mtime),
            size_bytes=stat_result.st_size,
        )
 
    def on_created(self, event):
        if event.is_directory:
            return
        try:
            stat_result = os.stat(event.src_path)
        except FileNotFoundError:
            return
        self.events_queue.put(self.construct_message(stat_result, event, is_new_file=True))
 
    def on_modified(self, event):
        if event.is_directory:
            return
        try:
            stat_result = os.stat(event.src_path)
        except FileNotFoundError:
            return
        self.events_queue.put(self.construct_message(stat_result, event, is_new_file=False))
 
RETRY_WAIT_SEC = 2
CREATE_CONNECTION_TIMEOUT_SEC = 2
EXIT_TIMEOUT_SEC = 2
 
def _run_logdump_thread(logger_name, logdump_dir, logdump_args, name, message_queue, record_class):
    # filter what content from user_logs gets put into db
    log = logging.getLogger(logger_name)
    glob_pat = logdump_dir + f'/{name}_*'
    has_no_logs = len(glob.glob(glob_pat)) == 0
    if has_no_logs:
        log.debug(f"No matching files found for {glob_pat}")
    while True:
        while has_no_logs := len(glob.glob(glob_pat)) == 0:
            time.sleep(RETRY_WAIT_SEC)
        try:
            args = logdump_args + ('--dir='+logdump_dir, '-J', '-f', name)
            log.debug(f"Running logdump command {repr(' '.join(args))} for {name} in follow mode")
            p = subprocess.Popen(args, stdout=subprocess.PIPE, stdin=subprocess.DEVNULL, stderr=subprocess.DEVNULL, text=True)
            for line in p.stdout:
                log.debug(f"Log line read: {line}")
                message = record_class.from_json(name, line)
                message_queue.put(message)
            if p.returncode != 0:
                raise RuntimeError(f"{name} logdump exited with {p.returncode} ({repr(' '.join(args))})")
        except Exception as e:
            log.exception(f"Exception in log/telem follower for {name}")
 
@xconf.config
class dbIngestConfig(BaseDbDeviceConfig):
    proclist : str = xconf.field(default="/opt/MagAOX/config/proclist_%s.txt", help="Path to process list file, %s will be replaced with the value of $MAGAOX_ROLE (or an empty string if absent from the environment)")
    logdump_exe : str = xconf.field(default="/opt/MagAOX/bin/logdump", help="logdump (a.k.a. teldump) executable to use")
 
class dbIngest(XDevice):
    config : dbIngestConfig
    telem_threads : list[tuple[str, threading.Thread]]
    telem_queue : queue.Queue
    fs_observer : BaseObserverSubclassCallable
    fs_queue : queue.Queue
    last_update_ts_sec : float
    startup_ts_sec : float
    records_since_startup : float
    _connections : list[psycopg.Connection]
    _should_connect : bool = True
    
    #add user_log support here
    user_log_threads : list[tuple[str, threading.Thread]]
    user_log_queue : queue.Queue
 
    def launch_followers(self, dev):
        telem_args = self.log.name + '.' + dev, '/opt/MagAOX/telem', (self.config.logdump_exe, '--ext=.bintel'), dev, self.telem_queuetelem_queue, Telem
        telem_thread = threading.Thread(target=_run_logdump_thread, args=telem_args, daemon=True)
        telem_thread.start()
        self.log.debug(f"Watching {dev} for incoming telem")
        self.telem_threadstelem_threads.append((dev, telem_thread))
 
        if dev == "observers":
            ULog_args = self.log.name + '.' + dev, '/opt/MagAOX/logs', (self.config.logdump_exe, '--ext=.binlog'), dev, self.user_log_queueuser_log_queue, UserLog
            user_log_thread = threading.Thread(target=_run_logdump_thread, args= ULog_args, daemon=True)
            user_log_thread.start()
            self.log.debug(f"Watching {dev} for incoming user logs")
            self.user_log_threadsuser_log_threads.append((dev, user_log_thread))
 
    def refresh_properties(self):
        self.properties['last_update']['timestamp'] = self.last_update_ts_seclast_update_ts_sec
        self.update_property(self.properties['last_update'])
        self.properties['records']['since_startup'] = self.records_since_startuprecords_since_startup
        self.properties['records']['per_sec'] = self.records_since_startuprecords_since_startup / (time.time() - self.startup_ts_secstartup_ts_sec)
        self.update_property(self.properties['records'])
 
    def setup(self):
        self.last_update_ts_seclast_update_ts_sec = time.time()
        self.records_since_startuprecords_since_startup = 0
        self.records_per_sec = 0.0
        last_update = properties.NumberVector(name="last_update", perm=constants.PropertyPerm.READ_ONLY)
        last_update.add_element(DefNumber(
            name="timestamp",
            _value=self.last_update_ts_seclast_update_ts_sec,
            min=0.0, max=1e200, format='%f',
            step=1e-6,
        ))
        self.add_property(last_update)
        
        records = properties.NumberVector(name="records", perm=constants.PropertyPerm.READ_ONLY)
        records.add_element(DefNumber(
            name="per_sec",
            _value=0.0,
            min=0.0, max=1e200, format='%f',
            step=1e-6,
        ))
        records.add_element(DefNumber(
            name="since_startup",
            _value=0,
            min=0, max=1_000_000_000, format='%i',
            step=1,
        ))
        self.add_property(records)
 
        role = os.environ.get('MAGAOX_ROLE', '')
        proclist = pathlib.Path(self.config.proclist.replace('%s', role))
        if not proclist.exists():
            raise RuntimeError(f"No process list at {proclist}")
 
        device_names = set()
 
        with proclist.open() as fh:
            for line in fh:
                if not line.strip():
                    continue
                if line.strip()[0] == '#':
                    continue
                parts = line.split(None, 1)
                if len(parts) != 2:
                    raise RuntimeError(f"Got malformed proclist line: {repr(line)}")
                device_names.add(parts[0])
 
        self.user_log_queueuser_log_queue = queue.Queue()
        self.user_log_threadsuser_log_threads = []
 
        self.telem_queuetelem_queue = queue.Queue()
        self.telem_threadstelem_threads = []
        for dev in device_names:
            self.launch_followers(dev)
        
        self.startup_ts_secstartup_ts_sec = time.time()
        
        self._connections_connections = self.config.connect_to_databases()
        self._should_connect_should_connect = False
 
        # rescan for inventory
        self.rescan_files()
 
        self.fs_queuefs_queue = queue.Queue()
        event_handler = NewXFilesHandler(self.config.hostname, self.fs_queuefs_queue, self.log.name + '.fs_observer')
        self.fs_observerfs_observer = Observer()
        for dirname in self.config.data_dirs:
            dirpath = self.config.common_path_prefix / dirname
            if not dirpath.exists():
                self.log.debug(f"No {dirpath} to watch")
                continue
            self.fs_observerfs_observer.schedule(event_handler, dirpath, recursive=True)
            self.log.info(f"Watching {dirpath} for changes")
        self.fs_observerfs_observer.start()
 
 
 
    def rescan_files(self):
        search_paths = [self.config.common_path_prefix / name for name in self.config.data_dirs]
        try:
            for conn in self._connections_connections:
                with conn.cursor() as cur:
                    # n.b. the state of the file inventory and which files
                    # are 'new' can be different depending on which
                    # database we're talking about, so we rescan once
                    # per connection
                    self.log.debug(f"Scanning for new files for {conn.info.dsn}")
                    ingest.update_file_inventory(
                        cur,
                        self.config.hostname,
                        search_paths,
                        self.config.ignore_patterns.files, self.config.ignore_patterns.directories
                    )
        except Exception:
            self.log.exception(f"Failed to rescan/inventory files for {conn.info.dsn}, attempting to reconnect")
            self._should_connect_should_connect = True
            return
        self.log.info(f"Completed startup rescan of file inventory for {self.config.hostname} from {search_paths}")
 
    def ingest_line(self, line):
        # avoid infinite loop of modifying log file and getting notified of the modification
        if self.log_file_name.encode('utf8') not in line:
            self.log.debug(line)
 
    def loop(self):
        if self._should_connect_should_connect:
            self._connections_connections = self.config.connect_to_databases(existing_connections=self._connections_connections)
            self._should_connect_should_connect = False
 
        telems = []
        try:
            while rec := self.telem_queuetelem_queue.get(timeout=0.1):
                telems.append(rec)
                self.records_since_startuprecords_since_startup += 1
        except queue.Empty:
            pass
 
        fs_events = []
        try:
            while rec := self.fs_queuefs_queue.get(timeout=0.1):
                fs_events.append(rec)
                self.records_since_startuprecords_since_startup += 1
        except queue.Empty:
            pass
 
        #update for userlog here too
        user_logs = []
        try:
            while rec := self.user_log_queueuser_log_queue.get(timeout=0.1):
                user_logs.append(rec)
                self.records_since_startuprecords_since_startup += 1
        except queue.Empty:
            pass
 
        try:
            for conn in self._connections_connections:
                self.log.debug(f"Batching ingest for {conn.info.dsn}")
                with conn.transaction():
                    cur = conn.cursor()
                    ingest.batch_telem(cur, telems)
                    ingest.batch_file_origins(cur, fs_events)
                    ingest.batch_user_log(cur, user_logs)
        except Exception as e:
            self.log.exception("Caught exception in batch ingest, reconnecting")
            self._should_connect_should_connect = True
 
        this_ts_sec = time.time()
        self.last_update_ts_seclast_update_ts_sec = this_ts_sec
        self.refresh_properties()