这篇文章上次修改于 412 天前，可能其部分内容已经发生变化，如有疑问可询问作者。
TL; DR
因为硬盘意外寄寄，导致我的 Grafana 数据完全丢失。在重新构建的过程中，我思考了许多
1. 我是否应该仅仅局限于商家公开的测试 IP/Looking Glass ？
不应该，这类站点有很多，比如古博。商家的测试 IP/LG 可以从一定程度上说明该商家的网络情况。但可能存在作弊的情况。
2. 我应该使用什么样的监控程序？
比如古博 这类使用 smokeping 的站点有着本质的缺点（比如 UI 十分质朴，功能较为单一等）。虽说这个站点大概率是我一个人使用，何必为难自己呢？
当然是 Grafana ~
服务列表收集
解决方案1
Ookla 出品的 Speedtest 应该是全球范围内涵盖国家/商家最多的测速服务。假如存在某方法获取到完整的 Speedtest 服务端列表，对其进行监控，应该是覆盖最全的全球网络监控服务了。
目前，Speedtest 仅仅支持使用与出口 IP 地理位置接近的测试节点。如下图所示，仅仅能通过 LAX 找到 LAX 同城或附近的部分测速节点。
如果想要获取完整的测速节点，很简单的办法——购买全球的 VPS 即可。
解决方案2
上述的方案1不是一个可行的解决方案（资金压力）。于是打开了 Web 页面进行分析。在一次刷新的过程中发现了下图的 API 请求。
https://www.speedtest.net/api/js/servers?engine=js&search=China&https_functional=true&limit=100经过分析，一个完整的 API 请求参数包括以下参数
- engine 表示通过何种方式请求，此时通过 Web 页面方式获取，js 也算正常
- search 表示搜索内容
- https_functional 表示服务端协议是否采用 HTTPS
- limit 表示最大大小（实测 1000 内可用）
请求的返回值是一个 json-list，其中的每项元素结构如下
{
"url":"http://suntechspeedtest.com.prod.hosts.ooklaserver.net:8080/speedtest/upload.php",
"lat":"22.2500",
"lon":"114.1667",
"distance":1,
"name":"Hong Kong",
"country":"China",
"cc":"HK",
"sponsor":"STC",
"id":"1536",
"preferred":0,
"https_functional":1,
"host":"suntechspeedtest.com.prod.hosts.ooklaserver.net:8080"
}其中，host 就是我们需要的数据。
编码与数据清洗
在 search 字段中，我采用了 ISO 3166-1 alpha-2 的搜索内容。ISO 3166-1 alpha-2 是国际标准化组织ISO 3166标准第一部分ISO 3166-1的二位字母表示方式，旨在为国家、属地、具特殊科学价值地点建立国际认可的代码。ISO 3166-1二位字母代码是目前应用最为广泛的国家代码，被大量应用于国家和地区顶级域名。
通过这种方式，可以避免上例中尴尬的数据结果。
import asyncio
import datetime
import json
import os
from typing import List
import aiohttp
URL = "https://www.speedtest.net/api/js/servers"
HEADERS = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.80 Safari/537.36"
}
TODAY = str(datetime.date.today())
SAVE_PATH = os.path.abspath("speedtest_server_list/{today}".format(today=TODAY))
async def fetch(client: aiohttp.ClientSession, location: str) -> List:
async with client.get(
url=URL,
params={
"engine": "js",
"search": location,
"limit": 1000,
},
headers=HEADERS,
) as resp:
assert resp.status == 200
return await resp.json()
async def do_request(task_queue: asyncio.Queue, result_queue: asyncio.Queue):
async with aiohttp.ClientSession() as client:
while not task_queue.empty():
task = await task_queue.get()
retries = task.get("retries")
if retries >= 5:
continue
try:
_data = await fetch(client, location=task.get("e_name"))
iso_3166_1_alpha_2 = task.get("iso_3166_1_alpha_2")
data = [_ for _ in _data if iso_3166_1_alpha_2 == _.get("cc")]
task.update({"servers": data, "nan_filter": _data})
await result_queue.put(task)
except Exception as e:
task.update({"retries": retries + 1})
await task_queue.put(task)
async def init_task() -> asyncio.Queue:
queue = asyncio.Queue()
with open("country.json", "r") as f:
countries = json.load(f)
for country in countries:
country.update({"retries": 0})
queue.put_nowait(country)
return queue
async def save(result_queue: asyncio.Queue):
t = []
while not result_queue.empty():
task = await result_queue.get()
nan_filter = task.get("nan_filter")
task.pop("nan_filter")
country_e_cname_json = "{}.json".format(task.get("e_name"))
with open(os.path.join(SAVE_PATH, country_e_cname_json), "w") as f:
json.dump(task, f, ensure_ascii=False, indent=4)
t.extend(nan_filter)
with open(os.path.join(SAVE_PATH, "all.json"), "w") as f:
json.dump(t, f, ensure_ascii=False, indent=4)
def main():
if not os.path.exists(SAVE_PATH):
os.makedirs(SAVE_PATH)
loop = asyncio.get_event_loop()
t_q = loop.run_until_complete(init_task())
r_q = asyncio.Queue()
tasks = [do_request(task_queue=t_q, result_queue=r_q) for _ in range(5)]
loop.run_until_complete(asyncio.gather(*tasks))
loop.run_until_complete(save(result_queue=r_q))
if __name__ == "__main__":
main()
代码中需要用到的 country.json 文件内容如下：
[
{
"c_name":"阿富汗",
"e_name":"Afghanistan",
"iso_3166_1_alpha_2":"AF",
"iso_3166_1_alpha_3":"AFG",
"iso_3166_1_numeric":"004"
}...
]结果将会保存至本地硬盘，过几天放出第一波数据集（结果包含了 4663 个不同的测速节点）。
没有评论