Coverage for chalicelib/core/alerts/alerts_processor.py: 0%

1import logging 

2 

3from pydantic_core._pydantic_core import ValidationError 

4 

5import schemas 

6from chalicelib.utils import pg_client, ch_client, exp_ch_helper 

7from chalicelib.utils.TimeUTC import TimeUTC 

8from chalicelib.core.alerts import alerts, alerts_listener 

9from chalicelib.core.alerts.modules import alert_helpers 

10from chalicelib.core.sessions import sessions_ch as sessions 

11 

12logger = logging.getLogger(__name__) 

13 

14LeftToDb = { 

15 schemas.AlertColumn.PERFORMANCE__DOM_CONTENT_LOADED__AVERAGE: { 

16 "table": lambda timestamp: f"{exp_ch_helper.get_main_events_table(timestamp)} AS pages", 

17 "formula": "COALESCE(AVG(NULLIF(CAST(`$properties`.dom_content_loaded_event_time AS Float32) ,0)),0)", 

18 "eventType": "LOCATION", 

19 "condition": "`$auto_captured`" 

20 }, 

21 schemas.AlertColumn.PERFORMANCE__FIRST_MEANINGFUL_PAINT__AVERAGE: { 

22 "table": lambda timestamp: f"{exp_ch_helper.get_main_events_table(timestamp)} AS pages", 

23 "formula": "COALESCE(AVG(NULLIF(CAST(`$properties`.first_contentful_paint_time AS Float32),0)),0)", 

24 "eventType": "LOCATION", 

25"condition": "`$auto_captured`" 

26 }, 

27 schemas.AlertColumn.PERFORMANCE__PAGE_LOAD_TIME__AVERAGE: { 

28 "table": lambda timestamp: f"{exp_ch_helper.get_main_events_table(timestamp)} AS pages", 

29 "formula": "AVG(NULLIF(CAST(`$properties`.load_event_time AS Float32) ,0))", 

30 "eventType": "LOCATION", 

31"condition": "`$auto_captured`" 

32 }, 

33 schemas.AlertColumn.PERFORMANCE__DOM_BUILD_TIME__AVERAGE: { 

34 "table": lambda timestamp: f"{exp_ch_helper.get_main_events_table(timestamp)} AS pages", 

35 "formula": "AVG(NULLIF(CAST(`$properties`.dom_building_time AS Float32),0))", 

36 "eventType": "LOCATION", 

37"condition": "`$auto_captured`" 

38 }, 

39 schemas.AlertColumn.PERFORMANCE__SPEED_INDEX__AVERAGE: { 

40 "table": lambda timestamp: f"{exp_ch_helper.get_main_events_table(timestamp)} AS pages", 

41 "formula": "AVG(NULLIF(speed_index,0))", 

42 "eventType": "LOCATION", 

43"condition": "`$auto_captured`" 

44 }, 

45 schemas.AlertColumn.PERFORMANCE__PAGE_RESPONSE_TIME__AVERAGE: { 

46 "table": lambda timestamp: f"{exp_ch_helper.get_main_events_table(timestamp)} AS pages", 

47 "formula": "AVG(NULLIF(CAST(`$properties`.response_time AS Float32),0))", 

48 "eventType": "LOCATION", 

49"condition": "`$auto_captured`" 

50 }, 

51 schemas.AlertColumn.PERFORMANCE__TTFB__AVERAGE: { 

52 "table": lambda timestamp: f"{exp_ch_helper.get_main_events_table(timestamp)} AS pages", 

53 "formula": "AVG(NULLIF(CAST(`$properties`.first_contentful_paint_time AS Float32),0))", 

54 "eventType": "LOCATION", 

55"condition": "`$auto_captured`" 

56 }, 

57 schemas.AlertColumn.PERFORMANCE__TIME_TO_RENDER__AVERAGE: { 

58 "table": lambda timestamp: f"{exp_ch_helper.get_main_events_table(timestamp)} AS pages", 

59 "formula": "AVG(NULLIF(CAST(`$properties`.visually_complete AS Float32),0))", 

60 "eventType": "LOCATION", 

61"condition": "`$auto_captured`" 

62 }, 

63 schemas.AlertColumn.PERFORMANCE__CRASHES__COUNT: { 

64 "table": lambda timestamp: f"{exp_ch_helper.get_main_sessions_table(timestamp)} AS sessions", 

65 "formula": "COUNT(DISTINCT session_id)", 

66 "condition": "duration>0 AND errors_count>0" 

67 }, 

68 schemas.AlertColumn.ERRORS__JAVASCRIPT__COUNT: { 

69 "table": lambda timestamp: f"{exp_ch_helper.get_main_events_table(timestamp)} AS errors", 

70 "eventType": "ERROR", 

71 "formula": "COUNT(DISTINCT session_id)", 

72 "condition": "source='js_exception' AND `$auto_captured`" 

73 }, 

74 schemas.AlertColumn.ERRORS__BACKEND__COUNT: { 

75 "table": lambda timestamp: f"{exp_ch_helper.get_main_events_table(timestamp)} AS errors", 

76 "eventType": "ERROR", 

77 "formula": "COUNT(DISTINCT session_id)", 

78 "condition": "source!='js_exception' AND `$auto_captured`" 

79 }, 

80} 

81 

82 

83def Build(a): 

84 now = TimeUTC.now() 

85 params = {"project_id": a["projectId"], "now": now} 

86 full_args = {} 

87 if a["seriesId"] is not None: 

88 a["filter"]["sort"] = "session_id" 

89 a["filter"]["order"] = schemas.SortOrderType.DESC 

90 a["filter"]["startDate"] = 0 

91 a["filter"]["endDate"] = TimeUTC.now() 

92 try: 

93 data = schemas.SessionsSearchPayloadSchema.model_validate(a["filter"]) 

94 except ValidationError: 

95 logger.warning("Validation error for:") 

96 logger.warning(a["filter"]) 

97 raise 

98 

99 full_args, query_part = sessions.search_query_parts_ch(data=data, error_status=None, errors_only=False, 

100 issue=None, project_id=a["projectId"], user_id=None, 

101 favorite_only=False) 

102 subQ = f"""SELECT COUNT(session_id) AS value  

103 {query_part}""" 

104 else: 

105 colDef = LeftToDb[a["query"]["left"]] 

106 params["event_type"] = LeftToDb[a["query"]["left"]].get("eventType") 

107 subQ = f"""SELECT {colDef["formula"]} AS value 

108 FROM {colDef["table"](now)} 

109 WHERE project_id = %(project_id)s  

110 {"AND event_type=%(event_type)s" if params["event_type"] else ""}  

111 {"AND " + colDef["condition"] if colDef.get("condition") else ""}""" 

112 

113 q = f"""SELECT coalesce(value,0) AS value, coalesce(value,0) {a["query"]["operator"]} {a["query"]["right"]} AS valid""" 

114 

115 if a["detectionMethod"] == schemas.AlertDetectionMethod.THRESHOLD: 

116 if a["seriesId"] is not None: 

117 q += f""" FROM ({subQ}) AS stat""" 

118 else: 

119 q += f""" FROM ({subQ}  

120 AND datetime>=toDateTime(%(startDate)s/1000)  

121 AND datetime<=toDateTime(%(now)s/1000) ) AS stat""" 

122 params = {**params, **full_args, "startDate": TimeUTC.now() - a["options"]["currentPeriod"] * 60 * 1000} 

123 else: 

124 if a["change"] == schemas.AlertDetectionType.CHANGE: 

125 if a["seriesId"] is not None: 

126 sub2 = subQ.replace("%(startDate)s", "%(timestamp_sub2)s").replace("%(endDate)s", "%(startDate)s") 

127 sub1 = f"SELECT (({subQ})-({sub2})) AS value" 

128 q += f" FROM ( {sub1} ) AS stat" 

129 params = {**params, **full_args, 

130 "startDate": TimeUTC.now() - a["options"]["currentPeriod"] * 60 * 1000, 

131 "timestamp_sub2": TimeUTC.now() - 2 * a["options"]["currentPeriod"] * 60 * 1000} 

132 else: 

133 sub1 = f"""{subQ} AND datetime>=toDateTime(%(startDate)s/1000) 

134 AND datetime<=toDateTime(%(now)s/1000)""" 

135 params["startDate"] = TimeUTC.now() - a["options"]["currentPeriod"] * 60 * 1000 

136 sub2 = f"""{subQ} AND datetime<toDateTime(%(startDate)s/1000)  

137 AND datetime>=toDateTime(%(timestamp_sub2)s/1000)""" 

138 params["timestamp_sub2"] = TimeUTC.now() - 2 * a["options"]["currentPeriod"] * 60 * 1000 

139 sub1 = f"SELECT (( {sub1} )-( {sub2} )) AS value" 

140 q += f" FROM ( {sub1} ) AS stat" 

141 

142 else: 

143 if a["seriesId"] is not None: 

144 sub2 = subQ.replace("%(startDate)s", "%(timestamp_sub2)s").replace("%(endDate)s", "%(startDate)s") 

145 sub1 = f"SELECT (({subQ})/NULLIF(({sub2}),0)-1)*100 AS value" 

146 q += f" FROM ({sub1}) AS stat" 

147 params = {**params, **full_args, 

148 "startDate": TimeUTC.now() - a["options"]["currentPeriod"] * 60 * 1000, 

149 "timestamp_sub2": TimeUTC.now() \ 

150 - (a["options"]["currentPeriod"] + a["options"]["currentPeriod"]) \ 

151 * 60 * 1000} 

152 else: 

153 sub1 = f"""{subQ} AND datetime>=toDateTime(%(startDate)s/1000) 

154 AND datetime<=toDateTime(%(now)s/1000)""" 

155 params["startDate"] = TimeUTC.now() - a["options"]["currentPeriod"] * 60 * 1000 

156 sub2 = f"""{subQ} AND datetime<toDateTime(%(startDate)s/1000) 

157 AND datetime>=toDateTime(%(timestamp_sub2)s/1000)""" 

158 params["timestamp_sub2"] = TimeUTC.now() \ 

159 - (a["options"]["currentPeriod"] + a["options"]["currentPeriod"]) * 60 * 1000 

160 sub1 = f"SELECT (({sub1})/NULLIF(({sub2}),0)-1)*100 AS value" 

161 q += f" FROM ({sub1}) AS stat" 

162 

163 return q, params 

164 

165 

166def process(): 

167 logger.info("> processing alerts on CH") 

168 notifications = [] 

169 all_alerts = alerts_listener.get_all_alerts() 

170 with pg_client.PostgresClient() as cur, ch_client.ClickHouseClient() as ch_cur: 

171 for alert in all_alerts: 

172 if alert["query"]["left"] != "CUSTOM": 

173 continue 

174 if alert_helpers.can_check(alert): 

175 query, params = Build(alert) 

176 try: 

177 query = ch_cur.format(query=query, parameters=params) 

178 except Exception as e: 

179 logger.error( 

180 f"!!!Error while building alert query for alertId:{alert['alertId']} name: {alert['name']}") 

181 logger.error(e) 

182 continue 

183 logger.debug(alert) 

184 logger.debug(query) 

185 try: 

186 # result = ch_cur.execute(query=query) 

187 result = 0 

188 if len(result) > 0: 

189 result = result[0] 

190 

191 if result["valid"]: 

192 logger.info("Valid alert, notifying users") 

193 notifications.append(alert_helpers.generate_notification(alert, result)) 

194 except Exception as e: 

195 logger.error(f"!!!Error while running alert query for alertId:{alert['alertId']}") 

196 logger.error(str(e)) 

197 logger.error(query) 

198 if len(notifications) > 0: 

199 cur.execute( 

200 cur.mogrify(f"""UPDATE public.alerts  

201 SET options = options||'{{"lastNotification":{TimeUTC.now()}}}'::jsonb  

202 WHERE alert_id IN %(ids)s;""", {"ids": tuple([n["alertId"] for n in notifications])})) 

203 if len(notifications) > 0: 

204 alerts.process_notifications(notifications)