Merge pull request 'proto: add RecordSpeedTest RPC + request/response messages' (#1) from feature/speedtest-telemetry into main
Reviewed-on: #1
This commit is contained in:
@@ -5,6 +5,7 @@ build:
|
||||
go build -o bin/stats-collector ./cmd/collector
|
||||
go build -o bin/stats-snapshot ./cmd/snapshot
|
||||
go build -o bin/stats-issue-token ./cmd/issue-token
|
||||
CGO_ENABLED=0 go build -o bin/speedtest-sync ./cmd/speedtest-sync
|
||||
|
||||
test:
|
||||
go test ./...
|
||||
|
||||
@@ -23,9 +23,61 @@ type nodeData struct {
|
||||
Recent []sample `json:"recent"`
|
||||
}
|
||||
|
||||
// SpeedtestSnapshot exposes recent speedtest rows + day/week/month median
|
||||
// aggregates per node. Recent uses storage.SpeedTest's capitalized field
|
||||
// names (no JSON tags on the struct); the frontend reads them as-is.
|
||||
type SpeedtestSnapshot struct {
|
||||
Recent []storage.SpeedTest `json:"recent"`
|
||||
AggDay map[string]storage.Agg `json:"agg_day"`
|
||||
AggWeek map[string]storage.Agg `json:"agg_week"`
|
||||
AggMonth map[string]storage.Agg `json:"agg_month"`
|
||||
GeneratedAt int64 `json:"generated_at"`
|
||||
}
|
||||
|
||||
type snapshot struct {
|
||||
GeneratedAt string `json:"generated_at"`
|
||||
Nodes map[string]nodeData `json:"nodes"`
|
||||
Speedtest *SpeedtestSnapshot `json:"speedtest,omitempty"`
|
||||
}
|
||||
|
||||
// buildSpeedtestSnapshot collects the last 200 speedtest rows + per-node
|
||||
// medians over 24h/7d/30d windows. Aggregate errors are logged but
|
||||
// non-fatal — the rest of the snapshot still ships.
|
||||
//
|
||||
// Side effect: prunes speedtests older than 30 days. Retention is colocated
|
||||
// with snapshot rebuild because both run on the same timer cadence and the
|
||||
// storage layer already has the DeleteSpeedTestsBefore primitive.
|
||||
func buildSpeedtestSnapshot(ctx context.Context, st *storage.Store) (*SpeedtestSnapshot, error) {
|
||||
if deleted, err := st.DeleteSpeedTestsBefore(ctx, time.Now().Add(-30*24*time.Hour).Unix()); err != nil {
|
||||
log.Printf("speedtests retention (non-fatal): %v", err)
|
||||
} else if deleted > 0 {
|
||||
log.Printf("speedtests retention: pruned %d rows older than 30 days", deleted)
|
||||
}
|
||||
recent, err := st.ListRecentSpeedTests(ctx, 200)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
now := time.Now().Unix()
|
||||
day, week, month := now-86400, now-7*86400, now-30*86400
|
||||
aggDay, errDay := st.AggregateSpeedTestsSince(ctx, day)
|
||||
if errDay != nil {
|
||||
log.Printf("aggregate day: %v", errDay)
|
||||
}
|
||||
aggWeek, errWeek := st.AggregateSpeedTestsSince(ctx, week)
|
||||
if errWeek != nil {
|
||||
log.Printf("aggregate week: %v", errWeek)
|
||||
}
|
||||
aggMonth, errMonth := st.AggregateSpeedTestsSince(ctx, month)
|
||||
if errMonth != nil {
|
||||
log.Printf("aggregate month: %v", errMonth)
|
||||
}
|
||||
return &SpeedtestSnapshot{
|
||||
Recent: recent,
|
||||
AggDay: aggDay,
|
||||
AggWeek: aggWeek,
|
||||
AggMonth: aggMonth,
|
||||
GeneratedAt: now,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func main() {
|
||||
@@ -63,6 +115,12 @@ func main() {
|
||||
s.Nodes[node] = nd
|
||||
}
|
||||
|
||||
if stSnap, err := buildSpeedtestSnapshot(ctx, st); err != nil {
|
||||
log.Printf("build speedtest snapshot: %v", err)
|
||||
} else {
|
||||
s.Speedtest = stSnap
|
||||
}
|
||||
|
||||
if err := os.MkdirAll(filepath.Dir(*out), 0755); err != nil {
|
||||
log.Fatalf("mkdir %s: %v", filepath.Dir(*out), err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"flag"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
pb "github.com/bergabruh/stats-gateway/gen/stats/v1"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
"google.golang.org/grpc/metadata"
|
||||
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
func main() {
|
||||
dbPath := flag.String("source-db", "/var/lib/librespeed-db/speedtest_telemetry.db",
|
||||
"Path to LibreSpeed SQLite (read-only)")
|
||||
statsAddr := flag.String("stats-addr", "stats.projectshitpost.fun:443",
|
||||
"stats-gateway gRPC address")
|
||||
cursorPath := flag.String("cursor", "/var/lib/speedtest-sync/cursor",
|
||||
"File holding the last synced source ID")
|
||||
tokenEnv := flag.String("token-env", "JWT_TOKEN",
|
||||
"Environment variable with JWT token")
|
||||
insecureFlag := flag.Bool("insecure", false, "Use insecure gRPC (testing only)")
|
||||
flag.Parse()
|
||||
|
||||
token := os.Getenv(*tokenEnv)
|
||||
if token == "" {
|
||||
log.Fatalf("env %s is empty", *tokenEnv)
|
||||
}
|
||||
|
||||
cursor := readCursor(*cursorPath)
|
||||
|
||||
db, err := sql.Open("sqlite", "file:"+*dbPath+"?mode=ro")
|
||||
if err != nil {
|
||||
log.Fatalf("open source db: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
rows, err := extractRows(ctx, db, cursor)
|
||||
if err != nil {
|
||||
log.Fatalf("extract: %v", err)
|
||||
}
|
||||
if len(rows) == 0 {
|
||||
log.Printf("no new rows since id=%d", cursor)
|
||||
return
|
||||
}
|
||||
|
||||
var dialOpt grpc.DialOption
|
||||
if *insecureFlag {
|
||||
dialOpt = grpc.WithTransportCredentials(insecure.NewCredentials())
|
||||
} else {
|
||||
dialOpt = grpc.WithTransportCredentials(credentials.NewTLS(nil))
|
||||
}
|
||||
conn, err := grpc.NewClient(*statsAddr, dialOpt)
|
||||
if err != nil {
|
||||
log.Fatalf("grpc dial: %v", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
client := pb.NewSpeedStatusClient(conn)
|
||||
|
||||
authCtx := metadata.AppendToOutgoingContext(ctx, "authorization", "Bearer "+token)
|
||||
sent := 0
|
||||
for _, r := range rows {
|
||||
if err := sendRow(authCtx, client, r); err != nil {
|
||||
log.Printf("send id=%d failed: %v (stopping; cursor not advanced)", r.SourceID, err)
|
||||
break
|
||||
}
|
||||
cursor = r.SourceID
|
||||
sent++
|
||||
}
|
||||
|
||||
if err := writeCursor(*cursorPath, cursor); err != nil {
|
||||
log.Printf("WARN: cursor write failed: %v", err)
|
||||
}
|
||||
log.Printf("synced %d rows, cursor=%d", sent, cursor)
|
||||
}
|
||||
|
||||
func readCursor(path string) int64 {
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
n, err := strconv.ParseInt(string(b), 10, 64)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func writeCursor(path string, id int64) error {
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(path, []byte(strconv.FormatInt(id, 10)), 0o600)
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
|
||||
pb "github.com/bergabruh/stats-gateway/gen/stats/v1"
|
||||
)
|
||||
|
||||
type extractedRow struct {
|
||||
SourceID int64
|
||||
Timestamp int64
|
||||
Node string
|
||||
Country string
|
||||
DlMbps float64
|
||||
UlMbps float64
|
||||
PingMs float64
|
||||
JitterMs float64
|
||||
}
|
||||
|
||||
// extractRows reads rows from the LibreSpeed source DB whose id > sinceID and
|
||||
// returns a privacy-filtered projection. The SELECT intentionally touches only
|
||||
// id, timestamp, extra, dl, ul, ping, jitter — never ip, ispinfo, ua, lang, log.
|
||||
func extractRows(ctx context.Context, db *sql.DB, sinceID int64) ([]extractedRow, error) {
|
||||
rows, err := db.QueryContext(ctx, `
|
||||
SELECT id, strftime('%s', timestamp), extra, dl, ul, ping, jitter
|
||||
FROM speedtest_users
|
||||
WHERE id > ? AND extra IS NOT NULL AND extra != ''
|
||||
ORDER BY id ASC`, sinceID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query source: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := []extractedRow{}
|
||||
for rows.Next() {
|
||||
var (
|
||||
id int64
|
||||
ts sql.NullString
|
||||
extraJSON sql.NullString
|
||||
dl, ul, ping, jitter sql.NullFloat64
|
||||
)
|
||||
if err := rows.Scan(&id, &ts, &extraJSON, &dl, &ul, &ping, &jitter); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var meta struct {
|
||||
Node string `json:"node"`
|
||||
Country string `json:"country"`
|
||||
}
|
||||
if !extraJSON.Valid {
|
||||
continue
|
||||
}
|
||||
if err := json.Unmarshal([]byte(extraJSON.String), &meta); err != nil {
|
||||
log.Printf("skip id=%d: bad extra json: %v", id, err)
|
||||
continue
|
||||
}
|
||||
if meta.Node != "fl" && meta.Node != "pl1" {
|
||||
log.Printf("skip id=%d: bad node %q", id, meta.Node)
|
||||
continue
|
||||
}
|
||||
|
||||
if !ts.Valid {
|
||||
log.Printf("skip id=%d: null timestamp", id)
|
||||
continue
|
||||
}
|
||||
var tsUnix int64
|
||||
if _, err := fmt.Sscanf(ts.String, "%d", &tsUnix); err != nil || tsUnix == 0 {
|
||||
log.Printf("skip id=%d: bad timestamp %q", id, ts.String)
|
||||
continue
|
||||
}
|
||||
|
||||
out = append(out, extractedRow{
|
||||
SourceID: id,
|
||||
Timestamp: tsUnix,
|
||||
Node: meta.Node,
|
||||
Country: nonEmpty(meta.Country, "XX"),
|
||||
DlMbps: dl.Float64,
|
||||
UlMbps: ul.Float64,
|
||||
PingMs: ping.Float64,
|
||||
JitterMs: jitter.Float64,
|
||||
})
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func nonEmpty(s, fallback string) string {
|
||||
if s == "" {
|
||||
return fallback
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func sendRow(ctx context.Context, client pb.SpeedStatusClient, r extractedRow) error {
|
||||
_, err := client.RecordSpeedTest(ctx, &pb.RecordSpeedTestRequest{
|
||||
Timestamp: r.Timestamp,
|
||||
Node: r.Node,
|
||||
Country: r.Country,
|
||||
DlMbps: r.DlMbps,
|
||||
UlMbps: r.UlMbps,
|
||||
PingMs: r.PingMs,
|
||||
JitterMs: r.JitterMs,
|
||||
})
|
||||
return err
|
||||
}
|
||||
@@ -0,0 +1,113 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
func TestExtractRows_FiltersExtraJSON(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
dbPath := filepath.Join(dir, "test.db")
|
||||
db, err := sql.Open("sqlite", dbPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
_, err = db.Exec(`
|
||||
CREATE TABLE speedtest_users (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
timestamp DATETIME, ip TEXT, ispinfo TEXT, extra TEXT,
|
||||
ua TEXT, lang TEXT, dl REAL, ul REAL, ping REAL, jitter REAL, log TEXT
|
||||
)`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
extra, _ := json.Marshal(map[string]string{"node": "fl", "country": "RU"})
|
||||
_, err = db.Exec(`INSERT INTO speedtest_users(timestamp, ip, ispinfo, extra, dl, ul, ping, jitter)
|
||||
VALUES(datetime('now'), '0.0.0.0', '', ?, 100.5, 20.1, 35.2, 2.4)`, string(extra))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
rows, err := extractRows(context.Background(), db, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(rows) != 1 {
|
||||
t.Fatalf("expected 1 row, got %d", len(rows))
|
||||
}
|
||||
if rows[0].Node != "fl" || rows[0].Country != "RU" {
|
||||
t.Errorf("extract mismatch: %+v", rows[0])
|
||||
}
|
||||
if rows[0].DlMbps != 100.5 {
|
||||
t.Errorf("dl mismatch: %v", rows[0].DlMbps)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractRows_SkipsBadExtra(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
dbPath := filepath.Join(dir, "test.db")
|
||||
db, err := sql.Open("sqlite", dbPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
if _, err := db.Exec(`CREATE TABLE speedtest_users (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
timestamp DATETIME, ip TEXT, ispinfo TEXT, extra TEXT,
|
||||
ua TEXT, lang TEXT, dl REAL, ul REAL, ping REAL, jitter REAL, log TEXT)`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := db.Exec(`INSERT INTO speedtest_users(timestamp, extra, dl, ul, ping, jitter)
|
||||
VALUES(datetime('now'), 'not-json', 1, 1, 1, 1)`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
rows, err := extractRows(context.Background(), db, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(rows) != 0 {
|
||||
t.Errorf("expected 0 rows (bad extra skipped), got %d", len(rows))
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractRows_SkipsBadTimestamp(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
dbPath := filepath.Join(dir, "test.db")
|
||||
db, err := sql.Open("sqlite", dbPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
if _, err := db.Exec(`CREATE TABLE speedtest_users (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
timestamp DATETIME, ip TEXT, ispinfo TEXT, extra TEXT,
|
||||
ua TEXT, lang TEXT, dl REAL, ul REAL, ping REAL, jitter REAL, log TEXT)`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
extra, _ := json.Marshal(map[string]string{"node": "fl", "country": "RU"})
|
||||
// NULL timestamp
|
||||
if _, err := db.Exec(`INSERT INTO speedtest_users(timestamp, extra, dl, ul, ping, jitter)
|
||||
VALUES(NULL, ?, 1, 1, 1, 1)`, string(extra)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
rows, err := extractRows(context.Background(), db, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(rows) != 0 {
|
||||
t.Errorf("expected 0 rows (null timestamp skipped), got %d", len(rows))
|
||||
}
|
||||
}
|
||||
+168
-9
@@ -450,6 +450,150 @@ func (x *Frame) GetPadding() []byte {
|
||||
return nil
|
||||
}
|
||||
|
||||
type RecordSpeedTestRequest struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
// Unix epoch seconds. Required.
|
||||
Timestamp int64 `protobuf:"varint,1,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
|
||||
// Node identifier: "fl" or "pl1". Required.
|
||||
Node string `protobuf:"bytes,2,opt,name=node,proto3" json:"node,omitempty"`
|
||||
// ISO-2 country code resolved from client IP at the edge node. "XX" if unknown.
|
||||
Country string `protobuf:"bytes,3,opt,name=country,proto3" json:"country,omitempty"`
|
||||
// Mbps. >= 0.
|
||||
DlMbps float64 `protobuf:"fixed64,4,opt,name=dl_mbps,json=dlMbps,proto3" json:"dl_mbps,omitempty"`
|
||||
// Mbps. >= 0.
|
||||
UlMbps float64 `protobuf:"fixed64,5,opt,name=ul_mbps,json=ulMbps,proto3" json:"ul_mbps,omitempty"`
|
||||
// Milliseconds. >= 0.
|
||||
PingMs float64 `protobuf:"fixed64,6,opt,name=ping_ms,json=pingMs,proto3" json:"ping_ms,omitempty"`
|
||||
// Milliseconds. >= 0.
|
||||
JitterMs float64 `protobuf:"fixed64,7,opt,name=jitter_ms,json=jitterMs,proto3" json:"jitter_ms,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *RecordSpeedTestRequest) Reset() {
|
||||
*x = RecordSpeedTestRequest{}
|
||||
mi := &file_stats_proto_msgTypes[6]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *RecordSpeedTestRequest) String() string {
|
||||
return protoimpl.X.MessageStringOf(x)
|
||||
}
|
||||
|
||||
func (*RecordSpeedTestRequest) ProtoMessage() {}
|
||||
|
||||
func (x *RecordSpeedTestRequest) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_stats_proto_msgTypes[6]
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
return ms
|
||||
}
|
||||
return mi.MessageOf(x)
|
||||
}
|
||||
|
||||
// Deprecated: Use RecordSpeedTestRequest.ProtoReflect.Descriptor instead.
|
||||
func (*RecordSpeedTestRequest) Descriptor() ([]byte, []int) {
|
||||
return file_stats_proto_rawDescGZIP(), []int{6}
|
||||
}
|
||||
|
||||
func (x *RecordSpeedTestRequest) GetTimestamp() int64 {
|
||||
if x != nil {
|
||||
return x.Timestamp
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *RecordSpeedTestRequest) GetNode() string {
|
||||
if x != nil {
|
||||
return x.Node
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *RecordSpeedTestRequest) GetCountry() string {
|
||||
if x != nil {
|
||||
return x.Country
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *RecordSpeedTestRequest) GetDlMbps() float64 {
|
||||
if x != nil {
|
||||
return x.DlMbps
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *RecordSpeedTestRequest) GetUlMbps() float64 {
|
||||
if x != nil {
|
||||
return x.UlMbps
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *RecordSpeedTestRequest) GetPingMs() float64 {
|
||||
if x != nil {
|
||||
return x.PingMs
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *RecordSpeedTestRequest) GetJitterMs() float64 {
|
||||
if x != nil {
|
||||
return x.JitterMs
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
type RecordSpeedTestResponse struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
// Server-assigned monotonic ID for the inserted row.
|
||||
Id int64 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *RecordSpeedTestResponse) Reset() {
|
||||
*x = RecordSpeedTestResponse{}
|
||||
mi := &file_stats_proto_msgTypes[7]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *RecordSpeedTestResponse) String() string {
|
||||
return protoimpl.X.MessageStringOf(x)
|
||||
}
|
||||
|
||||
func (*RecordSpeedTestResponse) ProtoMessage() {}
|
||||
|
||||
func (x *RecordSpeedTestResponse) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_stats_proto_msgTypes[7]
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
return ms
|
||||
}
|
||||
return mi.MessageOf(x)
|
||||
}
|
||||
|
||||
// Deprecated: Use RecordSpeedTestResponse.ProtoReflect.Descriptor instead.
|
||||
func (*RecordSpeedTestResponse) Descriptor() ([]byte, []int) {
|
||||
return file_stats_proto_rawDescGZIP(), []int{7}
|
||||
}
|
||||
|
||||
func (x *RecordSpeedTestResponse) GetId() int64 {
|
||||
if x != nil {
|
||||
return x.Id
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
var File_stats_proto protoreflect.FileDescriptor
|
||||
|
||||
const file_stats_proto_rawDesc = "" +
|
||||
@@ -486,11 +630,22 @@ const file_stats_proto_rawDesc = "" +
|
||||
"\aUNKNOWN\x10\x00\x12\t\n" +
|
||||
"\x05HELLO\x10\x01\x12\b\n" +
|
||||
"\x04DATA\x10\x02\x12\t\n" +
|
||||
"\x05CLOSE\x10\x032\xa7\x01\n" +
|
||||
"\x05CLOSE\x10\x03\"\xcc\x01\n" +
|
||||
"\x16RecordSpeedTestRequest\x12\x1c\n" +
|
||||
"\ttimestamp\x18\x01 \x01(\x03R\ttimestamp\x12\x12\n" +
|
||||
"\x04node\x18\x02 \x01(\tR\x04node\x12\x18\n" +
|
||||
"\acountry\x18\x03 \x01(\tR\acountry\x12\x17\n" +
|
||||
"\adl_mbps\x18\x04 \x01(\x01R\x06dlMbps\x12\x17\n" +
|
||||
"\aul_mbps\x18\x05 \x01(\x01R\x06ulMbps\x12\x17\n" +
|
||||
"\aping_ms\x18\x06 \x01(\x01R\x06pingMs\x12\x1b\n" +
|
||||
"\tjitter_ms\x18\a \x01(\x01R\bjitterMs\")\n" +
|
||||
"\x17RecordSpeedTestResponse\x12\x0e\n" +
|
||||
"\x02id\x18\x01 \x01(\x03R\x02id2\xff\x01\n" +
|
||||
"\vSpeedStatus\x125\n" +
|
||||
"\rRecentResults\x12\x10.stats.v1.Filter\x1a\x10.stats.v1.Sample0\x01\x121\n" +
|
||||
"\tAggregate\x12\x0f.stats.v1.Range\x1a\x13.stats.v1.Histogram\x12.\n" +
|
||||
"\x06Tunnel\x12\x0f.stats.v1.Frame\x1a\x0f.stats.v1.Frame(\x010\x01B9Z7github.com/bergabruh/stats-gateway/gen/stats/v1;statsv1b\x06proto3"
|
||||
"\x06Tunnel\x12\x0f.stats.v1.Frame\x1a\x0f.stats.v1.Frame(\x010\x01\x12V\n" +
|
||||
"\x0fRecordSpeedTest\x12 .stats.v1.RecordSpeedTestRequest\x1a!.stats.v1.RecordSpeedTestResponseB9Z7github.com/bergabruh/stats-gateway/gen/stats/v1;statsv1b\x06proto3"
|
||||
|
||||
var (
|
||||
file_stats_proto_rawDescOnce sync.Once
|
||||
@@ -505,7 +660,7 @@ func file_stats_proto_rawDescGZIP() []byte {
|
||||
}
|
||||
|
||||
var file_stats_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
|
||||
var file_stats_proto_msgTypes = make([]protoimpl.MessageInfo, 6)
|
||||
var file_stats_proto_msgTypes = make([]protoimpl.MessageInfo, 8)
|
||||
var file_stats_proto_goTypes = []any{
|
||||
(Frame_Kind)(0), // 0: stats.v1.Frame.Kind
|
||||
(*Filter)(nil), // 1: stats.v1.Filter
|
||||
@@ -514,6 +669,8 @@ var file_stats_proto_goTypes = []any{
|
||||
(*Histogram)(nil), // 4: stats.v1.Histogram
|
||||
(*Bucket)(nil), // 5: stats.v1.Bucket
|
||||
(*Frame)(nil), // 6: stats.v1.Frame
|
||||
(*RecordSpeedTestRequest)(nil), // 7: stats.v1.RecordSpeedTestRequest
|
||||
(*RecordSpeedTestResponse)(nil), // 8: stats.v1.RecordSpeedTestResponse
|
||||
}
|
||||
var file_stats_proto_depIdxs = []int32{
|
||||
5, // 0: stats.v1.Histogram.buckets:type_name -> stats.v1.Bucket
|
||||
@@ -521,11 +678,13 @@ var file_stats_proto_depIdxs = []int32{
|
||||
1, // 2: stats.v1.SpeedStatus.RecentResults:input_type -> stats.v1.Filter
|
||||
3, // 3: stats.v1.SpeedStatus.Aggregate:input_type -> stats.v1.Range
|
||||
6, // 4: stats.v1.SpeedStatus.Tunnel:input_type -> stats.v1.Frame
|
||||
2, // 5: stats.v1.SpeedStatus.RecentResults:output_type -> stats.v1.Sample
|
||||
4, // 6: stats.v1.SpeedStatus.Aggregate:output_type -> stats.v1.Histogram
|
||||
6, // 7: stats.v1.SpeedStatus.Tunnel:output_type -> stats.v1.Frame
|
||||
5, // [5:8] is the sub-list for method output_type
|
||||
2, // [2:5] is the sub-list for method input_type
|
||||
7, // 5: stats.v1.SpeedStatus.RecordSpeedTest:input_type -> stats.v1.RecordSpeedTestRequest
|
||||
2, // 6: stats.v1.SpeedStatus.RecentResults:output_type -> stats.v1.Sample
|
||||
4, // 7: stats.v1.SpeedStatus.Aggregate:output_type -> stats.v1.Histogram
|
||||
6, // 8: stats.v1.SpeedStatus.Tunnel:output_type -> stats.v1.Frame
|
||||
8, // 9: stats.v1.SpeedStatus.RecordSpeedTest:output_type -> stats.v1.RecordSpeedTestResponse
|
||||
6, // [6:10] is the sub-list for method output_type
|
||||
2, // [2:6] is the sub-list for method input_type
|
||||
2, // [2:2] is the sub-list for extension type_name
|
||||
2, // [2:2] is the sub-list for extension extendee
|
||||
0, // [0:2] is the sub-list for field type_name
|
||||
@@ -542,7 +701,7 @@ func file_stats_proto_init() {
|
||||
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
|
||||
RawDescriptor: unsafe.Slice(unsafe.StringData(file_stats_proto_rawDesc), len(file_stats_proto_rawDesc)),
|
||||
NumEnums: 1,
|
||||
NumMessages: 6,
|
||||
NumMessages: 8,
|
||||
NumExtensions: 0,
|
||||
NumServices: 1,
|
||||
},
|
||||
|
||||
@@ -22,6 +22,7 @@ const (
|
||||
SpeedStatus_RecentResults_FullMethodName = "/stats.v1.SpeedStatus/RecentResults"
|
||||
SpeedStatus_Aggregate_FullMethodName = "/stats.v1.SpeedStatus/Aggregate"
|
||||
SpeedStatus_Tunnel_FullMethodName = "/stats.v1.SpeedStatus/Tunnel"
|
||||
SpeedStatus_RecordSpeedTest_FullMethodName = "/stats.v1.SpeedStatus/RecordSpeedTest"
|
||||
)
|
||||
|
||||
// SpeedStatusClient is the client API for SpeedStatus service.
|
||||
@@ -36,6 +37,8 @@ type SpeedStatusClient interface {
|
||||
// Auth: Bearer JWT (HS256) в gRPC metadata "authorization: Bearer <jwt>".
|
||||
// Server validates: signature, exp, scope=="tunnel", optional revocation list.
|
||||
Tunnel(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[Frame, Frame], error)
|
||||
// RecordSpeedTest ingests a single speedtest sample from an edge node.
|
||||
RecordSpeedTest(ctx context.Context, in *RecordSpeedTestRequest, opts ...grpc.CallOption) (*RecordSpeedTestResponse, error)
|
||||
}
|
||||
|
||||
type speedStatusClient struct {
|
||||
@@ -88,6 +91,16 @@ func (c *speedStatusClient) Tunnel(ctx context.Context, opts ...grpc.CallOption)
|
||||
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
|
||||
type SpeedStatus_TunnelClient = grpc.BidiStreamingClient[Frame, Frame]
|
||||
|
||||
func (c *speedStatusClient) RecordSpeedTest(ctx context.Context, in *RecordSpeedTestRequest, opts ...grpc.CallOption) (*RecordSpeedTestResponse, error) {
|
||||
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
|
||||
out := new(RecordSpeedTestResponse)
|
||||
err := c.cc.Invoke(ctx, SpeedStatus_RecordSpeedTest_FullMethodName, in, out, cOpts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// SpeedStatusServer is the server API for SpeedStatus service.
|
||||
// All implementations must embed UnimplementedSpeedStatusServer
|
||||
// for forward compatibility.
|
||||
@@ -100,6 +113,8 @@ type SpeedStatusServer interface {
|
||||
// Auth: Bearer JWT (HS256) в gRPC metadata "authorization: Bearer <jwt>".
|
||||
// Server validates: signature, exp, scope=="tunnel", optional revocation list.
|
||||
Tunnel(grpc.BidiStreamingServer[Frame, Frame]) error
|
||||
// RecordSpeedTest ingests a single speedtest sample from an edge node.
|
||||
RecordSpeedTest(context.Context, *RecordSpeedTestRequest) (*RecordSpeedTestResponse, error)
|
||||
mustEmbedUnimplementedSpeedStatusServer()
|
||||
}
|
||||
|
||||
@@ -119,6 +134,9 @@ func (UnimplementedSpeedStatusServer) Aggregate(context.Context, *Range) (*Histo
|
||||
func (UnimplementedSpeedStatusServer) Tunnel(grpc.BidiStreamingServer[Frame, Frame]) error {
|
||||
return status.Error(codes.Unimplemented, "method Tunnel not implemented")
|
||||
}
|
||||
func (UnimplementedSpeedStatusServer) RecordSpeedTest(context.Context, *RecordSpeedTestRequest) (*RecordSpeedTestResponse, error) {
|
||||
return nil, status.Error(codes.Unimplemented, "method RecordSpeedTest not implemented")
|
||||
}
|
||||
func (UnimplementedSpeedStatusServer) mustEmbedUnimplementedSpeedStatusServer() {}
|
||||
func (UnimplementedSpeedStatusServer) testEmbeddedByValue() {}
|
||||
|
||||
@@ -176,6 +194,24 @@ func _SpeedStatus_Tunnel_Handler(srv interface{}, stream grpc.ServerStream) erro
|
||||
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
|
||||
type SpeedStatus_TunnelServer = grpc.BidiStreamingServer[Frame, Frame]
|
||||
|
||||
func _SpeedStatus_RecordSpeedTest_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
|
||||
in := new(RecordSpeedTestRequest)
|
||||
if err := dec(in); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if interceptor == nil {
|
||||
return srv.(SpeedStatusServer).RecordSpeedTest(ctx, in)
|
||||
}
|
||||
info := &grpc.UnaryServerInfo{
|
||||
Server: srv,
|
||||
FullMethod: SpeedStatus_RecordSpeedTest_FullMethodName,
|
||||
}
|
||||
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
|
||||
return srv.(SpeedStatusServer).RecordSpeedTest(ctx, req.(*RecordSpeedTestRequest))
|
||||
}
|
||||
return interceptor(ctx, in, info, handler)
|
||||
}
|
||||
|
||||
// SpeedStatus_ServiceDesc is the grpc.ServiceDesc for SpeedStatus service.
|
||||
// It's only intended for direct use with grpc.RegisterService,
|
||||
// and not to be introspected or modified (even as a copy)
|
||||
@@ -187,6 +223,10 @@ var SpeedStatus_ServiceDesc = grpc.ServiceDesc{
|
||||
MethodName: "Aggregate",
|
||||
Handler: _SpeedStatus_Aggregate_Handler,
|
||||
},
|
||||
{
|
||||
MethodName: "RecordSpeedTest",
|
||||
Handler: _SpeedStatus_RecordSpeedTest_Handler,
|
||||
},
|
||||
},
|
||||
Streams: []grpc.StreamDesc{
|
||||
{
|
||||
|
||||
@@ -11,8 +11,17 @@ require (
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/dustin/go-humanize v1.0.1 // indirect
|
||||
github.com/google/uuid v1.6.0 // indirect
|
||||
github.com/mattn/go-isatty v0.0.20 // indirect
|
||||
github.com/ncruces/go-strftime v1.0.0 // indirect
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
|
||||
golang.org/x/net v0.51.0 // indirect
|
||||
golang.org/x/sys v0.42.0 // indirect
|
||||
golang.org/x/text v0.34.0 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171 // indirect
|
||||
modernc.org/libc v1.72.0 // indirect
|
||||
modernc.org/mathutil v1.7.1 // indirect
|
||||
modernc.org/memory v1.11.0 // indirect
|
||||
modernc.org/sqlite v1.50.0 // indirect
|
||||
)
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
|
||||
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
|
||||
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
|
||||
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
|
||||
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
|
||||
@@ -10,8 +12,14 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
|
||||
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
|
||||
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
|
||||
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
|
||||
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
|
||||
github.com/mattn/go-sqlite3 v1.14.44 h1:3VSe+xafpbzsLbdr2AWlAZk9yRHiBhTBakioXaCKTF8=
|
||||
github.com/mattn/go-sqlite3 v1.14.44/go.mod h1:pjEuOr8IwzLJP2MfGeTb0A35jauH+C2kbHKBr7yXKVQ=
|
||||
github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
|
||||
github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
|
||||
go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
|
||||
go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
|
||||
go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I=
|
||||
@@ -26,6 +34,7 @@ go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09
|
||||
go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0=
|
||||
golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo=
|
||||
golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y=
|
||||
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
|
||||
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk=
|
||||
@@ -44,3 +53,11 @@ gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
modernc.org/libc v1.72.0 h1:IEu559v9a0XWjw0DPoVKtXpO2qt5NVLAnFaBbjq+n8c=
|
||||
modernc.org/libc v1.72.0/go.mod h1:tTU8DL8A+XLVkEY3x5E/tO7s2Q/q42EtnNWda/L5QhQ=
|
||||
modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
|
||||
modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
|
||||
modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI=
|
||||
modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
|
||||
modernc.org/sqlite v1.50.0 h1:eMowQSWLK0MeiQTdmz3lqoF5dqclujdlIKeJA11+7oM=
|
||||
modernc.org/sqlite v1.50.0/go.mod h1:m0w8xhwYUVY3H6pSDwc3gkJ/irZT/0YEXwBlhaxQEew=
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/metadata"
|
||||
"google.golang.org/grpc/status"
|
||||
|
||||
statsv1 "github.com/bergabruh/stats-gateway/gen/stats/v1"
|
||||
"github.com/bergabruh/stats-gateway/internal/storage"
|
||||
)
|
||||
|
||||
// RecordSpeedTest ingests a single client-side speedtest sample from an edge
|
||||
// node. Authenticates via Bearer JWT (same scheme as Tunnel), validates the
|
||||
// request, and persists via the storage layer.
|
||||
func (s *Service) RecordSpeedTest(ctx context.Context, req *statsv1.RecordSpeedTestRequest) (*statsv1.RecordSpeedTestResponse, error) {
|
||||
// 1. JWT auth — mirrors the Tunnel handler.
|
||||
md, ok := metadata.FromIncomingContext(ctx)
|
||||
if !ok {
|
||||
return nil, status.Error(codes.Unauthenticated, "no metadata")
|
||||
}
|
||||
authH := md.Get("authorization")
|
||||
if len(authH) == 0 {
|
||||
return nil, status.Error(codes.Unauthenticated, "no auth")
|
||||
}
|
||||
tok := strings.TrimPrefix(authH[0], "Bearer ")
|
||||
if tok == authH[0] {
|
||||
return nil, status.Error(codes.Unauthenticated, "expected Bearer")
|
||||
}
|
||||
if _, err := s.verifier.Verify(tok); err != nil {
|
||||
return nil, status.Errorf(codes.Unauthenticated, "auth: %v", err)
|
||||
}
|
||||
|
||||
// 2. Input validation.
|
||||
if req.Node != "fl" && req.Node != "pl1" {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "node: must be 'fl' or 'pl1', got %q", req.Node)
|
||||
}
|
||||
if !validCountry(req.Country) {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "country: must match ^[A-Z]{2}$ or 'XX', got %q", req.Country)
|
||||
}
|
||||
if req.DlMbps < 0 {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "dl_mbps: must be >= 0, got %v", req.DlMbps)
|
||||
}
|
||||
if req.UlMbps < 0 {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "ul_mbps: must be >= 0, got %v", req.UlMbps)
|
||||
}
|
||||
if req.PingMs < 0 {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "ping_ms: must be >= 0, got %v", req.PingMs)
|
||||
}
|
||||
if req.JitterMs < 0 {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "jitter_ms: must be >= 0, got %v", req.JitterMs)
|
||||
}
|
||||
if req.Timestamp <= 0 {
|
||||
return nil, status.Errorf(codes.InvalidArgument, "timestamp: must be > 0, got %d", req.Timestamp)
|
||||
}
|
||||
|
||||
// 3. Persist.
|
||||
id, err := s.store.SaveSpeedTest(ctx, storage.SpeedTest{
|
||||
Timestamp: req.Timestamp,
|
||||
Node: req.Node,
|
||||
Country: req.Country,
|
||||
DlMbps: req.DlMbps,
|
||||
UlMbps: req.UlMbps,
|
||||
PingMs: req.PingMs,
|
||||
JitterMs: req.JitterMs,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, status.Errorf(codes.Internal, "save: %v", err)
|
||||
}
|
||||
return &statsv1.RecordSpeedTestResponse{Id: id}, nil
|
||||
}
|
||||
|
||||
// validCountry returns true when c is two uppercase ASCII letters (ISO-3166-1
|
||||
// alpha-2 shape) or the literal "XX" sentinel for unknown.
|
||||
func validCountry(c string) bool {
|
||||
if len(c) != 2 {
|
||||
return false
|
||||
}
|
||||
if c == "XX" {
|
||||
return true
|
||||
}
|
||||
for i := 0; i < 2; i++ {
|
||||
if c[i] < 'A' || c[i] > 'Z' {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/metadata"
|
||||
"google.golang.org/grpc/status"
|
||||
|
||||
statsv1 "github.com/bergabruh/stats-gateway/gen/stats/v1"
|
||||
"github.com/bergabruh/stats-gateway/internal/auth"
|
||||
"github.com/bergabruh/stats-gateway/internal/storage"
|
||||
)
|
||||
|
||||
// authedCtx returns a context with a valid Bearer JWT for the given subject.
|
||||
// Uses the same secret as newTestServer ("test-secret") and scope "tunnel"
|
||||
// (the only scope accepted by *auth.Verifier.Verify).
|
||||
func authedCtx(t *testing.T, sub string) context.Context {
|
||||
t.Helper()
|
||||
tok, err := auth.Issue([]byte("test-secret"), sub, "tunnel", time.Minute)
|
||||
if err != nil {
|
||||
t.Fatalf("issue token: %v", err)
|
||||
}
|
||||
return metadata.AppendToOutgoingContext(context.Background(), "authorization", "Bearer "+tok)
|
||||
}
|
||||
|
||||
func newRecordSpeedTestClient(t *testing.T) (statsv1.SpeedStatusClient, func()) {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
st, err := storage.Open(filepath.Join(dir, "test.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("storage.Open: %v", err)
|
||||
}
|
||||
client, cleanup := newTestServer(t, st)
|
||||
return client, func() { cleanup(); st.Close() }
|
||||
}
|
||||
|
||||
func TestRecordSpeedTest_HappyPath(t *testing.T) {
|
||||
client, cleanup := newRecordSpeedTestClient(t)
|
||||
defer cleanup()
|
||||
|
||||
ctx := authedCtx(t, "fl-speedtest-sync")
|
||||
resp, err := client.RecordSpeedTest(ctx, &statsv1.RecordSpeedTestRequest{
|
||||
Timestamp: time.Now().Unix(),
|
||||
Node: "fl",
|
||||
Country: "RU",
|
||||
DlMbps: 100, UlMbps: 20, PingMs: 35, JitterMs: 2,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("RecordSpeedTest: %v", err)
|
||||
}
|
||||
if resp.Id <= 0 {
|
||||
t.Fatalf("expected id>0, got %d", resp.Id)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecordSpeedTest_RejectsBadNode(t *testing.T) {
|
||||
client, cleanup := newRecordSpeedTestClient(t)
|
||||
defer cleanup()
|
||||
|
||||
ctx := authedCtx(t, "fl-speedtest-sync")
|
||||
_, err := client.RecordSpeedTest(ctx, &statsv1.RecordSpeedTestRequest{
|
||||
Timestamp: time.Now().Unix(),
|
||||
Node: "moon", Country: "RU",
|
||||
DlMbps: 1, UlMbps: 1, PingMs: 1, JitterMs: 1,
|
||||
})
|
||||
if err == nil {
|
||||
t.Fatalf("expected InvalidArgument for bad node, got nil")
|
||||
}
|
||||
if st, ok := status.FromError(err); !ok || st.Code() != codes.InvalidArgument {
|
||||
t.Fatalf("expected InvalidArgument, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecordSpeedTest_RejectsNegativeMbps(t *testing.T) {
|
||||
client, cleanup := newRecordSpeedTestClient(t)
|
||||
defer cleanup()
|
||||
|
||||
ctx := authedCtx(t, "fl-speedtest-sync")
|
||||
_, err := client.RecordSpeedTest(ctx, &statsv1.RecordSpeedTestRequest{
|
||||
Timestamp: time.Now().Unix(), Node: "fl", Country: "RU",
|
||||
DlMbps: -1, UlMbps: 1, PingMs: 1, JitterMs: 1,
|
||||
})
|
||||
if err == nil {
|
||||
t.Fatalf("expected InvalidArgument for negative dl_mbps, got nil")
|
||||
}
|
||||
if st, ok := status.FromError(err); !ok || st.Code() != codes.InvalidArgument {
|
||||
t.Fatalf("expected InvalidArgument, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecordSpeedTest_RequiresAuth(t *testing.T) {
|
||||
client, cleanup := newRecordSpeedTestClient(t)
|
||||
defer cleanup()
|
||||
|
||||
_, err := client.RecordSpeedTest(context.Background(), &statsv1.RecordSpeedTestRequest{
|
||||
Timestamp: time.Now().Unix(), Node: "fl", Country: "RU",
|
||||
DlMbps: 1, UlMbps: 1, PingMs: 1, JitterMs: 1,
|
||||
})
|
||||
if err == nil {
|
||||
t.Fatalf("expected Unauthenticated, got nil")
|
||||
}
|
||||
if st, ok := status.FromError(err); !ok || st.Code() != codes.Unauthenticated {
|
||||
t.Fatalf("expected Unauthenticated, got %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
CREATE TABLE IF NOT EXISTS speedtests (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
timestamp INTEGER NOT NULL,
|
||||
node TEXT NOT NULL CHECK(node IN ('fl', 'pl1')),
|
||||
country TEXT NOT NULL,
|
||||
dl_mbps REAL NOT NULL CHECK(dl_mbps >= 0),
|
||||
ul_mbps REAL NOT NULL CHECK(ul_mbps >= 0),
|
||||
ping_ms REAL NOT NULL CHECK(ping_ms >= 0),
|
||||
jitter_ms REAL NOT NULL CHECK(jitter_ms >= 0)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_speedtests_ts ON speedtests(timestamp);
|
||||
CREATE INDEX IF NOT EXISTS idx_speedtests_node ON speedtests(node);
|
||||
@@ -0,0 +1,136 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
)
|
||||
|
||||
// Agg holds median aggregates for a single node over an arbitrary time
|
||||
// window. Counts and medians are computed Go-side (sort + middle pick)
|
||||
// because SQLite has no native percentile aggregate.
|
||||
type Agg struct {
|
||||
Count int64 `json:"count"`
|
||||
MedianDl float64 `json:"median_dl"`
|
||||
MedianUl float64 `json:"median_ul"`
|
||||
MedianPing float64 `json:"median_ping"`
|
||||
}
|
||||
|
||||
// SpeedTest is a single client-side speedtest result reported via the
|
||||
// RecordSpeedTest RPC. Timestamp is unix seconds. Node is 'fl' or 'pl1'
|
||||
// (DB-level CHECK). Country is ISO-3166-1 alpha-2 (validated at the
|
||||
// gRPC handler, not at the DB).
|
||||
type SpeedTest struct {
|
||||
ID int64
|
||||
Timestamp int64
|
||||
Node string
|
||||
Country string
|
||||
DlMbps float64
|
||||
UlMbps float64
|
||||
PingMs float64
|
||||
JitterMs float64
|
||||
}
|
||||
|
||||
// SaveSpeedTest inserts a speedtest row and returns the new id.
|
||||
func (s *Store) SaveSpeedTest(ctx context.Context, st SpeedTest) (int64, error) {
|
||||
res, err := s.db.ExecContext(ctx, `
|
||||
INSERT INTO speedtests(timestamp, node, country, dl_mbps, ul_mbps, ping_ms, jitter_ms)
|
||||
VALUES(?, ?, ?, ?, ?, ?, ?)`,
|
||||
st.Timestamp, st.Node, st.Country, st.DlMbps, st.UlMbps, st.PingMs, st.JitterMs)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("insert speedtest: %w", err)
|
||||
}
|
||||
return res.LastInsertId()
|
||||
}
|
||||
|
||||
// ListRecentSpeedTests returns up to limit speedtests ordered by id DESC
|
||||
// (newest first).
|
||||
func (s *Store) ListRecentSpeedTests(ctx context.Context, limit int) ([]SpeedTest, error) {
|
||||
rows, err := s.db.QueryContext(ctx, `
|
||||
SELECT id, timestamp, node, country, dl_mbps, ul_mbps, ping_ms, jitter_ms
|
||||
FROM speedtests ORDER BY id DESC LIMIT ?`, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := []SpeedTest{}
|
||||
for rows.Next() {
|
||||
var st SpeedTest
|
||||
if err := rows.Scan(&st.ID, &st.Timestamp, &st.Node, &st.Country,
|
||||
&st.DlMbps, &st.UlMbps, &st.PingMs, &st.JitterMs); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, st)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// DeleteSpeedTestsBefore deletes rows with timestamp < ts and returns the
|
||||
// number of rows removed. Used by the retention sweep (Task 14).
|
||||
func (s *Store) DeleteSpeedTestsBefore(ctx context.Context, ts int64) (int64, error) {
|
||||
res, err := s.db.ExecContext(ctx, `DELETE FROM speedtests WHERE timestamp < ?`, ts)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return res.RowsAffected()
|
||||
}
|
||||
|
||||
// AggregateSpeedTestsSince returns per-node median aggregates of dl/ul/ping
|
||||
// for rows with timestamp >= since. The result is keyed by node ("fl"/"pl1").
|
||||
// Reads only metric columns — no privacy-sensitive data (country, jitter
|
||||
// not used here).
|
||||
func (s *Store) AggregateSpeedTestsSince(ctx context.Context, since int64) (map[string]Agg, error) {
|
||||
rows, err := s.db.QueryContext(ctx, `
|
||||
SELECT node, dl_mbps, ul_mbps, ping_ms FROM speedtests
|
||||
WHERE timestamp >= ? ORDER BY node`, since)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
type bucket struct{ dl, ul, ping []float64 }
|
||||
buckets := map[string]*bucket{}
|
||||
for rows.Next() {
|
||||
var node string
|
||||
var dl, ul, ping float64
|
||||
if err := rows.Scan(&node, &dl, &ul, &ping); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
b, ok := buckets[node]
|
||||
if !ok {
|
||||
b = &bucket{}
|
||||
buckets[node] = b
|
||||
}
|
||||
b.dl = append(b.dl, dl)
|
||||
b.ul = append(b.ul, ul)
|
||||
b.ping = append(b.ping, ping)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
out := map[string]Agg{}
|
||||
for node, b := range buckets {
|
||||
out[node] = Agg{
|
||||
Count: int64(len(b.dl)),
|
||||
MedianDl: median(b.dl),
|
||||
MedianUl: median(b.ul),
|
||||
MedianPing: median(b.ping),
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// median sorts xs in place and returns the middle value (or mean of the
|
||||
// two middle values if even-length). Returns 0 for empty input.
|
||||
func median(xs []float64) float64 {
|
||||
if len(xs) == 0 {
|
||||
return 0
|
||||
}
|
||||
sort.Float64s(xs)
|
||||
mid := len(xs) / 2
|
||||
if len(xs)%2 == 0 {
|
||||
return (xs[mid-1] + xs[mid]) / 2
|
||||
}
|
||||
return xs[mid]
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// newSpeedTestDB mirrors the open pattern from sqlite_test.go since the
|
||||
// storage package has no dedicated test-DB helper — every test calls
|
||||
// Open(t.TempDir()/...) directly.
|
||||
func newSpeedTestDB(t *testing.T) *Store {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
st, err := Open(filepath.Join(dir, "test.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("open: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { st.Close() })
|
||||
return st
|
||||
}
|
||||
|
||||
func TestSpeedTest_SaveAndList(t *testing.T) {
|
||||
db := newSpeedTestDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
now := time.Now().Unix()
|
||||
id, err := db.SaveSpeedTest(ctx, SpeedTest{
|
||||
Timestamp: now, Node: "fl", Country: "RU",
|
||||
DlMbps: 100.5, UlMbps: 20.1, PingMs: 35.2, JitterMs: 2.4,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("SaveSpeedTest: %v", err)
|
||||
}
|
||||
if id <= 0 {
|
||||
t.Fatalf("expected positive id, got %d", id)
|
||||
}
|
||||
|
||||
rows, err := db.ListRecentSpeedTests(ctx, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("ListRecentSpeedTests: %v", err)
|
||||
}
|
||||
if len(rows) != 1 {
|
||||
t.Fatalf("expected 1 row, got %d", len(rows))
|
||||
}
|
||||
if rows[0].Country != "RU" || rows[0].Node != "fl" {
|
||||
t.Errorf("row mismatch: %+v", rows[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestSpeedTest_RetentionDelete(t *testing.T) {
|
||||
db := newSpeedTestDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
old := time.Now().Add(-31 * 24 * time.Hour).Unix()
|
||||
fresh := time.Now().Unix()
|
||||
|
||||
for _, ts := range []int64{old, fresh} {
|
||||
if _, err := db.SaveSpeedTest(ctx, SpeedTest{
|
||||
Timestamp: ts, Node: "fl", Country: "RU",
|
||||
DlMbps: 1, UlMbps: 1, PingMs: 1, JitterMs: 1,
|
||||
}); err != nil {
|
||||
t.Fatalf("save: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
deleted, err := db.DeleteSpeedTestsBefore(ctx, time.Now().Add(-30*24*time.Hour).Unix())
|
||||
if err != nil {
|
||||
t.Fatalf("DeleteSpeedTestsBefore: %v", err)
|
||||
}
|
||||
if deleted != 1 {
|
||||
t.Errorf("expected 1 deleted row, got %d", deleted)
|
||||
}
|
||||
|
||||
rows, _ := db.ListRecentSpeedTests(ctx, 10)
|
||||
if len(rows) != 1 {
|
||||
t.Errorf("expected 1 remaining row, got %d", len(rows))
|
||||
}
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"database/sql"
|
||||
"embed"
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/bergabruh/stats-gateway/internal/probes"
|
||||
@@ -42,13 +43,32 @@ func Open(path string) (*Store, error) {
|
||||
// Close closes the underlying database.
|
||||
func (s *Store) Close() error { return s.db.Close() }
|
||||
|
||||
// migrate applies every SQL file under migrations/ in lexical filename order.
|
||||
// Each file is executed wholesale (statements are CREATE ... IF NOT EXISTS,
|
||||
// so re-applying on an existing DB is a no-op).
|
||||
func (s *Store) migrate() error {
|
||||
body, err := migrations.ReadFile("migrations/001_init.sql")
|
||||
entries, err := migrations.ReadDir("migrations")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = s.db.Exec(string(body))
|
||||
return err
|
||||
names := make([]string, 0, len(entries))
|
||||
for _, e := range entries {
|
||||
if e.IsDir() {
|
||||
continue
|
||||
}
|
||||
names = append(names, e.Name())
|
||||
}
|
||||
sort.Strings(names)
|
||||
for _, name := range names {
|
||||
body, err := migrations.ReadFile("migrations/" + name)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read migration %s: %w", name, err)
|
||||
}
|
||||
if _, err := s.db.Exec(string(body)); err != nil {
|
||||
return fmt.Errorf("apply migration %s: %w", name, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// InsertSample stores a probe result for the given node.
|
||||
|
||||
@@ -14,6 +14,9 @@ service SpeedStatus {
|
||||
// Auth: Bearer JWT (HS256) в gRPC metadata "authorization: Bearer <jwt>".
|
||||
// Server validates: signature, exp, scope=="tunnel", optional revocation list.
|
||||
rpc Tunnel(stream Frame) returns (stream Frame);
|
||||
|
||||
// RecordSpeedTest ingests a single speedtest sample from an edge node.
|
||||
rpc RecordSpeedTest(RecordSpeedTestRequest) returns (RecordSpeedTestResponse);
|
||||
}
|
||||
|
||||
message Filter {
|
||||
@@ -60,3 +63,25 @@ message Frame {
|
||||
bytes payload = 3; // raw TCP bytes
|
||||
bytes padding = 4; // random padding для frame-length obfuscation
|
||||
}
|
||||
|
||||
message RecordSpeedTestRequest {
|
||||
// Unix epoch seconds. Required.
|
||||
int64 timestamp = 1;
|
||||
// Node identifier: "fl" or "pl1". Required.
|
||||
string node = 2;
|
||||
// ISO-2 country code resolved from client IP at the edge node. "XX" if unknown.
|
||||
string country = 3;
|
||||
// Mbps. >= 0.
|
||||
double dl_mbps = 4;
|
||||
// Mbps. >= 0.
|
||||
double ul_mbps = 5;
|
||||
// Milliseconds. >= 0.
|
||||
double ping_ms = 6;
|
||||
// Milliseconds. >= 0.
|
||||
double jitter_ms = 7;
|
||||
}
|
||||
|
||||
message RecordSpeedTestResponse {
|
||||
// Server-assigned monotonic ID for the inserted row.
|
||||
int64 id = 1;
|
||||
}
|
||||
|
||||
@@ -40,6 +40,7 @@
|
||||
</section>
|
||||
|
||||
<nav class="footer-nav">
|
||||
<a href="/speedtest/">Speedtest</a>
|
||||
<a href="/about/">About</a>
|
||||
</nav>
|
||||
</main>
|
||||
|
||||
@@ -0,0 +1,186 @@
|
||||
<!DOCTYPE html>
|
||||
<html lang="ru">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||
<title>Speedtest — PSP Speedtest Stats</title>
|
||||
<link rel="stylesheet" href="/css/main.css">
|
||||
<style>
|
||||
.speedtest table {
|
||||
width: 100%;
|
||||
border-collapse: collapse;
|
||||
margin-top: 1rem;
|
||||
font-variant-numeric: tabular-nums;
|
||||
}
|
||||
.speedtest th,
|
||||
.speedtest td {
|
||||
padding: 0.4rem 0.7rem;
|
||||
border-bottom: 1px solid var(--border);
|
||||
text-align: left;
|
||||
}
|
||||
.speedtest .muted {
|
||||
color: var(--muted);
|
||||
font-size: 0.9rem;
|
||||
}
|
||||
.speedtest .filters {
|
||||
margin: 1rem 0;
|
||||
color: var(--muted);
|
||||
}
|
||||
.speedtest .filters select,
|
||||
.speedtest .filters input {
|
||||
color: var(--fg);
|
||||
background: var(--bg);
|
||||
border: 1px solid var(--border);
|
||||
border-radius: 4px;
|
||||
padding: 0.2rem 0.4rem;
|
||||
font: inherit;
|
||||
}
|
||||
.speedtest h2 {
|
||||
font-size: 1.1em;
|
||||
margin-top: 2em;
|
||||
}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<header>
|
||||
<h1>Speedtest</h1>
|
||||
</header>
|
||||
<main>
|
||||
<section class="speedtest">
|
||||
<p class="muted">Updated <span id="ts">…</span>. Retention 30 days.</p>
|
||||
|
||||
<h2>Last 24h / 7d / 30d (median)</h2>
|
||||
<table class="agg">
|
||||
<thead><tr><th>Node</th><th>Period</th><th>Count</th><th>↓ Mbps</th><th>↑ Mbps</th><th>Ping ms</th></tr></thead>
|
||||
<tbody id="agg-tbody"></tbody>
|
||||
</table>
|
||||
|
||||
<h2>Recent tests (last 200)</h2>
|
||||
<div class="filters">
|
||||
Node:
|
||||
<select id="filter-node">
|
||||
<option value="">all</option>
|
||||
<option value="fl">fl</option>
|
||||
<option value="pl1">pl1</option>
|
||||
</select>
|
||||
Country:
|
||||
<input type="text" id="filter-country" maxlength="2" placeholder="RU" style="width: 4rem;">
|
||||
</div>
|
||||
<table class="recent">
|
||||
<thead><tr>
|
||||
<th>Time</th><th>Node</th><th>Country</th>
|
||||
<th>↓ Mbps</th><th>↑ Mbps</th><th>Ping ms</th><th>Jitter ms</th>
|
||||
</tr></thead>
|
||||
<tbody id="recent-tbody"></tbody>
|
||||
</table>
|
||||
</section>
|
||||
|
||||
<nav class="footer-nav">
|
||||
<a href="/">← Назад</a>
|
||||
<a href="/about/">About</a>
|
||||
</nav>
|
||||
</main>
|
||||
|
||||
<script>
|
||||
function clearChildren(el) {
|
||||
while (el.firstChild) el.removeChild(el.firstChild);
|
||||
}
|
||||
|
||||
function appendCell(row, text) {
|
||||
const td = document.createElement('td');
|
||||
td.textContent = text;
|
||||
row.appendChild(td);
|
||||
}
|
||||
|
||||
function fmtNumber(n, fixed) {
|
||||
if (typeof n !== 'number' || !isFinite(n)) return '0';
|
||||
return n.toFixed(fixed);
|
||||
}
|
||||
|
||||
function fmtTimestamp(unixSeconds) {
|
||||
if (typeof unixSeconds !== 'number') return '';
|
||||
return new Date(unixSeconds * 1000).toISOString().replace('T', ' ').slice(0, 19);
|
||||
}
|
||||
|
||||
function renderAggTable(snapshot) {
|
||||
const aggBody = document.getElementById('agg-tbody');
|
||||
clearChildren(aggBody);
|
||||
const periods = [
|
||||
{ key: 'agg_day', label: '24h' },
|
||||
{ key: 'agg_week', label: '7d' },
|
||||
{ key: 'agg_month', label: '30d' },
|
||||
];
|
||||
for (const { key, label } of periods) {
|
||||
const m = snapshot[key] || {};
|
||||
for (const node of ['fl', 'pl1']) {
|
||||
const a = m[node];
|
||||
if (!a) continue;
|
||||
const tr = document.createElement('tr');
|
||||
appendCell(tr, node);
|
||||
appendCell(tr, label);
|
||||
appendCell(tr, String(a.count | 0));
|
||||
appendCell(tr, fmtNumber(a.median_dl, 1));
|
||||
appendCell(tr, fmtNumber(a.median_ul, 1));
|
||||
appendCell(tr, fmtNumber(a.median_ping, 1));
|
||||
aggBody.appendChild(tr);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function renderRecentTable(rows) {
|
||||
const body = document.getElementById('recent-tbody');
|
||||
clearChildren(body);
|
||||
const limited = rows.slice(0, 200);
|
||||
for (const r of limited) {
|
||||
const tr = document.createElement('tr');
|
||||
appendCell(tr, fmtTimestamp(r.Timestamp));
|
||||
appendCell(tr, String(r.Node || ''));
|
||||
appendCell(tr, String(r.Country || ''));
|
||||
appendCell(tr, fmtNumber(r.DlMbps, 1));
|
||||
appendCell(tr, fmtNumber(r.UlMbps, 1));
|
||||
appendCell(tr, fmtNumber(r.PingMs, 1));
|
||||
appendCell(tr, fmtNumber(r.JitterMs, 1));
|
||||
body.appendChild(tr);
|
||||
}
|
||||
}
|
||||
|
||||
async function loadSnapshot() {
|
||||
let snapshot;
|
||||
try {
|
||||
const r = await fetch('/data/snapshot.json', { cache: 'no-store' });
|
||||
snapshot = await r.json();
|
||||
} catch (e) {
|
||||
document.getElementById('ts').textContent = 'load error';
|
||||
return;
|
||||
}
|
||||
const st = snapshot.speedtest;
|
||||
if (!st) {
|
||||
document.getElementById('ts').textContent = 'no data';
|
||||
return;
|
||||
}
|
||||
|
||||
document.getElementById('ts').textContent = fmtTimestamp(st.generated_at);
|
||||
renderAggTable(st);
|
||||
|
||||
const recent = Array.isArray(st.recent) ? st.recent : [];
|
||||
const filterNode = document.getElementById('filter-node');
|
||||
const filterCountry = document.getElementById('filter-country');
|
||||
|
||||
function applyFilters() {
|
||||
const fn = filterNode.value;
|
||||
const fc = filterCountry.value.toUpperCase();
|
||||
const filtered = recent.filter(row =>
|
||||
(!fn || row.Node === fn) &&
|
||||
(!fc || row.Country === fc));
|
||||
renderRecentTable(filtered);
|
||||
}
|
||||
|
||||
filterNode.addEventListener('change', applyFilters);
|
||||
filterCountry.addEventListener('input', applyFilters);
|
||||
applyFilters();
|
||||
}
|
||||
|
||||
loadSnapshot();
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
Reference in New Issue
Block a user