-
Notifications
You must be signed in to change notification settings - Fork 3
/
app.py
108 lines (78 loc) · 2.71 KB
/
app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
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
from fastapi import FastAPI, Response
import uvicorn
import asyncio
import random
from autometrics import autometrics, init
from autometrics.objectives import Objective, ObjectiveLatency, ObjectivePercentile
from prometheus_client import generate_latest
from git_utils import get_git_commit, get_git_branch
app = FastAPI()
VERSION = "0.0.4"
init(tracker="prometheus", version=VERSION, commit=get_git_commit(), branch=get_git_branch())
# Set up a metrics endpoint for Prometheus to scrape
# `generate_latest` returns the latest metrics data in the Prometheus text format
@app.get("/metrics")
def metrics():
return Response(generate_latest())
# Define Objectives for the api routes
API_SLO_HIGH_SUCCESS = Objective(
"Animal API Route SLO for High Success Rate",
success_rate=ObjectivePercentile.P99,
)
API_QUICK_RESPONSES = Objective(
"Animal API SLO for Low Latency",
latency=(ObjectiveLatency.Ms100, ObjectivePercentile.P99),
)
ANIMALS = ["snail", "rabbit", "panda", "beaver"]
@app.get("/")
@autometrics(objective=API_SLO_HIGH_SUCCESS)
def animals():
return {"animals": list_animals_helper()}
@app.get("/snail")
@autometrics(objective=API_QUICK_RESPONSES)
async def snail(max: int = 11):
await validate_animal()
# Snails are slow sometimes.
await snail_service(max)
return {"suggestion": "Let's take it easy"}
@app.get("/rabbit")
@autometrics(objective=API_QUICK_RESPONSES)
def rabbit():
# Rabbits are fast. They have very low latency
return {"suggestion": "Let's drink coffee and go for a jog"}
@app.get("/panda")
@autometrics(objective=API_SLO_HIGH_SUCCESS)
async def panda():
# Pandas are clumsy. They error sometimes
await validate_animal()
await clumsy_panda_service()
return {"suggestion": "Let's eat bamboo"}
@app.get("/beaver")
@autometrics(objective=API_SLO_HIGH_SUCCESS)
def beaver():
# Beavers are hard working. They never error
return {"suggestion": "Let's build a dam"}
@autometrics
def list_animals_helper():
"""Return all animals"""
return ANIMALS
@autometrics
async def clumsy_panda_service():
"""Randomly raise an error with a 50% chance"""
# Generate a random number between 1 and 2
error_chance = random.randint(1, 2)
# If the random number is 1, raise an error
if error_chance == 1:
raise Exception("Random error occurred!")
@autometrics
async def validate_animal():
delay = random.randint(0, 2) * 0.01
await asyncio.sleep(delay)
return True
@autometrics
async def snail_service(max: int):
"""Generate a random latency between 0 and `max * 10` ms"""
delay = random.randint(0, max) * 0.01
await asyncio.sleep(delay)
if __name__ == "__main__":
uvicorn.run(app, host="localhost", port=8080)