openrvdas.server.sqlite_server_api

  1#!/usr/bin/env python3
  2"""API implementation for interacting with a SQLite3 data store,
  3   by Kevin Pedigo, based on in_memory_server_api.py
  4
  5   See api_tool.py for a simple script that exercises this class.
  6   See also server/server_api.py for full documentation of the API.
  7"""
  8
  9import logging
 10import pprint
 11import sys
 12import sqlite3
 13import os
 14import yaml
 15import gzip
 16from datetime import datetime
 17
 18
 19from server.server_api import ServerAPI  # noqa: E402
 20
 21DEFAULT_MAX_TRIES = 3
 22ID_SEPARATOR = ':'
 23
 24# Not a runtime option because the API doesn't have any
 25DATABASE_BACKUPS = True
 26DATABASE_COMPRESS = True
 27
 28# Default location of SQLite database to use/create
 29DIR_PATH = os.path.dirname(__file__)
 30DEFAULT_DATABASE_PATH = os.path.join(DIR_PATH, 'openrvdas.sql')
 31
 32# Effectively "time zero" for POSIX systems.
 33EPOCH_TIME_ZERO = datetime(1970, 1, 1, 0, 0, 0)
 34
 35########################################################################
 36# Let's trust SQLite and forget about thread locking.
 37# https://www.sqlite.org/lockingv3.html
 38# https://www.sqlite.org/threadsafe.html
 39# SQLITE3, by default, when built from source, builds in serialized mode
 40# (can be safely used by multiple threads without restriction)
 41# All distros tested are good.  To test yours:
 42# strings `which sqlite3` | grep THREADSAFE (should = 1)
 43
 44
 45class SQLiteServerAPI(ServerAPI):
 46    ############################
 47    def __init__(self, database_path=DEFAULT_DATABASE_PATH,
 48                 no_create_database=False):
 49        """
 50        database_path  - If specified, the path to the SQLite database to use
 51
 52        no_create_database - If True, and database does not exist, throw an error
 53                             rather than creating a new database.
 54        """
 55        super().__init__()
 56
 57        # Where do we l
 58        self.database_path = database_path
 59        self.no_create_database = no_create_database
 60
 61        self.config = {}
 62        self.callbacks = []
 63        self.status = []
 64        self.server_messages = []
 65        self.cx = None
 66        self.timestamp = self._get_database_timestamp()
 67
 68    def _database_exists(self):
 69        """Return True if SQLite database at self.database_path exists,
 70        and can be read without errors."""
 71        read_only_database_path = ''.join(['file:', self.database_path, '?mode=ro'])
 72        try:
 73            cx = sqlite3.connect(read_only_database_path, uri=True)  # noqa: F841
 74        except sqlite3.OperationalError:
 75            return False
 76        except sqlite3.Error as err:
 77            # Some other error
 78            logging.error(f'Unknown SQLite database error on read: {err}')
 79            return False
 80        return True
 81
 82    def _create_database(self):
 83        """Try to create the database."""
 84        logging.debug(f'Creating SQLite database "{self.database_path}"')
 85        cx = sqlite3.connect(self.database_path)
 86        cu = cx.cursor()
 87        cu.execute('CREATE TABLE Cruise (highlander integer primary key not null, config blob, [loaded_time] datetime, compressed integer)')  # noqa E501
 88        cu.execute('CREATE TABLE lastupdate (highlander integer primary key not null, [timestamp] datetime)')  # noqa E501
 89        cu.execute('CREATE TABLE logmessages (timestamp datetime primary key not null, loglevel integer, cruise text, source text, user text, message text)')  # noqa E501
 90
 91        # We need a time or we think the database is not initialized
 92        cu.execute('INSERT INTO lastupdate (highlander, timestamp) VALUES (1, CURRENT_TIMESTAMP);')
 93
 94        # Close, or not necessarily executed?
 95        cx.close()
 96
 97    #############################
 98    # API methods below are used in querying/modifying the API for the
 99    # record of the running state of loggers.
100    ############################
101    def _get_connection(self):
102        """ Return SQLite connection or get one """
103
104        def dict_factory(cursor, row):
105            """ Factory method for sqlite row as dictionary """
106            d = {}
107            for idx, col in enumerate(cursor.description):
108                d[col[0]] = row[idx]
109            return d
110
111        # Return cached connection if it exists
112        if self.cx is not None:
113            return self.cx
114
115        # Otherwise establish a connection to database
116        if not self._database_exists():
117            if self.no_create_database:
118                raise sqlite3.OperationalError(f'No database "{self.database_path}" found, '
119                                               'and flag "no_create_database" is set.')
120            # Otherwise, create the missing database
121            self._create_database()
122
123        # Open the database for use
124        try:
125            cx = sqlite3.connect(self.database_path,
126                                 check_same_thread=False,
127                                 detect_types=sqlite3.PARSE_DECLTYPES)
128        except sqlite3.Error as err:
129            # Some other error
130            logging.error(f'SQLite database error: {err}')
131            raise err
132
133        # Database exists and is now open
134        cx.row_factory = dict_factory
135        cx.isolation_level = None
136        # cx.execute('PRAGMA ... etc...');
137        # See if database is initialized
138        try:
139            cx.execute('SELECT timestamp from lastupdate')
140        except sqlite3.OperationalError:
141            # Database or table missing
142            logging.error('System error: SQLite database exists but is not initialized!')
143            raise sqlite3.OperationError
144        except sqlite3.Error as err:
145            # Some other error
146            logging.error(f'SQLite database error: {err}')
147            raise err
148        else:
149            self.cx = cx
150            return self.cx
151
152    ##################################################################
153    def _sql_query(self, query, *args):
154        """ Query the SQLite database.  Do NOT use this method
155            for INSERT, UPDATE, or DELETE as it does not
156            commit nor update the timestamp """
157
158        cx = self._get_connection()
159        try:
160            res = cx.execute(query, args)
161            rows = res.fetchall()
162            return rows
163        except sqlite3.OperationalError:
164            # No such table.  We should be logging this.
165            logging.error('System error: SQLite database exists but has no tables?')
166            return None
167        except sqlite3.Error as err:
168            logging.error(f'SQLite database error: {err}')
169            raise err
170
171    ##################################################################
172    def _sql_cmd(self, query, *args):
173        """ Execute a SQL command that modifies the database """
174
175        # NOTE(kped):  Consider not using the cached connection
176        # in this function to help avoid possible threading
177        # concurrency issues.
178        cx = self._get_connection()
179        try:
180            res = cx.execute(query, args)
181            rows = res.fetchall()
182            # Update the database timestamp
183            Q = 'INSERT OR REPLACE INTO lastupdate VALUES (1, ?)'
184            now = datetime.utcnow()
185            cx.execute(Q, (now,))
186            cx.commit()
187            # Note:  If using WAL, checkpoint
188            return rows
189        except sqlite3.OperationalError:
190            # No such table
191            logging.error(f'No such SQLite table for query: {query}')
192            raise sqlite3.OperationalError
193        except sqlite3.Error as err:
194            logging.error(f'SQLite database error: {err}')
195            raise err
196
197    ##################################################################
198    def _save_config(self):
199        """Save our config object to the database"""
200
201        Q = 'INSERT OR REPLACE INTO cruise \
202             (highlander, config, compressed) \
203             VALUES (1, ?, ?)'
204
205        try:
206            ydump = yaml.dump(self.config, sort_keys=False)
207            logging.debug(f'YAML dump: "{ydump}"')
208            conf = bytes(ydump, 'utf-8')
209            logging.debug(f'YAML conf: "{conf}"')
210            if DATABASE_COMPRESS:
211                conf = gzip.compress(conf)
212            self._sql_cmd(Q, conf, DATABASE_COMPRESS)
213            cx = self._get_connection()
214            # VACUUM takes a millisecond or so, but keeps the database
215            # size small, which speeds us back up.
216            cx.execute('VACUUM',)
217        except Exception as err:
218            logging.warn(f'Failed to save SQLite database: {err}')
219            raise err
220        else:
221            # logging.info('Database save successful')
222            pass
223
224    ##################################################################
225    def _get_database_timestamp(self):
226        """Get the timestamp from the sqlite database"""
227
228        Q = 'SELECT timestamp from lastupdate'
229        cx = self._get_connection()
230        try:
231            res = cx.execute(Q)
232            row = res.fetchone()
233            if not row:  # if no timestamp row, return 'time zero'
234                return EPOCH_TIME_ZERO
235
236            timestamp = row['timestamp']
237            if isinstance(timestamp, str):
238                timestamp = datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S.%f")
239
240            # Sanity check that we get what we expect
241            elif not isinstance(timestamp, datetime):
242                logging.warning('Got non-datetime timestamp from database!')
243                logging.warning(f'Type: "{type(timestamp)}", value: "{timestamp}"')
244            return timestamp
245
246        except sqlite3.OperationalError as err:
247            # No such table
248            logging.error('System error: SQLite database exists but is not initialized!')
249            raise err
250        except sqlite3.Error as err:
251            logging.error(f'Unhandled SQLite database error: {err}')
252            raise err
253
254    ##################################################################
255    def _do_we_need_to_reload(self):
256        """ Check database timestamp and reload config if needed """
257
258        db_timestamp = self._get_database_timestamp()
259        if db_timestamp > self.timestamp or self.config == {}:
260            Q = """SELECT
261                       config, compressed
262                   FROM
263                       cruise
264                   WHERE
265                       highlander=1"""
266            rows = self._sql_query(Q)
267            row0 = None
268            try:
269                row0 = rows[0]
270            except IndexError:
271                return None
272
273            if 'config' not in row0:
274                return None
275            conf = row0['config']
276
277            # Wanted bzip, but build problems (probably install script)
278            if row0.get('compressed', 0):
279                conf = gzip.decompress(conf)
280
281            # Wanted JSON, but datetime objects aren't JSON
282            # serializable, so went with YAML.
283            conf = yaml.load(conf, Loader=yaml.FullLoader)
284            if 'loggers' not in conf:
285                return None
286
287            self.timestamp = db_timestamp
288            self.config = conf
289
290    # For each of the get_* function, check if the timestamp
291    # is newer than our current timestamp, pull config from
292    # the database.
293    ##################################################################
294    def get_configuration(self):
295        """" Return cruise config for specified cruise id. """
296
297        self._do_we_need_to_reload()
298        return self.config or None
299
300    ############################
301    def get_modes(self):
302        """ Return list of modes defined for given cruise. """
303
304        config = self.get_configuration()
305        if not config:
306            return None
307        return list(config.get('modes', []))
308
309    ############################
310    def get_active_mode(self):
311        """ Return cruise config for specified cruise id."""
312
313        config = self.get_configuration()
314        if not config:
315            return None
316        return config.get('active_mode')
317
318    ############################
319    def get_default_mode(self):
320        """ Get the name of the default mode for the specified cruise
321        from the. data store. """
322
323        config = self.get_configuration()
324        if not config:
325            return None
326        return config.get('default_mode')
327
328    ############################
329    def get_logger(self, logger):
330        """Retrieve the logger spec for the specified logger id."""
331
332        loggers = self.get_loggers()   # which calls self._get_configuration
333        if logger not in loggers:
334            raise ValueError(f'No logger "{logger}" found')
335        return loggers.get(logger)
336
337    ############################
338    def get_loggers(self):
339        """Get a dict of
340            {logger_id:{'configs':[<name_1>,<name_2>,...],
341            'active':<name>},...}
342        for all loggers.
343        """
344
345        config = self.get_configuration()
346        if not config:
347            return {}
348
349        if 'loggers' not in config:
350            raise ValueError('No loggers found')
351        logger_configs = config.get('loggers')
352        if logger_configs is None:
353            raise ValueError('No logger configurations found')
354
355        # Fetch and insert the currently active config for each logger
356        # Note that this only changes our copy, not the config itself
357        for logger in logger_configs:
358            if 'active' not in logger_configs[logger]:
359                mode = self.get_logger_config_name(logger)
360                logger_configs[logger]['active'] = mode
361        return logger_configs
362
363    ############################
364    def get_logger_config(self, config_name):
365        """Retrieve the config associated with the specified name."""
366
367        config = self.get_configuration()
368        if config is None:
369            return {}
370        logger_configs = config.get('configs')
371        if logger_configs is None:
372            raise ValueError('No "configs" section found')
373        logger_config = logger_configs.get(config_name)
374        if logger_config is None:
375            raise ValueError(f'No logger config "{config_name}" in config')
376        return logger_config
377
378    ############################
379    def get_logger_configs(self, mode=None):
380        """Retrieve the configs associated with a cruise id and mode from the
381        data store. If mode is omitted, retrieve configs associated with
382        the cruise's current logger configs."""
383
384        loggers = self.get_loggers()
385        if not loggers:
386            return None
387
388        output = {}
389        for logger in loggers:
390            logger_config_name = self.get_logger_config_name(logger, mode)
391            output[logger] = self.get_logger_config(logger_config_name)
392
393        return output
394
395    ############################
396    def get_logger_config_name(self, logger_id, mode=None):
397        """ Retrieve name of the config associated with the specified logger
398        in the specified mode.  If mode is omitted, retrieve name of logger's
399        current config. """
400
401        config = self.get_configuration()
402        if not config:
403            return {}
404        loggers = config.get('loggers')
405        if loggers is None:
406            raise ValueError('No loggers found in config')
407
408        # No mode, so we want the active mode
409        if mode is None:
410            logger = loggers.get(logger_id)
411            if logger is None:
412                raise ValueError(f'Logger id {logger_id} has no mode!')
413            conf_name = logger.get('active')
414            if conf_name is not None:
415                return conf_name
416
417        # Mode given or no active conf, so get the default for this mode
418        modes = config.get('modes')
419        mode_configs = modes.get(mode)
420        if mode_configs is None:
421            raise ValueError(f'Requested mode {mode} is not defined')
422        logger_config_name = mode_configs.get(logger_id)
423        if logger_config_name is None:
424            raise ValueError(f'Logger {logger_id} has no config defined in mode {mode}')
425        return logger_config_name
426
427    #############################
428    def get_logger_config_names(self, logger_id):
429        """ Retrieve list of config names that are valid for the
430            specified logger .
431        > api.get_logger_config_names('NBP1406', 'knud')
432              ["off", "knud->net", "knud->net/file", "knud->net/file/db"]
433        """
434        logger = self.get_logger(logger_id)
435        return logger.get('configs', [])
436
437    ############################
438    # Methods for manipulating the desired state via API to indicate
439    # current mode and which loggers should be in which configs.
440    ############################
441    def set_active_mode(self, mode):
442        """Set the current mode of the specified cruise in the data store."""
443
444        config = self.get_configuration()
445        modes = config.get('modes')
446        if not modes:
447            raise ValueError('Config has no modes')
448        if mode not in modes:
449            raise ValueError(f'Config has no mode "{mode}"')
450
451        self.config['active_mode'] = mode
452
453        # Update the API's working config's loggers
454        # to match the new mode
455        for logger, conf in modes[mode].items():
456            self.config['loggers'][logger]['active'] = conf
457
458        self._save_config()
459        logging.info('Signaling update')
460        self.signal_update()
461
462    ############################
463    def set_active_logger_config(self, logger, config_name):
464        """Set specified logger to new config. NOTE: we have no way to check
465        whether logger is compatible with config, so we rely on whoever is
466        calling us to have made that determination."""
467
468        # self.logger_config[logger] = config_name
469        # NOTE: We can check that config_name is in logger[configs]
470        self.config['loggers'][logger]['active'] = config_name
471        self._save_config()
472        logging.info('Signaling update')
473        self.signal_update()
474
475    ############################
476    # Methods for feeding data from LoggerServer back into the API
477    ############################
478    def update_status(self, status):
479        """Save/register the loggers' retrieved status report with the API."""
480        self.status.append(((datetime.utcnow()), status))
481        # NOTE(kped) Do we need to write this to the database?
482        # logger_manager never calls this....
483
484    ############################
485    # Methods for getting status data from API
486    ############################
487
488    def get_status(self, since_timestamp=None):
489        """Retrieve a dict of the most-recent status report from each
490        logger. If since_timestamp is specified, retrieve all status reports
491        since that time."""
492
493        # Start by getting set of loggers for cruise. Store as
494        # cruise_id:logger for ease of lookup.
495        try:
496            logger_set = set([logger
497                              for logger in self.get_loggers()])
498        except ValueError:
499            logger_set = set()
500
501        logging.debug(f'logger_set: {logger_set}')
502
503        # Step backwards through status messages until we run out of
504        # status messages or reach termination condition. If
505        # since_timestamp==None, our termination is when we have a status
506        # for each of our loggers. If since_timestamp is a number, our
507        # termination is when we've grabbed all the statuses with a
508        # timestamp greater than the specified number.
509        status = {}
510
511        status_index = len(self.status) - 1
512        logging.debug(f'starting at status index {status_index}')
513        while logger_set and status_index >= 0:
514            # record is a dict of 'cruise_id:logger' : {fields}
515            (timestamp, record) = self.status[status_index]
516            logging.debug('%d: %f: %s',
517                          status_index, timestamp, pprint.pformat(record))
518
519            # If we've been given a numeric timestamp and we've stepped back
520            # in time to or before that timestamp, we're done - break out.
521            if since_timestamp is not None and timestamp <= since_timestamp:
522                break
523
524            # Otherwise, examine ids in this record to see if they're for
525            # the cruise in question.
526            for id, fields in record.items():
527                # If id is cruise_id:logger that we're interested in, grab it.
528                logging.debug(f'Is {id} in {logger_set}?')
529                if id in logger_set:
530                    if timestamp not in status:
531                        status[timestamp] = {}
532                    status[timestamp][id] = fields
533
534                    # If since_timestamp==None, we only want the latest status
535                    # for each logger. So once we've found it, remove the id
536                    # from the logger_set we're lookings. We'll drop out of the
537                    # loop when the set is empty.
538                    if since_timestamp is None:
539                        logger_set.discard(id)
540            status_index -= 1
541
542        return status
543
544    ############################
545    # Methods for storing/retrieving messages from servers/loggers/etc.
546    ############################
547    def message_log(self, source, user, log_level, message):
548        """ Timestamp and store the passed message. """
549
550        now = datetime.utcnow()
551        self.server_messages.append((now, source, user,
552                                     log_level, message))
553
554        # Keep server_messages from over-eating memory
555        while len(self.server_messages) > 1000:
556            self.server_messages.pop(0)
557
558        Q = 'INSERT INTO logmessages \
559             (timestamp, loglevel, cruise, source, user, message) \
560             VALUES(?, ?, ?, ?, ?, ?)'
561
562        cruise = self.config.get('cruise', {})
563        cruise_id = cruise.get('id', 'none')
564
565        self._sql_cmd(Q, now, log_level, cruise_id, source, user, message)
566
567    ############################
568    def get_message_log(self, source=None, user=None, log_level=sys.maxsize,
569                        since_timestamp=None):
570        """Retrieve log messages from source at or above log_level since
571        timestamp. If source is omitted, retrieve from all sources. If
572        log_level is omitted, retrieve at all levels. If since_timestamp is
573        omitted, only retrieve most recent message.
574        """
575
576        # NOTE:  Should we pull this from the database?
577        #        No... if they want more history, look directly.
578        index = len(self.server_messages) - 1
579        messages = []
580        while index >= 0:
581            message = self.server_messages[index]
582            (timestamp, mesg_source, mesg_user,
583             mesg_log_level, mesg_message) = message
584            # Have we gone back too far? If so, we're done.
585            if since_timestamp is not None and timestamp <= since_timestamp:
586                break
587
588            if mesg_log_level < log_level:
589                continue
590            if user and not mesg_user == user:
591                continue
592            if source and not mesg_source == source:
593                continue
594
595            messages.insert(0, message)
596
597            # Are we only looking for last message, and do we have a message?
598            if since_timestamp is None and messages:
599                break
600            index -= 1
601
602        return messages
603
604    #############################
605    # Save a copy before empyting out the database
606    ##############################
607    def _backup_database(self):
608        """ Backup the database """
609
610        config = self.get_configuration()
611        if not config:
612            logging.debug('No configuration to back up')
613            return
614        cruise = config.get('cruise', {})
615        cruise_id = cruise.get('id', 'none')
616
617        dt = datetime.utcnow().strftime('%Y%m%d%H%M%S')
618        ourpath = os.path.dirname(__file__)
619        filename = f'openrvdas-{cruise_id}-{dt}.sql'
620        dbfile = os.path.join(ourpath, filename)
621        try:
622            cx = self._get_connection()
623            cx.execute('VACUUM INTO ?', (dbfile,))
624        except Exception as err:
625            logging.warn(f'Failed to backup SQLite database: {err}')
626            pass
627
628    #####################################
629    def load_configuration(self, config):
630        """Add a complete cruise configuration (id, modes, configs,
631        default) to the data store."""
632
633        # Loaded new config, (optionally) backup old one
634        if DATABASE_BACKUPS:
635            self._backup_database()
636
637        self.config = config
638        # self.config['loaded_time'] = datetime.utcnow().isoformat()
639        self.config['loaded_time'] = datetime.utcnow()
640
641        # Some syntactic sugar to simplify config definitions
642        configs = self.config.get('configs')
643        for config_name, config in configs.items():
644            if config is None:
645                raise ValueError(f'No logger for "{config_name}" in cruise definition')
646            if 'name' not in config:
647                self.config['configs'][config_name]['name'] = config_name
648
649        # Set cruise into default mode, if one is defined
650        if 'default_mode' in self.config:
651            active_mode = self.config['default_mode']
652            self.set_active_mode(active_mode)
653        else:
654            logging.warn('Cruise has no default mode')
655        # Why not send the entire config to the CDS?  Why
656        # just *almost* all of it?  JSON issue?
657        cruise = self.config.get('cruise')
658        if cruise:
659            for key in ['id', 'start', 'end']:
660                if key not in self.config:
661                    self.config[key] = cruise.get(key)
662        self._save_config()
663        self.signal_load()
664
665    ###############################
666    def delete_configuration(self):
667        """Remove the specified cruise from the data store."""
668        self.config = {}
669        # self.logger_config = {}
670        self.callbacks = []
671        self.status = []
672        self._save_config()
673
674    ############################
675    # Methods for manually constructing/modifying a cruise spec via API
676    def add_mode(self, cruise_id, mode):
677        logging.warn('Method "add_mode" not implemented')
678
679    def delete_mode(self, cruise_id, mode):
680        logging.warn('Method "delete_mode" not implemented')
681
682    def add_logger(self, cruise_id, logger_id, logger_spec):
683        logging.warn('Method "add_logger" not implemented')
684
685    def delete_logger(self, cruise_id, logger_id):
686        logging.warn('Method "delete_logger" not implemented')
687
688    def add_config(self, cruise_id, config, config_spec):
689        logging.warn('Method "add_config" not implemented')
690
691    def add_config_to_logger(self, cruise_id, config, logger_id):
692        logging.warn('Method "add_config_to_logger" not implemented')
693
694    def add_config_to_mode(self, cruise_id, config, logger_id, mode):
695        logging.warn('Method "add_config_to_mode" not implemented')
696
697    def delete_config(self, cruise_id, config_id):
698        logging.warn('Method "delete_config" not implemented')