forked from Reecepbcups/cosmos-cache
-
Notifications
You must be signed in to change notification settings - Fork 0
/
HELPERS.py
238 lines (185 loc) · 7.23 KB
/
HELPERS.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
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
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
import json
import re
from os import getenv
import httpx
import CONFIG
from CONFIG import KV_STORE
from HELPERS_TYPES import CallType, Mode
def ttl_block_only(cache_seconds: int = 0):
# this way on a new block, we delete all *;IsBlockOnly;* keys
return (
"IsBlockOnly"
if cache_seconds == Mode.FOR_BLOCK_TIME.value
else f"{cache_seconds}s"
)
total_calls = {
# RPC:
CallType.RPC_GET_CACHE.value: 0,
CallType.RPC_GET_OUTBOUND.value: 0,
# RPC Post:
CallType.RPC_POST_CACHE.value: 0,
CallType.RPC_POST_OUTBOUND.value: 0,
# REST:
CallType.REST_GET_CACHE.value: 0,
CallType.REST_GET_OUTBOUND.value: 0,
}
def increment_call_value(key: str, amount: int = 1):
global total_calls
if CONFIG.ENABLE_COUNTER == False:
return
if key not in total_calls:
total_calls[str(key)] = 0
if total_calls[key] >= CONFIG.INC_EVERY:
KV_STORE.incr(f"{key}", amount=total_calls[key])
total_calls[key] = 0
else:
total_calls[key] += amount
if CONFIG.DEBUGGING:
print(f"incremented {key} to {total_calls[key]}")
# NOTE: testing only
# print("testing only dump here")
# KV_STORE.dump()
def download_openapi_locally():
# TODO: What if there is no swagger API?
r = httpx.get(CONFIG.OPEN_API)
if r.status_code != 200:
return
file_loc = f"{CONFIG.PROJECT_DIR}/static/openapi.yml"
with open(file_loc, "w") as f:
f.write(r.text)
def get_swagger_code_from_source():
req = httpx.get(f"{CONFIG.REST_URL}")
html = req.text.replace(
"//unpkg.com/[email protected]/favicon-16x16.png",
"/static/favicon.png",
)
html = re.sub(r"<title>.*</title>", f"<title>{CONFIG.API_TITLE}</title>", html)
return html
def replace_rpc_text() -> str:
# we replace after on requests of the user, then replace this text to our cache endpoint at time of requests to root endpoint
try:
RPC_ROOT_HTML = httpx.get(f"{CONFIG.RPC_URL}/").text
except:
RPC_ROOT_HTML = httpx.get(f"{CONFIG.BACKUP_RPC_URL}/").text
RPC_TITLE = getenv("RPC_TITLE", "")
if len(RPC_TITLE) > 0:
RPC_ROOT_HTML = RPC_ROOT_HTML.replace(
"<html><body>",
f"<html><head><title>{RPC_TITLE}</title></head><body>",
)
# Puts text at the bottom, maybe put at the top in the future?
RPC_CUSTOM_TEXT = getenv("RPC_CUSTOM_TEXT", "")
if len(RPC_CUSTOM_TEXT) > 0:
RPC_ROOT_HTML = RPC_ROOT_HTML.replace(
"Available endpoints:<br><br>",
f"{RPC_CUSTOM_TEXT}<br>Available endpoints:<br><br>",
)
# add cache_info endpoint. THIS REMOVES BLANK 'Available endpoints:<br><br>'
RPC_ROOT_HTML = RPC_ROOT_HTML.replace(
"Available endpoints:<br><br>",
f'<a href="//{{BASE_URL}}/cache_info">//{{BASE_URL}}/cache_info</a><br><br>',
# we replace the BASE_URL on the call to the root endpoint
)
RPC_ROOT_HTML = RPC_ROOT_HTML.replace(
"/cache_info</a><br><br>",
f'/cache_info</a><br><a href="//{{BASE_URL}}/prices">//{{BASE_URL}}/prices</a><br><br>',
# we replace the BASE_URL on the call to the root endpoint
)
# Set RPC favicon to nothing
RPC_ROOT_HTML = RPC_ROOT_HTML.replace(
"<head>",
f'<head><link rel="icon" href="data:,">',
)
return RPC_ROOT_HTML
INITIAL_HTML = """<html><head><title>Cache Stats</title></head><body>"""
CLOSING_HTML = """</body></html>"""
def get_config_values():
KVs = [item for item in dir(CONFIG) if not item.startswith("__")]
items = {item: getattr(CONFIG, item) for item in KVs}
return f"""
{INITIAL_HTML}
<h2>Config Values</h2>
<p>{items}</p>
{CLOSING_HTML}
"""
def get_stats_html():
updates_every = CONFIG.INC_EVERY
# gets information about the kv store
rpc_get_cache = KV_STORE.get(CallType.RPC_GET_CACHE.value)
rpc_get_outbound = KV_STORE.get(CallType.RPC_GET_OUTBOUND.value)
rpc_post_cache = KV_STORE.get(CallType.RPC_POST_CACHE.value)
rpc_post_outbound = KV_STORE.get(CallType.RPC_POST_OUTBOUND.value)
rest_cache = KV_STORE.get(CallType.REST_GET_CACHE.value)
rest_outbound = KV_STORE.get(CallType.REST_GET_OUTBOUND.value)
# no rest post yet, not added.
# converts (1 so no div / 0 errors)
rpc_get_cache = 1 if rpc_get_cache == None else int(rpc_get_cache.decode("utf-8"))
rpc_get_outbound = (
1 if rpc_get_outbound == None else int(rpc_get_outbound.decode("utf-8"))
)
rpc_post_cache = (
1 if rpc_post_cache == None else int(rpc_post_cache.decode("utf-8"))
)
rpc_post_outbound = (
1 if rpc_post_outbound == None else int(rpc_post_outbound.decode("utf-8"))
)
rest_cache = 1 if rest_cache == None else int(rest_cache.decode("utf-8"))
rest_outbound = 1 if rest_outbound == None else int(rest_outbound.decode("utf-8"))
return f"""
{INITIAL_HTML}
<h2>Updates every {updates_every} calls</h2>
<h1>RPC GET Cache Stats</h1>
<p>RPC Cache Hits: {rpc_get_cache}</p>
<p>RPC outbound: {rpc_get_outbound}</p>
<p>Percent Cached: {round((rpc_get_cache / (rpc_get_cache + rpc_get_outbound)) * 100, 2)}%</p>
<br>
<h1>RPC POST Cache Stats</h1>
<p>RPC Cache Hits: {rpc_post_cache}</p>
<p>RPC outbound: {rpc_post_outbound}</p>
<p>Percent Cached: {round((rpc_post_cache / (rpc_post_cache + rpc_post_outbound)) * 100, 2)}%</p>
<br>
<h1>REST GET Cache Stats</h1>
<p>REST Cache Hits: {rest_cache}</p>
<p>REST outbound: {rest_outbound}</p>
<p>Percent Cached: {round((rest_cache / (rest_cache + rest_outbound)) * 100, 2)}%</p>
{CLOSING_HTML}
"""
def _hide_data(json: dict, str_path: str, cfg_value: str) -> dict:
"""
cfg_value is some string
path is the json path in string form. For example, ['result']['node_info'] is result.node_info
json is teh default json response
Given this, if the path exist in the json, edit said path and update it to be the cfg_value
Then return the updated JSON
else:
return the original JSON
"""
if len(str_path) == 0 or len(cfg_value) == 0:
return json
path = str_path.split(".")
parent = json
for key in path[:-1]:
parent = parent.get(key, {})
if not parent:
return json
parent[path[-1]] = cfg_value
return json
def hide_rpc_data(res: dict, endpoint_path: str):
if endpoint_path.lower().startswith("status"):
res = _hide_data(res, "result.node_info.listen_addr", CONFIG.RPC_LISTEN_ADDRESS)
res = _hide_data(
res, "result.node_info.other.rpc_address", CONFIG.RPC_LISTEN_ADDRESS
)
res = _hide_data(res, "result.node_info.moniker", CONFIG.NODE_MONIKER)
return res
def hide_rest_data(res: dict, endpoint_path: str):
if endpoint_path.lower().endswith("v1beta1/node_info"):
res = _hide_data(
res, "default_node_info.listen_addr", CONFIG.RPC_LISTEN_ADDRESS
)
res = _hide_data(
res, "default_node_info.other.rpc_address", CONFIG.RPC_LISTEN_ADDRESS
)
res = _hide_data(res, "default_node_info.moniker", CONFIG.NODE_MONIKER)
# hide application_version.build_deps?
return res