hat.event.server.runner

  1import asyncio
  2import contextlib
  3import logging
  4import types
  5import typing
  6
  7from hat import aio
  8from hat import json
  9from hat.drivers import tcp
 10import hat.monitor.component
 11
 12from hat.event import common
 13from hat.event.server.engine import create_engine
 14from hat.event.server.eventer_client import create_eventer_client
 15from hat.event.server.eventer_server import (create_eventer_server,
 16                                             EventerServer)
 17
 18
 19mlog: logging.Logger = logging.getLogger(__name__)
 20"""Module logger"""
 21
 22
 23class EventerServerData(typing.NamedTuple):
 24    server_id: common.ServerId
 25    addr: tcp.Address
 26
 27
 28class MainRunner(aio.Resource):
 29
 30    def __init__(self, conf: json.Data):
 31        self._conf = conf
 32        self._loop = asyncio.get_running_loop()
 33        self._async_group = aio.Group()
 34        self._backend = None
 35        self._eventer_server = None
 36        self._monitor_component = None
 37        self._eventer_client_runner = None
 38        self._engine_runner = None
 39
 40        self.async_group.spawn(self._run)
 41
 42    @property
 43    def async_group(self) -> aio.Group:
 44        return self._async_group
 45
 46    async def _run(self):
 47        try:
 48            await self._start()
 49            await self._loop.create_future()
 50
 51        except Exception as e:
 52            mlog.error("main runner loop error: %s", e, exc_info=e)
 53
 54        finally:
 55            self.close()
 56            await aio.uncancellable(self._stop())
 57
 58    async def _start(self):
 59        backend_conf = self._conf['backend']
 60        backend_info = common.import_backend_info(backend_conf['module'])
 61        self._backend = await aio.call(backend_info.create, backend_conf,
 62                                       self._on_backend_registered_events,
 63                                       self._on_backend_flushed_events)
 64        _bind_resource(self.async_group, self._backend)
 65
 66        self._eventer_server = await create_eventer_server(
 67            addr=tcp.Address(self._conf['eventer_server']['host'],
 68                             self._conf['eventer_server']['port']),
 69            backend=self._backend,
 70            server_id=self._conf['server_id'],
 71            server_token=self._conf.get('server_token'))
 72        _bind_resource(self.async_group, self._eventer_server)
 73
 74        if 'monitor_component' in self._conf:
 75            self._monitor_component = await hat.monitor.component.connect(
 76                addr=tcp.Address(self._conf['monitor_component']['host'],
 77                                 self._conf['monitor_component']['port']),
 78                name=self._conf['monitor_component']['name'],
 79                group=self._conf['monitor_component']['group'],
 80                runner_cb=self._create_monitor_runner,
 81                data={'server_id': self._conf['server_id'],
 82                      'eventer_server': self._conf['eventer_server'],
 83                      'server_token': self._conf.get('server_token')},
 84                state_cb=self._on_monitor_state)
 85            _bind_resource(self.async_group, self._monitor_component)
 86
 87            self._eventer_client_runner = EventerClientRunner(
 88                conf=self._conf,
 89                backend=self._backend,
 90                synced_cb=self._on_eventer_client_synced)
 91            _bind_resource(self.async_group, self._eventer_client_runner)
 92
 93            await self._eventer_client_runner.set_monitor_state(
 94                self._monitor_component.state)
 95
 96            await self._monitor_component.set_ready(True)
 97
 98        else:
 99            self._engine_runner = EngineRunner(
100                conf=self._conf,
101                backend=self._backend,
102                eventer_server=self._eventer_server)
103            _bind_resource(self.async_group, self._engine_runner)
104
105    async def _stop(self):
106        if self._engine_runner and not self._monitor_component:
107            await self._engine_runner.async_close()
108
109        if self._eventer_client_runner:
110            await self._eventer_client_runner.async_close()
111
112        if self._monitor_component:
113            await self._monitor_component.async_close()
114
115        if self._eventer_server:
116            with contextlib.suppress(Exception):
117                await self._backend.flush()
118
119            await self._eventer_server.async_close()
120
121        if self._backend:
122            await self._backend.async_close()
123
124    async def _create_monitor_runner(self, monitor_component):
125        self._engine_runner = EngineRunner(conf=self._conf,
126                                           backend=self._backend,
127                                           eventer_server=self._eventer_server)
128        return self._engine_runner
129
130    async def _on_backend_registered_events(self, events):
131        if not self._eventer_server:
132            return
133
134        await self._eventer_server.notify_events(events, False)
135
136    async def _on_backend_flushed_events(self, events):
137        if not self._eventer_server:
138            return
139
140        await self._eventer_server.notify_events(events, True)
141
142    async def _on_monitor_state(self, monitor_component, state):
143        if not self._eventer_client_runner:
144            return
145
146        await self._eventer_client_runner.set_monitor_state(state)
147
148    async def _on_eventer_client_synced(self, server_id, synced, counter):
149        if not self._engine_runner:
150            return
151
152        await self._engine_runner.set_synced(server_id, synced, counter)
153
154
155class EventerClientRunner(aio.Resource):
156
157    def __init__(self,
158                 conf: json.Data,
159                 backend: common.Backend,
160                 synced_cb: aio.AsyncCallable[[common.ServerId, bool, int],
161                                              None],
162                 reconnect_delay: float = 5):
163        self._conf = conf
164        self._backend = backend
165        self._synced_cb = synced_cb
166        self._reconnect_delay = reconnect_delay
167        self._async_group = aio.Group()
168        self._client_subgroups = {}
169
170    @property
171    def async_group(self) -> aio.Group:
172        return self._async_group
173
174    async def set_monitor_state(self, state: hat.monitor.component.State):
175        valid_server_data = set(_get_eventer_server_data(
176            group=self._conf['monitor_component']['group'],
177            server_token=self._conf.get('server_token'),
178            state=state))
179
180        for server_data in list(self._client_subgroups.keys()):
181            if server_data in valid_server_data:
182                continue
183
184            subgroup = self._client_subgroups.pop(server_data)
185            subgroup.close()
186
187        for server_data in valid_server_data:
188            subgroup = self._client_subgroups.get(server_data)
189            if subgroup and subgroup.is_open:
190                continue
191
192            subgroup = self.async_group.create_subgroup()
193            subgroup.spawn(self._client_loop, subgroup, server_data)
194            self._client_subgroups[server_data] = subgroup
195
196    async def _client_loop(self, async_group, server_data):
197        try:
198            while True:
199                try:
200                    eventer_client = await create_eventer_client(
201                        addr=server_data.addr,
202                        client_name=self._conf['monitor_component']['name'],
203                        server_id=server_data.server_id,
204                        backend=self._backend,
205                        client_token=self._conf.get('server_token'),
206                        synced_cb=self._on_synced)
207
208                except Exception:
209                    await asyncio.sleep(self._reconnect_delay)
210                    continue
211
212                try:
213                    await aio.call(self._synced_cb, server_data.server_id,
214                                   False, 0)
215
216                    await eventer_client.wait_closing()
217
218                finally:
219                    await aio.uncancellable(eventer_client.async_close())
220
221        except Exception as e:
222            mlog.error("eventer client runner loop error: %s", e, exc_info=e)
223            self.close()
224
225        finally:
226            async_group.close()
227
228    async def _on_synced(self, server_id, counter):
229        await aio.call(self._synced_cb, server_id, True, counter)
230
231
232class EngineRunner(aio.Resource):
233
234    def __init__(self,
235                 conf: json.Data,
236                 backend: common.Backend,
237                 eventer_server: EventerServer):
238        self._conf = conf
239        self._backend = backend
240        self._eventer_server = eventer_server
241        self._async_group = aio.Group()
242        self._engine = None
243        self._restart = asyncio.Event()
244
245        self.async_group.spawn(self._run)
246
247    @property
248    def async_group(self) -> aio.Group:
249        return self._async_group
250
251    async def set_synced(self,
252                         server_id: common.ServerId,
253                         synced: bool,
254                         counter: int):
255        if self._engine and self._engine.is_open:
256            source = common.Source(type=common.SourceType.SERVER, id=0)
257            event = common.RegisterEvent(
258                type=('event', str(self._conf['server_id']), 'synced',
259                      str(server_id)),
260                source_timestamp=None,
261                payload=common.EventPayloadJson(synced))
262
263            await self._engine.register(source, [event])
264
265        if synced and counter and self._conf.get('synced_restart_engine'):
266            self._restart.set()
267
268    async def _run(self):
269        try:
270            while True:
271                self._restart.clear()
272
273                self._engine = await create_engine(
274                    backend=self._backend,
275                    module_confs=self._conf['modules'],
276                    server_id=self._conf['server_id'])
277                await self._eventer_server.set_engine(self._engine)
278
279                async with self._async_group.create_subgroup() as subgroup:
280                    await asyncio.wait(
281                        [subgroup.spawn(self._engine.wait_closing),
282                         subgroup.spawn(self._restart.wait)],
283                        return_when=asyncio.FIRST_COMPLETED)
284
285                if not self._engine.is_open:
286                    break
287
288                await self._close()
289
290        except Exception as e:
291            mlog.error("engine runner loop error: %s", e, exc_info=e)
292
293        finally:
294            self.close()
295            await aio.uncancellable(self._close())
296
297    async def _close(self):
298        if self._engine:
299            await self._engine.async_close()
300
301        await self._eventer_server.set_engine(None)
302
303
304def _bind_resource(async_group, resource):
305    async_group.spawn(aio.call_on_done, resource.wait_closing(),
306                      async_group.close)
307
308
309def _get_eventer_server_data(group, server_token, state):
310    for info in state.components:
311        if info == state.info or info.group != group:
312            continue
313
314        server_id = json.get(info.data, 'server_id')
315        host = json.get(info.data, ['eventer_server', 'host'])
316        port = json.get(info.data, ['eventer_server', 'port'])
317        token = json.get(info.data, 'server_token')
318        if (not isinstance(server_id, int) or
319                not isinstance(host, str) or
320                not isinstance(port, int) or
321                not isinstance(token, (str, types.NoneType))):
322            continue
323
324        if server_token is not None and token != server_token:
325            continue
326
327        yield EventerServerData(server_id=server_id,
328                                addr=tcp.Address(host, port))