hat.syslog.server.backend

  1"""Backend implementation"""
  2
  3from collections.abc import Callable
  4from pathlib import Path
  5import asyncio
  6import contextlib
  7import itertools
  8import logging
  9
 10from hat import aio
 11from hat import util
 12
 13from hat.syslog.server import common
 14from hat.syslog.server import database
 15
 16
 17mlog: logging.Logger = logging.getLogger(__name__)
 18"""Module logger"""
 19
 20register_delay: float = 0.1
 21"""Registration delay in seconds"""
 22
 23register_queue_size: int = 50
 24"""Registration queue size"""
 25
 26register_queue_treshold: int = 10
 27"""Registration queue threshold"""
 28
 29
 30async def create_backend(path: Path,
 31                         low_size: int,
 32                         high_size: int,
 33                         enable_archive: bool,
 34                         disable_journal: bool
 35                         ) -> 'Backend':
 36    """Create backend"""
 37    db = await database.create_database(path, disable_journal)
 38    try:
 39        first_id = await db.get_first_id()
 40        last_id = await db.get_last_id()
 41    except BaseException:
 42        await aio.uncancellable(db.async_close())
 43        raise
 44
 45    backend = Backend()
 46    backend._path = path
 47    backend._low_size = low_size
 48    backend._high_size = high_size
 49    backend._enable_archive = enable_archive
 50    backend._disable_journal = disable_journal
 51    backend._db = db
 52    backend._first_id = first_id
 53    backend._last_id = last_id
 54    backend._async_group = aio.Group()
 55    backend._change_cbs = util.CallbackRegistry()
 56    backend._msg_queue = aio.Queue(register_queue_size)
 57    backend._executor = aio.create_executor()
 58
 59    backend._async_group.spawn(aio.call_on_cancel, db.async_close)
 60    backend._async_group.spawn(backend._loop)
 61
 62    mlog.debug('created backend with database %s', path)
 63    return backend
 64
 65
 66class Backend(aio.Resource):
 67
 68    @property
 69    def async_group(self) -> aio.Group:
 70        """Async group"""
 71        return self._async_group
 72
 73    @property
 74    def first_id(self) -> int | None:
 75        """First entry id"""
 76        return self._first_id
 77
 78    @property
 79    def last_id(self) -> int | None:
 80        """Last entry id"""
 81        return self._last_id
 82
 83    def register_change_cb(self,
 84                           cb: Callable[[list[common.Entry]], None]
 85                           ) -> util.RegisterCallbackHandle:
 86        """Register change callback
 87
 88        Callback is called if `first_id` changes and/or `last_id` changes
 89        and/or new entries are available (passed as argument to registered
 90        callback).
 91
 92        """
 93        return self._change_cbs.register(cb)
 94
 95    async def register(self,
 96                       timestamp: float,
 97                       msg: common.Msg):
 98        """Register message
 99
100        Registration adds msg to registration queue. If queue is full, wait
101        until message can be successfully added.
102
103        When message is added to empty queue, registration delay timer is
104        started. Once delay timer expires or if number of messages in queue
105        is greater than threshold, all messages are removed from queue and
106        inserted into sqlite database.
107
108        """
109        await self._msg_queue.put((timestamp, msg))
110
111    async def query(self,
112                    filter: common.Filter
113                    ) -> list[common.Entry]:
114        """Query entries"""
115        return await self._db.query(filter)
116
117    async def _loop(self):
118        try:
119            while True:
120                msgs = await self._get_msgs()
121                await self._process_msgs(msgs)
122
123        except Exception as e:
124            mlog.warn("backend loop error: %s", e, exc_info=e)
125
126        finally:
127            self.close()
128            self._msg_queue.close()
129            mlog.debug('backend loop closed')
130
131    async def _get_msgs(self):
132        loop = asyncio.get_running_loop()
133        msgs = []
134
135        msg = await self._msg_queue.get()
136        msgs.append(msg)
137
138        start = loop.time()
139        while True:
140            while not self._msg_queue.empty():
141                msgs.append(self._msg_queue.get_nowait())
142            timeout = register_delay - (loop.time() - start)
143            if timeout <= 0:
144                break
145            if len(msgs) >= register_queue_treshold:
146                break
147            async_group = aio.Group()
148            try:
149                f = async_group.spawn(self._msg_queue.get)
150                await aio.wait_for(asyncio.shield(f), timeout)
151            except asyncio.TimeoutError:
152                break
153            finally:
154                await aio.uncancellable(async_group.async_close())
155                if not f.cancelled():
156                    msgs.append(f.result())
157
158        while not self._msg_queue.empty():
159            msgs.append(self._msg_queue.get_nowait())
160        return msgs
161
162    async def _process_msgs(self, msgs):
163        mlog.debug("registering new messages (message count: %s)...",
164                   len(msgs))
165        entries = await self._db.add_msgs(msgs)
166        if not entries:
167            return
168        entries = list(reversed(entries))
169
170        self._last_id = entries[0].id
171        if self._first_id is None:
172            self._first_id = entries[-1].id
173
174        mlog.debug("backend state changed (first_id: %s; last_id: %s)",
175                   self._first_id, self._last_id)
176        self._change_cbs.notify(entries)
177
178        if self._high_size <= 0:
179            return
180        if self._last_id - self._first_id + 1 <= self._high_size:
181            return
182
183        mlog.debug("database cleanup starting...")
184        await self._db_cleanup()
185
186    async def _db_cleanup(self):
187        first_id = self._last_id - self._low_size + 1
188        if first_id > self._last_id:
189            first_id = None
190        if first_id <= self._first_id:
191            return
192
193        if self._enable_archive:
194            mlog.debug("archiving database entries...")
195            await self._archive_db(first_id)
196
197        await self._db.delete(first_id)
198        self._first_id = first_id
199        if self._first_id is None:
200            self._last_id = None
201
202        mlog.debug("backend state changed (first_id: %s; last_id: %s)",
203                   self._first_id, self._last_id)
204        self._change_cbs.notify([])
205
206    async def _archive_db(self, first_id):
207        archive_path = await self._async_group.spawn(
208            self._executor, _ext_get_new_archive_path, self._path)
209        archive = await database.create_database(
210            archive_path, self._disable_journal)
211        try:
212            entries = await self._db.query(common.Filter(
213                last_id=first_id - 1 if first_id is not None else None))
214            await archive.add_entries(entries)
215        finally:
216            await aio.uncancellable(archive.async_close())
217
218
219def _ext_get_new_archive_path(db_path):
220    last_index = 0
221
222    for i in db_path.parent.glob(db_path.name + '.*'):
223        with contextlib.suppress(ValueError):
224            index = int(i.name.split('.')[-1])
225            if index > last_index:
226                last_index = index
227
228    for i in itertools.count(last_index + 1):
229        new_path = db_path.parent / f"{db_path.name}.{i}"
230        if new_path.exists():
231            continue
232        return new_path