From 175d6020452e2bbbae6544d6c4e302b9bf117995 Mon Sep 17 00:00:00 2001 From: jinlong Date: Fri, 7 Aug 2026 15:23:17 +0800 Subject: [PATCH] feat(control-plane): expose node isolation in CubeOps/WebUI via Redis Stream - Add controlevents package: publish node isolation events to a shared Redis Stream so every CubeMaster replica converges in-memory views within seconds after a DB write. MySQL remains the source of truth; nodemeta.loopReload covers missed events. - Wire CubeMaster main.go: controlevents.Init is a fatal dependency in coreInit (Redis unreachable => CubeMaster refuses to start). - Extend nodemeta/isolation.go: append a best-effort event publish at the tail of the DB-write function; add a symmetric ApplySchedulingDisabledLocal for consumer callbacks (mutually exclusive via lockNodeLabels). - Register cube:v1:shared:master:control:events (Stream, shared scope, XADD MAXLEN ~ 10000) in rediskey + redis-key-spec.md. - CubeOps: add PUT/DELETE /api/v1/nodes/{nodeID}/isolation (thin pass-through to CubeMaster REST, no request body, idempotent); expose schedulingDisabled on every nodeView; extend CubeMasterClient interface and cubemaster.Client with IsolateNode/UnisolateNode (+ PUT helper). - Web: introduce IsolateConfirmDialog (Portal), add Isolate / Unisolate actions on Nodes list (Radix dropdown menu) and NodeDetail header, show an Isolated badge, format condition badges via new utils.ts helpers. Bilingual locales extended. - Docs: extend node-isolation.md (4 entry points, DB+Stream mechanism, multi-replica scope, 60s wait window clarified as create-RPC pipeline window). Signed-off-by: jinlong --- CubeMaster/cmd/cubemaster/app/main.go | 8 + CubeMaster/pkg/base/rediskey/rediskey.go | 7 + CubeMaster/pkg/controlevents/apply.go | 60 +++++ CubeMaster/pkg/controlevents/consumer.go | 174 ++++++++++++++ .../pkg/controlevents/controlevents_test.go | 218 ++++++++++++++++++ CubeMaster/pkg/controlevents/init.go | 50 ++++ CubeMaster/pkg/controlevents/publisher.go | 100 ++++++++ CubeMaster/pkg/controlevents/schema.go | 56 +++++ CubeMaster/pkg/nodemeta/isolation.go | 43 ++++ CubeOps/README.md | 2 + CubeOps/internal/cubemaster/client.go | 36 +++ CubeOps/internal/handler/cluster.go | 47 ++++ CubeOps/internal/handler/cluster_test.go | 74 +++++- CubeOps/internal/handler/cmiface.go | 2 + .../internal/handler/fake_cm_handler_test.go | 6 + CubeOps/internal/handler/fake_cm_test.go | 14 ++ docs/dev/redis-key-spec.md | 15 ++ docs/guide/node-isolation.md | 37 ++- docs/zh/dev/redis-key-spec.md | 15 ++ docs/zh/guide/node-isolation.md | 37 ++- web/src/api/client.ts | 11 + web/src/api/generated/schema.ts | 4 + .../components/nodes/IsolateConfirmDialog.tsx | 72 ++++++ web/src/lib/utils.ts | 24 ++ web/src/locales/en/nodeDetail.json | 14 ++ web/src/locales/en/nodes.json | 3 +- web/src/locales/zh/nodeDetail.json | 14 ++ web/src/locales/zh/nodes.json | 3 +- web/src/pages/NodeDetail.tsx | 124 ++++++++-- web/src/pages/Nodes.tsx | 125 +++++++++- 30 files changed, 1345 insertions(+), 50 deletions(-) create mode 100644 CubeMaster/pkg/controlevents/apply.go create mode 100644 CubeMaster/pkg/controlevents/consumer.go create mode 100644 CubeMaster/pkg/controlevents/controlevents_test.go create mode 100644 CubeMaster/pkg/controlevents/init.go create mode 100644 CubeMaster/pkg/controlevents/publisher.go create mode 100644 CubeMaster/pkg/controlevents/schema.go create mode 100644 web/src/components/nodes/IsolateConfirmDialog.tsx diff --git a/CubeMaster/cmd/cubemaster/app/main.go b/CubeMaster/cmd/cubemaster/app/main.go index e1c5aba88..95cec8db4 100644 --- a/CubeMaster/cmd/cubemaster/app/main.go +++ b/CubeMaster/cmd/cubemaster/app/main.go @@ -25,6 +25,7 @@ import ( "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/config" "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/log" "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/recov" + "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/controlevents" "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/cubelet/grpcconn" "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/errorcode" "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/instancecache" @@ -180,6 +181,13 @@ func coreInit(ctx context.Context, cfg *config.Config) error { log.G(ctx).Warnf("lifecycle init fail (non-fatal): %v", err) } + // controlevents fans out control-plane mutations (e.g. node isolation) to + // every Cubemaster replica via Redis Stream. Redis is a hard dependency: + // without it multi-replica cordon convergence is not guaranteed. + if err := controlevents.Init(ctx, nodemeta.ApplySchedulingDisabledLocal); err != nil { + return fmt.Errorf("controlevents init: %w", err) + } + scheduler.InitScheduler(ctx) if err := sandbox.Init(ctx, cfg); err != nil { diff --git a/CubeMaster/pkg/base/rediskey/rediskey.go b/CubeMaster/pkg/base/rediskey/rediskey.go index 72497887b..0c787dcf1 100644 --- a/CubeMaster/pkg/base/rediskey/rediskey.go +++ b/CubeMaster/pkg/base/rediskey/rediskey.go @@ -77,6 +77,13 @@ func SandboxLifecycleState(sandboxID string) string { return join(Prefix, Version, ScopeShared, "sandbox", "lifecycle", "state", sandboxID) } +// MasterControlEvents is the append-only control-plane event stream that +// CubeMaster replicas use to fan out mutations that must converge in every +// replica's in-memory view (e.g. node isolation / cordon). +func MasterControlEvents() string { + return join(Prefix, Version, ScopeShared, "master", "control", "events") +} + // ---- legacy key builders (read fallback / delete cleanup only) ---- // LegacyNodeMetric is the bare node ID used before namespacing. diff --git a/CubeMaster/pkg/controlevents/apply.go b/CubeMaster/pkg/controlevents/apply.go new file mode 100644 index 000000000..7dea98a2c --- /dev/null +++ b/CubeMaster/pkg/controlevents/apply.go @@ -0,0 +1,60 @@ +// Copyright (c) 2026 Tencent Inc. +// SPDX-License-Identifier: Apache-2.0 +// + +package controlevents + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/log" +) + +// ApplyFunc updates the local process view for a node cordon change without +// writing MySQL. Provided by the wiring layer (typically nodemeta) to avoid an +// import cycle between controlevents and nodemeta. +type ApplyFunc func(ctx context.Context, nodeID string, disabled bool) error + +// NewIsolationHandler returns a Handler that dispatches isolate/unisolate ops. +func NewIsolationHandler(apply ApplyFunc) Handler { + return func(ctx context.Context, ev Event) error { + return applyIsolationEvent(ctx, ev, apply) + } +} + +func applyIsolationEvent(ctx context.Context, ev Event, apply ApplyFunc) error { + if apply == nil { + return nil + } + if ev.NodeID == "" { + return fmt.Errorf("missing node_id") + } + + var disabled bool + switch ev.Op { + case OpNodeIsolate: + disabled = true + case OpNodeUnisolate: + disabled = false + default: + log.G(ctx).Debugf("controlevents: ignoring unknown op=%s", ev.Op) + return nil + } + + if len(ev.Payload) > 0 { + var p IsolationPayload + if err := json.Unmarshal(ev.Payload, &p); err != nil { + log.G(ctx).Warnf("controlevents: bad payload op=%s node=%s: %v", ev.Op, ev.NodeID, err) + // Fall through with op-derived disabled; payload is advisory. + } else if p.SchedulingDisabled != disabled { + // Prefer explicit payload when present and consistent with intent; + // if inconsistent, trust the op (stream field is authoritative). + log.G(ctx).Warnf("controlevents: payload scheduling_disabled=%v disagrees with op=%s node=%s; using op", + p.SchedulingDisabled, ev.Op, ev.NodeID) + } + } + + return apply(ctx, ev.NodeID, disabled) +} diff --git a/CubeMaster/pkg/controlevents/consumer.go b/CubeMaster/pkg/controlevents/consumer.go new file mode 100644 index 000000000..79657e24d --- /dev/null +++ b/CubeMaster/pkg/controlevents/consumer.go @@ -0,0 +1,174 @@ +// Copyright (c) 2026 Tencent Inc. +// SPDX-License-Identifier: Apache-2.0 +// + +package controlevents + +import ( + "context" + "fmt" + "time" + + "github.com/gomodule/redigo/redis" + "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/log" + "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/recov" +) + +const ( + defaultBlockMs = 5000 + defaultCount = 32 +) + +// Handler applies a decoded control-plane event to the local process view. +// Implementations must be idempotent and must not write MySQL. +type Handler func(ctx context.Context, ev Event) error + +// Consumer independently XREADs the control stream starting from "$" so every +// Cubemaster replica observes every new event (broadcast). Missed events are +// recovered via nodemeta's periodic DB reload. +type Consumer struct { + pool *redis.Pool + handler Handler + blockMs int + count int +} + +// NewConsumer builds a Consumer. pool must be non-nil; handler may be nil +// (events are then logged and dropped). +func NewConsumer(pool *redis.Pool, handler Handler) *Consumer { + return &Consumer{ + pool: pool, + handler: handler, + blockMs: defaultBlockMs, + count: defaultCount, + } +} + +// Run blocks until ctx is cancelled. It starts reading at "$" (new events only). +func (c *Consumer) Run(ctx context.Context) { + if c == nil || c.pool == nil { + return + } + lastID := "$" + log.G(ctx).Infof("controlevents: consumer started stream=%s from=%s", EventStreamKey, lastID) + for { + select { + case <-ctx.Done(): + log.G(ctx).Infof("controlevents: consumer stopped: %v", ctx.Err()) + return + default: + } + + events, nextID, err := c.readOnce(ctx, lastID) + if err != nil { + log.G(ctx).Warnf("controlevents: XREAD failed: %v", err) + select { + case <-ctx.Done(): + return + case <-time.After(time.Second): + } + continue + } + if nextID != "" { + lastID = nextID + } + for _, ev := range events { + c.dispatch(ctx, ev) + } + } +} + +func (c *Consumer) dispatch(ctx context.Context, ev Event) { + defer recov.HandleCrash(func(r interface{}) { + log.G(ctx).Errorf("controlevents: handler panic op=%s node=%s: %v", ev.Op, ev.NodeID, r) + }) + if c.handler == nil { + return + } + if err := c.handler(ctx, ev); err != nil { + log.G(ctx).Warnf("controlevents: handler failed op=%s node=%s: %v", ev.Op, ev.NodeID, err) + } +} + +func (c *Consumer) readOnce(ctx context.Context, lastID string) ([]Event, string, error) { + conn := c.pool.Get() + defer conn.Close() + if err := conn.Err(); err != nil { + return nil, "", err + } + + reply, err := conn.Do("XREAD", + "BLOCK", c.blockMs, + "COUNT", c.count, + "STREAMS", EventStreamKey, lastID, + ) + if err == redis.ErrNil || reply == nil { + return nil, "", nil + } + if err != nil { + return nil, "", err + } + + events, nextID, err := parseXReadReply(reply) + if err != nil { + return nil, "", err + } + _ = ctx + return events, nextID, nil +} + +// parseXReadReply decodes a redigo XREAD reply into events and the last stream ID. +// Reply shape: [[streamName, [[id, [k, v, ...]], ...]]] +func parseXReadReply(reply interface{}) ([]Event, string, error) { + streams, err := redis.Values(reply, nil) + if err != nil { + return nil, "", fmt.Errorf("decode streams: %w", err) + } + if len(streams) == 0 { + return nil, "", nil + } + + var out []Event + var lastID string + for _, streamRaw := range streams { + stream, err := redis.Values(streamRaw, nil) + if err != nil || len(stream) < 2 { + continue + } + entries, err := redis.Values(stream[1], nil) + if err != nil { + return nil, "", fmt.Errorf("decode entries: %w", err) + } + for _, entryRaw := range entries { + entry, err := redis.Values(entryRaw, nil) + if err != nil || len(entry) < 2 { + continue + } + id, err := redis.String(entry[0], nil) + if err != nil { + continue + } + fields, err := redis.Values(entry[1], nil) + if err != nil { + continue + } + ev := Event{StreamID: id} + for i := 0; i+1 < len(fields); i += 2 { + k, _ := redis.String(fields[i], nil) + switch k { + case FieldOp: + ev.Op, _ = redis.String(fields[i+1], nil) + case FieldNodeID: + ev.NodeID, _ = redis.String(fields[i+1], nil) + case FieldPayload: + ev.Payload, _ = redis.Bytes(fields[i+1], nil) + case FieldTimestamp: + ev.Timestamp, _ = redis.Int64(fields[i+1], nil) + } + } + out = append(out, ev) + lastID = id + } + } + return out, lastID, nil +} diff --git a/CubeMaster/pkg/controlevents/controlevents_test.go b/CubeMaster/pkg/controlevents/controlevents_test.go new file mode 100644 index 000000000..fe319bb84 --- /dev/null +++ b/CubeMaster/pkg/controlevents/controlevents_test.go @@ -0,0 +1,218 @@ +// Copyright (c) 2026 Tencent Inc. +// SPDX-License-Identifier: Apache-2.0 +// + +package controlevents + +import ( + "context" + "encoding/json" + "errors" + "sync" + "testing" +) + +type recordedCall struct { + cmd string + args []interface{} +} + +type fakeRedis struct { + mu sync.Mutex + calls []recordedCall + failXADD bool +} + +func (f *fakeRedis) Do(cmd string, args ...interface{}) (interface{}, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, recordedCall{cmd: cmd, args: args}) + if cmd == "XADD" && f.failXADD { + return nil, errors.New("XADD boom") + } + return "OK", nil +} + +func (f *fakeRedis) snapshot() []recordedCall { + f.mu.Lock() + defer f.mu.Unlock() + out := make([]recordedCall, len(f.calls)) + copy(out, f.calls) + return out +} + +func TestPublisher_PublishNodeIsolation_Isolate(t *testing.T) { + r := &fakeRedis{} + p := NewPublisher(r) + p.origin = "master-a" + + p.PublishNodeIsolation(context.Background(), "node-1", true) + + calls := r.snapshot() + if len(calls) != 1 || calls[0].cmd != "XADD" { + t.Fatalf("want 1 XADD, got %+v", calls) + } + args := calls[0].args + if args[0] != EventStreamKey { + t.Fatalf("stream key: got %v", args[0]) + } + if args[5] != FieldOp || args[6] != OpNodeIsolate { + t.Fatalf("op wrong: %+v", args) + } + if args[7] != FieldNodeID || args[8] != "node-1" { + t.Fatalf("node_id wrong: %+v", args) + } + // payload is last pair + var payload []byte + for i := 0; i+1 < len(args); i++ { + if args[i] == FieldPayload { + b, ok := args[i+1].([]byte) + if !ok { + t.Fatalf("payload type %T", args[i+1]) + } + payload = b + } + } + if len(payload) == 0 { + t.Fatal("missing payload") + } + var got IsolationPayload + if err := json.Unmarshal(payload, &got); err != nil { + t.Fatalf("payload json: %v", err) + } + if !got.SchedulingDisabled || got.Origin != "master-a" { + t.Fatalf("payload wrong: %+v", got) + } +} + +func TestPublisher_PublishNodeIsolation_Unisolate(t *testing.T) { + r := &fakeRedis{} + p := NewPublisher(r) + p.PublishNodeIsolation(context.Background(), "node-2", false) + + calls := r.snapshot() + if len(calls) != 1 { + t.Fatalf("want 1 call, got %d", len(calls)) + } + if calls[0].args[6] != OpNodeUnisolate { + t.Fatalf("want unisolate op, got %v", calls[0].args[6]) + } +} + +func TestPublisher_PublishNodeIsolation_XADDFailureSwallowed(t *testing.T) { + r := &fakeRedis{failXADD: true} + p := NewPublisher(r) + // Must not panic. + p.PublishNodeIsolation(context.Background(), "node-1", true) +} + +func TestPublisher_PublishNodeIsolation_EmptyNodeNoop(t *testing.T) { + r := &fakeRedis{} + p := NewPublisher(r) + p.PublishNodeIsolation(context.Background(), "", true) + if len(r.snapshot()) != 0 { + t.Fatal("expected no redis calls for empty node_id") + } +} + +func TestApplyIsolationEvent_IsolateUnisolateIdempotent(t *testing.T) { + var mu sync.Mutex + applied := make([]bool, 0, 4) + + apply := func(_ context.Context, nodeID string, disabled bool) error { + if nodeID != "n1" { + t.Fatalf("node_id=%s", nodeID) + } + mu.Lock() + applied = append(applied, disabled) + mu.Unlock() + return nil + } + h := NewIsolationHandler(apply) + + payload, _ := json.Marshal(IsolationPayload{SchedulingDisabled: true}) + for i := 0; i < 2; i++ { + if err := h(context.Background(), Event{ + Op: OpNodeIsolate, NodeID: "n1", Payload: payload, + }); err != nil { + t.Fatalf("isolate: %v", err) + } + } + payloadOff, _ := json.Marshal(IsolationPayload{SchedulingDisabled: false}) + if err := h(context.Background(), Event{ + Op: OpNodeUnisolate, NodeID: "n1", Payload: payloadOff, + }); err != nil { + t.Fatalf("unisolate: %v", err) + } + + mu.Lock() + defer mu.Unlock() + if len(applied) != 3 || !applied[0] || !applied[1] || applied[2] { + t.Fatalf("applied sequence wrong: %v", applied) + } +} + +func TestApplyIsolationEvent_UnknownOpIgnored(t *testing.T) { + called := false + h := NewIsolationHandler(func(context.Context, string, bool) error { + called = true + return nil + }) + if err := h(context.Background(), Event{Op: "node.labels", NodeID: "n1"}); err != nil { + t.Fatalf("unexpected err: %v", err) + } + if called { + t.Fatal("unknown op should not call apply") + } +} + +func TestApplyIsolationEvent_MissingNodeID(t *testing.T) { + h := NewIsolationHandler(func(context.Context, string, bool) error { return nil }) + if err := h(context.Background(), Event{Op: OpNodeIsolate}); err == nil { + t.Fatal("expected error for missing node_id") + } +} + +func TestParseXReadReply(t *testing.T) { + reply := []interface{}{ + []interface{}{ + []byte(EventStreamKey), + []interface{}{ + []interface{}{ + []byte("1710000000000-0"), + []interface{}{ + []byte(FieldOp), []byte(OpNodeIsolate), + []byte(FieldNodeID), []byte("node-9"), + []byte(FieldTimestamp), []byte("1710000000000"), + []byte(FieldPayload), []byte(`{"scheduling_disabled":true}`), + }, + }, + }, + }, + } + events, lastID, err := parseXReadReply(reply) + if err != nil { + t.Fatalf("parse: %v", err) + } + if lastID != "1710000000000-0" { + t.Fatalf("lastID=%s", lastID) + } + if len(events) != 1 { + t.Fatalf("events=%d", len(events)) + } + ev := events[0] + if ev.Op != OpNodeIsolate || ev.NodeID != "node-9" || ev.Timestamp != 1710000000000 { + t.Fatalf("event wrong: %+v", ev) + } + var p IsolationPayload + if err := json.Unmarshal(ev.Payload, &p); err != nil || !p.SchedulingDisabled { + t.Fatalf("payload wrong: %v %+v", err, p) + } +} + +func TestParseXReadReply_Empty(t *testing.T) { + events, lastID, err := parseXReadReply([]interface{}{}) + if err != nil || len(events) != 0 || lastID != "" { + t.Fatalf("got events=%v lastID=%q err=%v", events, lastID, err) + } +} diff --git a/CubeMaster/pkg/controlevents/init.go b/CubeMaster/pkg/controlevents/init.go new file mode 100644 index 000000000..1122bf323 --- /dev/null +++ b/CubeMaster/pkg/controlevents/init.go @@ -0,0 +1,50 @@ +// Copyright (c) 2026 Tencent Inc. +// SPDX-License-Identifier: Apache-2.0 +// + +package controlevents + +import ( + "context" + "fmt" + + "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/log" + "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/recov" + "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/wrapredis" +) + +// Init verifies Redis connectivity (fatal for the caller on failure), installs +// the package-level publisher, and starts a broadcast consumer goroutine. +// +// Unlike lifecycle.Init, Redis is a hard dependency: Cubemaster must not serve +// without the control-plane fan-out channel. +func Init(ctx context.Context, apply ApplyFunc) error { + pool := wrapredis.GetRedis() + if isNilPool(pool) { + return fmt.Errorf("controlevents: redis pool unavailable") + } + if _, err := pool.Do("PING"); err != nil { + return fmt.Errorf("controlevents: redis ping failed: %w", err) + } + // Confirm the Redis server accepts stream commands. TYPE on a missing key + // returns "none"; any transport/command error is fatal at startup. + if _, err := pool.Do("TYPE", EventStreamKey); err != nil { + return fmt.Errorf("controlevents: redis TYPE %s failed: %w", EventStreamKey, err) + } + + pub := NewPublisher(pool) + setDefaultPublisher(pub) + + handler := NewIsolationHandler(apply) + consumer := NewConsumer(pool.RedisConnPool, handler) + recov.GoWithRecover(func() { + consumer.Run(ctx) + }) + + log.G(ctx).Infof("controlevents: ready (stream=%s)", EventStreamKey) + return nil +} + +func isNilPool(w *wrapredis.RedisWrap) bool { + return w == nil || w.RedisConnPool == nil +} diff --git a/CubeMaster/pkg/controlevents/publisher.go b/CubeMaster/pkg/controlevents/publisher.go new file mode 100644 index 000000000..f206c1621 --- /dev/null +++ b/CubeMaster/pkg/controlevents/publisher.go @@ -0,0 +1,100 @@ +// Copyright (c) 2026 Tencent Inc. +// SPDX-License-Identifier: Apache-2.0 +// + +package controlevents + +import ( + "context" + "encoding/json" + "os" + "strconv" + "sync/atomic" + "time" + + "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/log" +) + +// redisDoer is the minimal redigo-shaped surface the publisher needs. +type redisDoer interface { + Do(cmd string, args ...interface{}) (interface{}, error) +} + +// Publisher performs Redis XADD writes for control-plane events. Runtime +// failures are logged and swallowed so a Redis hiccup cannot roll back a DB +// write that already succeeded. +type Publisher struct { + doer redisDoer + origin string + enabled atomic.Bool +} + +// NewPublisher wires a Publisher onto the supplied redis client. +func NewPublisher(doer redisDoer) *Publisher { + origin, _ := os.Hostname() + p := &Publisher{doer: doer, origin: origin} + p.enabled.Store(true) + return p +} + +// SetEnabled toggles all writes. +func (p *Publisher) SetEnabled(v bool) { + if p == nil { + return + } + p.enabled.Store(v) +} + +// PublishNodeIsolation emits node.isolate or node.unisolate after a successful +// DB cordon write. Best-effort: errors are warned, never returned. +func (p *Publisher) PublishNodeIsolation(ctx context.Context, nodeID string, disabled bool) { + if p == nil || !p.enabled.Load() || p.doer == nil || nodeID == "" { + return + } + op := OpNodeUnisolate + if disabled { + op = OpNodeIsolate + } + payload, err := json.Marshal(IsolationPayload{ + SchedulingDisabled: disabled, + UpdatedAtUnixMs: time.Now().UnixMilli(), + Origin: p.origin, + }) + if err != nil { + log.G(ctx).Warnf("controlevents: marshal isolation payload node=%s: %v", nodeID, err) + return + } + if _, err := p.xadd(op, nodeID, payload); err != nil { + log.G(ctx).Warnf("controlevents: XADD %s node=%s failed: %v", op, nodeID, err) + } +} + +func (p *Publisher) xadd(op, nodeID string, payload []byte) (interface{}, error) { + args := make([]interface{}, 0, 12) + args = append(args, + EventStreamKey, + "MAXLEN", "~", strconv.Itoa(EventStreamMaxLen), + "*", + FieldOp, op, + FieldNodeID, nodeID, + FieldTimestamp, time.Now().UnixMilli(), + ) + if len(payload) > 0 { + args = append(args, FieldPayload, payload) + } + return p.doer.Do("XADD", args...) +} + +var defaultPublisher atomic.Pointer[Publisher] + +func setDefaultPublisher(p *Publisher) { defaultPublisher.Store(p) } + +func getDefaultPublisher() *Publisher { return defaultPublisher.Load() } + +// PublishNodeIsolationDefault is the package-level entry used by nodemeta after +// a successful isolation write. Safe when Init has not run (no-op). +func PublishNodeIsolationDefault(ctx context.Context, nodeID string, disabled bool) { + if p := getDefaultPublisher(); p != nil { + p.PublishNodeIsolation(ctx, nodeID, disabled) + } +} diff --git a/CubeMaster/pkg/controlevents/schema.go b/CubeMaster/pkg/controlevents/schema.go new file mode 100644 index 000000000..5a85bdc6c --- /dev/null +++ b/CubeMaster/pkg/controlevents/schema.go @@ -0,0 +1,56 @@ +// Copyright (c) 2026 Tencent Inc. +// SPDX-License-Identifier: Apache-2.0 +// + +// Package controlevents owns the Cubemaster↔Cubemaster control-plane event +// channel backed by a Redis Stream. +// +// MySQL remains the source of truth for node metadata. The stream only +// accelerates multi-replica in-memory convergence after a successful DB write. +// Each Cubemaster replica independently XREADs the stream (broadcast); a shared +// consumer group is intentionally NOT used. +// +// Stream: cube:v1:shared:master:control:events +package controlevents + +import "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/rediskey" + +// EventStreamKey is the append-only control-plane event stream. +var EventStreamKey = rediskey.MasterControlEvents() + +const ( + // EventStreamMaxLen caps stream growth. Replicas bootstrap cordon state + // from MySQL (nodemeta reload), so trimmed events are recovered on the + // next full sync. + EventStreamMaxLen = 10000 +) + +// Event op codes carried in stream entries. +const ( + OpNodeIsolate = "node.isolate" + OpNodeUnisolate = "node.unisolate" +) + +// Stream entry field names. +const ( + FieldOp = "op" + FieldNodeID = "node_id" + FieldPayload = "payload" + FieldTimestamp = "ts" +) + +// IsolationPayload is the JSON body for node.isolate / node.unisolate events. +type IsolationPayload struct { + SchedulingDisabled bool `json:"scheduling_disabled"` + UpdatedAtUnixMs int64 `json:"updated_at_unix_ms,omitempty"` + Origin string `json:"origin,omitempty"` +} + +// Event is a decoded control-plane stream entry. +type Event struct { + StreamID string + Op string + NodeID string + Payload []byte + Timestamp int64 +} diff --git a/CubeMaster/pkg/nodemeta/isolation.go b/CubeMaster/pkg/nodemeta/isolation.go index 5669dfbdd..e3161dd85 100644 --- a/CubeMaster/pkg/nodemeta/isolation.go +++ b/CubeMaster/pkg/nodemeta/isolation.go @@ -14,6 +14,7 @@ import ( "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/constants" "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/log" "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/node" + "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/controlevents" "gorm.io/gorm" ) @@ -92,9 +93,51 @@ func SetNodeSchedulingDisabled(ctx context.Context, nodeID string, disabled bool out := cloneSnapshotWithCurrentHealth(snap, time.Now()) log.G(ctx).Infof("node isolation write node_id=%s disabled=%v changed=%v scheduling_disabled=%v", nodeID, disabled, changed, out.SchedulingDisabled) + + // Fan out to all Cubemaster replicas via Redis Stream (best-effort). + controlevents.PublishNodeIsolationDefault(ctx, nodeID, disabled) return out, nil } +// ApplySchedulingDisabledLocal updates the in-memory cordon view and localcache +// without touching MySQL. Used by controlevents consumers on every replica so +// isolation converges immediately; DB remains the source of truth via reload. +func ApplySchedulingDisabledLocal(ctx context.Context, nodeID string, disabled bool) error { + if nodeID == "" { + return fmt.Errorf("node_id is required") + } + if !Ready() { + return fmt.Errorf("nodemeta not ready") + } + + unlock := global.lockNodeLabels(nodeID) + defer unlock() + + global.mu.Lock() + snap, ok := global.nodes[nodeID] + if !ok { + global.mu.Unlock() + log.G(ctx).Debugf("controlevents apply skipped: unknown node_id=%s", nodeID) + return nil + } + labels := cloneStringMap(snap.Labels) + if labels == nil { + labels = map[string]string{} + } + if disabled { + labels[constants.LabelSchedulingDisabled] = constants.LabelSchedulingDisabledValue + } else { + delete(labels, constants.LabelSchedulingDisabled) + } + snap.Labels = labels + snap.labelsJSONCorrupt = false + global.mu.Unlock() + + syncLocalcache(snap) + log.G(ctx).Infof("node isolation applied from event node_id=%s disabled=%v", nodeID, disabled) + return nil +} + // stripAndPreserveSchedulingLabel merges cubelet labels while keeping the // control-plane cordon key from DB (cubelet cannot create/overwrite/delete it). func stripAndPreserveSchedulingLabel(existing, cubeletLabels map[string]string) map[string]string { diff --git a/CubeOps/README.md b/CubeOps/README.md index 5fbca4f40..9a70c0cc6 100644 --- a/CubeOps/README.md +++ b/CubeOps/README.md @@ -161,6 +161,8 @@ RBAC is reserved for future use — currently any valid JWT grants full access. - `GET /api/v1/cluster/versions` — Component version matrix - `GET /api/v1/nodes` — Node list - `GET /api/v1/nodes/{nodeID}` — Node detail +- `PUT /api/v1/nodes/{nodeID}/isolation` — Isolate (cordon) node +- `DELETE /api/v1/nodes/{nodeID}/isolation` — Unisolate (uncordon) node ### AgentHub - `GET /api/v1/agenthub/instances` — List agent instances diff --git a/CubeOps/internal/cubemaster/client.go b/CubeOps/internal/cubemaster/client.go index 900308222..93ced9b49 100644 --- a/CubeOps/internal/cubemaster/client.go +++ b/CubeOps/internal/cubemaster/client.go @@ -128,6 +128,18 @@ func (c *Client) GetNode(ctx context.Context, nodeID string) (json.RawMessage, e return c.get(ctx, fmt.Sprintf("/internal/meta/nodes/%s", escaped)) } +// IsolateNode cordons a node so CubeMaster stops scheduling new sandboxes onto it. +func (c *Client) IsolateNode(ctx context.Context, nodeID string) (json.RawMessage, error) { + escaped := url.PathEscape(nodeID) + return c.put(ctx, fmt.Sprintf("/internal/meta/nodes/%s/isolation", escaped), nil) +} + +// UnisolateNode removes the cordon so the node can receive new sandboxes again. +func (c *Client) UnisolateNode(ctx context.Context, nodeID string) (json.RawMessage, error) { + escaped := url.PathEscape(nodeID) + return c.delete(ctx, fmt.Sprintf("/internal/meta/nodes/%s/isolation", escaped)) +} + // ListSandboxes fetches the sandbox list from CubeMaster. func (c *Client) ListSandboxes(ctx context.Context) (json.RawMessage, error) { return c.post(ctx, "/cube/sandbox/list", map[string]interface{}{ @@ -322,6 +334,30 @@ func (c *Client) post(ctx context.Context, path string, body interface{}) (json. return readResponse(resp) } +func (c *Client) put(ctx context.Context, path string, body interface{}) (json.RawMessage, error) { + var bodyReader io.Reader + if body != nil { + b, err := json.Marshal(body) + if err != nil { + return nil, fmt.Errorf("marshal body: %w", err) + } + bodyReader = bytes.NewReader(b) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPut, c.baseURL+path, bodyReader) + if err != nil { + return nil, err + } + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + resp, err := c.http.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + return readResponse(resp) +} + func (c *Client) delete(ctx context.Context, path string) (json.RawMessage, error) { req, err := http.NewRequestWithContext(ctx, http.MethodDelete, c.baseURL+path, nil) if err != nil { diff --git a/CubeOps/internal/handler/cluster.go b/CubeOps/internal/handler/cluster.go index 1d4daa725..9b97855d4 100644 --- a/CubeOps/internal/handler/cluster.go +++ b/CubeOps/internal/handler/cluster.go @@ -27,6 +27,8 @@ func (h *ClusterHandler) Register(r *gin.RouterGroup) { r.GET("/cluster/versions", h.Versions) r.GET("/nodes", h.ListNodes) r.GET("/nodes/:nodeID", h.GetNode) + r.PUT("/nodes/:nodeID/isolation", h.IsolateNode) + r.DELETE("/nodes/:nodeID/isolation", h.UnisolateNode) } // --- Response types matching the frontend's expected format --- @@ -57,6 +59,7 @@ type nodeView struct { HostIP string `json:"hostIP"` InstanceType string `json:"instanceType"` Healthy bool `json:"healthy"` + SchedulingDisabled bool `json:"schedulingDisabled"` Capacity nodeResourcesView `json:"capacity"` Allocatable nodeResourcesView `json:"allocatable"` CpuSaturation float32 `json:"cpuSaturation"` @@ -113,6 +116,7 @@ type cmNodeSnapshot struct { HostIP string `json:"host_ip"` InstanceType string `json:"instance_type"` Healthy bool `json:"healthy"` + SchedulingDisabled bool `json:"scheduling_disabled"` Capacity cmNodeResources `json:"capacity"` Allocatable cmNodeResources `json:"allocatable"` MaxMvmNum int `json:"max_mvm_num"` @@ -253,6 +257,48 @@ func (h *ClusterHandler) GetNode(c *gin.Context) { httputil.WriteJSON(c, http.StatusOK, toNodeView(*resp.Data, used)) } +// IsolateNode handles PUT /nodes/{nodeID}/isolation. +func (h *ClusterHandler) IsolateNode(c *gin.Context) { + h.writeIsolation(c, true) +} + +// UnisolateNode handles DELETE /nodes/{nodeID}/isolation. +func (h *ClusterHandler) UnisolateNode(c *gin.Context) { + h.writeIsolation(c, false) +} + +func (h *ClusterHandler) writeIsolation(c *gin.Context, isolate bool) { + nodeID := c.Param("nodeID") + if nodeID == "" { + httputil.WriteError(c, http.StatusBadRequest, "nodeID is required") + return + } + var ( + data json.RawMessage + err error + ) + if isolate { + data, err = h.cm.IsolateNode(c.Request.Context(), nodeID) + } else { + data, err = h.cm.UnisolateNode(c.Request.Context(), nodeID) + } + if err != nil { + writeCMError(c, err) + return + } + var resp cmNodeResponse + if err := json.Unmarshal(data, &resp); err != nil { + httputil.WriteError(c, http.StatusInternalServerError, "failed to parse node response") + return + } + if resp.Data == nil { + httputil.WriteError(c, http.StatusNotFound, fmt.Sprintf("node %s not found", nodeID)) + return + } + used := h.fetchUsedResources(c.Request.Context()) + httputil.WriteJSON(c, http.StatusOK, toNodeView(*resp.Data, used)) +} + // Versions handles GET /cluster/versions. // // Empty/missing CubeMaster data returns an empty shell for the UI. Otherwise @@ -342,6 +388,7 @@ func toNodeView(s cmNodeSnapshot, usedMap map[string]struct { HostIP: s.HostIP, InstanceType: s.InstanceType, Healthy: s.Healthy, + SchedulingDisabled: s.SchedulingDisabled, Capacity: nodeResourcesView{CpuMilli: capCPU, MemoryMB: capMem}, Allocatable: nodeResourcesView{CpuMilli: allocCPU, MemoryMB: allocMem}, CpuSaturation: saturationPct(capCPU, allocCPU), diff --git a/CubeOps/internal/handler/cluster_test.go b/CubeOps/internal/handler/cluster_test.go index 4e6474d9d..0b0e58806 100644 --- a/CubeOps/internal/handler/cluster_test.go +++ b/CubeOps/internal/handler/cluster_test.go @@ -10,6 +10,7 @@ import ( "testing" "github.com/gin-gonic/gin" + "github.com/tencentcloud/CubeSandbox/CubeOps/internal/cubemaster" ) func newClusterRouter(t *testing.T, cm CubeMasterClient) *gin.Engine { @@ -88,7 +89,8 @@ func TestCluster_ListNodes_Success(t *testing.T) { getNodes: func(_ context.Context) (json.RawMessage, error) { return raw(`{"data": [ {"node_id": "n-1", "host_ip": "10.0.0.1", "instance_type": "cubebox", - "healthy": true, "capacity": {"milli_cpu": 4000, "memory_mb": 8192}, + "healthy": true, "scheduling_disabled": true, + "capacity": {"milli_cpu": 4000, "memory_mb": 8192}, "allocatable": {"milli_cpu": 4000, "memory_mb": 8192}, "max_mvm_num": 10, "quota_cpu": 4000, "quota_mem_mb": 8192, "create_concurrent_num": 5, "conditions": [], "local_templates": [], "versions": []} @@ -117,6 +119,12 @@ func TestCluster_ListNodes_Success(t *testing.T) { if nodes[0]["healthy"] != true { t.Errorf("healthy = %v, want true", nodes[0]["healthy"]) } + if nodes[0]["schedulingDisabled"] != true { + t.Errorf("schedulingDisabled = %v, want true", nodes[0]["schedulingDisabled"]) + } + if _, ok := nodes[0]["labels"]; ok { + t.Errorf("labels must not be exposed on nodeView: %v", nodes[0]["labels"]) + } // snake_case → camelCase for nested fields. cap, _ := nodes[0]["capacity"].(map[string]interface{}) if cap["cpuMilli"] != float64(4000) { @@ -124,6 +132,70 @@ func TestCluster_ListNodes_Success(t *testing.T) { } } +func TestCluster_IsolateNode_Success(t *testing.T) { + cm := &fakeCM{ + isolateNode: func(_ context.Context, id string) (json.RawMessage, error) { + return raw(`{"data":{"node_id":"` + id + `","host_ip":"10.0.0.1", + "healthy":true,"scheduling_disabled":true, + "capacity":{"milli_cpu":4000,"memory_mb":8192}, + "allocatable":{"milli_cpu":4000,"memory_mb":8192}, + "conditions":[],"local_templates":[],"versions":[]}}`), nil + }, + listSandboxes: func(_ context.Context) (json.RawMessage, error) { + return raw(`{"data":[]}`), nil + }, + } + w := httptestRecorder(t, newClusterRouter(t, cm), "PUT", "/api/v1/nodes/n-1/isolation") + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", w.Code, w.Body.String()) + } + var node map[string]interface{} + if err := json.Unmarshal(w.Body.Bytes(), &node); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if node["schedulingDisabled"] != true { + t.Errorf("schedulingDisabled = %v, want true", node["schedulingDisabled"]) + } +} + +func TestCluster_UnisolateNode_Success(t *testing.T) { + cm := &fakeCM{ + unisolateNode: func(_ context.Context, id string) (json.RawMessage, error) { + return raw(`{"data":{"node_id":"` + id + `","host_ip":"10.0.0.1", + "healthy":true,"scheduling_disabled":false, + "capacity":{"milli_cpu":4000,"memory_mb":8192}, + "allocatable":{"milli_cpu":4000,"memory_mb":8192}, + "conditions":[],"local_templates":[],"versions":[]}}`), nil + }, + listSandboxes: func(_ context.Context) (json.RawMessage, error) { + return raw(`{"data":[]}`), nil + }, + } + w := httptestRecorder(t, newClusterRouter(t, cm), "DELETE", "/api/v1/nodes/n-1/isolation") + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", w.Code, w.Body.String()) + } + var node map[string]interface{} + if err := json.Unmarshal(w.Body.Bytes(), &node); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if node["schedulingDisabled"] != false { + t.Errorf("schedulingDisabled = %v, want false", node["schedulingDisabled"]) + } +} + +func TestCluster_IsolateNode_NotFound(t *testing.T) { + cm := &fakeCM{ + isolateNode: func(_ context.Context, _ string) (json.RawMessage, error) { + return nil, &cubemaster.CMError{RetCode: 130404, RetMsg: "node not found"} + }, + } + w := httptestRecorder(t, newClusterRouter(t, cm), "PUT", "/api/v1/nodes/ghost/isolation") + if w.Code != http.StatusNotFound { + t.Errorf("status = %d, want 404; body=%s", w.Code, w.Body.String()) + } +} + func TestCluster_GetNode_NotFound(t *testing.T) { cm := &fakeCM{ getNode: func(_ context.Context, _ string) (json.RawMessage, error) { diff --git a/CubeOps/internal/handler/cmiface.go b/CubeOps/internal/handler/cmiface.go index 91e45d9b7..371605736 100644 --- a/CubeOps/internal/handler/cmiface.go +++ b/CubeOps/internal/handler/cmiface.go @@ -17,6 +17,8 @@ type CubeMasterClient interface { GetNodes(ctx context.Context) (json.RawMessage, error) ClusterVersions(ctx context.Context) (json.RawMessage, error) GetNode(ctx context.Context, nodeID string) (json.RawMessage, error) + IsolateNode(ctx context.Context, nodeID string) (json.RawMessage, error) + UnisolateNode(ctx context.Context, nodeID string) (json.RawMessage, error) ListSandboxes(ctx context.Context) (json.RawMessage, error) GetSandbox(ctx context.Context, sandboxID, instanceType string) (json.RawMessage, error) diff --git a/CubeOps/internal/handler/fake_cm_handler_test.go b/CubeOps/internal/handler/fake_cm_handler_test.go index e6e1bfeb5..1955ec2f7 100644 --- a/CubeOps/internal/handler/fake_cm_handler_test.go +++ b/CubeOps/internal/handler/fake_cm_handler_test.go @@ -35,6 +35,12 @@ func (f *fakeCMHandler) ClusterVersions(ctx context.Context) (json.RawMessage, e func (f *fakeCMHandler) GetNode(ctx context.Context, nodeID string) (json.RawMessage, error) { return nil, errMethodNotConfigured("GetNode") } +func (f *fakeCMHandler) IsolateNode(ctx context.Context, nodeID string) (json.RawMessage, error) { + return nil, errMethodNotConfigured("IsolateNode") +} +func (f *fakeCMHandler) UnisolateNode(ctx context.Context, nodeID string) (json.RawMessage, error) { + return nil, errMethodNotConfigured("UnisolateNode") +} func (f *fakeCMHandler) ListSandboxes(ctx context.Context) (json.RawMessage, error) { return nil, errMethodNotConfigured("ListSandboxes") } diff --git a/CubeOps/internal/handler/fake_cm_test.go b/CubeOps/internal/handler/fake_cm_test.go index bad11f046..6b8be1a59 100644 --- a/CubeOps/internal/handler/fake_cm_test.go +++ b/CubeOps/internal/handler/fake_cm_test.go @@ -23,6 +23,8 @@ type fakeCM struct { getNodes func(ctx context.Context) (json.RawMessage, error) clusterVersions func(ctx context.Context) (json.RawMessage, error) getNode func(ctx context.Context, nodeID string) (json.RawMessage, error) + isolateNode func(ctx context.Context, nodeID string) (json.RawMessage, error) + unisolateNode func(ctx context.Context, nodeID string) (json.RawMessage, error) listSandboxes func(ctx context.Context) (json.RawMessage, error) getSandbox func(ctx context.Context, sandboxID, instanceType string) (json.RawMessage, error) createSandbox func(ctx context.Context, body interface{}) (json.RawMessage, error) @@ -65,6 +67,18 @@ func (f *fakeCM) GetNode(ctx context.Context, nodeID string) (json.RawMessage, e } return f.getNode(ctx, nodeID) } +func (f *fakeCM) IsolateNode(ctx context.Context, nodeID string) (json.RawMessage, error) { + if f.isolateNode == nil { + return nil, errFakeNotConfigured + } + return f.isolateNode(ctx, nodeID) +} +func (f *fakeCM) UnisolateNode(ctx context.Context, nodeID string) (json.RawMessage, error) { + if f.unisolateNode == nil { + return nil, errFakeNotConfigured + } + return f.unisolateNode(ctx, nodeID) +} func (f *fakeCM) ListSandboxes(ctx context.Context) (json.RawMessage, error) { if f.listSandboxes == nil { return nil, errFakeNotConfigured diff --git a/docs/dev/redis-key-spec.md b/docs/dev/redis-key-spec.md index 22160b069..89267d002 100644 --- a/docs/dev/redis-key-spec.md +++ b/docs/dev/redis-key-spec.md @@ -64,6 +64,7 @@ The following are the standard keys currently registered in the system (`v1`). N | Sandbox lifecycle registry | `cube:v1:shared:sandbox:lifecycle:meta` | Hash | shared | CubeMaster | cube-lifecycle-manager | none (lifecycle via `HDEL`) | | Sandbox lifecycle events | `cube:v1:shared:sandbox:lifecycle:events` | Stream | shared | CubeMaster | cube-lifecycle-manager | MAXLEN ~ 100000 | | Sandbox lifecycle state | `cube:v1:shared:sandbox:lifecycle:state:{sandboxID}` | String | shared | cube-lifecycle-manager | cube-lifecycle-manager | SET TTL (default 60s) | +| Cubemaster control events | `cube:v1:shared:master:control:events` | Stream | shared | CubeMaster | CubeMaster (all replicas) | MAXLEN ~ 10000 | | CubeProxy replica registry | `cube:v1:shared:cube_proxy:registry` | Hash | shared | CubeProxy | cube-lifecycle-manager | none (evicted on heartbeat expiry via `HDEL`) | | CubeProxy replica heartbeat | `cube:v1:shared:cube_proxy:heartbeat` | Sorted Set | shared | CubeProxy | cube-lifecycle-manager | none (`ZREMRANGEBYSCORE` on expiry, default 15s) | @@ -121,6 +122,19 @@ See the `redis` tags on `InstanceInfoMap` in [`CubeMaster/pkg/base/types/redis.g | `paused` | Sandbox is paused | | `resuming` | Resume transition in progress | +**`master:control:events`** (CubeMaster control-plane fan-out) + +Stream fields: + +| field | Meaning | +| --- | --- | +| `op` | `node.isolate` \| `node.unisolate` (extensible) | +| `node_id` | Target compute node ID | +| `payload` | JSON `IsolationPayload` (`scheduling_disabled`, `updated_at_unix_ms`, `origin`) | +| `ts` | Event unix milliseconds | + +MySQL remains the source of truth for node labels / cordon. The stream accelerates in-memory convergence across Cubemaster replicas; each replica independently `XREAD`s (broadcast). See [`CubeMaster/pkg/controlevents`](https://github.com/tencentcloud/CubeSandbox/blob/master/CubeMaster/pkg/controlevents). + ## 6. TTL policy | Key type | Policy | Notes | @@ -132,6 +146,7 @@ See the `redis` tags on `InstanceInfoMap` in [`CubeMaster/pkg/base/types/redis.g | `sandbox:lifecycle:meta` | No TTL | Written on sandbox create, `HDEL` on destroy | | `sandbox:lifecycle:events` | MAXLEN ~ | Stream trimmed on each `XADD` (default ~100000) | | `sandbox:lifecycle:state` | SET TTL | `EX` on each write (cube-lifecycle-manager default 60s); released on rollback or sandbox delete | +| `master:control:events` | MAXLEN ~ | Stream trimmed on each `XADD` (default ~10000); Cubemaster replicas independently `XREAD` for broadcast | | `cube_proxy:registry` | No TTL (heartbeat-derived) | Written by each CubeProxy replica on startup; entries are `HDEL`'d by cube-lifecycle-manager once the corresponding heartbeat expires | | `cube_proxy:heartbeat` | Sorted Set expiry | Score = last heartbeat unix ms; entries older than `heartbeat_ttl` (default 15s) are removed via `ZREMRANGEBYSCORE` | | Cache keys (future) | TTL required | Must be declared on write and registered in this document | diff --git a/docs/guide/node-isolation.md b/docs/guide/node-isolation.md index c98ae6d5e..649a1e424 100644 --- a/docs/guide/node-isolation.md +++ b/docs/guide/node-isolation.md @@ -8,7 +8,7 @@ lang: en-US Node isolation (isolate) temporarily **stops CubeMaster from scheduling new sandboxes onto a compute node** during maintenance, upgrades, or troubleshooting. It behaves like Kubernetes `cordon`: the node can stay healthy and existing sandboxes keep running — it simply stops receiving new work. ::: tip Current entry points -WebUI / CubeOps / the public OpenAPI surface do **not** expose isolation yet. Use the **CubeMaster HTTP API**, or **`cubemastercli`** on the control node, to isolate and unisolate nodes. +Use the **WebUI Nodes detail page**, **CubeOps** `PUT/DELETE /api/v1/nodes/{nodeID}/isolation`, the **CubeMaster HTTP API**, or **`cubemastercli`** on the control node to isolate and unisolate nodes. ::: ## What you'll learn @@ -35,6 +35,8 @@ cube.cloud.tencentcloud.com/scheduling-disabled=true That label **cannot** be forged or cleared via the generic labels API or Cubelet registration — only the isolate / unisolate APIs on this page can change it. +Under the hood, the write lands in MySQL (source of truth). The handling Cubemaster replica updates its local scheduling cache immediately, then publishes a Redis Stream control event (`cube:v1:shared:master:control:events`) so **every Cubemaster replica** converges nearly in real time. Periodic DB reload remains a safety net if a replica briefly misses the stream. + ::: warning Isolation is not drain Isolation does **not** evict existing sandboxes. If your next step will interrupt sandbox networking or processes (for example, a Kubernetes compute-plane upgrade that recreates the Big Pod), **destroy sandboxes on that node yourself** after isolating, then proceed. See the [Kubernetes upgrade guide](./kubernetes/upgrade.md). ::: @@ -75,7 +77,23 @@ curl -s http://127.0.0.1:8089/internal/meta/nodes/ | jq '{ ## Isolate a node -### Option 1: HTTP API (best for scripts / automation) +### Option 1: WebUI (recommended for operators) + +Open **Nodes** → select the node → click **Isolate**. Confirm in the dialog. The page shows an **Isolated** badge when cordon is active; use **Unisolate** to clear it. + +### Option 2: CubeOps API + +```bash +# Isolate (requires CubeOps JWT) +curl -X PUT "http://127.0.0.1:3010/api/v1/nodes//isolation" \ + -H "Authorization: Bearer " + +# Unisolate +curl -X DELETE "http://127.0.0.1:3010/api/v1/nodes//isolation" \ + -H "Authorization: Bearer " +``` + +### Option 3: CubeMaster HTTP API (best for scripts on the control node) ```bash curl -X PUT "http://127.0.0.1:8089/internal/meta/nodes//isolation" @@ -103,7 +121,7 @@ A successful response looks like: The call is **idempotent**: repeating `PUT` on an already-isolated node is safe. No request body is required. -### Option 2: cubemastercli +### Option 4: cubemastercli ```bash # Isolate one node @@ -124,7 +142,7 @@ node node-1 isolated: scheduling_disabled=true ## Verify isolation -Query the node again and confirm `scheduling_disabled` is `true`: +In the WebUI, open the node detail page and confirm the **Isolated** badge is shown. Or query CubeMaster / CLI: ```bash curl -s http://127.0.0.1:8089/internal/meta/nodes/ | jq '.scheduling_disabled' @@ -135,7 +153,7 @@ cubemastercli --address 127.0.0.1 --port 8089 node list ``` ::: tip Wait window -After isolating, wait **≥ 60 seconds** so in-flight schedule / create windows can finish before you perform disruptive maintenance (reboot, upgrade, take-down, and so on). +After isolating, wait **≥ 60 seconds** so **in-flight** schedule / create RPCs that already selected the node can finish before you perform disruptive maintenance (reboot, upgrade, take-down, and so on). This wait is about the create pipeline window — **not** multi-replica cache lag. Other Cubemaster replicas pick up isolation via Redis Stream fan-out (with DB reload as fallback). ::: ## Unisolate a node @@ -143,7 +161,13 @@ After isolating, wait **≥ 60 seconds** so in-flight schedule / create windows When maintenance is done, remove the cordon so the node can receive new sandboxes again: ```bash -# HTTP +# WebUI: Nodes → node detail → Unisolate + +# CubeOps +curl -X DELETE "http://127.0.0.1:3010/api/v1/nodes//isolation" \ + -H "Authorization: Bearer " + +# CubeMaster HTTP curl -X DELETE "http://127.0.0.1:8089/internal/meta/nodes//isolation" # CLI @@ -176,6 +200,7 @@ Full steps: [Kubernetes upgrade guide](./kubernetes/upgrade.md). ## Scope and limitations - **Not a drain**: existing sandboxes are not migrated or destroyed automatically. +- **Multi-replica Cubemaster**: isolate/unisolate may hit any replica; MySQL is authoritative and Redis Stream broadcasts cordon changes so all replicas' schedulers converge quickly. - **Single-node / all-isolated clusters**: if no other schedulable node remains, new sandbox creates fail (no host selected). - **Orthogonal to health checks**: an isolated node can stay Healthy and may still appear in healthy-node listings; it is only excluded from the schedulable set. - **Independent of Kubernetes `kubectl cordon`**: this only affects CubeMaster scheduling; it does not cordon the Kubernetes Node. diff --git a/docs/zh/dev/redis-key-spec.md b/docs/zh/dev/redis-key-spec.md index 9ae351eba..4bcbbbd7b 100644 --- a/docs/zh/dev/redis-key-spec.md +++ b/docs/zh/dev/redis-key-spec.md @@ -64,6 +64,7 @@ cube:{ver}:{scope}:{resource}[:{sub}...]:{id} | 沙箱 lifecycle 注册表 | `cube:v1:shared:sandbox:lifecycle:meta` | Hash | shared | CubeMaster | cube-lifecycle-manager | 无（生命周期由 `HDEL` 管理） | | 沙箱 lifecycle 事件流 | `cube:v1:shared:sandbox:lifecycle:events` | Stream | shared | CubeMaster | cube-lifecycle-manager | MAXLEN ~ 100000 | | 沙箱 lifecycle 状态 | `cube:v1:shared:sandbox:lifecycle:state:{sandboxID}` | String | shared | cube-lifecycle-manager | cube-lifecycle-manager | SET TTL（默认 60s） | +| Cubemaster 控制面事件 | `cube:v1:shared:master:control:events` | Stream | shared | CubeMaster | CubeMaster（全部副本） | MAXLEN ~ 10000 | | CubeProxy 副本注册表 | `cube:v1:shared:cube_proxy:registry` | Hash | shared | CubeProxy | cube-lifecycle-manager | 无（心跳超时后由 `HDEL` 清理） | | CubeProxy 副本心跳 | `cube:v1:shared:cube_proxy:heartbeat` | Sorted Set | shared | CubeProxy | cube-lifecycle-manager | 无（`ZREMRANGEBYSCORE` 清理，默认 15s 过期） | @@ -121,6 +122,19 @@ cube:{ver}:{scope}:{resource}[:{sub}...]:{id} | `paused` | 沙箱已暂停 | | `resuming` | 恢复过渡中 | +**`master:control:events`**（CubeMaster 控制面广播） + +Stream 字段： + +| field | 含义 | +| --- | --- | +| `op` | `node.isolate` \| `node.unisolate`（可扩展） | +| `node_id` | 目标计算节点 ID | +| `payload` | JSON `IsolationPayload`（`scheduling_disabled`、`updated_at_unix_ms`、`origin`） | +| `ts` | 事件 unix 毫秒时间戳 | + +节点 labels / cordon 的权威源仍是 MySQL。Stream 用于加速各 Cubemaster 副本内存视图收敛；每个副本独立 `XREAD`（广播）。见 [`CubeMaster/pkg/controlevents`](https://github.com/tencentcloud/CubeSandbox/blob/master/CubeMaster/pkg/controlevents)。 + ## 6. TTL 策略 | Key 类型 | 策略 | 说明 | @@ -132,6 +146,7 @@ cube:{ver}:{scope}:{resource}[:{sub}...]:{id} | `sandbox:lifecycle:meta` | 无 TTL | 沙箱创建时写入，销毁时 `HDEL` | | `sandbox:lifecycle:events` | MAXLEN ~ | 每次 `XADD` 时裁剪（默认 ~100000） | | `sandbox:lifecycle:state` | SET TTL | 每次写入带 `EX`（cube-lifecycle-manager 默认 60s）；回滚或沙箱删除时释放 | +| `master:control:events` | MAXLEN ~ | 每次 `XADD` 时裁剪（默认 ~10000）；Cubemaster 各副本独立 `XREAD` 做广播 | | `cube_proxy:registry` | 无 TTL（依赖心跳） | 每个 CubeProxy 副本启动时写入；对应心跳过期后由 cube-lifecycle-manager 通过 `HDEL` 清理 | | `cube_proxy:heartbeat` | Sorted Set 过期 | Score 为最近一次心跳的 unix ms，超过 `heartbeat_ttl`（默认 15s）的条目由 `ZREMRANGEBYSCORE` 清理 | | 缓存类（未来新增） | 必须设 TTL | 写入时显式声明，并在文档中登记 | diff --git a/docs/zh/guide/node-isolation.md b/docs/zh/guide/node-isolation.md index 2a979911b..8218dc54d 100644 --- a/docs/zh/guide/node-isolation.md +++ b/docs/zh/guide/node-isolation.md @@ -8,7 +8,7 @@ lang: zh-CN 节点隔离（isolate）用于在维护、升级或排障时，**临时阻止 CubeMaster 向指定计算节点调度新沙箱**。它类似 Kubernetes 的 `cordon`：节点仍可保持健康、已有沙箱继续运行，只是不再接收新负载。 ::: tip 当前入口 -WebUI / CubeOps / 对外 OpenAPI **暂不提供**隔离操作。请通过 **CubeMaster HTTP 接口**，或控制节点上的 **`cubemastercli`** 完成隔离与取消隔离。 +可通过 **WebUI 节点详情页**、**CubeOps** `PUT/DELETE /api/v1/nodes/{nodeID}/isolation`、**CubeMaster HTTP 接口**，或控制节点上的 **`cubemastercli`** 完成隔离与取消隔离。 ::: ## 读完本页你会知道 @@ -35,6 +35,8 @@ cube.cloud.tencentcloud.com/scheduling-disabled=true 该 label **不能**通过普通 labels API 或 Cubelet 注册伪造 / 清除，只能走本文的隔离 / 取消隔离接口。 +实现上，写入会落到 MySQL（权威源）。处理请求的 Cubemaster 副本会立刻更新本机调度缓存，再通过 Redis Stream 控制面事件（`cube:v1:shared:master:control:events`）广播，使**所有 Cubemaster 副本**近实时收敛。周期性 DB reload 仍作为兜底。 + ::: warning 隔离 ≠ 清空节点 隔离**不会**驱逐存量沙箱。若你要做会中断沙箱网络或进程的操作（例如 K8s 计算面升级会 recreate Big Pod），需要在隔离之后**自行销毁**该节点上的沙箱，再进行维护。详见 [K8s 升级指南](./kubernetes/upgrade.md)。 ::: @@ -75,7 +77,23 @@ curl -s http://127.0.0.1:8089/internal/meta/nodes/ | jq '{ ## 隔离节点 -### 方式一：HTTP 接口（推荐脚本 / 自动化） +### 方式一：WebUI（推荐运维操作） + +打开 **节点** → 进入目标节点详情 → 点击 **隔离**，在确认框中确认。生效后页头会显示 **已隔离** 标记；用 **取消隔离** 恢复调度。 + +### 方式二：CubeOps API + +```bash +# 隔离（需要 CubeOps JWT） +curl -X PUT "http://127.0.0.1:3010/api/v1/nodes//isolation" \ + -H "Authorization: Bearer " + +# 取消隔离 +curl -X DELETE "http://127.0.0.1:3010/api/v1/nodes//isolation" \ + -H "Authorization: Bearer " +``` + +### 方式三：CubeMaster HTTP 接口（适合控制节点脚本） ```bash curl -X PUT "http://127.0.0.1:8089/internal/meta/nodes//isolation" @@ -103,7 +121,7 @@ curl -X PUT "http://127.0.0.1:8089/internal/meta/nodes//isolation" 接口**幂等**：对已隔离节点重复 `PUT` 是安全的。无需请求体。 -### 方式二：cubemastercli +### 方式四：cubemastercli ```bash # 隔离单个节点 @@ -124,7 +142,7 @@ node node-1 isolated: scheduling_disabled=true ## 确认隔离生效 -再次查询节点，确认 `scheduling_disabled` 为 `true`： +在 WebUI 节点详情页确认出现 **已隔离** 标记。或查询 CubeMaster / CLI： ```bash curl -s http://127.0.0.1:8089/internal/meta/nodes/ | jq '.scheduling_disabled' @@ -135,7 +153,7 @@ cubemastercli --address 127.0.0.1 --port 8089 node list ``` ::: tip 建议等待窗口 -隔离后建议再等待 **≥ 60 秒**，让进行中的调度 / 创建窗口结束，再对该节点做破坏性维护（重启、升级、下线等）。 +隔离后建议再等待 **≥ 60 秒**，让**已经选中该节点、尚在飞行中的**调度 / 创建 RPC 结束，再做破坏性维护（重启、升级、下线等）。这段等待针对的是创建流水线窗口，**不是**多副本缓存延迟。其它 Cubemaster 副本通过 Redis Stream 广播近实时生效（DB reload 为兜底）。 ::: ## 取消隔离 @@ -143,7 +161,13 @@ cubemastercli --address 127.0.0.1 --port 8089 node list 维护完成后，取消隔离，节点即可重新接收新沙箱： ```bash -# HTTP +# WebUI：节点详情 → 取消隔离 + +# CubeOps +curl -X DELETE "http://127.0.0.1:3010/api/v1/nodes//isolation" \ + -H "Authorization: Bearer " + +# CubeMaster HTTP curl -X DELETE "http://127.0.0.1:8089/internal/meta/nodes//isolation" # CLI @@ -176,6 +200,7 @@ cubemastercli --address 127.0.0.1 --port 8089 node unisolate ## 范围与限制 - **不是 drain**：不会自动迁移或销毁已有沙箱。 +- **Cubemaster 多副本**：isolate/unisolate 可打到任一副本；MySQL 为权威源，Redis Stream 广播 cordon 变更，使各副本调度视图快速收敛。 - **单节点 / 全部隔离**：若集群中没有其它可调度节点，新沙箱创建会失败（调度选不到节点）。 - **与健康检查正交**：隔离节点仍可保持 Healthy，仍会出现在健康节点列表中，只是不进入可调度集合。 - **与 Kubernetes `kubectl cordon` 无关**：本能力只影响 CubeMaster 调度，不会自动对 K8s Node 执行 cordon。 diff --git a/web/src/api/client.ts b/web/src/api/client.ts index f0b2f5ee5..cf3f05b8e 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -106,6 +106,7 @@ export interface ClusterNodeView { memorySaturationPct: number; heartbeatTime?: string | null; healthy: boolean; + schedulingDisabled: boolean; localTemplates: string[]; versions: ComponentVersionDto[]; } @@ -159,6 +160,9 @@ function mapTemplateDetail(dto: TemplateDetailDto): TemplateDetail { } function mapNode(dto: ApiNodeView): ClusterNodeView { + const extended = dto as ApiNodeView & { + schedulingDisabled?: boolean; + }; return { nodeID: dto.nodeID, hostname: undefined, @@ -187,6 +191,7 @@ function mapNode(dto: ApiNodeView): ClusterNodeView { memorySaturationPct: Math.round(dto.memorySaturation), heartbeatTime: dto.heartbeatTime, healthy: dto.healthy, + schedulingDisabled: extended.schedulingDisabled ?? false, localTemplates: dto.localTemplates ?? [], versions: dto.versions ?? [], }; @@ -279,6 +284,12 @@ export const clusterApi = { overview: () => ops('/cluster/overview'), nodes: () => ops('/nodes').then((items) => items.map(mapNode)), node: (id: string) => ops(`/nodes/${id}`).then(mapNode), + isolate: (id: string) => + ops(`/nodes/${encodeURIComponent(id)}/isolation`, { method: 'PUT' }).then(mapNode), + unisolate: (id: string) => + ops(`/nodes/${encodeURIComponent(id)}/isolation`, { + method: 'DELETE', + }).then(mapNode), config: () => ops<{ apiEndpoint: string; diff --git a/web/src/api/generated/schema.ts b/web/src/api/generated/schema.ts index bb0cdc787..fac54d937 100644 --- a/web/src/api/generated/schema.ts +++ b/web/src/api/generated/schema.ts @@ -407,6 +407,10 @@ export interface components { * @description Memory quota in MiB assigned to this node. */ quotaMemMB: number; + /** + * @description Whether CubeMaster will skip this node when scheduling new sandboxes. + */ + schedulingDisabled?: boolean; versions?: components["schemas"]["ComponentVersionView"][]; }; /** @description Request body for POST /sandboxes/{id}/resume (deprecated). */ diff --git a/web/src/components/nodes/IsolateConfirmDialog.tsx b/web/src/components/nodes/IsolateConfirmDialog.tsx new file mode 100644 index 000000000..feb7023e5 --- /dev/null +++ b/web/src/components/nodes/IsolateConfirmDialog.tsx @@ -0,0 +1,72 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (C) 2026 Tencent. All rights reserved. + +import { createPortal } from 'react-dom'; +import { useTranslation } from 'react-i18next'; +import { X } from 'lucide-react'; +import { ApiError } from '@/lib/api'; +import { Button } from '@/components/ui/button'; +import { Card, CardContent, CardHeader, CardTitle } from '@/components/ui/card'; + +export function formatIsolationError(err: unknown, fallback: string): string { + if (err instanceof ApiError) { + if (typeof err.body === 'object' && err.body && 'error' in err.body) { + const msg = (err.body as { error?: string }).error; + if (msg) return msg; + } + if (err.message) return err.message; + } + if (err instanceof Error && err.message) return err.message; + return fallback; +} + +type IsolateConfirmDialogProps = { + open: boolean; + onClose: () => void; + onConfirm: () => void; + pending?: boolean; + error?: string | null; +}; + +export function IsolateConfirmDialog({ + open, + onClose, + onConfirm, + pending = false, + error, +}: IsolateConfirmDialogProps) { + const { t } = useTranslation('nodeDetail'); + + if (!open) return null; + + return createPortal( +
+ + + {t('isolation.confirmTitle')} + + + +

{t('isolation.confirmDesc')}

+ {error &&

{error}

} +
+ + +
+ + +
, + document.body, + ); +} diff --git a/web/src/lib/utils.ts b/web/src/lib/utils.ts index b87cde9f7..c175c3992 100644 --- a/web/src/lib/utils.ts +++ b/web/src/lib/utils.ts @@ -31,6 +31,30 @@ export function short(id: string, head = 6, tail = 4): string { return `${id.slice(0, head)}…${id.slice(-tail)}`; } +export function formatCondition(type: string, status: string): string { + if (type === 'Ready') { + return status === 'True' ? 'Ready' : 'Not Ready'; + } + if (type.endsWith('Pressure')) { + const base = type.replace('Pressure', ''); + return status === 'True' ? `${base} Pressure` : `${base} OK`; + } + if (type === 'NetworkUnavailable') { + return status === 'True' ? 'Network Unavailable' : 'Network OK'; + } + return `${type}: ${status}`; +} + +export function getConditionTone(type: string, status: string): 'ok' | 'warn' | 'err' { + if (type === 'Ready') { + return status === 'True' ? 'ok' : 'err'; + } + if (type.endsWith('Pressure') || type === 'NetworkUnavailable') { + return status === 'False' ? 'ok' : 'warn'; + } + return status === 'True' ? 'ok' : 'warn'; +} + /** * Copy text to clipboard with execCommand fallback for HTTP (non-HTTPS) environments. * On success, dispatches a 'cube:toast' custom event so ToastProvider can show a notification. diff --git a/web/src/locales/en/nodeDetail.json b/web/src/locales/en/nodeDetail.json index 9264e10cf..471f7c3b7 100644 --- a/web/src/locales/en/nodeDetail.json +++ b/web/src/locales/en/nodeDetail.json @@ -23,5 +23,19 @@ }, "empty": { "sandboxes": "No sandboxes running on this node." + }, + "isolation": { + "badge": "Isolated", + "isolate": "Isolate", + "unisolate": "Unisolate", + "isolating": "Isolating…", + "unisolating": "Unisolating…", + "confirmTitle": "Isolate this node?", + "confirmDesc": "New sandboxes will no longer be scheduled to this node. Existing sandboxes are not affected.", + "confirm": "Isolate", + "cancel": "Cancel", + "failed": "Isolation request failed.", + "isolatedToast": "Node isolated", + "unisolatedToast": "Node unisolated" } } diff --git a/web/src/locales/en/nodes.json b/web/src/locales/en/nodes.json index 9b806edd8..130619fa2 100644 --- a/web/src/locales/en/nodes.json +++ b/web/src/locales/en/nodes.json @@ -3,5 +3,6 @@ "subtitle": "Host capacity, saturation and conditions across the fleet.", "noNodes": "No nodes registered.", "cpu": "CPU", - "memory": "Memory" + "memory": "Memory", + "isolated": "Isolated" } diff --git a/web/src/locales/zh/nodeDetail.json b/web/src/locales/zh/nodeDetail.json index 6878244ee..e41174ba9 100644 --- a/web/src/locales/zh/nodeDetail.json +++ b/web/src/locales/zh/nodeDetail.json @@ -23,5 +23,19 @@ }, "empty": { "sandboxes": "该节点上暂无运行中的沙箱。" + }, + "isolation": { + "badge": "已隔离", + "isolate": "隔离", + "unisolate": "取消隔离", + "isolating": "隔离中…", + "unisolating": "取消隔离中…", + "confirmTitle": "隔离该节点？", + "confirmDesc": "隔离后，该节点将不再接收新的沙箱调度。已有沙箱不受影响。", + "confirm": "确认隔离", + "cancel": "取消", + "failed": "隔离操作失败。", + "isolatedToast": "节点已隔离", + "unisolatedToast": "已取消隔离" } } diff --git a/web/src/locales/zh/nodes.json b/web/src/locales/zh/nodes.json index 191898f9b..5567de105 100644 --- a/web/src/locales/zh/nodes.json +++ b/web/src/locales/zh/nodes.json @@ -3,5 +3,6 @@ "subtitle": "集群中各主机的容量、饱和度与状态。", "noNodes": "暂无已注册节点。", "cpu": "CPU", - "memory": "内存" + "memory": "内存", + "isolated": "已隔离" } diff --git a/web/src/pages/NodeDetail.tsx b/web/src/pages/NodeDetail.tsx index fe388811c..91f99768a 100644 --- a/web/src/pages/NodeDetail.tsx +++ b/web/src/pages/NodeDetail.tsx @@ -1,14 +1,21 @@ // SPDX-License-Identifier: Apache-2.0 // Copyright (C) 2026 Tencent. All rights reserved. -import { useQuery } from '@tanstack/react-query'; +import { useState } from 'react'; +import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'; import { Link, useParams } from 'react-router-dom'; import { useTranslation } from 'react-i18next'; import { clusterApi, sandboxApi, templateApi } from '@/api/client'; +import { + formatIsolationError, + IsolateConfirmDialog, +} from '@/components/nodes/IsolateConfirmDialog'; import { Badge } from '@/components/ui/badge'; +import { Button } from '@/components/ui/button'; import { Skeleton } from '@/components/ui/skeleton'; -import { ArrowLeft, Package, Box, Activity } from 'lucide-react'; -import { cn, formatRelative } from '@/lib/utils'; +import { showToast } from '@/components/ui/ToastProvider'; +import { ArrowLeft, Package, Box, Activity, ShieldOff, ShieldCheck } from 'lucide-react'; +import { cn, formatRelative, formatCondition, getConditionTone } from '@/lib/utils'; // ── Resource bar ────────────────────────────────────────────────────────────── @@ -110,7 +117,8 @@ function ConditionRow({ message?: string; time?: string | null; }) { - const ok = status === 'True'; + const tone = getConditionTone(type, status); + const text = formatCondition(type, status); return (
@@ -118,20 +126,10 @@ function ConditionRow({ - {type} - - {status} - + {text}
{reason &&

{reason}

} {message && ( @@ -152,6 +150,9 @@ function ConditionRow({ export default function NodeDetailPage() { const { nodeID } = useParams<{ nodeID: string }>(); const { t } = useTranslation('nodeDetail'); + const qc = useQueryClient(); + const [confirmIsolate, setConfirmIsolate] = useState(false); + const [actionError, setActionError] = useState(null); const { data, isLoading, isError } = useQuery({ queryKey: ['node', nodeID], @@ -174,12 +175,44 @@ export default function NodeDetailPage() { enabled: !!data, }); + const invalidateNode = async () => { + await Promise.all([ + qc.invalidateQueries({ queryKey: ['node', nodeID] }), + qc.invalidateQueries({ queryKey: ['nodes'] }), + ]); + }; + + const isolate = useMutation({ + mutationFn: () => clusterApi.isolate(nodeID!), + onMutate: () => setActionError(null), + onSuccess: async () => { + setConfirmIsolate(false); + showToast(t('isolation.isolatedToast')); + await invalidateNode(); + }, + onError: (err) => { + setActionError(formatIsolationError(err, t('isolation.failed'))); + }, + }); + + const unisolate = useMutation({ + mutationFn: () => clusterApi.unisolate(nodeID!), + onMutate: () => setActionError(null), + onSuccess: async () => { + showToast(t('isolation.unisolatedToast')); + await invalidateNode(); + }, + onError: (err) => { + setActionError(formatIsolationError(err, t('isolation.failed'))); + }, + }); + // local templates with READY or RUNNING status only const localTemplateIDs = new Set(data?.localTemplates ?? []); const visibleLocalTemplates = (allTemplates ?? []).filter( - (t) => - localTemplateIDs.has(t.templateID) && - ['READY', 'RUNNING'].includes((t.status ?? '').toUpperCase()), + (tpl) => + localTemplateIDs.has(tpl.templateID) && + ['READY', 'RUNNING'].includes((tpl.status ?? '').toUpperCase()), ); const nodeSandboxes = (allSandboxes ?? []).filter((sb) => sb.clientID === data?.address); @@ -212,6 +245,7 @@ export default function NodeDetailPage() { const memUsed = data.resources.totalMemoryMB - data.resources.allocatableMemoryMB; const isReady = data.status.toLowerCase() === 'ready'; + const isolationPending = isolate.isPending || unisolate.isPending; return (
@@ -242,6 +276,9 @@ export default function NodeDetailPage() {

{data.hostname ?? data.nodeID}

+ {data.schedulingDisabled && ( + {t('isolation.badge')} + )}
{data.nodeID} @@ -259,14 +296,53 @@ export default function NodeDetailPage() { )}
-
- - - {formatRelative(data.heartbeatTime)} - +
+ {data.schedulingDisabled ? ( + + ) : ( + + )} +
+ + + {formatRelative(data.heartbeatTime)} + +
+ {actionError && ( +
+ {actionError} +
+ )} + + setConfirmIsolate(false)} + onConfirm={() => isolate.mutate()} + pending={isolate.isPending} + error={ + isolate.isError ? formatIsolationError(isolate.error, t('isolation.failed')) : null + } + /> + {/* resource KPIs */}
diff --git a/web/src/pages/Nodes.tsx b/web/src/pages/Nodes.tsx index f3567f5f0..1961160eb 100644 --- a/web/src/pages/Nodes.tsx +++ b/web/src/pages/Nodes.tsx @@ -1,15 +1,23 @@ // SPDX-License-Identifier: Apache-2.0 // Copyright (C) 2026 Tencent. All rights reserved. -import { useQuery } from '@tanstack/react-query'; +import { useState } from 'react'; +import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'; import { useTranslation } from 'react-i18next'; import { Link } from 'react-router-dom'; import { clusterApi } from '@/api/client'; +import { + formatIsolationError, + IsolateConfirmDialog, +} from '@/components/nodes/IsolateConfirmDialog'; +import * as DropdownMenu from '@radix-ui/react-dropdown-menu'; import { Card, CardHeader, CardTitle, CardDescription } from '@/components/ui/card'; import { Badge } from '@/components/ui/badge'; +import { Button } from '@/components/ui/button'; import { Skeleton } from '@/components/ui/skeleton'; -import { Cpu, HardDrive, Server } from 'lucide-react'; -import { cn, formatRelative } from '@/lib/utils'; +import { showToast } from '@/components/ui/ToastProvider'; +import { Cpu, HardDrive, Server, ShieldCheck, ShieldOff, MoreHorizontal } from 'lucide-react'; +import { cn, formatRelative, formatCondition, getConditionTone } from '@/lib/utils'; export default function NodesPage() { const { data, isLoading } = useQuery({ @@ -18,6 +26,28 @@ export default function NodesPage() { refetchInterval: 15_000, }); const { t } = useTranslation('nodes'); + const { t: td } = useTranslation('nodeDetail'); + const qc = useQueryClient(); + const [confirmNodeID, setConfirmNodeID] = useState(null); + + const isolate = useMutation({ + mutationFn: (nodeID: string) => clusterApi.isolate(nodeID), + onSuccess: async () => { + setConfirmNodeID(null); + showToast(td('isolation.isolatedToast')); + await qc.invalidateQueries({ queryKey: ['nodes'] }); + }, + }); + + const unisolate = useMutation({ + mutationFn: (nodeID: string) => clusterApi.unisolate(nodeID), + onSuccess: async () => { + showToast(td('isolation.unisolatedToast')); + await qc.invalidateQueries({ queryKey: ['nodes'] }); + }, + }); + + const isolationPending = isolate.isPending || unisolate.isPending; return (
@@ -47,7 +77,7 @@ export default function NodesPage() { -
+
{n.status.toLowerCase() === 'ready' && ( @@ -61,12 +91,72 @@ export default function NodesPage() { /> {n.hostname && n.hostname !== n.nodeID ? n.hostname : n.nodeID} + {n.schedulingDisabled && ( + {t('isolated')} + )} {n.hostname && n.hostname !== n.nodeID && ( {n.nodeID} )}
+ + + + + + { + e.preventDefault(); + e.stopPropagation(); + }} + onPointerDown={(e) => e.stopPropagation()} + onPointerUp={(e) => e.stopPropagation()} + > + {n.schedulingDisabled ? ( + { + unisolate.mutate(n.nodeID); + }} + > + + {unisolate.isPending && unisolate.variables === n.nodeID + ? td('isolation.unisolating') + : td('isolation.unisolate')} + + ) : ( + { + isolate.reset(); + setConfirmNodeID(n.nodeID); + }} + > + + {td('isolation.isolate')} + + )} + + +
@@ -93,15 +183,14 @@ export default function NodesPage() {
{n.conditions && n.conditions.length > 0 && ( -
+
{n.conditions.slice(0, 3).map((c, i) => (
- {c.type} - - {c.status} - - {formatRelative(c.lastTransitionTime)} - + + {formatCondition(c.type, c.status)} + + + {formatRelative(c.lastTransitionTime)}
))} @@ -117,6 +206,20 @@ export default function NodesPage() {
{t('noNodes')}
)} + + { + if (!isolate.isPending) setConfirmNodeID(null); + }} + onConfirm={() => { + if (confirmNodeID) isolate.mutate(confirmNodeID); + }} + pending={isolate.isPending} + error={ + isolate.isError ? formatIsolationError(isolate.error, td('isolation.failed')) : null + } + />
); }