bergamot/stats-gateway
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: stats-gateway gRPC service + dashboard + CLI

Browse Source
This commit is contained in:
BergaBruh
2026-05-05 21:04:32 +05:00
commit 4059e94759
44 changed files with 2513 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/probes/probes.go
+72
View File
@@ -0,0 +1,72 @@
package probes
import (
"context"
"crypto/tls"
"fmt"
"net"
"net/http"
"net/http/httptrace"
"time"
)
// Result holds timing measurements from a single HTTP probe.
type Result struct {
Node string // populated by storage on read; ignored by InsertSample (uses separate arg)
URL string
HTTPLatencyMs float64
TCPRttMs float64
DNSResolveMs float64
TLSHandshakeMs float64
ThroughputKbps int64
Timestamp time.Time
}
// ProbeHTTP performs an HTTP HEAD request to url, recording timing via httptrace.
func ProbeHTTP(ctx context.Context, url string) (*Result, error) {
r := &Result{URL: url, Timestamp: time.Now().UTC()}
var dnsStart, connStart, tlsStart time.Time
trace := &httptrace.ClientTrace{
DNSStart: func(_ httptrace.DNSStartInfo) { dnsStart = time.Now() },
DNSDone: func(_ httptrace.DNSDoneInfo) {
if !dnsStart.IsZero() {
r.DNSResolveMs = float64(time.Since(dnsStart).Microseconds()) / 1000
}
},
ConnectStart: func(_, _ string) { connStart = time.Now() },
ConnectDone: func(_, _ string, _ error) {
if !connStart.IsZero() {
r.TCPRttMs = float64(time.Since(connStart).Microseconds()) / 1000
}
},
TLSHandshakeStart: func() { tlsStart = time.Now() },
TLSHandshakeDone: func(_ tls.ConnectionState, _ error) {
if !tlsStart.IsZero() {
r.TLSHandshakeMs = float64(time.Since(tlsStart).Microseconds()) / 1000
}
},
}
req, err := http.NewRequestWithContext(httptrace.WithClientTrace(ctx, trace), "HEAD", url, nil)
if err != nil {
return nil, fmt.Errorf("new request: %w", err)
}
client := &http.Client{
Timeout: 10 * time.Second,
Transport: &http.Transport{
DialContext: (&net.Dialer{Timeout: 5 * time.Second}).DialContext,
},
}
start := time.Now()
resp, err := client.Do(req)
if err != nil {
return nil, fmt.Errorf("http: %w", err)
}
defer resp.Body.Close()
r.HTTPLatencyMs = float64(time.Since(start).Microseconds()) / 1000
return r, nil
}
internal/probes/probes_test.go
+33
View File
@@ -0,0 +1,33 @@
package probes
import (
"context"
"testing"
"time"
)
func TestProbeHTTP_Success(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
r, err := ProbeHTTP(ctx, "https://example.com/")
if err != nil {
t.Fatalf("expected success, got err: %v", err)
}
if r.HTTPLatencyMs <= 0 {
t.Errorf("expected positive latency, got %f", r.HTTPLatencyMs)
}
if r.TLSHandshakeMs <= 0 {
t.Errorf("expected positive TLS handshake time, got %f", r.TLSHandshakeMs)
}
}
func TestProbeHTTP_Timeout(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
_, err := ProbeHTTP(ctx, "https://10.255.255.1/") // unreachable
if err == nil {
t.Fatal("expected timeout error")
}
}
internal/service/service.go
+89
View File
@@ -0,0 +1,89 @@
package service
import (
"context"
"time"
statsv1 "github.com/bergabruh/stats-gateway/gen/stats/v1"
"github.com/bergabruh/stats-gateway/internal/storage"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
// Service implements statsv1.SpeedStatusServer with public RecentResults
// and Aggregate methods. Tunnel returns Unimplemented until Task 11 wires
// auth + handler.
type Service struct {
statsv1.UnimplementedSpeedStatusServer
store *storage.Store
}
func New(st *storage.Store) *Service { return &Service{store: st} }
func (s *Service) RecentResults(f *statsv1.Filter, stream statsv1.SpeedStatus_RecentResultsServer) error {
limit := int(f.Limit)
if limit <= 0 || limit > 1000 {
limit = 100
}
var since time.Time
if f.Since != "" {
t, err := time.Parse(time.RFC3339, f.Since)
if err != nil {
return status.Errorf(codes.InvalidArgument, "since: %v", err)
}
since = t
}
samples, err := s.store.QueryRecent(stream.Context(), f.Node, limit, since)
if err != nil {
return status.Errorf(codes.Internal, "query: %v", err)
}
for _, r := range samples {
err := stream.Send(&statsv1.Sample{
Node: r.Node,
Timestamp: r.Timestamp.Format(time.RFC3339Nano),
HttpLatencyMs: r.HTTPLatencyMs,
TcpRttMs: r.TCPRttMs,
DnsResolveMs: r.DNSResolveMs,
TlsHandshakeMs: r.TLSHandshakeMs,
ThroughputKbps: r.ThroughputKbps,
})
if err != nil {
return err
}
}
return nil
}
// aggregateBounds defines bucket upper bounds (ms) used by Aggregate.
var aggregateBounds = []float64{25, 50, 100, 250, 500, 1000, 5000}
func (s *Service) Aggregate(ctx context.Context, r *statsv1.Range) (*statsv1.Histogram, error) {
since, _ := time.Parse(time.RFC3339, r.Since)
until, _ := time.Parse(time.RFC3339, r.Until)
if until.IsZero() {
until = time.Now().UTC()
}
if since.IsZero() {
return nil, status.Error(codes.InvalidArgument, "since required")
}
samples, err := s.store.QueryRange(ctx, r.Node, since, until)
if err != nil {
return nil, status.Errorf(codes.Internal, "query: %v", err)
}
counts := make([]int64, len(aggregateBounds))
for _, sm := range samples {
for i, b := range aggregateBounds {
if sm.HTTPLatencyMs <= b {
counts[i]++
break
}
}
}
h := &statsv1.Histogram{}
for i, b := range aggregateBounds {
h.Buckets = append(h.Buckets, &statsv1.Bucket{UpperMs: b, Count: counts[i]})
}
return h, nil
}
internal/service/service_test.go
+124
View File
@@ -0,0 +1,124 @@
package service
import (
"context"
"net"
"path/filepath"
"testing"
"time"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/test/bufconn"
statsv1 "github.com/bergabruh/stats-gateway/gen/stats/v1"
"github.com/bergabruh/stats-gateway/internal/probes"
"github.com/bergabruh/stats-gateway/internal/storage"
)
func newTestServer(t *testing.T, st *storage.Store) (statsv1.SpeedStatusClient, func()) {
t.Helper()
lis := bufconn.Listen(1024 * 1024)
srv := grpc.NewServer()
statsv1.RegisterSpeedStatusServer(srv, New(st))
go srv.Serve(lis)
conn, err := grpc.NewClient("passthrough:///bufnet",
grpc.WithContextDialer(func(_ context.Context, _ string) (net.Conn, error) { return lis.Dial() }),
grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
t.Fatal(err)
}
cleanup := func() { conn.Close(); srv.Stop() }
return statsv1.NewSpeedStatusClient(conn), cleanup
}
func TestRecentResults_StreamsAllSamples(t *testing.T) {
dir := t.TempDir()
st, _ := storage.Open(filepath.Join(dir, "test.db"))
defer st.Close()
for i := 0; i < 3; i++ {
st.InsertSample(context.Background(), "fl", &probes.Result{
HTTPLatencyMs: float64(40 + i), Timestamp: time.Now().UTC().Add(time.Duration(i) * time.Second),
})
}
client, cleanup := newTestServer(t, st)
defer cleanup()
stream, err := client.RecentResults(context.Background(), &statsv1.Filter{Node: "fl", Limit: 10})
if err != nil {
t.Fatal(err)
}
count := 0
for {
_, err := stream.Recv()
if err != nil {
break
}
count++
}
if count != 3 {
t.Errorf("expected 3 samples, got %d", count)
}
}
func TestAggregate_BucketCounts(t *testing.T) {
dir := t.TempDir()
st, _ := storage.Open(filepath.Join(dir, "test.db"))
defer st.Close()
now := time.Now().UTC()
// latencies [10, 20, 30, 50, 100, 200] -> buckets [<=25:2, <=50:2, <=100:1, <=250:1, ...]
latencies := []float64{10, 20, 30, 50, 100, 200}
for i, ms := range latencies {
st.InsertSample(context.Background(), "fl", &probes.Result{
HTTPLatencyMs: ms, Timestamp: now.Add(time.Duration(i) * time.Second),
})
}
client, cleanup := newTestServer(t, st)
defer cleanup()
hist, err := client.Aggregate(context.Background(), &statsv1.Range{
Node: "fl",
Since: now.Add(-time.Minute).Format(time.RFC3339),
Until: now.Add(time.Minute).Format(time.RFC3339),
})
if err != nil {
t.Fatal(err)
}
// Find bucket counts by upper_ms
got := map[float64]int64{}
for _, b := range hist.Buckets {
got[b.UpperMs] = b.Count
}
want := map[float64]int64{25: 2, 50: 2, 100: 1, 250: 1, 500: 0, 1000: 0, 5000: 0}
for ms, want := range want {
if got[ms] != want {
t.Errorf("bucket <=%.0fms: got %d, want %d", ms, got[ms], want)
}
}
}
func TestTunnel_Unimplemented(t *testing.T) {
dir := t.TempDir()
st, _ := storage.Open(filepath.Join(dir, "test.db"))
defer st.Close()
client, cleanup := newTestServer(t, st)
defer cleanup()
stream, err := client.Tunnel(context.Background())
if err != nil {
t.Fatal(err)
}
stream.Send(&statsv1.Frame{Kind: statsv1.Frame_HELLO, Target: "1.2.3.4:80"})
_, err = stream.Recv()
if err == nil {
t.Fatal("expected Unimplemented error")
}
// Just check it's an error; the exact code/message will be checked indirectly when Task 11 wires real impl.
}
internal/storage/migrations/001_init.sql
+12
View File
@@ -0,0 +1,12 @@
CREATE TABLE IF NOT EXISTS samples (
id INTEGER PRIMARY KEY AUTOINCREMENT,
node TEXT NOT NULL,
ts TEXT NOT NULL,
http_latency_ms REAL NOT NULL,
tcp_rtt_ms REAL NOT NULL,
dns_resolve_ms REAL NOT NULL,
tls_handshake_ms REAL NOT NULL,
throughput_kbps INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_samples_node_ts ON samples (node, ts DESC);
internal/storage/sqlite.go
+114
View File
@@ -0,0 +1,114 @@
package storage
import (
"context"
"database/sql"
"embed"
"fmt"
"time"
"github.com/bergabruh/stats-gateway/internal/probes"
_ "github.com/mattn/go-sqlite3"
)
//go:embed migrations/*.sql
var migrations embed.FS
// tsLayout is RFC3339 with fixed 9-digit nanosecond fraction. Lexical
// ordering of formatted strings preserves chronological order — unlike
// time.RFC3339Nano which strips trailing zeros (".1Z" lex-sorts after
// ".100000001Z" because 'Z' > '0'). SQLite TEXT columns are compared
// lexically, so this matters for QueryRange boundaries.
const tsLayout = "2006-01-02T15:04:05.000000000Z07:00"
// Store wraps a SQLite database for probe sample persistence.
type Store struct {
db *sql.DB
}
// Open opens (or creates) the SQLite database at path and runs migrations.
func Open(path string) (*Store, error) {
db, err := sql.Open("sqlite3", path+"?_journal_mode=WAL&_busy_timeout=5000")
if err != nil {
return nil, err
}
s := &Store{db: db}
if err := s.migrate(); err != nil {
return nil, fmt.Errorf("migrate: %w", err)
}
return s, nil
}
// Close closes the underlying database.
func (s *Store) Close() error { return s.db.Close() }
func (s *Store) migrate() error {
body, err := migrations.ReadFile("migrations/001_init.sql")
if err != nil {
return err
}
_, err = s.db.Exec(string(body))
return err
}
// InsertSample stores a probe result for the given node.
func (s *Store) InsertSample(ctx context.Context, node string, r *probes.Result) error {
_, err := s.db.ExecContext(ctx, `
INSERT INTO samples (node, ts, http_latency_ms, tcp_rtt_ms, dns_resolve_ms, tls_handshake_ms, throughput_kbps)
VALUES (?, ?, ?, ?, ?, ?, ?)
`, node, r.Timestamp.Format(tsLayout),
r.HTTPLatencyMs, r.TCPRttMs, r.DNSResolveMs, r.TLSHandshakeMs, r.ThroughputKbps)
return err
}
// QueryRecent returns up to limit samples for node ordered by ts DESC.
// If since is zero, no lower bound is applied. If node is empty, all nodes are returned.
func (s *Store) QueryRecent(ctx context.Context, node string, limit int, since time.Time) ([]probes.Result, error) {
q := `SELECT node, ts, http_latency_ms, tcp_rtt_ms, dns_resolve_ms, tls_handshake_ms, throughput_kbps
FROM samples WHERE 1=1`
args := []any{}
if node != "" {
q += ` AND node = ?`
args = append(args, node)
}
if !since.IsZero() {
q += ` AND ts >= ?`
args = append(args, since.Format(tsLayout))
}
q += ` ORDER BY ts DESC LIMIT ?`
args = append(args, limit)
return s.scanSamples(ctx, q, args...)
}
// QueryRange returns samples for [since, until). Used by Aggregate (Task 5).
func (s *Store) QueryRange(ctx context.Context, node string, since, until time.Time) ([]probes.Result, error) {
q := `SELECT node, ts, http_latency_ms, tcp_rtt_ms, dns_resolve_ms, tls_handshake_ms, throughput_kbps
FROM samples WHERE ts >= ? AND ts < ?`
args := []any{since.Format(tsLayout), until.Format(tsLayout)}
if node != "" {
q += ` AND node = ?`
args = append(args, node)
}
q += ` ORDER BY ts ASC`
return s.scanSamples(ctx, q, args...)
}
func (s *Store) scanSamples(ctx context.Context, q string, args ...any) ([]probes.Result, error) {
rows, err := s.db.QueryContext(ctx, q, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []probes.Result
for rows.Next() {
var ts string
var r probes.Result
if err := rows.Scan(&r.Node, &ts, &r.HTTPLatencyMs, &r.TCPRttMs, &r.DNSResolveMs, &r.TLSHandshakeMs, &r.ThroughputKbps); err != nil {
return nil, err
}
r.Timestamp, _ = time.Parse(tsLayout, ts)
out = append(out, r)
}
return out, rows.Err()
}
internal/storage/sqlite_test.go
+106
View File
@@ -0,0 +1,106 @@
package storage
import (
"context"
"path/filepath"
"testing"
"time"
"github.com/bergabruh/stats-gateway/internal/probes"
)
func TestInsertAndQuery(t *testing.T) {
dir := t.TempDir()
st, err := Open(filepath.Join(dir, "test.db"))
if err != nil {
t.Fatal(err)
}
defer st.Close()
ctx := context.Background()
r := &probes.Result{
HTTPLatencyMs: 42.5,
TCPRttMs: 12.0,
DNSResolveMs: 3.0,
TLSHandshakeMs: 25.0,
Timestamp: time.Now().UTC(),
}
if err := st.InsertSample(ctx, "fl", r); err != nil {
t.Fatalf("insert: %v", err)
}
samples, err := st.QueryRecent(ctx, "fl", 10, time.Time{})
if err != nil {
t.Fatalf("query: %v", err)
}
if len(samples) != 1 {
t.Fatalf("expected 1 sample, got %d", len(samples))
}
if samples[0].HTTPLatencyMs != 42.5 {
t.Errorf("expected latency 42.5, got %f", samples[0].HTTPLatencyMs)
}
}
func TestQueryRange(t *testing.T) {
dir := t.TempDir()
st, err := Open(filepath.Join(dir, "test.db"))
if err != nil {
t.Fatal(err)
}
defer st.Close()
ctx := context.Background()
now := time.Now().UTC().Truncate(time.Second)
for i, dt := range []time.Duration{-2 * time.Hour, -1 * time.Hour, 0, 1 * time.Hour} {
r := &probes.Result{HTTPLatencyMs: float64(10 + i), Timestamp: now.Add(dt)}
if err := st.InsertSample(ctx, "fl", r); err != nil {
t.Fatal(err)
}
}
got, err := st.QueryRange(ctx, "fl", now.Add(-90*time.Minute), now.Add(30*time.Minute))
if err != nil {
t.Fatal(err)
}
// expect 2 samples: -1h and 0
if len(got) != 2 {
t.Fatalf("expected 2, got %d", len(got))
}
}
// TestQueryRange_SubsecondOrdering verifies fixed-width nanosecond format —
// regression guard for RFC3339Nano trailing-zero stripping bug.
func TestQueryRange_SubsecondOrdering(t *testing.T) {
dir := t.TempDir()
st, err := Open(filepath.Join(dir, "test.db"))
if err != nil {
t.Fatal(err)
}
defer st.Close()
ctx := context.Background()
base := time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC)
// Two timestamps where lexical ordering of RFC3339Nano disagrees with chronology:
// t1 = base + 100ms → RFC3339Nano formats as "...:00.1Z"
// t2 = base + 100ms + 1ns → RFC3339Nano formats as "...:00.100000001Z"
// Lexically ".1Z" > ".100000001Z" because 'Z' > '0', but t1 < t2.
t1 := base.Add(100 * time.Millisecond)
t2 := base.Add(100*time.Millisecond + time.Nanosecond)
for _, ts := range []time.Time{t1, t2} {
if err := st.InsertSample(ctx, "fl", &probes.Result{HTTPLatencyMs: 1.0, Timestamp: ts}); err != nil {
t.Fatal(err)
}
}
// Range starting just before t1 must include both samples.
got, err := st.QueryRange(ctx, "fl", base.Add(50*time.Millisecond), base.Add(200*time.Millisecond))
if err != nil {
t.Fatal(err)
}
if len(got) != 2 {
t.Fatalf("expected 2 samples in [t-50ms, t+200ms), got %d — lexical ordering bug?", len(got))
}
// Verify chronological ordering preserved on read.
if !got[0].Timestamp.Before(got[1].Timestamp) {
t.Errorf("expected got[0] < got[1] chronologically; got[0]=%v got[1]=%v", got[0].Timestamp, got[1].Timestamp)
}
}