amari.xlog: Initial draft

`xamari xlog` can be used to maintin extra log for an Amarisoft LTE
service: in addition to native logs, xlog contains results of periodic
queries done via WebSocket. For example with the following aguments

	xamari xlog <wsuri>  stats/10s ue_get/3s erab_get/3s

xlog will emit results of stats query - every 10 seconds, results from
ue_get query - every 3 seconds, and similarly for erab_get query.

The results are saved into xlog in JSON Lines format for easy future
processing. For the reference below is a copy of corresponding help
entries:

    xlte$ xamari help xlog
    Usage: xamari xlog [OPTIONS] <wsuri> <logspec>+
    Maintain extra log for a service.

    The service is queried periodically according to logspec and results are saved
    in JSON format to a file (see 'xamari help jsonlog').

    <wsuri> is URI (see 'xamari help websock') of an Amarisoft-service.
    <logspec> is specification of what to log. It has the following parts:

        <query>[<options>]/<period>

    The query specifies a message, that should be used to query service. For
    example "stats", "ue_get", "erab_get", ... Query part is mandatory.

    Options specifies additional flags for the query. Options part can be omitted.

    Period specifies periodicity of how often the service should be queried.
    Period is optional and defaults to 60 seconds.

    Example for <logspec>+:

        stats[samples,rf]/30s  ue_get[stats]  erab_get/10s  qos_flow_get

    Options:

        -h  --help            show this help

    xlte$ xamari help jsonlog
    Some commands produce logs with JSON entries. Such logs are organized with JSON
    Lines format(*) with each entry occupying one line.

    Logs in JSON Lines format are handy to exchange in between programs, and with
    corresponding tools, e.g. jq(+), they can be also displayed in human-readable
    form and inspected quickly.

    (*) https://jsonlines.org/
    (+) https://stedolan.github.io/jq/

Example output:

(xlte3.venv) kirr@deca:~/src/wendelin/xlte$ xamari xlog ws://localhost:9001 ue_get/3s erab_get/3s
{"meta": {"event": "start", "time": "Wed, 19 Oct 2022 15:05:50 -0000", "generator": "xlog ws://localhost:9001 ue_get[]/3.0s erab_get[]/3.0s"}}
{"meta": {"event": "service attach", "time": "Wed, 19 Oct 2022 15:05:51 -0000", "srv_name": "ENB", "srv_type": "ENB", "srv_version": "2022-09-16"}}
{"message":"config_get","version":"2022-09-16","type":"ENB","name":"ENB","license_id":"d9a961c166d2d4b15249fc559cdec925efbbe942d73b143aff","license_user":"rapid.space","logs":{"layers":{"PHY":{"level":"error","max_size":0,"key":false,"crypto":false,"payload":false},"MAC":{"level":"error","max_size":0,"key":false,"crypto":false,"payload":false},"RLC":{"level":"error","max_size":0,"key":false,"crypto":false,"payload":false},"PDCP":{"level":"error","max_size":0,"key":false,"crypto":false,"payload":false},"RRC":{"level":"debug","max_size":1,"key":false,"crypto":false,"payload":false},"NAS":{"level":"debug","max_size":1,"key":false,"crypto":false,"payload":false},"S1AP":{"level":"debug","max_size":1,"key":false,"crypto":false,"payload":false},"NGAP":{"level":"error","max_size":0,"key":false,"crypto":false,"payload":false},"GTPU":{"level":"error","max_size":0,"key":false,"crypto":false,"payload":false},"X2AP":{"level":"debug","max_size":1,"key":false,"crypto":false,"payload":false},"XnAP":{"level":"error","max_size":0,"key":false,"crypto":false,"payload":false},"M2AP":{"level":"error","max_size":0,"key":false,"crypto":false,"payload":false},"LPPa":{"level":"error","max_size":0,"key":false,"crypto":false,"payload":false},"NRPPa":{"level":"error","max_size":0,"key":false,"crypto":false,"payload":false},"TRX":{"level":"error","max_size":0,"key":false,"crypto":false,"payload":false}},"bcch":false,"mib":false,"signal":false,"cch":false,"rep":false,"dci_size":false,"csi":false,"cell_meas":false,"count":8192,"rotate":250000000,"path":"/var/log/lte"},"tai":9610.092,"global_enb_id":{"plmn":"00101","enb_id_type":"macro","enb_id":107216,"enb_name":"enb1a2d0"},"cells":{"1":{"n_antenna_dl":2,"n_antenna_ul":2,"n_layer_dl":2,"n_layer_ul":1,"gain":0,"ul_disabled":false,"rf_port":0,"dl_qam":256,"ul_qam":64,"ecgi":{"plmn":"00101","eci":27447297},"n_id_cell":1,"n_rb_dl":100,"n_rb_ul":100,"dl_earfcn":38350,"ul_earfcn":38350,"band":39,"dl_freq":1890000000,"ul_freq":1890000000,"mode":"TDD","uldl_config":2,"sp_config":7,"prach_sequence_index":204,"dl_cyclic_prefix":"normal","ul_cyclic_prefix":"normal","prach_config_index":4,"prach_freq_offset":11,"delta_pucch_shift":2,"n_rb_cqi":1,"n_cs_an":0,"pucch_allocation":[{"type":"2/2a/2b","rbs":1,"n":6}],"pucch_ack_nack_start":11,"pucch_reserved_rbs":[0,0,22,0,0,0,0,22,0,0],"sr_resource_count":44,"cqi_resource_count":48,"srs_resources":{"offsets":16,"freqs":20,"total":640},"gbr":{"dl_limit":7123840,"ul_limit":1797120},"tac":1,"plmn_list":[{"plmn":"00101","reserved":false}]}},"rx_channels":[{"gain":9,"freq":1890,"port":0},{"gain":9,"freq":1890,"port":0}],"tx_channels":[{"gain":60,"freq":1890,"port":0},{"gain":60,"freq":1890,"port":0}],"rf_ports":[{"sample_rate":30720000}],"message_id":1,"time":9610.087}
{"message":"ue_get","ue_list":[],"message_id":2,"time":9613.347}
{"message":"erab_get","erab_list":[],"timestamp":9613704,"message_id":3,"time":9613.704}
{"message":"ue_get","ue_list":[],"message_id":4,"time":9616.355}
{"message":"erab_get","erab_list":[],"timestamp":9616618,"message_id":5,"time":9616.618}
{"message":"ue_get","ue_list":[],"message_id":6,"time":9619.272}
{"message":"erab_get","erab_list":[],"timestamp":9619547,"message_id":7,"time":9619.547}
{"message":"ue_get","ue_list":[],"message_id":8,"time":9622.319}
{"message":"erab_get","erab_list":[],"timestamp":9622588,"message_id":9,"time":9622.588}
{"message":"ue_get","ue_list":[],"message_id":10,"time":9625.263}
{"message":"erab_get","erab_list":[],"timestamp":9625663,"message_id":11,"time":9625.663}
{"message":"ue_get","ue_list":[],"message_id":12,"time":9628.403}
{"message":"erab_get","erab_list":[],"timestamp":9628725,"message_id":13,"time":9628.725}
{"message":"ue_get","ue_list":[],"message_id":14,"time":9631.381}
{"message":"erab_get","erab_list":[],"timestamp":9631798,"message_id":15,"time":9631.798}
{"message":"ue_get","ue_list":[],"message_id":16,"time":9634.303}
{"message":"erab_get","erab_list":[],"timestamp":9634596,"message_id":17,"time":9634.596}
{"message":"ue_get","ue_list":[],"message_id":18,"time":9637.268}
{"message":"erab_get","erab_list":[],"timestamp":9637568,"message_id":19,"time":9637.568}

    (below an UE registers with eNB)

{"message":"ue_get","ue_list":[{"enb_ue_id":8,"mme_ue_id":107,"rnti":68,"cells":[{"cell_id":1}]}],"message_id":20,"time":9640.296}
{"message":"erab_get","erab_list":[{"erab_id":5,"qci":9,"dl_total_bytes":0,"ul_total_bytes":0,"enb_ue_id":8}],"timestamp":9640831,"message_id":21,"time":9640.831}
{"message":"ue_get","ue_list":[{"enb_ue_id":8,"mme_ue_id":107,"rnti":68,"cells":[{"cell_id":1}]}],"message_id":22,"time":9643.268}
{"message":"erab_get","erab_list":[{"erab_id":5,"qci":9,"dl_total_bytes":0,"ul_total_bytes":0,"enb_ue_id":8}],"timestamp":9643555,"message_id":23,"time":9643.555}
{"message":"ue_get","ue_list":[{"enb_ue_id":8,"mme_ue_id":107,"rnti":68,"cells":[{"cell_id":1}]}],"message_id":24,"time":9646.326}
{"message":"erab_get","erab_list":[{"erab_id":5,"qci":9,"dl_total_bytes":0,"ul_total_bytes":0,"enb_ue_id":8}],"timestamp":9646850,"message_id":25,"time":9646.85}
{"message":"ue_get","ue_list":[],"message_id":26,"time":9649.287}
{"message":"erab_get","erab_list":[],"timestamp":9649646,"message_id":27,"time":9649.646}
^CTraceback (most recent call last):
  File "/home/kirr/src/wendelin/venv/xlte3.venv/bin/xamari", line 33, in <module>
    sys.exit(load_entry_point('xlte', 'console_scripts', 'xamari')())
  File "/home/kirr/src/wendelin/xlte/amari/xamari.py", line 130, in main
    return command_module.main(argv)
  File "/home/kirr/src/wendelin/xlte/amari/xlog.py", line 275, in main
    xlog(wsuri, logspecv)
  File "/home/kirr/src/wendelin/xlte/amari/xlog.py", line 108, in xlog
    xl.xlog1()
  File "/home/kirr/src/wendelin/venv/xlte3.venv/lib/python3.9/site-packages/decorator.py", line 232, in fun
    return caller(func, *(extras + args), **kw)
  File "/home/kirr/src/wendelin/venv/xlte3.venv/lib/python3.9/site-packages/golang/__init__.py", line 103, in _
    return f(*argv, **kw)
  File "/home/kirr/src/wendelin/xlte/amari/xlog.py", line 157, in xlog1
    xl._xlog1(conn)
  File "/home/kirr/src/wendelin/xlte/amari/xlog.py", line 208, in _xlog1
    time.sleep(δtsleep)
KeyboardInterrupt

amari/help.py

@@ -39,5 +39,18 @@ for example
     ws://[2a11:9ac0:d::1]:9002
 """
 
+help_jsonlog = """\
+Some commands produce logs with JSON entries. Such logs are organized with JSON
+Lines format(*) with each entry occupying one line.
+
+Logs in JSON Lines format are handy to exchange in between programs, and with
+corresponding tools, e.g. jq(+), they can be also displayed in human-readable
+form and inspected quickly.
+
+(*) https://jsonlines.org/
+(+) https://stedolan.github.io/jq/
+"""
+
 
 topic_dict['websock']   = "specifying WebSocket URI of a service",      help_websock
+topic_dict['jsonlog']   = "log with JSON entries",                      help_jsonlog

amari/xamari.py

@@ -38,7 +38,7 @@ def register_command(cmdname):
     command_module = importlib.import_module('xlte.amari.' + cmdname)
     command_dict[cmdname] = command_module
 
-for _ in ():
+for _ in ('xlog',):
     register_command(_)
 
 

amari/xlog.py

+# -*- coding: utf-8 -*-
+# Copyright (C) 2022  Nexedi SA and Contributors.
+#                     Kirill Smelkov <kirr@nexedi.com>
+#
+# This program is free software: you can Use, Study, Modify and Redistribute
+# it under the terms of the GNU General Public License version 3, or (at your
+# option) any later version, as published by the Free Software Foundation.
+#
+# You can also Link and Combine this program with other software covered by
+# the terms of any of the Free Software licenses or any of the Open Source
+# Initiative approved licenses and Convey the resulting work. Corresponding
+# source of such a combination shall include the source code for all other
+# software used.
+#
+# This program is distributed WITHOUT ANY WARRANTY; without even the implied
+# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+#
+# See COPYING file for full licensing terms.
+# See https://www.nexedi.com/licensing for rationale and options.
+"""Package xlog provides additional extra logging facilities to Amarisoft LTE stack.
+
+- use xlog and LogSpec to organize logging of information available via WebSocket access(*).
+  The information is logged in JSON Lines format. See 'xamari help xlog' for details.
+
+
+(*) for example result of stats, ue_get and erab_get queries.
+"""
+
+# TODO log file + rotate
+# TODO log loading -> DataFrame
+
+
+from xlte import amari
+
+import json
+import traceback
+from email.utils import formatdate
+from golang import func, defer
+from golang import time
+from golang.gcompat import qq
+
+import logging; log = logging.getLogger('xlte.amari.xlog')
+
+
+# LogSpec represents one specification of what to log.
+# For example stats[rf]/10s.
+class LogSpec:
+    # .query        e.g. 'stats'
+    # .optv         [] with flags to send with query
+    # .period       how often to issue the query (seconds)
+
+    DEFAULT_PERIOD = 60
+
+    def __str__(spec):
+        return "%s[%s]/%ss" % (spec.query, ','.join(spec.optv), spec.period)
+
+    # LogSpec.parse parses text into LogSpec.
+    @staticmethod
+    def parse(text):
+        def bad(reason):
+            raise ValueError("invalid logspec %s: %s" % (qq(text), reason))
+
+        optv = []
+        period = LogSpec.DEFAULT_PERIOD
+        query = text
+        _ = query.rfind('/')
+        if _ != -1:
+            tail  = query[_+1:]
+            query = query[:_]
+            if not tail.endswith('s'):
+                bad("invalid period")
+            try:
+                period = float(tail[:-1])
+            except ValueError:
+                bad("invalid period")
+
+        _ = query.find('[')
+        if _ != -1:
+            tail  = query[_:]
+            query = query[:_]
+            _ = tail.find(']')
+            if _ == -1:
+                bad("missing closing ]")
+            optv = tail[1:_-1].split(',')
+            tail = tail[_+1:]
+
+        for c in '[]/ ':
+            if c in query:
+                bad("invalid query")
+
+        spec = LogSpec()
+        spec.query  = query
+        spec.optv   = optv
+        spec.period = period
+        return spec
+
+
+# xlog queries service @wsuri periodically according to queries specified by
+# logspecv and logs the result.
+def xlog(wsuri, logspecv):
+    xl = _XLogger(wsuri, logspecv)
+
+    slogspecv = ' '.join(['%s' % _ for _ in logspecv])
+    xl.jemit("start", {"generator": "xlog %s %s" % (wsuri, slogspecv)})
+
+    while 1:
+        try:
+            xl.xlog1()
+        except Exception as ex:
+            if not isinstance(ex, amari.ConnError):
+                log.exception('xlog failure:')
+                try:
+                    xl.jemit("xlog failure", {"traceback": traceback.format_exc()})
+                except:
+                    # e.g. disk full in xl.jemit itself
+                    log.exception('xlog failure (second level):')
+
+        time.sleep(3)
+
+
+class _XLogger:
+    def __init__(xl, wsuri, logspecv):
+        xl.wsuri    = wsuri
+        xl.logspecv = logspecv
+
+    # emit saves line to the log.
+    def emit(xl, line):
+        assert '\n' not in line, line
+        print(line)
+
+    # jemit emits line corresponding to event to the log.
+    def jemit(xl, event, args_dict):
+        d = {"event": event, "time": formatdate(time.now())}  # RFC 822 / UTC
+        d.update(args_dict)
+        d = {"meta": d}
+        xl.emit(json.dumps(d))
+
+    # xlog1 performs one cycle of attach/log,log,log.../detach.
+    @func
+    def xlog1(xl):
+        # connect to the service
+        try:
+            conn = amari.connect(xl.wsuri)
+        except Exception as ex:
+            xl.jemit("service connect failure", {"reason": str(ex)})
+            if not isinstance(ex, amari.ConnError):
+                raise
+            return
+        defer(conn.close)
+
+        # emit "service attach"/"service detach"
+        srv_info = {"srv_name": conn.srv_name,
+                    "srv_type": conn.srv_type,
+                    "srv_version": conn.srv_version}
+        xl.jemit("service attach", srv_info)
+        try:
+            xl._xlog1(conn)
+        except Exception as ex:
+            d = srv_info.copy()
+            d['reason'] = str(ex)
+            xl.jemit("service detach", d)
+            if not isinstance(ex, amari.ConnError):
+                raise
+
+
+    def _xlog1(xl, conn):
+        # emit config_get after attach
+        _, cfg_raw = conn.req_('config_get', {})
+        xl.emit(cfg_raw)
+
+
+        # loop emitting requested logspecs
+        t0 = time.now()
+        tnextv = []     # [i] - next time to arm for logspecv[i] relative to t0
+        for l in xl.logspecv:
+            tnextv.append(l.period)
+
+        while 1:
+            # go through all logspecs in the order they were given
+            # pick logspec with soonest arm time
+            # execute corresponding query
+            #
+            # by going logspecs in the order they were given, we execute queries in
+            # stable order, e.g. for `stats/10s ue_get/10` - always stats first,
+            # then ue_get next.
+            #
+            # XXX linear search instead of heapq, but len(logspecv) is usually very small.
+            tmin = float('inf')
+            imin = None
+            for i, t in enumerate(tnextv):
+                if t < tmin:
+                    tmin = t
+                    imin = i
+
+            logspec = xl.logspecv[imin]
+            tnextv[imin] += logspec.period
+
+            opts = {}
+            for opt in logspec.optv:
+                opts[opt] = True
+
+            # issue queries with planned schedule
+            # TODO detect time overruns and correct schedule correspondingly
+            tnow = time.now()
+            tarm = t0 + tmin
+            δtsleep = tarm - tnow
+            if δtsleep > 0:
+                time.sleep(δtsleep)
+
+            _, resp_raw = conn.req_(logspec.query, opts)
+            xl.emit(resp_raw)
+
+
+
+# ----------------------------------------
+import sys, getopt
+
+summary = "maintain extra log for a service"
+
+def usage(out):
+    print("""\
+Usage: xamari xlog [OPTIONS] <wsuri> <logspec>+
+Maintain extra log for a service.
+
+The service is queried periodically according to logspec and results are saved
+in JSON format to a file (see 'xamari help jsonlog').
+
+<wsuri> is URI (see 'xamari help websock') of an Amarisoft-service.
+<logspec> is specification of what to log. It has the following parts:
+
+    <query>[<options>]/<period>
+
+The query specifies a message, that should be used to query service. For
+example "stats", "ue_get", "erab_get", ... Query part is mandatory.
+
+Options specifies additional flags for the query. Options part can be omitted.
+
+Period specifies periodicity of how often the service should be queried.
+Period is optional and defaults to %d seconds.
+
+Example for <logspec>+:
+
+    stats[samples,rf]/30s  ue_get[stats]  erab_get/10s  qos_flow_get
+
+
+Options:
+
+    -h  --help            show this help
+""" % LogSpec.DEFAULT_PERIOD, file=out)
+
+
+def main(argv):
+    try:
+        optv, argv = getopt.getopt(argv[1:], "h", ["help"])
+    except getopt.GetoptError as e:
+        print(e, file=sys.stderr)
+        usage(sys.stderr)
+        sys.exit(2)
+
+
+    for opt, arg in optv:
+        if opt in ("-h", "--help"):
+            usage(sys.stdout)
+            sys.exit(0)
+
+    if len(argv) < 2:
+        usage(sys.stderr)
+        sys.exit(2)
+
+    wsuri = argv[0]
+    logspecv = []
+    for arg in argv[1:]:
+        logspecv.append( LogSpec.parse(arg) )
+
+    xlog(wsuri, logspecv)

[Kirill Smelkov]

/cc @lu.xu, @tomo