From ab8751d436c7ccc05012c2f3786dd96302c89926 Mon Sep 17 00:00:00 2001 From: iRedTea Date: Tue, 13 Oct 2026 23:06:56 +0300 Subject: [PATCH] feat: live reading during the speed test, and a 15 s limit that keeps what was measured --- internal/proto/proto.go | 3 + internal/server/handlers_panel_stat.go | 180 ++++++++++++++++++++-- internal/server/server.go | 2 + internal/server/web/templates/status.html | 152 +++++++++++++++--- 4 files changed, 305 insertions(+), 32 deletions(-) diff --git a/internal/proto/proto.go b/internal/proto/proto.go index a17ed1b..30080b0 100644 --- a/internal/proto/proto.go +++ b/internal/proto/proto.go @@ -193,4 +193,7 @@ type SysSpeedResult struct { DownBps float64 `json:"down_bps"` // bits per second UpBps float64 `json:"up_bps"` Relay bool `json:"relay,omitempty"` // the streams went through the relay + // Partial marks a test the server cut short: the numbers are what it had + // counted on its own side by then. + Partial bool `json:"partial,omitempty"` } diff --git a/internal/server/handlers_panel_stat.go b/internal/server/handlers_panel_stat.go index 4eeb3fb..f51b9de 100644 --- a/internal/server/handlers_panel_stat.go +++ b/internal/server/handlers_panel_stat.go @@ -20,8 +20,9 @@ const ( // statCacheFor is how long one sample answers every open status page, so // several tabs cost the agent a single reading. statCacheFor = 2 * time.Second - // speedTaskTimeout covers the agent's ping + down + up windows with slack. - speedTaskTimeout = 30 * time.Second + // speedTaskTimeout is how long a speed test may take. One that is not done + // by then is answered with what the server has counted itself. + speedTaskTimeout = 15 * time.Second // speedDownCap bounds one download stream; speedDownFor ends it if the // agent does not hang up first. speedDownCap = 256 << 20 @@ -141,6 +142,8 @@ func (s *Server) handleAgentSpeed(w http.ResponseWriter, r *http.Request) { return } relay := r.URL.Query().Get("relay") != "" + m := s.meters.start(id) + defer s.meters.stop(id, m) var args []byte if relay { rl := s.relays.get(id) @@ -153,8 +156,14 @@ func (s *Server) handleAgentSpeed(w http.ResponseWriter, r *http.Request) { } res, err := s.runTask(id, proto.TaskSysSpeed, args, speedTaskTimeout) if err != nil { - writeJSON(w, proto.TaskResult{Error: err.Error()}) - return + part, ok := m.result() + if !ok { + writeJSON(w, proto.TaskResult{Error: err.Error()}) + return + } + part.Relay = relay + data, _ := json.Marshal(part) + res = proto.TaskResult{OK: true, Data: data} } res = oldAgentError(res) if relay && res.OK { @@ -168,6 +177,148 @@ func (s *Server) handleAgentSpeed(w http.ResponseWriter, r *http.Request) { writeJSON(w, res) } +// speedMeter counts what flows through the streams of one speed test as the +// server sees it: for the live reading on the page, and for the result of a +// test that ran out of time. +type speedMeter struct { + mu sync.Mutex + start time.Time + phase string // "ping", "down", "up" + pingAt time.Time + pingMs float64 + down speedCount + up speedCount +} + +// speedCount is the bytes of one direction and when they went through. The +// first piece only starts the clock. +type speedCount struct { + n, skip int64 + first, last time.Time +} + +func (c *speedCount) add(n int) { + now := time.Now() + if c.first.IsZero() { + c.first, c.skip = now, int64(n) + } + c.n += int64(n) + c.last = now +} + +func (c *speedCount) bps() float64 { + d := c.last.Sub(c.first).Seconds() + if d <= 0 { + return 0 + } + return float64(c.n-c.skip) * 8 / d +} + +// ping notes one of the agent's ping probes: they come one right after +// another, so the gap between two is its round trip. +func (m *speedMeter) ping() { + if m == nil { + return + } + m.mu.Lock() + now := time.Now() + if !m.pingAt.IsZero() { + if ms := float64(now.Sub(m.pingAt).Microseconds()) / 1000; m.pingMs == 0 || ms < m.pingMs { + m.pingMs = ms + } + } + m.pingAt = now + m.mu.Unlock() +} + +func (m *speedMeter) add(up bool, n int) { + if m == nil || n <= 0 { + return + } + m.mu.Lock() + if up { + m.phase = "up" + m.up.add(n) + } else { + m.phase = "down" + m.down.add(n) + } + m.mu.Unlock() +} + +// result is the test as counted so far; false with nothing measured yet. +func (m *speedMeter) result() (proto.SysSpeedResult, bool) { + m.mu.Lock() + defer m.mu.Unlock() + res := proto.SysSpeedResult{PingMs: m.pingMs, DownBps: m.down.bps(), UpBps: m.up.bps(), Partial: true} + return res, res.DownBps > 0 +} + +// meterWriter counts what it passes on; a nil meter counts nothing. +type meterWriter struct { + w io.Writer + m *speedMeter + up bool +} + +func (c meterWriter) Write(p []byte) (int, error) { + n, err := c.w.Write(p) + c.m.add(c.up, n) + return n, err +} + +// speedMeters holds the meter of each agent under test. +type speedMeters struct { + mu sync.Mutex + m map[string]*speedMeter +} + +func (c *speedMeters) start(agentID string) *speedMeter { + m := &speedMeter{start: time.Now(), phase: "ping"} + c.mu.Lock() + if c.m == nil { + c.m = map[string]*speedMeter{} + } + c.m[agentID] = m + c.mu.Unlock() + return m +} + +func (c *speedMeters) get(agentID string) *speedMeter { + c.mu.Lock() + defer c.mu.Unlock() + return c.m[agentID] +} + +// stop forgets the agent's meter if it is still m. +func (c *speedMeters) stop(agentID string, m *speedMeter) { + c.mu.Lock() + if c.m[agentID] == m { + delete(c.m, agentID) + } + c.mu.Unlock() +} + +// GET /api/panel/agents/{id}/speed/progress - the bytes counted so far in the +// agent's running speed test, for the page to turn into a live rate. An empty +// phase means no test is running. +func (s *Server) handleSpeedProgress(w http.ResponseWriter, r *http.Request) { + var out struct { + Phase string `json:"phase"` + Down int64 `json:"down"` + Up int64 `json:"up"` + Ms int64 `json:"ms"` // since the test began, by the server's clock + } + if m := s.meters.get(r.PathValue("id")); m != nil { + m.mu.Lock() + out.Phase, out.Down, out.Up = m.phase, m.down.n, m.up.n + out.Ms = time.Since(m.start).Milliseconds() + m.mu.Unlock() + } + w.Header().Set("Cache-Control", "no-store") + writeJSON(w, out) +} + // speedBlock is one chunk of incompressible bytes, made once and sent over and // over, so a speed test costs the server no CPU beyond the copy. var speedBlock = sync.OnceValue(func() []byte { @@ -186,16 +337,21 @@ func speedStreamHeaders(w http.ResponseWriter) { // GET /api/agent/speed/down?bytes=N - stream N bytes (default: as much as fits // in the time window) for the agent to time. bytes=0 is the ping probe. func (s *Server) handleSpeedDown(w http.ResponseWriter, r *http.Request) { - if s.authenticateAgent(r) == nil { + a := s.authenticateAgent(r) + if a == nil { http.Error(w, "unauthorized", http.StatusUnauthorized) return } + m := s.meters.get(a.ID) left := speedDownCap if v := r.URL.Query().Get("bytes"); v != "" { if n, err := strconv.Atoi(v); err == nil && n >= 0 && n < left { left = n } } + if left == 0 { + m.ping() + } speedStreamHeaders(w) block := speedBlock() deadline := time.Now().Add(speedDownFor) @@ -208,17 +364,20 @@ func (s *Server) handleSpeedDown(w http.ResponseWriter, r *http.Request) { if err != nil { return } + m.add(false, n) left -= n } } // POST /api/agent/speed/up - swallow the agent's upload. func (s *Server) handleSpeedUp(w http.ResponseWriter, r *http.Request) { - if s.authenticateAgent(r) == nil { + a := s.authenticateAgent(r) + if a == nil { http.Error(w, "unauthorized", http.StatusUnauthorized) return } - _, _ = io.Copy(io.Discard, http.MaxBytesReader(w, r.Body, speedUpCap)) + sink := meterWriter{w: io.Discard, m: s.meters.get(a.ID), up: true} + _, _ = io.Copy(sink, http.MaxBytesReader(w, r.Body, speedUpCap)) w.WriteHeader(http.StatusNoContent) } @@ -323,13 +482,14 @@ func (s *Server) handleSpeedRelayFeed(w http.ResponseWriter, r *http.Request) { // GET /api/panel/agents/{id}/speed/up - the agent's upload, for the browser to // time. Ends when the agent is done sending. func (s *Server) handleSpeedRelayDrain(w http.ResponseWriter, r *http.Request) { - rl := s.relays.get(r.PathValue("id")) + id := r.PathValue("id") + rl := s.relays.get(id) if rl == nil { http.Error(w, "gone", http.StatusGone) return } speedStreamHeaders(w) - _, _ = io.Copy(w, rl.upR) + _, _ = io.Copy(meterWriter{w: w, m: s.meters.get(id), up: true}, rl.upR) // The browser may have left: don't let the agent keep sending to nobody. rl.upR.CloseWithError(io.ErrClosedPipe) } @@ -350,7 +510,7 @@ func (s *Server) handleSpeedRelayDown(w http.ResponseWriter, r *http.Request) { speedStreamHeaders(w) // The agent waits on the headers, the browser may not have sent a byte yet. _ = http.NewResponseController(w).Flush() - _, _ = io.Copy(w, rl.downR) + _, _ = io.Copy(meterWriter{w: w, m: s.meters.get(a.ID)}, rl.downR) // The agent hung up at the end of its window: stop the browser's upload. rl.downR.CloseWithError(io.ErrClosedPipe) } diff --git a/internal/server/server.go b/internal/server/server.go index 972595b..cfea288 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -55,6 +55,7 @@ type Server struct { stats statCache // last load sample per agent, for the status page relays speedRelays // speed tests that run through the browser + meters speedMeters // byte counts of the speed tests under way } // New builds the server, parsing embedded templates. @@ -234,6 +235,7 @@ func (s *Server) Handler() http.Handler { mux.Handle("GET /api/panel/agents/presence", s.requireAdmin(s.handlePresence)) mux.Handle("GET /api/panel/agents/{id}/stat", s.requireAdmin(s.handleAgentStat)) mux.Handle("POST /api/panel/agents/{id}/speed", s.requireAdmin(s.handleAgentSpeed)) + mux.Handle("GET /api/panel/agents/{id}/speed/progress", s.requireAdmin(s.handleSpeedProgress)) mux.Handle("GET /api/panel/ping", s.requireAdmin(s.handlePanelPing)) mux.Handle("POST /api/panel/agents/{id}/speed/relay", s.requireAdmin(s.handleSpeedRelayOpen)) mux.Handle("POST /api/panel/agents/{id}/speed/down", s.requireAdmin(s.handleSpeedRelayFeed)) diff --git a/internal/server/web/templates/status.html b/internal/server/web/templates/status.html index 6a0954c..6d90b4f 100644 --- a/internal/server/web/templates/status.html +++ b/internal/server/web/templates/status.html @@ -19,6 +19,24 @@ .speed { display:flex; gap:28px; flex-wrap:wrap; margin-top:16px; } .speed b { display:block; font-size:20px; font-variant-numeric:tabular-nums; } .speed span { color:var(--muted); font-size:13px; } + .live { display:flex; align-items:center; gap:28px; flex-wrap:wrap; margin-top:20px; } + .gauge { position:relative; width:240px; flex:none; } + .gauge svg { display:block; width:100%; } + .gauge path { fill:none; stroke-width:14; stroke-linecap:round; } + .gauge .track { stroke:#0c0f13; } + .gauge .fill { stroke:var(--accent); stroke-dasharray:251.4; stroke-dashoffset:251.4; transition:stroke-dashoffset .25s linear, stroke .3s; } + .gauge .num { position:absolute; left:0; right:0; bottom:26px; text-align:center; font-size:38px; font-weight:600; line-height:1; font-variant-numeric:tabular-nums; } + .gauge .unit { position:absolute; left:0; right:0; bottom:4px; text-align:center; color:var(--muted); font-size:13px; } + .live .side { flex:1; min-width:220px; } + .live .phase { font-size:15px; margin-bottom:8px; } + .live .phase i { font-style:normal; color:var(--accent); margin-right:6px; } + .live.up .phase i { color:var(--ok); } + .live.up .gauge .fill { stroke:var(--ok); } + .spark { display:block; width:100%; height:64px; } + .spark polyline { fill:none; stroke:var(--accent); stroke-width:2; vector-effect:non-scaling-stroke; stroke-linejoin:round; } + .spark polygon { fill:var(--accent); opacity:.15; } + .live.up .spark polyline { stroke:var(--ok); } + .live.up .spark polygon { fill:var(--ok); } label.check { display:inline-flex; align-items:center; gap:8px; margin:0; color:var(--text); font-size:14px; cursor:pointer; } @@ -59,7 +77,7 @@

Скорость соединения

Клиент замеряет свой канал до хоста панели - тот же, по которому идут терминал, файлы и http-проброс. С галочкой «До браузера» данные идут дальше, через панель до этого браузера и обратно: так виден весь путь клиент - панель - браузер. - Замер занимает около 7 секунд и на это время нагружает канал.

+ Замер занимает около 7 секунд, не дольше 20, и на это время нагружает канал.

+