openrvdas.server.cached_data_server

   1#!/usr/bin/env python3
   2
   3"""Accept data in DASRecord or dict format via the cache_record() method,
   4then serve it to anyone who connects via a websocket. A
   5CachedDataServer can be instantiated by running this script from the
   6command line and providing one or more UDP ports on which to listen
   7for timestamped text data that it can parse into key:value pairs. It
   8may also be instantiated as part of a CachedDataWriter that can be
   9invoked on the command line via the listen.py script of via a
  10configuration file.
  11The following direct invocation of this script
  12```
  13    logger/utils/cached_data_server.py \
  14      --udp 6225 \
  15      --port 8766 \
  16      --disk_cache /var/tmp/openrvdas/disk_cache \
  17      --back_seconds 3600 \
  18      --v
  19```
  20says to
  211. Listen on the UDP port specified by --udp for JSON-encoded,
  22   timestamped, field:value pairs. (See the definition for cache_record(),
  23   below for formats understood.)
  242. Store the received data in memory, retaining the most recent 3600
  25   seconds for each field (default is 86400 seconds = 24 hours).
  26   (The total number of values cached per field is also limited by the
  27   ``max_records`` parameter and defaults to 2880, equivalent to two
  28   records per minute for 24 hours. It may be overridden to "infinite"
  29   by setting ``--max_records=0`` on the command line.)
  303. Periodically back up the in-memory cache to a disk-based cache at
  31   /var/tmp/openrvdas/disk_cache (By default, back up every 60 seconds;
  32   this can be overridden with the --cleanup_interval argument).
  334. Wait for clients to connect to the websocket at port 8766 and serve
  34   them the requested data. Web clients may issue JSON-encoded
  35   requests of the following formats (see the definition of
  36   serve_requests() for insight):
  37```
  38   {'type':'fields'}   - return a list of fields for which cache has data
  39   {'type':'describe',
  40    'fields':['field_1', 'field_2', 'field_3']}
  41       - return a dict of metadata descriptions for each specified field. If
  42         'fields' is omitted, return a dict of metadata for *all* fields
  43   {'type':'subscribe',
  44    'fields':{'field_1':{'seconds':50},
  45              'field_2':{'seconds':0, 'back_records':10},
  46              'field_3':{'seconds':-1}}}
  47       - subscribe to updates for field_1, field_2 and field_3. Allowable
  48         values for 'seconds':
  49            0  - provide only new values that arrive after subscription
  50           -1  - provide the most recent value, and then all future new ones
  51           num - provide num seconds of back data, then all future new ones
  52         If 'seconds' is missing, use '0' as the default.
  53
  54         If 'back_records' is present it must be a number greater than or equal
  55         to zero. If present and non-zero, the CDS will try to provide at least
  56         that many "back records" when it first returns, even if it has to go
  57         back further than the interval specified in 'seconds'.
  58   {'type':'ready'}
  59       - indicate that client is ready to receive the next set of updates
  60         for subscribed fields.
  61   {'type':'publish', 'data':{'timestamp':1555468528.452,
  62                              'fields':{'field_1':'value_1',
  63                                        'field_2':'value_2'}}}
  64       - submit new data to the cache (an alternative way to get data
  65         in without the same record size limits of a UDP packet).
  66```
  67
  68HTTP GET interface (requires websockets >= 12):
  69
  70In addition to the WebSocket API, the server accepts plain HTTP GET
  71requests on the same port. This allows shell scripts and other simple
  72tools to retrieve the latest cached values without implementing a
  73WebSocket handshake:
  74
  75```
  76   GET /fields
  77       Returns a JSON object listing all field names currently in cache:
  78       {"fields": ["field_1", "field_2", ...]}
  79
  80       Example:
  81         curl http://localhost:8766/fields
  82
  83   GET /latest/<field_1>[,<field_2>,...]
  84       Returns the most recent cached timestamp and value for each
  85       requested field as a JSON object:
  86       {"field_1": {"timestamp": 1555468528.452, "value": 21.3},
  87        "field_2": {"timestamp": 1555468530.001, "value": "A"},
  88        "field_3": null}   <- null when the field is not in cache
  89
  90       Example (single field):
  91         curl http://localhost:8766/latest/S330Lat
  92
  93       Example (multiple fields):
  94         curl http://localhost:8766/latest/S330Lat,S330Lon,MwxAirTemp
  95```
  96
  97All other plain HTTP requests receive 400 Bad Request. WebSocket
  98connections are unaffected. On systems with websockets < 12 the HTTP
  99routes are unavailable and behaviour is unchanged.
 100
 101CAUTION: HTTP GET requests run inside the same asyncio event loop as
 102all WebSocket connections. Occasional queries (e.g. populating an elog
 103entry on demand) are fine. High-frequency polling will delay WebSocket
 104data delivery to connected clients; use the WebSocket subscription API
 105for any use case requiring frequent or continuous updates.
 106"""
 107import asyncio
 108import json
 109import logging
 110import os
 111import os.path
 112import re
 113import threading
 114import time
 115
 116try:
 117    import websockets
 118    try:
 119        from websockets.exceptions import ConnectionClosed
 120    except ImportError:
 121        from websockets.connection import ConnectionClosed
 122    # websockets 12+ (asyncio server) exposes http11.Response for process_request handlers
 123    try:
 124        from websockets.http11 import Response as _WsResponse, Headers as _WsHeaders
 125        _WEBSOCKETS_HAS_HTTP11 = True
 126    except ImportError:
 127        _WEBSOCKETS_HAS_HTTP11 = False
 128except ModuleNotFoundError:
 129    raise ModuleNotFoundError('CachedDataServer requires websockets module.\n'
 130                              'Please run "pip3 install websockets".')
 131
 132from logger.utils.stderr_logging import StdErrLoggingHandler, DEFAULT_LOGGING_FORMAT  # noqa: E402
 133from logger.utils.das_record import DASRecord                 # noqa: E402
 134
 135logging.basicConfig(format=DEFAULT_LOGGING_FORMAT)
 136
 137
 138############################
 139class RecordCache:
 140    """Structure for storing/retrieving record data and metadata."""
 141
 142    def __init__(self):
 143        """
 144        In-memory storage for key:value pairs.
 145        """
 146        self.data = {}
 147        self.data_lock = threading.Lock()  # When operating on whole dict
 148
 149        self.metadata = {}
 150        self.metadata_lock = threading.Lock()
 151
 152        # Disk files we've tried to write to but failed.
 153        self.failed_files = set()
 154
 155        # Create a lock for each key so threads don't step on each other
 156        self.locks = {key: threading.Lock() for key in self.keys()}
 157
 158    ############################
 159    def cache_record(self, record):
 160        """Add the passed record to the cache.
 161        Expects passed records to be in one of two formats:
 162        1) DASRecord
 163        2) A dict encoding optionally a source data_id and timestamp and a
 164           mandatory 'fields' key of field_name: value pairs. This is the format
 165           emitted by default by ParseTransform:
 166      ```
 167           {
 168             'data_id': ...,    # optional
 169             'timestamp': ...,  # optional - use time.time() if missing
 170             'fields': {
 171               field_name: value,
 172               field_name: value,
 173               ...
 174             }
 175           }
 176      ```
 177        A twist on format (2) is that the values may either be a singleton
 178        (int, float, string, etc) or a list. If the value is a singleton,
 179        it is taken at face value. If it is a list, it is assumed to be a
 180        list of (value, timestamp) tuples, in which case the top-level
 181        timestamp, if any, is ignored.
 182      ```
 183           {
 184             'data_id': ...,  # optional
 185             'fields': {
 186                field_name: [(timestamp, value), (timestamp, value),...],
 187                field_name: [(timestamp, value), (timestamp, value),...],
 188                ...
 189             }
 190           }
 191      ```
 192        In addition to a 'fields' field, a record may contain a 'metadata'
 193        field. If present, the data server will look for a 'fields' dict
 194        inside the metadata dict and add the key-value pairs there to its
 195        cache of metadata about the fields:
 196      ```
 197           {'data_id': 's330',
 198            'fields': {'S330CourseMag': 244.29,
 199                       'S330CourseTrue': 219.61,
 200                       'S330Mode': 'A',
 201                       'S330SpeedKm': 16.5,
 202                       'S330SpeedKt': 8.9},
 203            'metadata': {'fields': {
 204              'S330CourseMag': {'description': 'Magnetic course',
 205                                'device': 's330',
 206                                'device_type': 'Seapath330',
 207                                'device_type_field': 'CourseMag',
 208                                'units': 'degrees'},
 209              'S330CourseTrue': {'description': 'True course',
 210                                 ...}
 211              }
 212            }}
 213      ```
 214        This metadata field will be generated sent at intervals by a
 215        RecordParser (and its enclosing ParseTransform) if the parser's
 216        ``metadata_interval`` value is not None.
 217        """
 218        logging.debug('cache_record() received: %s', record)
 219        if not record:
 220            logging.debug('cache_record() received empty record.')
 221            return
 222
 223        # If we've been passed a DASRecord, the field:value pairs are in a
 224        # field called, uh, 'fields'; if we've been passed a dict, look
 225        # for its 'fields' key.
 226        if isinstance(record, DASRecord):
 227            record_timestamp = record.timestamp
 228            fields = record.fields
 229            metadata = record.metadata
 230        elif isinstance(record, dict):
 231            record_timestamp = record.get('timestamp', time.time())
 232            fields = record.get('fields')
 233            metadata = record.get('metadata')
 234            if fields is None:
 235                logging.debug(
 236                    'Dict record passed to cache_record() has no '
 237                    '"fields" key, which either means it\'s not a dict '
 238                    'you should be passing, or it is in the old "field_dict" '
 239                    'format that assumes key:value pairs are at the top '
 240                    'level.')
 241                logging.debug('The record in question: %s', str(record))
 242                return
 243        else:
 244            logging.warning(
 245                'Received non-DASRecord, non-dict input (type: %s): %s',
 246                type(record),
 247                record)
 248            return
 249
 250        # Add values from record to cache
 251        for field, value in fields.items():
 252            if field not in self.locks:
 253                self.locks[field] = threading.Lock()
 254            with self.locks[field]:
 255                if field not in self.data:
 256                    self.data[field] = []
 257
 258                if isinstance(value, list):
 259                    # Okay, for this field we have a list of values - iterate
 260                    # through
 261                    for val in value:
 262                        # If element in the list is itself a list or a tuple,
 263                        # we'll assume it's a (timestamp, value) pair. Otherwise,
 264                        # use the default timestamp of 'now'.
 265                        if type(val) in [list, tuple]:
 266                            self._add_tuple(field, val)
 267                        else:
 268                            self._add_tuple(field, (record_timestamp, value))
 269                else:
 270                    # If type(value) is *not* a list, assume it's the value
 271                    # itself. Add it using the default timestamp.
 272                    self._add_tuple(field, (record_timestamp, value))
 273
 274            # Is there any metadata to add? Cache whatever is in the
 275            # metadata.data.fields dict. Blithely overwrite whatever might
 276            # be there already.
 277            if metadata:
 278                metadata_fields = metadata.get('fields', {})
 279                with self.metadata_lock:
 280                    for mfield, value in metadata_fields.items():
 281                        self.metadata[mfield] = value
 282
 283    ############################
 284    def _add_tuple(self, field, value_tuple):
 285        self.data[field].append(value_tuple)
 286
 287    ############################
 288    def keys(self):
 289        """Return a list of all keys in the cache."""
 290        return list(self.data.keys())
 291
 292    ############################
 293    def get_metadata(self, fields=None):
 294        """Return a dict of metadata for the specified list of fields. If no
 295        fields are specified, return metadata for all fields.
 296        """
 297        with self.metadata_lock:
 298            if fields:
 299                return {
 300                    field: self.metadata.get(
 301                        field, {}) for field in fields}
 302            else:
 303                return self.metadata
 304
 305    ############################
 306    def cleanup(self, oldest=0, max_records=0, min_back_records=0):
 307        """Remove any data from cache with a timestamp older than 'oldest'
 308        seconds, but keep at least one (most recent) value.
 309        If max_records is non-zero, truncate to that many of the most
 310        recent records. Always, though, keep at least min_back_records.
 311        """
 312        logging.debug('Cleaning up cache')
 313        fields = self.keys()
 314        for field in fields:
 315            if field not in self.locks:
 316                self.locks[field] = threading.Lock()
 317            with self.locks[field]:
 318                value_list = self.data[field]
 319
 320                if len(value_list) <= min_back_records:
 321                    continue
 322
 323                # If max_records is specified, truncate to keep that many of
 324                # most recent records.
 325                if max_records > min_back_records and len(value_list) > max_records:
 326                    value_list = value_list[-max_records:]
 327
 328                # Iterate until find value that's not too old, but leave at least
 329                # min_back_records.
 330                for i in range(len(value_list) - min_back_records):
 331                    if value_list[i][0] > oldest:
 332                        break
 333
 334                # But keep at least one value
 335                last_index = min(i, len(value_list) - 1)
 336                self.data[field] = value_list[last_index:]
 337
 338    ############################
 339    def save_to_disk(self, disk_cache):
 340        """Create one JSON-encoded cache file per field in the directory named
 341        by disk_cache.
 342        """
 343        logging.debug('Saving to cache.')
 344        if not disk_cache:
 345            logging.warning('save_to_disk called, but no disk_cache defined')
 346            return
 347
 348        if not os.path.exists(disk_cache):
 349            try:
 350                os.makedirs(disk_cache)
 351            except OSError as e:
 352                logging.error('Unable to create disk cache directory "%s": %s', disk_cache, e)
 353                return
 354
 355        fields = self.keys()
 356        for field in fields:
 357            disk_filename = disk_cache + '/' + field
 358            if disk_filename in self.failed_files:
 359                continue
 360            if field not in self.locks:
 361                self.locks[field] = threading.Lock()
 362            with self.locks[field]:
 363                try:
 364                    with open(disk_filename, 'w') as cache_file:
 365                        json.dump(self.data[field], cache_file)
 366                except (PermissionError, IOError, OSError) as e:
 367                    logging.warning('Unable to write disk cache file %s: %s', disk_filename, e)
 368                    self.failed_files.add(disk_filename)
 369
 370                # This is BAD practice; but use it to figure out what else might go wrong
 371                # so we can add it to specific exceptions above.s
 372                except Exception as e:
 373                    logging.warning('Unanticipated exception writing disk cache file %s: %s',
 374                                    disk_filename, e)
 375                    self.failed_files.add(disk_filename)
 376
 377    ############################
 378    def load_from_disk(self, disk_cache):
 379        """Load the data dict from directory of JSON-encoded cache files.
 380        """
 381        logging.info('Loading from disk at %s', disk_cache)
 382        if not disk_cache:
 383            logging.info('load_from_disk called, but no disk_cache defined')
 384            return
 385        try:
 386            if not os.path.exists(disk_cache):
 387                logging.info('load_from_disk: no cache found at "%s"', disk_cache)
 388                return
 389
 390            field_files = [f for f in os.listdir(disk_cache)
 391                           if os.path.isfile(os.path.join(disk_cache, f))]
 392            logging.debug('Got cached fields: %s', field_files)
 393            for field in field_files:
 394                if field not in self.locks:
 395                    self.locks[field] = threading.Lock()
 396                try:
 397                    with self.locks[field]:
 398                        with open(disk_cache + '/' + field, 'r') as cache_file:
 399                            self.data[field] = json.load(cache_file)
 400
 401                except (json.decoder.JSONDecodeError, UnicodeDecodeError):
 402                    logging.warning('Failed to parse cache for %s', field)
 403        except OSError as e:
 404            logging.error('Unable to access disk cache at %s: %s', disk_cache, e)
 405
 406
 407############################
 408class WebSocketConnection:
 409    """Handle the websocket connection, serving data as requested."""
 410    ############################
 411
 412    def __init__(self, websocket, cache, interval):
 413        self.websocket = websocket
 414        self.cache = cache
 415        self.interval = interval
 416        self.quit_flag = False
 417
 418    ############################
 419    def closed(self):
 420        """Has our client closed the connection?"""
 421        return self.quit_flag
 422
 423    ############################
 424    def quit(self):
 425        """Close the connection from our end and quit."""
 426        self.quit_flag = True
 427
 428    ############################
 429    def get_matching_field_names(self, field_name):
 430        """If a wildcard field is present, returns a list
 431        (matching_field_names) of all the fields that match the
 432        pattern. Otherwise, it just returns the field_name as the sole
 433        entry in the list.
 434
 435        field_name - the name of the field as specified in the subscription request
 436        """
 437
 438        matching_field_names = set()
 439
 440        # If the field name is a wildcard
 441        if '*' in field_name:
 442            field_name = field_name.replace("*", ".+")
 443
 444            for field in self.cache.keys():
 445                if re.search(field_name, field):
 446                    matching_field_names.add(field)
 447
 448        # If here, the field name is not a wildcard
 449        else:
 450            matching_field_names.add(field_name)
 451
 452        return list(matching_field_names)
 453
 454    ############################
 455
 456    async def send_json_response(self, response, is_error=False):
 457        logging.debug('CachedDataServer sending %d bytes',
 458                      len(json.dumps(response)))
 459        await self.websocket.send(json.dumps(response))
 460        if is_error:
 461            logging.warning(response)
 462
 463    ############################
 464    async def serve_requests(self):
 465        """Wait for requests and serve data, if it exists, from
 466        cache. Requests are in JSON with request type encoded in
 467        'request_type' field. Recognized request types are:
 468        ```
 469        fields - return a (JSON encoded) list of fields for which cache
 470            has data.
 471        describe - return a (JSON encoded) dict of metadata for the listed
 472            fields.
 473        publish - look for a field called 'data' and expect its value to
 474            be a dict containing data in one of the formats accepted by
 475            cache_record().
 476        subscribe - look for a field called 'fields' in the request whose
 477            value is a dict of the format
 478            ```
 479              {field_name:{seconds:600, back_records:10},
 480               field_name:{seconds:0},...}
 481            ```
 482            The entire specification may also have a field called
 483            'interval', specifying how often server should provide
 484            updates. Will default to what was specified on command line
 485            with --interval flag (which itself defaults to 1 second
 486            intervals).
 487            ```
 488            ```
 489            A subscription will instruct the CachedDataServer to begin
 490            serving JSON messages of the format
 491            ```
 492              {
 493                field_name: [(timestamp, value), (timestamp, value),...],
 494                field_name: [(timestamp, value), (timestamp, value),...],
 495                field_name: [(timestamp, value), (timestamp, value),...],
 496              }
 497            ```
 498            Initially provide the number of seconds worth of back data
 499            requested, and on subsequent calls, return all data that have
 500            arrived since last call.
 501            NOTE: if the 'seconds' field is -1, server will only ever provide
 502            the single most recent value for the relevant field.
 503        ready - client has processed the previous data message and is ready
 504            for more.
 505        ```
 506        """
 507        # The field details specified in a subscribe request
 508        requested_fields = {}
 509
 510        # A map from field_name:latest_timestamp_sent. If latest_timestamp_sent is -1
 511        # then we'll always send just the most recent value we have for the field,
 512        # regardless of how many there are, or whether we've sent it before.
 513        field_timestamps = {}
 514
 515        # Output format requested by the client; set by a subscribe message.
 516        # Default matches the subscribe handler's own default so a 'ready'
 517        # received before any 'subscribe' degrades gracefully.
 518        requested_format = 'field_dict'
 519
 520        interval = self.interval  # Use the default interval, uh, by default
 521
 522        while not self.quit_flag:
 523            now = time.time()
 524            try:
 525                logging.debug('Waiting for client')
 526                raw_request = await self.websocket.recv()
 527                request = json.loads(raw_request)
 528
 529                # Make sure we've received a dict
 530                if not isinstance(request, dict):
 531                    await self.send_json_response(
 532                        {'status': 400, 'error': 'non-dict request received'},
 533                        is_error=True)
 534
 535                # Make sure request dict has a 'type' field
 536                elif 'type' not in request:
 537                    await self.send_json_response(
 538                        {'status': 400, 'error': 'no "type" field found in request'},
 539                        is_error=True)
 540
 541                # Let's see what type of request it is
 542
 543                # Send client a list of the variable names we're able to serve.
 544                elif request['type'] == 'fields':
 545                    logging.debug('fields request')
 546                    await self.send_json_response(
 547                        {'type': 'fields', 'status': 200,
 548                         'data': self.cache.keys()})
 549
 550                # Send client a dict of metadata descriptions; if they've
 551                # specified a set of fields, give just metadata for those;
 552                # otherwise send everything.
 553                elif request['type'] == 'describe':
 554                    logging.debug('describe request')
 555                    fields = request.get('fields')
 556                    result = self.cache.get_metadata(fields)
 557                    await self.send_json_response(
 558                        {'type': 'describe', 'status': 200, 'data': result})
 559
 560                # Client wants to publish to cache and provides a dict of data
 561                elif request['type'] == 'publish':
 562                    logging.debug('publish request')
 563                    data = request.get('data')
 564                    if data is None:
 565                        await self.send_json_response(
 566                            {'type': 'publish', 'status': 400,
 567                             'error': 'no data field found in request'},
 568                            is_error=True)
 569                    elif not isinstance(data, dict):
 570                        await self.send_json_response(
 571                            {'type': 'publish', 'status': 400,
 572                             'error': 'request has non-dict data field'},
 573                            is_error=True)
 574                    else:
 575                        self.cache.cache_record(data)
 576                        await self.send_json_response({'type': 'publish', 'status': 200})
 577
 578                # Client wants to subscribe, and provides a dict of requested
 579                # fields
 580                elif request['type'] == 'subscribe':
 581                    logging.debug('subscribe request')
 582                    # Have they given us a new subscription interval?
 583                    requested_interval = request.get('interval')
 584                    if requested_interval is not None:
 585                        try:
 586                            interval = float(requested_interval)
 587                        except ValueError:
 588                            await self.send_json_response(
 589                                {'type': 'subscribe', 'status': 400,
 590                                 'error': 'non-numeric interval requested'},
 591                                is_error=True)
 592                            continue
 593
 594                    # Which fields do they want?
 595                    raw_requested_fields = request.get('fields')
 596                    if not raw_requested_fields:
 597                        await self.send_json_response(
 598                            {'type': 'subscribe', 'status': 400,
 599                             'error': 'no fields found in subscribe request'},
 600                            is_error=True)
 601                        continue
 602
 603                    # What format do they want output in? field_dict?
 604                    # record_list? By default, use field_dict.
 605                    requested_format = request.get('format', 'field_dict')
 606
 607                    # Parse out request field names and number of back seconds
 608                    # requested. Encode that as 'last timestamp sent', unless back
 609                    # seconds == -1. If -1, save it as -1, so that we know we're
 610                    # always just sending the the most recent field value. Stores
 611                    # all fields, including expanded entries from a wildcard, in the
 612                    # requested_fields dict.
 613
 614                    now = time.time()
 615
 616                    # Reset requested field_timestamps and field_back_records
 617                    requested_fields = {}
 618                    field_timestamps = {}    # last timestamp seen
 619
 620                    logging.debug('Subscription requested')
 621                    for field_name, field_spec in raw_requested_fields.items():
 622                        matching_field_names = self.get_matching_field_names(field_name)
 623
 624                        for matching_field_name in matching_field_names:
 625                            requested_fields[matching_field_name] = field_spec
 626                            # If we don't have a field spec dict
 627                            if isinstance(field_spec, dict):
 628                                back_records = field_spec.get('back_records', 0)
 629                                back_seconds = field_spec.get('seconds', 0)
 630                            else:
 631                                back_records = 0
 632                                back_seconds = 0
 633
 634                            # Now figure out what's the latest timestamp we have for this
 635                            # field name that respects the back_records and back_seconds
 636                            # specification.
 637                            field_timestamps[matching_field_name] = 0  # if nothing else
 638
 639                            if field_name not in self.cache.locks:
 640                                logging.debug('No data for requested field %s', matching_field_name)
 641                                continue
 642                            with self.cache.locks[field_name]:
 643                                field_cache = self.cache.data.get(matching_field_name)
 644                                if field_cache is None:
 645                                    logging.debug('No cached data for %s', matching_field_name)
 646                                    continue
 647
 648                                logging.debug('    %s: %d records available; %d requested, '
 649                                              '%d seconds', matching_field_name, len(field_cache),
 650                                              back_records, back_seconds)
 651                                # If no data for requested field, skip.
 652                                if not field_cache or not field_cache[-1]:
 653                                    continue
 654
 655                                # If special case 0, they only want records that come after
 656                                # this point in time. Set the  last timestamp seen as the
 657                                # most-recently seen timestamp, or zero if no entries.
 658                                if back_seconds == 0:
 659                                    if len(field_cache) > 0:
 660                                        field_timestamps[matching_field_name] = field_cache[-1][0]
 661                                    else:
 662                                        field_timestamps[matching_field_name] = 0
 663                                    continue
 664
 665                                # If special case -1, they want just single most recent
 666                                # value. Set the last timestamp seen as the second to last
 667                                # timestamp if multiple entries, or as zero, if only 1.
 668                                if back_seconds == -1:
 669                                    if len(field_cache) > 1:
 670                                        field_timestamps[matching_field_name] = field_cache[-2][0]
 671                                    else:
 672                                        field_timestamps[matching_field_name] = 0
 673                                    continue
 674
 675                                # We've been told to return at least 'back_records' records; if
 676                                # there aren't at least that many, leave field_timestamps[field]
 677                                # at zero to return all we've got.
 678                                if len(field_cache) <= back_records:
 679                                    continue
 680
 681                                # If here, we've got at least 'back_records' records, and want to
 682                                # search backward to include the last 'back_seconds' seconds of
 683                                # them. Could do more efficiently with some sort of binary search.
 684                                this_record_index = len(field_cache) - back_records - 1
 685                                while this_record_index >= 0:
 686                                    # Recall that each element is (timestamp, value)
 687                                    this_timestamp = field_cache[this_record_index][0]
 688
 689                                    if now - this_timestamp > back_seconds:
 690                                        # Set our 'last seen' timestamp as timestamp of previous
 691                                        # record and stop looking.
 692                                        prev_timestamp = field_cache[this_record_index-1][0]
 693                                        field_timestamps[matching_field_name] = prev_timestamp
 694                                        break
 695                                    this_record_index -= 1
 696
 697                    if raw_requested_fields and not requested_fields:
 698                        logging.info('Request doesn\'t match any existing fields')
 699
 700                    # Let client know request succeeded
 701                    await self.send_json_response({'type': 'subscribe', 'status': 200})
 702
 703                # Client just letting us know it's ready for more. If there are
 704                # fields that have been requested, send along any new data for
 705                # them.
 706                elif request['type'] == 'ready':
 707                    logging.debug('Websocket got ready...')
 708                    if not field_timestamps:
 709                        # Client has told us that they're ready, but there are no
 710                        # fields that match their request. Let them know, then
 711                        # pause a moment before we try fielding their next
 712                        # request.
 713                        await self.send_json_response(
 714                            {'type': 'ready', 'status': 400,
 715                             'error': 'client ready, but no matching fields found (yet).'},
 716                            is_error=False)
 717                        await asyncio.sleep(self.interval * 5)
 718
 719                    ##########
 720                    results = {}
 721                    if requested_format == 'field_dict':
 722                        for field_name, field_spec in requested_fields.items():
 723                            if field_name not in self.cache.locks:
 724                                logging.debug('No data for requested field %s', field_name)
 725                                continue
 726
 727                            with self.cache.locks[field_name]:
 728                                field_cache = self.cache.data.get(field_name)
 729                                if field_cache is None:
 730                                    logging.debug(
 731                                        'No cached data for %s', field_name)
 732                                    continue
 733
 734                                # If no data for requested field, skip.
 735                                if not field_cache or not field_cache[-1]:
 736                                    continue
 737
 738                                # If special case -1, they want just single most recent
 739                                # value, then future results. Grab last value, then set its
 740                                # timestamp as the last one we've seen.
 741                                back_seconds = field_spec.get('back_seconds', 0)
 742                                if back_seconds == -1:
 743                                    last_value = field_cache[-1]
 744                                    results[field_name] = [last_value]
 745                                    # ts of last value
 746                                    field_timestamps[field_name] = last_value[0]
 747                                    continue
 748
 749                                # Otherwise - if no data newer than the latest
 750                                # timestamp we've already sent, skip,
 751                                latest_timestamp = field_timestamps.get(field_name, 0)
 752                                if not field_cache[-1][0] > latest_timestamp:
 753                                    continue
 754
 755                                # Otherwise, copy over records arrived since
 756                                # latest_timestamp and update the latest_timestamp sent
 757                                # (first element of last pair in field_cache).
 758                                field_results = [
 759                                    pair for pair in field_cache if pair[0] > latest_timestamp]
 760                                results[field_name] = field_results
 761                                if field_results:
 762                                    field_timestamps[field_name] = field_results[-1][0]
 763
 764                    ##########
 765                    # If not outputting data as a field dict, output as a list
 766                    # of records.
 767                    elif requested_format == 'record_list':
 768                        records = {}
 769                        for field_name, field_spec in requested_fields.items():
 770                            if field_name not in self.cache.locks:
 771                                logging.debug(
 772                                    'No data for requested field %s', field_name)
 773                                continue
 774                            with self.cache.locks[field_name]:
 775                                latest_timestamp = field_timestamps.get(
 776                                    field_name, 0)
 777                                field_cache = self.cache.data.get(
 778                                    field_name, None)
 779
 780                                if not field_cache or not field_cache[-1]:
 781                                    logging.debug(
 782                                        'No cached data for %s', field_name)
 783                                    continue
 784
 785                                # If latest_timestamp is special case -1, they want just
 786                                # single most recent value, then future results. Grab
 787                                # last value, then set its timestamp as the last one
 788                                # we've seen.
 789                                elif latest_timestamp == -1:
 790                                    last_ts, last_value = field_cache[-1]
 791                                    if last_ts not in records:
 792                                        records[last_ts] = {}
 793                                    records[last_ts][field_name] = last_value
 794                                    field_timestamps[field_name] = last_ts
 795                                    continue
 796
 797                                # Otherwise - if no data newer than the latest
 798                                # timestamp we've already sent, skip,
 799                                elif not field_cache[-1][0] > latest_timestamp:
 800                                    continue
 801
 802                                # Otherwise, copy over records arrived since
 803                                # latest_timestamp and update the latest_timestamp sent
 804                                # (first element of last pair in field_cache).
 805                                else:
 806                                    # Get the new (ts, value) pairs for this
 807                                    # field
 808                                    field_results = [
 809                                        pair for pair in field_cache if pair[0] > latest_timestamp]
 810
 811                                    # We know field_results is non-empty because of previous
 812                                    # elif, so new latest timestamp is last ts
 813                                    # in it.
 814                                    field_timestamps[field_name] = field_results[-1][0]
 815
 816                                    # Collate values by timestamp, folding into values for
 817                                    # other fields.
 818                                    for ts, value in field_results:
 819                                        if ts not in records:
 820                                            records[ts] = {}
 821                                        records[ts][field_name] = value
 822
 823                        # Create and send a list with one DASRecord-like dict for
 824                        # each timestamp.
 825                        results = [{'timestamp': ts, 'fields': records[ts]}
 826                                   for ts in sorted(records)]
 827
 828                    # If unknown requested format
 829                    else:
 830                        mesg = (
 831                            'Unrecognized requested format: %s; valid formats are '
 832                            '"field_dict" and "record_list"' %
 833                            requested_format)
 834                        logging.warning(mesg)
 835                        await self.send_json_response({'status': 400, 'error': mesg}, is_error=True)
 836
 837                    logging.debug(
 838                        'Websocket results: %s...',
 839                        str(results)[
 840                            0:100])
 841
 842                    # Package up what results we have (if any) and send them
 843                    # off
 844                    await self.send_json_response({'type': 'data', 'status': 200,
 845                                                   'data': results})
 846
 847                    # New results or not, take a nap before trying to fetch
 848                    # more results
 849                    elapsed = time.time() - now
 850                    time_to_sleep = max(0, interval - elapsed)
 851                    logging.debug('Sleeping %g seconds', time_to_sleep)
 852                    await asyncio.sleep(time_to_sleep)
 853
 854                # If unrecognized request type - whine, then iterate
 855                else:
 856                    await self.send_json_response(
 857                        {'status': 400,
 858                         'error': 'unrecognized request type: %s' % request['type']},
 859                        is_error=True)
 860
 861            # If we got bad input, complain and loop
 862            except json.JSONDecodeError:
 863                await self.send_json_response(
 864                    {'status': 400, 'error': 'received unparseable JSON'},
 865                    is_error=True)
 866                logging.warning('unparseable JSON: %s', raw_request)
 867
 868            # If our connection closed, complain and exit gracefully
 869            except ConnectionClosed:
 870                logging.info('Client closed connection')
 871                self.quit()
 872
 873
 874##########################################################################
 875class CachedDataServer:
 876    """Class that caches field:value pairs passed to it in either a
 877    DASRecord or a simple dict. It also establishes a websocket server
 878    on the specified port and serves the cached values to clients that
 879    connect via a websocket.
 880
 881    WebSocket API:
 882
 883    The server listens for two types of requests:
 884    1. If the request is the string "variables", return a list of the
 885       names of the variables the server has in cache and is able to
 886       serve. The server will continue listening for follow up messages,
 887       most likely this one:
 888    2. If the request is a python dict, assume it is of the form:
 889    ```
 890        {field_1_name: {'seconds': num_secs},
 891         field_2_name: {'seconds': num_secs},
 892         ...}
 893    ```
 894       where seconds is a float representing the number of seconds of
 895       back data being requested.
 896       This field dict is passed to serve_fields(), which will to retrieve
 897       num_secs of back data for each of the specified fields and return it
 898       as a JSON-encoded dict of the form:
 899    ```
 900         {
 901           field_1_name: [(timestamp, value), (timestamp, value), ...],
 902           field_2_name: [(timestamp, value), (timestamp, value), ...],
 903           ...
 904         }
 905    ```
 906    The server will then await a "ready" message from the client, and when
 907    received, will loop and send a JSON-encoded dict of all the
 908    (timestamp, value) tuples that have come in since the previous
 909    request. It will continue this behavior indefinitely, waiting for a
 910    "ready" request and sending updates.
 911
 912    HTTP GET API (requires websockets >= 12):
 913
 914    Plain HTTP GET requests are also accepted on the same port, allowing
 915    simple tools to retrieve the latest cached value for one or more
 916    fields without a WebSocket handshake:
 917
 918      GET /fields
 919          Returns {"fields": ["field_1", "field_2", ...]}
 920
 921      GET /latest/<field_1>[,<field_2>,...]
 922          Returns {"field_1": {"timestamp": T, "value": V}, ...}
 923          Fields not present in the cache are returned as null.
 924
 925    CAUTION: HTTP GET requests are handled inside the same asyncio event
 926    loop that drives all WebSocket connections. Each request briefly
 927    acquires the cache's threading lock, blocking the loop for the
 928    duration of the lookup. For occasional one-off queries (e.g. populating
 929    an elog entry) this is negligible. High-frequency polling from scripts
 930    or automated tools will delay WebSocket data delivery to all connected
 931    clients. Use the WebSocket subscription API for any use case that
 932    requires frequent or continuous updates.
 933    """
 934
 935    ############################
 936    def __init__(
 937            self,
 938            port,
 939            interval=1,
 940            back_seconds=60 * 60,
 941            max_records=60 * 24,
 942            min_back_records=100,
 943            cleanup_interval=60,
 944            disk_cache=None):
 945        """
 946        port         Port on which to serve websocket connections
 947        interval     How frequently to serve updates
 948        back_seconds
 949                     How many seconds of back data to retain
 950        max_records
 951                     Maximum number of records to store for each variable
 952        min_back_records
 953                     Minimum number of back records to keep when purging old data
 954        cleanup_interval
 955                     How many seconds between calls to cleanup old cache entries
 956                     and save to disk (if disk_cache is specified)
 957        disk_cache   If not None, name of directory in which to backup values
 958                     from in-memory cache
 959        """
 960        self.port = port
 961        self.interval = interval
 962        self.back_seconds = back_seconds
 963        self.max_records = max_records
 964        self.min_back_records = min_back_records
 965        self.cleanup_interval = cleanup_interval
 966
 967        self.cache = RecordCache()
 968
 969        # If they've given us the name of a disk cache, try loading our
 970        # RecordCache from it.
 971        self.disk_cache = disk_cache
 972        if disk_cache:
 973            self.cache.load_from_disk(disk_cache)
 974
 975        # List where we'll store our websocket connections so that we can
 976        # keep track of which are still open, and signal them to close
 977        # when we're done.
 978        self._connections = []
 979        self._connection_lock = threading.Lock()
 980
 981        self.quit_flag = False
 982
 983        # Start a thread to loop through, cleaning up the cache and (if we've
 984        # been given a disk_cache file) backing memory up to it.
 985        threading.Thread(target=self.cleanup_loop, daemon=True).start()
 986
 987        # Fire up the thread that's going to the websocket server in our
 988        # event loop. Calling quit() it will close any remaining
 989        # connections and stop the event loop, terminating the server.
 990        self.server_thread = threading.Thread(
 991            target=self._start_event_loop, daemon=True)
 992        self.server_thread.start()
 993
 994    def _start_event_loop(self):
 995        """Initialize and run the asyncio event loop in the current thread."""
 996        try:
 997            self.event_loop = asyncio.new_event_loop()
 998            asyncio.set_event_loop(self.event_loop)  # Bind the loop to this thread
 999            self._run_websocket_server()
1000        except Exception as e:
1001            logging.error('Failed to start event loop: %s', str(e))
1002
1003    ############################
1004    def __del__(self):
1005        if self.event_loop:
1006            self.event_loop.stop()
1007            self.event_loop.close()
1008
1009    ############################
1010    def cache_record(self, record):
1011        """Cache the passed record."""
1012        self.cache.cache_record(record)
1013
1014    ############################
1015    def cleanup_loop(self):
1016        """Clear out records older than oldest seconds."""
1017        while not self.quit_flag:
1018            time.sleep(self.cleanup_interval)
1019
1020            # What's the oldest record we should retain?
1021            oldest = time.time() - self.back_seconds
1022            self.cache.cleanup(oldest=oldest, max_records=self.max_records,
1023                               min_back_records=self.min_back_records)
1024
1025            # If we're using a disk cache, save things now
1026            if self.disk_cache:
1027                self.cache.save_to_disk(self.disk_cache)
1028
1029    ############################
1030    def _run_websocket_server(self):
1031        """Start serving on the specified websocket."""
1032        async def start_server():
1033            logging.info('Starting WebSocketServer on port %d', self.port)
1034            try:
1035                extra_kwargs = {}
1036                if _WEBSOCKETS_HAS_HTTP11:
1037                    # Handle plain HTTP GET requests on the WebSocket port.
1038                    # Supports two routes for simple data retrieval without a
1039                    # full WebSocket handshake (see issue #367):
1040                    #
1041                    #   GET /fields
1042                    #       Returns JSON list of all cached field names.
1043                    #       Example: {"fields": ["Temp", "Pressure", ...]}
1044                    #
1045                    #   GET /latest/<field1>[,<field2>,...]
1046                    #       Returns the most recent timestamp+value for each
1047                    #       requested field.
1048                    #       Example: {"Temp": {"timestamp": 1234567890.0,
1049                    #                          "value": 21.3},
1050                    #                 "Pressure": null}   <- null if not cached
1051                    #
1052                    # All other non-WebSocket requests receive a 400 response.
1053                    # Proxy warnings are preserved from the original handler.
1054                    async def _handle_non_ws_request(connection, request):
1055                        upgrade = request.headers.get('Upgrade', '')
1056                        if upgrade.lower() == 'websocket':
1057                            return None  # let normal WebSocket upgrade proceed
1058
1059                        via = request.headers.get('Via', '')
1060                        if via:
1061                            logging.warning(
1062                                'WebSocket upgrade on port %d was blocked by an HTTP '
1063                                'proxy (Via: %s) which stripped the Upgrade header. '
1064                                'WebSocket connections cannot be established through '
1065                                'this proxy. Fix: enable SSL/WSS, configure the proxy '
1066                                'to pass WebSocket upgrades, or bypass the proxy for '
1067                                'this host.',
1068                                self.port, via)
1069
1070                        path = request.path
1071
1072                        if path == '/fields':
1073                            body = json.dumps(
1074                                {'fields': sorted(self.cache.keys())}
1075                            ).encode()
1076                            return _WsResponse(
1077                                200, 'OK',
1078                                _WsHeaders([('Content-Type', 'application/json')]),
1079                                body)
1080
1081                        if path.startswith('/latest/'):
1082                            field_names = path[len('/latest/'):].split(',')
1083                            result = {}
1084                            with self.cache.data_lock:
1085                                for field in field_names:
1086                                    entries = self.cache.data.get(field)
1087                                    if entries:
1088                                        timestamp, value = entries[-1]
1089                                        result[field] = {
1090                                            'timestamp': timestamp,
1091                                            'value': value,
1092                                        }
1093                                    else:
1094                                        result[field] = None
1095                            body = json.dumps(result).encode()
1096                            return _WsResponse(
1097                                200, 'OK',
1098                                _WsHeaders([('Content-Type', 'application/json')]),
1099                                body)
1100
1101                        logging.debug(
1102                            'Plain HTTP request on WebSocket port %d '
1103                            '(path %r is not a recognised CDS HTTP route). '
1104                            'Check nginx proxy_set_header Upgrade config.',
1105                            self.port, path)
1106                        return _WsResponse(
1107                            400, 'Bad Request',
1108                            _WsHeaders([('Content-Type', 'text/plain')]),
1109                            b'Expected a WebSocket upgrade or a recognised CDS '
1110                            b'HTTP route (/fields, /latest/<field,...>)\n')
1111
1112                    extra_kwargs['process_request'] = _handle_non_ws_request
1113
1114                self.websocket_server = await websockets.serve(
1115                    self._serve_websocket_data,
1116                    host='',
1117                    port=self.port,
1118                    **extra_kwargs
1119                )
1120                logging.info('WebSocket server running on port %d', self.port)
1121                await self.websocket_server.wait_closed()
1122            except OSError as e:
1123                logging.fatal('Failed to open websocket on port %d: %s', self.port, e)
1124                raise e
1125
1126        # Use asyncio.run to manage the event loop
1127        asyncio.run(start_server())
1128
1129    ############################
1130    def quit(self):
1131        """Exit the loop and shut down all loggers.
1132        """
1133        # Close any connections
1134        with self._connection_lock:
1135            self.quit_flag = True
1136
1137            for connection in self._connections:
1138                connection.quit()
1139        logging.info('WebSocketServer closed')
1140
1141        # Stop the event loop that's serving connections
1142        self.event_loop.stop()
1143
1144        # Wait for thread that's running the server to finish
1145        self.server_thread.join()
1146
1147    ############################
1148    """Top-level coroutine for running CachedDataServer."""
1149    async def _serve_websocket_data(self, websocket, unused_loop_arg=None):
1150        # Legacy websocket code passes in event loop as third argument; we don't need it,
1151        # but include so code works both pre and post WS14.
1152
1153        # Here is where we see the anomalous behavior - when constructed
1154        # directly, self.cache is as it should be: a shared cache. But
1155        # when invoked indirectly, e.g. as part of a listener via
1156        #
1157        #    listener = ListenerFromLoggerConfig(config)
1158        #    proc = multiprocessing.Process(target=listener.run, daemon=True)
1159        #    proc.start()
1160        #
1161        # then self.cache always appears ins in its initial (empty) state.
1162        connection = WebSocketConnection(websocket, self.cache, self.interval)
1163
1164        # Stash the connection so we can tell it to exit when we receive a
1165        # quit(). But first do some cleanup, getting rid of old
1166        # connections that have closed.
1167        with self._connection_lock:
1168            index = 0
1169            while index < len(self._connections):
1170                if self._connections[index].closed():
1171                    logging.debug('Disposing of closed connection.')
1172                    self._connections.pop(index)
1173                else:
1174                    index += 1
1175            # Now add the new connection
1176            self._connections.append(connection)
1177
1178        # If client disconnects, tell connection to quit
1179        try:
1180            await connection.serve_requests()
1181        except ConnectionClosed:
1182            logging.warning('client disconnected')
1183        except KeyboardInterrupt:
1184            logging.warning('Keyboard Interrupt')
1185
1186        connection.quit()
1187        await websocket.close()
1188
1189
1190##########################################################################
1191##########################################################################
1192if __name__ == '__main__':
1193    import argparse
1194
1195    from logger.readers.composed_reader import ComposedReader
1196    from logger.readers.udp_reader import UDPReader
1197    from logger.transforms.from_json_transform import FromJSONTransform
1198
1199    parser = argparse.ArgumentParser()
1200    parser.add_argument('--port', dest='port', required=True,
1201                        action='store', type=int,
1202                        help='Websocket port on which to serve data')
1203
1204    parser.add_argument('--udp', dest='udp', default=None, action='store',
1205                        help='Comma-separated list of network ports to listen '
1206                        'for data on, e.g. 6221,6224. Prefix by group id '
1207                        'to specify multicast.')
1208
1209    parser.add_argument('--disk_cache', dest='disk_cache', default=None,
1210                        action='store', help='If specified, periodically '
1211                        'backup the in-memory cache to disk. On restart, '
1212                        'data will be reloaded from this cache.')
1213
1214    parser.add_argument('--back_seconds', dest='back_seconds', action='store',
1215                        type=float, default=24 * 60 * 60,
1216                        help='Maximum number of seconds of old data to keep '
1217                        'for serving to new clients.')
1218
1219    parser.add_argument('--max_records', dest='max_records', action='store',
1220                        type=int, default=24 * 60 * 2,
1221                        help='Maximum number of records to store per variable.')
1222
1223    parser.add_argument('--min_back_records', dest='min_back_records', action='store',
1224                        type=float, default=64,
1225                        help='Minimum number of back records to keep when purging old data.')
1226
1227    parser.add_argument('--cleanup_interval', dest='cleanup_interval',
1228                        action='store', type=float, default=60,
1229                        help='How often to clean old data out of the cache.')
1230
1231    parser.add_argument('--interval', dest='interval', action='store',
1232                        type=float, default=0.5,
1233                        help='How many seconds to sleep between successive '
1234                        'sends of data to clients.')
1235
1236    parser.add_argument('--stderr_file', dest='stderr_file', default=None,
1237                        help='Optional file to which stderr messages should '
1238                        'be written.')
1239
1240    parser.add_argument('-v', '--verbosity', dest='verbosity', default=0,
1241                        action='count', help='Increase output verbosity')
1242    args = parser.parse_args()
1243
1244    # Set logging verbosity
1245    LOG_LEVELS = {0: logging.WARNING, 1: logging.INFO, 2: logging.DEBUG}
1246    log_level = LOG_LEVELS[min(args.verbosity, max(LOG_LEVELS))]
1247    logging.getLogger().setLevel(log_level)
1248
1249    if args.stderr_file:
1250        from logger.writers.text_file_writer import TextFileWriter  # noqa: E402
1251        stderr_writer = [TextFileWriter(filename=args.stderr_file,
1252                                        split_by_date=True)]
1253        logging.getLogger().addHandler(StdErrLoggingHandler(stderr_writer))
1254
1255    logging.info('Starting CachedDataServer')
1256    server = CachedDataServer(port=args.port,
1257                              interval=args.interval,
1258                              back_seconds=args.back_seconds,
1259                              max_records=args.max_records,
1260                              min_back_records=args.min_back_records,
1261                              cleanup_interval=args.cleanup_interval,
1262                              disk_cache=args.disk_cache)
1263
1264    # Only create reader(s) if they've given us a network to read from;
1265    # otherwise, count on data coming from websocket publish
1266    # connections.
1267    if args.udp:
1268        readers = []
1269        # Readers may either be just a port (to listen for broadcast) or
1270        # a multicast_group:port to listen for multicast.
1271        for udp_spec in args.udp.split(','):
1272            group_port = udp_spec.split(':')
1273            port = int(group_port[-1])
1274            multicast_group = group_port[-2] if len(group_port) == 2 else ''
1275            readers.append(UDPReader(port=port, mc_group=multicast_group))
1276        transform = FromJSONTransform()
1277        reader = ComposedReader(readers=readers, transforms=[transform])
1278
1279    # Loop, reading data and writing it to the cache
1280    try:
1281        while True:
1282            if args.udp:
1283                record = reader.read()
1284                server.cache_record(record)
1285            else:
1286                time.sleep(args.interval)
1287
1288    except KeyboardInterrupt:
1289        logging.warning('Received KeyboardInterrupt - shutting down')
1290        if args.disk_cache:
1291            logging.warning(
1292                'Will try to save to disk cache prior to shutdown...')
1293            server.cache.save_to_disk(args.disk_cache)
1294        server.quit()