Mercurial > code > home > repos > background_loop

comparison background_loop.py @ 16:a1ed58edfccc

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
does its own task; returns Loop that you can query for status and call runNow on; rewrote metrics naming
author drewp@bigasterisk.com
date Sun, 24 Jul 2022 01:01:13 -0700
parents 263ccb9e00df
children
comparison
equal deleted inserted replaced
15:ba8e238a4e53 16:a1ed58edfccc
1 # dev copy 1 # dev copy
2 import asyncio 2 import asyncio
3 import logging 3 import logging
4 import time 4 import time
5 import traceback 5 import traceback
6 from typing import Awaitable, Callable 6 from typing import Any, Awaitable, Callable, Union
7 7
8 from prometheus_client import Gauge, Summary 8 from prometheus_client import Gauge, Summary
9 9
10 log = logging.getLogger() 10 log = logging.getLogger()
11 11
23 traceback.print_exc() 23 traceback.print_exc()
24 up_metric.set(0) 24 up_metric.set(0)
25 time.sleep(sleep_period) 25 time.sleep(sleep_period)
26 26
27 27
28 async def loop_forever( 28 _created = []
29 func: Callable[[bool], Awaitable[None]], 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,
30 sleep_period: float, 95 sleep_period: float,
31 up_metric: Gauge, 96 metric_prefix='background_loop',
32 call_metric: Summary, 97 up_metric=None,
98 call_metric=None,
33 ): 99 ):
34 """ 100 """
35 sleep_period is the sleep time after however long func takes to run 101 sleep_period is the sleep time after however long func takes to run
36 """ 102 """
37 @call_metric.time() 103 if up_metric is not None or call_metric is not None:
38 async def call_func(first_run): 104 raise NotImplementedError('remove old-style metrics')
39 if asyncio.iscoroutinefunction(func):
40 await func(first_run)
41 else:
42 func(first_run)
43 105
44 first_run = True 106 loop = Loop(func, sleep_period, metric_prefix)
45 while True: 107 _created.append(asyncio.create_task(loop._run()))
46 try: 108 return loop
47 await call_func(first_run)
48 up_metric.set(1)
49 first_run = False
50 except Exception as ex:
51 log.error(ex)
52 traceback.print_exc()
53 up_metric.set(0)
54 # todo: something that reveals error ratio
55 await asyncio.sleep(sleep_period)