package node import ( "context" "fmt" "log/slog" "os" "os/exec" "path/filepath" "strings" "sync" "time" "github.com/wild-cloud/wild-central/daemon/internal/config" "github.com/wild-cloud/wild-central/daemon/internal/setup" "github.com/wild-cloud/wild-central/daemon/internal/tools" "gopkg.in/yaml.v3" ) // Manager handles node configuration and state management type Manager struct { dataDir string configMgr *config.Manager talosctl *tools.Talosctl } // NewManager creates a new node manager func NewManager(dataDir string, instanceName string) *Manager { var talosctl *tools.Talosctl // If instanceName is provided, use instance-specific talosconfig // Otherwise, create basic talosctl (will use --insecure mode) if instanceName != "" { talosconfigPath := tools.GetTalosconfigPath(dataDir, instanceName) talosctl = tools.NewTalosconfigWithConfig(talosconfigPath) } else { talosctl = tools.NewTalosctl() } return &Manager{ dataDir: dataDir, configMgr: config.NewManager(), talosctl: talosctl, } } // Node represents a cluster node configuration type Node struct { Hostname string `yaml:"hostname" json:"hostname"` Role string `yaml:"role" json:"role"` // controlplane or worker TargetIP string `yaml:"targetIp" json:"target_ip"` CurrentIP string `yaml:"currentIp,omitempty" json:"current_ip,omitempty"` // For maintenance mode detection Interface string `yaml:"interface,omitempty" json:"interface,omitempty"` Disk string `yaml:"disk" json:"disk"` Version string `yaml:"version,omitempty" json:"version,omitempty"` SchematicID string `yaml:"schematicId,omitempty" json:"schematic_id,omitempty"` Maintenance bool `yaml:"maintenance,omitempty" json:"maintenance"` // Explicit maintenance mode flag Configured bool `yaml:"configured,omitempty" json:"configured"` Applied bool `yaml:"applied,omitempty" json:"applied"` } // HardwareInfo contains discovered hardware information type HardwareInfo struct { IP string `json:"ip"` Interface string `json:"interface"` Disks []tools.DiskInfo `json:"disks"` SelectedDisk string `json:"selected_disk"` MaintenanceMode bool `json:"maintenance_mode"` } // ApplyOptions contains options for node apply type ApplyOptions struct { // No options needed - apply always regenerates and auto-fetches templates } // GetInstancePath returns the path to an instance's nodes directory func (m *Manager) GetInstancePath(instanceName string) string { return tools.GetInstancePath(m.dataDir, instanceName) } // nodeConfigFile is the operator-authored portion stored in config/nodes/{hostname}.yaml type nodeConfigFile struct { Role string `yaml:"role"` Disk string `yaml:"disk"` Interface string `yaml:"interface,omitempty"` TargetIP string `yaml:"targetIp,omitempty"` SchematicID string `yaml:"schematicId,omitempty"` Version string `yaml:"version,omitempty"` Maintenance bool `yaml:"maintenance,omitempty"` } // nodeCacheFile is the daemon-written state stored in cache/nodes/{hostname}.yaml type nodeCacheFile struct { Applied bool `yaml:"applied,omitempty"` Configured bool `yaml:"configured,omitempty"` CurrentIP string `yaml:"currentIp,omitempty"` } // List returns all nodes for an instance func (m *Manager) List(instanceName string) ([]Node, error) { return m.listFromNodeFiles(instanceName) } // listFromNodeFiles reads nodes from config/nodes/*.yaml + cache/nodes/*.yaml func (m *Manager) listFromNodeFiles(instanceName string) ([]Node, error) { configDir := tools.GetNodeConfigDirPath(m.dataDir, instanceName) entries, err := os.ReadDir(configDir) if os.IsNotExist(err) { return []Node{}, nil } if err != nil { return nil, fmt.Errorf("failed to read node config dir: %w", err) } var nodes []Node for _, entry := range entries { if entry.IsDir() || filepath.Ext(entry.Name()) != ".yaml" { continue } hostname := entry.Name()[:len(entry.Name())-5] // strip .yaml data, err := os.ReadFile(filepath.Join(configDir, entry.Name())) if err != nil { slog.Warn("failed to read node config", "hostname", hostname, "error", err) continue } var cfg nodeConfigFile if err := yaml.Unmarshal(data, &cfg); err != nil { slog.Warn("failed to parse node config", "hostname", hostname, "error", err) continue } node := Node{ Hostname: hostname, Role: cfg.Role, Disk: cfg.Disk, Interface: cfg.Interface, TargetIP: cfg.TargetIP, SchematicID: cfg.SchematicID, Version: cfg.Version, Maintenance: cfg.Maintenance, } // Merge cache cacheData, err := os.ReadFile(tools.GetNodeCachePath(m.dataDir, instanceName, hostname)) if err == nil { var cache nodeCacheFile if err := yaml.Unmarshal(cacheData, &cache); err == nil { node.Applied = cache.Applied node.Configured = cache.Configured node.CurrentIP = cache.CurrentIP } } nodes = append(nodes, node) } return nodes, nil } // Get returns a specific node by hostname func (m *Manager) Get(instanceName, hostname string) (*Node, error) { // Get all nodes nodes, err := m.List(instanceName) if err != nil { return nil, err } // Find node by hostname for _, node := range nodes { if node.Hostname == hostname { return &node, nil } } return nil, fmt.Errorf("node %s not found", hostname) } // Add registers a new node func (m *Manager) Add(instanceName string, node *Node) error { slog.Info("adding node", "component", "node", "instance", instanceName, "hostname", node.Hostname, "role", node.Role) // Validate node data if node.Hostname == "" { return fmt.Errorf("hostname is required") } if node.Role != "controlplane" && node.Role != "worker" { return fmt.Errorf("role must be 'controlplane' or 'worker'") } if node.Disk == "" { return fmt.Errorf("disk is required") } // Check if node already exists - ERROR if yes existing, err := m.Get(instanceName, node.Hostname) if err == nil && existing != nil { return fmt.Errorf("node %s already exists", node.Hostname) } // Set maintenance=true if currentIP provided (node in maintenance mode) if node.CurrentIP != "" { node.Maintenance = true } return m.addNodeFile(instanceName, node) } // addNodeFile writes a new node's config to config/nodes/{hostname}.yaml (new format) func (m *Manager) addNodeFile(instanceName string, node *Node) error { // Fill talos defaults from instance.yaml if missing if node.SchematicID == "" || node.Version == "" { instanceYaml := tools.GetInstanceYamlPath(m.dataDir, instanceName) if data, err := os.ReadFile(instanceYaml); err == nil { var root struct { Cluster struct { Nodes struct { Talos struct { SchematicID string `yaml:"schematicId"` Version string `yaml:"version"` } `yaml:"talos"` } `yaml:"nodes"` } `yaml:"cluster"` } if err := yaml.Unmarshal(data, &root); err == nil { if node.SchematicID == "" { node.SchematicID = root.Cluster.Nodes.Talos.SchematicID } if node.Version == "" { node.Version = root.Cluster.Nodes.Talos.Version } } } } configDir := tools.GetNodeConfigDirPath(m.dataDir, instanceName) if err := os.MkdirAll(configDir, 0755); err != nil { return fmt.Errorf("failed to create node config dir: %w", err) } cfg := nodeConfigFile{ Role: node.Role, Disk: node.Disk, Interface: node.Interface, TargetIP: node.TargetIP, SchematicID: node.SchematicID, Version: node.Version, Maintenance: node.Maintenance, } data, err := yaml.Marshal(cfg) if err != nil { return fmt.Errorf("failed to marshal node config: %w", err) } return os.WriteFile(tools.GetNodeConfigPath(m.dataDir, instanceName, node.Hostname), data, 0644) } // Delete removes a node from config.yaml // If skipReset is false, the node will be reset before deletion (with 30s timeout) func (m *Manager) Delete(instanceName, nodeIdentifier string, skipReset bool) error { slog.Info("deleting node", "component", "node", "instance", instanceName, "node", nodeIdentifier, "skipReset", skipReset) // Get node to find hostname node, err := m.Get(instanceName, nodeIdentifier) if err != nil { return err } // Reset node first unless skipReset is true if !skipReset { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() // Use goroutine to respect context timeout done := make(chan error, 1) go func() { done <- m.Reset(instanceName, nodeIdentifier) }() select { case err := <-done: if err != nil { return fmt.Errorf("failed to reset node before deletion (use skip_reset=true to force delete): %w", err) } case <-ctx.Done(): return fmt.Errorf("node reset timed out after 30 seconds (use skip_reset=true to force delete)") } } // Remove the k8s node object (best-effort — node may already be gone) kubeconfigPath := tools.GetKubeconfigPath(m.dataDir, instanceName) cmd := exec.Command("kubectl", "delete", "node", node.Hostname, "--ignore-not-found") tools.WithKubeconfig(cmd, kubeconfigPath) if out, err := cmd.CombinedOutput(); err != nil { slog.Warn("failed to delete k8s node object (continuing)", "component", "node", "hostname", node.Hostname, "error", err, "output", strings.TrimSpace(string(out))) } // Delete node from config.yaml return m.deleteFromConfig(instanceName, node.Hostname) } // deleteFromConfig removes a node entry from config func (m *Manager) deleteFromConfig(instanceName, hostname string) error { configPath := tools.GetNodeConfigPath(m.dataDir, instanceName, hostname) if err := os.Remove(configPath); err != nil && !os.IsNotExist(err) { return fmt.Errorf("failed to delete node config: %w", err) } cachePath := tools.GetNodeCachePath(m.dataDir, instanceName, hostname) if err := os.Remove(cachePath); err != nil && !os.IsNotExist(err) { slog.Warn("failed to delete node cache", "component", "node", "hostname", hostname, "error", err) } return nil } // DetectHardware queries node hardware information via talosctl // Automatically detects maintenance mode by trying insecure first, then secure func (m *Manager) DetectHardware(nodeIP string) (*HardwareInfo, error) { // Try insecure first (maintenance mode) hwInfo, err := m.detectHardwareWithMode(nodeIP, true) if err == nil { return hwInfo, nil } // Fall back to secure (configured node) return m.detectHardwareWithMode(nodeIP, false) } // detectHardwareWithMode queries node hardware with specified connection mode func (m *Manager) detectHardwareWithMode(nodeIP string, insecure bool) (*HardwareInfo, error) { // Try to get default interface (with default route) iface, err := m.talosctl.GetDefaultInterface(nodeIP, insecure) if err != nil { // Fall back to physical interface iface, err = m.talosctl.GetPhysicalInterface(nodeIP, insecure) if err != nil { return nil, fmt.Errorf("failed to detect interface: %w", err) } } // Get disks disks, err := m.talosctl.GetDisks(nodeIP, insecure) if err != nil { return nil, fmt.Errorf("failed to detect disks: %w", err) } // Select first disk as default var selectedDisk string if len(disks) > 0 { selectedDisk = disks[0].Path } return &HardwareInfo{ IP: nodeIP, Interface: iface, Disks: disks, SelectedDisk: selectedDisk, MaintenanceMode: insecure, // If we used insecure, it's in maintenance mode }, nil } // Apply generates configuration and applies it to node // This follows the wild-node-apply flow: // 1. Auto-fetch templates if missing // 2. Generate node-specific patch file from template // 3. Merge base config + patch → final config (talosctl machineconfig patch) // 4. Apply final config to node (talosctl apply-config --insecure if maintenance mode) // 5. Update state: currentIP=targetIP, maintenance=false, applied=true func (m *Manager) Apply(instanceName, nodeIdentifier string, opts ApplyOptions) error { // Get node configuration node, err := m.Get(instanceName, nodeIdentifier) if err != nil { return err } setupDir := tools.GetNodeSetupCachePath(m.dataDir, instanceName) // Ensure node has version and schematicId (use cluster defaults if missing) if node.Version == "" || node.SchematicID == "" { m.fillTalosDefaults(instanceName, node) } // Always auto-extract templates from embedded files if they don't exist templatesDir := filepath.Join(setupDir, "patch.templates") if !m.templatesExist(templatesDir) { if err := m.extractEmbeddedTemplates(templatesDir); err != nil { return fmt.Errorf("failed to extract templates: %w", err) } } // Determine base configuration file (generated by cluster config generation) var baseConfig string if node.Role == "controlplane" { baseConfig = tools.GetControlplaneConfigPath(m.dataDir, instanceName) } else { baseConfig = tools.GetWorkerConfigPath(m.dataDir, instanceName) } // Check if base config exists if _, err := os.Stat(baseConfig); err != nil { return fmt.Errorf("base configuration not found: %s (run cluster config generation first)", baseConfig) } // Generate node-specific patch file patchFile, err := m.generateNodePatch(instanceName, node, setupDir) if err != nil { return fmt.Errorf("failed to generate node patch: %w", err) } // Generate final machine configuration (base + patch) finalConfig, err := m.generateFinalConfig(node, baseConfig, patchFile, setupDir) if err != nil { return fmt.Errorf("failed to generate final configuration: %w", err) } // Mark as configured node.Configured = true if err := m.updateNodeStatus(instanceName, node); err != nil { return fmt.Errorf("failed to update node status: %w", err) } slog.Info("applying node config", "component", "node", "instance", instanceName, "hostname", node.Hostname, "role", node.Role) // Apply configuration to node // Determine which IP to use and whether node is in maintenance mode // // Three scenarios: // 1. Production node (already applied, maintenance=false): use targetIP, no --insecure // 2. IP changing (currentIP != targetIP): use currentIP, --insecure (always maintenance) // 3. Fresh/maintenance node (never applied OR maintenance=true): use targetIP, --insecure var deployIP string var maintenanceMode bool if node.CurrentIP != "" && node.CurrentIP != node.TargetIP { // Scenario 2: IP is changing - node is at currentIP, moving to targetIP deployIP = node.CurrentIP maintenanceMode = true } else if node.Maintenance || !node.Applied { // Scenario 3: Explicit maintenance mode OR never been applied (fresh node) // Fresh nodes need --insecure because they have self-signed certificates deployIP = node.TargetIP maintenanceMode = true } else { // Scenario 1: Production node at target IP (already applied, not in maintenance) deployIP = node.TargetIP maintenanceMode = false } // Apply config talosconfigPath := tools.GetTalosconfigPath(m.dataDir, instanceName) if err := m.talosctl.ApplyConfig(deployIP, finalConfig, maintenanceMode, talosconfigPath); err != nil { return fmt.Errorf("failed to apply config to %s: %w", deployIP, err) } // Post-application updates: move to production IP, exit maintenance mode node.Applied = true node.CurrentIP = node.TargetIP // Node now on production IP node.Maintenance = false // Exit maintenance mode if err := m.updateNodeStatus(instanceName, node); err != nil { return fmt.Errorf("failed to update node status: %w", err) } slog.Info("node config applied", "component", "node", "instance", instanceName, "hostname", node.Hostname, "ip", node.TargetIP) return nil } // generateNodePatch creates a node-specific patch file from template func (m *Manager) generateNodePatch(instanceName string, node *Node, setupDir string) (string, error) { // Determine template file based on role var templateFile string if node.Role == "controlplane" { templateFile = filepath.Join(setupDir, "patch.templates", "controlplane.yaml") } else { templateFile = filepath.Join(setupDir, "patch.templates", "worker.yaml") } // Read template templateContent, err := os.ReadFile(templateFile) if err != nil { return "", fmt.Errorf("failed to read template %s: %w", templateFile, err) } // Stage 1: Apply simple variable substitutions (like v.PoC does with sed) patchContent := string(templateContent) patchContent = strings.ReplaceAll(patchContent, "{{NODE_NAME}}", node.Hostname) patchContent = strings.ReplaceAll(patchContent, "{{NODE_IP}}", node.TargetIP) patchContent = strings.ReplaceAll(patchContent, "{{SCHEMATIC_ID}}", node.SchematicID) patchContent = strings.ReplaceAll(patchContent, "{{VERSION}}", node.Version) // Stage 2: Process through gomplate with merged global+instance config context merged, err := config.LoadMergedInstanceConfig(m.dataDir, instanceName) if err != nil { return "", fmt.Errorf("failed to load merged config: %w", err) } mergedYAML, err := yaml.Marshal(merged) if err != nil { return "", fmt.Errorf("failed to marshal merged config: %w", err) } gomplate := tools.NewGomplate() processedPatch, err := gomplate.RenderTemplate(patchContent, string(mergedYAML)) if err != nil { return "", fmt.Errorf("failed to process template with gomplate: %w", err) } // Wipe disk when node needs a fresh install (maintenance or never applied) if node.Maintenance || !node.Applied { var patchMap map[string]interface{} if err := yaml.Unmarshal([]byte(processedPatch), &patchMap); err == nil { if machine, ok := patchMap["machine"].(map[string]interface{}); ok { if install, ok := machine["install"].(map[string]interface{}); ok { install["wipe"] = true if patched, err := yaml.Marshal(patchMap); err == nil { processedPatch = string(patched) } } } } } // Create patch directory patchDir := filepath.Join(setupDir, "patch") if err := os.MkdirAll(patchDir, 0755); err != nil { return "", fmt.Errorf("failed to create patch directory: %w", err) } // Write patch file patchFile := filepath.Join(patchDir, node.Hostname+".yaml") if err := os.WriteFile(patchFile, []byte(processedPatch), 0644); err != nil { return "", fmt.Errorf("failed to write patch file: %w", err) } return patchFile, nil } // generateFinalConfig merges base config + patch to create final machine config func (m *Manager) generateFinalConfig(node *Node, baseConfig, patchFile, setupDir string) (string, error) { // Create final config directory finalDir := filepath.Join(setupDir, "final") if err := os.MkdirAll(finalDir, 0755); err != nil { return "", fmt.Errorf("failed to create final directory: %w", err) } finalConfig := filepath.Join(finalDir, node.Hostname+".yaml") // Use talosctl machineconfig patch to merge base + patch // talosctl machineconfig patch base.yaml --patch @patch.yaml -o final.yaml cmd := exec.Command("talosctl", "machineconfig", "patch", baseConfig, "--patch", "@"+patchFile, "-o", finalConfig) output, err := cmd.CombinedOutput() if err != nil { return "", fmt.Errorf("failed to patch machine config: %w\nOutput: %s", err, string(output)) } return finalConfig, nil } // templatesExist checks if patch templates exist in the instance directory func (m *Manager) templatesExist(templatesDir string) bool { controlplaneTemplate := filepath.Join(templatesDir, "controlplane.yaml") workerTemplate := filepath.Join(templatesDir, "worker.yaml") _, err1 := os.Stat(controlplaneTemplate) _, err2 := os.Stat(workerTemplate) return err1 == nil && err2 == nil } // extractEmbeddedTemplates extracts patch templates from embedded files to instance directory func (m *Manager) extractEmbeddedTemplates(destDir string) error { // Create destination directory if err := os.MkdirAll(destDir, 0755); err != nil { return fmt.Errorf("failed to create templates directory: %w", err) } // Get embedded template files controlplaneData, err := setup.GetClusterNodesFile("patch.templates/controlplane.yaml") if err != nil { return fmt.Errorf("failed to get controlplane template: %w", err) } workerData, err := setup.GetClusterNodesFile("patch.templates/worker.yaml") if err != nil { return fmt.Errorf("failed to get worker template: %w", err) } // Write templates if err := os.WriteFile(filepath.Join(destDir, "controlplane.yaml"), controlplaneData, 0644); err != nil { return fmt.Errorf("failed to write controlplane template: %w", err) } if err := os.WriteFile(filepath.Join(destDir, "worker.yaml"), workerData, 0644); err != nil { return fmt.Errorf("failed to write worker template: %w", err) } return nil } // updateNodeStatus updates node status flags func (m *Manager) updateNodeStatus(instanceName string, node *Node) error { return m.updateNodeStatusFiles(instanceName, node) } // updateNodeStatusFiles writes cache (applied/configured/currentIp) and updates maintenance in config (new format) func (m *Manager) updateNodeStatusFiles(instanceName string, node *Node) error { cacheDir := filepath.Dir(tools.GetNodeCachePath(m.dataDir, instanceName, node.Hostname)) if err := os.MkdirAll(cacheDir, 0755); err != nil { return fmt.Errorf("failed to create node cache dir: %w", err) } cache := nodeCacheFile{ Applied: node.Applied, Configured: node.Configured, CurrentIP: node.CurrentIP, } cacheData, err := yaml.Marshal(cache) if err != nil { return fmt.Errorf("failed to marshal node cache: %w", err) } if err := os.WriteFile(tools.GetNodeCachePath(m.dataDir, instanceName, node.Hostname), cacheData, 0644); err != nil { return fmt.Errorf("failed to write node cache: %w", err) } // Update maintenance in config/nodes/{hostname}.yaml configPath := tools.GetNodeConfigPath(m.dataDir, instanceName, node.Hostname) configData, err := os.ReadFile(configPath) if err != nil { return fmt.Errorf("failed to read node config: %w", err) } var cfg nodeConfigFile if err := yaml.Unmarshal(configData, &cfg); err != nil { return fmt.Errorf("failed to parse node config: %w", err) } cfg.Maintenance = node.Maintenance updatedData, err := yaml.Marshal(cfg) if err != nil { return fmt.Errorf("failed to marshal node config: %w", err) } return os.WriteFile(configPath, updatedData, 0644) } // Update modifies existing node configuration with partial updates func (m *Manager) Update(instanceName string, hostname string, updates map[string]interface{}) error { return m.updateNodeFile(instanceName, hostname, updates) } // updateNodeFile applies partial updates to config/nodes/{hostname}.yaml (new format) func (m *Manager) updateNodeFile(instanceName, hostname string, updates map[string]interface{}) error { configPath := tools.GetNodeConfigPath(m.dataDir, instanceName, hostname) data, err := os.ReadFile(configPath) if err != nil { return fmt.Errorf("node %s not found: %w", hostname, err) } var cfg nodeConfigFile if err := yaml.Unmarshal(data, &cfg); err != nil { return fmt.Errorf("failed to parse node config: %w", err) } for key, value := range updates { switch key { case "target_ip": if strVal, ok := value.(string); ok { cfg.TargetIP = strVal } case "current_ip": // currentIp is cache; update via cache file (handled by updateNodeStatus) case "disk": if strVal, ok := value.(string); ok { cfg.Disk = strVal } case "interface": if strVal, ok := value.(string); ok { cfg.Interface = strVal } case "version": if strVal, ok := value.(string); ok { cfg.Version = strVal } case "schematic_id": if strVal, ok := value.(string); ok { cfg.SchematicID = strVal } case "maintenance": if boolVal, ok := value.(bool); ok { cfg.Maintenance = boolVal } } } out, err := yaml.Marshal(cfg) if err != nil { return fmt.Errorf("failed to marshal node config: %w", err) } return os.WriteFile(configPath, out, 0644) } // FetchTemplates extracts patch templates from embedded files to instance func (m *Manager) FetchTemplates(instanceName string) error { destDir := filepath.Join(tools.GetNodeSetupCachePath(m.dataDir, instanceName), "patch.templates") return m.extractEmbeddedTemplates(destDir) } // fillTalosDefaults reads cluster talos version/schematicId defaults into node from instance.yaml func (m *Manager) fillTalosDefaults(instanceName string, node *Node) { instanceYaml := tools.GetInstanceYamlPath(m.dataDir, instanceName) if data, err := os.ReadFile(instanceYaml); err == nil { var root struct { Cluster struct { Nodes struct { Talos struct { SchematicID string `yaml:"schematicId"` Version string `yaml:"version"` } `yaml:"talos"` } `yaml:"nodes"` } `yaml:"cluster"` } if err := yaml.Unmarshal(data, &root); err == nil { if node.SchematicID == "" { node.SchematicID = root.Cluster.Nodes.Talos.SchematicID } if node.Version == "" { node.Version = root.Cluster.Nodes.Talos.Version } } } } // NodeHealth represents the health status of a node type NodeHealth struct { Node string `json:"node"` Services []tools.ServiceStatus `json:"services"` DmesgErrors []tools.DmesgError `json:"dmesgErrors"` Healthy bool `json:"healthy"` } // Health checks node health by querying Talos service statuses and scanning dmesg for errors func (m *Manager) Health(instanceName, nodeIdentifier string) (*NodeHealth, error) { node, err := m.Get(instanceName, nodeIdentifier) if err != nil { return nil, fmt.Errorf("node not found: %w", err) } if !node.Applied || node.Maintenance { return nil, fmt.Errorf("health check requires an applied, non-maintenance node") } ip := node.TargetIP if ip == "" { return nil, fmt.Errorf("no IP address available for node %s", node.Hostname) } // Fetch services and dmesg concurrently var services []tools.ServiceStatus var dmesgRaw string var svcErr, dmesgErr error var wg sync.WaitGroup wg.Add(2) go func() { defer wg.Done() services, svcErr = m.talosctl.GetServices(ip) }() go func() { defer wg.Done() dmesgRaw, dmesgErr = m.talosctl.GetDmesg(ip) }() wg.Wait() if svcErr != nil { return nil, fmt.Errorf("failed to get services: %w", svcErr) } var dmesgErrors []tools.DmesgError if dmesgErr == nil { dmesgErrors = tools.ParseDmesgErrors(dmesgRaw) } if dmesgErrors == nil { dmesgErrors = []tools.DmesgError{} } // Compute overall health healthy := len(dmesgErrors) == 0 for _, svc := range services { if !svc.Healthy && svc.HealthMessage != "" { healthy = false break } } return &NodeHealth{ Node: node.Hostname, Services: services, DmesgErrors: dmesgErrors, Healthy: healthy, }, nil } // Reboot restarts a node without wiping state func (m *Manager) Reboot(instanceName, nodeIdentifier string) error { slog.Info("rebooting node", "component", "node", "instance", instanceName, "node", nodeIdentifier) node, err := m.Get(instanceName, nodeIdentifier) if err != nil { return fmt.Errorf("node not found: %w", err) } rebootIP := node.TargetIP if rebootIP == "" { rebootIP = node.CurrentIP } if rebootIP == "" { return fmt.Errorf("no IP address available for node %s", node.Hostname) } if err := m.talosctl.Reboot(rebootIP); err != nil { return fmt.Errorf("failed to reboot node: %w", err) } slog.Info("node reboot initiated", "component", "node", "instance", instanceName, "hostname", node.Hostname, "ip", rebootIP) return nil } // Reset resets a node to maintenance mode with resilient handling. // For control plane nodes, it first removes the etcd member from the cluster // via a healthy peer, then resets the node. The node stays in config.yaml // so it can be reconfigured and rejoined. func (m *Manager) Reset(instanceName, nodeIdentifier string) error { slog.Info("resetting node", "component", "node", "instance", instanceName, "node", nodeIdentifier) // Get node node, err := m.Get(instanceName, nodeIdentifier) if err != nil { return fmt.Errorf("node not found: %w", err) } // Determine IP to reset resetIP := node.CurrentIP if resetIP == "" { resetIP = node.TargetIP } if resetIP == "" { return fmt.Errorf("no IP address available for node %s", node.Hostname) } // For control plane nodes, remove from etcd first via a healthy peer if node.Role == "controlplane" { if err := m.removeEtcdMember(instanceName, node); err != nil { // Log but don't fail — the node may already be removed from etcd, // or etcd may be unreachable on this node. The reset should still proceed. slog.Warn("etcd member removal failed (continuing with reset)", "component", "node", "instance", instanceName, "hostname", node.Hostname, "error", err) } } // Reset the node via talosctl if err := m.talosctl.Reset(resetIP); err != nil { return fmt.Errorf("failed to reset node: %w", err) } // Update node status to maintenance mode but keep in config for reconfiguration node.Maintenance = true node.Configured = false node.Applied = false node.CurrentIP = "" node.Version = "" if err := m.updateNodeStatus(instanceName, node); err != nil { return fmt.Errorf("failed to update node status: %w", err) } slog.Info("node reset to maintenance mode", "component", "node", "instance", instanceName, "hostname", node.Hostname) return nil } // removeEtcdMember removes a control plane node's etcd member via a healthy peer. func (m *Manager) removeEtcdMember(instanceName string, targetNode *Node) error { // Find a healthy control plane peer to run the etcd remove-member command nodes, err := m.List(instanceName) if err != nil { return fmt.Errorf("failed to list nodes: %w", err) } for _, peer := range nodes { // Skip the target node itself, non-control-plane nodes, and unapplied nodes if peer.Hostname == targetNode.Hostname || peer.Role != "controlplane" || !peer.Applied || peer.Maintenance { continue } peerIP := peer.CurrentIP if peerIP == "" { peerIP = peer.TargetIP } if peerIP == "" { continue } slog.Info("removing etcd member via peer", "component", "node", "instance", instanceName, "target", targetNode.Hostname, "peer", peer.Hostname) if err := m.talosctl.EtcdRemoveMember(peerIP, targetNode.Hostname); err != nil { slog.Warn("etcd remove-member failed via peer, trying next", "component", "node", "peer", peer.Hostname, "error", err) continue } slog.Info("etcd member removed successfully", "component", "node", "instance", instanceName, "target", targetNode.Hostname, "via", peer.Hostname) return nil } return fmt.Errorf("no healthy control plane peer available to remove etcd member for %s", targetNode.Hostname) }