From 4059e94759b26c92b5d652c5af74698239bc185b Mon Sep 17 00:00:00 2001 From: BergaBruh Date: Tue, 5 May 2026 21:04:32 +0500 Subject: [PATCH] Initial commit: stats-gateway gRPC service + dashboard + CLI --- .gitignore | 13 + LICENSE | 21 + Makefile | 12 + README.md | 36 ++ cmd/collector/main.go | 47 ++ cmd/server/main.go | 38 ++ cmd/snapshot/main.go | 89 ++++ docs/AUTH.md | 44 ++ gen/stats/v1/stats.pb.go | 557 +++++++++++++++++++++++ gen/stats/v1/stats_grpc.pb.go | 205 +++++++++ go.mod | 16 + go.sum | 40 ++ internal/probes/probes.go | 72 +++ internal/probes/probes_test.go | 33 ++ internal/service/service.go | 89 ++++ internal/service/service_test.go | 124 +++++ internal/storage/migrations/001_init.sql | 12 + internal/storage/sqlite.go | 114 +++++ internal/storage/sqlite_test.go | 106 +++++ proto/Makefile | 14 + proto/stats.proto | 62 +++ psp-stats-cli/.gitignore | 1 + psp-stats-cli/Makefile | 10 + psp-stats-cli/README.md | 59 +++ psp-stats-cli/cmd/aggregate.go | 64 +++ psp-stats-cli/cmd/recent.go | 65 +++ psp-stats-cli/go.mod | 18 + psp-stats-cli/go.sum | 38 ++ psp-stats-cli/main.go | 45 ++ stats-site/build.sh | 41 ++ stats-site/config.toml | 10 + stats-site/content/_index.md | 3 + stats-site/layouts/index.html | 49 ++ stats-site/static/about/index.html | 26 ++ stats-site/static/css/main.css | 99 ++++ stats-site/static/data/.gitkeep | 1 + stats-site/static/js/charts.js | 137 ++++++ systemd/stats-collector.service | 12 + systemd/stats-collector.timer | 11 + systemd/stats-gateway.service | 24 + systemd/stats-site-build.service | 15 + systemd/stats-site-build.timer | 11 + systemd/stats-snapshot.service | 19 + systemd/stats-snapshot.timer | 11 + 44 files changed, 2513 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 Makefile create mode 100644 README.md create mode 100644 cmd/collector/main.go create mode 100644 cmd/server/main.go create mode 100644 cmd/snapshot/main.go create mode 100644 docs/AUTH.md create mode 100644 gen/stats/v1/stats.pb.go create mode 100644 gen/stats/v1/stats_grpc.pb.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 internal/probes/probes.go create mode 100644 internal/probes/probes_test.go create mode 100644 internal/service/service.go create mode 100644 internal/service/service_test.go create mode 100644 internal/storage/migrations/001_init.sql create mode 100644 internal/storage/sqlite.go create mode 100644 internal/storage/sqlite_test.go create mode 100644 proto/Makefile create mode 100644 proto/stats.proto create mode 100644 psp-stats-cli/.gitignore create mode 100644 psp-stats-cli/Makefile create mode 100644 psp-stats-cli/README.md create mode 100644 psp-stats-cli/cmd/aggregate.go create mode 100644 psp-stats-cli/cmd/recent.go create mode 100644 psp-stats-cli/go.mod create mode 100644 psp-stats-cli/go.sum create mode 100644 psp-stats-cli/main.go create mode 100755 stats-site/build.sh create mode 100644 stats-site/config.toml create mode 100644 stats-site/content/_index.md create mode 100644 stats-site/layouts/index.html create mode 100644 stats-site/static/about/index.html create mode 100644 stats-site/static/css/main.css create mode 100644 stats-site/static/data/.gitkeep create mode 100644 stats-site/static/js/charts.js create mode 100644 systemd/stats-collector.service create mode 100644 systemd/stats-collector.timer create mode 100644 systemd/stats-gateway.service create mode 100644 systemd/stats-site-build.service create mode 100644 systemd/stats-site-build.timer create mode 100644 systemd/stats-snapshot.service create mode 100644 systemd/stats-snapshot.timer diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5f91857 --- /dev/null +++ b/.gitignore @@ -0,0 +1,13 @@ +bin/ +*.exe +*.test +*.out +*.log +.DS_Store +.idea/ +.vscode/ +*.swp +*~ + +# experiment artifacts +data/snapshot.json diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..67a9325 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 bergamot + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..6931cfd --- /dev/null +++ b/Makefile @@ -0,0 +1,12 @@ +.PHONY: build test clean + +build: + go build -o bin/stats-gateway ./cmd/server + go build -o bin/stats-collector ./cmd/collector + go build -o bin/stats-snapshot ./cmd/snapshot + +test: + go test ./... + +clean: + rm -rf bin/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..2eb363c --- /dev/null +++ b/README.md @@ -0,0 +1,36 @@ +# stats-gateway + +gRPC service collecting and publishing HTTP/TCP/DNS/TLS-handshake timing +metrics for two geo-distributed OpenSpeedTest nodes (FL — Finland, PL1 — Poland). +Powers the dashboard at . + +## Components + +- **`cmd/server`** — gRPC server (TLS terminated by nginx, listens on loopback). +- **`cmd/collector`** — periodic prober (HTTP HEAD with httptrace), runs every 5 min via systemd timer. +- **`cmd/snapshot`** — generates `snapshot.json` for the dashboard, runs every 10 min. +- **`internal/probes`** — timing logic (DNS, TCP, TLS handshake, HTTP). +- **`internal/storage`** — SQLite WAL with embedded migrations. +- **`internal/service`** — gRPC handlers (`RecentResults`, `Aggregate`, `Tunnel`). +- **`proto/`** — protobuf definitions; generated Go in `gen/`. +- **`stats-site/`** — static dashboard (vanilla JS + canvas charts, no framework deps). +- **`psp-stats-cli/`** — Go CLI for querying the API. +- **`systemd/`** — ops units for running the service + timers. +- **`docs/AUTH.md`** — auth doc for the `Tunnel` method (Bearer JWT/HS256, used by internal CI agents). + +## Build + + make build # ./bin/{stats-gateway,stats-collector,stats-snapshot} + make test + cd psp-stats-cli && make build + +## API + + grpcurl -d '{"node":"fl","limit":5}' \ + stats.projectshitpost.fun:443 stats.v1.SpeedStatus/RecentResults + +See `proto/stats.proto`. + +## License + +MIT — see [LICENSE](./LICENSE). diff --git a/cmd/collector/main.go b/cmd/collector/main.go new file mode 100644 index 0000000..a4b2c29 --- /dev/null +++ b/cmd/collector/main.go @@ -0,0 +1,47 @@ +package main + +import ( + "context" + "flag" + "log" + "time" + + "github.com/bergabruh/stats-gateway/internal/probes" + "github.com/bergabruh/stats-gateway/internal/storage" +) + +func main() { + dbPath := flag.String("db", "/var/lib/stats-gateway/stats.db", "sqlite path") + flag.Parse() + + st, err := storage.Open(*dbPath) + if err != nil { + log.Fatalf("open db: %v", err) + } + defer st.Close() + + targets := []struct { + Node string + URL string + }{ + {"fl", "https://fl.speedtest.projectshitpost.fun/"}, + {"pl1", "https://speedtest.projectshitpost.fun/"}, + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + for _, t := range targets { + r, err := probes.ProbeHTTP(ctx, t.URL) + if err != nil { + log.Printf("probe %s: %v", t.Node, err) + continue + } + if err := st.InsertSample(ctx, t.Node, r); err != nil { + log.Printf("store %s: %v", t.Node, err) + continue + } + log.Printf("probe %s ok: http=%.1fms tls=%.1fms", + t.Node, r.HTTPLatencyMs, r.TLSHandshakeMs) + } +} diff --git a/cmd/server/main.go b/cmd/server/main.go new file mode 100644 index 0000000..0a5d8eb --- /dev/null +++ b/cmd/server/main.go @@ -0,0 +1,38 @@ +package main + +import ( + "flag" + "log" + "net" + + "google.golang.org/grpc" + + statsv1 "github.com/bergabruh/stats-gateway/gen/stats/v1" + "github.com/bergabruh/stats-gateway/internal/service" + "github.com/bergabruh/stats-gateway/internal/storage" +) + +func main() { + addr := flag.String("addr", "127.0.0.1:50051", "listen address") + dbPath := flag.String("db", "/var/lib/stats-gateway/stats.db", "sqlite path") + flag.Parse() + + st, err := storage.Open(*dbPath) + if err != nil { + log.Fatalf("storage: %v", err) + } + defer st.Close() + + lis, err := net.Listen("tcp", *addr) + if err != nil { + log.Fatalf("listen: %v", err) + } + + srv := grpc.NewServer() + statsv1.RegisterSpeedStatusServer(srv, service.New(st)) + + log.Printf("stats-gateway listening on %s", *addr) + if err := srv.Serve(lis); err != nil { + log.Fatalf("serve: %v", err) + } +} diff --git a/cmd/snapshot/main.go b/cmd/snapshot/main.go new file mode 100644 index 0000000..070c3e6 --- /dev/null +++ b/cmd/snapshot/main.go @@ -0,0 +1,89 @@ +package main + +import ( + "context" + "encoding/json" + "flag" + "log" + "os" + "path/filepath" + "time" + + "github.com/bergabruh/stats-gateway/internal/storage" +) + +type sample struct { + TS string `json:"ts"` + HTTPMs float64 `json:"http_ms"` + TLSMs float64 `json:"tls_ms"` + TCPMs float64 `json:"tcp_ms"` +} + +type nodeData struct { + Recent []sample `json:"recent"` +} + +type snapshot struct { + GeneratedAt string `json:"generated_at"` + Nodes map[string]nodeData `json:"nodes"` +} + +func main() { + dbPath := flag.String("db", "/var/lib/stats-gateway/stats.db", "") + out := flag.String("out", "/var/www/stats-site/data/snapshot.json", "") + limitPerNode := flag.Int("limit", 288, "max samples per node (288 = ~24h at 5min interval)") + flag.Parse() + + st, err := storage.Open(*dbPath) + if err != nil { + log.Fatalf("open db: %v", err) + } + defer st.Close() + + s := snapshot{ + GeneratedAt: time.Now().UTC().Format(time.RFC3339), + Nodes: map[string]nodeData{}, + } + ctx := context.Background() + for _, node := range []string{"fl", "pl1"} { + samples, err := st.QueryRecent(ctx, node, *limitPerNode, time.Time{}) + if err != nil { + log.Printf("query %s: %v", node, err) + continue + } + nd := nodeData{Recent: make([]sample, 0, len(samples))} + for _, sm := range samples { + nd.Recent = append(nd.Recent, sample{ + TS: sm.Timestamp.Format(time.RFC3339Nano), + HTTPMs: sm.HTTPLatencyMs, + TLSMs: sm.TLSHandshakeMs, + TCPMs: sm.TCPRttMs, + }) + } + s.Nodes[node] = nd + } + + if err := os.MkdirAll(filepath.Dir(*out), 0755); err != nil { + log.Fatalf("mkdir %s: %v", filepath.Dir(*out), err) + } + + // Atomic write: write to .tmp then rename. + tmp := *out + ".tmp" + f, err := os.Create(tmp) + if err != nil { + log.Fatalf("create %s: %v", tmp, err) + } + enc := json.NewEncoder(f) + enc.SetIndent("", " ") + if err := enc.Encode(s); err != nil { + f.Close() + os.Remove(tmp) + log.Fatalf("encode: %v", err) + } + f.Close() + if err := os.Rename(tmp, *out); err != nil { + log.Fatalf("rename: %v", err) + } + + log.Printf("snapshot: nodes=%d → %s", len(s.Nodes), *out) +} diff --git a/docs/AUTH.md b/docs/AUTH.md new file mode 100644 index 0000000..4e7df0a --- /dev/null +++ b/docs/AUTH.md @@ -0,0 +1,44 @@ +# Auth для Tunnel-метода + +Bearer JWT (HS256), HMAC signing secret в `/etc/stats-gateway/jwt.key`. + +## Claims + +- `sub` — client-id (`bergamot-laptop`, `bergamot-phone`, ...) +- `scope` — `tunnel` (только этот scope разрешён для Tunnel-метода) +- `iat`, `exp` — стандартные UNIX timestamps +- `jti` — random hex для аудита/revocation + +## Выпуск нового токена + +На gitea-узле от пользователя с доступом к `jwt.key`: + + sudo -u stats /usr/local/bin/stats-issue-token --sub --days + +(Helper-скрипт `issue-token.sh` устанавливается на сервер в Task 11 как Go-binary `stats-issue-token`.) + +Локальный путь во время разработки — `experiments/grpc-tunnel/issue-token.sh`. Семантика идентична Go-варианту. + +## Revocation + +Опции (по убыванию агрессивности): + +1. **Polite revocation** (один client) — добавить `jti` отзываемого токена в `/etc/stats-gateway/revoked.txt` + (формат: один jti на строку, `#` в начале строки — комментарий). Сервис подгружает на каждом запросе. +2. **Mass revocation** (все клиенты) — `openssl rand -hex 32 > /etc/stats-gateway/jwt.key`, + `systemctl restart stats-gateway`. Затем переиздать токены живым клиентам и раздать оффлайн. + +Polite revocation реализуется в Task 11 (через `LoadRevokedFile`). Для старта — только rotation. + +## Ротация secret + +Раз в год минимум. Перед сменой: +1. Выпустить новые токены клиентам с новым ключом (НЕ deploy ещё на сервер) +2. Раздать токены оффлайн (мессенджер, не plain SMS) +3. Заменить ключ на сервере, restart +4. Старые токены немедленно невалидны + +## Storage клиентского токена + +На клиенте — `~/.psp/token` mode 600. CLI читает только этот файл (см. `psp-stats-cli connect --token-file`). +Не коммитить токены в репозитории, не передавать через email/Slack без e2e. diff --git a/gen/stats/v1/stats.pb.go b/gen/stats/v1/stats.pb.go new file mode 100644 index 0000000..0d326cf --- /dev/null +++ b/gen/stats/v1/stats.pb.go @@ -0,0 +1,557 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11 +// protoc v7.34.1 +// source: stats.proto + +package statsv1 + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// Control message. First frame from client must be HELLO with target. +type Frame_Kind int32 + +const ( + Frame_UNKNOWN Frame_Kind = 0 + Frame_HELLO Frame_Kind = 1 + Frame_DATA Frame_Kind = 2 + Frame_CLOSE Frame_Kind = 3 +) + +// Enum value maps for Frame_Kind. +var ( + Frame_Kind_name = map[int32]string{ + 0: "UNKNOWN", + 1: "HELLO", + 2: "DATA", + 3: "CLOSE", + } + Frame_Kind_value = map[string]int32{ + "UNKNOWN": 0, + "HELLO": 1, + "DATA": 2, + "CLOSE": 3, + } +) + +func (x Frame_Kind) Enum() *Frame_Kind { + p := new(Frame_Kind) + *p = x + return p +} + +func (x Frame_Kind) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (Frame_Kind) Descriptor() protoreflect.EnumDescriptor { + return file_stats_proto_enumTypes[0].Descriptor() +} + +func (Frame_Kind) Type() protoreflect.EnumType { + return &file_stats_proto_enumTypes[0] +} + +func (x Frame_Kind) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use Frame_Kind.Descriptor instead. +func (Frame_Kind) EnumDescriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{5, 0} +} + +type Filter struct { + state protoimpl.MessageState `protogen:"open.v1"` + Node string `protobuf:"bytes,1,opt,name=node,proto3" json:"node,omitempty"` // "fl" | "pl1" | "" (all) + Limit int32 `protobuf:"varint,2,opt,name=limit,proto3" json:"limit,omitempty"` // 1..1000 + Since string `protobuf:"bytes,3,opt,name=since,proto3" json:"since,omitempty"` // RFC3339, optional + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Filter) Reset() { + *x = Filter{} + mi := &file_stats_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Filter) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Filter) ProtoMessage() {} + +func (x *Filter) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[0] + 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 Filter.ProtoReflect.Descriptor instead. +func (*Filter) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{0} +} + +func (x *Filter) GetNode() string { + if x != nil { + return x.Node + } + return "" +} + +func (x *Filter) GetLimit() int32 { + if x != nil { + return x.Limit + } + return 0 +} + +func (x *Filter) GetSince() string { + if x != nil { + return x.Since + } + return "" +} + +type Sample struct { + state protoimpl.MessageState `protogen:"open.v1"` + Node string `protobuf:"bytes,1,opt,name=node,proto3" json:"node,omitempty"` + Timestamp string `protobuf:"bytes,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"` // RFC3339 + HttpLatencyMs float64 `protobuf:"fixed64,3,opt,name=http_latency_ms,json=httpLatencyMs,proto3" json:"http_latency_ms,omitempty"` + TcpRttMs float64 `protobuf:"fixed64,4,opt,name=tcp_rtt_ms,json=tcpRttMs,proto3" json:"tcp_rtt_ms,omitempty"` + DnsResolveMs float64 `protobuf:"fixed64,5,opt,name=dns_resolve_ms,json=dnsResolveMs,proto3" json:"dns_resolve_ms,omitempty"` + TlsHandshakeMs float64 `protobuf:"fixed64,6,opt,name=tls_handshake_ms,json=tlsHandshakeMs,proto3" json:"tls_handshake_ms,omitempty"` + ThroughputKbps int64 `protobuf:"varint,7,opt,name=throughput_kbps,json=throughputKbps,proto3" json:"throughput_kbps,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Sample) Reset() { + *x = Sample{} + mi := &file_stats_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Sample) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Sample) ProtoMessage() {} + +func (x *Sample) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[1] + 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 Sample.ProtoReflect.Descriptor instead. +func (*Sample) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{1} +} + +func (x *Sample) GetNode() string { + if x != nil { + return x.Node + } + return "" +} + +func (x *Sample) GetTimestamp() string { + if x != nil { + return x.Timestamp + } + return "" +} + +func (x *Sample) GetHttpLatencyMs() float64 { + if x != nil { + return x.HttpLatencyMs + } + return 0 +} + +func (x *Sample) GetTcpRttMs() float64 { + if x != nil { + return x.TcpRttMs + } + return 0 +} + +func (x *Sample) GetDnsResolveMs() float64 { + if x != nil { + return x.DnsResolveMs + } + return 0 +} + +func (x *Sample) GetTlsHandshakeMs() float64 { + if x != nil { + return x.TlsHandshakeMs + } + return 0 +} + +func (x *Sample) GetThroughputKbps() int64 { + if x != nil { + return x.ThroughputKbps + } + return 0 +} + +type Range struct { + state protoimpl.MessageState `protogen:"open.v1"` + Node string `protobuf:"bytes,1,opt,name=node,proto3" json:"node,omitempty"` + Since string `protobuf:"bytes,2,opt,name=since,proto3" json:"since,omitempty"` + Until string `protobuf:"bytes,3,opt,name=until,proto3" json:"until,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Range) Reset() { + *x = Range{} + mi := &file_stats_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Range) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Range) ProtoMessage() {} + +func (x *Range) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[2] + 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 Range.ProtoReflect.Descriptor instead. +func (*Range) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{2} +} + +func (x *Range) GetNode() string { + if x != nil { + return x.Node + } + return "" +} + +func (x *Range) GetSince() string { + if x != nil { + return x.Since + } + return "" +} + +func (x *Range) GetUntil() string { + if x != nil { + return x.Until + } + return "" +} + +type Histogram struct { + state protoimpl.MessageState `protogen:"open.v1"` + Buckets []*Bucket `protobuf:"bytes,1,rep,name=buckets,proto3" json:"buckets,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Histogram) Reset() { + *x = Histogram{} + mi := &file_stats_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Histogram) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Histogram) ProtoMessage() {} + +func (x *Histogram) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[3] + 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 Histogram.ProtoReflect.Descriptor instead. +func (*Histogram) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{3} +} + +func (x *Histogram) GetBuckets() []*Bucket { + if x != nil { + return x.Buckets + } + return nil +} + +type Bucket struct { + state protoimpl.MessageState `protogen:"open.v1"` + UpperMs float64 `protobuf:"fixed64,1,opt,name=upper_ms,json=upperMs,proto3" json:"upper_ms,omitempty"` + Count int64 `protobuf:"varint,2,opt,name=count,proto3" json:"count,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Bucket) Reset() { + *x = Bucket{} + mi := &file_stats_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Bucket) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Bucket) ProtoMessage() {} + +func (x *Bucket) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[4] + 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 Bucket.ProtoReflect.Descriptor instead. +func (*Bucket) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{4} +} + +func (x *Bucket) GetUpperMs() float64 { + if x != nil { + return x.UpperMs + } + return 0 +} + +func (x *Bucket) GetCount() int64 { + if x != nil { + return x.Count + } + return 0 +} + +type Frame struct { + state protoimpl.MessageState `protogen:"open.v1"` + Kind Frame_Kind `protobuf:"varint,1,opt,name=kind,proto3,enum=stats.v1.Frame_Kind" json:"kind,omitempty"` + Target string `protobuf:"bytes,2,opt,name=target,proto3" json:"target,omitempty"` // "ip:port" — only used in HELLO + Payload []byte `protobuf:"bytes,3,opt,name=payload,proto3" json:"payload,omitempty"` // raw TCP bytes + Padding []byte `protobuf:"bytes,4,opt,name=padding,proto3" json:"padding,omitempty"` // random padding для frame-length obfuscation + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Frame) Reset() { + *x = Frame{} + mi := &file_stats_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Frame) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Frame) ProtoMessage() {} + +func (x *Frame) ProtoReflect() protoreflect.Message { + mi := &file_stats_proto_msgTypes[5] + 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 Frame.ProtoReflect.Descriptor instead. +func (*Frame) Descriptor() ([]byte, []int) { + return file_stats_proto_rawDescGZIP(), []int{5} +} + +func (x *Frame) GetKind() Frame_Kind { + if x != nil { + return x.Kind + } + return Frame_UNKNOWN +} + +func (x *Frame) GetTarget() string { + if x != nil { + return x.Target + } + return "" +} + +func (x *Frame) GetPayload() []byte { + if x != nil { + return x.Payload + } + return nil +} + +func (x *Frame) GetPadding() []byte { + if x != nil { + return x.Padding + } + return nil +} + +var File_stats_proto protoreflect.FileDescriptor + +const file_stats_proto_rawDesc = "" + + "\n" + + "\vstats.proto\x12\bstats.v1\"H\n" + + "\x06Filter\x12\x12\n" + + "\x04node\x18\x01 \x01(\tR\x04node\x12\x14\n" + + "\x05limit\x18\x02 \x01(\x05R\x05limit\x12\x14\n" + + "\x05since\x18\x03 \x01(\tR\x05since\"\xf9\x01\n" + + "\x06Sample\x12\x12\n" + + "\x04node\x18\x01 \x01(\tR\x04node\x12\x1c\n" + + "\ttimestamp\x18\x02 \x01(\tR\ttimestamp\x12&\n" + + "\x0fhttp_latency_ms\x18\x03 \x01(\x01R\rhttpLatencyMs\x12\x1c\n" + + "\n" + + "tcp_rtt_ms\x18\x04 \x01(\x01R\btcpRttMs\x12$\n" + + "\x0edns_resolve_ms\x18\x05 \x01(\x01R\fdnsResolveMs\x12(\n" + + "\x10tls_handshake_ms\x18\x06 \x01(\x01R\x0etlsHandshakeMs\x12'\n" + + "\x0fthroughput_kbps\x18\a \x01(\x03R\x0ethroughputKbps\"G\n" + + "\x05Range\x12\x12\n" + + "\x04node\x18\x01 \x01(\tR\x04node\x12\x14\n" + + "\x05since\x18\x02 \x01(\tR\x05since\x12\x14\n" + + "\x05until\x18\x03 \x01(\tR\x05until\"7\n" + + "\tHistogram\x12*\n" + + "\abuckets\x18\x01 \x03(\v2\x10.stats.v1.BucketR\abuckets\"9\n" + + "\x06Bucket\x12\x19\n" + + "\bupper_ms\x18\x01 \x01(\x01R\aupperMs\x12\x14\n" + + "\x05count\x18\x02 \x01(\x03R\x05count\"\xb2\x01\n" + + "\x05Frame\x12(\n" + + "\x04kind\x18\x01 \x01(\x0e2\x14.stats.v1.Frame.KindR\x04kind\x12\x16\n" + + "\x06target\x18\x02 \x01(\tR\x06target\x12\x18\n" + + "\apayload\x18\x03 \x01(\fR\apayload\x12\x18\n" + + "\apadding\x18\x04 \x01(\fR\apadding\"3\n" + + "\x04Kind\x12\v\n" + + "\aUNKNOWN\x10\x00\x12\t\n" + + "\x05HELLO\x10\x01\x12\b\n" + + "\x04DATA\x10\x02\x12\t\n" + + "\x05CLOSE\x10\x032\xa7\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" + +var ( + file_stats_proto_rawDescOnce sync.Once + file_stats_proto_rawDescData []byte +) + +func file_stats_proto_rawDescGZIP() []byte { + file_stats_proto_rawDescOnce.Do(func() { + file_stats_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_stats_proto_rawDesc), len(file_stats_proto_rawDesc))) + }) + return file_stats_proto_rawDescData +} + +var file_stats_proto_enumTypes = make([]protoimpl.EnumInfo, 1) +var file_stats_proto_msgTypes = make([]protoimpl.MessageInfo, 6) +var file_stats_proto_goTypes = []any{ + (Frame_Kind)(0), // 0: stats.v1.Frame.Kind + (*Filter)(nil), // 1: stats.v1.Filter + (*Sample)(nil), // 2: stats.v1.Sample + (*Range)(nil), // 3: stats.v1.Range + (*Histogram)(nil), // 4: stats.v1.Histogram + (*Bucket)(nil), // 5: stats.v1.Bucket + (*Frame)(nil), // 6: stats.v1.Frame +} +var file_stats_proto_depIdxs = []int32{ + 5, // 0: stats.v1.Histogram.buckets:type_name -> stats.v1.Bucket + 0, // 1: stats.v1.Frame.kind:type_name -> stats.v1.Frame.Kind + 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 + 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 +} + +func init() { file_stats_proto_init() } +func file_stats_proto_init() { + if File_stats_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_stats_proto_rawDesc), len(file_stats_proto_rawDesc)), + NumEnums: 1, + NumMessages: 6, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_stats_proto_goTypes, + DependencyIndexes: file_stats_proto_depIdxs, + EnumInfos: file_stats_proto_enumTypes, + MessageInfos: file_stats_proto_msgTypes, + }.Build() + File_stats_proto = out.File + file_stats_proto_goTypes = nil + file_stats_proto_depIdxs = nil +} diff --git a/gen/stats/v1/stats_grpc.pb.go b/gen/stats/v1/stats_grpc.pb.go new file mode 100644 index 0000000..ffb3507 --- /dev/null +++ b/gen/stats/v1/stats_grpc.pb.go @@ -0,0 +1,205 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.6.1 +// - protoc v7.34.1 +// source: stats.proto + +package statsv1 + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.64.0 or later. +const _ = grpc.SupportPackageIsVersion9 + +const ( + SpeedStatus_RecentResults_FullMethodName = "/stats.v1.SpeedStatus/RecentResults" + SpeedStatus_Aggregate_FullMethodName = "/stats.v1.SpeedStatus/Aggregate" + SpeedStatus_Tunnel_FullMethodName = "/stats.v1.SpeedStatus/Tunnel" +) + +// SpeedStatusClient is the client API for SpeedStatus service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +type SpeedStatusClient interface { + // RecentResults streams the last N samples for given filter. + RecentResults(ctx context.Context, in *Filter, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Sample], error) + // Aggregate returns histogram for a time range. + Aggregate(ctx context.Context, in *Range, opts ...grpc.CallOption) (*Histogram, error) + // Tunnel is a bidirectional stream for authenticated clients. + // Auth: Bearer JWT (HS256) в gRPC metadata "authorization: Bearer ". + // Server validates: signature, exp, scope=="tunnel", optional revocation list. + Tunnel(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[Frame, Frame], error) +} + +type speedStatusClient struct { + cc grpc.ClientConnInterface +} + +func NewSpeedStatusClient(cc grpc.ClientConnInterface) SpeedStatusClient { + return &speedStatusClient{cc} +} + +func (c *speedStatusClient) RecentResults(ctx context.Context, in *Filter, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Sample], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &SpeedStatus_ServiceDesc.Streams[0], SpeedStatus_RecentResults_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[Filter, Sample]{ClientStream: stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type SpeedStatus_RecentResultsClient = grpc.ServerStreamingClient[Sample] + +func (c *speedStatusClient) Aggregate(ctx context.Context, in *Range, opts ...grpc.CallOption) (*Histogram, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(Histogram) + err := c.cc.Invoke(ctx, SpeedStatus_Aggregate_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *speedStatusClient) Tunnel(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[Frame, Frame], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &SpeedStatus_ServiceDesc.Streams[1], SpeedStatus_Tunnel_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[Frame, Frame]{ClientStream: stream} + return x, nil +} + +// 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] + +// SpeedStatusServer is the server API for SpeedStatus service. +// All implementations must embed UnimplementedSpeedStatusServer +// for forward compatibility. +type SpeedStatusServer interface { + // RecentResults streams the last N samples for given filter. + RecentResults(*Filter, grpc.ServerStreamingServer[Sample]) error + // Aggregate returns histogram for a time range. + Aggregate(context.Context, *Range) (*Histogram, error) + // Tunnel is a bidirectional stream for authenticated clients. + // Auth: Bearer JWT (HS256) в gRPC metadata "authorization: Bearer ". + // Server validates: signature, exp, scope=="tunnel", optional revocation list. + Tunnel(grpc.BidiStreamingServer[Frame, Frame]) error + mustEmbedUnimplementedSpeedStatusServer() +} + +// UnimplementedSpeedStatusServer must be embedded to have +// forward compatible implementations. +// +// NOTE: this should be embedded by value instead of pointer to avoid a nil +// pointer dereference when methods are called. +type UnimplementedSpeedStatusServer struct{} + +func (UnimplementedSpeedStatusServer) RecentResults(*Filter, grpc.ServerStreamingServer[Sample]) error { + return status.Error(codes.Unimplemented, "method RecentResults not implemented") +} +func (UnimplementedSpeedStatusServer) Aggregate(context.Context, *Range) (*Histogram, error) { + return nil, status.Error(codes.Unimplemented, "method Aggregate not implemented") +} +func (UnimplementedSpeedStatusServer) Tunnel(grpc.BidiStreamingServer[Frame, Frame]) error { + return status.Error(codes.Unimplemented, "method Tunnel not implemented") +} +func (UnimplementedSpeedStatusServer) mustEmbedUnimplementedSpeedStatusServer() {} +func (UnimplementedSpeedStatusServer) testEmbeddedByValue() {} + +// UnsafeSpeedStatusServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to SpeedStatusServer will +// result in compilation errors. +type UnsafeSpeedStatusServer interface { + mustEmbedUnimplementedSpeedStatusServer() +} + +func RegisterSpeedStatusServer(s grpc.ServiceRegistrar, srv SpeedStatusServer) { + // If the following call panics, it indicates UnimplementedSpeedStatusServer was + // embedded by pointer and is nil. This will cause panics if an + // unimplemented method is ever invoked, so we test this at initialization + // time to prevent it from happening at runtime later due to I/O. + if t, ok := srv.(interface{ testEmbeddedByValue() }); ok { + t.testEmbeddedByValue() + } + s.RegisterService(&SpeedStatus_ServiceDesc, srv) +} + +func _SpeedStatus_RecentResults_Handler(srv interface{}, stream grpc.ServerStream) error { + m := new(Filter) + if err := stream.RecvMsg(m); err != nil { + return err + } + return srv.(SpeedStatusServer).RecentResults(m, &grpc.GenericServerStream[Filter, Sample]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type SpeedStatus_RecentResultsServer = grpc.ServerStreamingServer[Sample] + +func _SpeedStatus_Aggregate_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(Range) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(SpeedStatusServer).Aggregate(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: SpeedStatus_Aggregate_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(SpeedStatusServer).Aggregate(ctx, req.(*Range)) + } + return interceptor(ctx, in, info, handler) +} + +func _SpeedStatus_Tunnel_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(SpeedStatusServer).Tunnel(&grpc.GenericServerStream[Frame, Frame]{ServerStream: stream}) +} + +// 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] + +// 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) +var SpeedStatus_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "stats.v1.SpeedStatus", + HandlerType: (*SpeedStatusServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "Aggregate", + Handler: _SpeedStatus_Aggregate_Handler, + }, + }, + Streams: []grpc.StreamDesc{ + { + StreamName: "RecentResults", + Handler: _SpeedStatus_RecentResults_Handler, + ServerStreams: true, + }, + { + StreamName: "Tunnel", + Handler: _SpeedStatus_Tunnel_Handler, + ServerStreams: true, + ClientStreams: true, + }, + }, + Metadata: "stats.proto", +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..f23b1d8 --- /dev/null +++ b/go.mod @@ -0,0 +1,16 @@ +module github.com/bergabruh/stats-gateway + +go 1.25.0 + +require ( + github.com/mattn/go-sqlite3 v1.14.44 + google.golang.org/grpc v1.81.0 + google.golang.org/protobuf v1.36.11 +) + +require ( + 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 +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..1b580a2 --- /dev/null +++ b/go.sum @@ -0,0 +1,40 @@ +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/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= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +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-sqlite3 v1.14.44 h1:3VSe+xafpbzsLbdr2AWlAZk9yRHiBhTBakioXaCKTF8= +github.com/mattn/go-sqlite3 v1.14.44/go.mod h1:pjEuOr8IwzLJP2MfGeTb0A35jauH+C2kbHKBr7yXKVQ= +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= +go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= +go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= +go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= +go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= +go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= +go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= +go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= +go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= +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.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= +golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171 h1:ggcbiqK8WWh6l1dnltU4BgWGIGo+EVYxCaAPih/zQXQ= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.81.0 h1:W3G9N3KQf3BU+YuCtGKJk0CmxQNbAISICD/9AORxLIw= +google.golang.org/grpc v1.81.0/go.mod h1:xGH9GfzOyMTGIOXBJmXt+BX/V0kcdQbdcuwQ/zNw42I= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= diff --git a/internal/probes/probes.go b/internal/probes/probes.go new file mode 100644 index 0000000..a417616 --- /dev/null +++ b/internal/probes/probes.go @@ -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 +} diff --git a/internal/probes/probes_test.go b/internal/probes/probes_test.go new file mode 100644 index 0000000..2edae08 --- /dev/null +++ b/internal/probes/probes_test.go @@ -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") + } +} diff --git a/internal/service/service.go b/internal/service/service.go new file mode 100644 index 0000000..23ab978 --- /dev/null +++ b/internal/service/service.go @@ -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 +} diff --git a/internal/service/service_test.go b/internal/service/service_test.go new file mode 100644 index 0000000..45208bf --- /dev/null +++ b/internal/service/service_test.go @@ -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. +} diff --git a/internal/storage/migrations/001_init.sql b/internal/storage/migrations/001_init.sql new file mode 100644 index 0000000..b0b9817 --- /dev/null +++ b/internal/storage/migrations/001_init.sql @@ -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); diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go new file mode 100644 index 0000000..fc4eb90 --- /dev/null +++ b/internal/storage/sqlite.go @@ -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() +} diff --git a/internal/storage/sqlite_test.go b/internal/storage/sqlite_test.go new file mode 100644 index 0000000..300292f --- /dev/null +++ b/internal/storage/sqlite_test.go @@ -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) + } +} diff --git a/proto/Makefile b/proto/Makefile new file mode 100644 index 0000000..4b7b667 --- /dev/null +++ b/proto/Makefile @@ -0,0 +1,14 @@ +PROTOC := protoc +GEN_OUT := ../gen/stats/v1 + +.PHONY: gen +gen: + mkdir -p $(GEN_OUT) + $(PROTOC) \ + --go_out=$(GEN_OUT) --go_opt=paths=source_relative \ + --go-grpc_out=$(GEN_OUT) --go-grpc_opt=paths=source_relative \ + -I . stats.proto + +.PHONY: clean +clean: + rm -rf $(GEN_OUT) diff --git a/proto/stats.proto b/proto/stats.proto new file mode 100644 index 0000000..f34738d --- /dev/null +++ b/proto/stats.proto @@ -0,0 +1,62 @@ +syntax = "proto3"; + +package stats.v1; +option go_package = "github.com/bergabruh/stats-gateway/gen/stats/v1;statsv1"; + +service SpeedStatus { + // RecentResults streams the last N samples for given filter. + rpc RecentResults(Filter) returns (stream Sample); + + // Aggregate returns histogram for a time range. + rpc Aggregate(Range) returns (Histogram); + + // Tunnel is a bidirectional stream for authenticated clients. + // Auth: Bearer JWT (HS256) в gRPC metadata "authorization: Bearer ". + // Server validates: signature, exp, scope=="tunnel", optional revocation list. + rpc Tunnel(stream Frame) returns (stream Frame); +} + +message Filter { + string node = 1; // "fl" | "pl1" | "" (all) + int32 limit = 2; // 1..1000 + string since = 3; // RFC3339, optional +} + +message Sample { + string node = 1; + string timestamp = 2; // RFC3339 + double http_latency_ms = 3; + double tcp_rtt_ms = 4; + double dns_resolve_ms = 5; + double tls_handshake_ms = 6; + int64 throughput_kbps = 7; +} + +message Range { + string node = 1; + string since = 2; + string until = 3; +} + +message Histogram { + repeated Bucket buckets = 1; +} + +message Bucket { + double upper_ms = 1; + int64 count = 2; +} + +message Frame { + // Control message. First frame from client must be HELLO with target. + enum Kind { + UNKNOWN = 0; + HELLO = 1; + DATA = 2; + CLOSE = 3; + } + Kind kind = 1; + string target = 2; // "ip:port" — only used in HELLO + bytes payload = 3; // raw TCP bytes + bytes padding = 4; // random padding для frame-length obfuscation +} diff --git a/psp-stats-cli/.gitignore b/psp-stats-cli/.gitignore new file mode 100644 index 0000000..e660fd9 --- /dev/null +++ b/psp-stats-cli/.gitignore @@ -0,0 +1 @@ +bin/ diff --git a/psp-stats-cli/Makefile b/psp-stats-cli/Makefile new file mode 100644 index 0000000..341ea8d --- /dev/null +++ b/psp-stats-cli/Makefile @@ -0,0 +1,10 @@ +.PHONY: build test clean + +build: + go build -o bin/psp-stats-cli . + +test: + go test ./... + +clean: + rm -rf bin/ diff --git a/psp-stats-cli/README.md b/psp-stats-cli/README.md new file mode 100644 index 0000000..d95c833 --- /dev/null +++ b/psp-stats-cli/README.md @@ -0,0 +1,59 @@ +# psp-stats-cli + +CLI для запросов к PSP Speedtest Stats (https://stats.projectshitpost.fun/). + +## Установка + +``` +go install github.com/bergabruh/psp-stats-cli@latest +``` + +или собрать из исходников: + +``` +git clone +cd psp-stats-cli && make build +./bin/psp-stats-cli help +``` + +## Команды + +### recent + +``` +psp-stats-cli recent --node fl --limit 10 +psp-stats-cli recent --since 2026-05-01T00:00:00Z +``` + +Показывает последние N samples в текстовом формате `timestamp node http=X.Xms tls=...`. + +Флаги: +- `--node fl|pl1` — фильтр по узлу (пусто = оба) +- `--limit N` — максимум N записей (по умолчанию 20, макс 1000) +- `--since RFC3339` — нижняя граница по времени +- `--server host:port` — gRPC-эндпоинт (по умолчанию `stats.projectshitpost.fun:443`) +- `--timeout duration` — таймаут запроса (по умолчанию 10s) + +### aggregate + +``` +psp-stats-cli aggregate --node pl1 --since 2026-05-01T00:00:00Z +``` + +Показывает гистограмму HTTP-latency по бакетам {25, 50, 100, 250, 500, 1000, 5000} ms. + +Флаги: +- `--node fl|pl1` — фильтр по узлу (пусто = оба) +- `--since RFC3339` — нижняя граница (по умолчанию 24 часа назад) +- `--until RFC3339` — верхняя граница (по умолчанию сейчас) +- `--server host:port` — gRPC-эндпоинт +- `--timeout duration` — таймаут запроса (по умолчанию 10s) + +## Сервер + +По умолчанию `stats.projectshitpost.fun:443`. Override через `--server host:port`. +Используется TLS 1.3 + HTTP/2 + gRPC. Public методы аутентификации не требуют. + +## API + +См. [`stats-gateway` proto](https://gitea.projectshitpost.fun/bergabruh/stats-gateway). diff --git a/psp-stats-cli/cmd/aggregate.go b/psp-stats-cli/cmd/aggregate.go new file mode 100644 index 0000000..608a5bf --- /dev/null +++ b/psp-stats-cli/cmd/aggregate.go @@ -0,0 +1,64 @@ +package cmd + +import ( + "context" + "crypto/tls" + "flag" + "fmt" + "log" + "strings" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + + statsv1 "github.com/bergabruh/stats-gateway/gen/stats/v1" +) + +func Aggregate(args []string) { + fs := flag.NewFlagSet("aggregate", flag.ExitOnError) + server := fs.String("server", "stats.projectshitpost.fun:443", "gRPC endpoint host:port") + node := fs.String("node", "", "fl|pl1 (empty = both)") + since := fs.String("since", "", "RFC3339 lower bound (defaults to 24h ago)") + until := fs.String("until", "", "RFC3339 upper bound (defaults to now)") + timeout := fs.Duration("timeout", 10*time.Second, "") + fs.Parse(args) + + if *since == "" { + // Default to 24h ago for convenience. + *since = time.Now().UTC().Add(-24 * time.Hour).Format(time.RFC3339) + } + + conn, err := grpc.NewClient(*server, + grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{}))) + if err != nil { + log.Fatalf("dial: %v", err) + } + defer conn.Close() + + client := statsv1.NewSpeedStatusClient(conn) + ctx, cancel := context.WithTimeout(context.Background(), *timeout) + defer cancel() + + hist, err := client.Aggregate(ctx, &statsv1.Range{ + Node: *node, + Since: *since, + Until: *until, + }) + if err != nil { + log.Fatalf("Aggregate: %v", err) + } + + fmt.Printf("Histogram of HTTP latency (ms) — node=%q since=%s\n", *node, *since) + for _, b := range hist.Buckets { + bar := "" + if b.Count > 0 { + n := int(b.Count) + if n > 60 { + n = 60 + } + bar = strings.Repeat("#", n) + } + fmt.Printf(" <= %6.0fms %5d %s\n", b.UpperMs, b.Count, bar) + } +} diff --git a/psp-stats-cli/cmd/recent.go b/psp-stats-cli/cmd/recent.go new file mode 100644 index 0000000..25f208a --- /dev/null +++ b/psp-stats-cli/cmd/recent.go @@ -0,0 +1,65 @@ +package cmd + +import ( + "context" + "crypto/tls" + "flag" + "fmt" + "io" + "log" + "os" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + + statsv1 "github.com/bergabruh/stats-gateway/gen/stats/v1" +) + +func Recent(args []string) { + fs := flag.NewFlagSet("recent", flag.ExitOnError) + server := fs.String("server", "stats.projectshitpost.fun:443", "gRPC endpoint host:port") + node := fs.String("node", "", "fl|pl1 (empty = both)") + limit := fs.Int("limit", 20, "max samples (1..1000)") + since := fs.String("since", "", "RFC3339 lower bound (optional)") + timeout := fs.Duration("timeout", 10*time.Second, "") + fs.Parse(args) + + conn, err := grpc.NewClient(*server, + grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{}))) + if err != nil { + log.Fatalf("dial: %v", err) + } + defer conn.Close() + + client := statsv1.NewSpeedStatusClient(conn) + ctx, cancel := context.WithTimeout(context.Background(), *timeout) + defer cancel() + + stream, err := client.RecentResults(ctx, &statsv1.Filter{ + Node: *node, + Limit: int32(*limit), + Since: *since, + }) + if err != nil { + log.Fatalf("RecentResults: %v", err) + } + + count := 0 + for { + s, err := stream.Recv() + if err == io.EOF { + break + } + if err != nil { + fmt.Fprintf(os.Stderr, "recv: %v\n", err) + os.Exit(1) + } + fmt.Printf("%s %-3s http=%6.1fms tls=%6.1fms tcp=%6.1fms dns=%5.1fms\n", + s.Timestamp, s.Node, s.HttpLatencyMs, s.TlsHandshakeMs, s.TcpRttMs, s.DnsResolveMs) + count++ + } + if count == 0 { + fmt.Fprintln(os.Stderr, "(no samples)") + } +} diff --git a/psp-stats-cli/go.mod b/psp-stats-cli/go.mod new file mode 100644 index 0000000..5d90f67 --- /dev/null +++ b/psp-stats-cli/go.mod @@ -0,0 +1,18 @@ +module github.com/bergabruh/psp-stats-cli + +go 1.25.0 + +require ( + github.com/bergabruh/stats-gateway v0.0.0 + google.golang.org/grpc v1.81.0 +) + +require ( + 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 + google.golang.org/protobuf v1.36.11 // indirect +) + +replace github.com/bergabruh/stats-gateway => .. diff --git a/psp-stats-cli/go.sum b/psp-stats-cli/go.sum new file mode 100644 index 0000000..5045847 --- /dev/null +++ b/psp-stats-cli/go.sum @@ -0,0 +1,38 @@ +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/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= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +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= +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= +go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= +go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= +go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= +go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= +go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= +go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= +go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= +go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= +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.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= +golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171 h1:ggcbiqK8WWh6l1dnltU4BgWGIGo+EVYxCaAPih/zQXQ= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.81.0 h1:W3G9N3KQf3BU+YuCtGKJk0CmxQNbAISICD/9AORxLIw= +google.golang.org/grpc v1.81.0/go.mod h1:xGH9GfzOyMTGIOXBJmXt+BX/V0kcdQbdcuwQ/zNw42I= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= diff --git a/psp-stats-cli/main.go b/psp-stats-cli/main.go new file mode 100644 index 0000000..f151426 --- /dev/null +++ b/psp-stats-cli/main.go @@ -0,0 +1,45 @@ +package main + +import ( + "fmt" + "os" + + "github.com/bergabruh/psp-stats-cli/cmd" +) + +func main() { + if len(os.Args) < 2 { + usage() + os.Exit(2) + } + switch os.Args[1] { + case "recent": + cmd.Recent(os.Args[2:]) + case "aggregate": + cmd.Aggregate(os.Args[2:]) + case "-h", "--help", "help": + usage() + case "version": + fmt.Println("psp-stats-cli 0.1.0") + default: + fmt.Fprintf(os.Stderr, "unknown command: %s\n\n", os.Args[1]) + usage() + os.Exit(2) + } +} + +func usage() { + fmt.Print(`psp-stats-cli — query PSP Speedtest Stats + +Usage: + psp-stats-cli [flags] + +Commands: + recent Show last N probe samples + aggregate Show histogram for time range + version Show CLI version + help Show this help + +Run 'psp-stats-cli --help' for command-specific flags. +`) +} diff --git a/stats-site/build.sh b/stats-site/build.sh new file mode 100755 index 0000000..44b2061 --- /dev/null +++ b/stats-site/build.sh @@ -0,0 +1,41 @@ +#!/usr/bin/env bash +# Renders stats-site (Hugo project) without requiring Hugo, deploys atomically +# to /var/www/stats-site/. The Hugo project uses only two template variables; +# we substitute them with sed. +# +# Usage: build.sh [--src DIR] [--dest DIR] +# --src default /opt/stats-site +# --dest default /var/www/stats-site +set -euo pipefail + +SRC="${SRC:-/opt/stats-site}" +DEST="${DEST:-/var/www/stats-site}" + +# Hugo variables (mirror config.toml). +SITE_TITLE='PSP Speedtest Stats' +SITE_DESC='Метрики латентности и времени отклика двух geo-распределённых OpenSpeedTest нод. Snapshot обновляется каждые 10 минут.' + +[ -d "$SRC" ] || { echo "src $SRC not found" >&2; exit 1; } + +TMP=$(mktemp -d) +trap "rm -rf $TMP" EXIT + +# 1. Copy static/ verbatim +cp -r "$SRC/static/." "$TMP/" + +# 2. Render layouts/index.html with sed-substitution +sed \ + -e "s|{{ \.Site\.Title }}|${SITE_TITLE}|g" \ + -e "s|{{ \.Site\.Params\.description }}|${SITE_DESC}|g" \ + "$SRC/layouts/index.html" > "$TMP/index.html" + +# 3. Atomic rsync, preserving runtime data (snapshot.json) and ownership. +# --delete removes files in DEST that aren't in TMP (cleans old assets), +# --exclude protects data/ (snapshot generator owns that). +rsync -a --delete \ + --exclude='data/snapshot.json' \ + --exclude='data/snapshot.json.tmp' \ + --chown=deploy:www-data \ + "$TMP/" "$DEST/" + +echo "site built: $(find $DEST -type f | wc -l) files in $DEST" diff --git a/stats-site/config.toml b/stats-site/config.toml new file mode 100644 index 0000000..319d19b --- /dev/null +++ b/stats-site/config.toml @@ -0,0 +1,10 @@ +baseURL = "https://stats.projectshitpost.fun/" +title = "PSP Speedtest Stats" +languageCode = "ru" +defaultContentLanguage = "ru" + +[params] + description = "Метрики латентности и времени отклика двух geo-распределённых OpenSpeedTest нод. Snapshot обновляется каждые 10 минут." + +[markup.goldmark.renderer] + unsafe = true diff --git a/stats-site/content/_index.md b/stats-site/content/_index.md new file mode 100644 index 0000000..ed79bcf --- /dev/null +++ b/stats-site/content/_index.md @@ -0,0 +1,3 @@ +--- +title: "PSP Speedtest Stats" +--- diff --git a/stats-site/layouts/index.html b/stats-site/layouts/index.html new file mode 100644 index 0000000..94a626e --- /dev/null +++ b/stats-site/layouts/index.html @@ -0,0 +1,49 @@ + + + + + + {{ .Site.Title }} — Доступность FL и PL1 + + + + + +
+

{{ .Site.Title }}

+

Network probe stats — обновляется каждые 10 минут.

+
+ +
+
+
+

FL — Финляндия

+
+

Загрузка…

+
+ +
+ +
+

PL1 — Польша

+
+

Загрузка…

+
+ +
+
+ +
+

Snapshot age: —

+

Probes: HTTP HEAD, TCP RTT, DNS resolve, TLS handshake.

+

Источник: stats-gateway (open-source).

+
+ + +
+ + + + diff --git a/stats-site/static/about/index.html b/stats-site/static/about/index.html new file mode 100644 index 0000000..4ee2cab --- /dev/null +++ b/stats-site/static/about/index.html @@ -0,0 +1,26 @@ + + + + + + About — PSP Speedtest Stats + + + +
+

About

+
+
+

Этот dashboard показывает результаты регулярных HTTP-проб двух + speedtest-нод (FL — Финляндия, PL1 — Польша) от gitea-CI/CD-узла, + раз в 5 минут. Данные хранятся в SQLite на gitea-сервере и + публикуются как snapshot.json раз в 10 минут.

+ +

Цель — мониторить деградации latency/throughput для собственного + использования. API сервиса описан в + stats-gateway.

+ +

← Назад

+
+ + diff --git a/stats-site/static/css/main.css b/stats-site/static/css/main.css new file mode 100644 index 0000000..248b7b5 --- /dev/null +++ b/stats-site/static/css/main.css @@ -0,0 +1,99 @@ +:root { + color-scheme: light dark; + --fg: #1a1a1a; + --bg: #fafafa; + --muted: #555; + --border: #ddd; + --accent: #2563eb; +} + +@media (prefers-color-scheme: dark) { + :root { + --fg: #e8e8e8; + --bg: #111; + --muted: #aaa; + --border: #333; + --accent: #60a5fa; + } +} + +* { box-sizing: border-box; } + +body { + font-family: system-ui, -apple-system, "Segoe UI", sans-serif; + max-width: 1200px; + margin: 2em auto; + padding: 0 1em; + color: var(--fg); + background: var(--bg); + line-height: 1.5; +} + +header h1 { + font-size: 1.4em; + margin-bottom: 0.2em; +} + +.subtitle { + color: var(--muted); + margin-top: 0; +} + +.nodes { + display: grid; + grid-template-columns: 1fr 1fr; + gap: 1.5em; + margin: 2em 0; +} + +@media (max-width: 720px) { + .nodes { grid-template-columns: 1fr; } +} + +.node { + padding: 1em; + border: 1px solid var(--border); + border-radius: 8px; + background: color-mix(in srgb, var(--bg) 92%, var(--fg) 8%); +} + +.node h2 { + margin-top: 0; + font-size: 1.15em; +} + +.metrics .loading { color: var(--muted); font-style: italic; } + +.metrics .row { + display: flex; + justify-content: space-between; + padding: 0.25em 0; + border-bottom: 1px dashed var(--border); +} + +.metrics .row:last-child { border-bottom: none; } + +.metrics .label { color: var(--muted); } + +.metrics .value { font-variant-numeric: tabular-nums; } + +.chart { + display: block; + width: 100%; + height: 180px; + margin-top: 1em; +} + +.meta { + color: var(--muted); + font-size: 0.9em; + margin: 2em 0; +} + +.footer-nav a { + color: var(--accent); + text-decoration: none; + margin-right: 1em; +} + +.footer-nav a:hover { text-decoration: underline; } diff --git a/stats-site/static/data/.gitkeep b/stats-site/static/data/.gitkeep new file mode 100644 index 0000000..ef0f55f --- /dev/null +++ b/stats-site/static/data/.gitkeep @@ -0,0 +1 @@ +# Snapshot files deployed at runtime by stats-snapshot.timer diff --git a/stats-site/static/js/charts.js b/stats-site/static/js/charts.js new file mode 100644 index 0000000..ea48f18 --- /dev/null +++ b/stats-site/static/js/charts.js @@ -0,0 +1,137 @@ +(function () { + 'use strict'; + + const SNAPSHOT_URL = '/data/snapshot.json'; + + async function load() { + let data; + try { + const r = await fetch(SNAPSHOT_URL, { cache: 'no-cache' }); + if (!r.ok) throw new Error('HTTP ' + r.status); + data = await r.json(); + } catch (e) { + console.error('snapshot load failed', e); + document.querySelectorAll('.metrics').forEach(el => { + el.innerHTML = '

Snapshot не загружен (' + e.message + ').

'; + }); + return; + } + + renderAge(data.generated_at); + + for (const node of ['fl', 'pl1']) { + const nd = data.nodes && data.nodes[node]; + if (!nd || !nd.recent || nd.recent.length === 0) { + renderMetrics(node, null); + continue; + } + const samples = nd.recent; + renderMetrics(node, samples); + renderChart(node, samples); + } + } + + function renderAge(generatedAt) { + const el = document.querySelector('[data-bind="age"]'); + if (!el || !generatedAt) return; + const ts = new Date(generatedAt); + const ageS = Math.round((Date.now() - ts.getTime()) / 1000); + if (ageS < 90) el.textContent = ageS + 's'; + else if (ageS < 5400) el.textContent = Math.round(ageS / 60) + 'min'; + else el.textContent = Math.round(ageS / 3600) + 'h'; + } + + function renderMetrics(node, samples) { + const block = document.querySelector('.metrics[data-node="' + node + '"]'); + if (!block) return; + if (!samples || samples.length === 0) { + block.innerHTML = '

Нет данных за окно.

'; + return; + } + // most-recent sample + const latest = samples[0]; + // p50/p95 of http_ms over the window + const httpVals = samples.map(s => s.http_ms).filter(v => typeof v === 'number').sort((a, b) => a - b); + const p50 = httpVals[Math.floor(httpVals.length * 0.5)] || 0; + const p95 = httpVals[Math.floor(httpVals.length * 0.95)] || 0; + + block.innerHTML = [ + row('Last HTTP', fmtMs(latest.http_ms)), + row('Last TLS handshake', fmtMs(latest.tls_ms)), + row('p50 HTTP (24h)', fmtMs(p50)), + row('p95 HTTP (24h)', fmtMs(p95)), + row('Samples (24h)', String(samples.length)), + ].join(''); + } + + function row(label, value) { + return '
' + label + '' + value + '
'; + } + + function fmtMs(v) { + if (typeof v !== 'number') return '—'; + if (v < 10) return v.toFixed(2) + ' ms'; + if (v < 100) return v.toFixed(1) + ' ms'; + return Math.round(v) + ' ms'; + } + + function renderChart(node, samples) { + const canvas = document.querySelector('.chart[data-node="' + node + '"]'); + if (!canvas || !canvas.getContext) return; + const ctx = canvas.getContext('2d'); + // High-DPI scaling + const dpr = window.devicePixelRatio || 1; + const cssW = canvas.clientWidth || canvas.width; + const cssH = canvas.height; + canvas.width = cssW * dpr; + canvas.height = cssH * dpr; + ctx.scale(dpr, dpr); + + ctx.clearRect(0, 0, cssW, cssH); + + const points = samples + .map(s => ({ t: new Date(s.ts).getTime(), v: s.http_ms })) + .filter(p => !isNaN(p.t) && typeof p.v === 'number') + .sort((a, b) => a.t - b.t); + if (points.length === 0) return; + + const tMin = points[0].t; + const tMax = points[points.length - 1].t; + const tSpan = Math.max(1, tMax - tMin); + const vMax = Math.max(...points.map(p => p.v)) * 1.1 || 1; + + // axes + const pad = 24; + const w = cssW - pad * 2; + const h = cssH - pad * 2; + const muted = getComputedStyle(document.documentElement).getPropertyValue('--muted').trim() || '#555'; + const accent = getComputedStyle(document.documentElement).getPropertyValue('--accent').trim() || '#2563eb'; + + ctx.strokeStyle = muted; + ctx.lineWidth = 1; + ctx.beginPath(); + ctx.moveTo(pad, pad); + ctx.lineTo(pad, pad + h); + ctx.lineTo(pad + w, pad + h); + ctx.stroke(); + + // y-axis label (max value) + ctx.fillStyle = muted; + ctx.font = '10px system-ui, sans-serif'; + ctx.textAlign = 'right'; + ctx.fillText(Math.round(vMax) + ' ms', pad - 2, pad + 8); + ctx.fillText('0', pad - 2, pad + h); + + // points + ctx.fillStyle = accent; + for (const p of points) { + const x = pad + ((p.t - tMin) / tSpan) * w; + const y = pad + h - (p.v / vMax) * h; + ctx.beginPath(); + ctx.arc(x, y, 1.5, 0, Math.PI * 2); + ctx.fill(); + } + } + + document.addEventListener('DOMContentLoaded', load); +})(); diff --git a/systemd/stats-collector.service b/systemd/stats-collector.service new file mode 100644 index 0000000..cbd1166 --- /dev/null +++ b/systemd/stats-collector.service @@ -0,0 +1,12 @@ +[Unit] +Description=stats-gateway probe collector +After=network-online.target +Wants=network-online.target + +[Service] +Type=oneshot +User=stats +Group=stats +ExecStart=/usr/local/bin/stats-collector --db /var/lib/stats-gateway/stats.db +StandardOutput=journal +StandardError=journal diff --git a/systemd/stats-collector.timer b/systemd/stats-collector.timer new file mode 100644 index 0000000..9a771e3 --- /dev/null +++ b/systemd/stats-collector.timer @@ -0,0 +1,11 @@ +[Unit] +Description=stats-gateway probe every 5 minutes + +[Timer] +OnBootSec=2min +OnUnitActiveSec=5min +RandomizedDelaySec=30s +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/systemd/stats-gateway.service b/systemd/stats-gateway.service new file mode 100644 index 0000000..9c4094c --- /dev/null +++ b/systemd/stats-gateway.service @@ -0,0 +1,24 @@ +[Unit] +Description=stats-gateway gRPC service +After=network.target +Requires=network.target + +[Service] +Type=simple +User=stats +Group=stats +ExecStart=/usr/local/bin/stats-gateway --addr 127.0.0.1:50051 --db /var/lib/stats-gateway/stats.db +Restart=on-failure +RestartSec=5s +StandardOutput=journal +StandardError=journal + +# hardening +NoNewPrivileges=true +ProtectSystem=strict +ReadWritePaths=/var/lib/stats-gateway +ProtectHome=true +PrivateTmp=true + +[Install] +WantedBy=multi-user.target diff --git a/systemd/stats-site-build.service b/systemd/stats-site-build.service new file mode 100644 index 0000000..63ce6c1 --- /dev/null +++ b/systemd/stats-site-build.service @@ -0,0 +1,15 @@ +[Unit] +Description=Render stats-site templates and deploy to /var/www/stats-site + +[Service] +Type=oneshot +User=root +Group=root +ExecStart=/opt/stats-site/build.sh +StandardOutput=journal +StandardError=journal +NoNewPrivileges=true +ProtectSystem=strict +ReadWritePaths=/var/www/stats-site +ProtectHome=true +PrivateTmp=true diff --git a/systemd/stats-site-build.timer b/systemd/stats-site-build.timer new file mode 100644 index 0000000..956fcf1 --- /dev/null +++ b/systemd/stats-site-build.timer @@ -0,0 +1,11 @@ +[Unit] +Description=Run stats-site rebuild hourly + +[Timer] +OnBootSec=5min +OnUnitActiveSec=1h +RandomizedDelaySec=2min +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/systemd/stats-snapshot.service b/systemd/stats-snapshot.service new file mode 100644 index 0000000..02918ad --- /dev/null +++ b/systemd/stats-snapshot.service @@ -0,0 +1,19 @@ +[Unit] +Description=Generate stats snapshot.json for stats-site dashboard +After=stats-gateway.service +Wants=stats-gateway.service + +[Service] +Type=oneshot +User=stats +Group=www-data +ExecStart=/usr/local/bin/stats-snapshot --db /var/lib/stats-gateway/stats.db --out /var/www/stats-site/data/snapshot.json +StandardOutput=journal +StandardError=journal + +# Need write access to /var/www/stats-site/data +ReadWritePaths=/var/www/stats-site/data +NoNewPrivileges=true +ProtectSystem=strict +ProtectHome=true +PrivateTmp=true diff --git a/systemd/stats-snapshot.timer b/systemd/stats-snapshot.timer new file mode 100644 index 0000000..1479c06 --- /dev/null +++ b/systemd/stats-snapshot.timer @@ -0,0 +1,11 @@ +[Unit] +Description=Run stats snapshot every 10 minutes + +[Timer] +OnBootSec=3min +OnUnitActiveSec=10min +RandomizedDelaySec=30s +Persistent=true + +[Install] +WantedBy=timers.target