TL; DR

因为硬盘意外寄寄,导致我的 Grafana 数据完全丢失。在重新构建的过程中,我思考了许多

1. 我是否应该仅仅局限于商家公开的测试 IP/Looking Glass ?

不应该,这类站点有很多,比如古博。商家的测试 IP/LG 可以从一定程度上说明该商家的网络情况。但可能存在作弊的情况。

2. 我应该使用什么样的监控程序?

比如古博 这类使用 smokeping 的站点有着本质的缺点(比如 UI 十分质朴,功能较为单一等)。虽说这个站点大概率是我一个人使用,何必为难自己呢?

当然是 Grafana ~

服务列表收集

解决方案1

Ookla 出品的 Speedtest 应该是全球范围内涵盖国家/商家最多的测速服务。假如存在某方法获取到完整的 Speedtest 服务端列表,对其进行监控,应该是覆盖最全的全球网络监控服务了。

目前,Speedtest 仅仅支持使用与出口 IP 地理位置接近的测试节点。如下图所示,仅仅能通过 LAX 找到 LAX 同城或附近的部分测速节点。
H6BwIP.png

如果想要获取完整的测速节点,很简单的办法——购买全球的 VPS 即可。

解决方案2

上述的方案1不是一个可行的解决方案(资金压力)。于是打开了 Web 页面进行分析。在一次刷新的过程中发现了下图的 API 请求。
H66kjS.png

https://www.speedtest.net/api/js/servers?engine=js&search=China&https_functional=true&limit=100

经过分析,一个完整的 API 请求参数包括以下参数

  1. engine 表示通过何种方式请求,此时通过 Web 页面方式获取,js 也算正常
  2. search 表示搜索内容
  3. https_functional 表示服务端协议是否采用 HTTPS
  4. 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 个不同的测速节点)。