aaaamoooo
80140f391f
Assign/unassign, customer-form alignment, suspend/resume and credential rotation no longer rebuild (and restart the node's sing-box) per change. They mark the node rebuild-pending; a 1s background flusher applies exactly one rebuild with the final state when the FIXED window — 10s from the first pending change, never extended by later changes — elapses. Manual deploy, repair, node edits and server reconciles stay immediate and consume the mark. Removal clears it, and the flusher discards marks on deleted / removing / draft nodes, so a pending apply can never resurrect a node. The mark is persisted in the data file, so a panel restart inside the window cannot strand a node: the first post-boot pass flushes whatever is due.
1576 lines
43 KiB
Go
1576 lines
43 KiB
Go
package main
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"math"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
)
|
||
|
||
const HeartbeatReconcileIntervalMs = 120000
|
||
|
||
// ---------- small finders / helpers ----------
|
||
|
||
func orchFindServer(d *Data, id string) *Server {
|
||
for _, s := range d.Servers {
|
||
if s.ID == id {
|
||
return s
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func orchFindNode(d *Data, id string) *Node {
|
||
for _, n := range d.Nodes {
|
||
if n.ID == id {
|
||
return n
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func orchFindCustomer(d *Data, id string) *Customer {
|
||
for _, c := range d.Customers {
|
||
if c.ID == id {
|
||
return c
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func orchNotFound(label string) *HTTPError { return httpErr(404, "%s不存在", label) }
|
||
|
||
func xbTruncRunes(s string, n int) string {
|
||
r := []rune(s)
|
||
if len(r) > n {
|
||
return string(r[:n])
|
||
}
|
||
return s
|
||
}
|
||
|
||
func forwardNode(node *Node) bool {
|
||
topology := node.Topology
|
||
if topology == "" {
|
||
topology = "forward"
|
||
}
|
||
return topology == "forward"
|
||
}
|
||
|
||
// manualExit reports whether a forward node's exit is an unmanaged landing
|
||
// endpoint (round-2 item 8): no exit server, no exit resource, nothing to
|
||
// deploy on the far side.
|
||
func manualExit(node *Node) bool {
|
||
return forwardNode(node) && node.ExitMode == "manual"
|
||
}
|
||
|
||
func tagForNode(node *Node) string {
|
||
var sb strings.Builder
|
||
for _, r := range node.ID {
|
||
if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' {
|
||
sb.WriteRune(r)
|
||
}
|
||
}
|
||
s := sb.String()
|
||
if len(s) > 18 {
|
||
s = s[len(s)-18:]
|
||
}
|
||
return "ng-" + s
|
||
}
|
||
|
||
func activeNodeClients(d *Data, node *Node) []*NodeClient {
|
||
out := []*NodeClient{}
|
||
for _, nc := range d.NodeClients {
|
||
if nc.NodeID == node.ID && nc.Status == "active" {
|
||
out = append(out, nc)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func NodePortInUse(d *Data, serverID string, port int, excludeNodeID string) bool {
|
||
for _, n := range d.Nodes {
|
||
if n.ID == excludeNodeID {
|
||
continue
|
||
}
|
||
if n.ServerID == serverID && n.Port == port {
|
||
return true
|
||
}
|
||
if n.ExitServerID != nil && *n.ExitServerID == serverID && n.ExitPort != nil && *n.ExitPort == port {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
// ---------- ports ----------
|
||
|
||
func AllocatePort(d *Data, server *Server) (int, error) {
|
||
start := server.PortRangeStart
|
||
if start == 0 {
|
||
start = 20000
|
||
}
|
||
end := server.PortRangeEnd
|
||
if end == 0 {
|
||
end = 50000
|
||
}
|
||
if start < 1024 || end > 65535 || start > end {
|
||
return 0, fmt.Errorf("服务器 %s 的端口范围无效", server.Name)
|
||
}
|
||
span := end - start + 1
|
||
offset := xbRandInt(span)
|
||
for i := 0; i < span; i++ {
|
||
candidate := start + ((offset + i) % span)
|
||
if !NodePortInUse(d, server.ID, candidate, "") {
|
||
return candidate, nil
|
||
}
|
||
}
|
||
return 0, fmt.Errorf("服务器 %s 没有可用端口", server.Name)
|
||
}
|
||
|
||
// anyToJSNumber converts a decoded-JSON value the way JS Number() would;
|
||
// ok=false mirrors NaN.
|
||
func anyToJSNumber(v any) (float64, bool) {
|
||
switch t := v.(type) {
|
||
case json.Number:
|
||
f, err := t.Float64()
|
||
if err != nil {
|
||
return 0, false
|
||
}
|
||
return f, true
|
||
case float64:
|
||
return t, true
|
||
case float32:
|
||
return float64(t), true
|
||
case int:
|
||
return float64(t), true
|
||
case int64:
|
||
return float64(t), true
|
||
case string:
|
||
f := orchJSNumberStr(t)
|
||
if math.IsNaN(f) {
|
||
return 0, false
|
||
}
|
||
return f, true
|
||
case bool:
|
||
if t {
|
||
return 1, true
|
||
}
|
||
return 0, true
|
||
}
|
||
return 0, false
|
||
}
|
||
|
||
func jsFalsy(v any) bool {
|
||
if v == nil {
|
||
return true
|
||
}
|
||
if f, ok := anyToJSNumber(v); ok && f == 0 {
|
||
return true
|
||
}
|
||
if s, isStr := v.(string); isStr && s == "" {
|
||
return true
|
||
}
|
||
if b, isBool := v.(bool); isBool && !b {
|
||
return true
|
||
}
|
||
return false
|
||
}
|
||
|
||
func SelectPort(d *Data, server *Server, mode string, requested any, excludeNodeID string) (int, error) {
|
||
if mode != "fixed" || jsFalsy(requested) {
|
||
return AllocatePort(d, server)
|
||
}
|
||
f, ok := anyToJSNumber(requested)
|
||
if !ok || f != math.Trunc(f) || f < 1024 || f > 65535 {
|
||
return 0, httpErr(400, "服务器 %s 的固定端口无效", server.Name)
|
||
}
|
||
port := int(f)
|
||
if NodePortInUse(d, server.ID, port, excludeNodeID) {
|
||
return 0, httpErr(409, "服务器 %s 的端口 %d 已被占用", server.Name, port)
|
||
}
|
||
return port, nil
|
||
}
|
||
|
||
// ---------- jobs / audit ----------
|
||
|
||
func QueueJob(d *Data, serverID, nodeID, action string, payload *JobPayload) *Job {
|
||
now := NowISO()
|
||
job := &Job{
|
||
ID: NewID("job"),
|
||
ServerID: serverID,
|
||
NodeID: nodeID,
|
||
Action: action,
|
||
Payload: payload,
|
||
Status: "queued",
|
||
Attempts: 0,
|
||
CreatedAt: now,
|
||
UpdatedAt: now,
|
||
Error: nil,
|
||
}
|
||
d.Jobs = append(d.Jobs, job)
|
||
return job
|
||
}
|
||
|
||
func Audit(d *Data, actor, action, target string, detail map[string]any) {
|
||
if detail == nil {
|
||
detail = map[string]any{}
|
||
}
|
||
entry := &Activity{
|
||
ID: NewID("evt"),
|
||
At: NowISO(),
|
||
Actor: actor,
|
||
Action: action,
|
||
Target: target,
|
||
Detail: detail,
|
||
}
|
||
d.Activity = append([]*Activity{entry}, d.Activity...)
|
||
if len(d.Activity) > 2000 {
|
||
d.Activity = d.Activity[:2000]
|
||
}
|
||
}
|
||
|
||
// ---------- resource assembly ----------
|
||
|
||
func payloadWithResource(resource map[string]any, kind string, keyMaterial map[string]KeyPair) *JobPayload {
|
||
raw, err := marshalJSON(resource)
|
||
if err != nil {
|
||
panic(err)
|
||
}
|
||
payload := &JobPayload{Resource: raw, Kind: kind}
|
||
if keyMaterial != nil {
|
||
payload.KeyMaterial = keyMaterial
|
||
}
|
||
return payload
|
||
}
|
||
|
||
// ensureCustomerCredentials fills a missing customer-level credential set
|
||
// (defensive: the store-load migration and customer creation provide one).
|
||
func ensureCustomerCredentials(d *Data, customerID string) {
|
||
customer := findCustomer(d, customerID)
|
||
if customer != nil && (customer.Credentials == nil || customer.Credentials.ClientID == "") {
|
||
creds := NewCustomerCredentials()
|
||
customer.Credentials = &creds
|
||
}
|
||
}
|
||
|
||
func resourceForNode(d *Data, node *Node, clients []*NodeClient) (map[string]any, error) {
|
||
server := orchFindServer(d, node.ServerID)
|
||
if server == nil {
|
||
return nil, orchNotFound("服务器")
|
||
}
|
||
networkMode := node.NetworkMode
|
||
if networkMode == "" {
|
||
networkMode = "ipv4"
|
||
}
|
||
bindAddress, err := ResolveBindAddress(networkMode, server)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var exitServer *Server
|
||
exitAddress := ""
|
||
if manualExit(node) {
|
||
exitAddress = node.ExitAddress
|
||
} else if forwardNode(node) {
|
||
exitServer = orchFindServer(d, xbDeref(node.ExitServerID))
|
||
if exitServer == nil {
|
||
return nil, orchNotFound("出口服务器")
|
||
}
|
||
}
|
||
if clients == nil {
|
||
clients = activeNodeClients(d, node)
|
||
}
|
||
// Credentials live on the customer (round-2 item 2); stored bindings
|
||
// carry none. Stamp each customer's set onto throwaway copies for the
|
||
// resource builder so the stored records stay clean.
|
||
resolved := make([]*NodeClient, len(clients))
|
||
for i, nc := range clients {
|
||
ensureCustomerCredentials(d, nc.CustomerID)
|
||
cp := *nc
|
||
if customer := findCustomer(d, nc.CustomerID); customer != nil && customer.Credentials != nil {
|
||
creds := *customer.Credentials
|
||
cp.Credentials = &creds
|
||
} else {
|
||
cp.Credentials = nil
|
||
}
|
||
resolved[i] = &cp
|
||
}
|
||
return BuildNodeResource(BuildNodeResourceParams{
|
||
ResourceID: node.ResourceID,
|
||
TagPrefix: tagForNode(node),
|
||
Node: node,
|
||
NodeClients: resolved,
|
||
Server: server,
|
||
BindAddress: bindAddress,
|
||
ExitServer: exitServer,
|
||
ExitAddress: exitAddress,
|
||
})
|
||
}
|
||
|
||
func resourceForNodeExit(d *Data, node *Node) (map[string]any, error) {
|
||
server := orchFindServer(d, xbDeref(node.ExitServerID))
|
||
if server == nil {
|
||
return nil, orchNotFound("出口服务器")
|
||
}
|
||
networkMode := node.NetworkMode
|
||
if networkMode == "" {
|
||
networkMode = "ipv4"
|
||
}
|
||
bindAddress, err := ResolveBindAddress(networkMode, server)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return BuildNodeExitResource(BuildNodeExitResourceParams{
|
||
ResourceID: node.ExitResourceID,
|
||
TagPrefix: tagForNode(node),
|
||
Node: node,
|
||
Server: server,
|
||
BindAddress: bindAddress,
|
||
})
|
||
}
|
||
|
||
func cancelQueuedDeletes(d *Data, nodeID, reason string) bool {
|
||
cancelled := false
|
||
for _, job := range d.Jobs {
|
||
if job.NodeID == nodeID && job.Action == "delete_resource" && job.Status == "queued" {
|
||
job.Status = "failed"
|
||
job.Error = StrPtr(reason)
|
||
job.UpdatedAt = NowISO()
|
||
cancelled = true
|
||
}
|
||
}
|
||
return cancelled
|
||
}
|
||
|
||
func RebuildNode(d *Data, node *Node, includeExit bool) error {
|
||
// Round-3 item 20: a node whose bindings drop to zero is rebuilt
|
||
// like any other (its inbound just ends up with no customer users);
|
||
// it is NOT auto-removed. The pre-item-20 auto-removal branch is
|
||
// gone; the AutoRemoving revival paths in assign/resume stay for
|
||
// records an older panel left in that state.
|
||
clients := activeNodeClients(d, node)
|
||
node.AutoRemoving = BoolPtr(false)
|
||
resource, err := resourceForNode(d, node, clients)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
payload := payloadWithResource(resource, "node", KeyMaterialForNode(d, node))
|
||
QueueJob(d, node.ServerID, node.ID, "apply_resource", payload)
|
||
if forwardNode(node) && includeExit && !manualExit(node) {
|
||
exitResource, err := resourceForNodeExit(d, node)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
QueueJob(d, xbDeref(node.ExitServerID), node.ID, "apply_resource", payloadWithResource(exitResource, "node-exit", nil))
|
||
}
|
||
node.Status = "deploying"
|
||
node.Error = nil
|
||
node.UpdatedAt = NowISO()
|
||
return nil
|
||
}
|
||
|
||
func DeriveNodeStatus(node *Node) {
|
||
if !node.EntryReady {
|
||
if node.Status != "failed" {
|
||
node.Status = "deploying"
|
||
}
|
||
return
|
||
}
|
||
if forwardNode(node) && !node.ExitReady && !manualExit(node) {
|
||
if node.Status != "degraded" {
|
||
node.Status = "deploying"
|
||
}
|
||
return
|
||
}
|
||
node.Status = "active"
|
||
node.Error = nil
|
||
}
|
||
|
||
// ---------- coalesced rebuilds (round-3 item 23) ----------
|
||
|
||
// RebuildCoalesceWindow is the FIXED coalescing window for binding-class
|
||
// rebuilds: it starts at a node's first pending change and is never
|
||
// extended by later changes inside it, so a burst of assign/unassign
|
||
// operations can never postpone the dispatch indefinitely. When the
|
||
// window elapses the flusher applies exactly one rebuild with the
|
||
// node's final state.
|
||
const RebuildCoalesceWindow = 10 * time.Second
|
||
|
||
// markRebuildPending records a binding-class change for the coalesced
|
||
// flusher instead of rebuilding (and restarting the node's sing-box)
|
||
// immediately. The window is anchored at the first pending change:
|
||
// re-marking inside the window only widens the exit flag.
|
||
func markRebuildPending(node *Node, includeExit bool) {
|
||
if !node.RebuildPending {
|
||
node.RebuildPending = true
|
||
node.RebuildPendingAt = NowISO()
|
||
}
|
||
if includeExit {
|
||
node.RebuildPendingExit = true
|
||
}
|
||
node.UpdatedAt = NowISO()
|
||
}
|
||
|
||
// clearRebuildPending drops a pending coalesced rebuild — used when an
|
||
// immediate action (manual deploy, repair, node edit, removal, server
|
||
// reconcile) already ships the node's current state or ends its life.
|
||
func clearRebuildPending(node *Node) {
|
||
node.RebuildPending = false
|
||
node.RebuildPendingAt = ""
|
||
node.RebuildPendingExit = false
|
||
}
|
||
|
||
// FlushPendingRebuilds applies one rebuild per node whose fixed window
|
||
// has elapsed, with the node's state as of now. Pending marks on nodes
|
||
// that were deleted, are being removed, or fell back to draft are
|
||
// discarded — a pending apply must never resurrect a node. Called once
|
||
// per second from the panel's background loop; because the mark is
|
||
// persisted in the data file, a panel restart inside the window simply
|
||
// means the first post-boot pass flushes whatever is due.
|
||
func (o *Orchestrator) FlushPendingRebuilds(now time.Time) (int, error) {
|
||
flushed := 0
|
||
err := o.tx(func(d *Data) error {
|
||
for _, node := range d.Nodes {
|
||
if !node.RebuildPending {
|
||
continue
|
||
}
|
||
at, ok := ParseISO(node.RebuildPendingAt)
|
||
if !ok || now.Sub(at) < RebuildCoalesceWindow {
|
||
continue
|
||
}
|
||
includeExit := node.RebuildPendingExit
|
||
clearRebuildPending(node)
|
||
if node.PendingDelete || node.Status == "removing" || node.Status == "draft" {
|
||
continue
|
||
}
|
||
if err := RebuildNode(d, node, includeExit); err != nil {
|
||
node.Error = StrPtr(err.Error())
|
||
Audit(d, "system", "rebuild_flush_failed", node.ID, map[string]any{"error": err.Error()})
|
||
continue
|
||
}
|
||
flushed++
|
||
}
|
||
return nil
|
||
})
|
||
return flushed, err
|
||
}
|
||
|
||
// ---------- customer suspend / resume ----------
|
||
|
||
func SuspendCustomerResources(d *Data, customer *Customer, actor string) (int, error) {
|
||
if actor == "" {
|
||
actor = "system"
|
||
}
|
||
affected := []string{}
|
||
seen := map[string]bool{}
|
||
for _, nc := range d.NodeClients {
|
||
if nc.CustomerID == customer.ID && nc.Status == "active" {
|
||
nc.Status = "suspended"
|
||
nc.UpdatedAt = NowISO()
|
||
if !seen[nc.NodeID] {
|
||
seen[nc.NodeID] = true
|
||
affected = append(affected, nc.NodeID)
|
||
}
|
||
}
|
||
}
|
||
for _, nodeID := range affected {
|
||
node := orchFindNode(d, nodeID)
|
||
if node != nil && node.Status != "draft" && node.Status != "removing" {
|
||
markRebuildPending(node, false)
|
||
}
|
||
}
|
||
Audit(d, actor, "suspend_customer_resources", customer.ID, map[string]any{"nodes": len(affected)})
|
||
return len(affected), nil
|
||
}
|
||
|
||
func ResumeCustomerResources(d *Data, customer *Customer, actor string) (int, error) {
|
||
if actor == "" {
|
||
actor = "admin"
|
||
}
|
||
affected := []string{}
|
||
seen := map[string]bool{}
|
||
for _, nc := range d.NodeClients {
|
||
if nc.CustomerID == customer.ID && nc.Status == "suspended" {
|
||
nc.Status = "active"
|
||
nc.UpdatedAt = NowISO()
|
||
if !seen[nc.NodeID] {
|
||
seen[nc.NodeID] = true
|
||
affected = append(affected, nc.NodeID)
|
||
}
|
||
}
|
||
}
|
||
for _, nodeID := range affected {
|
||
node := orchFindNode(d, nodeID)
|
||
if node == nil {
|
||
continue
|
||
}
|
||
if node.Status == "removing" && !boolVal(node.AutoRemoving) {
|
||
continue
|
||
}
|
||
if node.Status == "draft" && !boolVal(node.AutoRemoving) {
|
||
continue
|
||
}
|
||
if node.Status == "removing" && boolVal(node.AutoRemoving) {
|
||
deleteRunning := false
|
||
for _, job := range d.Jobs {
|
||
if job.NodeID == nodeID && job.Action == "delete_resource" && job.Status == "running" {
|
||
deleteRunning = true
|
||
break
|
||
}
|
||
}
|
||
cancelQueuedDeletes(d, nodeID, "客户恢复，清理任务已取消")
|
||
node.AutoRemoving = BoolPtr(false)
|
||
if !deleteRunning {
|
||
DeriveNodeStatus(node)
|
||
node.UpdatedAt = NowISO()
|
||
continue
|
||
}
|
||
}
|
||
includeExit := forwardNode(node) && !node.ExitReady
|
||
markRebuildPending(node, includeExit)
|
||
}
|
||
Audit(d, actor, "resume_customer_resources", customer.ID, map[string]any{"nodes": len(affected)})
|
||
return len(affected), nil
|
||
}
|
||
|
||
// ---------- usage delta accounting ----------
|
||
|
||
type DeltaResult struct {
|
||
UpDelta int64
|
||
DownDelta int64
|
||
}
|
||
|
||
// UsageHolder is the small interface ApplyDelta operates on; *UsageSlot and
|
||
// *Node are adapted onto it by SlotHolder / NodeHolder.
|
||
type UsageHolder interface {
|
||
GetCounters() *UsageCounters
|
||
SetCounters(c *UsageCounters)
|
||
GetRetired() []string
|
||
SetRetired(v []string)
|
||
}
|
||
|
||
type slotUsageHolder struct{ slot *UsageSlot }
|
||
|
||
func (h slotUsageHolder) GetCounters() *UsageCounters { return h.slot.UsageCounters }
|
||
func (h slotUsageHolder) SetCounters(c *UsageCounters) { h.slot.UsageCounters = c }
|
||
func (h slotUsageHolder) GetRetired() []string { return h.slot.RetiredUsageEpochs }
|
||
func (h slotUsageHolder) SetRetired(v []string) { h.slot.RetiredUsageEpochs = v }
|
||
|
||
type nodeUsageHolder struct{ node *Node }
|
||
|
||
func (h nodeUsageHolder) GetCounters() *UsageCounters { return h.node.UsageCounters }
|
||
func (h nodeUsageHolder) SetCounters(c *UsageCounters) { h.node.UsageCounters = c }
|
||
func (h nodeUsageHolder) GetRetired() []string { return h.node.RetiredUsageEpochs }
|
||
func (h nodeUsageHolder) SetRetired(v []string) { h.node.RetiredUsageEpochs = v }
|
||
|
||
func SlotHolder(s *UsageSlot) UsageHolder { return slotUsageHolder{slot: s} }
|
||
func NodeHolder(n *Node) UsageHolder { return nodeUsageHolder{node: n} }
|
||
|
||
// ApplyDelta mirrors applyDelta; a nil result means the sample is skipped.
|
||
func ApplyDelta(holder UsageHolder, uplink, downlink int64, sampleEpoch *string) *DeltaResult {
|
||
if sampleEpoch != nil && *sampleEpoch == "" {
|
||
sampleEpoch = nil
|
||
}
|
||
previous := holder.GetCounters()
|
||
if previous == nil {
|
||
previous = &UsageCounters{}
|
||
}
|
||
retired := holder.GetRetired()
|
||
if sampleEpoch != nil {
|
||
for _, e := range retired {
|
||
if e == *sampleEpoch {
|
||
return nil
|
||
}
|
||
}
|
||
}
|
||
restarted := sampleEpoch != nil && previous.Epoch != nil && *sampleEpoch != *previous.Epoch
|
||
if !restarted && sampleEpoch != nil && previous.Epoch != nil &&
|
||
!strings.HasSuffix(*sampleEpoch, ":unknown") &&
|
||
(uplink < previous.Uplink || downlink < previous.Downlink) {
|
||
return nil
|
||
}
|
||
var upDelta, downDelta int64
|
||
if !restarted && uplink >= previous.Uplink {
|
||
upDelta = uplink - previous.Uplink
|
||
} else {
|
||
upDelta = uplink
|
||
}
|
||
if !restarted && downlink >= previous.Downlink {
|
||
downDelta = downlink - previous.Downlink
|
||
} else {
|
||
downDelta = downlink
|
||
}
|
||
if restarted && previous.Epoch != nil {
|
||
next := append(append([]string{}, retired...), *previous.Epoch)
|
||
if len(next) > 8 {
|
||
next = next[len(next)-8:]
|
||
}
|
||
holder.SetRetired(next)
|
||
}
|
||
holder.SetCounters(&UsageCounters{Uplink: uplink, Downlink: downlink, Epoch: sampleEpoch})
|
||
return &DeltaResult{UpDelta: upDelta, DownDelta: downDelta}
|
||
}
|
||
|
||
// ---------- Reality key backfill ----------
|
||
|
||
func MergeKeyMaterialBackfill(d *Data, node *Node, artifacts map[string]json.RawMessage) {
|
||
if artifacts == nil {
|
||
return
|
||
}
|
||
reportedRaw, hasReported := artifacts["keyMaterial"]
|
||
delete(artifacts, "keyMaterial")
|
||
if hasReported && len(reportedRaw) > 0 {
|
||
var reported map[string]any
|
||
if err := unmarshalJSONUseNumber(reportedRaw, &reported); err == nil && reported != nil {
|
||
if d.KeyStore == nil {
|
||
d.KeyStore = map[string]*KeyEntry{}
|
||
}
|
||
for keyID, pairAny := range reported {
|
||
pair, isObj := pairAny.(map[string]any)
|
||
if !isObj {
|
||
continue
|
||
}
|
||
priv, privOK := pair["privateKey"].(string)
|
||
pub, pubOK := pair["publicKey"].(string)
|
||
if !privOK || !pubOK {
|
||
continue
|
||
}
|
||
existing := d.KeyStore[keyID]
|
||
if existing == nil {
|
||
d.KeyStore[keyID] = &KeyEntry{PrivateKey: priv, PublicKey: pub, UpdatedAt: NowISO()}
|
||
} else if existing.PrivateKey != priv || existing.PublicKey != pub {
|
||
Audit(d, "agent", "reality_key_conflict", node.ID, map[string]any{"keyId": keyID})
|
||
}
|
||
}
|
||
}
|
||
}
|
||
keyID := ""
|
||
if node.Reality != nil {
|
||
keyID = node.Reality.KeyID
|
||
}
|
||
if keyID != "" && d.KeyStore != nil {
|
||
if entry, ok := d.KeyStore[keyID]; ok && entry != nil {
|
||
raw, err := marshalJSON(entry.PublicKey)
|
||
if err == nil {
|
||
artifacts["realityPublicKey"] = raw
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// ---------- deploy validation ----------
|
||
|
||
// orchJSNumberStr mirrors JS Number(string): "" is 0, unparsable is NaN.
|
||
func orchJSNumberStr(s string) float64 {
|
||
t := strings.TrimSpace(s)
|
||
if t == "" {
|
||
return 0
|
||
}
|
||
f, err := strconv.ParseFloat(t, 64)
|
||
if err != nil {
|
||
return math.NaN()
|
||
}
|
||
return f
|
||
}
|
||
|
||
func versionPart(parts []string, i int) float64 {
|
||
if i >= len(parts) {
|
||
return math.NaN()
|
||
}
|
||
return orchJSNumberStr(parts[i])
|
||
}
|
||
|
||
func ValidateNodeDeploy(d *Data, node *Node) error {
|
||
direct := !forwardNode(node)
|
||
if direct {
|
||
if herr := ValidateEntryProtocol(node.Protocol); herr != nil {
|
||
return herr
|
||
}
|
||
} else {
|
||
if herr := ValidateProtocolPair(node.Protocol, xbDeref(node.ExitProtocol)); herr != nil {
|
||
return herr
|
||
}
|
||
}
|
||
server := orchFindServer(d, node.ServerID)
|
||
if server == nil {
|
||
return orchNotFound("服务器")
|
||
}
|
||
var exitServer *Server
|
||
if !direct && !manualExit(node) {
|
||
exitServer = orchFindServer(d, xbDeref(node.ExitServerID))
|
||
if exitServer == nil {
|
||
return orchNotFound("出口服务器")
|
||
}
|
||
}
|
||
targets := []*Server{server}
|
||
if exitServer != nil {
|
||
targets = append(targets, exitServer)
|
||
}
|
||
for _, target := range targets {
|
||
if target.Enabled != nil && !*target.Enabled {
|
||
return httpErr(409, "设备 %s 已停用", target.Name)
|
||
}
|
||
}
|
||
// Round-3 item 20: no client-count gate. Nodes are built and deployed
|
||
// first and bound later (the 3x-ui model); bindings may go to zero
|
||
// and back without tearing the node down. A zero-client inbound
|
||
// simply has no users who can connect — including socks5-auth,
|
||
// whose inbound always carries its node-level fixed account
|
||
// (item 21), so it never degenerates into an open proxy.
|
||
switch node.Protocol {
|
||
case "vless-ws-tls", "hysteria2", "anytls":
|
||
if server.TLSDomain == "" {
|
||
return httpErr(400, "%s 需要先填写设备 TLS 域名并在节点上安装证书", node.Protocol)
|
||
}
|
||
certOK := false
|
||
if server.Engine != nil {
|
||
for _, c := range server.Engine.Certificates {
|
||
if c == server.TLSDomain {
|
||
certOK = true
|
||
break
|
||
}
|
||
}
|
||
}
|
||
if !certOK {
|
||
return httpErr(409, "入口证书未由 Agent 确认。在入口机运行 ng-agent update && ng-agent cert，证书申请成功后刷新设备状态")
|
||
}
|
||
}
|
||
if node.Protocol == "anytls" {
|
||
var agent *Agent
|
||
for _, item := range d.Agents {
|
||
if item.ServerID == server.ID && item.Status == "active" {
|
||
agent = item
|
||
break
|
||
}
|
||
}
|
||
version := ""
|
||
if agent != nil {
|
||
version = agent.Version
|
||
}
|
||
parts := strings.Split(version, ".")
|
||
p0, p1, p2 := versionPart(parts, 0), versionPart(parts, 1), versionPart(parts, 2)
|
||
versionOK := p0 > 0 || (p0 == 0 && (p1 > 6 || (p1 == 6 && p2 >= 1)))
|
||
if agent == nil || !versionOK {
|
||
return httpErr(409, "AnyTLS 入口需要新版 Agent：请在 %s 运行 ng-agent update", server.Name)
|
||
}
|
||
singBoxInstalled := server.Engine != nil && server.Engine.SingBoxInstalled
|
||
if !singBoxInstalled {
|
||
return httpErr(409, "AnyTLS 入口缺少 sing-box 及统计组件：请在 %s 运行 ng-agent engine install，完成后等待 Agent 心跳并刷新页面", server.Name)
|
||
}
|
||
}
|
||
servers := []*Server{server}
|
||
if exitServer != nil {
|
||
servers = []*Server{server, exitServer}
|
||
}
|
||
nowMs := time.Now().UnixMilli()
|
||
unavailable := []*Server{}
|
||
for _, s := range servers {
|
||
stale := false
|
||
if s.LastSeenAt != nil {
|
||
if ms, ok := ParseISOMs(*s.LastSeenAt); ok && nowMs-ms > 120000 {
|
||
stale = true
|
||
}
|
||
}
|
||
if s.Status != "online" || stale {
|
||
unavailable = append(unavailable, s)
|
||
}
|
||
}
|
||
if len(unavailable) > 0 {
|
||
parts := make([]string, len(unavailable))
|
||
for i, s := range unavailable {
|
||
parts[i] = fmt.Sprintf("%s（Agent 离线；请在该机器运行 ng-agent doctor）", s.Name)
|
||
}
|
||
return httpErr(409, "目标设备尚未就绪：%s", strings.Join(parts, "、"))
|
||
}
|
||
networkMode := node.NetworkMode
|
||
if networkMode == "" {
|
||
networkMode = "ipv4"
|
||
}
|
||
if networkMode == "ipv6" {
|
||
missing := []*Server{}
|
||
for _, s := range servers {
|
||
if s.PublicAddressV6 == "" {
|
||
missing = append(missing, s)
|
||
}
|
||
}
|
||
if len(missing) > 0 {
|
||
names := make([]string, len(missing))
|
||
for i, s := range missing {
|
||
names[i] = s.Name
|
||
}
|
||
return httpErr(400, "IPv6 模式需要先为这些服务器填写 IPv6 地址：%s", strings.Join(names, "、"))
|
||
}
|
||
}
|
||
if NodePortInUse(d, node.ServerID, node.Port, node.ID) {
|
||
return httpErr(409, "端口 %d 已被该服务器上的其他节点占用", node.Port)
|
||
}
|
||
if !direct && !manualExit(node) && node.ExitPort != nil && NodePortInUse(d, xbDeref(node.ExitServerID), *node.ExitPort, node.ID) {
|
||
return httpErr(409, "出口端口 %d 已被该服务器上的其他节点占用", *node.ExitPort)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ---------- reconciliation ----------
|
||
|
||
type ReconcileOptions struct {
|
||
OnlyNodeIDs map[string]bool
|
||
FailInFlight *bool // nil mirrors the JS default: true
|
||
}
|
||
|
||
func reconcileServerData(d *Data, serverID, actor string, opts ReconcileOptions) (int, error) {
|
||
server := orchFindServer(d, serverID)
|
||
if server == nil {
|
||
return 0, orchNotFound("服务器")
|
||
}
|
||
// A disabled device receives no new deployments, including reconciles.
|
||
if server.Enabled != nil && !*server.Enabled {
|
||
return 0, nil
|
||
}
|
||
failInFlight := true
|
||
if opts.FailInFlight != nil {
|
||
failInFlight = *opts.FailInFlight
|
||
}
|
||
if failInFlight {
|
||
for _, job := range d.Jobs {
|
||
if job.ServerID == serverID && (job.Status == "queued" || job.Status == "running") {
|
||
job.Status = "failed"
|
||
job.Error = StrPtr("Agent 重新注册，旧任务已由对账任务替代")
|
||
job.UpdatedAt = NowISO()
|
||
}
|
||
}
|
||
}
|
||
count := 0
|
||
for _, node := range d.Nodes {
|
||
if node.ServerID != serverID && xbDeref(node.ExitServerID) != serverID {
|
||
continue
|
||
}
|
||
if opts.OnlyNodeIDs != nil && !opts.OnlyNodeIDs[node.ID] {
|
||
continue
|
||
}
|
||
isEntry := node.ServerID == serverID
|
||
isExit := xbDeref(node.ExitServerID) == serverID && forwardNode(node)
|
||
if !isEntry && !isExit {
|
||
continue
|
||
}
|
||
if node.Status == "draft" {
|
||
continue
|
||
}
|
||
if node.Status == "removing" {
|
||
if isEntry {
|
||
QueueJob(d, serverID, node.ID, "delete_resource", &JobPayload{ResourceID: node.ResourceID, Kind: "node"})
|
||
}
|
||
if isExit && node.ExitResourceID != "" {
|
||
QueueJob(d, serverID, node.ID, "delete_resource", &JobPayload{ResourceID: node.ExitResourceID, Kind: "node-exit"})
|
||
}
|
||
} else {
|
||
clients := activeNodeClients(d, node)
|
||
if isEntry {
|
||
resource, err := resourceForNode(d, node, clients)
|
||
if err != nil {
|
||
return count, err
|
||
}
|
||
QueueJob(d, serverID, node.ID, "apply_resource", payloadWithResource(resource, "node", KeyMaterialForNode(d, node)))
|
||
}
|
||
if isExit {
|
||
resource, err := resourceForNodeExit(d, node)
|
||
if err != nil {
|
||
return count, err
|
||
}
|
||
QueueJob(d, serverID, node.ID, "apply_resource", payloadWithResource(resource, "node-exit", nil))
|
||
}
|
||
// The reconcile just shipped the node's current state; a
|
||
// coalesced pending mark for it is satisfied.
|
||
clearRebuildPending(node)
|
||
node.Status = "deploying"
|
||
}
|
||
node.UpdatedAt = NowISO()
|
||
count++
|
||
}
|
||
Audit(d, actor, "reconcile_server", server.ID, map[string]any{"nodes": count})
|
||
return count, nil
|
||
}
|
||
|
||
// ---------- Orchestrator ----------
|
||
|
||
type Orchestrator struct {
|
||
store *Store
|
||
|
||
mu sync.Mutex
|
||
heartbeatReconcileAt map[string]int64
|
||
}
|
||
|
||
// tx adapts the store's (any, error) transaction signature to the
|
||
// error-only closures used by the orchestrator methods.
|
||
func (o *Orchestrator) tx(fn func(d *Data) error) error {
|
||
_, err := o.store.Transaction(func(d *Data) (any, error) { return nil, fn(d) })
|
||
return err
|
||
}
|
||
|
||
func (o *Orchestrator) transient(fn func(d *Data) error) error {
|
||
o.store.Transient(func(d *Data) { _ = fn(d) })
|
||
return nil
|
||
}
|
||
|
||
func NewOrchestrator(store *Store) *Orchestrator {
|
||
return &Orchestrator{store: store, heartbeatReconcileAt: map[string]int64{}}
|
||
}
|
||
|
||
func cloneNodeRecord(n *Node) *Node {
|
||
raw, err := marshalJSON(n)
|
||
if err != nil {
|
||
panic(err)
|
||
}
|
||
var out Node
|
||
if err := unmarshalJSONUseNumber(raw, &out); err != nil {
|
||
panic(err)
|
||
}
|
||
return &out
|
||
}
|
||
|
||
func cloneNodeClientRecord(nc *NodeClient) *NodeClient {
|
||
raw, err := marshalJSON(nc)
|
||
if err != nil {
|
||
panic(err)
|
||
}
|
||
var out NodeClient
|
||
if err := unmarshalJSONUseNumber(raw, &out); err != nil {
|
||
panic(err)
|
||
}
|
||
return &out
|
||
}
|
||
|
||
func cloneJobRecord(j *Job) *Job {
|
||
raw, err := marshalJSON(j)
|
||
if err != nil {
|
||
panic(err)
|
||
}
|
||
var out Job
|
||
if err := unmarshalJSONUseNumber(raw, &out); err != nil {
|
||
panic(err)
|
||
}
|
||
return &out
|
||
}
|
||
|
||
func (o *Orchestrator) DeployNode(nodeID, actor string) (*Node, error) {
|
||
if actor == "" {
|
||
actor = "admin"
|
||
}
|
||
var result *Node
|
||
err := o.tx(func(d *Data) error {
|
||
node := orchFindNode(d, nodeID)
|
||
if node == nil {
|
||
return orchNotFound("节点")
|
||
}
|
||
if err := ValidateNodeDeploy(d, node); err != nil {
|
||
return err
|
||
}
|
||
// Manual deploy is immediate (item 23 constraint 1): it ships the
|
||
// current state now and consumes any coalesced pending mark.
|
||
clearRebuildPending(node)
|
||
if err := RebuildNode(d, node, true); err != nil {
|
||
return err
|
||
}
|
||
Audit(d, actor, "deploy_node", node.ID, map[string]any{"name": node.Name})
|
||
result = cloneNodeRecord(node)
|
||
return nil
|
||
})
|
||
return result, err
|
||
}
|
||
|
||
func (o *Orchestrator) RemoveNode(nodeID, actor string, deleteRecord bool) (bool, error) {
|
||
if actor == "" {
|
||
actor = "admin"
|
||
}
|
||
err := o.tx(func(d *Data) error {
|
||
node := orchFindNode(d, nodeID)
|
||
if node == nil {
|
||
return orchNotFound("节点")
|
||
}
|
||
node.PendingDelete = deleteRecord
|
||
node.AutoRemoving = BoolPtr(false)
|
||
// A pending coalesced apply must never resurrect a node that is
|
||
// being removed (item 23 constraint 2).
|
||
clearRebuildPending(node)
|
||
if node.Status != "draft" {
|
||
QueueJob(d, node.ServerID, node.ID, "delete_resource", &JobPayload{ResourceID: node.ResourceID, Kind: "node"})
|
||
if forwardNode(node) && node.ExitResourceID != "" {
|
||
QueueJob(d, xbDeref(node.ExitServerID), node.ID, "delete_resource", &JobPayload{ResourceID: node.ExitResourceID, Kind: "node-exit"})
|
||
}
|
||
node.Status = "removing"
|
||
node.Error = nil
|
||
} else if deleteRecord {
|
||
DropRealityKey(d, node)
|
||
kept := []*Node{}
|
||
for _, n := range d.Nodes {
|
||
if n.ID != node.ID {
|
||
kept = append(kept, n)
|
||
}
|
||
}
|
||
d.Nodes = kept
|
||
keptClients := []*NodeClient{}
|
||
for _, nc := range d.NodeClients {
|
||
if nc.NodeID != node.ID {
|
||
keptClients = append(keptClients, nc)
|
||
}
|
||
}
|
||
d.NodeClients = keptClients
|
||
}
|
||
node.UpdatedAt = NowISO()
|
||
action := "remove_node"
|
||
if deleteRecord {
|
||
action = "delete_node"
|
||
}
|
||
Audit(d, actor, action, node.ID, map[string]any{"name": node.Name})
|
||
return nil
|
||
})
|
||
return err == nil, err
|
||
}
|
||
|
||
func (o *Orchestrator) RepairNode(nodeID, actor string) (bool, error) {
|
||
if actor == "" {
|
||
actor = "admin"
|
||
}
|
||
err := o.tx(func(d *Data) error {
|
||
node := orchFindNode(d, nodeID)
|
||
if node == nil {
|
||
return orchNotFound("节点")
|
||
}
|
||
if node.Status != "failed" && node.Status != "degraded" {
|
||
return httpErr(409, "节点状态正常，无需修复")
|
||
}
|
||
clearRebuildPending(node)
|
||
if err := RebuildNode(d, node, true); err != nil {
|
||
return err
|
||
}
|
||
Audit(d, actor, "repair_node", node.ID, map[string]any{"name": node.Name})
|
||
return nil
|
||
})
|
||
return err == nil, err
|
||
}
|
||
|
||
// assignCustomerData is the transaction-free core of AssignCustomer,
|
||
// shared with the customer form's node alignment (AlignCustomerNodes).
|
||
func assignCustomerData(d *Data, nodeID, customerID, actor string) (*NodeClient, error) {
|
||
node := orchFindNode(d, nodeID)
|
||
if node == nil {
|
||
return nil, orchNotFound("节点")
|
||
}
|
||
customer := orchFindCustomer(d, customerID)
|
||
if customer == nil {
|
||
return nil, orchNotFound("客户")
|
||
}
|
||
if customer.Status != "active" {
|
||
return nil, httpErr(409, "只能给状态正常的客户分配节点")
|
||
}
|
||
var nc *NodeClient
|
||
for _, entry := range d.NodeClients {
|
||
if entry.NodeID == nodeID && entry.CustomerID == customerID {
|
||
nc = entry
|
||
break
|
||
}
|
||
}
|
||
if nc != nil && nc.Status == "active" {
|
||
return nil, httpErr(409, "该客户已分配到此节点")
|
||
}
|
||
if nc != nil {
|
||
nc.Status = "active"
|
||
nc.UpdatedAt = NowISO()
|
||
} else {
|
||
now := NowISO()
|
||
ensureCustomerCredentials(d, customerID)
|
||
nc = &NodeClient{
|
||
ID: NewID("ncl"),
|
||
NodeID: nodeID,
|
||
CustomerID: customerID,
|
||
Status: "active",
|
||
CreatedAt: now,
|
||
UpdatedAt: now,
|
||
}
|
||
d.NodeClients = append(d.NodeClients, nc)
|
||
}
|
||
if (node.Status == "removing" || node.Status == "draft") && !boolVal(node.AutoRemoving) {
|
||
// Record the assignment only.
|
||
} else {
|
||
if node.Status == "removing" && boolVal(node.AutoRemoving) {
|
||
cancelQueuedDeletes(d, node.ID, "已取消：节点恢复服务")
|
||
node.AutoRemoving = BoolPtr(false)
|
||
}
|
||
markRebuildPending(node, false)
|
||
}
|
||
Audit(d, actor, "assign_customer", node.ID, map[string]any{"customerId": customerID, "customer": customer.Name})
|
||
return nc, nil
|
||
}
|
||
|
||
func unassignCustomerData(d *Data, nodeID, customerID, actor string) error {
|
||
node := orchFindNode(d, nodeID)
|
||
if node == nil {
|
||
return orchNotFound("节点")
|
||
}
|
||
index := -1
|
||
for i, entry := range d.NodeClients {
|
||
if entry.NodeID == nodeID && entry.CustomerID == customerID {
|
||
index = i
|
||
break
|
||
}
|
||
}
|
||
if index < 0 {
|
||
return httpErr(404, "该客户未分配到此节点")
|
||
}
|
||
d.NodeClients = append(d.NodeClients[:index], d.NodeClients[index+1:]...)
|
||
if node.Status != "draft" && node.Status != "removing" {
|
||
markRebuildPending(node, false)
|
||
}
|
||
Audit(d, actor, "unassign_customer", node.ID, map[string]any{"customerId": customerID})
|
||
return nil
|
||
}
|
||
|
||
// AlignCustomerNodes mirrors alignCustomerNodes in lib/orchestrator.js:
|
||
// full-set alignment of one customer's node bindings (customer form's
|
||
// 关联入站 multi-select). Suspended customers get suspended bindings.
|
||
func AlignCustomerNodes(d *Data, customer *Customer, desiredIds []string, actor string) error {
|
||
desired := []string{}
|
||
seen := map[string]bool{}
|
||
for _, nodeID := range desiredIds {
|
||
if nodeID == "" || seen[nodeID] {
|
||
continue
|
||
}
|
||
seen[nodeID] = true
|
||
node := orchFindNode(d, nodeID)
|
||
if node == nil || node.PendingDelete {
|
||
return httpErr(400, "节点不存在")
|
||
}
|
||
desired = append(desired, nodeID)
|
||
}
|
||
rebuild := map[string]bool{}
|
||
kept := []*NodeClient{}
|
||
for _, nc := range d.NodeClients {
|
||
if nc.CustomerID != customer.ID {
|
||
kept = append(kept, nc)
|
||
continue
|
||
}
|
||
if seen[nc.NodeID] {
|
||
kept = append(kept, nc)
|
||
continue
|
||
}
|
||
Audit(d, actor, "unassign_customer", nc.NodeID, map[string]any{"customerId": customer.ID})
|
||
rebuild[nc.NodeID] = true
|
||
}
|
||
d.NodeClients = kept
|
||
for _, nodeID := range desired {
|
||
exists := false
|
||
for _, nc := range d.NodeClients {
|
||
if nc.CustomerID == customer.ID && nc.NodeID == nodeID {
|
||
exists = true
|
||
break
|
||
}
|
||
}
|
||
if exists {
|
||
continue
|
||
}
|
||
node := orchFindNode(d, nodeID)
|
||
status := "suspended"
|
||
if customer.Status == "active" {
|
||
status = "active"
|
||
}
|
||
now := NowISO()
|
||
ensureCustomerCredentials(d, customer.ID)
|
||
d.NodeClients = append(d.NodeClients, &NodeClient{
|
||
ID: NewID("ncl"),
|
||
NodeID: nodeID,
|
||
CustomerID: customer.ID,
|
||
Status: status,
|
||
CreatedAt: now,
|
||
UpdatedAt: now,
|
||
})
|
||
Audit(d, actor, "assign_customer", node.ID, map[string]any{"customerId": customer.ID, "customer": customer.Name})
|
||
if status == "active" {
|
||
rebuild[nodeID] = true
|
||
}
|
||
}
|
||
for nodeID := range rebuild {
|
||
node := orchFindNode(d, nodeID)
|
||
if node == nil {
|
||
continue
|
||
}
|
||
if (node.Status == "removing" || node.Status == "draft") && !boolVal(node.AutoRemoving) {
|
||
continue
|
||
}
|
||
if node.Status == "removing" && boolVal(node.AutoRemoving) {
|
||
cancelQueuedDeletes(d, node.ID, "已取消：节点恢复服务")
|
||
node.AutoRemoving = BoolPtr(false)
|
||
}
|
||
markRebuildPending(node, false)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (o *Orchestrator) AssignCustomer(nodeID, customerID, actor string) (*NodeClient, error) {
|
||
if actor == "" {
|
||
actor = "admin"
|
||
}
|
||
var result *NodeClient
|
||
err := o.tx(func(d *Data) error {
|
||
nc, err := assignCustomerData(d, nodeID, customerID, actor)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
result = cloneNodeClientRecord(nc)
|
||
return nil
|
||
})
|
||
return result, err
|
||
}
|
||
|
||
func (o *Orchestrator) UnassignCustomer(nodeID, customerID, actor string) (bool, error) {
|
||
if actor == "" {
|
||
actor = "admin"
|
||
}
|
||
err := o.tx(func(d *Data) error {
|
||
return unassignCustomerData(d, nodeID, customerID, actor)
|
||
})
|
||
return err == nil, err
|
||
}
|
||
|
||
func (o *Orchestrator) ReconcileServer(serverID, actor string) (int, error) {
|
||
if actor == "" {
|
||
actor = "system"
|
||
}
|
||
var count int
|
||
err := o.tx(func(d *Data) error {
|
||
var err error
|
||
count, err = reconcileServerData(d, serverID, actor, ReconcileOptions{})
|
||
return err
|
||
})
|
||
return count, err
|
||
}
|
||
|
||
func (o *Orchestrator) ReconcileFromHeartbeat(serverID string, reportedResourceIds []string) (int, error) {
|
||
if reportedResourceIds == nil {
|
||
return 0, nil
|
||
}
|
||
o.mu.Lock()
|
||
last := o.heartbeatReconcileAt[serverID]
|
||
o.mu.Unlock()
|
||
if time.Now().UnixMilli()-last < HeartbeatReconcileIntervalMs {
|
||
return 0, nil
|
||
}
|
||
reported := map[string]bool{}
|
||
limit := len(reportedResourceIds)
|
||
if limit > 1000 {
|
||
limit = 1000
|
||
}
|
||
for _, v := range reportedResourceIds[:limit] {
|
||
if v != "" {
|
||
reported[v] = true
|
||
}
|
||
}
|
||
triggered := false
|
||
var count int
|
||
err := o.tx(func(d *Data) error {
|
||
server := orchFindServer(d, serverID)
|
||
if server == nil {
|
||
return nil
|
||
}
|
||
missing := map[string]bool{}
|
||
for _, node := range d.Nodes {
|
||
if node.Status == "draft" || node.Status == "removing" {
|
||
continue
|
||
}
|
||
if node.ServerID == serverID && node.ResourceID != "" && !reported[node.ResourceID] {
|
||
missing[node.ID] = true
|
||
}
|
||
if xbDeref(node.ExitServerID) == serverID && forwardNode(node) && node.ExitResourceID != "" && !reported[node.ExitResourceID] {
|
||
missing[node.ID] = true
|
||
}
|
||
}
|
||
if len(missing) == 0 {
|
||
return nil
|
||
}
|
||
for _, job := range d.Jobs {
|
||
if job.ServerID == serverID && (job.Status == "queued" || job.Status == "running") {
|
||
return nil
|
||
}
|
||
}
|
||
triggered = true
|
||
fail := false
|
||
var err error
|
||
count, err = reconcileServerData(d, serverID, "system:heartbeat", ReconcileOptions{OnlyNodeIDs: missing, FailInFlight: &fail})
|
||
return err
|
||
})
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
if triggered {
|
||
o.mu.Lock()
|
||
o.heartbeatReconcileAt[serverID] = time.Now().UnixMilli()
|
||
o.mu.Unlock()
|
||
}
|
||
return count, nil
|
||
}
|
||
|
||
type CompleteJobResult struct {
|
||
Artifacts map[string]json.RawMessage
|
||
}
|
||
|
||
func (o *Orchestrator) CompleteJob(jobID string, success bool, result CompleteJobResult, errorMessage *string) (*Job, error) {
|
||
var out *Job
|
||
err := o.tx(func(d *Data) error {
|
||
var job *Job
|
||
for _, item := range d.Jobs {
|
||
if item.ID == jobID {
|
||
job = item
|
||
break
|
||
}
|
||
}
|
||
if job == nil {
|
||
return orchNotFound("任务")
|
||
}
|
||
if job.Status != "running" {
|
||
if job.Status == "completed" && success {
|
||
out = cloneJobRecord(job)
|
||
return nil
|
||
}
|
||
return httpErr(409, "任务已被重试或取消，请等待当前任务完成")
|
||
}
|
||
now := NowISO()
|
||
if success {
|
||
job.Status = "completed"
|
||
} else {
|
||
job.Status = "failed"
|
||
}
|
||
job.UpdatedAt = now
|
||
job.CompletedAt = StrPtr(now)
|
||
job.LeaseUntil = nil
|
||
if success {
|
||
job.Error = nil
|
||
} else {
|
||
msg := "Agent operation failed"
|
||
if errorMessage != nil {
|
||
msg = *errorMessage
|
||
}
|
||
job.Error = StrPtr(xbTruncRunes(RedactSecrets(msg), 1800))
|
||
}
|
||
node := orchFindNode(d, job.NodeID)
|
||
if node == nil {
|
||
out = cloneJobRecord(job)
|
||
return nil
|
||
}
|
||
kind := job.Payload.KindOf()
|
||
switch {
|
||
case success && job.Action == "apply_resource":
|
||
if kind == "node-exit" {
|
||
node.ExitReady = true
|
||
} else {
|
||
node.EntryReady = true
|
||
artifacts := result.Artifacts
|
||
if artifacts == nil {
|
||
artifacts = map[string]json.RawMessage{}
|
||
}
|
||
node.Artifacts = artifacts
|
||
MergeKeyMaterialBackfill(d, node, node.Artifacts)
|
||
}
|
||
DeriveNodeStatus(node)
|
||
case success && job.Action == "delete_resource":
|
||
if kind == "node-exit" {
|
||
node.ExitReady = false
|
||
} else {
|
||
node.EntryReady = false
|
||
}
|
||
pending := false
|
||
for _, item := range d.Jobs {
|
||
if item.NodeID == node.ID && item.Action == "delete_resource" && (item.Status == "queued" || item.Status == "running") {
|
||
pending = true
|
||
break
|
||
}
|
||
}
|
||
if !pending && node.Status == "removing" {
|
||
if node.PendingDelete {
|
||
DropRealityKey(d, node)
|
||
kept := []*Node{}
|
||
for _, n := range d.Nodes {
|
||
if n.ID != node.ID {
|
||
kept = append(kept, n)
|
||
}
|
||
}
|
||
d.Nodes = kept
|
||
keptClients := []*NodeClient{}
|
||
for _, nc := range d.NodeClients {
|
||
if nc.NodeID != node.ID {
|
||
keptClients = append(keptClients, nc)
|
||
}
|
||
}
|
||
d.NodeClients = keptClients
|
||
} else {
|
||
node.Status = "draft"
|
||
node.Error = nil
|
||
}
|
||
}
|
||
case !success:
|
||
if kind == "node-exit" {
|
||
node.Status = "degraded"
|
||
} else {
|
||
node.Status = "failed"
|
||
}
|
||
node.Error = job.Error
|
||
}
|
||
node.UpdatedAt = NowISO()
|
||
out = cloneJobRecord(job)
|
||
return nil
|
||
})
|
||
return out, err
|
||
}
|
||
|
||
// ---------- usage reporting ----------
|
||
|
||
type UsageSample struct {
|
||
Kind string `json:"kind"`
|
||
ID string `json:"id"`
|
||
Uplink json.Number `json:"uplink"`
|
||
Downlink json.Number `json:"downlink"`
|
||
Epoch *string `json:"epoch"`
|
||
}
|
||
|
||
// safeJSInt mirrors Number.isSafeInteger(v) && v >= 0 over a JSON number.
|
||
func safeJSInt(n json.Number) (int64, bool) {
|
||
if n == "" {
|
||
return 0, false
|
||
}
|
||
f, err := n.Float64()
|
||
if err != nil {
|
||
return 0, false
|
||
}
|
||
if f != math.Trunc(f) || f < 0 || math.Abs(f) > 9007199254740991 {
|
||
return 0, false
|
||
}
|
||
return int64(f), true
|
||
}
|
||
|
||
func usageErrorPtr(errText string) *string {
|
||
if errText == "" {
|
||
return nil
|
||
}
|
||
return StrPtr(xbTruncRunes(errText, 240))
|
||
}
|
||
|
||
func (o *Orchestrator) RecordUsage(serverID string, samples []UsageSample, errText string, epoch string) error {
|
||
if len(samples) == 0 {
|
||
return o.transient(func(d *Data) error {
|
||
server := orchFindServer(d, serverID)
|
||
if server != nil {
|
||
server.Usage = &ServerUsage{LastReportAt: NowISO(), SampleCount: 0, Error: usageErrorPtr(errText)}
|
||
}
|
||
return nil
|
||
})
|
||
}
|
||
return o.tx(func(d *Data) error {
|
||
reportedAt := NowISO()
|
||
server := orchFindServer(d, serverID)
|
||
if server != nil {
|
||
server.Usage = &ServerUsage{LastReportAt: reportedAt, SampleCount: len(samples), Error: usageErrorPtr(errText)}
|
||
}
|
||
limit := len(samples)
|
||
if limit > 1000 {
|
||
limit = 1000
|
||
}
|
||
for _, sample := range samples[:limit] {
|
||
uplink, okUp := safeJSInt(sample.Uplink)
|
||
downlink, okDown := safeJSInt(sample.Downlink)
|
||
if !okUp || !okDown {
|
||
continue
|
||
}
|
||
var sampleEpoch *string
|
||
if sample.Epoch != nil {
|
||
sampleEpoch = StrPtr(xbTruncRunes(*sample.Epoch, 80))
|
||
} else if epoch != "" {
|
||
sampleEpoch = StrPtr(epoch)
|
||
}
|
||
switch sample.Kind {
|
||
case "user":
|
||
customer := orchFindCustomer(d, sample.ID)
|
||
if customer == nil {
|
||
continue
|
||
}
|
||
key := "srv:" + serverID
|
||
if customer.ServerUsage == nil {
|
||
customer.ServerUsage = map[string]*UsageSlot{}
|
||
}
|
||
slot := customer.ServerUsage[key]
|
||
if slot == nil {
|
||
slot = &UsageSlot{}
|
||
customer.ServerUsage[key] = slot
|
||
}
|
||
prevCounters := slot.UsageCounters
|
||
prevSampleISO := slot.LastSampleAt
|
||
restarted := prevCounters != nil && prevCounters.Epoch != nil && sampleEpoch != nil && *sampleEpoch != *prevCounters.Epoch
|
||
delta := ApplyDelta(SlotHolder(slot), uplink, downlink, sampleEpoch)
|
||
if delta == nil {
|
||
continue
|
||
}
|
||
customer.UsedUplinkBytes += delta.UpDelta
|
||
customer.UsedDownlinkBytes += delta.DownDelta
|
||
customer.UsedBytes += delta.UpDelta + delta.DownDelta
|
||
customer.LastUsageAt = StrPtr(reportedAt)
|
||
customer.UpdatedAt = reportedAt
|
||
slot.LastSampleAt = StrPtr(reportedAt)
|
||
if delta.UpDelta > 0 || delta.DownDelta > 0 {
|
||
slot.LastActiveAt = StrPtr(reportedAt)
|
||
customer.LastOnlineAt = StrPtr(reportedAt)
|
||
}
|
||
if !restarted && prevCounters != nil && prevSampleISO != nil {
|
||
if prevMs, ok1 := ParseISOMs(*prevSampleISO); ok1 {
|
||
if nowMs, ok2 := ParseISOMs(reportedAt); ok2 {
|
||
dt := float64(nowMs-prevMs) / 1000
|
||
if dt >= 5 {
|
||
up := float64(int64(math.Round(float64(uplink-prevCounters.Uplink) / dt)))
|
||
down := float64(int64(math.Round(float64(downlink-prevCounters.Downlink) / dt)))
|
||
slot.SpeedUpBps = &up
|
||
slot.SpeedDownBps = &down
|
||
slot.SpeedAt = StrPtr(reportedAt)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
case "inbound":
|
||
node := orchFindNode(d, sample.ID)
|
||
if node == nil || node.ServerID != serverID {
|
||
continue
|
||
}
|
||
prevCounters := node.UsageCounters
|
||
prevSampleISO := node.LastSampleAt
|
||
restarted := prevCounters != nil && prevCounters.Epoch != nil && sampleEpoch != nil && *sampleEpoch != *prevCounters.Epoch
|
||
delta := ApplyDelta(NodeHolder(node), uplink, downlink, sampleEpoch)
|
||
if delta == nil {
|
||
continue
|
||
}
|
||
metered := node.MeteredTraffic
|
||
var mu, md int64
|
||
if metered != nil {
|
||
mu, md = metered.Uplink, metered.Downlink
|
||
}
|
||
node.MeteredTraffic = &MeteredTraffic{Uplink: mu + delta.UpDelta, Downlink: md + delta.DownDelta}
|
||
node.LastUsageAt = StrPtr(reportedAt)
|
||
node.LastSampleAt = StrPtr(reportedAt)
|
||
if delta.UpDelta > 0 || delta.DownDelta > 0 {
|
||
node.LastActiveAt = StrPtr(reportedAt)
|
||
}
|
||
if !restarted && prevCounters != nil && prevSampleISO != nil {
|
||
if prevMs, ok1 := ParseISOMs(*prevSampleISO); ok1 {
|
||
if nowMs, ok2 := ParseISOMs(reportedAt); ok2 {
|
||
dt := float64(nowMs-prevMs) / 1000
|
||
if dt >= 5 {
|
||
up := float64(int64(math.Round(float64(uplink-prevCounters.Uplink) / dt)))
|
||
down := float64(int64(math.Round(float64(downlink-prevCounters.Downlink) / dt)))
|
||
node.SpeedUpBps = &up
|
||
node.SpeedDownBps = &down
|
||
node.SpeedAt = StrPtr(reportedAt)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
})
|
||
}
|