Star9
29
代码
代码
Issues4
Pull Requests7
项目讨论
Wiki
分析
Star9
29
已合并
feat(webhook): add sandbox lifecycle notifications #174
hu-zhangying创建于 18 天前
feat(webhook): add sandbox lifecycle notifications #174
已合并
hu-zhangying创建于 18 天前
从hu-zhangying/Conchdev合入到openeuler/Conchdev
共 23 个文件变更+987-115
Mdocs/user/template.md+15-0
@@ -23,6 +23,21 @@ conch template create \
23 --initrd /var/lib/conch/conch.initrd23 --initrd /var/lib/conch/conch.initrd
24```24```
25 25 
26+### CLI API 请求超时
27+ 
28+所有通过 `conch` CLI 调用 conchd API 的请求共用 `CONCH_API_TIMEOUT` 环境变量。未设置时默认超时为 2 分钟；设置为正的 Go duration（例如 `30m`）后，当前命令中的所有 CLI 到 conchd API 请求都会使用该值。这个设置适用于 `template create`、`template pull`、`template push` 及其他使用 conchd API 的 CLI 命令。
29+ 
30+例如，首次拉取或转换较大镜像时可执行：
31+ 
32+```bash
33+CONCH_API_TIMEOUT=30m conch template create \
34+ --source hub.oepkgs.net/openeuler/python:latest \
35+ --kernel /var/lib/conch/kernel \
36+ --initrd /var/lib/conch/conch.initrd
37+```
38+ 
39+`conch template push --timeout <duration>` 是 push 命令的局部覆盖，优先于 `CONCH_API_TIMEOUT`；它仅影响该次 push 请求。
40+ 
26创建、拉取和 checkpoint 都会建立由 digest 唯一派生的内部 canonical image record，例如41创建、拉取和 checkpoint 都会建立由 digest 唯一派生的内部 canonical image record，例如
27`localhost/conch/template:sha256-1111...`。这个 record 是 containerd GC 的引用根，不需要用户命名。42`localhost/conch/template:sha256-1111...`。这个 record 是 containerd GC 的引用根，不需要用户命名。
28该本地命名空间由 Template 生命周期独占，pull 操作不允许把它作为远端输入，普通 `conch image rm` 也不能删除 canonical record。43该本地命名空间由 Template 生命周期独占，pull 操作不允许把它作为远端输入，普通 `conch image rm` 也不能删除 canonical record。
Adocs/user/webhooks.md+125-0
@@ -0,0 +1,125 @@
1+# Sandbox 生命周期 Webhook
2+ 
3+conchd 可在 Sandbox 创建、主动删除或非预期退出后，将生命周期事件异步发送到已注册的 HTTP/HTTPS 回调地址。Webhook 配置仅保存在当前 conchd 进程内存中；conchd 重启后必须重新注册。
4+ 
5+## 1. 注册 Webhook
6+ 
7+conchd 默认通过 Unix socket 提供 API。以下示例中的 socket 路径请替换为 `server.work_dir/conchd.sock`。
8+ 
9+```bash
10+curl --unix-socket /var/run/conch/conchd.sock \
11+ -X POST http://localhost/api/v1/events/webhooks \
12+ -H 'Content-Type: application/json' \
13+ -d '{
14+ "name": "reliability-service",
15+ "url": "https://reliability.example.com/conch/events",
16+ "events": [
17+ "sandbox.lifecycle.created",
18+ "sandbox.lifecycle.killed"
19+ ]
20+ }'
21+```
22+ 
23+`name` 和 `url` 必填。`url` 必须是 HTTP 或 HTTPS 地址；`events` 省略时订阅全部当前支持的事件。
24+ 
25+成功时返回 `201 Created`：
26+ 
27+```json
28+{
29+ "webhook_id": "wh_0123456789abcdef0123456789abcdef",
30+ "name": "reliability-service",
31+ "url": "https://reliability.example.com/conch/events",
32+ "events": ["sandbox.lifecycle.created", "sandbox.lifecycle.killed"],
33+ "createdAt": "2026-08-21T10:20:30Z"
34+}
35+```
36+ 
37+## 2. 查询和删除
38+ 
39+查询当前 conchd 实例注册的全部 Webhook：
40+ 
41+```bash
42+curl --unix-socket /var/run/conch/conchd.sock \
43+ http://localhost/api/v1/events/webhooks
44+```
45+ 
46+成功时返回 `200 OK`：
47+ 
48+```json
49+{
50+ "webhooks": [
51+ {
52+ "webhook_id": "wh_0123456789abcdef0123456789abcdef",
53+ "name": "reliability-service",
54+ "url": "https://reliability.example.com/conch/events",
55+ "events": ["sandbox.lifecycle.created", "sandbox.lifecycle.killed"],
56+ "createdAt": "2026-08-21T10:20:30Z"
57+ }
58+ ]
59+}
60+```
61+ 
62+删除一个 Webhook：
63+ 
64+```bash
65+curl --unix-socket /var/run/conch/conchd.sock \
66+ -X DELETE \
67+ http://localhost/api/v1/events/webhooks/wh_0123456789abcdef0123456789abcdef
68+```
69+ 
70+删除成功时返回 `200 OK`：
71+ 
72+```json
73+{
74+ "webhook_id": "wh_0123456789abcdef0123456789abcdef",
75+ "status": "deleted"
76+}
77+```
78+ 
79+删除返回成功后，conchd 不会再为该 `webhook_id` 创建新的投递任务；已经开始的投递可以继续完成。
80+ 
81+## 3. 事件载荷
82+ 
83+回调地址会收到 `POST` 请求和 JSON 请求体：
84+ 
85+```json
86+{
87+ "event_id": "evt_0123456789abcdef0123456789abcdef",
88+ "version": "v1",
89+ "type": "sandbox.lifecycle.created",
90+ "timestamp": "2026-08-21T10:20:30Z",
91+ "sandbox_id": "sandbox-001",
92+ "event_data": {
93+ "execution": {
94+ "created_at": "2026-08-21T10:20:30Z",
95+ "vcpu_num": 2,
96+ "ram_mb": 512
97+ }
98+ }
99+}
100+```
101+ 
102+`sandbox.lifecycle.killed` 在通用的 `event_data.execution` 之外，还包含 `event_data.kill_reason`：主动删除为 `request`，Sandbox 非预期退出为 `orphaned`。
103+ 
104+支持的事件与发送时机如下：
105+ 
106+| 事件类型 | 发送时机 |
107+| --- | --- |
108+| `sandbox.lifecycle.created` | Sandbox 已创建成功，且状态已持久化为 `READY` 后。 |
109+| `sandbox.lifecycle.killed` | 主动删除完成后，`kill_reason` 为 `request`。 |
110+| `sandbox.lifecycle.killed` | Sandbox 非预期退出且状态已持久化为 `UNKNOWN` 后，`kill_reason` 为 `orphaned`。 |
111+ 
112+## 4. 请求头、重试与幂等
113+ 
114+每次投递包含以下请求头：
115+ 
116+| 请求头 | 说明 |
117+| --- | --- |
118+| `Content-Type: application/json` | 事件正文的媒体类型。 |
119+| `conch-webhook-id` | 触发本次投递的 `webhook_id`。 |
120+ 
121+一次逻辑事件的所有重试使用相同的 `event_id`。接收端应以 `event_id` 去重并实现幂等处理。
122+ 
123+conchd 对每个匹配 Webhook 最多尝试投递 3 次。任意 2xx 响应均视为成功；网络错误、超时或非 2xx 响应均视为失败。投递为异步操作，不阻塞 Sandbox 生命周期操作。三次均失败时 conchd 记录错误日志，但不会保存事件或投递任务。
124+ 
125+第一阶段不提供回调请求签名、投递 ID 或事件持久化。请将回调端部署在受控网络中，并保护 conchd 的 Unix socket 访问权限。
Minternal/conchruntime/service.go+71-22
@@ -2,8 +2,6 @@ package conchruntime
2 2 
3import (3import (
4 "context"4 "context"
5- "crypto/rand"
6- "encoding/hex"
7 "errors"5 "errors"
8 "fmt"6 "fmt"
9 "os"7 "os"
@@ -18,13 +16,14 @@ import (
18 agentprotocol "github.com/openeuler/Conch/internal/agent/protocol"16 agentprotocol "github.com/openeuler/Conch/internal/agent/protocol"
19 "github.com/openeuler/Conch/internal/apperror"17 "github.com/openeuler/Conch/internal/apperror"
20 "github.com/openeuler/Conch/internal/daemon/state"18 "github.com/openeuler/Conch/internal/daemon/state"
19+ "github.com/openeuler/Conch/internal/id"
21 conchimage "github.com/openeuler/Conch/internal/image"20 conchimage "github.com/openeuler/Conch/internal/image"
22 "github.com/openeuler/Conch/internal/image/erofsconvert"21 "github.com/openeuler/Conch/internal/image/erofsconvert"
23 "github.com/openeuler/Conch/internal/netstack"22 "github.com/openeuler/Conch/internal/netstack"
24 "github.com/openeuler/Conch/internal/runtimeapi"23 "github.com/openeuler/Conch/internal/runtimeapi"
25 "github.com/openeuler/Conch/internal/sandbox"24 "github.com/openeuler/Conch/internal/sandbox"
26- "github.com/openeuler/Conch/internal/sandboxid"
27 conchtemplate "github.com/openeuler/Conch/internal/template"25 conchtemplate "github.com/openeuler/Conch/internal/template"
26+ "github.com/openeuler/Conch/internal/webhook"
28 "github.com/openeuler/Conch/pkg/ulog"27 "github.com/openeuler/Conch/pkg/ulog"
29)28)
30 29 
@@ -44,13 +43,14 @@ type SnapshotOps interface {
44}43}
45 44 
46type Service struct {45type Service struct {
47- Sandbox SandboxOps46+ Sandbox SandboxOps
48- Containerd *containerdclient.Client47+ Containerd *containerdclient.Client
49- Snapshot SnapshotOps48+ Snapshot SnapshotOps
50- Store state.Store49+ Store state.Store
51- Templates conchtemplate.Store50+ Templates conchtemplate.Store
52- SandboxDefaults SandboxDefaults51+ SandboxDefaults SandboxDefaults
53- lifecycleLocks sandboxLifecycleLocks52+ WebhookDispatcher *webhook.Dispatcher
53+ lifecycleLocks sandboxLifecycleLocks
54}54}
55 55 
56type sandboxLifecycleLock struct {56type sandboxLifecycleLock struct {
@@ -113,13 +113,13 @@ func (s *Service) CreateSandbox(ctx context.Context, opts SandboxCreateOptions)
113 }113 }
114 opts.SandboxID = strings.TrimSpace(opts.SandboxID)114 opts.SandboxID = strings.TrimSpace(opts.SandboxID)
115 if opts.SandboxID == "" {115 if opts.SandboxID == "" {
116- id, err := NewID()116+ id, err := id.New()
117 if err != nil {117 if err != nil {
118 return SandboxCreateResult{}, err118 return SandboxCreateResult{}, err
119 }119 }
120 opts.SandboxID = id120 opts.SandboxID = id
121 } else {121 } else {
122- if err := sandboxid.Validate(opts.SandboxID); err != nil {122+ if err := id.Validate(opts.SandboxID); err != nil {
123 return SandboxCreateResult{}, sandbox.ErrInvalidArgument.Wrap(123 return SandboxCreateResult{}, sandbox.ErrInvalidArgument.Wrap(
124 fmt.Errorf("invalid sandbox_id: %w", err),124 fmt.Errorf("invalid sandbox_id: %w", err),
125 )125 )
@@ -222,6 +222,7 @@ func (s *Service) CreateSandbox(ctx context.Context, opts SandboxCreateOptions)
222 deleteErr,222 deleteErr,
223 )223 )
224 }224 }
225+ s.publishLifecycleEvent(webhook.EventSandboxCreated, rec, "")
225 return SandboxCreateResult{226 return SandboxCreateResult{
226 SandboxID: opts.SandboxID,227 SandboxID: opts.SandboxID,
227 IP: createResult.IP,228 IP: createResult.IP,
@@ -323,6 +324,14 @@ func (s *Service) RemoveSandbox(ctx context.Context, sandboxID string) error {
323 }324 }
324 unlock := s.lifecycleLocks.lock(sandboxID)325 unlock := s.lifecycleLocks.lock(sandboxID)
325 defer unlock()326 defer unlock()
327+ var rec state.SandboxRecord
328+ if s.Store != nil {
329+ var getErr error
330+ rec, getErr = s.getSandbox(ctx, sandboxID)
331+ if getErr != nil && !errors.Is(getErr, state.ErrNotFound) {
332+ return getErr
333+ }
334+ }
326 err := s.Sandbox.Delete(sandbox.DeleteRequest{SandboxID: sandboxID})335 err := s.Sandbox.Delete(sandbox.DeleteRequest{SandboxID: sandboxID})
327 if err != nil && errors.Is(err, sandbox.ErrNotFound) {336 if err != nil && errors.Is(err, sandbox.ErrNotFound) {
328 err = nil337 err = nil
@@ -331,11 +340,59 @@ func (s *Service) RemoveSandbox(ctx context.Context, sandboxID string) error {
331 return err340 return err
332 }341 }
333 if s.Store != nil {342 if s.Store != nil {
334- return s.Store.DeleteSandbox(ctx, sandboxID)343+ if err := s.Store.DeleteSandbox(ctx, sandboxID); err != nil {
344+ return err
345+ }
346+ }
347+ if rec.SandboxID != "" {
348+ s.publishLifecycleEvent(webhook.EventSandboxKilled, rec, "request")
335 }349 }
336 return nil350 return nil
337}351}
338 352 
353+// HandleSandboxUnexpectedExit records the loss of a sandbox and emits its lifecycle event.
354+// It is called by sandbox.Manager after the runtime resources have been cleaned up.
355+func (s *Service) HandleSandboxUnexpectedExit(sandboxID string) {
356+ if s == nil || s.Store == nil {
357+ return
358+ }
359+ unlock := s.lifecycleLocks.lock(sandboxID)
360+ defer unlock()
361+ rec, err := s.getSandbox(context.Background(), sandboxID)
362+ if errors.Is(err, state.ErrNotFound) {
363+ return
364+ }
365+ if err != nil {
366+ ulog.GetLogger().Error("failed to read sandbox after unexpected exit", ulog.F("sandbox_id", sandboxID), ulog.F("error", err))
367+ return
368+ }
369+ if rec.State == state.SandboxUnknown {
370+ return
371+ }
372+ rec.State = state.SandboxUnknown
373+ if err := s.upsertSandbox(context.Background(), rec); err != nil {
374+ ulog.GetLogger().Error("failed to persist sandbox after unexpected exit", ulog.F("sandbox_id", sandboxID), ulog.F("error", err))
375+ return
376+ }
377+ s.publishLifecycleEvent(webhook.EventSandboxKilled, rec, "orphaned")
378+}
379+ 
380+func (s *Service) publishLifecycleEvent(eventType string, rec state.SandboxRecord, killReason string) {
381+ if s == nil || s.WebhookDispatcher == nil {
382+ return
383+ }
384+ event, err := webhook.NewEvent(eventType, rec.SandboxID, killReason, webhook.Execution{
385+ CreatedAt: time.Unix(0, rec.CreatedAt).UTC().Format(time.RFC3339),
386+ VCPUNum: rec.VCPUNum,
387+ RamMB: rec.RamMB,
388+ })
389+ if err != nil {
390+ ulog.GetLogger().Error("failed to create sandbox lifecycle event", ulog.F("sandbox_id", rec.SandboxID), ulog.F("error", err))
391+ return
392+ }
393+ s.WebhookDispatcher.Publish(event)
394+}
395+ 
339func (s *Service) SuspendSandbox(ctx context.Context, sandboxID string) error {396func (s *Service) SuspendSandbox(ctx context.Context, sandboxID string) error {
340 if s == nil || s.Sandbox == nil {397 if s == nil || s.Sandbox == nil {
341 return fmt.Errorf("sandbox service is not configured")398 return fmt.Errorf("sandbox service is not configured")
@@ -673,7 +730,7 @@ func (s *Service) createTemplateFromSource(ctx context.Context, opts TemplateCre
673 return templateBuildResult{}, fmt.Errorf("label rootfs source image: %w", err)730 return templateBuildResult{}, fmt.Errorf("label rootfs source image: %w", err)
674 }731 }
675 732 
676- buildID, err := NewID()733+ buildID, err := id.New()
677 if err != nil {734 if err != nil {
678 return templateBuildResult{}, err735 return templateBuildResult{}, err
679 }736 }
@@ -872,14 +929,6 @@ func combineOperationErrors(primary error, secondary ...error) error {
872 return fmt.Errorf("%w; additional operation failures: %v", primary, additional)929 return fmt.Errorf("%w; additional operation failures: %v", primary, additional)
873}930}
874 931 
875-func NewID() (string, error) {
876- var data [16]byte
877- if _, err := rand.Read(data[:]); err != nil {
878- return "", fmt.Errorf("generate id: %w", err)
879- }
880- return hex.EncodeToString(data[:]), nil
881-}
882- 
883func copyMap(in map[string]string) map[string]string {932func copyMap(in map[string]string) map[string]string {
884 if len(in) == 0 {933 if len(in) == 0 {
885 return nil934 return nil
Minternal/conchruntime/service_test.go+103-2
@@ -2,8 +2,11 @@ package conchruntime
2 2 
3import (3import (
4 "context"4 "context"
5+ "encoding/json"
5 "errors"6 "errors"
6 "fmt"7 "fmt"
8+ "net/http"
9+ "net/http/httptest"
7 "os"10 "os"
8 "os/exec"11 "os/exec"
9 "path/filepath"12 "path/filepath"
@@ -20,12 +23,13 @@ import (
20 agentprotocol "github.com/openeuler/Conch/internal/agent/protocol"23 agentprotocol "github.com/openeuler/Conch/internal/agent/protocol"
21 "github.com/openeuler/Conch/internal/apperror"24 "github.com/openeuler/Conch/internal/apperror"
22 "github.com/openeuler/Conch/internal/daemon/state"25 "github.com/openeuler/Conch/internal/daemon/state"
26+ "github.com/openeuler/Conch/internal/id"
23 conchimage "github.com/openeuler/Conch/internal/image"27 conchimage "github.com/openeuler/Conch/internal/image"
24 "github.com/openeuler/Conch/internal/netstack"28 "github.com/openeuler/Conch/internal/netstack"
25 "github.com/openeuler/Conch/internal/runtimeapi"29 "github.com/openeuler/Conch/internal/runtimeapi"
26 "github.com/openeuler/Conch/internal/sandbox"30 "github.com/openeuler/Conch/internal/sandbox"
27- "github.com/openeuler/Conch/internal/sandboxid"
28 conchtemplate "github.com/openeuler/Conch/internal/template"31 conchtemplate "github.com/openeuler/Conch/internal/template"
32+ "github.com/openeuler/Conch/internal/webhook"
29)33)
30 34 
31type fakeSandboxOps struct {35type fakeSandboxOps struct {
@@ -42,6 +46,103 @@ type fakeSandboxOps struct {
42 createHook func()46 createHook func()
43}47}
44 48 
49+func TestSandboxLifecycleEventsPublishedAfterCreateAndDelete(t *testing.T) {
50+ store, err := state.OpenBolt(filepath.Join(t.TempDir(), "state.db"))
51+ if err != nil {
52+ t.Fatalf("OpenBolt: %v", err)
53+ }
54+ t.Cleanup(func() { _ = store.Close() })
55+ events := make(chan webhook.Event, 2)
56+ receiver := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
57+ var event webhook.Event
58+ if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
59+ t.Errorf("decode event: %v", err)
60+ return
61+ }
62+ events <- event
63+ }))
64+ defer receiver.Close()
65+ dispatcher := webhook.NewDispatcher()
66+ if _, err := dispatcher.Create(runtimeapi.WebhookCreateOptions{Name: "receiver", URL: receiver.URL}); err != nil {
67+ t.Fatalf("register webhook: %v", err)
68+ }
69+ templateID := digest.FromString("lifecycle-event-template").String()
70+ svc := New(&fakeSandboxOps{createResult: sandbox.CreateResult{BootIndexDigest: templateID}}, nil, store)
71+ svc.WebhookDispatcher = dispatcher
72+ created, err := svc.CreateSandbox(context.Background(), SandboxCreateOptions{SandboxID: "sandbox-events", TemplateID: templateID, VCPUNum: 2, VCPUMax: 2, RamMB: 512})
73+ if err != nil {
74+ t.Fatalf("CreateSandbox: %v", err)
75+ }
76+ select {
77+ case event := <-events:
78+ if event.Type != webhook.EventSandboxCreated || event.EventData.KillReason != "" || event.SandboxID != created.SandboxID || event.EventData.Execution.VCPUNum != 2 || event.EventData.Execution.RamMB != 512 {
79+ t.Fatalf("created event = %#v", event)
80+ }
81+ case <-time.After(time.Second):
82+ t.Fatal("created event not delivered")
83+ }
84+ if err := svc.RemoveSandbox(context.Background(), created.SandboxID); err != nil {
85+ t.Fatalf("RemoveSandbox: %v", err)
86+ }
87+ select {
88+ case event := <-events:
89+ if event.Type != webhook.EventSandboxKilled || event.EventData.KillReason != "request" || event.SandboxID != created.SandboxID {
90+ t.Fatalf("killed event = %#v", event)
91+ }
92+ case <-time.After(time.Second):
93+ t.Fatal("killed event not delivered")
94+ }
95+}
96+ 
97+func TestHandleSandboxUnexpectedExitMarksUnknownAndPublishesOnce(t *testing.T) {
98+ store, err := state.OpenBolt(filepath.Join(t.TempDir(), "state.db"))
99+ if err != nil {
100+ t.Fatalf("OpenBolt: %v", err)
101+ }
102+ t.Cleanup(func() { _ = store.Close() })
103+ templateID := digest.FromString("orphaned-event-template").String()
104+ record := state.SandboxRecord{SandboxID: "sandbox-orphaned", State: state.SandboxReady, CreatedAt: time.Now().UnixNano(), CheckpointHeadTemplateID: templateID, VCPUNum: 2, RamMB: 512}
105+ if err := store.UpsertSandbox(context.Background(), record); err != nil {
106+ t.Fatalf("seed sandbox: %v", err)
107+ }
108+ events := make(chan webhook.Event, 2)
109+ receiver := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
110+ var event webhook.Event
111+ if err := json.NewDecoder(r.Body).Decode(&event); err == nil {
112+ events <- event
113+ }
114+ }))
115+ defer receiver.Close()
116+ dispatcher := webhook.NewDispatcher()
117+ if _, err := dispatcher.Create(runtimeapi.WebhookCreateOptions{Name: "receiver", URL: receiver.URL}); err != nil {
118+ t.Fatalf("register webhook: %v", err)
119+ }
120+ svc := New(nil, nil, store)
121+ svc.WebhookDispatcher = dispatcher
122+ svc.HandleSandboxUnexpectedExit(record.SandboxID)
123+ svc.HandleSandboxUnexpectedExit(record.SandboxID)
124+ select {
125+ case event := <-events:
126+ if event.Type != webhook.EventSandboxKilled || event.EventData.KillReason != "orphaned" || event.SandboxID != record.SandboxID {
127+ t.Fatalf("orphaned event = %#v", event)
128+ }
129+ case <-time.After(time.Second):
130+ t.Fatal("orphaned event not delivered")
131+ }
132+ select {
133+ case event := <-events:
134+ t.Fatalf("unexpected duplicate orphaned event = %#v", event)
135+ case <-time.After(100 * time.Millisecond):
136+ }
137+ updated, err := store.GetSandbox(context.Background(), record.SandboxID)
138+ if err != nil {
139+ t.Fatalf("GetSandbox: %v", err)
140+ }
141+ if updated.State != state.SandboxUnknown {
142+ t.Fatalf("state = %q, want %q", updated.State, state.SandboxUnknown)
143+ }
144+}
145+ 
45type serializedDeleteOps struct {146type serializedDeleteOps struct {
46 fakeSandboxOps147 fakeSandboxOps
47 firstEntered chan struct{}148 firstEntered chan struct{}
@@ -654,7 +755,7 @@ func TestCreateSandboxGeneratesIDWhenSandboxIDIsNotProvided(t *testing.T) {
654 if err != nil {755 if err != nil {
655 t.Fatalf("CreateSandbox() error = %v", err)756 t.Fatalf("CreateSandbox() error = %v", err)
656 }757 }
657- if len(result.SandboxID) != 32 || sandboxid.Validate(result.SandboxID) != nil {758+ if len(result.SandboxID) != 32 || id.Validate(result.SandboxID) != nil {
658 t.Fatalf("generated sandbox ID = %q, want 32-character safe ID", result.SandboxID)759 t.Fatalf("generated sandbox ID = %q, want 32-character safe ID", result.SandboxID)
659 }760 }
660 if ops.req.SandboxID != result.SandboxID {761 if ops.req.SandboxID != result.SandboxID {
Minternal/daemon/daemon.go+88-10
@@ -32,6 +32,7 @@ import (
32 conchsnapshot "github.com/openeuler/Conch/internal/snapshot"32 conchsnapshot "github.com/openeuler/Conch/internal/snapshot"
33 "github.com/openeuler/Conch/internal/util"33 "github.com/openeuler/Conch/internal/util"
34 "github.com/openeuler/Conch/internal/volume"34 "github.com/openeuler/Conch/internal/volume"
35+ "github.com/openeuler/Conch/internal/webhook"
35 "github.com/openeuler/Conch/pkg/ulog"36 "github.com/openeuler/Conch/pkg/ulog"
36)37)
37 38 
@@ -47,16 +48,17 @@ const (
47)48)
48 49 
49type Daemon struct {50type Daemon struct {
50- router *http.ServeMux51+ router *http.ServeMux
51- containerdHost *containerdhost.Host52+ containerdHost *containerdhost.Host
52- stateStore state.Store53+ stateStore state.Store
53- runtimeService *conchruntime.Service54+ runtimeService *conchruntime.Service
54- volumeManager *volume.Manager55+ webhookDispatcher *webhook.Dispatcher
55- daemonClient *containerdclient.Client56+ volumeManager *volume.Manager
56- httpServer *http.Server57+ daemonClient *containerdclient.Client
57- listener net.Listener58+ httpServer *http.Server
58- unixSocketPath string59+ listener net.Listener
59- cleanupOnce sync.Once60+ unixSocketPath string
61+ cleanupOnce sync.Once
60 62 
61 // TODO: need ListCachedBuilds()63 // TODO: need ListCachedBuilds()
62}64}
@@ -156,6 +158,8 @@ func New(cfg *config.Config) (*Daemon, error) {
156 s.daemonClient = daemonClient158 s.daemonClient = daemonClient
157 159 
158 s.runtimeService = conchruntime.New(host.SandboxManager(), host.Client(), store)160 s.runtimeService = conchruntime.New(host.SandboxManager(), host.Client(), store)
161+ s.webhookDispatcher = webhook.NewDispatcher()
162+ s.runtimeService.WebhookDispatcher = s.webhookDispatcher
159 s.runtimeService.Snapshot = host.SnapshotServer()163 s.runtimeService.Snapshot = host.SnapshotServer()
160 s.runtimeService.Templates = host.TemplateStore()164 s.runtimeService.Templates = host.TemplateStore()
161 s.runtimeService.SetSandboxDefaults(runtimeapi.SandboxDefaults{165 s.runtimeService.SetSandboxDefaults(runtimeapi.SandboxDefaults{
@@ -168,6 +172,7 @@ func New(cfg *config.Config) (*Daemon, error) {
168 172 
169 manager := host.SandboxManager()173 manager := host.SandboxManager()
170 if manager != nil {174 if manager != nil {
175+ manager.UnexpectedExitHandler = s.runtimeService.HandleSandboxUnexpectedExit
171 records, err := store.ListSandboxes(ctx)176 records, err := store.ListSandboxes(ctx)
172 if err != nil {177 if err != nil {
173 cleanupErr := host.Close()178 cleanupErr := host.Close()
@@ -215,6 +220,9 @@ func New(cfg *config.Config) (*Daemon, error) {
215}220}
216 221 
217func (s *Daemon) routes() {222func (s *Daemon) routes() {
223+ s.router.HandleFunc("POST /api/v1/events/webhooks", s.handleCreateWebhook)
224+ s.router.HandleFunc("GET /api/v1/events/webhooks", s.handleListWebhooks)
225+ s.router.HandleFunc("DELETE /api/v1/events/webhooks/{webhookID}", s.handleDeleteWebhook)
218 // sandbox226 // sandbox
219 s.router.HandleFunc("GET /api/v1/sandboxes", s.handleListSandbox)227 s.router.HandleFunc("GET /api/v1/sandboxes", s.handleListSandbox)
220 s.router.HandleFunc("POST /api/v1/sandboxes", s.handleCreateSandbox)228 s.router.HandleFunc("POST /api/v1/sandboxes", s.handleCreateSandbox)
@@ -455,6 +463,76 @@ func (s *Daemon) handleCreateSandbox(w http.ResponseWriter, r *http.Request) {
455 _ = json.NewEncoder(w).Encode(sandboxResponseFromCreate(result))463 _ = json.NewEncoder(w).Encode(sandboxResponseFromCreate(result))
456}464}
457 465 
466+// Webhook management handlers configure the daemon-local in-memory dispatcher.
467+func (s *Daemon) handleCreateWebhook(w http.ResponseWriter, r *http.Request) {
468+ logger := ulog.GetLogger()
469+ logger.Debug("Handling create webhook request")
470+ 
471+ if s.webhookDispatcher == nil {
472+ writeAPIError(w, errServiceUnavailable.New())
473+ return
474+ }
475+ var req webhookCreateRequest
476+ if !decodeJSONBody(w, r, &req) {
477+ return
478+ }
479+ hook, err := s.webhookDispatcher.Create(runtimeapi.WebhookCreateOptions{
480+ Name: req.Name, URL: req.URL, Events: req.Events,
481+ })
482+ if err != nil {
483+ writeAPIError(w, err)
484+ return
485+ }
486+ logger.Info("Webhook created successfully", ulog.F("webhook_id", hook.WebhookID), ulog.F("webhook_name", hook.Name))
487+ w.Header().Set("Content-Type", "application/json")
488+ w.WriteHeader(http.StatusCreated)
489+ _ = json.NewEncoder(w).Encode(webhookResponseFromRecord(hook))
490+}
491+ 
492+func (s *Daemon) handleListWebhooks(w http.ResponseWriter, r *http.Request) {
493+ logger := ulog.GetLogger()
494+ logger.Debug("Handling list webhooks request")
495+ 
496+ if s.webhookDispatcher == nil {
497+ writeAPIError(w, errServiceUnavailable.New())
498+ return
499+ }
500+ w.Header().Set("Content-Type", "application/json")
501+ records := s.webhookDispatcher.List()
502+ hooks := make([]webhookResponse, 0, len(records))
503+ for _, record := range records {
504+ hooks = append(hooks, webhookResponseFromRecord(record))
505+ }
506+ logger.Debug("Webhooks listed successfully", ulog.F("webhook_count", len(hooks)))
507+ _ = json.NewEncoder(w).Encode(listWebhooksResponse{Webhooks: hooks})
508+}
509+ 
510+func webhookResponseFromRecord(record runtimeapi.WebhookRecord) webhookResponse {
511+ return webhookResponse{
512+ WebhookID: record.WebhookID, Name: record.Name, URL: record.URL,
513+ Events: append([]string(nil), record.Events...),
514+ CreatedAt: record.CreatedAt.UTC().Format(time.RFC3339),
515+ }
516+}
517+ 
518+func (s *Daemon) handleDeleteWebhook(w http.ResponseWriter, r *http.Request) {
519+ logger := ulog.GetLogger()
520+ webhookID := strings.TrimSpace(r.PathValue("webhookID"))
521+ logger.Debug("Handling delete webhook request", ulog.F("webhook_id", webhookID))
522+ 
523+ if s.webhookDispatcher == nil {
524+ writeAPIError(w, errServiceUnavailable.New())
525+ return
526+ }
527+ if webhookID == "" || !s.webhookDispatcher.Delete(webhookID) {
528+ writeAPIError(w, webhook.ErrNotFound.New())
529+ return
530+ }
531+ logger.Info("Webhook deleted successfully", ulog.F("webhook_id", webhookID))
532+ w.Header().Set("Content-Type", "application/json")
533+ _ = json.NewEncoder(w).Encode(deleteWebhookResponse{WebhookID: webhookID, Status: "deleted"})
534+}
535+ 
458func (s *Daemon) handleUpdateSandboxNetwork(w http.ResponseWriter, r *http.Request) {536func (s *Daemon) handleUpdateSandboxNetwork(w http.ResponseWriter, r *http.Request) {
459 sandboxID := r.PathValue("sandboxID")537 sandboxID := r.PathValue("sandboxID")
460 if sandboxID == "" {538 if sandboxID == "" {
Ainternal/daemon/handler_webhook_test.go+57-0
@@ -0,0 +1,57 @@
1+package daemon
2+ 
3+import (
4+ "bytes"
5+ "encoding/json"
6+ "net/http"
7+ "net/http/httptest"
8+ "testing"
9+ 
10+ "github.com/openeuler/Conch/internal/webhook"
11+)
12+ 
13+func TestWebhookManagementHandlers(t *testing.T) {
14+ server := &Daemon{router: http.NewServeMux(), webhookDispatcher: webhook.NewDispatcher()}
15+ server.routes()
16+ create := httptest.NewRecorder()
17+ server.router.ServeHTTP(create, httptest.NewRequest(http.MethodPost, "/api/v1/events/webhooks", bytes.NewBufferString(`{"name":"reliability","url":"https://example.test/events","events":["sandbox.lifecycle.killed"]}`)))
18+ if create.Code != http.StatusCreated {
19+ t.Fatalf("create status = %d, body = %s", create.Code, create.Body.String())
20+ }
21+ var hook webhookResponse
22+ if err := json.NewDecoder(create.Body).Decode(&hook); err != nil {
23+ t.Fatalf("decode create: %v", err)
24+ }
25+ list := httptest.NewRecorder()
26+ server.router.ServeHTTP(list, httptest.NewRequest(http.MethodGet, "/api/v1/events/webhooks", nil))
27+ if list.Code != http.StatusOK {
28+ t.Fatalf("list status = %d", list.Code)
29+ }
30+ var response listWebhooksResponse
31+ if err := json.NewDecoder(list.Body).Decode(&response); err != nil || len(response.Webhooks) != 1 {
32+ t.Fatalf("list = %#v, err = %v", response, err)
33+ }
34+ deleteResponse := httptest.NewRecorder()
35+ server.router.ServeHTTP(deleteResponse, httptest.NewRequest(http.MethodDelete, "/api/v1/events/webhooks/"+hook.WebhookID, nil))
36+ if deleteResponse.Code != http.StatusOK {
37+ t.Fatalf("delete status = %d", deleteResponse.Code)
38+ }
39+ missing := httptest.NewRecorder()
40+ server.router.ServeHTTP(missing, httptest.NewRequest(http.MethodDelete, "/api/v1/events/webhooks/"+hook.WebhookID, nil))
41+ if missing.Code != http.StatusNotFound {
42+ t.Fatalf("missing delete status = %d", missing.Code)
43+ }
44+ var missingError apiErrorResponse
45+ if err := json.NewDecoder(missing.Body).Decode(&missingError); err != nil || missingError.Code != "webhook.not_found" {
46+ t.Fatalf("missing delete error = %#v, err = %v", missingError, err)
47+ }
48+ invalid := httptest.NewRecorder()
49+ server.router.ServeHTTP(invalid, httptest.NewRequest(http.MethodPost, "/api/v1/events/webhooks", bytes.NewBufferString(`{"name":"","url":"https://example.test"}`)))
50+ if invalid.Code != http.StatusBadRequest {
51+ t.Fatalf("invalid create status = %d", invalid.Code)
52+ }
53+ var invalidError apiErrorResponse
54+ if err := json.NewDecoder(invalid.Body).Decode(&invalidError); err != nil || invalidError.Code != "webhook.invalid_argument" {
55+ t.Fatalf("invalid create error = %#v, err = %v", invalidError, err)
56+ }
57+}
Minternal/daemon/types.go+24-0
@@ -5,6 +5,30 @@ import (
5 "github.com/openeuler/Conch/internal/volume"5 "github.com/openeuler/Conch/internal/volume"
6)6)
7 7 
8+// webhookCreateRequest is the daemon HTTP API payload for registering a Webhook.
9+type webhookCreateRequest struct {
10+ Name string `json:"name"`
11+ URL string `json:"url"`
12+ Events []string `json:"events"`
13+}
14+ 
15+type webhookResponse struct {
16+ WebhookID string `json:"webhook_id"`
17+ Name string `json:"name"`
18+ URL string `json:"url"`
19+ Events []string `json:"events"`
20+ CreatedAt string `json:"createdAt"`
21+}
22+ 
23+type listWebhooksResponse struct {
24+ Webhooks []webhookResponse `json:"webhooks"`
25+}
26+ 
27+type deleteWebhookResponse struct {
28+ WebhookID string `json:"webhook_id"`
29+ Status string `json:"status"`
30+}
31+ 
8type pullImageRequest struct {32type pullImageRequest struct {
9 ImageName string `json:"image_name"`33 ImageName string `json:"image_name"`
10 PlainHTTP bool `json:"plain_http,omitempty"`34 PlainHTTP bool `json:"plain_http,omitempty"`
Rinternal/sandboxid/sandboxid.go→internal/id/id.go+17-4
@@ -1,6 +1,9 @@
1-package sandboxid1+// Package id provides generation and validation for Conch identifiers.
2+package id
2 3 
3import (4import (
5+ "crypto/rand"
6+ "encoding/hex"
4 "fmt"7 "fmt"
5 "regexp"8 "regexp"
6)9)
@@ -13,11 +16,21 @@ const (
13 16 
14var pattern = regexp.MustCompile(`^` + chars + `+$`)17var pattern = regexp.MustCompile(`^` + chars + `+$`)
15 18 
16-func Validate(id string) error {19+func New() (string, error) { return NewWithPrefix("") }
17- if len(id) < MinLength || len(id) > MaxLength {20+ 
21+func NewWithPrefix(prefix string) (string, error) {
22+ var data [16]byte
23+ if _, err := rand.Read(data[:]); err != nil {
24+ return "", fmt.Errorf("generate id: %w", err)
25+ }
26+ return prefix + hex.EncodeToString(data[:]), nil
27+}
28+ 
29+func Validate(value string) error {
30+ if len(value) < MinLength || len(value) > MaxLength {
18 return fmt.Errorf("length must be between %d and %d characters", MinLength, MaxLength)31 return fmt.Errorf("length must be between %d and %d characters", MinLength, MaxLength)
19 }32 }
20- if !pattern.MatchString(id) {33+ if !pattern.MatchString(value) {
21 return fmt.Errorf("only %s are allowed", chars)34 return fmt.Errorf("only %s are allowed", chars)
22 }35 }
23 return nil36 return nil
Rinternal/sandboxid/sandboxid_test.go→internal/id/id_test.go+24-1
@@ -1,4 +1,4 @@
1-package sandboxid1+package id
2 2 
3import (3import (
4 "strings"4 "strings"
@@ -29,3 +29,26 @@ func TestValidate(t *testing.T) {
29 })29 })
30 }30 }
31}31}
32+ 
33+func TestNew(t *testing.T) {
34+ value, err := New()
35+ if err != nil {
36+ t.Fatalf("New() error = %v", err)
37+ }
38+ if len(value) != 32 {
39+ t.Fatalf("New() length = %d, want 32", len(value))
40+ }
41+ if err := Validate(value); err != nil {
42+ t.Fatalf("New() generated invalid ID %q: %v", value, err)
43+ }
44+}
45+ 
46+func TestNewWithPrefix(t *testing.T) {
47+ value, err := NewWithPrefix("wh_")
48+ if err != nil {
49+ t.Fatalf("NewWithPrefix() error = %v", err)
50+ }
51+ if len(value) != len("wh_")+32 || !strings.HasPrefix(value, "wh_") {
52+ t.Fatalf("NewWithPrefix() = %q, want wh_ prefix and 32 hex characters", value)
53+ }
54+}
Minternal/runtimeapi/types.go+16-0
@@ -19,6 +19,22 @@ type SandboxNetworkUpdateOptions struct {
19 Network *SandboxNetworkConfig19 Network *SandboxNetworkConfig
20}20}
21 21 
22+// WebhookCreateOptions describes a Webhook registration for this conchd instance.
23+type WebhookCreateOptions struct {
24+ Name string
25+ URL string
26+ Events []string
27+}
28+ 
29+// WebhookRecord is the runtime representation of an in-memory Webhook registration.
30+type WebhookRecord struct {
31+ WebhookID string
32+ Name string
33+ URL string
34+ Events []string
35+ CreatedAt time.Time
36+}
37+ 
22// ImageRecord.Kind values exposed by the image API. These classify the38// ImageRecord.Kind values exposed by the image API. These classify the
23// user-visible image record, not the io.conch.kind annotation stored on Boot39// user-visible image record, not the io.conch.kind annotation stored on Boot
24// Index component descriptors.40// Index component descriptors.
Minternal/sandbox/manager.go+19-13
@@ -31,17 +31,18 @@ type Config struct {
31}31}
32 32 
33type Manager struct {33type Manager struct {
34- sandboxes sync.Map // map[string]*sandboxEntry34+ sandboxes sync.Map // map[string]*sandboxEntry
35- pool *netstack.Pool35+ pool *netstack.Pool
36- daemonClient *containerdclient.Client36+ daemonClient *containerdclient.Client
37- boot BootPreparer37+ boot BootPreparer
38- checkpointCapture CheckpointCapture38+ checkpointCapture CheckpointCapture
39- vsockSignalRetry time.Duration39+ vsockSignalRetry time.Duration
40- vsockSignalTimeout time.Duration40+ vsockSignalTimeout time.Duration
41- requestTimeout time.Duration41+ requestTimeout time.Duration
42- cidAllocator *CIDAllocator42+ cidAllocator *CIDAllocator
43- volumeManager *volume.Manager43+ volumeManager *volume.Manager
44- vmmBinaries map[string]string44+ vmmBinaries map[string]string
45+ UnexpectedExitHandler UnexpectedExitHandler
45}46}
46 47 
47type sandboxLifecycleState uint848type sandboxLifecycleState uint8
@@ -68,6 +69,8 @@ type sandboxEntry struct {
68 sbx *Sandbox69 sbx *Sandbox
69}70}
70 71 
72+type UnexpectedExitHandler func(sandboxID string)
73+ 
71func New(74func New(
72 ctx context.Context,75 ctx context.Context,
73 client *containerdclient.Client,76 client *containerdclient.Client,
@@ -593,15 +596,18 @@ func (m *Manager) trackSandbox(ctx context.Context, mapKey string, entry *sandbo
593func (m *Manager) handleSandboxExit(mapKey string, entry *sandboxEntry, sandboxID string, sbx *Sandbox) {596func (m *Manager) handleSandboxExit(mapKey string, entry *sandboxEntry, sandboxID string, sbx *Sandbox) {
594 logger := ulog.GetLogger()597 logger := ulog.GetLogger()
595 entry.mu.Lock()598 entry.mu.Lock()
596- defer entry.mu.Unlock()
597 if !m.isCurrentSandboxEntry(mapKey, entry) || entry.sbx != sbx {599 if !m.isCurrentSandboxEntry(mapKey, entry) || entry.sbx != sbx {
600+ entry.mu.Unlock()
598 return601 return
599 }602 }
600- 
601 if err := m.cleanupSandbox(context.Background(), sbx, sandboxID); err != nil {603 if err := m.cleanupSandbox(context.Background(), sbx, sandboxID); err != nil {
602 logger.Warn("failed to cleanup sandbox after wait", ulog.F("sandbox_id", sandboxID), ulog.F("error", err))604 logger.Warn("failed to cleanup sandbox after wait", ulog.F("sandbox_id", sandboxID), ulog.F("error", err))
603 }605 }
604 m.sandboxes.CompareAndDelete(mapKey, entry)606 m.sandboxes.CompareAndDelete(mapKey, entry)
607+ entry.mu.Unlock()
608+ if m.UnexpectedExitHandler != nil {
609+ m.UnexpectedExitHandler(sandboxID)
610+ }
605}611}
606 612 
607func buildSandboxCreateResult(leaseID string, req CreateRequest, sbx *Sandbox, boot PreparedBoot, runtimeIDs createRuntimeIDs, volumeDevices []volume.Device) CreateResult {613func buildSandboxCreateResult(leaseID string, req CreateRequest, sbx *Sandbox, boot PreparedBoot, runtimeIDs createRuntimeIDs, volumeDevices []volume.Device) CreateResult {
Minternal/sandbox/manager_test.go+21-0
@@ -73,6 +73,27 @@ func TestHandleSandboxExitCleansSuspendedSandbox(t *testing.T) {
73 }73 }
74}74}
75 75 
76+func TestHandleSandboxExitCallsUnexpectedHandlerOnce(t *testing.T) {
77+ m, entry, sbx := newExitTestSandbox(func(context.Context) error { return nil })
78+ called := make(chan string, 2)
79+ m.UnexpectedExitHandler = func(id string) { called <- id }
80+ m.handleSandboxExit("sandbox-a", entry, "sandbox-a", sbx)
81+ m.handleSandboxExit("sandbox-a", entry, "sandbox-a", sbx)
82+ select {
83+ case id := <-called:
84+ if id != "sandbox-a" {
85+ t.Fatalf("handler sandbox ID = %q", id)
86+ }
87+ case <-time.After(time.Second):
88+ t.Fatal("unexpected-exit handler was not called")
89+ }
90+ select {
91+ case id := <-called:
92+ t.Fatalf("handler called more than once for %q", id)
93+ case <-time.After(100 * time.Millisecond):
94+ }
95+}
96+ 
76func TestWaitForSandboxExitCleansSandboxOnVirtiofsExit(t *testing.T) {97func TestWaitForSandboxExitCleansSandboxOnVirtiofsExit(t *testing.T) {
77 cleanupDone := make(chan struct{})98 cleanupDone := make(chan struct{})
78 m, entry, sbx := newExitTestSandbox(func(context.Context) error {99 m, entry, sbx := newExitTestSandbox(func(context.Context) error {
Minternal/sandbox/sandbox.go+1-2
@@ -247,8 +247,7 @@ func CreateSandbox(
247}247}
248 248 
249func (s *Sandbox) Wait(ctx context.Context) error {249func (s *Sandbox) Wait(ctx context.Context) error {
250- s.process.Wait()250+ return s.process.Wait()
251- return nil
252}251}
253 252 
254func (s *Sandbox) Stop(ctx context.Context) error {253func (s *Sandbox) Stop(ctx context.Context) error {
Minternal/vmm/clh/cloud_hypervisor.go+3-3
@@ -213,11 +213,11 @@ func (c *CLHClient) Cleanup() {
213 }213 }
214}214}
215 215 
216-func (c *CLHClient) WaitForCreateReady(ctx context.Context, processExited <-chan error) error {216+func (c *CLHClient) WaitForCreateReady(ctx context.Context, _ driver.ProcessExit) error {
217 return c.waitForSourceEvent(ctx, "vm", EventBooted)217 return c.waitForSourceEvent(ctx, "vm", EventBooted)
218}218}
219 219 
220-func (c *CLHClient) WaitForRestoreReady(ctx context.Context, processExited <-chan error) error {220+func (c *CLHClient) WaitForRestoreReady(ctx context.Context, _ driver.ProcessExit) error {
221 return nil221 return nil
222}222}
223 223 
@@ -548,7 +548,7 @@ func (c *CLHClient) requestApi(method, fullCommand, requestBody string) error {
548 return nil548 return nil
549}549}
550 550 
551-func (c *CLHClient) CheckAgentAlive(ctx context.Context, processExited <-chan error) error {551+func (c *CLHClient) CheckAgentAlive(ctx context.Context, processExited driver.ProcessExit) error {
552 // TODO: call conch-init GetHealth552 // TODO: call conch-init GetHealth
553 return nil553 return nil
554}554}
Minternal/vmm/driver/types.go+8-3
@@ -2,6 +2,11 @@ package driver
2 2 
3import "context"3import "context"
4 4 
5+type ProcessExit interface {
6+ Done() <-chan struct{}
7+ Err() error
8+}
9+ 
5type ResourceArgs struct {10type ResourceArgs struct {
6 // CPU11 // CPU
7 CPUBoot int6412 CPUBoot int64
@@ -46,9 +51,9 @@ type Adapter interface {
46 BuildStartCmd(args *ResourceArgs, restore bool) (string, error)51 BuildStartCmd(args *ResourceArgs, restore bool) (string, error)
47 PrepareLaunch(args *ResourceArgs, restore bool) error52 PrepareLaunch(args *ResourceArgs, restore bool) error
48 AfterProcessStart()53 AfterProcessStart()
49- WaitForCreateReady(ctx context.Context, processExited <-chan error) error54+ WaitForCreateReady(ctx context.Context, processExited ProcessExit) error
50- WaitForRestoreReady(ctx context.Context, processExited <-chan error) error55+ WaitForRestoreReady(ctx context.Context, processExited ProcessExit) error
51- CheckAgentAlive(ctx context.Context, processExited <-chan error) error56+ CheckAgentAlive(ctx context.Context, processExited ProcessExit) error
52 PauseVM() error57 PauseVM() error
53 ResumeVM() error58 ResumeVM() error
54 DeleteVM() error59 DeleteVM() error
Minternal/vmm/process.go+26-23
@@ -56,8 +56,9 @@ type Process struct {
56 apiReadyMu sync.Mutex56 apiReadyMu sync.Mutex
57 apiReady bool57 apiReady bool
58 // Exit *utils.SetOnce[struct{}]58 // Exit *utils.SetOnce[struct{}]
59- adapter vmmAdapter59+ adapter vmmAdapter
60- exitSignal chan error60+ exitDone chan struct{}
61+ exitErr error
61}62}
62 63 
63func SandboxVmmSocketPath(sandboxId string) (string, error) {64func SandboxVmmSocketPath(sandboxId string) (string, error) {
@@ -105,7 +106,7 @@ func NewProcess(
105 VsockSocketPath: vmmResourceArgs.VsockSocketPath,106 VsockSocketPath: vmmResourceArgs.VsockSocketPath,
106 VmmSocketPath: vmmSocketPath,107 VmmSocketPath: vmmSocketPath,
107 adapter: adapter,108 adapter: adapter,
108- exitSignal: make(chan error, 1),109+ exitDone: make(chan struct{}),
109 }110 }
110 111 
111 startScript, err := adapter.BuildStartCmd(vmmResourceArgs, restore)112 startScript, err := adapter.BuildStartCmd(vmmResourceArgs, restore)
@@ -169,8 +170,7 @@ func (p *Process) startCmd(
169 // Check if process was killed by a signal170 // Check if process was killed by a signal
170 if status, ok := exitErr.Sys().(syscall.WaitStatus); ok && status.Signaled() && (status.Signal() == syscall.SIGKILL || status.Signal() == syscall.SIGTERM) {171 if status, ok := exitErr.Sys().(syscall.WaitStatus); ok && status.Signaled() && (status.Signal() == syscall.SIGKILL || status.Signal() == syscall.SIGTERM) {
171 logger.Debug("VMM process killed by signal")172 logger.Debug("VMM process killed by signal")
172- p.exitSignal <- nil173+ p.recordExit(nil)
173- close(p.exitSignal)
174 return174 return
175 }175 }
176 }176 }
@@ -178,20 +178,18 @@ func (p *Process) startCmd(
178 logger.Warn("VMM process error",178 logger.Warn("VMM process error",
179 ulog.F("error", errMsg),179 ulog.F("error", errMsg),
180 )180 )
181- p.exitSignal <- errMsg181+ p.recordExit(errMsg)
182- close(p.exitSignal)
183 return182 return
184 }183 }
185 logger.Debug("VMM process exited normally")184 logger.Debug("VMM process exited normally")
186- p.exitSignal <- nil185+ p.recordExit(nil)
187- close(p.exitSignal)
188 }()186 }()
189 187 
190 return nil188 return nil
191}189}
192 190 
193func (p *Process) waitForAgentAlive(ctx context.Context) error {191func (p *Process) waitForAgentAlive(ctx context.Context) error {
194- return p.adapter.CheckAgentAlive(ctx, p.exitSignal)192+ return p.adapter.CheckAgentAlive(ctx, p)
195}193}
196 194 
197func (p *Process) Create(ctx context.Context) error {195func (p *Process) Create(ctx context.Context) error {
@@ -204,7 +202,7 @@ func (p *Process) Create(ctx context.Context) error {
204 return errors.Join(fmt.Errorf("error starting vmm process: %w", err), vmmStopErr)202 return errors.Join(fmt.Errorf("error starting vmm process: %w", err), vmmStopErr)
205 }203 }
206 204 
207- if err := p.adapter.WaitForCreateReady(ctx, p.exitSignal); err != nil {205+ if err := p.adapter.WaitForCreateReady(ctx, p); err != nil {
208 vmmStopErr := p.Stop()206 vmmStopErr := p.Stop()
209 return errors.Join(fmt.Errorf("error waiting for vmm create readiness: %w", err), vmmStopErr)207 return errors.Join(fmt.Errorf("error waiting for vmm create readiness: %w", err), vmmStopErr)
210 }208 }
@@ -234,7 +232,7 @@ func (p *Process) Restore(ctx context.Context, snapshotPath string) error {
234 return errors.Join(fmt.Errorf("error starting vmm process: %w", err), vmmStopErr)232 return errors.Join(fmt.Errorf("error starting vmm process: %w", err), vmmStopErr)
235 }233 }
236 234 
237- if err := p.adapter.WaitForRestoreReady(ctx, p.exitSignal); err != nil {235+ if err := p.adapter.WaitForRestoreReady(ctx, p); err != nil {
238 vmmStopErr := p.Stop()236 vmmStopErr := p.Stop()
239 return errors.Join(fmt.Errorf("error waiting for vmm restore readiness: %w", err), vmmStopErr)237 return errors.Join(fmt.Errorf("error waiting for vmm restore readiness: %w", err), vmmStopErr)
240 }238 }
@@ -284,7 +282,7 @@ func (p *Process) Stop() error {
284 }282 }
285 283 
286 select {284 select {
287- case <-p.exitSignal:285+ case <-p.exitDone:
288 // Already exited286 // Already exited
289 p.adapter.Cleanup()287 p.adapter.Cleanup()
290 return errors.Join(errs...)288 return errors.Join(errs...)
@@ -330,7 +328,7 @@ func (p *Process) Stop() error {
330 ulog.F("pid", p.cmd.Process.Pid),328 ulog.F("pid", p.cmd.Process.Pid),
331 )329 )
332 330 
333- <-p.exitSignal331+ <-p.exitDone
334 p.adapter.Cleanup()332 p.adapter.Cleanup()
335 return errors.Join(errs...)333 return errors.Join(errs...)
336}334}
@@ -366,15 +364,9 @@ func (p *Process) CreateSnapshot(ctx context.Context, snapfilePath string) error
366func (p *Process) Wait() error {364func (p *Process) Wait() error {
367 logger := ulog.GetLogger()365 logger := ulog.GetLogger()
368 366 
369- // Blocks until single reaper goroutine (in startCmd) sends result.367+ // Blocks until the single reaper goroutine records its result.
370- // This ensures only one part of code calls OS wait syscall.368+ <-p.exitDone
371- err, ok := <-p.exitSignal369+ err := p.Err()
372- if !ok {
373- // Channel closed, process already reaped.
374- logger.Debug("Process already reaped")
375- p.adapter.Cleanup()
376- return nil
377- }
378 p.adapter.Cleanup()370 p.adapter.Cleanup()
379 if err != nil {371 if err != nil {
380 logger.Error("VMM process wait error",372 logger.Error("VMM process wait error",
@@ -384,3 +376,14 @@ func (p *Process) Wait() error {
384 }376 }
385 return nil377 return nil
386}378}
379+ 
380+func (p *Process) Done() <-chan struct{} { return p.exitDone }
381+ 
382+// Err returns the VMM exit result after Done has been closed.
383+func (p *Process) Err() error { return p.exitErr }
384+ 
385+func (p *Process) recordExit(err error) {
386+ // recordExit is called only by the single cmd.Wait reaper started in startCmd.
387+ p.exitErr = err
388+ close(p.exitDone)
389+}
Minternal/vmm/process_test.go+14-13
@@ -12,6 +12,7 @@ import (
12 "time"12 "time"
13 13 
14 "github.com/openeuler/Conch/internal/config"14 "github.com/openeuler/Conch/internal/config"
15+ "github.com/openeuler/Conch/internal/vmm/driver"
15)16)
16 17 
17func TestSandboxSocketPathUsesShortStableName(t *testing.T) {18func TestSandboxSocketPathUsesShortStableName(t *testing.T) {
@@ -110,15 +111,16 @@ type blockingDaemonClient struct {
110}111}
111 112 
112func (c *blockingDaemonClient) BuildStartCmd(*ResourceArgs, bool) (string, error) { return "", nil }113func (c *blockingDaemonClient) BuildStartCmd(*ResourceArgs, bool) (string, error) { return "", nil }
113-func (c *blockingDaemonClient) CheckAgentAlive(ctx context.Context, processExited <-chan error) error {114+func (c *blockingDaemonClient) CheckAgentAlive(ctx context.Context, processExited driver.ProcessExit) error {
114 select {115 select {
115 case <-c.release:116 case <-c.release:
116 return nil117 return nil
117- case waitErr, ok := <-processExited:118+ case <-processExited.Done():
118- if !ok || waitErr == nil {119+ if waitErr := processExited.Err(); waitErr == nil {
119 return errors.New("vmm process exited before conch-init became ready")120 return errors.New("vmm process exited before conch-init became ready")
121+ } else {
122+ return errors.Join(errors.New("vmm process exited before conch-init became ready"), waitErr)
120 }123 }
121- return errors.Join(errors.New("vmm process exited before conch-init became ready"), waitErr)
122 case <-ctx.Done():124 case <-ctx.Done():
123 return ctx.Err()125 return ctx.Err()
124 }126 }
@@ -132,10 +134,10 @@ func (c *blockingDaemonClient) PrepareLaunch(*ResourceArgs, bool) error {
132 return nil134 return nil
133}135}
134func (c *blockingDaemonClient) AfterProcessStart() {}136func (c *blockingDaemonClient) AfterProcessStart() {}
135-func (c *blockingDaemonClient) WaitForCreateReady(context.Context, <-chan error) error {137+func (c *blockingDaemonClient) WaitForCreateReady(context.Context, driver.ProcessExit) error {
136 return nil138 return nil
137}139}
138-func (c *blockingDaemonClient) WaitForRestoreReady(context.Context, <-chan error) error {140+func (c *blockingDaemonClient) WaitForRestoreReady(context.Context, driver.ProcessExit) error {
139 return nil141 return nil
140}142}
141func (c *blockingDaemonClient) Cleanup() { c.cleanupCalls.Add(1) }143func (c *blockingDaemonClient) Cleanup() { c.cleanupCalls.Add(1) }
@@ -151,9 +153,9 @@ func TestStopIgnoresProcessDoneWhenProcessAlreadyFinished(t *testing.T) {
151 153 
152 client := &blockingDaemonClient{release: make(chan struct{})}154 client := &blockingDaemonClient{release: make(chan struct{})}
153 process := &Process{155 process := &Process{
154- cmd: cmd,156+ cmd: cmd,
155- adapter: client,157+ adapter: client,
156- exitSignal: make(chan error, 1),158+ exitDone: make(chan struct{}),
157 }159 }
158 160 
159 if err := process.Stop(); err != nil {161 if err := process.Stop(); err != nil {
@@ -168,14 +170,13 @@ func TestWaitForAgentAliveReturnsProcessExitError(t *testing.T) {
168 processErr := errors.New("stratovirt exited after creating qmp socket")170 processErr := errors.New("stratovirt exited after creating qmp socket")
169 client := &blockingDaemonClient{release: make(chan struct{})}171 client := &blockingDaemonClient{release: make(chan struct{})}
170 process := &Process{172 process := &Process{
171- adapter: client,173+ adapter: client,
172- exitSignal: make(chan error, 1),174+ exitDone: make(chan struct{}),
173 }175 }
174 ctx, cancel := context.WithTimeout(context.Background(), time.Second)176 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
175 defer cancel()177 defer cancel()
176 178 
177- process.exitSignal <- processErr179+ process.recordExit(processErr)
178- close(process.exitSignal)
179 t.Cleanup(func() { close(client.release) })180 t.Cleanup(func() { close(client.release) })
180 181 
181 err := process.waitForAgentAlive(ctx)182 err := process.waitForAgentAlive(ctx)
Minternal/vmm/stratovirt/stratovirt.go+25-14
@@ -123,16 +123,20 @@ func (s *StratovirtClient) AfterProcessStart() {}
123 123 
124func (s *StratovirtClient) Cleanup() {}124func (s *StratovirtClient) Cleanup() {}
125 125 
126-func (s *StratovirtClient) WaitForCreateReady(ctx context.Context, processExited <-chan error) error {126+func (s *StratovirtClient) WaitForCreateReady(ctx context.Context, processExited driver.ProcessExit) error {
127 return waitForVmmSocket(ctx, s.socketPath, processExited)127 return waitForVmmSocket(ctx, s.socketPath, processExited)
128}128}
129 129 
130-func (s *StratovirtClient) WaitForRestoreReady(ctx context.Context, processExited <-chan error) error {130+func (s *StratovirtClient) WaitForRestoreReady(ctx context.Context, processExited driver.ProcessExit) error {
131 return waitForVmmSocket(ctx, s.socketPath, processExited)131 return waitForVmmSocket(ctx, s.socketPath, processExited)
132}132}
133 133 
134-func waitForVmmSocket(ctx context.Context, socketPath string, processExited <-chan error) error {134+func waitForVmmSocket(ctx context.Context, socketPath string, processExited driver.ProcessExit) error {
135 logger := ulog.GetLogger()135 logger := ulog.GetLogger()
136+ var exitDone <-chan struct{}
137+ if processExited != nil {
138+ exitDone = processExited.Done()
139+ }
136 140 
137 delay := 2 * time.Millisecond141 delay := 2 * time.Millisecond
138 const maxDelay = 100 * time.Millisecond142 const maxDelay = 100 * time.Millisecond
@@ -147,12 +151,13 @@ func waitForVmmSocket(ctx context.Context, socketPath string, processExited <-ch
147 case <-ctx.Done():151 case <-ctx.Done():
148 timer.Stop()152 timer.Stop()
149 return fmt.Errorf("cancelled waiting for vmm socket %s: %w", socketPath, ctx.Err())153 return fmt.Errorf("cancelled waiting for vmm socket %s: %w", socketPath, ctx.Err())
150- case waitErr, ok := <-processExited:154+ case <-exitDone:
151 timer.Stop()155 timer.Stop()
152- if !ok || waitErr == nil {156+ if waitErr := processExited.Err(); waitErr == nil {
153 return fmt.Errorf("vmm process exited before vmm socket %s was ready", socketPath)157 return fmt.Errorf("vmm process exited before vmm socket %s was ready", socketPath)
158+ } else {
159+ return fmt.Errorf("vmm process exited before vmm socket %s was ready: %w", socketPath, waitErr)
154 }160 }
155- return fmt.Errorf("vmm process exited before vmm socket %s was ready: %w", socketPath, waitErr)
156 case <-timer.C:161 case <-timer.C:
157 }162 }
158 163 
@@ -406,16 +411,21 @@ func (s *StratovirtClient) executeQMPCommandWithResponse(command string, argumen
406 return response, nil411 return response, nil
407}412}
408 413 
409-func waitForAgentRetry(ctx context.Context, processExited <-chan error, delay time.Duration) error {414+func waitForAgentRetry(ctx context.Context, processExited driver.ProcessExit, delay time.Duration) error {
415+ var exitDone <-chan struct{}
416+ if processExited != nil {
417+ exitDone = processExited.Done()
418+ }
410 if delay <= 0 {419 if delay <= 0 {
411 select {420 select {
412 case <-ctx.Done():421 case <-ctx.Done():
413 return fmt.Errorf("cancelled waiting for conch-init ready: %w", ctx.Err())422 return fmt.Errorf("cancelled waiting for conch-init ready: %w", ctx.Err())
414- case waitErr, ok := <-processExited:423+ case <-exitDone:
415- if !ok || waitErr == nil {424+ if waitErr := processExited.Err(); waitErr == nil {
416 return fmt.Errorf("vmm process exited before conch-init became ready")425 return fmt.Errorf("vmm process exited before conch-init became ready")
426+ } else {
427+ return fmt.Errorf("vmm process exited before conch-init became ready: %w", waitErr)
417 }428 }
418- return fmt.Errorf("vmm process exited before conch-init became ready: %w", waitErr)
419 default:429 default:
420 return nil430 return nil
421 }431 }
@@ -427,17 +437,18 @@ func waitForAgentRetry(ctx context.Context, processExited <-chan error, delay ti
427 select {437 select {
428 case <-ctx.Done():438 case <-ctx.Done():
429 return fmt.Errorf("cancelled waiting for conch-init ready: %w", ctx.Err())439 return fmt.Errorf("cancelled waiting for conch-init ready: %w", ctx.Err())
430- case waitErr, ok := <-processExited:440+ case <-exitDone:
431- if !ok || waitErr == nil {441+ if waitErr := processExited.Err(); waitErr == nil {
432 return fmt.Errorf("vmm process exited before conch-init became ready")442 return fmt.Errorf("vmm process exited before conch-init became ready")
443+ } else {
444+ return fmt.Errorf("vmm process exited before conch-init became ready: %w", waitErr)
433 }445 }
434- return fmt.Errorf("vmm process exited before conch-init became ready: %w", waitErr)
435 case <-timer.C:446 case <-timer.C:
436 return nil447 return nil
437 }448 }
438}449}
439 450 
440-func (s *StratovirtClient) CheckAgentAlive(ctx context.Context, processExited <-chan error) error {451+func (s *StratovirtClient) CheckAgentAlive(ctx context.Context, processExited driver.ProcessExit) error {
441 logger := ulog.GetLogger()452 logger := ulog.GetLogger()
442 453 
443 for i := 0; i < 60; i++ {454 for i := 0; i < 60; i++ {
Minternal/vmm/stratovirt/stratovirt_test.go+10-3
@@ -166,12 +166,11 @@ func TestWaitForVmmSocketWaitsUntilPathExists(t *testing.T) {
166func TestWaitForVmmSocketReturnsProcessExitError(t *testing.T) {166func TestWaitForVmmSocketReturnsProcessExitError(t *testing.T) {
167 socketPath := filepath.Join(t.TempDir(), "qmp.sock")167 socketPath := filepath.Join(t.TempDir(), "qmp.sock")
168 processErr := errors.New("stratovirt exited before creating qmp socket")168 processErr := errors.New("stratovirt exited before creating qmp socket")
169- processExited := make(chan error, 1)169+ processExited := &testProcessExit{done: make(chan struct{}), err: processErr}
170 ctx, cancel := context.WithTimeout(context.Background(), time.Second)170 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
171 defer cancel()171 defer cancel()
172 172 
173- processExited <- processErr173+ close(processExited.done)
174- close(processExited)
175 174 
176 err := waitForVmmSocket(ctx, socketPath, processExited)175 err := waitForVmmSocket(ctx, socketPath, processExited)
177 if !errors.Is(err, processErr) {176 if !errors.Is(err, processErr) {
@@ -181,3 +180,11 @@ func TestWaitForVmmSocketReturnsProcessExitError(t *testing.T) {
181 t.Fatalf("waitForVmmSocket() error = %q, want early exit context", err.Error())180 t.Fatalf("waitForVmmSocket() error = %q, want early exit context", err.Error())
182 }181 }
183}182}
183+ 
184+type testProcessExit struct {
185+ done chan struct{}
186+ err error
187+}
188+ 
189+func (p *testProcessExit) Done() <-chan struct{} { return p.done }
190+func (p *testProcessExit) Err() error { return p.err }
Minternal/volume/virtiofs.go+2-2
@@ -15,7 +15,7 @@ import (
15 "github.com/moby/sys/mountinfo"15 "github.com/moby/sys/mountinfo"
16 "golang.org/x/sys/unix"16 "golang.org/x/sys/unix"
17 17 
18- "github.com/openeuler/Conch/internal/sandboxid"18+ "github.com/openeuler/Conch/internal/id"
19 "github.com/openeuler/Conch/pkg/ulog"19 "github.com/openeuler/Conch/pkg/ulog"
20)20)
21 21 
@@ -298,7 +298,7 @@ func (b *virtiofsBackend) CleanupStaleResources() error {
298 return errors.Join(append(errs, err)...)298 return errors.Join(append(errs, err)...)
299 }299 }
300 for _, entry := range entries {300 for _, entry := range entries {
301- if !entry.IsDir() || sandboxid.Validate(entry.Name()) != nil {301+ if !entry.IsDir() || id.Validate(entry.Name()) != nil {
302 continue302 continue
303 }303 }
304 if cleanupErr := b.Cleanup(entry.Name(), nil); cleanupErr != nil {304 if cleanupErr := b.Cleanup(entry.Name(), nil); cleanupErr != nil {
Ainternal/webhook/errors.go+8-0
@@ -0,0 +1,8 @@
1+package webhook
2+ 
3+import "github.com/openeuler/Conch/internal/apperror"
4+ 
5+var (
6+ ErrInvalidArgument = apperror.Define(apperror.InvalidArgument, "webhook.invalid_argument", "invalid webhook argument")
7+ ErrNotFound = apperror.Define(apperror.NotFound, "webhook.not_found", "webhook not found")
8+)
Ainternal/webhook/webhook.go+191-0
@@ -0,0 +1,191 @@
1+package webhook
2+ 
3+import (
4+ "bytes"
5+ "encoding/json"
6+ "fmt"
7+ "net/http"
8+ "net/url"
9+ "sort"
10+ "strings"
11+ "sync"
12+ "time"
13+ 
14+ "github.com/openeuler/Conch/internal/id"
15+ "github.com/openeuler/Conch/internal/runtimeapi"
16+ "github.com/openeuler/Conch/pkg/ulog"
17+)
18+ 
19+const (
20+ EventSandboxCreated = "sandbox.lifecycle.created"
21+ EventSandboxKilled = "sandbox.lifecycle.killed"
22+)
23+ 
24+var supportedEvents = map[string]struct{}{
25+ EventSandboxCreated: {},
26+ EventSandboxKilled: {},
27+}
28+ 
29+type Execution struct {
30+ CreatedAt string `json:"created_at"`
31+ VCPUNum int64 `json:"vcpu_num"`
32+ RamMB int64 `json:"ram_mb"`
33+}
34+ 
35+type EventData struct {
36+ KillReason string `json:"kill_reason,omitempty"`
37+ Execution Execution `json:"execution"`
38+}
39+ 
40+type Event struct {
41+ EventID string `json:"event_id"`
42+ Version string `json:"version"`
43+ Type string `json:"type"`
44+ Timestamp string `json:"timestamp"`
45+ SandboxID string `json:"sandbox_id"`
46+ EventData EventData `json:"event_data"`
47+}
48+ 
49+// Dispatcher stores webhook registrations only in memory and dispatches events asynchronously.
50+type Dispatcher struct {
51+ mu sync.RWMutex
52+ webhooks map[string]runtimeapi.WebhookRecord
53+ client *http.Client
54+}
55+ 
56+func NewDispatcher() *Dispatcher {
57+ return &Dispatcher{
58+ webhooks: make(map[string]runtimeapi.WebhookRecord),
59+ client: &http.Client{Timeout: 10 * time.Second},
60+ }
61+}
62+ 
63+func (d *Dispatcher) Create(opts runtimeapi.WebhookCreateOptions) (runtimeapi.WebhookRecord, error) {
64+ name := strings.TrimSpace(opts.Name)
65+ if name == "" {
66+ return runtimeapi.WebhookRecord{}, ErrInvalidArgument.Wrap(fmt.Errorf("name is required"))
67+ }
68+ parsedURL, err := url.ParseRequestURI(strings.TrimSpace(opts.URL))
69+ if err != nil || parsedURL.Host == "" || (parsedURL.Scheme != "http" && parsedURL.Scheme != "https") {
70+ return runtimeapi.WebhookRecord{}, ErrInvalidArgument.Wrap(fmt.Errorf("url must be a valid HTTP or HTTPS URL"))
71+ }
72+ events, err := normalizeEvents(opts.Events)
73+ if err != nil {
74+ return runtimeapi.WebhookRecord{}, ErrInvalidArgument.Wrap(err)
75+ }
76+ webhookID, err := id.NewWithPrefix("wh_")
77+ if err != nil {
78+ return runtimeapi.WebhookRecord{}, err
79+ }
80+ hook := runtimeapi.WebhookRecord{WebhookID: webhookID, Name: name, URL: parsedURL.String(), Events: events, CreatedAt: time.Now().UTC()}
81+ d.mu.Lock()
82+ d.webhooks[hook.WebhookID] = hook
83+ d.mu.Unlock()
84+ return hook, nil
85+}
86+ 
87+func (d *Dispatcher) List() []runtimeapi.WebhookRecord {
88+ if d == nil {
89+ return []runtimeapi.WebhookRecord{}
90+ }
91+ d.mu.RLock()
92+ hooks := make([]runtimeapi.WebhookRecord, 0, len(d.webhooks))
93+ for _, hook := range d.webhooks {
94+ hook.Events = append([]string(nil), hook.Events...)
95+ hooks = append(hooks, hook)
96+ }
97+ d.mu.RUnlock()
98+ sort.Slice(hooks, func(i, j int) bool { return hooks[i].CreatedAt.Before(hooks[j].CreatedAt) })
99+ return hooks
100+}
101+ 
102+func (d *Dispatcher) Delete(webhookID string) bool {
103+ if d == nil {
104+ return false
105+ }
106+ d.mu.Lock()
107+ _, found := d.webhooks[webhookID]
108+ delete(d.webhooks, webhookID)
109+ d.mu.Unlock()
110+ return found
111+}
112+ 
113+func (d *Dispatcher) Publish(event Event) {
114+ if d == nil || !isSupportedEvent(event.Type) {
115+ return
116+ }
117+ d.mu.RLock()
118+ for _, hook := range d.webhooks {
119+ if subscribesTo(hook, event.Type) {
120+ go d.deliver(hook, event)
121+ }
122+ }
123+ d.mu.RUnlock()
124+}
125+ 
126+func (d *Dispatcher) deliver(hook runtimeapi.WebhookRecord, event Event) {
127+ body, err := json.Marshal(event)
128+ if err != nil {
129+ ulog.GetLogger().Error("failed to marshal webhook event", ulog.F("event_id", event.EventID), ulog.F("webhook_id", hook.WebhookID), ulog.F("error", err))
130+ return
131+ }
132+ var lastErr error
133+ for attempt := 0; attempt < 3; attempt++ {
134+ req, err := http.NewRequest(http.MethodPost, hook.URL, bytes.NewReader(body))
135+ if err == nil {
136+ req.Header.Set("Content-Type", "application/json")
137+ req.Header.Set("conch-webhook-id", hook.WebhookID)
138+ resp, doErr := d.client.Do(req)
139+ if doErr == nil {
140+ resp.Body.Close()
141+ if resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices {
142+ return
143+ }
144+ lastErr = fmt.Errorf("unexpected HTTP status %d", resp.StatusCode)
145+ } else {
146+ lastErr = doErr
147+ }
148+ } else {
149+ lastErr = err
150+ }
151+ }
152+ ulog.GetLogger().Error("webhook delivery failed after retries", ulog.F("event_id", event.EventID), ulog.F("webhook_id", hook.WebhookID), ulog.F("url", hook.URL), ulog.F("error", lastErr))
153+}
154+ 
155+func NewEvent(eventType, sandboxID, killReason string, execution Execution) (Event, error) {
156+ eventID, err := id.NewWithPrefix("evt_")
157+ if err != nil {
158+ return Event{}, err
159+ }
160+ return Event{EventID: eventID, Version: "v1", Type: eventType, Timestamp: time.Now().UTC().Format(time.RFC3339), SandboxID: sandboxID, EventData: EventData{KillReason: killReason, Execution: execution}}, nil
161+}
162+ 
163+func normalizeEvents(events []string) ([]string, error) {
164+ if len(events) == 0 {
165+ return []string{EventSandboxCreated, EventSandboxKilled}, nil
166+ }
167+ seen := make(map[string]struct{}, len(events))
168+ result := make([]string, 0, len(events))
169+ for _, event := range events {
170+ event = strings.TrimSpace(event)
171+ if !isSupportedEvent(event) {
172+ return nil, fmt.Errorf("unsupported event %q", event)
173+ }
174+ if _, exists := seen[event]; !exists {
175+ seen[event] = struct{}{}
176+ result = append(result, event)
177+ }
178+ }
179+ return result, nil
180+}
181+ 
182+func subscribesTo(hook runtimeapi.WebhookRecord, eventType string) bool {
183+ for _, event := range hook.Events {
184+ if event == eventType {
185+ return true
186+ }
187+ }
188+ return false
189+}
190+ 
191+func isSupportedEvent(event string) bool { _, ok := supportedEvents[event]; return ok }
Ainternal/webhook/webhook_test.go+119-0
@@ -0,0 +1,119 @@
1+package webhook
2+ 
3+import (
4+ "encoding/json"
5+ "net/http"
6+ "net/http/httptest"
7+ "sync/atomic"
8+ "testing"
9+ "time"
10+ 
11+ "github.com/openeuler/Conch/internal/runtimeapi"
12+)
13+ 
14+func TestCreateValidatesAndDefaultsEvents(t *testing.T) {
15+ dispatcher := NewDispatcher()
16+ if _, err := dispatcher.Create(runtimeapi.WebhookCreateOptions{Name: " ", URL: "https://example.test"}); err == nil {
17+ t.Fatal("Create accepted an empty name")
18+ }
19+ if _, err := dispatcher.Create(runtimeapi.WebhookCreateOptions{Name: "test", URL: "file:///tmp/events"}); err == nil {
20+ t.Fatal("Create accepted a non-HTTP URL")
21+ }
22+ if _, err := dispatcher.Create(runtimeapi.WebhookCreateOptions{Name: "test", URL: "https://example.test", Events: []string{"unknown"}}); err == nil {
23+ t.Fatal("Create accepted an unsupported event")
24+ }
25+ hook, err := dispatcher.Create(runtimeapi.WebhookCreateOptions{Name: "test", URL: "https://example.test"})
26+ if err != nil {
27+ t.Fatalf("Create: %v", err)
28+ }
29+ if hook.WebhookID == "" || len(hook.Events) != 2 || hook.CreatedAt.IsZero() {
30+ t.Fatalf("webhook = %#v, want webhook ID, timestamps and default subscriptions", hook)
31+ }
32+ if !dispatcher.Delete(hook.WebhookID) || dispatcher.Delete(hook.WebhookID) {
33+ t.Fatal("Delete did not report existing then absent webhook")
34+ }
35+}
36+ 
37+func TestPublishRetriesAndPreservesEventID(t *testing.T) {
38+ var calls atomic.Int32
39+ events := make(chan Event, 3)
40+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
41+ calls.Add(1)
42+ if r.Header.Get("Content-Type") != "application/json" || r.Header.Get("conch-webhook-id") == "" {
43+ t.Errorf("headers = %#v", r.Header)
44+ }
45+ var event Event
46+ if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
47+ t.Errorf("decode event: %v", err)
48+ }
49+ events <- event
50+ if calls.Load() < 3 {
51+ w.WriteHeader(http.StatusBadGateway)
52+ }
53+ }))
54+ defer server.Close()
55+ 
56+ dispatcher := NewDispatcher()
57+ if _, err := dispatcher.Create(runtimeapi.WebhookCreateOptions{Name: "receiver", URL: server.URL, Events: []string{EventSandboxCreated}}); err != nil {
58+ t.Fatalf("Create: %v", err)
59+ }
60+ event, err := NewEvent(EventSandboxCreated, "sandbox-a", "", Execution{CreatedAt: "2026-08-21T10:00:00Z", VCPUNum: 2, RamMB: 512})
61+ if err != nil {
62+ t.Fatalf("NewEvent: %v", err)
63+ }
64+ dispatcher.Publish(event)
65+ for i := 0; i < 3; i++ {
66+ select {
67+ case received := <-events:
68+ if received.EventID != event.EventID || received.Version != "v1" || received.SandboxID != "sandbox-a" {
69+ t.Fatalf("event = %#v", received)
70+ }
71+ case <-time.After(time.Second):
72+ t.Fatalf("delivery %d did not arrive", i+1)
73+ }
74+ }
75+ if got := calls.Load(); got != 3 {
76+ t.Fatalf("calls = %d, want 3", got)
77+ }
78+}
79+ 
80+func TestDeletePreventsSubsequentPublish(t *testing.T) {
81+ events := make(chan Event, 2)
82+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
83+ var event Event
84+ if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
85+ t.Errorf("decode event: %v", err)
86+ return
87+ }
88+ events <- event
89+ }))
90+ defer server.Close()
91+ dispatcher := NewDispatcher()
92+ hook, err := dispatcher.Create(runtimeapi.WebhookCreateOptions{Name: "receiver", URL: server.URL})
93+ if err != nil {
94+ t.Fatalf("Create: %v", err)
95+ }
96+ first, err := NewEvent(EventSandboxCreated, "sandbox-a", "", Execution{})
97+ if err != nil {
98+ t.Fatalf("NewEvent: %v", err)
99+ }
100+ dispatcher.Publish(first)
101+ select {
102+ case <-events:
103+ case <-time.After(time.Second):
104+ t.Fatal("first event not delivered")
105+ }
106+ if !dispatcher.Delete(hook.WebhookID) {
107+ t.Fatal("Delete returned false")
108+ }
109+ second, err := NewEvent(EventSandboxCreated, "sandbox-a", "", Execution{})
110+ if err != nil {
111+ t.Fatalf("NewEvent: %v", err)
112+ }
113+ dispatcher.Publish(second)
114+ select {
115+ case event := <-events:
116+ t.Fatalf("event delivered after deletion: %#v", event)
117+ case <-time.After(100 * time.Millisecond):
118+ }
119+}