openrvdas.logger.writers.composed_writer

  1#!/usr/bin/env python3
  2
  3import logging
  4import threading
  5
  6from logger.writers.writer import Writer  # noqa: E402
  7
  8
  9class ComposedWriter(Writer):
 10    ############################
 11    def __init__(self, transforms=[], writers=[], **kwargs):
 12        """
 13        Apply zero or more Transforms (in series) to passed records, then
 14        write them (in parallel threads) using the specified Writers.
 15
 16        ```
 17        transforms     A single Transform, a list of Transforms, or None.
 18
 19        writers        A single Writer or a list of Writers.
 20        ```
 21        Example:
 22        ```
 23        writer = ComposedWriter(transforms=[TimestampTransform(),
 24                                            PrefixTransform('gyr1')],
 25                                writers=[NetworkWriter(':6221'),
 26                                         LogfileWriter('/logs/gyr1')],
 27                                )
 28        ```
 29        NOTE: we make the rash assumption that transforms are thread-safe,
 30        that is, that no mischief or corrupted internal state will result if
 31        more than one thread calls a transform at the same time. To be
 32        thread-safe, a transform must protect any changes to its internal
 33        state with a non-re-entrant thread lock, as described in the threading
 34        module. We do *not* make this assumption of our writers, and impose a
 35        lock to prevent a writer's write() method from being called a second
 36        time if the first has not yet completed.
 37        """
 38        super().__init__(**kwargs)  # processes 'quiet' and type hints
 39
 40        # Make transforms a list if it's not. Even if it's only one transform.
 41        if not isinstance(transforms, list):
 42            self.transforms = [transforms]
 43        else:
 44            self.transforms = transforms
 45
 46        # Make writers a list if it's not. Even if it's only one writer.
 47        if not isinstance(writers, list):
 48            self.writers = [writers]
 49        else:
 50            self.writers = writers
 51
 52        # One lock per writer, to prevent us from accidental re-entry if a
 53        # new write is requested before the previous one has completed.
 54        self.writer_lock = [threading.Lock() for w in self.writers]
 55        self.exceptions = [None for w in self.writers]
 56
 57    ############################
 58
 59    def _run_writer(self, index, record):
 60        """Internal: grab the appropriate lock and call the appropriate
 61        write() method. If there's an exception, save it."""
 62        with self.writer_lock[index]:
 63            try:
 64                self.writers[index].write(record)
 65            except Exception as e:
 66                self.exceptions[index] = e
 67
 68    ############################
 69    def apply_transforms(self, record):
 70        """Internal: apply the transforms in series."""
 71        if record:
 72            for t in self.transforms:
 73                record = t.transform(record)
 74                if not record:
 75                    break
 76        return record
 77
 78    ############################
 79    def write(self, record):
 80        """Transform the passed record and dispatch it to writers."""
 81        # Transforms run in series
 82        record = self.apply_transforms(record)
 83        if record is None:
 84            return
 85
 86        # No idea why someone would instantiate without writers, but it's
 87        # plausible. Try to be accommodating.
 88        if not self.writers:
 89            return
 90
 91        # If we only have one writer, there's no point making things
 92        # complicated. Just write and return.
 93        if len(self.writers) == 1:
 94            self.writers[0].write(record)
 95            return
 96
 97        # Fire record off to write() requests for each writer.
 98        writer_threads = []
 99        for i in range(len(self.writers)):
100            try:
101                writer_name = str(type(self.writers[i]))
102                t = threading.Thread(target=self._run_writer, args=(i, record),
103                                     name=writer_name, daemon=True)
104                t.start()
105            except (OSError, RuntimeError) as e:
106                logging.error('ComposedWriter failed to write to %s: %s',
107                              writer_name, e)
108                t = None
109            writer_threads.append(t)
110
111        # Wait for all writes to complete
112        for t in writer_threads:
113            if t:
114                t.join()
115
116        # Were there any exceptions? Arbitrarily raise the first one in list
117        exceptions = [e for e in self.exceptions if e]
118        for e in exceptions:
119            logging.error(e)
120        if exceptions:
121            raise exceptions[0]