Coverage for an_website/uptime/uptime.py: 67.797%

1# This program is free software: you can redistribute it and/or modify 

2# it under the terms of the GNU Affero General Public License as 

3# published by the Free Software Foundation, either version 3 of the 

4# License, or (at your option) any later version. 

5# 

6# This program is distributed in the hope that it will be useful, 

7# but WITHOUT ANY WARRANTY; without even the implied warranty of 

8# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

9# GNU Affero General Public License for more details. 

10# 

11# You should have received a copy of the GNU Affero General Public License 

12# along with this program. If not, see <https://www.gnu.org/licenses/>. 

13 

14"""The uptime page that shows the time the website is running.""" 

15 

16import logging 

17import math 

18import time 

19from typing import Final, TypedDict 

20 

21import regex 

22from elasticsearch import AsyncElasticsearch 

23from tornado.web import HTTPError, RedirectHandler 

24 

25from .. import EPOCH, EVENT_ELASTICSEARCH, NAME, UPTIME 

26from ..utils.base_request_handler import BaseRequestHandler 

27from ..utils.request_handler import APIRequestHandler, HTMLRequestHandler 

28from ..utils.utils import ModuleInfo, time_to_str 

29 

30LOGGER: Final = logging.getLogger(__name__) 

31 

32 

33class AvailabilityDict(TypedDict): # noqa: D101 

34 # pylint: disable=missing-class-docstring 

35 up: int 

36 down: int 

37 total: int 

38 percentage: None | float 

39 

40 

41def get_module_info() -> ModuleInfo: 

42 """Create and return the ModuleInfo for this module.""" 

43 return ModuleInfo( 

44 handlers=( 

45 (r"/betriebszeit", UptimeHandler), 

46 (r"/betriebszeit/verfuegbarkeit.svg", AvailabilityChartHandler), 

47 (r"/api/betriebszeit", UptimeAPIHandler), 

48 (r"/api/uptime/*", RedirectHandler, {"url": "/api/betriebszeit"}), 

49 ), 

50 name="Betriebszeit", 

51 description="Die Dauer, die die Webseite am Stück in Betrieb ist", 

52 path="/betriebszeit", 

53 aliases=("/uptime",), 

54 keywords=("Uptime", "Betriebszeit", "Zeit"), 

55 ) 

56 

57 

58class EsAvailabilityKwargs(TypedDict): 

59 """Subset of kwargs to Elasticsearch search.""" 

60 

61 index: str 

62 query: dict[str, dict[str, list[dict[str, dict[str, dict[str, str]]]]]] 

63 size: int 

64 aggs: dict[str, dict[str, dict[str, str]]] 

65 

66 

67ES_AVAILABILITY_KWARGS: Final[EsAvailabilityKwargs] = { 

68 "index": "heartbeat-*,synthetics-*", 

69 "query": { 

70 "bool": { 

71 "filter": [ 

72 {"range": {"@timestamp": {"gte": "now-1M"}}}, 

73 {"term": {"monitor.type": {"value": "http"}}}, 

74 { 

75 "term": { 

76 "service.name": {"value": NAME.removesuffix("-dev")} 

77 } 

78 }, 

79 ] 

80 } 

81 }, 

82 "size": 0, 

83 "aggs": { 

84 "up": {"sum": {"field": "summary.up"}}, 

85 "down": {"sum": {"field": "summary.down"}}, 

86 }, 

87} 

88 

89 

90async def get_availability_data( 

91 elasticsearch: AsyncElasticsearch, 

92) -> None | tuple[int, int]: # (up, down) 

93 """Get the availability data.""" 

94 try: 

95 data = await elasticsearch.search(**ES_AVAILABILITY_KWARGS) 

96 except Exception: # pylint: disable=broad-exception-caught 

97 LOGGER.exception("Getting availability data from Elasticsearch failed.") 

98 return None 

99 data.setdefault("aggregations", {"up": {"value": 0}, "down": {"value": 0}}) 

100 return ( 

101 int(data["aggregations"]["up"]["value"]), 

102 int(data["aggregations"]["down"]["value"]), 

103 ) 

104 

105 

106def get_availability_dict(up: int, down: int) -> AvailabilityDict: 

107 """Get the availability data as a dict.""" 

108 return { 

109 "up": up, 

110 "down": down, 

111 "total": up + down, 

112 "percentage": 100 * up / (up + down) if up + down else None, 

113 } 

114 

115 

116AVAILABILITY_CHART: Final[str] = regex.sub( 

117 r"\s+", 

118 " ", 

119 """ 

120<svg height="20" 

121 width="20" 

122 viewBox="0 0 20 20" 

123 xmlns="http://www.w3.org/2000/svg" 

124><circle r="10" cx="10" cy="10" fill="red" /> 

125 <circle r="5" cx="10" cy="10" fill="transparent" 

126 stroke="green" 

127 stroke-width="10" 

128 stroke-dasharray="%2.2f 31.4159" 

129 transform="rotate(-90) translate(-20)" /> 

130</svg> 

131""".strip(), 

132) 

133 

134 

135class UptimeHandler(HTMLRequestHandler): 

136 """The request handler for the uptime page.""" 

137 

138 COMPUTE_ETAG = False 

139 POSSIBLE_CONTENT_TYPES = ( 

140 *HTMLRequestHandler.POSSIBLE_CONTENT_TYPES, 

141 *APIRequestHandler.POSSIBLE_CONTENT_TYPES, 

142 ) 

143 

144 async def get(self, *, head: bool = False) -> None: 

145 """Handle GET requests.""" 

146 self.set_header("Cache-Control", "no-cache") 

147 if head: 

148 return 

149 if self.content_type in APIRequestHandler.POSSIBLE_CONTENT_TYPES: 

150 return await self.finish(await self.get_uptime_data()) 

151 await self.render("pages/uptime.html", **(await self.get_uptime_data())) 

152 

153 async def get_uptime_data( 

154 self, 

155 ) -> dict[str, str | float | AvailabilityDict]: 

156 """Get uptime data.""" 

157 availability_data = ( 

158 await get_availability_data(self.elasticsearch) 

159 if EVENT_ELASTICSEARCH.is_set() 

160 else None 

161 ) or (0, 0) 

162 return { 

163 "uptime": (uptime := UPTIME.get()), 

164 "uptime_str": time_to_str(uptime), 

165 "start_time": time.time() - uptime - EPOCH, 

166 "availability": get_availability_dict(*availability_data), 

167 } 

168 

169 

170class AvailabilityChartHandler(BaseRequestHandler): 

171 """The request handler for the availability chart.""" 

172 

173 POSSIBLE_CONTENT_TYPES = ("image/svg+xml",) 

174 COMPUTE_ETAG = False 

175 

176 async def get(self, *, head: bool = False) -> None: 

177 """Handle GET requests.""" 

178 if not (availability := self.get_argument("a", None)): 

179 if not EVENT_ELASTICSEARCH.is_set(): 

180 raise HTTPError(503) 

181 availability_data = await get_availability_data(self.elasticsearch) 

182 if not availability_data: 

183 raise HTTPError(503) 

184 self.redirect( 

185 self.fix_url( 

186 self.request.full_url(), 

187 query_args={ 

188 "a": int( 

189 ( 

190 get_availability_dict(*availability_data)[ 

191 "percentage" 

192 ] 

193 or 0 

194 ) 

195 * 100 

196 ) 

197 / (100 * 100), 

198 }, 

199 ), 

200 permanent=False, 

201 ) 

202 return 

203 self.set_header( 

204 "Cache-Control", f"public, min-fresh={60 * 60 * 24 * 14}, immutable" 

205 ) 

206 if head: 

207 return 

208 await self.finish( 

209 AVAILABILITY_CHART % (math.pi * 10 * float(availability)) 

210 ) 

211 

212 

213class UptimeAPIHandler(APIRequestHandler, UptimeHandler): 

214 """The request handler for the uptime API.""" 

215 

216 POSSIBLE_CONTENT_TYPES = APIRequestHandler.POSSIBLE_CONTENT_TYPES