|
2215
|
1 # dev copy
|
|
|
2 import asyncio
|
|
|
3 import logging
|
|
|
4 import time
|
|
|
5 import traceback
|
|
|
6 from typing import Any, Awaitable, Callable, Union
|
|
|
7
|
|
|
8 from prometheus_client import Gauge, Summary
|
|
|
9
|
|
|
10 log = logging.getLogger()
|
|
|
11
|
|
|
12
|
|
|
13 # throw this away (when net_routes is rewritten)
|
|
|
14 def loop_forever_sync(func, sleep_period, up_metric):
|
|
|
15 first_run = True
|
|
|
16 while True:
|
|
|
17 try:
|
|
|
18 func(first_run)
|
|
|
19 up_metric.set(1)
|
|
|
20 first_run = False
|
|
|
21 except Exception as ex:
|
|
|
22 log.error(ex)
|
|
|
23 traceback.print_exc()
|
|
|
24 up_metric.set(0)
|
|
|
25 time.sleep(sleep_period)
|
|
|
26
|
|
|
27
|
|
|
28 _created = []
|
|
|
29
|
|
|
30 # todo: some tricky typing
|
|
|
31 F_RET = Any
|
|
|
32 F_KWARGS = Any
|
|
|
33
|
|
|
34 UserAsyncFunc = Callable[..., # always called with at least kwarg first_run=bool
|
|
|
35 Awaitable[F_RET]]
|
|
|
36 UserSyncFunc = Callable[..., # see above
|
|
|
37 F_RET]
|
|
|
38 UserFunc = Union[UserAsyncFunc, UserSyncFunc]
|
|
|
39
|
|
|
40
|
|
|
41 class Loop:
|
|
|
42
|
|
|
43 def __init__(
|
|
|
44 self,
|
|
|
45 func: UserFunc,
|
|
|
46 sleep_period: float,
|
|
|
47 metric_prefix: str,
|
|
|
48 ):
|
|
|
49 self.func = func
|
|
|
50 self.sleep_period = sleep_period
|
|
|
51
|
|
|
52 self.up_metric = Gauge(f'{metric_prefix}_up', 'not erroring')
|
|
|
53 self.call_metric = Summary(f'{metric_prefix}_calls', 'calls')
|
|
|
54 self.lastSuccessRun = 0
|
|
|
55 self.everSucceeded = False
|
|
|
56
|
|
|
57 async def call_func(self, first_run: bool, call_kwargs={}) -> F_RET:
|
|
|
58 with self.call_metric.time():
|
|
|
59 if asyncio.iscoroutinefunction(self.func):
|
|
|
60 ret = await self.func(first_run=first_run, **call_kwargs)
|
|
|
61 else:
|
|
|
62 ret = self.func(first_run=first_run, **call_kwargs)
|
|
|
63 return ret
|
|
|
64
|
|
|
65 async def _run(self):
|
|
|
66 self.first_run = True
|
|
|
67 while True:
|
|
|
68 await self._runOne()
|
|
|
69 await asyncio.sleep(self.sleep_period)
|
|
|
70
|
|
|
71 async def runNow(self, **kwargs: F_KWARGS) -> F_RET:
|
|
|
72 """unlike background runs, you get to pass more args and get your func's
|
|
|
73 return value.
|
|
|
74 """
|
|
|
75 return await self._runOne(call_kwargs=kwargs)
|
|
|
76
|
|
|
77 async def _runOne(self, call_kwargs={}):
|
|
|
78 try:
|
|
|
79 result = await self.call_func(self.first_run, call_kwargs)
|
|
|
80 self.lastSuccessRun = time.time()
|
|
|
81 self.up_metric.set(1)
|
|
|
82 self.everSucceeded = True
|
|
|
83 self.first_run = False
|
|
|
84 except Exception as ex:
|
|
|
85 log.error(ex)
|
|
|
86 traceback.print_exc()
|
|
|
87 self.up_metric.set(0)
|
|
|
88 result = None
|
|
|
89 # todo: something that reveals error ratio
|
|
|
90 return result
|
|
|
91
|
|
|
92
|
|
|
93 def loop_forever(
|
|
|
94 func: UserFunc,
|
|
|
95 sleep_period: float,
|
|
|
96 metric_prefix='background_loop',
|
|
|
97 up_metric=None,
|
|
|
98 call_metric=None,
|
|
|
99 ):
|
|
|
100 """
|
|
|
101 sleep_period is the sleep time after however long func takes to run
|
|
|
102 """
|
|
|
103 if up_metric is not None or call_metric is not None:
|
|
|
104 raise NotImplementedError('remove old-style metrics')
|
|
|
105
|
|
|
106 loop = Loop(func, sleep_period, metric_prefix)
|
|
|
107 _created.append(asyncio.create_task(loop._run()))
|
|
|
108 return loop
|
