-
Notifications
You must be signed in to change notification settings - Fork 21
Expand file tree
/
Copy pathflask_app.py
More file actions
146 lines (99 loc) · 2.63 KB
/
flask_app.py
File metadata and controls
146 lines (99 loc) · 2.63 KB
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
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
#env AGENT_KEY=agnetkeyhere FLASK_APP=examples/demo/flask_app.py flask run -p 5010
from __future__ import print_function
import os
import time
import sys
import threading
import subprocess
import collections
import random
import traceback
from flask import Flask
try:
# python 2
from urllib2 import urlopen
except ImportError:
# python 3
from urllib.request import urlopen
sys.path.append(".")
import stackimpact
# StackImpact agent initialization
agent = stackimpact.start(
agent_key = os.environ['AGENT_KEY'],
dashboard_address = os.environ['DASHBOARD_ADDRESS'],
app_name = 'ExamplePythonFlaskApp',
app_version = '1.0.0',
debug = True)
# Simulate CPU intensive work
def simulate_cpu():
duration = 10 * 60 * 60
usage = 10
while True:
for j in range(0, duration):
for i in range(0, usage * 15000):
text = "text1" + str(i)
text = text + "text2"
time.sleep(1 - usage/100)
t = threading.Thread(target=simulate_cpu)
t.start()
# Simulate memory leak
def simulate_mem_leak():
while True:
mem1 = []
for j in range(0, 1800):
mem2 = []
for i in range(0, 1000):
obj1 = {'v': random.randint(0, 1000000)}
mem1.append(obj1)
obj2 = {'v': random.randint(0, 1000000)}
mem2.append(obj2)
time.sleep(1)
t = threading.Thread(target=simulate_mem_leak)
t.start()
# Simulate lock
def simulate_lock():
lock = threading.Lock()
def lock_wait():
lock.acquire()
lock.release()
while True:
lock.acquire()
t = threading.Thread(target=lock_wait)
t.start()
time.sleep(1)
lock.release()
time.sleep(1)
t = threading.Thread(target=simulate_lock)
t.start()
# Simulate exceptions
def simulate_exceptions():
while True:
try:
raise ValueError('some error')
except:
traceback.print_exc()
pass
time.sleep(2)
t = threading.Thread(target=simulate_exceptions)
t.start()
# Simulate http server
def simulate_http_traffic():
while True:
try:
urlopen('http://localhost:5010', timeout=10)
time.sleep(2)
except:
traceback.print_exc()
pass
t = threading.Thread(target=simulate_http_traffic)
t.start()
def cpu_work():
for i in range(0, 1000000):
text = "text1" + str(i)
text = text + "text2"
app = Flask(__name__)
@app.route('/')
def hello_world():
time.sleep(0.5)
cpu_work()
return 'Hello'