diff --git a/VERSION.txt b/VERSION.txt index bbdeab6..1750564 100644 --- a/VERSION.txt +++ b/VERSION.txt @@ -1 +1 @@ -0.0.5 +0.0.6 diff --git a/deploy/nginx.conf.example b/deploy/nginx.conf.example index ad9b658..4ecf2ad 100644 --- a/deploy/nginx.conf.example +++ b/deploy/nginx.conf.example @@ -36,6 +36,9 @@ server { # Long-poll behaviour: no buffering, generous read timeout. proxy_buffering off; proxy_cache off; + # Pass uploads on as they arrive: the speed test through the browser + # times them on the client. + proxy_request_buffering off; proxy_read_timeout 120s; proxy_send_timeout 120s; } diff --git a/internal/agent/speed.go b/internal/agent/speed.go index 93e6114..bd4df53 100644 --- a/internal/agent/speed.go +++ b/internal/agent/speed.go @@ -4,6 +4,7 @@ import ( "context" "crypto/rand" "encoding/json" + "errors" "io" "net/http" "time" @@ -14,14 +15,22 @@ import ( const ( // speedWindow is how long each direction is measured for. speedWindow = 3 * time.Second + // speedWait is how long the download may take to start: in a relayed test + // its bytes come from the browser. + speedWait = 10 * time.Second // speedUpCap bounds the upload, whatever the link does in speedWindow. speedUpCap = 48 << 20 ) // doSpeed measures the agent's HTTP link to the server: round-trip time of an -// empty request, then how much flows each way in speedWindow. -func (a *Agent) doSpeed(ctx context.Context) proto.TaskResult { - res, err := a.measureSpeed(ctx) +// empty request, then how much flows each way in speedWindow. In a relayed +// test the server pipes both streams to the browser, so they run end to end. +func (a *Agent) doSpeed(ctx context.Context, env proto.TaskEnvelope) proto.TaskResult { + var args proto.SysSpeedArgs + if len(env.Args) > 0 { + _ = json.Unmarshal(env.Args, &args) + } + res, err := a.measureSpeed(ctx, args.Relay) if err != nil { return proto.TaskResult{Error: err.Error()} } @@ -29,8 +38,12 @@ func (a *Agent) doSpeed(ctx context.Context) proto.TaskResult { return proto.TaskResult{OK: true, Data: data} } -func (a *Agent) measureSpeed(ctx context.Context) (proto.SysSpeedResult, error) { - var res proto.SysSpeedResult +func (a *Agent) measureSpeed(ctx context.Context, relay bool) (proto.SysSpeedResult, error) { + res := proto.SysSpeedResult{Relay: relay} + streams := "/api/agent/speed" + if relay { + streams += "/relay" + } // Ping: the first request may pay for a new connection, so keep the best. for i := 0; i < 4; i++ { start := time.Now() @@ -46,24 +59,16 @@ func (a *Agent) measureSpeed(ctx context.Context) (proto.SysSpeedResult, error) } } - // Down: read what the server streams until the window closes. - dctx, cancel := context.WithTimeout(ctx, speedWindow) - resp, err := a.speedRequest(dctx, http.MethodGet, "/api/agent/speed/down", nil) + down, err := a.speedDown(ctx, streams+"/down") if err != nil { - cancel() return res, err } - start := time.Now() - n, _ := io.Copy(io.Discard, resp.Body) - elapsed := time.Since(start) - resp.Body.Close() - cancel() - res.DownBps = bps(n, elapsed) + res.DownBps = down // Up: send until the window closes or the cap is hit. body := &speedSource{deadline: time.Now().Add(speedWindow), left: speedUpCap} - start = time.Now() - resp, err = a.speedRequest(ctx, http.MethodPost, "/api/agent/speed/up", body) + start := time.Now() + resp, err := a.speedRequest(ctx, http.MethodPost, streams+"/up", body) if err != nil { return res, err } @@ -73,6 +78,40 @@ func (a *Agent) measureSpeed(ctx context.Context) (proto.SysSpeedResult, error) return res, nil } +// speedDown reads what the server streams for speedWindow, counted from the +// first byte, and returns the rate. +func (a *Agent) speedDown(ctx context.Context, path string) (float64, error) { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + t := time.AfterFunc(speedWait, cancel) + defer t.Stop() + resp, err := a.speedRequest(ctx, http.MethodGet, path, nil) + if err != nil { + return 0, err + } + defer resp.Body.Close() + var ( + n int64 + start time.Time + buf = make([]byte, 64<<10) + ) + for { + k, err := resp.Body.Read(buf) + if k > 0 && start.IsZero() { + start = time.Now() + t.Reset(speedWindow) + } + n += int64(k) + if err != nil { + break + } + } + if start.IsZero() { + return 0, errors.New("нет данных для замера") + } + return bps(n, time.Since(start)), nil +} + func (a *Agent) speedRequest(ctx context.Context, method, path string, body io.Reader) (*http.Response, error) { req, err := http.NewRequestWithContext(ctx, method, a.cfg.Host+path, body) if err != nil { diff --git a/internal/agent/task.go b/internal/agent/task.go index c6778c9..35a7055 100644 --- a/internal/agent/task.go +++ b/internal/agent/task.go @@ -21,7 +21,7 @@ func (a *Agent) runTask(ctx context.Context, taskID string) { case proto.TaskSysUpdate: res = a.doUpdate() case proto.TaskSysSpeed: - res = a.doSpeed(ctx) + res = a.doSpeed(ctx, env) default: res = a.execTask(env) } diff --git a/internal/proto/proto.go b/internal/proto/proto.go index cd3008b..a17ed1b 100644 --- a/internal/proto/proto.go +++ b/internal/proto/proto.go @@ -179,10 +179,18 @@ type SysStatResult struct { DiskFree uint64 `json:"disk_free"` } +// SysSpeedArgs are the arguments of TaskSysSpeed. +type SysSpeedArgs struct { + // Relay makes the agent use the relayed streams, which the server pipes + // to and from the browser. + Relay bool `json:"relay,omitempty"` +} + // SysSpeedResult is returned for TaskSysSpeed: the agent's HTTP link to the // server as the agent measured it. Down is server -> agent. type SysSpeedResult struct { PingMs float64 `json:"ping_ms"` DownBps float64 `json:"down_bps"` // bits per second UpBps float64 `json:"up_bps"` + Relay bool `json:"relay,omitempty"` // the streams went through the relay } diff --git a/internal/server/handlers_panel.go b/internal/server/handlers_panel.go index 0d84280..27c41b0 100644 --- a/internal/server/handlers_panel.go +++ b/internal/server/handlers_panel.go @@ -185,6 +185,7 @@ func (s *Server) handleAgentDelete(w http.ResponseWriter, r *http.Request) { } _ = s.db.DeleteAgent(id) s.stats.drop(id) + s.relays.drop(id, nil) s.hub.Forget(id) http.Redirect(w, r, "/", http.StatusSeeOther) } diff --git a/internal/server/handlers_panel_stat.go b/internal/server/handlers_panel_stat.go index fac0f84..42d8575 100644 --- a/internal/server/handlers_panel_stat.go +++ b/internal/server/handlers_panel_stat.go @@ -2,6 +2,7 @@ package server import ( "crypto/rand" + "encoding/json" "errors" "io" "net/http" @@ -70,8 +71,8 @@ func (s *Server) handleStatusPage(w http.ResponseWriter, r *http.Request) { } // runTask runs a one-shot task on the agent and waits for its result. -func (s *Server) runTask(agentID, typ string, timeout time.Duration) (proto.TaskResult, error) { - taskID, ch := s.hub.CreateTask(agentID, typ, nil) +func (s *Server) runTask(agentID, typ string, args []byte, timeout time.Duration) (proto.TaskResult, error) { + taskID, ch := s.hub.CreateTask(agentID, typ, args) defer s.hub.DiscardTask(taskID) t := time.NewTimer(timeout) defer t.Stop() @@ -118,7 +119,7 @@ func (s *Server) handleAgentStat(w http.ResponseWriter, r *http.Request) { e.mu.Lock() defer e.mu.Unlock() if time.Since(e.at) > statCacheFor { - res, err := s.runTask(id, proto.TaskSysStat, statTaskTimeout) + res, err := s.runTask(id, proto.TaskSysStat, nil, statTaskTimeout) if err != nil { // Cached too, so pages queued behind this one don't each wait it out. res = proto.TaskResult{Error: err.Error()} @@ -129,17 +130,39 @@ func (s *Server) handleAgentStat(w http.ResponseWriter, r *http.Request) { } // POST /api/panel/agents/{id}/speed - have the agent measure its link to us. +// With relay=1 its streams run through the relay opened for it, to and from +// the browser. func (s *Server) handleAgentSpeed(w http.ResponseWriter, r *http.Request) { id, ok := s.onlineAgent(w, r) if !ok { return } - res, err := s.runTask(id, proto.TaskSysSpeed, speedTaskTimeout) + relay := r.URL.Query().Get("relay") != "" + var args []byte + if relay { + rl := s.relays.get(id) + if rl == nil { + writeJSON(w, proto.TaskResult{Error: "замер через браузер не открыт"}) + return + } + defer s.relays.drop(id, rl) + args, _ = json.Marshal(proto.SysSpeedArgs{Relay: true}) + } + res, err := s.runTask(id, proto.TaskSysSpeed, args, speedTaskTimeout) if err != nil { writeJSON(w, proto.TaskResult{Error: err.Error()}) return } - writeJSON(w, oldAgentError(res)) + res = oldAgentError(res) + if relay && res.OK { + // An agent that predates the relay ignores the argument and measures + // its own link. + var d proto.SysSpeedResult + if json.Unmarshal(res.Data, &d) != nil || !d.Relay { + res = proto.TaskResult{Error: "клиент не умеет замер через браузер - обновите его (нужен 0.0.6+)"} + } + } + writeJSON(w, res) } // speedBlock is one chunk of incompressible bytes, made once and sent over and @@ -150,6 +173,13 @@ var speedBlock = sync.OnceValue(func() []byte { return b }) +func speedStreamHeaders(w http.ResponseWriter) { + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Cache-Control", "no-store") + // Keep a proxy in front from buffering or compressing the stream. + w.Header().Set("X-Accel-Buffering", "no") +} + // 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) { @@ -163,10 +193,7 @@ func (s *Server) handleSpeedDown(w http.ResponseWriter, r *http.Request) { left = n } } - w.Header().Set("Content-Type", "application/octet-stream") - w.Header().Set("Cache-Control", "no-store") - // Keep a proxy in front from buffering or compressing the stream. - w.Header().Set("X-Accel-Buffering", "no") + speedStreamHeaders(w) block := speedBlock() deadline := time.Now().Add(speedDownFor) for left > 0 && time.Now().Before(deadline) && r.Context().Err() == nil { @@ -191,3 +218,148 @@ func (s *Server) handleSpeedUp(w http.ResponseWriter, r *http.Request) { _, _ = io.Copy(io.Discard, http.MaxBytesReader(w, r.Body, speedUpCap)) w.WriteHeader(http.StatusNoContent) } + +// speedRelay joins the agent's speed test streams to the browser's, so the +// bytes travel browser - panel - client. Both pipes are unbuffered: a slow +// reader holds its writer back, and the rate is that of the whole path. +type speedRelay struct { + downR *io.PipeReader // browser -> agent + downW *io.PipeWriter + upR *io.PipeReader // agent -> browser + upW *io.PipeWriter +} + +func (rl *speedRelay) close() { + rl.downR.CloseWithError(io.ErrClosedPipe) + rl.downW.CloseWithError(io.ErrClosedPipe) + rl.upR.CloseWithError(io.ErrClosedPipe) + rl.upW.CloseWithError(io.ErrClosedPipe) +} + +// speedRelays holds the relay of each agent under test, one per agent. +type speedRelays struct { + mu sync.Mutex + m map[string]*speedRelay +} + +// open starts a fresh relay for the agent, ending the one it had. +func (c *speedRelays) open(agentID string) { + rl := &speedRelay{} + rl.downR, rl.downW = io.Pipe() + rl.upR, rl.upW = io.Pipe() + c.mu.Lock() + if c.m == nil { + c.m = map[string]*speedRelay{} + } + old := c.m[agentID] + c.m[agentID] = rl + c.mu.Unlock() + if old != nil { + old.close() + } +} + +func (c *speedRelays) get(agentID string) *speedRelay { + c.mu.Lock() + defer c.mu.Unlock() + return c.m[agentID] +} + +// drop ends the agent's relay; with rl given, only if it is still that one. +func (c *speedRelays) drop(agentID string, rl *speedRelay) { + c.mu.Lock() + cur := c.m[agentID] + if cur != nil && (rl == nil || rl == cur) { + delete(c.m, agentID) + } else { + cur = nil + } + c.mu.Unlock() + if cur != nil { + cur.close() + } +} + +// GET /api/panel/ping - an empty answer for the browser to time. +func (s *Server) handlePanelPing(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-store") + w.WriteHeader(http.StatusNoContent) +} + +// POST /api/panel/agents/{id}/speed/relay - open the relay of a speed test +// that runs through the browser. The test itself ends it. +func (s *Server) handleSpeedRelayOpen(w http.ResponseWriter, r *http.Request) { + id, ok := s.onlineAgent(w, r) + if !ok { + return + } + s.relays.open(id) + writeJSON(w, proto.TaskResult{OK: true}) +} + +// POST /api/panel/agents/{id}/speed/down - a piece of the browser's upload, +// passed on to the agent. 410 once the agent has stopped reading. +func (s *Server) handleSpeedRelayFeed(w http.ResponseWriter, r *http.Request) { + rl := s.relays.get(r.PathValue("id")) + if rl == nil { + http.Error(w, "gone", http.StatusGone) + return + } + if _, err := io.Copy(rl.downW, http.MaxBytesReader(w, r.Body, speedUpCap)); err != nil { + http.Error(w, "gone", http.StatusGone) + return + } + w.WriteHeader(http.StatusNoContent) +} + +// 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")) + if rl == nil { + http.Error(w, "gone", http.StatusGone) + return + } + speedStreamHeaders(w) + _, _ = io.Copy(w, rl.upR) + // The browser may have left: don't let the agent keep sending to nobody. + rl.upR.CloseWithError(io.ErrClosedPipe) +} + +// GET /api/agent/speed/relay/down - what the browser uploads, for the agent to +// time. +func (s *Server) handleSpeedRelayDown(w http.ResponseWriter, r *http.Request) { + a := s.authenticateAgent(r) + if a == nil { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + rl := s.relays.get(a.ID) + if rl == nil { + http.NotFound(w, r) + return + } + 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) + // The agent hung up at the end of its window: stop the browser's upload. + rl.downR.CloseWithError(io.ErrClosedPipe) +} + +// POST /api/agent/speed/relay/up - the agent's upload, passed on to the browser. +func (s *Server) handleSpeedRelayUp(w http.ResponseWriter, r *http.Request) { + a := s.authenticateAgent(r) + if a == nil { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + rl := s.relays.get(a.ID) + if rl == nil { + http.NotFound(w, r) + return + } + _, _ = io.Copy(rl.upW, http.MaxBytesReader(w, r.Body, speedUpCap)) + rl.upW.Close() + w.WriteHeader(http.StatusNoContent) +} diff --git a/internal/server/server.go b/internal/server/server.go index 24a9475..67537b2 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -53,7 +53,8 @@ type Server struct { tunnelMu sync.Mutex // serializes port changes tunnelPort atomic.Int64 - stats statCache // last load sample per agent, for the status page + stats statCache // last load sample per agent, for the status page + relays speedRelays // speed tests that run through the browser } // New builds the server, parsing embedded templates. @@ -232,6 +233,10 @@ 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/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)) + mux.Handle("GET /api/panel/agents/{id}/speed/up", s.requireAdmin(s.handleSpeedRelayDrain)) // Panel port forwarding (admin-only). mux.Handle("POST /agents/{id}/forwards/add", s.requireAdmin(s.handleForwardAdd)) @@ -247,6 +252,8 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /api/agent/task/{id}/result", s.handleTaskResult) mux.HandleFunc("GET /api/agent/speed/down", s.handleSpeedDown) mux.HandleFunc("POST /api/agent/speed/up", s.handleSpeedUp) + mux.HandleFunc("GET /api/agent/speed/relay/down", s.handleSpeedRelayDown) + mux.HandleFunc("POST /api/agent/speed/relay/up", s.handleSpeedRelayUp) return s.logRequests(mux) } diff --git a/internal/server/web/templates/status.html b/internal/server/web/templates/status.html index c76d51d..4aee3fb 100644 --- a/internal/server/web/templates/status.html +++ b/internal/server/web/templates/status.html @@ -19,6 +19,7 @@ .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; } + label.check { display:inline-flex; align-items:center; gap:8px; margin:0; color:var(--text); font-size:14px; cursor:pointer; } @@ -57,15 +58,19 @@

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

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

+
@@ -134,17 +139,93 @@ load(); var btn = $('speed-btn'), note = $('speed-note'); + function json(r) { if (!r.ok) throw new Error('HTTP ' + r.status); return r.json(); } + function post(path) { return fetch(base + path, {method: 'POST', credentials: 'same-origin'}).then(json); } + + // The browser's round trip to the panel: the first request may pay for a + // new connection, so keep the best. + function panelPing() { + var best = Infinity, left = 4; + function one() { + var t = performance.now(); + return fetch('/api/panel/ping', {credentials: 'same-origin', cache: 'no-store'}) + .then(function (r) { if (!r.ok) throw new Error('HTTP ' + r.status); return r.arrayBuffer(); }) + .then(function () { + best = Math.min(best, performance.now() - t); + return --left > 0 ? one() : best; + }); + } + return one(); + } + // Keeps the client's download fed until stopped or the panel says the + // client has read enough; the pieces grow while they go through quickly. + function feed(stop) { + var block = new Uint8Array(65536), n = 4; + crypto.getRandomValues(block); + function one() { + var parts = [], t = performance.now(); + for (var i = 0; i < n; i++) parts.push(block); + return fetch(base + '/speed/down', {method: 'POST', body: new Blob(parts), credentials: 'same-origin', signal: stop.signal}) + .then(function (r) { + if (!r.ok) return; + if (performance.now() - t < 500 && n < 256) n *= 2; + return one(); + }); + } + one().catch(function () {}); + } + // Times what the client sends, from its first byte to the end of the stream. + function drain(onFirst) { + return fetch(base + '/speed/up', {credentials: 'same-origin', cache: 'no-store'}).then(function (r) { + if (!r.ok) throw new Error('HTTP ' + r.status); + var rd = r.body.getReader(), n = 0, start = 0; + function pump() { + return rd.read().then(function (c) { + if (c.done) return start && n ? n * 8000 / Math.max(1, performance.now() - start) : 0; + if (!start) { start = performance.now(); onFirst(); } + n += c.value.length; + return pump(); + }); + } + return pump(); + }); + } + // The test through the browser: the client's download is what this page + // uploads, and its upload is timed here, where it arrives. + function relayTest() { + return panelPing().then(function (ping) { + return post('/speed/relay').then(function (res) { + if (!res.ok) return res; + var stop = new AbortController(); + var up = drain(function () { stop.abort(); }); + up.catch(function () {}); + feed(stop); + return post('/speed?relay=1').then(function (res) { + stop.abort(); + if (!res.ok) return res; + return up.then(function (bps) { + res.data.ping_ms += ping; + res.data.up_bps = bps; + return res; + }); + }, function (e) { stop.abort(); throw e; }); + }); + }); + } + btn.onclick = function () { + var relay = $('speed-relay').checked; btn.disabled = true; note.textContent = 'Идёт замер…'; - fetch(base + '/speed', {method: 'POST', credentials: 'same-origin'}) - .then(function (r) { if (!r.ok) throw new Error('HTTP ' + r.status); return r.json(); }) + (relay ? relayTest() : post('/speed')) .then(function (res) { if (!res.ok) { note.textContent = res.error || 'ошибка'; return; } var d = res.data; $('sp-ping').textContent = d.ping_ms.toFixed(d.ping_ms < 10 ? 1 : 0) + ' мс'; $('sp-down').textContent = rate(d.down_bps); $('sp-up').textContent = rate(d.up_bps); + $('sp-down-lbl').textContent = relay ? 'браузер → клиент' : 'хост → клиент'; + $('sp-up-lbl').textContent = relay ? 'клиент → браузер' : 'клиент → хост'; $('speed').hidden = false; note.textContent = ''; })