toki
695df72eac
- Add PHASE.md for cli-agent-group-grade-routing milestone - Add SDD for automation-runtime-bridge - Update edge configrefresh classify logic and tests - Update node config_set, node.go, and run_manager for runtime config refresh - Archive old plan/review docs for 06+05_node_drain_runtime_config
454 lines
15 KiB
Go
454 lines
15 KiB
Go
package configrefresh
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"reflect"
|
||
"sort"
|
||
"strings"
|
||
|
||
"iop/apps/edge/internal/edgevalidate"
|
||
"iop/packages/go/config"
|
||
)
|
||
|
||
// applyRuntimeNormalization normalizes candidate config so that classification
|
||
// uses the same effective values as a running serve process.
|
||
func applyRuntimeNormalization(cfg *config.EdgeConfig) {
|
||
// Normalize logging.path – must match edgecmd.ResolveEdgeLogPath exactly:
|
||
// - empty → resolve to binary-dir/logs/edge.log
|
||
// - absolute → preserve exactly
|
||
// - relative → preserve exactly (include ./ segments)
|
||
// Serve keeps explicit relative paths like "./logs/edge.log" unchanged,
|
||
// so classifying against a cleaned value produces false restart_required.
|
||
if cfg.Logging.Path == "" {
|
||
cfg.Logging.Path = defaultEdgeLogPath()
|
||
}
|
||
// NOTE: explicit absolute / relative paths are intentionally left untouched.
|
||
|
||
// Normalize bootstrap.artifact_dir – resolve relative paths to absolute.
|
||
cfg.Bootstrap.ArtifactDir = resolveBootstrapArtifactDir(cfg.Bootstrap.ArtifactDir)
|
||
}
|
||
|
||
func defaultEdgeLogPath() string {
|
||
if exe, err := os.Executable(); err == nil {
|
||
return filepath.Join(filepath.Dir(exe), "logs", "edge.log")
|
||
}
|
||
if wd, err := os.Getwd(); err == nil {
|
||
return filepath.Join(wd, "logs", "edge.log")
|
||
}
|
||
return "logs/edge.log"
|
||
}
|
||
|
||
func resolveBootstrapArtifactDir(dir string) string {
|
||
if dir == "" {
|
||
dir = "artifacts"
|
||
}
|
||
if filepath.IsAbs(dir) {
|
||
return dir
|
||
}
|
||
return filepath.Join(binaryDir(), dir)
|
||
}
|
||
|
||
func binaryDir() string {
|
||
if exe, err := os.Executable(); err == nil {
|
||
return filepath.Dir(exe)
|
||
}
|
||
if wd, err := os.Getwd(); err == nil {
|
||
return wd
|
||
}
|
||
return "."
|
||
}
|
||
|
||
// LoadCandidate loads and validates an EdgeConfig from the given path, then
|
||
// applies the same runtime normalization that iop-edge serve uses. This
|
||
// ensures that classification compares effective (normalized) values and
|
||
// prevents false restart_required results when source-level YAML is
|
||
// identical but paths are relative or empty.
|
||
func LoadCandidate(path string) (*config.EdgeConfig, error) {
|
||
cfg, err := config.LoadEdge(path)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if err := edgevalidate.ValidateEdgeConfig(cfg); err != nil {
|
||
return nil, err
|
||
}
|
||
applyRuntimeNormalization(cfg)
|
||
return cfg, nil
|
||
}
|
||
|
||
// providerKey identifies a provider entry for diffing.
|
||
type providerKey struct {
|
||
NodeKey string
|
||
Type string
|
||
Category config.Category
|
||
Adapter string
|
||
Models []string
|
||
Health string
|
||
Capacity int
|
||
MaxQueue int
|
||
QueueTimeoutMS int
|
||
LifecycleCapabilities []string
|
||
}
|
||
|
||
func buildProviderIndex(cfg *config.EdgeConfig) map[string]providerKey {
|
||
idx := make(map[string]providerKey)
|
||
for i, node := range cfg.Nodes {
|
||
nodeKey := nodeIdentity(node, i)
|
||
for _, p := range node.Providers {
|
||
idx[p.ID] = providerKey{
|
||
NodeKey: nodeKey,
|
||
Type: p.Type,
|
||
Category: p.Category,
|
||
Adapter: p.Adapter,
|
||
Models: append([]string(nil), p.Models...),
|
||
Health: p.Health,
|
||
Capacity: p.Capacity,
|
||
MaxQueue: p.MaxQueue,
|
||
QueueTimeoutMS: p.QueueTimeoutMS,
|
||
LifecycleCapabilities: append([]string(nil), p.LifecycleCapabilities...),
|
||
}
|
||
}
|
||
}
|
||
return idx
|
||
}
|
||
|
||
type nodeKey struct {
|
||
Alias string
|
||
Token string
|
||
AgentKind string
|
||
Adapters config.AdaptersConf
|
||
Runtime config.RuntimeConf
|
||
}
|
||
|
||
func buildNodeIndex(cfg *config.EdgeConfig) map[string]nodeKey {
|
||
idx := make(map[string]nodeKey, len(cfg.Nodes))
|
||
for i, node := range cfg.Nodes {
|
||
idx[nodeIdentity(node, i)] = nodeKey{
|
||
Alias: node.Alias,
|
||
Token: node.Token,
|
||
AgentKind: node.AgentKind,
|
||
Adapters: node.Adapters,
|
||
Runtime: node.Runtime,
|
||
}
|
||
}
|
||
return idx
|
||
}
|
||
|
||
func nodeIdentity(node config.NodeDefinition, index int) string {
|
||
if node.ID != "" {
|
||
return node.ID
|
||
}
|
||
if node.Alias != "" {
|
||
return node.Alias
|
||
}
|
||
return fmt.Sprintf("#%d", index)
|
||
}
|
||
|
||
func buildModelIndex(cfg *config.EdgeConfig) map[string]config.ModelCatalogEntry {
|
||
idx := make(map[string]config.ModelCatalogEntry, len(cfg.Models))
|
||
for _, model := range cfg.Models {
|
||
idx[model.ID] = model
|
||
}
|
||
return idx
|
||
}
|
||
|
||
func appendIfChanged[T comparable](changes *[]Change, path string, class Status, previous, next T) {
|
||
if previous == next {
|
||
return
|
||
}
|
||
*changes = append(*changes, Change{
|
||
Path: path,
|
||
Class: class,
|
||
Previous: fmt.Sprint(previous),
|
||
Next: fmt.Sprint(next),
|
||
})
|
||
}
|
||
|
||
func appendDeepIfChanged(changes *[]Change, path string, class Status, previous, next interface{}) {
|
||
if reflect.DeepEqual(previous, next) {
|
||
return
|
||
}
|
||
*changes = append(*changes, Change{
|
||
Path: path,
|
||
Class: class,
|
||
Previous: fmt.Sprintf("%v", previous),
|
||
Next: fmt.Sprintf("%v", next),
|
||
})
|
||
}
|
||
|
||
// Classify computes the diff between current and candidate configs and returns
|
||
// the classified change list and an overall status.
|
||
func Classify(current, candidate *config.EdgeConfig) Result {
|
||
var changes []Change
|
||
|
||
// --- restart_required fields ---
|
||
appendIfChanged(&changes, "edge.id", StatusRestartRequired, current.Edge.ID, candidate.Edge.ID)
|
||
appendIfChanged(&changes, "edge.name", StatusRestartRequired, current.Edge.Name, candidate.Edge.Name)
|
||
appendIfChanged(&changes, "server.listen", StatusRestartRequired, current.Server.Listen, candidate.Server.Listen)
|
||
appendIfChanged(&changes, "server.advertise_host", StatusRestartRequired, current.Server.AdvertiseHost, candidate.Server.AdvertiseHost)
|
||
appendIfChanged(&changes, "bootstrap.listen", StatusRestartRequired, current.Bootstrap.Listen, candidate.Bootstrap.Listen)
|
||
appendIfChanged(&changes, "bootstrap.artifact_base_url", StatusRestartRequired, current.Bootstrap.ArtifactBaseURL, candidate.Bootstrap.ArtifactBaseURL)
|
||
appendIfChanged(&changes, "bootstrap.artifact_dir", StatusRestartRequired, current.Bootstrap.ArtifactDir, candidate.Bootstrap.ArtifactDir)
|
||
appendDeepIfChanged(&changes, "tls", StatusRestartRequired, current.TLS, candidate.TLS)
|
||
appendDeepIfChanged(&changes, "logging", StatusRestartRequired, current.Logging, candidate.Logging)
|
||
appendIfChanged(&changes, "metrics.port", StatusRestartRequired, current.Metrics.Port, candidate.Metrics.Port)
|
||
appendDeepIfChanged(&changes, "console", StatusRestartRequired, current.Console, candidate.Console)
|
||
appendDeepIfChanged(&changes, "control_plane", StatusRestartRequired, current.ControlPlane, candidate.ControlPlane)
|
||
appendDeepIfChanged(&changes, "refresh", StatusRestartRequired, current.Refresh, candidate.Refresh)
|
||
appendDeepIfChanged(&changes, "openai", StatusRestartRequired, current.OpenAI, candidate.OpenAI)
|
||
appendDeepIfChanged(&changes, "a2a", StatusRestartRequired, current.A2A, candidate.A2A)
|
||
|
||
currentNodes := buildNodeIndex(current)
|
||
candidateNodes := buildNodeIndex(candidate)
|
||
for key, cur := range currentNodes {
|
||
next, exists := candidateNodes[key]
|
||
if !exists {
|
||
changes = append(changes, Change{
|
||
Path: fmt.Sprintf("nodes[%q]", key),
|
||
Class: StatusRestartRequired,
|
||
Previous: "present",
|
||
Next: "absent",
|
||
})
|
||
continue
|
||
}
|
||
appendIfChanged(&changes, fmt.Sprintf("nodes[%q].alias", key), StatusRestartRequired, cur.Alias, next.Alias)
|
||
appendIfChanged(&changes, fmt.Sprintf("nodes[%q].token", key), StatusRestartRequired, cur.Token, next.Token)
|
||
appendIfChanged(&changes, fmt.Sprintf("nodes[%q].agent_kind", key), StatusRestartRequired, cur.AgentKind, next.AgentKind)
|
||
appendDeepIfChanged(&changes, fmt.Sprintf("nodes[%q].adapters", key), StatusRestartRequired, cur.Adapters, next.Adapters)
|
||
// Runtime tuning is classified per field: node-wide concurrency is applied
|
||
// live, while workspace_root requires a node restart because the store DB
|
||
// path is fixed at bootstrap.
|
||
appendIfChanged(&changes, fmt.Sprintf("nodes[%q].runtime.concurrency", key), StatusApplied, cur.Runtime.Concurrency, next.Runtime.Concurrency)
|
||
appendIfChanged(&changes, fmt.Sprintf("nodes[%q].runtime.workspace_root", key), StatusRestartRequired, cur.Runtime.WorkspaceRoot, next.Runtime.WorkspaceRoot)
|
||
}
|
||
for key := range candidateNodes {
|
||
if _, exists := currentNodes[key]; !exists {
|
||
changes = append(changes, Change{
|
||
Path: fmt.Sprintf("nodes[%q]", key),
|
||
Class: StatusRestartRequired,
|
||
Previous: "absent",
|
||
Next: "present",
|
||
})
|
||
}
|
||
}
|
||
|
||
// --- mutable applied fields: provider capacity ---
|
||
currentProviders := buildProviderIndex(current)
|
||
candidateProviders := buildProviderIndex(candidate)
|
||
|
||
for provID, cur := range currentProviders {
|
||
next, exists := candidateProviders[provID]
|
||
if !exists {
|
||
changes = append(changes, Change{
|
||
Path: fmt.Sprintf("nodes[].providers[%q]", provID),
|
||
Class: StatusRestartRequired,
|
||
Previous: "present",
|
||
Next: "absent",
|
||
})
|
||
continue
|
||
}
|
||
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].node", provID), StatusRestartRequired, cur.NodeKey, next.NodeKey)
|
||
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].type", provID), StatusRestartRequired, cur.Type, next.Type)
|
||
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].category", provID), StatusRestartRequired, cur.Category, next.Category)
|
||
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].adapter", provID), StatusRestartRequired, cur.Adapter, next.Adapter)
|
||
appendDeepIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].models", provID), StatusRestartRequired, cur.Models, next.Models)
|
||
appendIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].health", provID), StatusRestartRequired, cur.Health, next.Health)
|
||
appendDeepIfChanged(&changes, fmt.Sprintf("nodes[].providers[%q].lifecycle_capabilities", provID), StatusRestartRequired, cur.LifecycleCapabilities, next.LifecycleCapabilities)
|
||
}
|
||
for provID := range candidateProviders {
|
||
if _, exists := currentProviders[provID]; !exists {
|
||
changes = append(changes, Change{
|
||
Path: fmt.Sprintf("nodes[].providers[%q]", provID),
|
||
Class: StatusRestartRequired,
|
||
Previous: "absent",
|
||
Next: "present",
|
||
})
|
||
}
|
||
}
|
||
for provID, cp := range candidateProviders {
|
||
cur, exists := currentProviders[provID]
|
||
if !exists {
|
||
continue
|
||
}
|
||
if cur.Capacity != cp.Capacity {
|
||
changes = append(changes, Change{
|
||
Path: fmt.Sprintf("nodes[].providers[%q].capacity", provID),
|
||
Class: StatusApplied,
|
||
Previous: fmt.Sprintf("%d", cur.Capacity),
|
||
Next: fmt.Sprintf("%d", cp.Capacity),
|
||
})
|
||
}
|
||
if cur.MaxQueue != cp.MaxQueue {
|
||
changes = append(changes, Change{
|
||
Path: fmt.Sprintf("nodes[].providers[%q].max_queue", provID),
|
||
Class: StatusApplied,
|
||
Previous: fmt.Sprintf("%d", cur.MaxQueue),
|
||
Next: fmt.Sprintf("%d", cp.MaxQueue),
|
||
})
|
||
}
|
||
if cur.QueueTimeoutMS != cp.QueueTimeoutMS {
|
||
changes = append(changes, Change{
|
||
Path: fmt.Sprintf("nodes[].providers[%q].queue_timeout_ms", provID),
|
||
Class: StatusApplied,
|
||
Previous: fmt.Sprintf("%d", cur.QueueTimeoutMS),
|
||
Next: fmt.Sprintf("%d", cp.QueueTimeoutMS),
|
||
})
|
||
}
|
||
}
|
||
|
||
currentModels := buildModelIndex(current)
|
||
candidateModels := buildModelIndex(candidate)
|
||
for modelID, cur := range currentModels {
|
||
next, exists := candidateModels[modelID]
|
||
if !exists {
|
||
changes = append(changes, Change{
|
||
Path: fmt.Sprintf("models[%q]", modelID),
|
||
Class: StatusApplied,
|
||
Previous: "present",
|
||
Next: "absent",
|
||
})
|
||
continue
|
||
}
|
||
appendIfChanged(&changes, fmt.Sprintf("models[%q].display_name", modelID), StatusApplied, cur.DisplayName, next.DisplayName)
|
||
appendDeepIfChanged(&changes, fmt.Sprintf("models[%q].providers", modelID), StatusApplied, cur.Providers, next.Providers)
|
||
}
|
||
for modelID := range candidateModels {
|
||
if _, exists := currentModels[modelID]; !exists {
|
||
changes = append(changes, Change{
|
||
Path: fmt.Sprintf("models[%q]", modelID),
|
||
Class: StatusApplied,
|
||
Previous: "absent",
|
||
Next: "present",
|
||
})
|
||
}
|
||
}
|
||
|
||
sort.SliceStable(changes, func(i, j int) bool {
|
||
if changes[i].Path == changes[j].Path {
|
||
return changes[i].Class < changes[j].Class
|
||
}
|
||
return changes[i].Path < changes[j].Path
|
||
})
|
||
|
||
// Determine overall status: any restart_required wins over applied.
|
||
hasRestart := false
|
||
for _, c := range changes {
|
||
if c.Class == StatusRestartRequired {
|
||
hasRestart = true
|
||
break
|
||
}
|
||
}
|
||
|
||
overallStatus := StatusApplied
|
||
summary := "no changes detected"
|
||
if hasRestart {
|
||
overallStatus = StatusRestartRequired
|
||
summary = "some changes require a process restart"
|
||
} else if len(changes) > 0 {
|
||
summary = "all changes can be applied without restart"
|
||
}
|
||
|
||
result := Result{
|
||
Status: overallStatus,
|
||
Changes: changes,
|
||
Summary: summary,
|
||
}
|
||
deriveReport(&result)
|
||
return result
|
||
}
|
||
|
||
// deriveReport populates the ops-report fields (changed node/provider/model ids
|
||
// and restart-required paths) from the classified change list. All slices are
|
||
// initialized non-nil so the JSON response stays stable for ops assertions.
|
||
func deriveReport(r *Result) {
|
||
nodes := newStringSet()
|
||
providers := newStringSet()
|
||
models := newStringSet()
|
||
var restartPaths []string
|
||
for _, c := range r.Changes {
|
||
if c.Class == StatusRestartRequired {
|
||
restartPaths = append(restartPaths, c.Path)
|
||
}
|
||
// Provider paths are nested under nodes[].providers[...]; check them
|
||
// before the broader nodes[ prefix so they are attributed correctly.
|
||
if id, ok := extractBracketID(c.Path, "providers["); ok {
|
||
providers.add(id)
|
||
continue
|
||
}
|
||
if id, ok := extractBracketID(c.Path, "nodes["); ok {
|
||
nodes.add(id)
|
||
continue
|
||
}
|
||
if id, ok := extractBracketID(c.Path, "models["); ok {
|
||
models.add(id)
|
||
}
|
||
}
|
||
r.ChangedNodes = nodes.sorted()
|
||
r.ChangedProviders = providers.sorted()
|
||
r.ChangedModels = models.sorted()
|
||
sort.Strings(restartPaths)
|
||
if restartPaths == nil {
|
||
restartPaths = []string{}
|
||
}
|
||
r.RestartRequiredPaths = restartPaths
|
||
}
|
||
|
||
// extractBracketID returns the quoted id immediately following prefix in path,
|
||
// e.g. extractBracketID(`nodes["node-1"].alias`, "nodes[") returns "node-1".
|
||
func extractBracketID(path, prefix string) (string, bool) {
|
||
i := strings.Index(path, prefix)
|
||
if i < 0 {
|
||
return "", false
|
||
}
|
||
rest := path[i+len(prefix):]
|
||
if len(rest) == 0 || rest[0] != '"' {
|
||
return "", false
|
||
}
|
||
rest = rest[1:]
|
||
j := strings.IndexByte(rest, '"')
|
||
if j <= 0 {
|
||
return "", false
|
||
}
|
||
return rest[:j], true
|
||
}
|
||
|
||
type stringSet struct {
|
||
m map[string]struct{}
|
||
}
|
||
|
||
func newStringSet() stringSet { return stringSet{m: make(map[string]struct{})} }
|
||
|
||
func (s stringSet) add(v string) { s.m[v] = struct{}{} }
|
||
|
||
func (s stringSet) sorted() []string {
|
||
out := make([]string, 0, len(s.m))
|
||
for v := range s.m {
|
||
out = append(out, v)
|
||
}
|
||
sort.Strings(out)
|
||
return out
|
||
}
|
||
|
||
// Evaluate loads the candidate config from req.ConfigPath, validates it, and
|
||
// classifies changes against current. On success it also returns the loaded
|
||
// candidate so the caller can apply mutable changes in apply mode.
|
||
func Evaluate(_ context.Context, current *config.EdgeConfig, req Request) (Result, *config.EdgeConfig, error) {
|
||
if req.Mode != ModeDryRun && req.Mode != ModeApply {
|
||
return RejectedResult(req, fmt.Sprintf("invalid mode %q: must be dry_run or apply", req.Mode)), nil, nil
|
||
}
|
||
|
||
candidate, err := LoadCandidate(req.ConfigPath)
|
||
if err != nil {
|
||
return RejectedResult(req, fmt.Sprintf("load candidate config: %v", err)), nil, nil
|
||
}
|
||
|
||
result := Classify(current, candidate)
|
||
result.Mode = req.Mode
|
||
result.RequestID = req.RequestID
|
||
if result.Changes == nil {
|
||
result.Changes = []Change{}
|
||
}
|
||
return result, candidate, nil
|
||
}
|