in static/Reliability/300_Health_Checks_and_Dependencies/Code/Python/server_healthcheck.py [0:0]
def do_GET(self):
print("path: ", self.path)
# Default request URL without additional path info (main response page)
if self.path == '/':
message = "<h1>Enjoy some classic television</h1>"
message += "<h1>What to watch next....</h1>"
# Generate User ID between 1 and 4
# This currently uses a randomly generated user.
# In the future maybe allow student to supply the user ID as input
user_id = str(random.randint(1, 4))
# Error handling:
# surround the call to RecommendationService in a try catch
try:
# Call the getRecommendation API on the RecommendationService
response = call_getRecommendation(self.region, user_id)
# Parses value of recommendation from DynamoDB JSON return value
# {'Item': {
# 'ServiceAPI': {'S': 'getRecommendation'},
# 'UserID': {'N': '1'},
# 'Result': {'S': 'M*A*S*H'}, ...
tv_show = response['Item']['Result']['S']
user_name = response['Item']['UserName']['S']
message += recommendation_message (user_name, tv_show, True)
# Error handling:
# If the service dependency fails, and we cannot make a personalized recommendation
# then give a pre-selected (static) recommendation
# and report diagnostic information
except Exception as e:
message += recommendation_message ('Valued Customer', 'I Love Lucy', False)
message += '<br><br><br><h2>Diagnostic Info:</h2>'
message += '<br>We are unable to provide personalized recommendations'
message += '<br>If this persists, please report the following info to us:'
message += str(traceback.format_exception_only(e.__class__, e))
# Add metadata - this is useful in the lab to see
# info about the EC2 instance and Availability Zone
message += get_metadata()
# Send successful response status code
self.send_response(200)
# Send headers
self.send_header('Content-type', 'text/html')
self.end_headers()
# Write html output
self.wfile.write(
bytes(
html.format(Title="Resiliency workshop", Content=message),
"utf-8"
)
)
# Healthcheck request - will be used by the Elastic Load Balancer
elif self.path == '/healthcheck':
is_healthy = False
error_msg = ''
TEST = 'test'
# Make a request to RecommendationService using a predefined
# test call as part of health assessment for this server
try:
# call RecommendationService using the test user
user_id = str(0)
response = call_getRecommendation(self.region, user_id)
# Parses value of recommendation from DynamoDB JSON return value
tv_show = response['Item']['Result']['S']
user_name = response['Item']['UserName']['S']
# Server is healthy of RecommendationService returned the expected response
is_healthy = (tv_show == TEST) and (user_name == TEST)
# If the service dependency fails, capture diagnostic info
except Exception as e:
error_msg += str(traceback.format_exception_only(e.__class__, e))
# Based on the health assessment
# If it succeeded return a healthy code
# If it failed return a server failure code
message = ""
if (is_healthy):
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
message += "<h1>Success</h1>"
# Add metadata
message += get_metadata()
else:
self.send_response(503)
self.send_header('Content-type', 'text/html')
self.end_headers()
message += "<h1>Fail</h1>"
message += "<h3>Error message:</h3>"
message += error_msg
# Add metadata
message += get_metadata()
self.wfile.write(
bytes(
html.format(Title="healthcheck", Content=message),
"utf-8"
)
)
return