cmd/speedtest-sync: read LibreSpeed local SQLite, sync to stats via gRPC
This commit is contained in:
BergaBruh
@@ -0,0 +1,115 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"flag"
|
||||
"log"
|
||||
"os"
|
||||
"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(parentDir(path), 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(path, []byte(strconv.FormatInt(id, 10)), 0o600)
|
||||
}
|
||||
|
||||
func parentDir(p string) string {
|
||||
for i := len(p) - 1; i >= 0; i-- {
|
||||
if p[i] == '/' {
|
||||
return p[:i]
|
||||
}
|
||||
}
|
||||
return "."
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
var tsUnix int64
|
||||
if ts.Valid {
|
||||
fmt.Sscanf(ts.String, "%d", &tsUnix)
|
||||
}
|
||||
if tsUnix == 0 {
|
||||
tsUnix = time.Now().Unix()
|
||||
}
|
||||
|
||||
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,74 @@
|
||||
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, _ := sql.Open("sqlite", dbPath)
|
||||
defer db.Close()
|
||||
|
||||
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)`)
|
||||
db.Exec(`INSERT INTO speedtest_users(timestamp, extra, dl, ul, ping, jitter)
|
||||
VALUES(datetime('now'), 'not-json', 1, 1, 1, 1)`)
|
||||
|
||||
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))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user