[conch-agent/conchd] refactor vsock to host-pull model and add CID allocator--AtomGit

Mcmd/conch-agent/main.go+78-110

	@@ -3,13 +3,11 @@ package main
3	import (	3	import (
4	"flag"	4	"flag"
5	"fmt"	5	"fmt"
6	- "math/rand"
7	"net"	6	"net"
8	"os"	7	"os"
9	"strings"	8	"strings"
10	"sync"	9	"sync"
11	"syscall"	10	"syscall"
12	- "time"
13		11
14	"golang.org/x/sys/unix"	12	"golang.org/x/sys/unix"
15	"google.golang.org/grpc"	13	"google.golang.org/grpc"
	@@ -20,67 +18,18 @@ import (
20	)	18	)
21		19
22	const (	20	const (
23	- ServerPort = ":4064"	21	+ ServerPort = ":4064"
24	- ServerVersion = "0.0.1"	22	+ ServerVersion = "0.0.2"
25	- vsockCIDHost = unix.VMADDR_CID_HOST // CID 2 = host	23	+ vsockReadyPort = 4065
26	- vsockReadyPort = 4065 // must match conchd side
27	- vsockReadyTimeout = 30 * time.Second // timeout for ready signal loop
28	- vsockRetryBase = 50 * time.Millisecond // base retry interval
29	- vsockRetryJitter = 25 * time.Millisecond // max jitter (+/- 25ms)
30	)	24	)
31		25
32	var (	26	var (
33	currentSandboxID string	27	currentSandboxID string
34	- agentLogger ulog.Logger	28	+ rootLogger ulog.Logger
35	mu sync.Mutex	29	mu sync.Mutex
		30	+ isSafe = true
36	)	31	)
37		32
38	-// updateSandboxID reads sandbox_id from cmdline and updates global logger context
39	-func updateSandboxID() {
40	- mu.Lock()
41	- defer mu.Unlock()
42	-
43	- id := getSandboxIDFromCmdline()
44	- if id == "" {
45	- return
46	- }
47	-
48	- // Always update if it's the first time or if it changed
49	- if id == currentSandboxID {
50	- return
51	- }
52	-
53	- currentSandboxID = id
54	- // Update the global logger with the new sandboxId field.
55	- // ulog.With returns a new logger instance with the added field.
56	- agentLogger = ulog.With(ulog.F("sandboxId", id))
57	- ulog.SetLogger(agentLogger)
58	-
59	- agentLogger.Info("Updated sandbox_id from cmdline", ulog.F("sandbox_id", id))
60	-}
61	-
62	-// monitorTimeDrift detects snapshot resume by monitoring system clock jumps
63	-func monitorTimeDrift() {
64	- lastTick := time.Now()
65	- ticker := time.NewTicker(500 * time.Millisecond)
66	- defer ticker.Stop()
67	-
68	- for {
69	- <-ticker.C
70	- now := time.Now()
71	- elapsed := now.Sub(lastTick)
72	-
73	- // If elapsed time since last tick is significantly larger than expected interval (500ms),
74	- // it indicates the VM was likely frozen/resumed.
75	- if elapsed > 2*time.Second {
76	- ulog.Info("Snapshot boot or significant time jump detected", ulog.F("elapsed", elapsed))
77	- updateSandboxID()
78	- }
79	- lastTick = now
80	- }
81	-}
82	-
83	-// getSandboxIDFromCmdline reads conch.sandbox_id from /proc/cmdline
84	func getSandboxIDFromCmdline() string {	33	func getSandboxIDFromCmdline() string {
85	data, err := os.ReadFile("/proc/cmdline")	34	data, err := os.ReadFile("/proc/cmdline")
86	if err != nil {	35	if err != nil {
	@@ -94,63 +43,76 @@ func getSandboxIDFromCmdline() string {
94	return ""	43	return ""
95	}	44	}
96		45
97	-// vsockDial connects to the host via AF_VSOCK using unix package	46	+func createVsockListener(port uint32) (int, error) {
98	-func vsockDial(cid uint32, port uint32) (int, error) {
99	fd, err := unix.Socket(unix.AF_VSOCK, unix.SOCK_STREAM, 0)	47	fd, err := unix.Socket(unix.AF_VSOCK, unix.SOCK_STREAM, 0)
100	if err != nil {	48	if err != nil {
101	- return -1, fmt.Errorf("socket: %w", err)	49	+ return -1, fmt.Errorf("failed to create vsock socket: %w", err)
102	}	50	}
103		51
104	sa := &unix.SockaddrVM{	52	sa := &unix.SockaddrVM{
105	- CID: cid,	53	+ CID: unix.VMADDR_CID_ANY,
106	Port: port,	54	Port: port,
107	}	55	}
108		56
109	- err = unix.Connect(fd, sa)	57	+ if err := unix.Bind(fd, sa); err != nil {
110	- if err != nil {
111	unix.Close(fd)	58	unix.Close(fd)
112	- return -1, err	59	+ return -1, fmt.Errorf("failed to bind vsock socket to port %d: %w", port, err)
		60	+ }
		61	+
		62	+ if err := unix.Listen(fd, 5); err != nil {
		63	+ unix.Close(fd)
		64	+ return -1, fmt.Errorf("failed to listen on vsock port %d: %w", port, err)
113	}	65	}
114		66
115	return fd, nil	67	return fd, nil
116	}	68	}
117		69
118	-// sendReadySignalLoop sends ready signal to host via vsock.	70	+func acceptVsockConnection(fd int) (int, error) {
119	-// Runs a loop with 50ms interval until ACK received or timeout:	71	+ nfd, _, err := unix.Accept(fd)
120	-// - Fresh create: signals within ~50ms after gRPC is bound.	72	+ if err != nil {
121	-// - Snapshot restore: goroutine resumes from freeze, signals within ~50ms.	73	+ return -1, fmt.Errorf("failed to accept vsock connection: %w", err)
122	-//	74	+ }
123	-// After wait for ACK from conchd, the listener socket is removed.	75	+ return nfd, nil
124	-// Subsequent vsockDial attempts fail immediately (ECONNREFUSED in kernel),	76	+}
125	-// costing ~20 failed syscalls/second — negligible overhead, zero network traffic.	77	+
126	-func sendReadySignalLoop(logger ulog.Logger, timeout time.Duration) {	78	+func listenVsockLoop(handler VsockHandler) {
127	- deadline := time.Now().Add(timeout)	79	+ logger := ulog.GetLogger()
		80	+ logger.Info("Starting vsock listener loop", ulog.F("port", vsockReadyPort))
		81	+
		82	+ fd, err := createVsockListener(vsockReadyPort)
		83	+ if err != nil {
		84	+ logger.Error("Failed to create vsock listener", ulog.F("error", err))
		85	+ return
		86	+ }
		87	+ defer unix.Close(fd)
		88	+
		89	+ logger.Info("Vsock listener started", ulog.F("port", vsockReadyPort))
		90	+
128	for {	91	for {
129	- if time.Now().After(deadline) {	92	+ connFd, err := acceptVsockConnection(fd)
130	- logger.Warn("vsock ready signal loop timed out", ulog.F("timeout", timeout))	93	+ if err != nil {
131	- return	94	+ logger.Error("Failed to accept vsock connection", ulog.F("error", err))
		95	+ continue
132	}	96	}
133		97
134	- fd, err := vsockDial(vsockCIDHost, vsockReadyPort)	98	+ buf := make([]byte, 1024)
135	- if err == nil {	99	+ n, err := unix.Read(connFd, buf)
136	- logger.Info("Sent agent readiness signal via vsock")	100	+ if err != nil {
		101	+ logger.Error("Failed to read from vsock connection", ulog.F("error", err))
		102	+ } else if n > 0 {
		103	+ message := string(buf[:n])
		104	+ logger.Info("Received message via vsock", ulog.F("message", message))
137		105
138	- // Send the signal payload	106	+ response := handler.HandleMessage(message)
139	- unix.Write(fd, []byte("READY\n"))	107	+ if response != "" {
140	-	108	+ unix.Write(connFd, []byte(response))
141	- // Wait for ACK from conchd. When received, it means the host has
142	- // successfully recorded our readiness and we can stop the loop.
143	- buf := make([]byte, 8)
144	- n, _ := unix.Read(fd, buf)
145	- if n > 0 && string(buf[:n]) == "ACK" {
146	- logger.Info("Received ACK from host, stopping vsock signal loop")
147	- unix.Close(fd)
148	- return
149	}	109	}
150	- unix.Close(fd)
151	}	110	}
152	- jitter := time.Duration(rand.Int63n(int64(vsockRetryJitter)*2)) - vsockRetryJitter	111	+ logger = ulog.GetLogger()
153	- time.Sleep(vsockRetryBase + jitter)	112	+ if err := unix.Close(connFd); err != nil {
		113	+ logger.Error("Failed to close vsock connection", ulog.F("error", err))
		114	+ }
		115	+ logger.Info("Continuing to listen for next signal via vsock")
154	}	116	}
155	}	117	}
156		118
	@@ -170,14 +132,23 @@ func main() {
170	if err != nil {	132	if err != nil {
171	panic(err)	133	panic(err)
172	}	134	}
		135	+
		136	+ rootLogger = ulog.GetLogger()
173	defer func() {	137	defer func() {
174	- logger := ulog.GetLogger()	138	+ if closer, ok := rootLogger.(interface{ Close() error }); ok {
175	- if closer, ok := logger.(interface{ Close() error }); ok {
176	_ = closer.Close()	139	_ = closer.Close()
177	}	140	}
178	}()	141	}()
179		142
180	- updateSandboxID()	143	+ // Get initial sandbox ID from cmdline
		144	+ initialSandboxID := getSandboxIDFromCmdline()
		145	+ if initialSandboxID != "" {
		146	+ mu.Lock()
		147	+ currentSandboxID = initialSandboxID
		148	+ mu.Unlock()
		149	+ }
		150	+
		151	+ ulog.SetLogger(rootLogger.With(ulog.F("sandboxId", currentSandboxID)))
181	logger := ulog.GetLogger()	152	logger := ulog.GetLogger()
182		153
183	logger.Info("Starting conch-agent",	154	logger.Info("Starting conch-agent",
	@@ -223,21 +194,18 @@ func main() {
223	pb.RegisterAgentServiceServer(grpcServer, &AgentServer{Version: ServerVersion})	194	pb.RegisterAgentServiceServer(grpcServer, &AgentServer{Version: ServerVersion})
224		195
225	reflection.Register(grpcServer)	196	reflection.Register(grpcServer)
226	- logger.Info("Agent gRPC server listening",
227	- ulog.F("address", listener.Addr()),
228	- ulog.F("version", ServerVersion),
229	- )
230		197
231	- // Start vsock ready signal loop AFTER gRPC is bound	198	+ go func() {
232	- go sendReadySignalLoop(logger, vsockReadyTimeout)	199	+ logger.Info("gRPC server starting", ulog.F("address", listener.Addr()))
233		200
234	- // Start snapshot boot monitor	201	+ if err := grpcServer.Serve(listener); err != nil {
235	- go monitorTimeDrift()	202	+ mu.Lock()
236	-	203	+ isSafe = false
237	- if err := grpcServer.Serve(listener); err != nil {	204	+ mu.Unlock()
238	- logger.Fatal("Failed to serve gRPC",	205	+ logger.Error("gRPC server failed", ulog.F("error", err))
239	- ulog.F("error", err),	206	+ }
240	- )	207	+ }()
241	- }
242		208
		209	+ vsockHandler := NewVsockHandler(ServerVersion, checkGRPCHealth)
		210	+ listenVsockLoop(vsockHandler)
243	}	211	}

Acmd/conch-agent/vsock.go+77-0

@@ -0,0 +1,77 @@
	1	+package main
	2	+
	3	+import (
	4	+ "strings"
	5	+ "sync"
	6	+
	7	+ "github.com/openeuler/Conch/pkg/ulog"
	8	+)
	9	+
	10	+type VsockHandler interface {
	11	+ HandleMessage(message string) string
	12	+ GetSandboxID() string
	13	+ SetSandboxID(id string)
	14	+}
	15	+
	16	+type VsockHandlerImpl struct {
	17	+ mu sync.Mutex
	18	+ sandboxID string
	19	+ version string
	20	+ healthFunc func() bool
	21	+}
	22	+
	23	+func NewVsockHandler(version string, healthFunc func() bool) *VsockHandlerImpl {
	24	+ return &VsockHandlerImpl{
	25	+ version: version,
	26	+ healthFunc: healthFunc,
	27	+ }
	28	+}
	29	+
	30	+func (h *VsockHandlerImpl) HandleMessage(message string) string {
	31	+ logger := ulog.GetLogger()
	32	+ logger.Info("Handling vsock message", ulog.F("message", message))
	33	+
	34	+ if strings.Contains(message, "SANDBOX_ID:") {
	35	+ parts := strings.Split(message, "SANDBOX_ID:")
	36	+ if len(parts) > 1 {
	37	+ newSandboxID := strings.TrimSpace(parts[1])
	38	+ if newSandboxID != "" {
	39	+ h.SetSandboxID(newSandboxID)
	40	+
	41	+ newCtxLogger := rootLogger.ReplaceField("sandboxId", newSandboxID)
	42	+ ulog.SetLogger(newCtxLogger)
	43	+
	44	+ logger = ulog.GetLogger()
	45	+ logger.Info("Updated sandbox_id from vsock", ulog.F("new_sandbox_id", newSandboxID))
	46	+
	47	+ if h.healthFunc() {
	48	+ response := "OK\nREADY:" + h.version + "\n"
	49	+ logger.Info("gRPC and network healthy, sent READY back with version", ulog.F("version", h.version))
	50	+ return response
	51	+ } else {
	52	+ logger.Error("gRPC or network not responding")
	53	+ return "NOT_READY\n"
	54	+ }
	55	+ }
	56	+ }
	57	+ }
	58	+ return ""
	59	+}
	60	+
	61	+func (h *VsockHandlerImpl) GetSandboxID() string {
	62	+ h.mu.Lock()
	63	+ defer h.mu.Unlock()
	64	+ return h.sandboxID
	65	+}
	66	+
	67	+func (h *VsockHandlerImpl) SetSandboxID(id string) {
	68	+ h.mu.Lock()
	69	+ defer h.mu.Unlock()
	70	+ h.sandboxID = id
	71	+}
	72	+
	73	+func checkGRPCHealth() bool {
	74	+ mu.Lock()
	75	+ defer mu.Unlock()
	76	+ return isSafe
	77	+}

Mconfig/config.yaml+5-0

	@@ -29,3 +29,8 @@ network:
29	containerd:	29	containerd:
30	socket: /run/containerd/containerd.sock	30	socket: /run/containerd/containerd.sock
31	default_namespace: default	31	default_namespace: default
		32	+
		33	+sandbox:
		34	+ vsock_signal_retry: 10ms
		35	+ vsock_signal_timeout: 60s
		36	+ request_timeout: 60s

Minternal/config/config.go+22-1

	@@ -4,6 +4,7 @@ import (
4	"fmt"	4	"fmt"
5	"os"	5	"os"
6	"path/filepath"	6	"path/filepath"
		7	+ "time"
7		8
8	"github.com/openeuler/Conch/pkg/ulog"	9	"github.com/openeuler/Conch/pkg/ulog"
9	"gopkg.in/yaml.v3"	10	"gopkg.in/yaml.v3"
	@@ -16,6 +17,7 @@ type Config struct {
16	Server ServerConfig `yaml:"server"`	17	Server ServerConfig `yaml:"server"`
17	Network NetworkConfig `yaml:"network"`	18	Network NetworkConfig `yaml:"network"`
18	Containerd ContainerdConfig `yaml:"containerd"`	19	Containerd ContainerdConfig `yaml:"containerd"`
		20	+ Sandbox SandboxConfig `yaml:"sandbox"`
19	}	21	}
20		22
21	// AppConfig holds application-specific configuration	23	// AppConfig holds application-specific configuration
	@@ -52,6 +54,12 @@ type ContainerdConfig struct {
52	DefaultNamespace string `yaml:"default_namespace"`	54	DefaultNamespace string `yaml:"default_namespace"`
53	}	55	}
54		56
		57	+type SandboxConfig struct {
		58	+ VsockSignalRetry time.Duration `yaml:"vsock_signal_retry"`
		59	+ VsockSignalTimeout time.Duration `yaml:"vsock_signal_timeout"`
		60	+ RequestTimeout time.Duration `yaml:"request_timeout"`
		61	+}
		62	+
55	// DefaultConfig returns the default configuration	63	// DefaultConfig returns the default configuration
56	func DefaultConfig() *Config {	64	func DefaultConfig() *Config {
57	defaultUnixSocket := "/var/run/conchd/conchd.sock"	65	defaultUnixSocket := "/var/run/conchd/conchd.sock"
	@@ -82,6 +90,11 @@ func DefaultConfig() *Config {
82	Socket: "/run/containerd/containerd.sock",	90	Socket: "/run/containerd/containerd.sock",
83	DefaultNamespace: "default",	91	DefaultNamespace: "default",
84	},	92	},
		93	+ Sandbox: SandboxConfig{
		94	+ VsockSignalRetry: 10 * time.Millisecond,
		95	+ VsockSignalTimeout: 60 * time.Second,
		96	+ RequestTimeout: 60 * time.Second,
		97	+ },
85	}	98	}
86	}	99	}
87		100
	@@ -149,7 +162,15 @@ func LoadConfig(configPath string) (*Config, error) {
149	if cfg.Containerd.DefaultNamespace == "" {	162	if cfg.Containerd.DefaultNamespace == "" {
150	cfg.Containerd.DefaultNamespace = defaultCfg.Containerd.DefaultNamespace	163	cfg.Containerd.DefaultNamespace = defaultCfg.Containerd.DefaultNamespace
151	}	164	}
152	-	165	+ if cfg.Sandbox.VsockSignalRetry == 0 {
		166	+ cfg.Sandbox.VsockSignalRetry = defaultCfg.Sandbox.VsockSignalRetry
		167	+ }
		168	+ if cfg.Sandbox.VsockSignalTimeout == 0 {
		169	+ cfg.Sandbox.VsockSignalTimeout = defaultCfg.Sandbox.VsockSignalTimeout
		170	+ }
		171	+ if cfg.Sandbox.RequestTimeout == 0 {
		172	+ cfg.Sandbox.RequestTimeout = defaultCfg.Sandbox.RequestTimeout
		173	+ }
153	return &cfg, nil	174	return &cfg, nil
154	}	175	}
155		176

Minternal/handler.go+4-1

	@@ -120,7 +120,7 @@ func NewServer(cfg config.Config) (Server, error) {
120	return nil, fmt.Errorf("failed to init network pool: %w", err)	120	return nil, fmt.Errorf("failed to init network pool: %w", err)
121	}	121	}
122		122
123	- s.SetSandboxManager(sandbox.NewManager(pool, daemonClient))	123	+ s.SetSandboxManager(sandbox.NewManager(pool, daemonClient, cfg.Sandbox.VsockSignalRetry, cfg.Sandbox.VsockSignalTimeout, cfg.Sandbox.RequestTimeout))
124	go pool.Populate(ctx)	124	go pool.Populate(ctx)
125		125
126	handleSignals(ctx, cancel, s)	126	handleSignals(ctx, cancel, s)
	@@ -217,6 +217,9 @@ func (s *Server) Cleanup() {
217	if err := m.CleanupPool(); err != nil {	217	if err := m.CleanupPool(); err != nil {
218	logger.Error("Server cleanup error", ulog.F("error", err))	218	logger.Error("Server cleanup error", ulog.F("error", err))
219	}	219	}
		220	+ if err := m.CleanupCIDMap(); err != nil {
		221	+ logger.Error("CID map cleanup error", ulog.F("error", err))
		222	+ }
220	}	223	}
221	snapshot.CleanupAllViews()	224	snapshot.CleanupAllViews()
222	if err := snapshot.Close(); err != nil {	225	if err := snapshot.Close(); err != nil {

Ainternal/sandbox/cid_allocator.go+294-0

@@ -0,0 +1,294 @@
	1	+package sandbox
	2	+
	3	+import (
	4	+ "encoding/json"
	5	+ "fmt"
	6	+ "os"
	7	+ "path/filepath"
	8	+ "sync"
	9	+
	10	+ "github.com/openeuler/Conch/pkg/ulog"
	11	+ "golang.org/x/sys/unix"
	12	+)
	13	+
	14	+const (
	15	+ CIDMapDir = "/var/run/conch/maptable"
	16	+ CIDMapFile = "cidmap.json"
	17	+ MinCID = 3
	18	+ MaxCID = uint32(4294967294) // uint32 max - 1, 预留一个避免边界问题
	19	+ CIDMapFilePerm = 0644
	20	+)
	21	+
	22	+type CIDMap struct {
	23	+ NextCID uint32 `json:"next_cid"`
	24	+ CidMap map[string]uint32 `json:"cid_map"` // sandboxId -> cid
	25	+}
	26	+
	27	+type CIDAllocator struct {
	28	+ mu sync.Mutex
	29	+ filePath string
	30	+}
	31	+
	32	+func NewCIDAllocator() *CIDAllocator {
	33	+ if err := os.MkdirAll(CIDMapDir, 0755); err != nil {
	34	+ ulog.Error("failed to create cidmap directory", ulog.F("error", err))
	35	+ }
	36	+
	37	+ filePath := filepath.Join(CIDMapDir, CIDMapFile)
	38	+ allocator := &CIDAllocator{
	39	+ filePath: filePath,
	40	+ }
	41	+
	42	+ if err := allocator.initFile(); err != nil {
	43	+ ulog.Error("failed to init cidmap file", ulog.F("error", err))
	44	+ }
	45	+
	46	+ return allocator
	47	+}
	48	+
	49	+func (a *CIDAllocator) initFile() error {
	50	+ if _, err := os.Stat(a.filePath); os.IsNotExist(err) {
	51	+ initialMap := CIDMap{
	52	+ NextCID: MinCID,
	53	+ CidMap: make(map[string]uint32),
	54	+ }
	55	+ return a.writeCIDMap(&initialMap)
	56	+ }
	57	+ return nil
	58	+}
	59	+
	60	+func (a CIDAllocator) readCIDMap() (CIDMap, error) {
	61	+ data, err := os.ReadFile(a.filePath)
	62	+ if err != nil {
	63	+ return nil, fmt.Errorf("failed to read cidmap file: %w", err)
	64	+ }
	65	+
	66	+ if len(data) == 0 {
	67	+ return &CIDMap{
	68	+ NextCID: MinCID,
	69	+ CidMap: make(map[string]uint32),
	70	+ }, nil
	71	+ }
	72	+
	73	+ var cidMap CIDMap
	74	+ if err := json.Unmarshal(data, &cidMap); err != nil {
	75	+ return nil, fmt.Errorf("failed to unmarshal cidmap: %w", err)
	76	+ }
	77	+
	78	+ if cidMap.CidMap == nil {
	79	+ cidMap.CidMap = make(map[string]uint32)
	80	+ }
	81	+
	82	+ return &cidMap, nil
	83	+}
	84	+
	85	+func (a CIDAllocator) writeCIDMap(cidMap CIDMap) error {
	86	+ data, err := json.MarshalIndent(cidMap, "", " ")
	87	+ if err != nil {
	88	+ return fmt.Errorf("failed to marshal cidmap: %w", err)
	89	+ }
	90	+
	91	+ return os.WriteFile(a.filePath, data, CIDMapFilePerm)
	92	+}
	93	+
	94	+func (a *CIDAllocator) acquireFileLock() (int, error) {
	95	+ fd, err := unix.Open(a.filePath, unix.O_RDWR\|unix.O_CREAT, CIDMapFilePerm)
	96	+ if err != nil {
	97	+ return -1, fmt.Errorf("failed to open cidmap file: %w", err)
	98	+ }
	99	+
	100	+ if err := unix.Flock(fd, unix.LOCK_EX); err != nil {
	101	+ unix.Close(fd)
	102	+ return -1, fmt.Errorf("failed to acquire exclusive lock: %w", err)
	103	+ }
	104	+
	105	+ return fd, nil
	106	+}
	107	+
	108	+func (a *CIDAllocator) releaseFileLock(fd int) {
	109	+ unix.Flock(fd, unix.LOCK_UN)
	110	+ unix.Close(fd)
	111	+}
	112	+
	113	+func (a *CIDAllocator) AllocateCID(sandboxId string) (uint32, error) {
	114	+ a.mu.Lock()
	115	+ defer a.mu.Unlock()
	116	+
	117	+ fd, err := a.acquireFileLock()
	118	+ if err != nil {
	119	+ return 0, err
	120	+ }
	121	+ defer a.releaseFileLock(fd)
	122	+
	123	+ cidMap, err := a.readCIDMapLocked(fd)
	124	+ if err != nil {
	125	+ return 0, err
	126	+ }
	127	+
	128	+ if existingCID, ok := cidMap.CidMap[sandboxId]; ok {
	129	+ return existingCID, nil
	130	+ }
	131	+
	132	+ if cidMap.NextCID > MaxCID {
	133	+ return 0, fmt.Errorf("CID allocation failed: reached maximum CID limit (%d). Please cleanup old sandboxes to release CIDs. Current active sandboxes: %d",
	134	+ MaxCID, len(cidMap.CidMap))
	135	+ }
	136	+
	137	+ cid := cidMap.NextCID
	138	+ cidMap.CidMap[sandboxId] = cid
	139	+ cidMap.NextCID = cid + 1
	140	+
	141	+ if err := a.writeCIDMapLocked(fd, cidMap); err != nil {
	142	+ return 0, err
	143	+ }
	144	+
	145	+ ulog.Info("allocated CID", ulog.F("sandbox_id", sandboxId), ulog.F("cid", cid))
	146	+ return cid, nil
	147	+}
	148	+
	149	+func (a CIDAllocator) readCIDMapLocked(fd int) (CIDMap, error) {
	150	+ if _, err := unix.Seek(fd, 0, 0); err != nil {
	151	+ return nil, fmt.Errorf("failed to seek to beginning: %w", err)
	152	+ }
	153	+
	154	+ var buf []byte
	155	+ for {
	156	+ chunk := make([]byte, 1024)
	157	+ n, err := unix.Read(fd, chunk)
	158	+ if err != nil {
	159	+ return nil, fmt.Errorf("failed to read cidmap: %w", err)
	160	+ }
	161	+ if n == 0 {
	162	+ break
	163	+ }
	164	+ buf = append(buf, chunk[:n]...)
	165	+ }
	166	+
	167	+ if len(buf) == 0 {
	168	+ return &CIDMap{
	169	+ NextCID: MinCID,
	170	+ CidMap: make(map[string]uint32),
	171	+ }, nil
	172	+ }
	173	+
	174	+ var cidMap CIDMap
	175	+ if err := json.Unmarshal(buf, &cidMap); err != nil {
	176	+ return nil, fmt.Errorf("failed to unmarshal cidmap: %w", err)
	177	+ }
	178	+
	179	+ if cidMap.CidMap == nil {
	180	+ cidMap.CidMap = make(map[string]uint32)
	181	+ }
	182	+
	183	+ return &cidMap, nil
	184	+}
	185	+
	186	+func (a CIDAllocator) writeCIDMapLocked(fd int, cidMap CIDMap) error {
	187	+ data, err := json.MarshalIndent(cidMap, "", " ")
	188	+ if err != nil {
	189	+ return fmt.Errorf("failed to marshal cidmap: %w", err)
	190	+ }
	191	+
	192	+ if _, err := unix.Seek(fd, 0, 0); err != nil {
	193	+ return fmt.Errorf("failed to seek: %w", err)
	194	+ }
	195	+
	196	+ if err := unix.Ftruncate(fd, 0); err != nil {
	197	+ return fmt.Errorf("failed to truncate: %w", err)
	198	+ }
	199	+
	200	+ if _, err := unix.Write(fd, data); err != nil {
	201	+ return fmt.Errorf("failed to write cidmap: %w", err)
	202	+ }
	203	+
	204	+ return nil
	205	+}
	206	+
	207	+func (a *CIDAllocator) ReleaseCID(sandboxId string) error {
	208	+ a.mu.Lock()
	209	+ defer a.mu.Unlock()
	210	+
	211	+ fd, err := a.acquireFileLock()
	212	+ if err != nil {
	213	+ return err
	214	+ }
	215	+ defer a.releaseFileLock(fd)
	216	+
	217	+ cidMap, err := a.readCIDMapLocked(fd)
	218	+ if err != nil {
	219	+ return err
	220	+ }
	221	+
	222	+ delete(cidMap.CidMap, sandboxId)
	223	+
	224	+ return a.writeCIDMapLocked(fd, cidMap)
	225	+}
	226	+
	227	+func (a *CIDAllocator) GetCID(sandboxId string) (uint32, bool) {
	228	+ a.mu.Lock()
	229	+ defer a.mu.Unlock()
	230	+
	231	+ fd, err := a.acquireFileLock()
	232	+ if err != nil {
	233	+ return 0, false
	234	+ }
	235	+ defer a.releaseFileLock(fd)
	236	+
	237	+ cidMap, err := a.readCIDMapLocked(fd)
	238	+ if err != nil {
	239	+ return 0, false
	240	+ }
	241	+
	242	+ cid, ok := cidMap.CidMap[sandboxId]
	243	+ return cid, ok
	244	+}
	245	+
	246	+func (a *CIDAllocator) GetActiveCount() int {
	247	+ a.mu.Lock()
	248	+ defer a.mu.Unlock()
	249	+
	250	+ fd, err := a.acquireFileLock()
	251	+ if err != nil {
	252	+ return 0
	253	+ }
	254	+ defer a.releaseFileLock(fd)
	255	+
	256	+ cidMap, err := a.readCIDMapLocked(fd)
	257	+ if err != nil {
	258	+ return 0
	259	+ }
	260	+
	261	+ return len(cidMap.CidMap)
	262	+}
	263	+
	264	+func (a *CIDAllocator) ListActiveSandboxes() []string {
	265	+ a.mu.Lock()
	266	+ defer a.mu.Unlock()
	267	+
	268	+ fd, err := a.acquireFileLock()
	269	+ if err != nil {
	270	+ return nil
	271	+ }
	272	+ defer a.releaseFileLock(fd)
	273	+
	274	+ cidMap, err := a.readCIDMapLocked(fd)
	275	+ if err != nil {
	276	+ return nil
	277	+ }
	278	+
	279	+ sandboxIds := make([]string, 0, len(cidMap.CidMap))
	280	+ for sandboxId := range cidMap.CidMap {
	281	+ sandboxIds = append(sandboxIds, sandboxId)
	282	+ }
	283	+ return sandboxIds
	284	+}
	285	+
	286	+func (a *CIDAllocator) Cleanup() error {
	287	+ a.mu.Lock()
	288	+ defer a.mu.Unlock()
	289	+
	290	+ if err := os.Remove(a.filePath); err != nil && !os.IsNotExist(err) {
	291	+ return fmt.Errorf("failed to remove cidmap file: %w", err)
	292	+ }
	293	+ return nil
	294	+}

Minternal/sandbox/manager.go+179-55

	@@ -4,28 +4,36 @@ import (
4	"context"	4	"context"
5	"fmt"	5	"fmt"
6	"net"	6	"net"
7	- "os"	7	+ "strings"
8	"sync"	8	"sync"
9	"syscall"	9	"syscall"
		10	+ "time"
10		11
11	"github.com/openeuler/Conch/internal/daemon"	12	"github.com/openeuler/Conch/internal/daemon"
12	"github.com/openeuler/Conch/internal/image"	13	"github.com/openeuler/Conch/internal/image"
13	"github.com/openeuler/Conch/internal/sandbox/network"	14	"github.com/openeuler/Conch/internal/sandbox/network"
14	"github.com/openeuler/Conch/internal/snapshot"	15	"github.com/openeuler/Conch/internal/snapshot"
15	- "github.com/openeuler/Conch/internal/snapshot/common"
16	"github.com/openeuler/Conch/pkg/ulog"	16	"github.com/openeuler/Conch/pkg/ulog"
17	)	17	)
18		18
19	type Manager struct {	19	type Manager struct {
20	- sandboxes sync.Map	20	+ sandboxes sync.Map
21	- pool *network.Pool	21	+ pool *network.Pool
22	- daemonClient *daemon.Client	22	+ daemonClient *daemon.Client
		23	+ vsockSignalRetry time.Duration
		24	+ vsockSignalTimeout time.Duration
		25	+ requestTimeout time.Duration
		26	+ cidAllocator *CIDAllocator
23	}	27	}
24		28
25	-func NewManager(p network.Pool, daemonClient daemon.Client) *Manager {	29	+func NewManager(p network.Pool, daemonClient daemon.Client, vsockSignalRetry, vsockSignalTimeout, requestTimeout time.Duration) *Manager {
26	return &Manager{	30	return &Manager{
27	- pool: p,	31	+ pool: p,
28	- daemonClient: daemonClient,	32	+ daemonClient: daemonClient,
		33	+ vsockSignalRetry: vsockSignalRetry,
		34	+ vsockSignalTimeout: vsockSignalTimeout,
		35	+ requestTimeout: requestTimeout,
		36	+ cidAllocator: NewCIDAllocator(),
29	}	37	}
30	}	38	}
31		39
	@@ -50,67 +58,153 @@ type SandboxPauseRequest struct {
50	}	58	}
51		59
52	const (	60	const (
53	- // vsockReadyPort is the vsock port used for agent readiness signaling.	61	+ vsockReadyPort = 4065
54	- // Agent connects to (CID=2, vsockReadyPort) after gRPC is ready.	62	+ expectedAgentVersion = "0.0.2"
55	- // cloud-hypervisor forwards this to unix socket: <vsockPath>_<vsockReadyPort>
56	- vsockReadyPort = 4065
57	)	63	)
58		64
59	func sandboxMapKey(namespace, sandboxID string) string {	65	func sandboxMapKey(namespace, sandboxID string) string {
60	return namespace + ":" + sandboxID	66	return namespace + ":" + sandboxID
61	}	67	}
62		68
63	-func createSandboxWithVsockReady(ctx context.Context, snapshotConf snapshot.SnapshotConfig, namespace, vmmName, sandboxId string, vcpuNum int64, pool network.Pool) (*Sandbox, error) {	69	+func createSandboxWithVsockSend(ctx context.Context, snapshotConf snapshot.SnapshotConfig, namespace, vmmName, sandboxId string, vcpuNum int64, pool network.Pool, vsockSignalRetry, vsockSignalTimeout time.Duration, resume bool, vsockCID uint32, vsockSocketPath string) (*Sandbox, error) {
64	logger := ulog.GetLogger()	70	logger := ulog.GetLogger()
65		71
66	- if err := os.MkdirAll(VsockSocketDir, 0755); err != nil {	72	+ var sbx *Sandbox
67	- return nil, fmt.Errorf("failed to create vsock socket directory: %w", err)	73	+ var createErr error
		74	+ if resume {
		75	+ sbx, createErr = ResumeSandbox(ctx, snapshotConf, namespace, vmmName, sandboxId, vcpuNum, pool, vsockCID, vsockSocketPath)
		76	+ } else {
		77	+ sbx, createErr = CreateSandbox(ctx, snapshotConf, namespace, vmmName, sandboxId, vcpuNum, pool, vsockCID, vsockSocketPath)
68	}	78	}
69	-	79	+ if createErr != nil {
70	- vsockReadyPath := SandboxVsockSocketPath(sandboxId) + fmt.Sprintf("_%d", vsockReadyPort)	80	+ return nil, fmt.Errorf("failed to create sandbox: %w", createErr)
71	- os.Remove(vsockReadyPath)
72	-
73	- logger.Info("creating vsock ready listener", ulog.F("path", vsockReadyPath), ulog.F("sandboxId", sandboxId))
74	- readyListener, listenErr := net.Listen("unix", vsockReadyPath)
75	- if listenErr != nil {
76	- return nil, fmt.Errorf("failed to create vsock ready listener: %w", listenErr)
77	- }
78	- defer readyListener.Close()
79	- defer os.Remove(vsockReadyPath)
80	-
81	- sbx, err := CreateSandbox(ctx, snapshotConf, namespace, vmmName, sandboxId, vcpuNum, pool)
82	- if err != nil {
83	- return nil, fmt.Errorf("failed to create sandbox: %w", err)
84	}	81	}
85		82
86	readyCh := make(chan struct{}, 1)	83	readyCh := make(chan struct{}, 1)
87	- go func() {	84	+ go waitForVsockAgentReady(ctx, sbx, sandboxId, vsockSocketPath, vsockSignalRetry, vsockSignalTimeout, readyCh)
88	- conn, acceptErr := readyListener.Accept()
89	- if acceptErr != nil {
90	- logger.Debug("vsock ready listener accept error (expected on cleanup)", ulog.F("error", acceptErr), ulog.F("sandboxId", sandboxId))
91	- return
92	- }
93	- logger.Info("vsock connection accepted from agent", ulog.F("sandboxId", sandboxId))
94	- conn.Write([]byte("ACK"))
95	- conn.Close()
96	- close(readyCh)
97	- }()
98		85
99	select {	86	select {
100	case <-readyCh:	87	case <-readyCh:
101	- logger.Info("agent ready signal received via vsock", ulog.F("sandboxId", sandboxId))	88	+ logger.Info("Vsock signal sent successfully", ulog.F("sandboxId", sandboxId))
102	case <-ctx.Done():	89	case <-ctx.Done():
103	- return nil, fmt.Errorf("timeout waiting for agent ready: %w", ctx.Err())	90	+ return sbx, ctx.Err()
104	}	91	}
105	-
106	return sbx, nil	92	return sbx, nil
107	}	93	}
108		94
		95	+func waitForVsockAgentReady(ctx context.Context, sbx *Sandbox, sandboxId, vsockSocketPath string, vsockSignalRetry, vsockSignalTimeout time.Duration, readyCh chan struct{}) {
		96	+ logger := ulog.GetLogger()
		97	+ payload := fmt.Sprintf("I AM SANDBOX_ID:%s\n", sandboxId)
		98	+
		99	+ timer := time.NewTimer(vsockSignalTimeout)
		100	+ defer timer.Stop()
		101	+
		102	+ for {
		103	+ select {
		104	+ case <-timer.C:
		105	+ logger.Error("vsock signal attempts timed out", ulog.F("sandboxId", sandboxId), ulog.F("timeout", vsockSignalTimeout))
		106	+ return
		107	+ case <-ctx.Done():
		108	+ return
		109	+ default:
		110	+ conn, err := net.Dial("unix", vsockSocketPath)
		111	+ if err != nil {
		112	+ logger.Debug("failed to connect to vsock socket, retrying...", ulog.F("sandboxId", sandboxId), ulog.F("error", err))
		113	+ time.Sleep(vsockSignalRetry)
		114	+ continue
		115	+ }
		116	+
		117	+ _, err = conn.Write([]byte("CONNECT 4065\n"))
		118	+ if err != nil {
		119	+ logger.Debug("failed to write CONNECT command, retrying...", ulog.F("sandboxId", sandboxId), ulog.F("error", err))
		120	+ conn.Close()
		121	+ time.Sleep(vsockSignalRetry)
		122	+ continue
		123	+ }
		124	+
		125	+ _, err = conn.Write([]byte(payload))
		126	+ if err != nil {
		127	+ logger.Warn("failed to send payload, retrying...", ulog.F("sandboxId", sandboxId), ulog.F("error", err))
		128	+ conn.Close()
		129	+ time.Sleep(vsockSignalRetry)
		130	+ continue
		131	+ }
		132	+
		133	+ logger.Debug("payload sent, waiting for OK receipt", ulog.F("sandboxId", sandboxId))
		134	+ conn.SetReadDeadline(time.Now().Add(1 * time.Second))
		135	+
		136	+ respBuf := make([]byte, 64)
		137	+ n, readErr := conn.Read(respBuf)
		138	+ if readErr != nil {
		139	+ logger.Debug("failed to read receipt, retrying...", ulog.F("sandboxId", sandboxId), ulog.F("error", readErr))
		140	+ conn.Close()
		141	+ time.Sleep(vsockSignalRetry)
		142	+ continue
		143	+ }
		144	+
		145	+ vmmMsg := string(respBuf[:n])
		146	+ if !strings.Contains(vmmMsg, "OK") {
		147	+ logger.Debug("Unexpected response from VMM proxy", ulog.F("msg", vmmMsg))
		148	+ conn.Close()
		149	+ time.Sleep(vsockSignalRetry)
		150	+ continue
		151	+ }
		152	+
		153	+ conn.SetReadDeadline(time.Now().Add(2 * time.Second))
		154	+
		155	+ readyBuf := make([]byte, 64)
		156	+ rn, rerr := conn.Read(readyBuf)
		157	+ if rerr != nil {
		158	+ logger.Debug("Waiting for Agent READY signal timed out", ulog.F("error", rerr))
		159	+ conn.Close()
		160	+ time.Sleep(vsockSignalRetry)
		161	+ continue
		162	+ }
		163	+
		164	+ agentMsg := string(readyBuf[:rn])
		165	+
		166	+ if strings.Contains(agentMsg, "NOT_READY") {
		167	+ logger.Error("Agent gRPC service not started", ulog.F("sandboxId", sandboxId))
		168	+ conn.Close()
		169	+ time.Sleep(vsockSignalRetry)
		170	+ continue
		171	+ }
		172	+
		173	+ if strings.Contains(agentMsg, "READY:") {
		174	+ parts := strings.SplitN(agentMsg, "READY:", 2)
		175	+ agentVersion := ""
		176	+ if len(parts) > 1 {
		177	+ agentVersion = strings.TrimSpace(parts[1])
		178	+ }
		179	+
		180	+ if agentVersion != expectedAgentVersion {
		181	+ logger.Warn("Received agent signal but version mismatch",
		182	+ ulog.F("sandboxId", sandboxId),
		183	+ ulog.F("agent_version", agentVersion),
		184	+ ulog.F("expected_version", expectedAgentVersion))
		185	+ } else {
		186	+ logger.Info("Sandbox Agent is officially READY!",
		187	+ ulog.F("sandboxId", sandboxId),
		188	+ ulog.F("agent_version", agentVersion))
		189	+ }
		190	+
		191	+ conn.SetReadDeadline(time.Time{})
		192	+ sbx.vsockConn = conn
		193	+ close(readyCh)
		194	+ return
		195	+ }
		196	+ logger.Warn("Received unknown message from Agent", ulog.F("msg", agentMsg))
		197	+ conn.Close()
		198	+ time.Sleep(vsockSignalRetry)
		199	+ }
		200	+ }
		201	+}
		202	+
109	func (m *Manager) Create(req SandboxCreateRequest) (string, error) {	203	func (m *Manager) Create(req SandboxCreateRequest) (string, error) {
110	logger := ulog.GetLogger()	204	logger := ulog.GetLogger()
111	logger.Debug("creating sandbox in manager")	205	logger.Debug("creating sandbox in manager")
112		206
113	- ctx, cancel := context.WithTimeoutCause(context.Background(), common.RequestTimeout, fmt.Errorf("request timed out"))	207	+ ctx, cancel := context.WithTimeoutCause(context.Background(), m.requestTimeout, fmt.Errorf("request timed out"))
114	defer cancel()	208	defer cancel()
115		209
116	var sbx *Sandbox	210	var sbx *Sandbox
	@@ -132,9 +226,18 @@ func (m *Manager) Create(req SandboxCreateRequest) (string, error) {
132		226
133	var snapshotConf *snapshot.SnapshotConfig	227	var snapshotConf *snapshot.SnapshotConfig
134		228
		229	+ vsockCID, err := m.AllocateUniqueCID(req.SandboxId)
		230	+ if err != nil {
		231	+ return "", fmt.Errorf("failed to create sandbox: CID allocation error: %v", err)
		232	+ }
		233	+ vsockSocketPath, err := SandboxVsockSocketPath(key)
		234	+ if err != nil {
		235	+ return "", fmt.Errorf("failed to create sandbox: vsock socket path error: %v", err)
		236	+ }
		237	+
135	if resume {	238	if resume {
136	logger.Debug("creating sandbox by snapshotId")	239	logger.Debug("creating sandbox by snapshotId")
137	- snapshotConf, err = snapshot.AcquireResumeWorkspace(context.Background(), namespace, key, parentIDs, memOpt)	240	+ snapshotConf, err = snapshot.AcquireResumeWorkspace(context.Background(), namespace, key, parentIDs, vsockCID, vsockSocketPath, memOpt)
138	} else {	241	} else {
139	logger.Debug("creating sandbox by image", ulog.F("imageName", req.ImageName))	242	logger.Debug("creating sandbox by image", ulog.F("imageName", req.ImageName))
140	snapshotConf, err = snapshot.Prepare(context.Background(), namespace, key, parentIDs, memOpt)	243	snapshotConf, err = snapshot.Prepare(context.Background(), namespace, key, parentIDs, memOpt)
	@@ -154,12 +257,12 @@ func (m *Manager) Create(req SandboxCreateRequest) (string, error) {
154	}	257	}
155	}()	258	}()
156		259
157	- if resume {	260	+ sbx, err = createSandboxWithVsockSend(ctx, snapshotConf, namespace, req.VmmName, req.SandboxId, req.VcpuNum, m.pool, m.vsockSignalRetry, m.vsockSignalTimeout, resume, vsockCID, vsockSocketPath)
158	- sbx, err = ResumeSandbox(ctx, snapshotConf, namespace, req.VmmName, req.SandboxId, req.VcpuNum, m.pool)	261	+
159	- } else {
160	- sbx, err = createSandboxWithVsockReady(ctx, snapshotConf, namespace, req.VmmName, req.SandboxId, req.VcpuNum, m.pool)
161	- }
162	if err != nil {	262	if err != nil {
		263	+ if releaseErr := m.ReleaseCID(req.SandboxId); releaseErr != nil {
		264	+ logger.Warn("failed to release CID on create failure", ulog.F("sandbox_id", req.SandboxId), ulog.F("error", releaseErr))
		265	+ }
163	return "", fmt.Errorf("failed to create sandbox: %w", err)	266	return "", fmt.Errorf("failed to create sandbox: %w", err)
164	}	267	}
165	peerIP = sbx.slot.VpeerIPString()	268	peerIP = sbx.slot.VpeerIPString()
	@@ -179,6 +282,10 @@ func (m *Manager) Create(req SandboxCreateRequest) (string, error) {
179		282
180	snapshot.Remove(context.Background(), sbx.namespace, req.SandboxId)	283	snapshot.Remove(context.Background(), sbx.namespace, req.SandboxId)
181		284
		285	+ if releaseErr := m.ReleaseCID(req.SandboxId); releaseErr != nil {
		286	+ logger.Warn("failed to release CID", ulog.F("sandbox_id", req.SandboxId), ulog.F("error", releaseErr))
		287	+ }
		288	+
182	m.sandboxes.Delete(mapKey)	289	m.sandboxes.Delete(mapKey)
183	}()	290	}()
184		291
	@@ -227,11 +334,12 @@ func (m *Manager) resolveNamespace(namespace string) string {
227	func (m *Manager) Delete(req SandboxDeleteRequest) error {	334	func (m *Manager) Delete(req SandboxDeleteRequest) error {
228	logger := ulog.GetLogger()	335	logger := ulog.GetLogger()
229		336
230	- ctx, cancel := context.WithTimeoutCause(context.Background(), common.RequestTimeout, fmt.Errorf("request timed out"))	337	+ ctx, cancel := context.WithTimeoutCause(context.Background(), m.requestTimeout, fmt.Errorf("request timed out"))
231	defer cancel()	338	defer cancel()
232		339
233	namespace := m.resolveNamespace(req.Namespace)	340	namespace := m.resolveNamespace(req.Namespace)
234	- sbxVal, exists := m.sandboxes.Load(sandboxMapKey(namespace, req.SandboxId))	341	+ mapKey := sandboxMapKey(namespace, req.SandboxId)
		342	+ sbxVal, exists := m.sandboxes.Load(mapKey)
235	if !exists {	343	if !exists {
236	return fmt.Errorf("sandbox %s not found", req.SandboxId)	344	return fmt.Errorf("sandbox %s not found", req.SandboxId)
237	}	345	}
	@@ -241,12 +349,16 @@ func (m *Manager) Delete(req SandboxDeleteRequest) error {
241	return fmt.Errorf("invalid sandbox type for %s", req.SandboxId)	349	return fmt.Errorf("invalid sandbox type for %s", req.SandboxId)
242	}	350	}
243		351
244	- m.sandboxes.Delete(sandboxMapKey(sbx.namespace, req.SandboxId))	352	+ m.sandboxes.Delete(mapKey)
245	go func() {	353	go func() {
246	err := sbx.Stop(ctx)	354	err := sbx.Stop(ctx)
247	if err != nil {	355	if err != nil {
248	logger.Error("sandbox stop error", ulog.F("sandboxId", req.SandboxId), ulog.F("error", err))	356	logger.Error("sandbox stop error", ulog.F("sandboxId", req.SandboxId), ulog.F("error", err))
249	}	357	}
		358	+
		359	+ if releaseErr := m.ReleaseCID(req.SandboxId); releaseErr != nil {
		360	+ logger.Warn("failed to release CID", ulog.F("sandbox_id", req.SandboxId), ulog.F("error", releaseErr))
		361	+ }
250	}()	362	}()
251	return nil	363	return nil
252	}	364	}
	@@ -254,7 +366,7 @@ func (m *Manager) Delete(req SandboxDeleteRequest) error {
254	func (m *Manager) Pause(req SandboxPauseRequest) (string, error) {	366	func (m *Manager) Pause(req SandboxPauseRequest) (string, error) {
255	logger := ulog.GetLogger()	367	logger := ulog.GetLogger()
256		368
257	- ctx, cancel := context.WithTimeoutCause(context.Background(), common.RequestTimeout, fmt.Errorf("request timed out"))	369	+ ctx, cancel := context.WithTimeoutCause(context.Background(), m.requestTimeout, fmt.Errorf("request timed out"))
258	defer cancel()	370	defer cancel()
259		371
260	namespace := m.resolveNamespace(req.Namespace)	372	namespace := m.resolveNamespace(req.Namespace)
	@@ -320,3 +432,15 @@ func (m *Manager) CleanupPool() error {
320		432
321	return nil	433	return nil
322	}	434	}
		435	+
		436	+func (m *Manager) AllocateUniqueCID(sandboxId string) (uint32, error) {
		437	+ return m.cidAllocator.AllocateCID(sandboxId)
		438	+}
		439	+
		440	+func (m *Manager) ReleaseCID(sandboxId string) error {
		441	+ return m.cidAllocator.ReleaseCID(sandboxId)
		442	+}
		443	+
		444	+func (m *Manager) CleanupCIDMap() error {
		445	+ return m.cidAllocator.Cleanup()
		446	+}

Minternal/sandbox/sandbox.go+30-15

	@@ -4,6 +4,8 @@ import (
4	"context"	4	"context"
5	"errors"	5	"errors"
6	"fmt"	6	"fmt"
		7	+ "net"
		8	+ "os"
7	"path/filepath"	9	"path/filepath"
8		10
9	"github.com/openeuler/Conch/internal/sandbox/network"	11	"github.com/openeuler/Conch/internal/sandbox/network"
	@@ -20,8 +22,12 @@ const (
20	)	22	)
21		23
22	// SandboxVsockSocketPath returns the vsock socket path for a sandbox.	24	// SandboxVsockSocketPath returns the vsock socket path for a sandbox.
23	-func SandboxVsockSocketPath(sandboxId string) string {	25	+func SandboxVsockSocketPath(sandboxId string) (string, error) {
24	- return filepath.Join(VsockSocketDir, fmt.Sprintf("conch-vmm-%s.vsock", sandboxId))	26	+ if err := os.MkdirAll(VsockSocketDir, 0755); err != nil {
		27	+ return "", fmt.Errorf("failed to create vsock socket directory: %w", err)
		28	+ }
		29	+
		30	+ return filepath.Join(VsockSocketDir, fmt.Sprintf("conch-vmm-%s.vsock", sandboxId)), nil
25	}	31	}
26		32
27	type Execution struct {	33	type Execution struct {
	@@ -34,12 +40,14 @@ type Sandbox struct {
34	snapshotConf *snapshot.SnapshotConfig	40	snapshotConf *snapshot.SnapshotConfig
35	namespace string	41	namespace string
36	slot *network.Slot	42	slot *network.Slot
		43	+ vsockConn net.Conn
37	}	44	}
38		45
39	func ResumeSandbox(	46	func ResumeSandbox(
40	ctx context.Context,	47	ctx context.Context,
41	snapshotConf *snapshot.SnapshotConfig,	48	snapshotConf *snapshot.SnapshotConfig,
42	namespace, vmmName, sandboxId string, vcpuNum int64, pool *network.Pool,	49	namespace, vmmName, sandboxId string, vcpuNum int64, pool *network.Pool,
		50	+ vsockCID uint32, vsockSocketPath string,
43	) (s *Sandbox, e error) {	51	) (s *Sandbox, e error) {
44	cleanup := NewCleanup()	52	cleanup := NewCleanup()
45	defer func() {	53	defer func() {
	@@ -65,16 +73,18 @@ func ResumeSandbox(
65	snapfilePath := snapshotConf.SnapDir()	73	snapfilePath := snapshotConf.SnapDir()
66		74
67	vmmResourceArgs := &vmm.ResourceArgs{	75	vmmResourceArgs := &vmm.ResourceArgs{
68	- CPUBoot: defaultCPUBoot,	76	+ CPUBoot: defaultCPUBoot,
69	- CPUMax: vcpuNum,	77	+ CPUMax: vcpuNum,
70	- MemorySize: snapshotConf.MemSize,	78	+ MemorySize: snapshotConf.MemSize,
71	- MemoryPath: snapshotConf.SnapshotMemFile(),	79	+ MemoryPath: snapshotConf.SnapshotMemFile(),
72	- NamespaceID: slot.NamespaceID(),	80	+ NamespaceID: slot.NamespaceID(),
73	- TapName: slot.TapName(),	81	+ TapName: slot.TapName(),
74	- KernelPath: snapshotConf.KernelFile(),	82	+ KernelPath: snapshotConf.KernelFile(),
75	- SnapfilePath: snapfilePath,	83	+ SnapfilePath: snapfilePath,
76	- InitrdPath: snapshotConf.InitrdFile(),	84	+ InitrdPath: snapshotConf.InitrdFile(),
77	- PmemPaths: snapshotConf.PmemFiles(),	85	+ PmemPaths: snapshotConf.PmemFiles(),
		86	+ VsockCID: vsockCID,
		87	+ VsockSocketPath: vsockSocketPath,
78	}	88	}
79		89
80	vmmHandle, vmmErr := vmm.NewProcess(	90	vmmHandle, vmmErr := vmm.NewProcess(
	@@ -98,7 +108,7 @@ func ResumeSandbox(
98	}	108	}
99		109
100	cleanup.Add(func(ctx context.Context) error {	110	cleanup.Add(func(ctx context.Context) error {
101	- filesErr := cleanupFiles(sbx.process.VmmSocketPath)	111	+ filesErr := cleanupFiles(sbx.process.VmmSocketPath, sbx.process.VsockSocketPath)
102	if filesErr != nil {	112	if filesErr != nil {
103	return fmt.Errorf("failed to cleanup files: %w", filesErr)	113	return fmt.Errorf("failed to cleanup files: %w", filesErr)
104	}	114	}
	@@ -117,6 +127,7 @@ func CreateSandbox(
117	ctx context.Context,	127	ctx context.Context,
118	snapshotConf *snapshot.SnapshotConfig,	128	snapshotConf *snapshot.SnapshotConfig,
119	namespace, vmmName, sandboxId string, vcpuNum int64, pool *network.Pool,	129	namespace, vmmName, sandboxId string, vcpuNum int64, pool *network.Pool,
		130	+ vsockCID uint32, vsockSocketPath string,
120	) (s *Sandbox, e error) {	131	) (s *Sandbox, e error) {
121		132
122	cleanup := NewCleanup()	133	cleanup := NewCleanup()
	@@ -150,8 +161,8 @@ func CreateSandbox(
150	KernelPath: snapshotConf.KernelFile(),	161	KernelPath: snapshotConf.KernelFile(),
151	InitrdPath: snapshotConf.InitrdFile(),	162	InitrdPath: snapshotConf.InitrdFile(),
152	PmemPaths: snapshotConf.PmemFiles(),	163	PmemPaths: snapshotConf.PmemFiles(),
153	- VsockCID: uint32(slot.Idx + vsockCIDOffset),	164	+ VsockCID: vsockCID,
154	- VsockSocketPath: SandboxVsockSocketPath(sandboxId),	165	+ VsockSocketPath: vsockSocketPath,
155	SandboxId: sandboxId,	166	SandboxId: sandboxId,
156	}	167	}
157		168
	@@ -206,6 +217,10 @@ func (s *Sandbox) Stop(ctx context.Context) error {
206	}	217	}
207		218
208	func (s *Sandbox) Close(ctx context.Context) error {	219	func (s *Sandbox) Close(ctx context.Context) error {
		220	+ if s.vsockConn != nil {
		221	+ s.vsockConn.Close()
		222	+ s.vsockConn = nil
		223	+ }
209	err := s.cleanup.Run(ctx)	224	err := s.cleanup.Run(ctx)
210	if err != nil {	225	if err != nil {
211	return fmt.Errorf("failed to cleanup sandbox: %w", err)	226	return fmt.Errorf("failed to cleanup sandbox: %w", err)

Minternal/snapshot/common/const.go+0-3

	@@ -1,7 +1,5 @@
1	package common	1	package common
2		2
3	-import "time"
4	-
5	const (	3	const (
6	DirMode = 0750	4	DirMode = 0750
7	FileMode = 0640	5	FileMode = 0640
	@@ -18,7 +16,6 @@ const (
18	PemSuffix = ".erofs"	16	PemSuffix = ".erofs"
19		17
20	ContainerdSock = "/run/containerd/containerd.sock"	18	ContainerdSock = "/run/containerd/containerd.sock"
21	- RequestTimeout = 60 * time.Second
22		19
23	SnapshotMountRootfs = "rootfs"	20	SnapshotMountRootfs = "rootfs"
24	SnapshotMountMem = "mem"	21	SnapshotMountMem = "mem"

Minternal/snapshot/config_updater.go+14-9

	@@ -11,7 +11,7 @@ import (
11	type configUpdater struct{}	11	type configUpdater struct{}
12		12
13	// updateSnapshotConfig updates the snapshot configuration file with new paths.	13	// updateSnapshotConfig updates the snapshot configuration file with new paths.
14	-func (cu *configUpdater) updateSnapshotConfig(configFilePath, kernelPath, initrdPath, memoryPath string, pmemPaths []string) error {	14	+func (cu *configUpdater) updateSnapshotConfig(configFilePath, kernelPath, initrdPath, memoryPath string, pmemPaths []string, cid uint32, socketPath string) error {
15	data, err := os.ReadFile(configFilePath)	15	data, err := os.ReadFile(configFilePath)
16	if err != nil {	16	if err != nil {
17	return fmt.Errorf("error open snapshot config file %s : %w", configFilePath, err)	17	return fmt.Errorf("error open snapshot config file %s : %w", configFilePath, err)
	@@ -25,7 +25,7 @@ func (cu *configUpdater) updateSnapshotConfig(configFilePath, kernelPath, initrd
25	cu.updatePayloadPaths(config, kernelPath, initrdPath)	25	cu.updatePayloadPaths(config, kernelPath, initrdPath)
26	cu.updateMemoryZone(config, memoryPath)	26	cu.updateMemoryZone(config, memoryPath)
27	cu.updatePmemDevices(config, pmemPaths)	27	cu.updatePmemDevices(config, pmemPaths)
28	- cu.removeVsockConfig(config)	28	+ cu.updateVsockConfig(config, cid, socketPath)
29	updatedData, err := json.MarshalIndent(config, "", " ")	29	updatedData, err := json.MarshalIndent(config, "", " ")
30	if err != nil {	30	if err != nil {
31	return fmt.Errorf("error marshal config: %w", err)	31	return fmt.Errorf("error marshal config: %w", err)
	@@ -89,10 +89,15 @@ func (cu *configUpdater) updatePmemDevices(config map[string]interface{}, pmemPa
89	config["pmem"] = pmemArray	89	config["pmem"] = pmemArray
90	}	90	}
91		91
92	-// removeVsockConfig sets the vsock field to nil (null in JSON).	92	+// updateVsockConfig updates the vsock configuration with new cid and socket path.
93	-func (cu *configUpdater) removeVsockConfig(config map[string]interface{}) {	93	+func (cu *configUpdater) updateVsockConfig(config map[string]interface{}, cid uint32, socketPath string) {
94	- if _, ok := config["vsock"]; ok {	94	+ if cid == 0 \|\| socketPath == "" {
95	- slog.Info("Removing vsock configuration from config")	95	+ return
96	- config["vsock"] = nil	96	+ }
97	- }	97	+ vsock, ok := config["vsock"].(map[string]interface{})
98	-}	98	+ if !ok {
		99	+ return
		100	+ }
		101	+ vsock["cid"] = cid
		102	+ vsock["socket"] = socketPath
		103	+}

Minternal/snapshot/server.go+5-1

	@@ -275,6 +275,8 @@ func (s *server) AcquireResumeWorkspace(
275	ctx context.Context,	275	ctx context.Context,
276	namespace, key string,	276	namespace, key string,
277	parents ParentSnapshotIDs,	277	parents ParentSnapshotIDs,
		278	+ cid uint32,
		279	+ socketPath string,
278	opts ...Opt,	280	opts ...Opt,
279	) (_ *SnapshotConfig, err error) {	281	) (_ *SnapshotConfig, err error) {
280	memKey := getMemKeyFromRootfs(key)	282	memKey := getMemKeyFromRootfs(key)
	@@ -353,6 +355,8 @@ func (s *server) AcquireResumeWorkspace(
353	conf.InitrdFile(),	355	conf.InitrdFile(),
354	conf.SnapshotMemFile(),	356	conf.SnapshotMemFile(),
355	conf.PmemFiles(),	357	conf.PmemFiles(),
		358	+ cid,
		359	+ socketPath,
356	); err != nil {	360	); err != nil {
357	return nil, fmt.Errorf("update snapshot config failed: %v", err)	361	return nil, fmt.Errorf("update snapshot config failed: %v", err)
358	}	362	}
	@@ -431,7 +435,7 @@ func (s *server) Commit(ctx context.Context, namespace, snapshotID, key string,
431		435
432	configUpdater := &configUpdater{}	436	configUpdater := &configUpdater{}
433	configFilePath := filepath.Join(conf.SnapDir(), common.SnapshotConfigFileName)	437	configFilePath := filepath.Join(conf.SnapDir(), common.SnapshotConfigFileName)
434	- if err := configUpdater.updateSnapshotConfig(configFilePath, viewConf.KernelFile(), viewConf.InitrdFile(), viewConf.SnapshotMemFile(), viewConf.PmemFiles()); err != nil {	438	+ if err := configUpdater.updateSnapshotConfig(configFilePath, viewConf.KernelFile(), viewConf.InitrdFile(), viewConf.SnapshotMemFile(), viewConf.PmemFiles(),0,""); err != nil {
435	return fmt.Errorf("update snapshot config failed: %v", err)	439	return fmt.Errorf("update snapshot config failed: %v", err)
436	}	440	}
437		441

Minternal/snapshot/snapshot.go+3-2

	@@ -109,13 +109,14 @@ func AcquireView(ctx context.Context, namespace, key string, parents ParentSnaps
109	// AcquireResumeWorkspace prepares a snapshot-based restore workspace.	109	// AcquireResumeWorkspace prepares a snapshot-based restore workspace.
110	// Rootfs and VM are mounted as shared views, while mem is mounted as an active	110	// Rootfs and VM are mounted as shared views, while mem is mounted as an active
111	// layer so the snapshot config can be updated before restore.	111	// layer so the snapshot config can be updated before restore.
112	-func AcquireResumeWorkspace(ctx context.Context, namespace, key string, parents ParentSnapshotIDs, opts ...Opt) (*SnapshotConfig, error) {	112	+func AcquireResumeWorkspace(ctx context.Context, namespace, key string, parents ParentSnapshotIDs, cid uint32, socketPath string, opts ...Opt) (*SnapshotConfig, error) {
113	if gServer.snt == nil {	113	if gServer.snt == nil {
114	return nil, fmt.Errorf("server not init")	114	return nil, fmt.Errorf("server not init")
115	}	115	}
116	- return gServer.AcquireResumeWorkspace(ctx, namespace, key, parents, opts...)	116	+ return gServer.AcquireResumeWorkspace(ctx, namespace, key, parents, cid, socketPath, opts...)
117	}	117	}
118		118
		119	+
119	// ResolveParentSnapshotIDs resolves parent mem/vm snapshots from rootfs snapshot.	120	// ResolveParentSnapshotIDs resolves parent mem/vm snapshots from rootfs snapshot.
120	func ResolveParentSnapshotIDs(namespace, rootfs string) (ParentSnapshotIDs, error) {	121	func ResolveParentSnapshotIDs(namespace, rootfs string) (ParentSnapshotIDs, error) {
121	if gServer.snt == nil {	122	if gServer.snt == nil {

Mpkg/ulog/ulog.go+63-17

	@@ -60,6 +60,7 @@ type Logger interface {
60	Fatal(msg string, fields ...Field)	60	Fatal(msg string, fields ...Field)
61	With(fields ...Field) Logger	61	With(fields ...Field) Logger
62	WithContext(ctx context.Context) Logger	62	WithContext(ctx context.Context) Logger
		63	+ ReplaceField(key string, value interface{}) Logger
63	}	64	}
64		65
65	// Field represents a key-value pair for structured logging	66	// Field represents a key-value pair for structured logging
	@@ -81,6 +82,7 @@ type ulog struct {
81	filePath string	82	filePath string
82	writer writer	83	writer writer
83	fields []Field	84	fields []Field
		85	+ sandboxId string
84	maxFileSize int64	86	maxFileSize int64
85	rotation int	87	rotation int
86	}	88	}
	@@ -372,6 +374,11 @@ func WithContext(ctx context.Context) Logger {
372	return GetLogger().WithContext(ctx)	374	return GetLogger().WithContext(ctx)
373	}	375	}
374		376
		377	+// ReplaceField returns a new logger with the specified field replaced
		378	+func ReplaceField(key string, value interface{}) Logger {
		379	+ return GetLogger().ReplaceField(key, value)
		380	+}
		381	+
375	// log formats and writes a log entry	382	// log formats and writes a log entry
376	func (l *ulog) log(level LogLevel, msg string, fields ...Field) {	383	func (l *ulog) log(level LogLevel, msg string, fields ...Field) {
377	if l.level > level {	384	if l.level > level {
	@@ -400,13 +407,10 @@ func (l *ulog) log(level LogLevel, msg string, fields ...Field) {
400	}	407	}
401		408
402	// SandboxId (as a fixed part if present)	409	// SandboxId (as a fixed part if present)
403	- for _, f := range l.fields {	410	+ if l.sandboxId != "" {
404	- if f.Key == "sandboxId" {	411	+ b.WriteString("[")
405	- b.WriteString("[")	412	+ b.WriteString(l.sandboxId)
406	- b.WriteString(fmt.Sprintf("%v", f.Value))	413	+ b.WriteString("] ")
407	- b.WriteString("] ")
408	- break
409	- }
410	}	414	}
411		415
412	// Message	416	// Message
	@@ -417,6 +421,7 @@ func (l *ulog) log(level LogLevel, msg string, fields ...Field) {
417	allFields := make([]Field, 0, len(fields)+len(l.fields))	421	allFields := make([]Field, 0, len(fields)+len(l.fields))
418	allFields = append(allFields, l.fields...)	422	allFields = append(allFields, l.fields...)
419	allFields = append(allFields, fields...)	423	allFields = append(allFields, fields...)
		424	+
420	b.WriteString(" ")	425	b.WriteString(" ")
421	for i, f := range allFields {	426	for i, f := range allFields {
422	if i > 0 {	427	if i > 0 {
	@@ -511,23 +516,64 @@ func (l *ulog) Fatal(msg string, fields ...Field) {
511	// With returns a new logger with additional fields	516	// With returns a new logger with additional fields
512	func (l *ulog) With(fields ...Field) Logger {	517	func (l *ulog) With(fields ...Field) Logger {
513	newLogger := &ulog{	518	newLogger := &ulog{
514	- level: l.level,	519	+ level: l.level,
515	- output: l.output,	520	+ output: l.output,
516	- filePath: l.filePath,	521	+ filePath: l.filePath,
517	- writer: l.writer,	522	+ writer: l.writer,
		523	+ sandboxId: l.sandboxId,
		524	+ }
		525	+ newLogger.fields = make([]Field, 0, len(l.fields)+len(fields))
		526	+ newLogger.fields = append(newLogger.fields, l.fields...)
		527	+ for _, f := range fields {
		528	+ if f.Key == "sandboxId" {
		529	+ newLogger.sandboxId = fmt.Sprintf("%v", f.Value)
		530	+ } else {
		531	+ newLogger.fields = append(newLogger.fields, f)
		532	+ }
		533	+ }
		534	+ return newLogger
		535	+}
		536	+
		537	+// ReplaceField returns a new logger with the specified field replaced.
		538	+// If the field key doesn't exist, it adds the field.
		539	+func (l *ulog) ReplaceField(key string, value interface{}) Logger {
		540	+ newLogger := &ulog{
		541	+ level: l.level,
		542	+ output: l.output,
		543	+ filePath: l.filePath,
		544	+ writer: l.writer,
		545	+ sandboxId: l.sandboxId,
		546	+ }
		547	+ if key == "sandboxId" {
		548	+ newLogger.sandboxId = fmt.Sprintf("%v", value)
		549	+ newLogger.fields = append([]Field{}, l.fields...)
		550	+ return newLogger
		551	+ }
		552	+
		553	+ newLogger.fields = make([]Field, 0, len(l.fields))
		554	+ found := false
		555	+ for _, f := range l.fields {
		556	+ if f.Key == key {
		557	+ newLogger.fields = append(newLogger.fields, F(key, value))
		558	+ found = true
		559	+ } else {
		560	+ newLogger.fields = append(newLogger.fields, f)
		561	+ }
		562	+ }
		563	+ if !found {
		564	+ newLogger.fields = append(newLogger.fields, F(key, value))
518	}	565	}
519	- newLogger.fields = append([]Field{}, l.fields...)
520	- newLogger.fields = append(newLogger.fields, fields...)
521	return newLogger	566	return newLogger
522	}	567	}
523		568
524	// WithContext returns a new logger with context fields	569	// WithContext returns a new logger with context fields
525	func (l *ulog) WithContext(ctx context.Context) Logger {	570	func (l *ulog) WithContext(ctx context.Context) Logger {
526	newLogger := &ulog{	571	newLogger := &ulog{
527	- level: l.level,	572	+ level: l.level,
528	- output: l.output,	573	+ output: l.output,
529	- filePath: l.filePath,	574	+ filePath: l.filePath,
530	- writer: l.writer,	575	+ writer: l.writer,
		576	+ sandboxId: l.sandboxId,
531	}	577	}
532	newLogger.fields = append([]Field{}, l.fields...)	578	newLogger.fields = append([]Field{}, l.fields...)
533		579