openrvdas.server.logger_runner

  1#!/usr/bin/env python3
  2"""Low-level class to run a logger config in its own process and write
  3its stderr to a file.
  4
  5Can be run from the command line as follows:
  6```
  7   server/logger_runner.py \
  8     --config test/NBP1406/NBP1406_cruise.yaml:gyr1->net \
  9     --stderr_file /var/log/openrvdas/gyr1.stderr
 10```
 11
 12But its main intended use is to be invoked by another module to start
 13a logger in its own, non-blocking process:
 14```
 15    runner = LoggerRunner(config=config, name=logger,
 16                          stderr_file=stderr_file,
 17                          logger_log_level=self.logger_log_level)
 18    self.logger_runner_map[logger] = runner
 19    self.logger_runner_map[logger].start()
 20```
 21Simulated Serial Ports:
 22
 23The NBP1406_cruise.yaml file above specifies configs that read from
 24simulated serial ports and write to UDP port 6224. To get the configs
 25to actually run, you'll need to run
 26
 27```
 28  logger/utils/simulate_data.py --config test/NBP1406/simulate_NBP1406.yaml
 29```
 30in a separate terminal window to create the virtual serial ports the
 31sample config references and feed simulated data through them.)
 32
 33To verify that the scripts are actually working as intended, you can
 34create a network listener on port 6224 in yet another window:
 35```
 36  logger/listener/listen.py --network :6224
 37```
 38"""
 39import logging
 40import multiprocessing
 41import os
 42import pprint
 43import signal
 44import time
 45
 46from importlib import reload
 47from logging.handlers import RotatingFileHandler
 48from setproctitle import setproctitle
 49
 50# Add the openrvdas/ directory to module search path
 51from logger.utils.read_config import read_config  # noqa: E402
 52from logger.utils.stderr_logging import DEFAULT_LOGGING_FORMAT  # noqa: E402
 53from logger.listener.listen import ListenerFromLoggerConfig  # noqa: E402
 54
 55# For writing to cached data server
 56from logger.transforms.to_das_record_transform import ToDASRecordTransform  # noqa: E402
 57from logger.writers.cached_data_writer import CachedDataWriter  # noqa: E402
 58from logger.writers.composed_writer import ComposedWriter  # noqa: E402
 59from logger.utils.stderr_logging import StdErrLoggingHandler  # noqa: E402
 60
 61# Rotate stderr logs out so that their sizes remain manageable. Plan to keep all
 62# stderr logs, but don't swamp if something goes awry. Note: these values
 63# should probably be extracted to a settings.py file somewhere.
 64STDERR_MAX_BYTES = 1000000  # 10M
 65STDERR_BACKUP_COUNT = 100  # 100 backups should be plenty
 66
 67
 68################################################################################
 69def kill_handler(self, signum):
 70    """Translate an external signal (such as we'd get from os.kill) into a
 71    KeyboardInterrupt, which will signal the start() loop to exit nicely."""
 72    logging.info('Received external kill')
 73    raise KeyboardInterrupt('Received external kill signal')
 74
 75
 76################################################################################
 77def config_from_filename(filename):
 78    """Load a logger configuration from a filename. If there's a ':' in
 79    the config file name, then we expect what is before the colon to be
 80    a cruise definition, and what is after to be the name of a
 81    configuration inside that definition.
 82    """
 83    config_name = None
 84    if filename.find(':') > 0:
 85        (filename, config_name) = filename.split(':', maxsplit=1)
 86    config = read_config(filename)
 87
 88    if config_name:
 89        config_dict = config.get('configs')
 90        if not config_dict:
 91            raise ValueError('Configuration name "%s" specified, but no '
 92                             '"configs" section found in file "%s"'
 93                             % (config_name, filename))
 94        config = config_dict.get(config_name)
 95        if not config:
 96            raise ValueError('Configuration name "%s" not found in file "%s"'
 97                             % (config_name, filename))
 98    logging.info('Loaded config file: %s', pprint.pformat(config))
 99    return config
100
101
102################################################################################
103def config_is_runnable(config):
104    """Is this logger configuration runnable? (Or, e.g. does it just have
105    a name and no readers/transforms/writers?)
106    """
107    if not config:
108        return False
109    return 'readers' in config or 'writers' in config
110
111
112################################################################################
113def run_logger(logger, config, stderr_filename=None, stderr_data_server=None,
114               log_level=logging.INFO):
115    """Run a logger, sending its stderr to a cached data server if so indicated
116
117    logger -    Name of logger
118
119    config -    Config dict
120
121    stderr_filename  - If not None, send stderr to this file.
122
123    stderr_data_server  - If not None, host:port of cached data server to
124                send stderr messages to.
125
126    log_level - Level at which logger should be logging (e.g logging.WARNING,
127                logging.INFO, etc.
128    """
129    # Reset logging to its freshly-imported state
130    reload(logging)
131
132    if stderr_filename:
133        stderr_handlers = [RotatingFileHandler(stderr_filename,
134                                               maxBytes=STDERR_MAX_BYTES,
135                                               backupCount=STDERR_BACKUP_COUNT)]
136    else:
137        stderr_handlers = []
138    logging.basicConfig(
139        handlers=stderr_handlers,
140        level=log_level,
141        format=DEFAULT_LOGGING_FORMAT)
142
143    if stderr_data_server:
144        field_name = 'stderr:logger:' + logger
145        cds_writer = ComposedWriter(
146            transforms=ToDASRecordTransform(data_id='stderr', field_name=field_name),
147            writers=CachedDataWriter(data_server=stderr_data_server))
148        logging.getLogger().addHandler(StdErrLoggingHandler(cds_writer))
149
150    # Set the name of the process for ps
151    config_name = config.get('name', 'no_name')
152    setproctitle('openrvdas/server/logger_runner.py:' + config_name)
153    logging.info(f'Starting logger {logger} config {config_name}')
154
155    try:
156        if config_is_runnable(config):
157            listener = ListenerFromLoggerConfig(config=config)
158            try:
159                listener.run()
160            except KeyboardInterrupt:
161                logging.warning(f'Received quit for {config_name}')
162    except Exception as e:
163        logging.fatal(e)
164
165    # Allow a moment for stderr_writers to finish up
166    time.sleep(0.25)
167
168
169################################################################################
170class LoggerRunner:
171    ############################
172    def __init__(self, config, name=None, stderr_filename=None,
173                 stderr_data_server=None, logger_log_level=logging.WARNING):
174        """Create a LoggerRunner.
175        ```
176        config   - Python dict containing the logger configuration to be run
177
178        name     - Optional name to give to logger process.
179
180        stderr_filename - Optional name of file to write stderr to.
181
182        stderr_data_server - Optional host:port of a cached data server to
183                   send encoded stderr messages to.
184
185        logger_log_level - At what logging level our logger should operate.
186        ```
187        """
188        self.config = config
189        self.name = name or config.get('name', 'Unnamed logger')
190        self.stderr_filename = stderr_filename
191        self.stderr_data_server = stderr_data_server
192        self.logger_log_level = logger_log_level
193
194        self.process = None     # this is hold the logger process
195        self.failed = False     # flag - has logger failed?
196        self.quit_flag = False  # flag - has quit been signaled?
197
198        # Set the signal handler so that an external break will get
199        # translated into a KeyboardInterrupt. But signal only works if
200        # we're in the main thread - catch if we're not, and just assume
201        # everything's gonna be okay and we'll get shut down with a proper
202        # "quit()" call otherwise.
203        try:
204            signal.signal(signal.SIGTERM, kill_handler)
205        except ValueError:
206            logging.debug('LoggerRunner not running in main thread; '
207                          'shutting down with Ctl-C may not work.')
208
209    ############################
210    def start(self):
211        """Start a listener subprocess."""
212        self.quit_flag = False
213        self.failed = False
214
215        # We're going to go ahead and create the process, even if the
216        # config is not runnable, just so we can get log messages that the
217        # config has been started.
218
219        # If config is not runnable, just say so and be done with it.
220        # if not self.is_runnable():
221        #  logging.info('Process %s is complete. Not running.', self.name)
222        #  return
223
224        run_logger_kwargs = {
225            'logger': self.name,
226            'config': self.config,
227            'stderr_filename': self.stderr_filename,
228            'stderr_data_server': self.stderr_data_server,
229            'log_level': self.logger_log_level
230        }
231        self.process = multiprocessing.Process(target=run_logger,
232                                               kwargs=run_logger_kwargs,
233                                               daemon=True)
234        self.process.start()
235
236    ############################
237    def is_runnable(self):
238        """Is this logger configuration runnable? (Or, e.g. does it just have
239        a name and no readers/transforms/writers?)
240        """
241        return config_is_runnable(self.config)
242
243    ############################
244    def is_alive(self):
245        """Is the logger in question alive?"""
246        return self.process and self.process.is_alive()
247
248    ############################
249    def is_failed(self):
250        """Return whether the logger has failed."""
251        return self.failed
252
253    ############################
254    def quit(self):
255        """Signal loop exit and try to cleanly terminate the process."""
256        self.quit_flag = True
257        if self.process:
258            # First attempt: terminate gracefully
259            self.process.terminate()
260            self.process.join(timeout=5)
261
262            if self.process.is_alive():
263                # Escalation: send SIGKILL (Unix only)
264                try:
265                    os.kill(self.process.pid, signal.SIGKILL)
266                except OSError:
267                    pass  # process may have already exited
268                self.process.join(timeout=5)
269
270                if self.process.is_alive():
271                    # If it's *still* alive, warn, and just live with the undead process
272                    logging.error(f'Process {self.process.pid} could not be killed')
273
274        self.process = None
275        self.failed = False
276
277
278################################################################################
279if __name__ == '__main__':
280    import argparse
281    parser = argparse.ArgumentParser()
282    parser.add_argument('--config', dest='config', action='store', required=True,
283                        help='Logger configuration to run. May either be the '
284                        'name of a file containing a single logger configuration '
285                        'or filename:config_name, for a file containing a cruise '
286                        'definition followed by the name of the specific '
287                        'configuration inside that definition.')
288
289    parser.add_argument('--name', dest='name', action='store', default=None,
290                        help='Name to give to logger process.')
291
292    parser.add_argument('--stderr_filename', dest='stderr_filename', default=None,
293                        help='Optional filename to which stderr should be '
294                        'written. Will attempt to create path if it does not '
295                        'exist.')
296
297    parser.add_argument('--stderr_data_server', dest='stderr_data_server', default=None,
298                        help='Optional host:port of a cached data server to which '
299                        ' stderr messages should be written.')
300
301    parser.add_argument('-v', '--verbosity', dest='verbosity',
302                        default=0, action='count',
303                        help='Increase output verbosity')
304
305    parser.add_argument('-V', '--logger_verbosity', dest='logger_verbosity',
306                        default=0, action='count',
307                        help='Increase output verbosity of component loggers')
308
309    args = parser.parse_args()
310
311    # Set up logging first of all
312
313    LOG_LEVELS = {0: logging.WARNING, 1: logging.INFO, 2: logging.DEBUG}
314    log_level = LOG_LEVELS[min(args.verbosity, max(LOG_LEVELS))]
315    logging.basicConfig(format=DEFAULT_LOGGING_FORMAT)
316    logging.getLogger().setLevel(log_level)
317
318    # What level do we want our component loggers to write?
319    logger_log_level = LOG_LEVELS[min(args.logger_verbosity, max(LOG_LEVELS))]
320
321    config = config_from_filename(args.config)
322
323    # Finally, create our runner and run it
324    runner = LoggerRunner(config=config,
325                          name=args.name,
326                          stderr_filename=args.stderr_filename,
327                          stderr_data_server=args.stderr_data_server,
328                          logger_log_level=logger_log_level)
329    runner.start()
330
331    # Wait for it to complete
332    runner.process.join()