llmgw/internal/cli/agent.go

package cli

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"text/tabwriter"
	"time"

	"github.com/spf13/cobra"

	"gitea.mynodes.xyz/ilker/llmgw/internal/agent"
	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
	"gitea.mynodes.xyz/ilker/llmgw/internal/config"
	"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
	"gitea.mynodes.xyz/ilker/llmgw/internal/version"
)

func newAgentCmd() *cobra.Command {
	var configPath string

	var (
		server   string
		nodeID   string
		mgmtAddr string
	)

	cmd := &cobra.Command{
		Use:   "agent",
		Short: "Node agent: discovery and engine supervision",
		Long: "Runs on every GPU node. It discovers the hardware, connects to the control\n" +
			"plane, and starts, supervises and stops inference engines on demand.\n\n" +
			"The agent dials out and keeps one stream open. Engines keep serving if the\n" +
			"agent restarts or the control plane goes away: the router talks to them\n" +
			"directly, and orphan adoption picks them up again.",
		RunE: func(cmd *cobra.Command, _ []string) error {
			log := logger()

			// The cluster's settings for this agent over its file, as the
			// control plane last sent them. Settings that no longer load
			// beside the file are set aside rather than keeping the agent
			// from starting: the control plane sends them again, and says
			// why they are refused.
			settings := agent.ReadSettings()
			cfg, err := config.LoadAgentWith(configPath, settings)
			if err != nil && settings != nil {
				log.Warn("the cluster's settings kept for this agent do not load beside its agent.yaml; starting without them",
					"err", err)
				settings = nil
				cfg, err = config.LoadAgent(configPath)
			}
			if err != nil {
				return err
			}
			cfg.Settings = string(settings)
			if server != "" {
				cfg.Server = server
			}
			if nodeID != "" {
				cfg.NodeID = nodeID
			}
			if mgmtAddr != "" {
				cfg.MgmtAddr = mgmtAddr
			}
			if cfg.NodeID == "" {
				host, err := os.Hostname()
				if err != nil {
					return fmt.Errorf("node_id is unset and the hostname is unavailable: %w", err)
				}
				cfg.NodeID = host
			}
			if cfg.Hostname == "" {
				cfg.Hostname, _ = os.Hostname()
			}
			cfg.Version = version.Version

			// Before the agent, so its tracer is the configured one.
			shutdownTracing, err := obs.SetupTracing(cmd.Context(), obs.TracingOptions{
				Endpoint: cfg.Tracing.Endpoint, SampleRatio: cfg.Tracing.SampleRatio,
				Service: "llmgw-agent", Version: version.Version, Instance: cfg.NodeID,
			})
			if err != nil {
				return err
			}
			defer func() {
				// Bounded: a collector that is gone must not hold up a shutdown.
				sctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
				defer cancel()
				_ = shutdownTracing(sctx)
			}()

			httpClient, err := agent.Client(cfg.TLS, 30*time.Second)
			if err != nil {
				return err
			}

			a, err := agent.New(cfg, httpClient, log)
			if err != nil {
				return err
			}
			a.SetSettingsCheck(func(b []byte) error {
				_, err := config.LoadAgentWith(configPath, b)
				return err
			})

			ctx, stop := signal.NotifyContext(cmd.Context(), syscall.SIGINT, syscall.SIGTERM)
			defer stop()
			go a.WatchFile(ctx, configPath, func() error {
				_, err := config.LoadAgentWith(configPath, agent.ReadSettings())
				return err
			})

			log.Info("agent starting",
				"node", cfg.NodeID, "server", cfg.Server,
				"fabrics", len(cfg.Fabrics), "nvml", discovery.NVMLSupported())

			err = a.Run(ctx)
			if errors.Is(err, context.Canceled) {
				return nil
			}
			return err
		},
	}
	cmd.Flags().StringVar(&configPath, "config", "", "configuration file")
	cmd.Flags().StringVar(&server, "server", "", "control plane address (overrides the config)")
	cmd.Flags().StringVar(&nodeID, "node-id", "", "node id (defaults to the hostname)")
	cmd.Flags().StringVar(&mgmtAddr, "mgmt-addr", "", "address engines are reachable on")
	cmd.AddCommand(newInventoryCmd())
	return cmd
}

func newInventoryCmd() *cobra.Command {
	var (
		asJSON  bool
		smiBin  string
		ibRoot  string
		skipGPU bool
		useNVML bool
	)

	cmd := &cobra.Command{
		Use:   "inventory",
		Short: "Show this node's GPUs and InfiniBand adapters",
		Long: "Reports what the scheduler would see on this node: GPU classes, health,\n" +
			"and the fabric each adapter is attached to.\n\n" +
			"Run it before adding a node to the cluster — it is the fastest way to find\n" +
			"a card another process is already using, or a port with no subnet manager.",
		RunE: func(cmd *cobra.Command, _ []string) error {
			inv, err := discovery.Discover(cmd.Context(), discovery.Options{
				NvidiaSMI:  smiBin,
				IBRoot:     ibRoot,
				PreferNVML: useNVML,
				SkipGPUs:   skipGPU,
			}, logger())
			if err != nil {
				// Not fatal: a node whose GPUs cannot be read still has IB
				// worth showing, and the error is itself the useful output.
				fmt.Fprintf(os.Stderr, "%v\n", err)
				inv.Source = "unavailable"
			}

			if asJSON {
				enc := json.NewEncoder(os.Stdout)
				enc.SetIndent("", "  ")
				return enc.Encode(inv)
			}
			printInventory(inv)
			return nil
		},
	}

	f := cmd.Flags()
	f.BoolVar(&asJSON, "json", false, "emit JSON")
	f.StringVar(&smiBin, "nvidia-smi", "", "path to nvidia-smi")
	f.StringVar(&ibRoot, "ib-root", discovery.SysfsRoot, "sysfs InfiniBand directory")
	f.BoolVar(&skipGPU, "no-gpu", false, "skip GPU discovery")
	f.BoolVar(&useNVML, "nvml", true, "prefer the NVML backend when this build has one (it is the only source of NVLink topology)")

	return cmd
}

func printInventory(inv discovery.Inventory) {
	w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)

	fmt.Fprintf(w, "GPUs (%d, via %s)\n", len(inv.GPUs), inv.Source)
	if len(inv.GPUs) == 0 {
		fmt.Fprintln(w, "  none")
	} else {
		fmt.Fprintln(w, "  IDX\tCLASS\tNAME\tMEM\tCC\tHEALTH")
		for _, g := range inv.GPUs {
			health := "ok"
			if !g.Health.OK() {
				health = g.Health.Problems[0]
			}
			fmt.Fprintf(w, "  %s\t%s\t%s\t%d MiB\t%s\t%s\n",
				discovery.DeviceLabel(g.Index), g.Class(), g.Name, g.MemoryTotalMB, g.ComputeCapability, health)
		}
		fmt.Fprintf(w, "  classes: %v\n", inv.Classes())
		for _, g := range inv.GPUs {
			if len(g.NVLinkPeers) > 0 {
				fmt.Fprintf(w, "  GPU %s NVLink peers: %v\n", discovery.DeviceLabel(g.Index), g.NVLinkPeers)
			}
		}
	}

	fmt.Fprintf(w, "\nInfiniBand adapters (%d)\n", len(inv.HCAs))
	if len(inv.HCAs) == 0 {
		// Worth stating plainly: such a node is its own single-node island and
		// can never host a multi-node replica.
		fmt.Fprintln(w, "  none — this node cannot host multi-node replicas")
	} else {
		fmt.Fprintln(w, "  HCA\tPORT\tSTATE\tRATE\tLINK\tGID PREFIX\tNUMA")
		for _, h := range inv.HCAs {
			for _, p := range h.Ports {
				fmt.Fprintf(w, "  %s\t%d\t%s\t%d Gb/s\t%s\t%s\t%d\n",
					h.Name, p.Num, p.State, p.RateGbps, p.LinkLayer, p.GIDPrefix, h.NUMANode)
			}
		}
		fmt.Fprintln(w, "\n  Fabric identity is not derived from the GID prefix by default:")
		fmt.Fprintln(w, "  OpenSM's default is fe80:: on every fabric, so a shared prefix proves")
		fmt.Fprintln(w, "  nothing. Label fabrics explicitly in agent.yaml (plan §5.2).")
	}

	_ = w.Flush()
}