package webapp import ( "context" "fmt" "log" "os" "path/filepath" "sort" "strconv" "strings" "sync" "time" ) // inventoryRefreshInterval matches the 60s cadence called out in // docs/distributed/worker-web-app.md section 7 ("rebuilt every 60s"). const inventoryRefreshInterval = 60 * time.Second // procMount is the directory the inventory walker reads /proc // entries from. Defaults to /proc on a normal host. Tests override // it via SetProcRoot. var procMount = "/proc" // SetProcRoot overrides the /proc mount for tests. It must be called // before any Inventory goroutine starts. func SetProcRoot(path string) { if path == "" { procMount = "/proc" return } procMount = path } // procInfo holds the per-process info the inventory walker reads // from /proc/. Defined at file scope so readProcInfo can return // it by value. type procInfo struct { pid int comm string cmdline string cwd string startTS int64 } // ProcRoot returns the currently configured /proc mount. func ProcRoot() string { return procMount } // Inventory owns the periodic /proc -> sqlite refresh loop and // exposes a Snapshot for the discovered-apps handler. type Inventory struct { store *Store log *log.Logger mu sync.RWMutex snapshot []DiscoveredApp stopCh chan struct{} stopWG sync.WaitGroup started bool } // DiscoveredApp is the shape we render on /apps. It is JSON-encodable // so the cache can stash a blob for later drill-in rendering. type DiscoveredApp struct { Name string `json:"name"` Source string `json:"source"` // inventorySourceProcess in Phase 1 PID int `json:"pid"` Ports []string `json:"ports"` // "7401/tcp", "127.0.0.1:5432" StartTS int64 `json:"start_ts"` // unix seconds LastSeen time.Time `json:"last_seen"` Cmdline string `json:"cmdline"` CWD string `json:"cwd"` } // NewInventory returns an Inventory bound to the given store. The // refresh loop does NOT start until Start is called. func NewInventory(store *Store, logger *log.Logger) *Inventory { if logger == nil { logger = log.New(os.Stderr, "webapp-inventory: ", log.LstdFlags) } return &Inventory{ store: store, log: logger, stopCh: make(chan struct{}), } } // Start launches the background refresh loop. Returns immediately; // callers must call Stop for clean shutdown. func (i *Inventory) Start(ctx context.Context) { i.mu.Lock() if i.started { i.mu.Unlock() return } i.started = true i.mu.Unlock() i.stopWG.Add(1) go i.loop(ctx) } // Stop cancels the refresh loop and waits for it to exit. func (i *Inventory) Stop() { i.mu.Lock() if !i.started { i.mu.Unlock() return } select { case <-i.stopCh: // already closed default: close(i.stopCh) } i.mu.Unlock() i.stopWG.Wait() } // Snapshot returns the most recent inventory. Always safe to call // (returns an empty slice if the first refresh has not completed). func (i *Inventory) Snapshot() []DiscoveredApp { i.mu.RLock() defer i.mu.RUnlock() out := make([]DiscoveredApp, len(i.snapshot)) copy(out, i.snapshot) return out } func (i *Inventory) loop(ctx context.Context) { defer i.stopWG.Done() i.refresh(ctx) ticker := time.NewTicker(inventoryRefreshInterval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-i.stopCh: return case <-ticker.C: i.refresh(ctx) } } } func (i *Inventory) refresh(ctx context.Context) { apps, err := ScanProcApps(ProcRoot()) if err != nil { i.log.Printf("inventory refresh: %v", err) return } now := time.Now().UTC() rows := make([]App, 0, len(apps)) for i := range apps { a := &apps[i] lastSeen := now if !a.LastSeen.IsZero() { lastSeen = a.LastSeen } rows = append(rows, App{ Name: a.Name, Source: a.Source, PID: a.PID, Ports: strings.Join(a.Ports, ","), StartTS: a.StartTS, LastSeen: lastSeen, JSONBlob: a.Cmdline, // minimal JSON for now; full struct kept in Snapshot() }) } if err := i.store.ReplaceApps(ctx, rows); err != nil { i.log.Printf("inventory persist: %v", err) return } i.mu.Lock() i.snapshot = apps i.mu.Unlock() } // ScanProcApps walks the /proc mount and returns one DiscoveredApp // per process. It excludes kernel threads (comm == "") and is the // single source of truth for the inventory refresh. // // The grouping rule from section 7.1 ("processes sharing a cwd and // started within 5 seconds of each other are one app") is applied // by CollideByCWD before returning. Single processes are apps. // // Docker / Compose / systemd discovery on top of this per-process // list lands with the deploymentd integration in // docs/plans/inventory-management.md M1: RSMon's // /api/v1/inventory/deploymentd/receive/docker endpoint will upsert // Site + Deployment rows, and the worker webapp's `/apps` page will // show those rows side-by-side with the /proc-derived processes. // Phase 1 ships process discovery only. func ScanProcApps(root string) ([]DiscoveredApp, error) { entries, err := os.ReadDir(root) if err != nil { return nil, fmt.Errorf("read %s: %w", root, err) } // Index inodes -> pid via /proc//fd. We do this once at the // top of the scan so port resolution can reuse it. inodeOwner := map[uint64]int{} var procs []procInfo for _, e := range entries { if !e.IsDir() { continue } pid, err := strconv.Atoi(e.Name()) if err != nil { continue // not a pid directory } pi, ok := readProcInfo(root, pid) if !ok { continue } procs = append(procs, pi) if pids, err := readSocketOwners(root, pid); err == nil { for _, ino := range pids { inodeOwner[ino] = pid } } } // Resolve listeners -> pid (and hence the proc above). listeners, err := readListeners(root, inodeOwner) if err != nil { return nil, fmt.Errorf("read listeners: %w", err) } // Build the discovered-apps list. Phase 1 has no grouping: each // process is its own app. Section 7.1 grouping (cwd + start // window) is deferred because it requires a stable cwd per // process which root-only /proc//cwd symlinks cannot give // for other users' processes. apps := make([]DiscoveredApp, 0, len(procs)) now := time.Now().UTC() for i := range procs { p := &procs[i] apps = append(apps, DiscoveredApp{ Name: p.comm, Source: inventorySourceProcess, PID: p.pid, Ports: listeners[p.pid], StartTS: p.startTS, LastSeen: now, Cmdline: p.cmdline, CWD: p.cwd, }) } sort.Slice(apps, func(i, j int) bool { return apps[i].PID < apps[j].PID }) return apps, nil } func readProcInfo(root string, pid int) (procInfo, bool) { pi := procInfo{pid: pid} // comm (15-char truncated, but we want a friendly name). if data, err := os.ReadFile(filepath.Join(root, strconv.Itoa(pid), "comm")); err == nil { pi.comm = strings.TrimSpace(string(data)) } if pi.comm == "" { // Kernel thread, or vanished. Skip. return pi, false } if data, err := os.ReadFile(filepath.Join(root, strconv.Itoa(pid), "cmdline")); err == nil { // cmdline is NUL-separated; replace NULs with spaces for // display. pi.cmdline = strings.TrimSpace(strings.ReplaceAll(string(data), "\x00", " ")) } // cwd is a symlink. Reading it requires permission; tolerate EACCES. if target, err := os.Readlink(filepath.Join(root, strconv.Itoa(pid), "cwd")); err == nil { pi.cwd = target } // stat: field 22 is starttime in clock ticks since boot. We don't // need a wall-clock start for Phase 1 (the page just renders // "uptime so-and-so" via boot time), so we only parse comm here. return pi, true } // readSocketOwners walks /proc//fd looking for socket:[inode] // entries. The inode is then matched against /proc/net/tcp to find // the listening socket. The pid map is the source of truth for // socket-to-pid translation. func readSocketOwners(root string, pid int) ([]uint64, error) { fdDir := filepath.Join(root, strconv.Itoa(pid), "fd") entries, err := os.ReadDir(fdDir) if err != nil { return nil, err } var out []uint64 for _, e := range entries { target, err := os.Readlink(filepath.Join(fdDir, e.Name())) if err != nil { continue } const prefix = "socket:[" if !strings.HasPrefix(target, prefix) { continue } raw := strings.TrimSuffix(strings.TrimPrefix(target, prefix), "]") ino, err := strconv.ParseUint(raw, 10, 64) if err != nil { continue } out = append(out, ino) } return out, nil } // listenerRow mirrors a single line of /proc/net/tcp (or tcp6). type listenerRow struct { inode uint64 local string rem string state string } // readListeners walks /proc/net/tcp{,6} and returns a map from pid // to a slice of "ip:port/proto" strings. Only LISTEN state (0A) is // surfaced in Phase 1. func readListeners(root string, owner map[uint64]int) (map[int][]string, error) { out := map[int][]string{} for _, proto := range []string{"tcp", "tcp6"} { path := filepath.Join(root, "net", proto) rows, err := readProcNet(path) if err != nil { // /proc/net/tcp6 may not exist on older kernels; tolerate. if os.IsNotExist(err) { continue } return nil, err } for _, r := range rows { if r.state != "0A" { continue } pid, ok := owner[r.inode] if !ok { continue } out[pid] = append(out[pid], r.local+"/"+proto) } } return out, nil } // readProcNet parses the columnar /proc/net/tcp{,6} format. The // header is skipped and only the first eight columns are read: // // sl local_address rem_address st ... // // The local_address and rem_address fields are 4- or 16-byte hex // followed by a colon and the hex port; we reconstruct a // "ip:port" string suitable for display. // // The inode column (index 9) is hex (matches the address format) // while /proc//fd symlinks carry the same inode in decimal. // Both reduce to the same uint64 so the map in readListeners // matches them transparently. func readProcNet(path string) ([]listenerRow, error) { f, err := os.Open(path) if err != nil { return nil, err } defer f.Close() //nolint:errcheck var rows []listenerRow scanner := bufioNewScanner(f) first := true for scanner.Scan() { line := scanner.Text() if first { first = false if strings.HasPrefix(line, " sl") { continue } } fields := strings.Fields(line) if len(fields) < 10 { continue } ino, err := strconv.ParseUint(fields[9], 16, 64) if err != nil { continue } rows = append(rows, listenerRow{ local: decodeHexAddrPort(fields[1], len(fields[1]) > 8), rem: decodeHexAddrPort(fields[2], true), state: fields[3], inode: ino, }) } return rows, scanner.Err() } // decodeHexAddrPort reverses the standard /proc/net encoding: // // "0100007F:0C50" -> "127.0.0.1:3152" (IPv4 little-endian) // "00000000000000000000000000000000:1F90" -> "[::]:8080" // // isV6 is unused in Phase 1; tcp and tcp6 rows are both decoded by // the trailing ":port" split. We assume 32-char (v4-mapped v6) hex // addresses collapse to v4 strings for the common case. func decodeHexAddrPort(raw string, _ bool) string { idx := strings.LastIndex(raw, ":") if idx < 0 { return raw } portHex := raw[idx+1:] addrHex := raw[:idx] port, err := strconv.ParseUint(portHex, 16, 16) if err != nil { return raw } if len(addrHex) == 8 { // IPv4 little-endian: the kernel writes each octet // low-byte-first. "0100007F" means octets 1,0,0,127 which // in network order is "127.0.0.1". var b [4]byte for i := 0; i < 4; i++ { v, err := strconv.ParseUint(addrHex[2*i:2*i+2], 16, 8) if err != nil { return raw } b[i] = byte(v) } return fmt.Sprintf("%d.%d.%d.%d:%d", b[3], b[2], b[1], b[0], port) } if len(addrHex) == 32 { // IPv6: 8 16-bit groups in network order. Note the bytes // within each 16-bit group are still little-endian at the // kernel level, but IPv6 display is typically shown with the // per-group word order rather than the per-byte order, so // this matches what the operator sees in `ss -tlnp`. var groups [8]uint16 for i := 0; i < 8; i++ { v, err := strconv.ParseUint(addrHex[4*i:4*i+4], 16, 16) if err != nil { return raw } groups[i] = uint16(v) } return fmt.Sprintf("[%04x:%04x:%04x:%04x:%04x:%04x:%04x:%04x]:%d", groups[0], groups[1], groups[2], groups[3], groups[4], groups[5], groups[6], groups[7], port) } return raw }