HealthCheckExecutor
The HealthCheckExecutor encapsulates the functionality to perform a health check on
a site and properly notify users or update statuspage.io accordingly.
A Healthcheck represents a simple request to the defined url. If a non-200 the
request generates an exception or a non-200 response, the site is determined to
be down.
If statuspage_operator is present and the HealthCheckSettings have a component_id
set, statuspage.io will be updated according to the following rules.
- If the site returns a 2xx response and statuspage.io lists the component as
non-operational:
- The component's status will be set to operational
- Any open incidents associated with this component will be marked as resolved
- If the site returns a non-2xx response or an exception and statuspage.io lists
the component as operational:
- The component's status will be set to operational
- An incident will be opened using the
name and associated with
this component.
Attributes:
| Name |
Type |
Description |
statuspage_operator |
StatusPageOperator
|
The name of the site being checked
|
notifier |
Notifier
|
The url to be fetched as part of the check
|
Source code in pi_monitor/healthchecks.py
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189 | class HealthCheckExecutor:
"""HealthCheckExecutor
The HealthCheckExecutor encapsulates the functionality to perform a health check on
a site and properly notify users or update statuspage.io accordingly.
A Healthcheck represents a simple request to the defined `url`. If a non-200 the
request generates an exception or a non-200 response, the site is determined to
be down.
If `statuspage_operator` is present and the HealthCheckSettings have a component_id
set, statuspage.io will be updated according to the following rules.
- If the site returns a 2xx response and statuspage.io lists the component as
non-operational:
- The component's status will be set to operational
- Any open incidents associated with this component will be marked as resolved
- If the site returns a non-2xx response or an exception and statuspage.io lists
the component as operational:
- The component's status will be set to operational
- An incident will be opened using the `name` and associated with
this component.
Attributes:
statuspage_operator: The name of the site being checked
notifier: The url to be fetched as part of the check
"""
statuspage_operator: StatusPageOperator
notifier: Notifier
def __init__(
self,
status_operator: StatusPageOperator,
notifier: Notifier,
):
"""Constructor
Constructs an instance of the HealthCheckExecutor with the given
[StatusPageOperator][pi_monitor.StatusPageOperator] and
[Notifier][pi_monitor.Notifier].
Attributes:
statuspage_operator: The name of the site being checked
notifier: The url to be fetched as part of the check
"""
self.statuspage_operator = status_operator
self.notifier = notifier
def execute_health_check(self, check_settings: HealthCheckSettings):
"""Execute a health check
Executes a health check using the provided HealthCheckSettings.
Args:
check_settings: An instance of
[HealthCheckSettings][pi_monitor.HealthCheckSettings]
"""
logger.info("Checking %s...", check_settings.name)
send_notification = False
http_result = self._get_http(check_settings.url)
if http_result.success:
# Good Check
logger.info("Status OK")
op_level = OpLevel.Operational
else:
# Bad check
op_level = OpLevel.Full_Outage
logger.warning(http_result.message)
send_notification = True
notification_text = http_result.message
if check_settings.status_page and check_settings.status_page.component_id != "":
status_result = self._update_status_page(check_settings, op_level)
send_notification = (
status_result.incident_result.incident_created
or status_result.incident_result.incident_resolved
)
notification_text = status_result.incident_result.incident.description
if (
send_notification
and notification_text is not None
and notification_text != ""
):
logger.info("Sending notification: %s", notification_text)
self.notifier.notify(check_settings.name, notification_text)
def _get_http(self, url: str) -> HttpGetResult:
"""Retrieve data from the URL
Attempt to get data from the provided URL
Args:
url: The url to be retrieved
Returns:
An [HttpGetResult][pi_monitor.HttpGetResult]
"""
if not url or url == "":
result = HttpGetResult(False, "no url defined")
return result
try:
logger.debug("Requesting %s", url)
r = requests.get(url)
result = self._process_response(r)
except Exception as e:
logger.error("Request failed exception %s", e)
result = HttpGetResult(False, "Unknown status failure")
return result
def _process_response(self, response: requests.Response) -> HttpGetResult:
"""Process the HTTP Requests response
Convert the provided Response object from the requests module into an
[HttpGetResult][healthchecks.HttpGetResult].
Args:
response: The [requests.Response] object from the HTTP operation
Returns:
An [HttpGetResult][healthchecks.HttpGetResult]
"""
result = HttpGetResult(response.status_code == 200)
if not result.success:
logger.info(
"Request failed with Response Code %d: %s",
response.status_code,
response.text,
)
result.message = f"{response.status_code} {response.text}"
return result
result.raw_response = response.text
return result
def _update_status_page(
self, check_settings: HealthCheckSettings, op_level: OpLevel
) -> StatusResult:
incident = Incident()
incident.name = check_settings.name
description_dict = {
OpLevel.Operational: "Operating Normally",
OpLevel.Degraded: "Service Degraded",
OpLevel.Partial_Outage: "Partial Service Outage",
OpLevel.Full_Outage: "Major Service Outage",
}
incident.description = description_dict[op_level]
return self.statuspage_operator.update_component_status(
check_settings.status_page.component_id, op_level, incident
)
|
__init__(status_operator, notifier)
Constructor
Constructs an instance of the HealthCheckExecutor with the given
StatusPageOperator and
Notifier.
Attributes:
| Name |
Type |
Description |
statuspage_operator |
|
The name of the site being checked
|
notifier |
|
The url to be fetched as part of the check
|
Source code in pi_monitor/healthchecks.py
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80 | def __init__(
self,
status_operator: StatusPageOperator,
notifier: Notifier,
):
"""Constructor
Constructs an instance of the HealthCheckExecutor with the given
[StatusPageOperator][pi_monitor.StatusPageOperator] and
[Notifier][pi_monitor.Notifier].
Attributes:
statuspage_operator: The name of the site being checked
notifier: The url to be fetched as part of the check
"""
self.statuspage_operator = status_operator
self.notifier = notifier
|
execute_health_check(check_settings)
Execute a health check
Executes a health check using the provided HealthCheckSettings.
Parameters:
Source code in pi_monitor/healthchecks.py
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122 | def execute_health_check(self, check_settings: HealthCheckSettings):
"""Execute a health check
Executes a health check using the provided HealthCheckSettings.
Args:
check_settings: An instance of
[HealthCheckSettings][pi_monitor.HealthCheckSettings]
"""
logger.info("Checking %s...", check_settings.name)
send_notification = False
http_result = self._get_http(check_settings.url)
if http_result.success:
# Good Check
logger.info("Status OK")
op_level = OpLevel.Operational
else:
# Bad check
op_level = OpLevel.Full_Outage
logger.warning(http_result.message)
send_notification = True
notification_text = http_result.message
if check_settings.status_page and check_settings.status_page.component_id != "":
status_result = self._update_status_page(check_settings, op_level)
send_notification = (
status_result.incident_result.incident_created
or status_result.incident_result.incident_resolved
)
notification_text = status_result.incident_result.incident.description
if (
send_notification
and notification_text is not None
and notification_text != ""
):
logger.info("Sending notification: %s", notification_text)
self.notifier.notify(check_settings.name, notification_text)
|