127 lines
4.3 KiB
Go
127 lines
4.3 KiB
Go
package node
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"go.uber.org/zap"
|
|
|
|
"iop/apps/node/internal/adapters"
|
|
"iop/apps/node/internal/router"
|
|
"iop/apps/node/internal/transport"
|
|
iop "iop/proto/gen/iop"
|
|
)
|
|
|
|
// OnConfigRefresh handles a config refresh request from edge.
|
|
func (n *Node) OnConfigRefresh(ctx context.Context, _ *transport.Session, req *iop.NodeConfigRefreshRequest) (*iop.NodeConfigRefreshResponse, error) {
|
|
n.logger.Info("config refresh request received", zap.String("request_id", req.GetRequestId()))
|
|
|
|
if req.GetConfig() == nil {
|
|
if len(req.GetChangedPaths()) == 0 {
|
|
return &iop.NodeConfigRefreshResponse{
|
|
RequestId: req.GetRequestId(),
|
|
Status: iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED,
|
|
}, nil
|
|
}
|
|
return &iop.NodeConfigRefreshResponse{
|
|
RequestId: req.GetRequestId(),
|
|
Status: iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_RESTART_REQUIRED,
|
|
RestartRequiredPaths: req.GetChangedPaths(),
|
|
}, nil
|
|
}
|
|
|
|
nextSet, err := adapters.BuildConfigSet(req.GetConfig(), n.logger)
|
|
if err != nil {
|
|
n.logger.Error("config refresh: failed to build next config set", zap.Error(err))
|
|
return &iop.NodeConfigRefreshResponse{
|
|
RequestId: req.GetRequestId(),
|
|
Status: iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_FAILED,
|
|
Error: fmt.Sprintf("build config set: %s", err.Error()),
|
|
}, nil
|
|
}
|
|
|
|
n.configSetMu.Lock()
|
|
defer n.configSetMu.Unlock()
|
|
|
|
// Runtime concurrency is no longer applied live to admission gates.
|
|
// Config metadata is preserved in currentConfigSet.Runtime.
|
|
// Per-adapter capacity from Capabilities().MaxConcurrency is the sole admission source.
|
|
diff := adapters.DiffConfigSets(n.currentConfigSet, nextSet)
|
|
|
|
if len(diff.Added) == 0 && len(diff.Updated) == 0 && len(diff.Removed) == 0 {
|
|
n.logger.Info("config refresh: no adapter changes detected")
|
|
if n.currentConfigSet != nil {
|
|
n.currentConfigSet.Runtime = nextSet.Runtime
|
|
}
|
|
return &iop.NodeConfigRefreshResponse{
|
|
RequestId: req.GetRequestId(),
|
|
Status: iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED,
|
|
}, nil
|
|
}
|
|
|
|
n.logger.Info("config refresh: adapter diff calculated",
|
|
zap.Strings("added", diff.Added),
|
|
zap.Strings("updated", diff.Updated),
|
|
zap.Strings("removed", diff.Removed),
|
|
)
|
|
|
|
// Start the new registry
|
|
if err := nextSet.Registry.Start(ctx); err != nil {
|
|
n.logger.Error("config refresh: failed to start new adapters", zap.Error(err))
|
|
return &iop.NodeConfigRefreshResponse{
|
|
RequestId: req.GetRequestId(),
|
|
Status: iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_FAILED,
|
|
Error: fmt.Sprintf("start next registry: %s", err.Error()),
|
|
}, nil
|
|
}
|
|
|
|
// Live swap registry in router
|
|
if mr, ok := n.router.(router.MutableRouter); ok {
|
|
mr.SetRegistry(nextSet.Registry)
|
|
} else {
|
|
n.logger.Warn("router does not implement MutableRouter, cannot swap registry")
|
|
}
|
|
|
|
// Update existing adapter gates capacity
|
|
n.adapterGatesMu.Lock()
|
|
for key, gate := range n.adapterGates {
|
|
if adapter, ok := nextSet.Registry.Get(key); ok {
|
|
if caps, err := adapter.Capabilities(ctx); err == nil {
|
|
gate.updateCapacity(caps.MaxConcurrency)
|
|
}
|
|
}
|
|
}
|
|
n.adapterGatesMu.Unlock()
|
|
|
|
// Registry.Stop stops every lifecycle adapter in the old registry. Pre-swap
|
|
// runs still hold old adapter instances, so the old registry stop is deferred
|
|
// until those runs drain; in-flight executions then finish on their original
|
|
// adapter snapshot while new requests already route to nextSet.
|
|
oldSet := n.currentConfigSet
|
|
pendingRuns := n.runs.snapshotActive()
|
|
|
|
if len(pendingRuns) == 0 {
|
|
if oldSet != nil && oldSet.Registry != nil {
|
|
if err := oldSet.Registry.Stop(context.Background()); err != nil {
|
|
n.logger.Warn("config refresh: failed to stop old adapters", zap.Error(err))
|
|
}
|
|
}
|
|
} else if oldSet != nil && oldSet.Registry != nil {
|
|
n.logger.Info("config refresh: deferring old registry stop until active runs drain",
|
|
zap.Int("pending_runs", len(pendingRuns)),
|
|
)
|
|
go func() {
|
|
waitHandles(context.Background(), pendingRuns)
|
|
if err := oldSet.Registry.Stop(context.Background()); err != nil {
|
|
n.logger.Warn("config refresh: failed to stop old adapters after drain", zap.Error(err))
|
|
}
|
|
}()
|
|
}
|
|
|
|
n.currentConfigSet = nextSet
|
|
|
|
return &iop.NodeConfigRefreshResponse{
|
|
RequestId: req.GetRequestId(),
|
|
Status: iop.NodeConfigRefreshStatus_NODE_CONFIG_REFRESH_STATUS_APPLIED,
|
|
}, nil
|
|
}
|