Commit 62fdae15 authored by Kirill Smelkov's avatar Kirill Smelkov

amari.xlog: Unify start with sync

Let's use "sync(reason=start)" instead of dedicated "start" event for
uniformity. Periodic syncs are now "sync(reason=periodic)" and after
logrotation support there will be also "pre-logrotate" and
"post-logrotate" reasons. Emit "sync(reason=stop)" at xlog shutdown for
uniformity and to make it more clear from looking at just enb.xlog about
what is xlog state at the end.

Stop requiring "start" to be present in the header - we will soon rework
xlog reader to look around for nearby sync automatically so that reading
could be started from any position in the stream.
parent f5626f73
...@@ -154,7 +154,7 @@ def _read(logm): ...@@ -154,7 +154,7 @@ def _read(logm):
x = LogError.EOF # represent EOF as LogError x = LogError.EOF # represent EOF as LogError
# ignore sync events # ignore sync events
if isinstance(x, xlog.Event) and x.event == "sync": if isinstance(x, xlog.SyncEvent):
continue continue
# handle messages that update current Measurement # handle messages that update current Measurement
......
...@@ -40,12 +40,12 @@ ...@@ -40,12 +40,12 @@
# Queries are specific to monitored LTE service. # Queries are specific to monitored LTE service.
# Events are specific to xlog itself and can be as follows: # Events are specific to xlog itself and can be as follows:
# #
# - "start" when xlog starts
# - "service attach" when xlog successfully connects to monitored LTE service # - "service attach" when xlog successfully connects to monitored LTE service
# - "service detach" when xlog disconnects from monitored LTE service # - "service detach" when xlog disconnects from monitored LTE service
# - "service connect failure" when xlog tries to connect to monitored LTE service # - "service connect failure" when xlog tries to connect to monitored LTE service
# with unsuccessful result. # with unsuccessful result.
# - "sync" emitted periodically with current state of # - "sync" emitted periodically and when xlogs starts,
# stops (TODO and rotate logs). Comes with current state of
# connection to LTE service and xlog setup # connection to LTE service and xlog setup
# - "xlog failure" on internal xlog error # - "xlog failure" on internal xlog error
...@@ -128,6 +128,7 @@ class LogSpec: ...@@ -128,6 +128,7 @@ class LogSpec:
# xlog queries service @wsuri periodically according to queries specified by # xlog queries service @wsuri periodically according to queries specified by
# logspecv and logs the result. # logspecv and logs the result.
@func
def xlog(ctx, wsuri, logspecv): def xlog(ctx, wsuri, logspecv):
# make sure we always have meta.sync - either the caller specifies it # make sure we always have meta.sync - either the caller specifies it
# explicitly, or we add it automatically to come first with default # explicitly, or we add it automatically to come first with default
...@@ -155,8 +156,11 @@ def xlog(ctx, wsuri, logspecv): ...@@ -155,8 +156,11 @@ def xlog(ctx, wsuri, logspecv):
xl = _XLogger(wsuri, logspecv, lsync.period) xl = _XLogger(wsuri, logspecv, lsync.period)
slogspecv = ' '.join(['%s' % _ for _ in logspecv]) # emit sync at start/stop
xl.jemit("start", {"generator": "xlog %s %s" % (wsuri, slogspecv)}) xl.jemit_sync("detached", "start", {})
def _():
xl.jemit_sync("detached", "stop", {})
defer(_)
while 1: while 1:
try: try:
...@@ -203,9 +207,10 @@ class _XLogger: ...@@ -203,9 +207,10 @@ class _XLogger:
# jemit_sync emits line with sync event to the log. # jemit_sync emits line with sync event to the log.
# TODO logrotate at this point # TODO logrotate at this point
def jemit_sync(xl, state, args_dict): def jemit_sync(xl, state, reason, args_dict):
tnow = time.now() tnow = time.now()
d = {"state": state, d = {"state": state,
"reason": reason,
"generator": "xlog %s %s" % (xl.wsuri, ' '.join(['%s' % _ for _ in xl.logspecv]))} "generator": "xlog %s %s" % (xl.wsuri, ' '.join(['%s' % _ for _ in xl.logspecv]))}
d.update(args_dict) d.update(args_dict)
xl.jemit("sync", d) xl.jemit("sync", d)
...@@ -217,7 +222,7 @@ class _XLogger: ...@@ -217,7 +222,7 @@ class _XLogger:
# emit sync periodically even in detached state # emit sync periodically even in detached state
# this is useful to still know e.g. intended logspec if the service is stopped for a long time # this is useful to still know e.g. intended logspec if the service is stopped for a long time
if time.now() - xl.tsync >= xl.δt_sync: if time.now() - xl.tsync >= xl.δt_sync:
xl.jemit_sync("detached", {}) xl.jemit_sync("detached", "periodic", {})
# connect to the service # connect to the service
try: try:
...@@ -314,7 +319,7 @@ class _XLogger: ...@@ -314,7 +319,7 @@ class _XLogger:
raise ctx.err() raise ctx.err()
if logspec.query == 'meta.sync': if logspec.query == 'meta.sync':
xl.jemit_sync("attached", srv_info) xl.jemit_sync("attached", "periodic", srv_info)
else: else:
resp_raw = req_(ctx, logspec.query, opts) resp_raw = req_(ctx, logspec.query, opts)
xl.emit(resp_raw) xl.emit(resp_raw)
...@@ -429,6 +434,13 @@ class Message(xdict): ...@@ -429,6 +434,13 @@ class Message(xdict):
# .timestamp seconds since epoch # .timestamp seconds since epoch
pass pass
# SyncEvent specializes Event and represents "sync" event in xlog.
class SyncEvent(Event):
# .state
# .reason
# .generator
pass
# Reader(r) creates new reader that will read xlog data from r. # Reader(r) creates new reader that will read xlog data from r.
@func(Reader) @func(Reader)
...@@ -436,23 +448,6 @@ def __init__(xr, r): ...@@ -436,23 +448,6 @@ def __init__(xr, r):
xr._r = r xr._r = r
xr._lineno = 0 xr._lineno = 0
# parse header
try:
head = xr._jread1()
if head is None:
raise xr._err("header: unexpected EOF")
meta = head.get1("meta", dict)
ev0, t0 = xr._parse_metahead(meta)
if ev0 != "start":
raise xr._err("header: starts with meta.event=%s ; expected `start`" % ev0)
gen = meta.get1("generator", str)
# TODO parse generator -> ._xlogspecv
except:
xr._r.close()
raise
# close release resources associated with the Reader. # close release resources associated with the Reader.
@func(Reader) @func(Reader)
def close(xr): def close(xr):
...@@ -469,6 +464,16 @@ def read(xr): # -> Event|Message|None ...@@ -469,6 +464,16 @@ def read(xr): # -> Event|Message|None
x.__class__ = Event x.__class__ = Event
meta = x.get1("meta", dict) meta = x.get1("meta", dict)
x.event, x.timestamp = xr._parse_metahead(meta) x.event, x.timestamp = xr._parse_metahead(meta)
if x.event in {"sync", "start"}: # for backward compatibility with old logs meta:start
x.__class__ = SyncEvent # is reported to users as sync(start) event
x.generator = meta.get1("generator", str)
if x.event == "start":
x.state = "detached"
x.reason = "start"
else:
x.state = meta.get1("state", str)
x.reason = meta.get1("reason", str)
# TODO parse generator -> .logspecv
return x return x
if "message" in x: if "message" in x:
......
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# Copyright (C) 2022 Nexedi SA and Contributors. # Copyright (C) 2022-2023 Nexedi SA and Contributors.
# Kirill Smelkov <kirr@nexedi.com> # Kirill Smelkov <kirr@nexedi.com>
# #
# This program is free software: you can Use, Study, Modify and Redistribute # 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 # it under the terms of the GNU General Public License version 3, or (at your
...@@ -38,8 +38,14 @@ zzzqqqrrrr ...@@ -38,8 +38,14 @@ zzzqqqrrrr
xr = xlog.Reader(io.BytesIO(data)) xr = xlog.Reader(io.BytesIO(data))
defer(xr.close) defer(xr.close)
# TODO check xr.tstart == 0.01 # :1
# TODO check xr.xlogspecv == ue_get[]/3.0s erab_get[]/3.0s _ = xr.read()
assert type(_) is xlog.SyncEvent
assert _.event == "start"
assert _.timestamp == 0.01
assert _ == {"meta": {"event": "start",
"time": 0.01,
"generator": "xlog ws://localhost:9001 ue_get[]/3.0s erab_get[]/3.0s"}}
# :2 # :2
_ = xr.read() _ = xr.read()
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment