Something went wrong. Try again.
router flow monitor
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266package ctl
import ( "context" "encoding/json" "errors" "fmt" "io" "net" "net/netip" "os" "strconv" "sync" "syscall" "time"
"github.com/charmbracelet/log")
// requestTimeout bounds one connection so a stuck client cannot pin a// server goroutineconst requestTimeout = 5 * time.Second
// maxRequestSize bounds one request, real ones are a few hundred bytesconst maxRequestSize = 64 << 10
// maxConns bounds the connections served at once, further clients wait in// the listen backlogconst maxConns = 16
// Server answers requests on a unix sockettype Server struct { handler Handler listener net.Listener path string // created is the socket file Listen made, close unlinks the path only // while it still names this file created os.FileInfo closeOnce sync.Once wg sync.WaitGroup}
// Listen creates the socket at path, replacing a stale one left by an// earlier run, and refuses a path that is not a socket or that another// process still listens on, the socket is only reachable by the ownerfunc Listen(path string, handler Handler) (*Server, error) { if err := removeStale(path); err != nil { return nil, err } ln, err := net.ListenUnix("unix", &net.UnixAddr{Name: path, Net: "unix"}) if err != nil { return nil, fmt.Errorf("listen control socket %q: %w", path, err) } ln.SetUnlinkOnClose(false) created, err := os.Lstat(path) if err != nil { ln.Close() return nil, fmt.Errorf("stat control socket %q: %w", path, err) } s := &Server{handler: handler, listener: ln, path: path, created: created} if err := os.Chmod(path, 0o600); err != nil { s.close() return nil, fmt.Errorf("chmod control socket %q: %w", path, err) } return s, nil}
// removeStale unlinks a socket at path that no process listens on and// refuses anything else, the file type is checked before the dial because// dialing a regular file fails with ECONNREFUSED as wellfunc removeStale(path string) error { fi, err := os.Lstat(path) if errors.Is(err, os.ErrNotExist) { return nil } if err != nil { return fmt.Errorf("stat control socket %q: %w", path, err) } if fi.Mode().Type() != os.ModeSocket { return fmt.Errorf("control socket path %q exists and is not a socket", path) } conn, err := net.Dial("unix", path) switch { case err == nil: conn.Close() return fmt.Errorf("control socket %q is in use by another process", path) case errors.Is(err, os.ErrNotExist): return nil case !errors.Is(err, syscall.ECONNREFUSED): return fmt.Errorf("probe control socket %q: %w", path, err) } if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { return fmt.Errorf("remove stale control socket %q: %w", path, err) } return nil}
// close unlinks the socket file unless another process has put a different// file at the path since Listen, the listener closes after the unlink, which// means the file is gone once Serve sees the listener closedfunc (s *Server) close() { s.closeOnce.Do(func() { if fi, err := os.Lstat(s.path); err == nil && os.SameFile(fi, s.created) { _ = os.Remove(s.path) } s.listener.Close() })}
// Close closes the socket of a server that is not serving, Serve closes it// itself when it returns, closing twice is harmlessfunc (s *Server) Close() error { s.close() return nil}
// Addr returns the socket pathfunc (s *Server) Addr() string { return s.listener.Addr().String()}
// Serve answers connections until ctx is done or accept fails with an error// that is not temporary, and closes the socket when it returnsfunc (s *Server) Serve(ctx context.Context) error { stop := context.AfterFunc(ctx, s.close) defer stop() defer s.wg.Wait() defer s.close()
slots := make(chan struct{}, maxConns) var delay time.Duration for { select { case slots <- struct{}{}: case <-ctx.Done(): return ctx.Err() } conn, err := s.listener.Accept() if err != nil { <-slots if ctx.Err() != nil { return ctx.Err() } // accept still marks fd exhaustion with the deprecated // Temporary, which net/http retries on as well var te interface{ Temporary() bool } if !errors.As(err, &te) || !te.Temporary() { return fmt.Errorf("accept control connection: %w", err) } delay = min(max(2*delay, 5*time.Millisecond), time.Second) log.Error("accept control connection", "err", err, "retry_in", delay) select { case <-time.After(delay): case <-ctx.Done(): return ctx.Err() } continue } delay = 0 s.wg.Go(func() { defer func() { <-slots }() defer conn.Close() s.serveConn(conn) }) }}
func (s *Server) serveConn(conn net.Conn) { _ = conn.SetDeadline(time.Now().Add(requestTimeout))
var req Request body := &io.LimitedReader{R: conn, N: maxRequestSize} if err := json.NewDecoder(body).Decode(&req); err != nil { msg := "decode request: " + err.Error() if body.N == 0 { msg = fmt.Sprintf("request exceeds %d bytes", maxRequestSize) } writeResponse(conn, Response{Error: msg}) return } writeResponse(conn, s.dispatch(req))}
func writeResponse(conn net.Conn, resp Response) { enc := json.NewEncoder(conn) _ = enc.Encode(resp)}
func (s *Server) dispatch(req Request) Response { data, err := s.handle(req) if err != nil { return Response{Error: err.Error()} } raw, err := json.Marshal(data) if err != nil { return Response{Error: "encode response: " + err.Error()} } return Response{Data: raw}}
func (s *Server) handle(req Request) (any, error) { switch req.Command { case CmdStatus: return s.handler.Status() case CmdFlowsTop: n := 20 if v := req.Args["n"]; v != "" { parsed, err := strconv.Atoi(v) if err != nil || parsed < 1 { return nil, fmt.Errorf("n must be a positive integer, got %q", v) } n = parsed } by := req.Args["by"] if by == "" { by = "bytes" } if by != "bytes" && by != "packets" { return nil, fmt.Errorf("by must be bytes or packets, got %q", by) } rows, err := s.handler.FlowsTop(n, by) if err != nil { return nil, err } if rows == nil { rows = []FlowRow{} } return rows, nil case CmdFlowsCount: return s.handler.FlowsCount(), nil case CmdRIBLookup: addr, err := netip.ParseAddr(req.Args["addr"]) if err != nil { return nil, fmt.Errorf("addr: %w", err) } route, ok, err := s.handler.RIBLookup(addr) if err != nil { return nil, err } if !ok { return nil, fmt.Errorf("no route for %s", addr) } return route, nil case CmdRIBSummary: return s.handler.RIBSummary() case CmdSetRate: rate, err := strconv.ParseUint(req.Args["rate"], 10, 32) if err != nil || rate == 0 { return nil, fmt.Errorf("rate must be a positive integer, got %q", req.Args["rate"]) } if err := s.handler.SetSampleRate(uint32(rate)); err != nil { return nil, err } return Sampling{Rate: uint32(rate)}, nil case CmdConfigShow: return s.handler.ConfigShow(), nil case CmdReloadMMDB: if err := s.handler.ReloadMMDB(); err != nil { return nil, err } return "reloaded", nil default: return nil, fmt.Errorf("unknown command %q", req.Command) }}