366 lines
10 KiB
Go
366 lines
10 KiB
Go
package server
|
|
|
|
import (
|
|
"crypto/rand"
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/x-grk/longpolling-remote-access/internal/proto"
|
|
)
|
|
|
|
const (
|
|
// statTaskTimeout bounds how long a status request waits for the agent.
|
|
statTaskTimeout = 8 * time.Second
|
|
// 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
|
|
// speedDownCap bounds one download stream; speedDownFor ends it if the
|
|
// agent does not hang up first.
|
|
speedDownCap = 256 << 20
|
|
speedDownFor = 6 * time.Second
|
|
speedUpCap = 64 << 20
|
|
)
|
|
|
|
// statCache holds the last sample per agent and serializes its refresh.
|
|
type statCache struct {
|
|
mu sync.Mutex
|
|
m map[string]*statEntry
|
|
}
|
|
|
|
type statEntry struct {
|
|
mu sync.Mutex
|
|
at time.Time
|
|
res proto.TaskResult
|
|
}
|
|
|
|
func (c *statCache) entry(agentID string) *statEntry {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
if c.m == nil {
|
|
c.m = map[string]*statEntry{}
|
|
}
|
|
e := c.m[agentID]
|
|
if e == nil {
|
|
e = &statEntry{}
|
|
c.m[agentID] = e
|
|
}
|
|
return e
|
|
}
|
|
|
|
func (c *statCache) drop(agentID string) {
|
|
c.mu.Lock()
|
|
delete(c.m, agentID)
|
|
c.mu.Unlock()
|
|
}
|
|
|
|
func (s *Server) handleStatusPage(w http.ResponseWriter, r *http.Request) {
|
|
a, err := s.db.GetAgent(r.PathValue("id"))
|
|
if err != nil || a == nil {
|
|
http.NotFound(w, r)
|
|
return
|
|
}
|
|
s.render(w, "status.html", map[string]any{"Agent": a})
|
|
}
|
|
|
|
// runTask runs a one-shot task on the agent and waits for its result.
|
|
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()
|
|
select {
|
|
case res := <-ch:
|
|
return res, nil
|
|
case <-t.C:
|
|
return proto.TaskResult{}, errors.New("таймаут ожидания клиента")
|
|
}
|
|
}
|
|
|
|
// onlineAgent answers the request itself and returns false when the agent is
|
|
// unknown or offline.
|
|
func (s *Server) onlineAgent(w http.ResponseWriter, r *http.Request) (string, bool) {
|
|
id := r.PathValue("id")
|
|
a, err := s.db.GetAgent(id)
|
|
if err != nil || a == nil {
|
|
http.NotFound(w, r)
|
|
return "", false
|
|
}
|
|
if !s.online(a) {
|
|
writeJSON(w, proto.TaskResult{Error: "машина офлайн"})
|
|
return "", false
|
|
}
|
|
return id, true
|
|
}
|
|
|
|
// oldAgentError replaces the "unknown task" answer of an agent that predates
|
|
// the task.
|
|
func oldAgentError(res proto.TaskResult) proto.TaskResult {
|
|
if !res.OK && strings.Contains(res.Error, "неизвестная операция") {
|
|
res.Error = "клиент не поддерживает эту функцию - обновите его (нужен 0.0.5+)"
|
|
}
|
|
return res
|
|
}
|
|
|
|
// GET /api/panel/agents/{id}/stat - CPU, RAM and disk of the machine right now.
|
|
func (s *Server) handleAgentStat(w http.ResponseWriter, r *http.Request) {
|
|
id, ok := s.onlineAgent(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
e := s.stats.entry(id)
|
|
e.mu.Lock()
|
|
defer e.mu.Unlock()
|
|
if time.Since(e.at) > statCacheFor {
|
|
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()}
|
|
}
|
|
e.res, e.at = oldAgentError(res), time.Now()
|
|
}
|
|
writeJSON(w, e.res)
|
|
}
|
|
|
|
// 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
|
|
}
|
|
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
|
|
}
|
|
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
|
|
// over, so a speed test costs the server no CPU beyond the copy.
|
|
var speedBlock = sync.OnceValue(func() []byte {
|
|
b := make([]byte, 64<<10)
|
|
rand.Read(b)
|
|
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) {
|
|
if s.authenticateAgent(r) == nil {
|
|
http.Error(w, "unauthorized", http.StatusUnauthorized)
|
|
return
|
|
}
|
|
left := speedDownCap
|
|
if v := r.URL.Query().Get("bytes"); v != "" {
|
|
if n, err := strconv.Atoi(v); err == nil && n >= 0 && n < left {
|
|
left = n
|
|
}
|
|
}
|
|
speedStreamHeaders(w)
|
|
block := speedBlock()
|
|
deadline := time.Now().Add(speedDownFor)
|
|
for left > 0 && time.Now().Before(deadline) && r.Context().Err() == nil {
|
|
chunk := block
|
|
if len(chunk) > left {
|
|
chunk = chunk[:left]
|
|
}
|
|
n, err := w.Write(chunk)
|
|
if err != nil {
|
|
return
|
|
}
|
|
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 {
|
|
http.Error(w, "unauthorized", http.StatusUnauthorized)
|
|
return
|
|
}
|
|
_, _ = 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)
|
|
}
|