Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -188,6 +188,7 @@ crash.*.log

# Containers and local runtime data
/data/
/.octobus-mbs-mac/
/playground/
/qemu-compose.yml
docker-compose.override.yml
Expand Down
23 changes: 18 additions & 5 deletions internal/protocol/gateway.go
Original file line number Diff line number Diff line change
Expand Up @@ -1594,24 +1594,29 @@ func (g *Gateway) invokeOnDemand(ctx context.Context, item store.ExposedMethod,
if !info.Mode().IsRegular() {
return nil, status.Errorf(codes.Internal, "runtime entry %q is not a regular file", item.Service.NodeEntry)
}
secretFile, closeSecret, err := secretReadFile(inst.SecretJSON)
secretArg, secretFile, closeSecret, err := runtimeSecretArg(workdir, inst.SecretJSON)
if err != nil {
return nil, status.Errorf(codes.Internal, "prepare secret fd: %v", err)
return nil, status.Errorf(codes.Internal, "prepare runtime secret: %v", err)
}
defer closeSecret()
cmd := exec.CommandContext(ctx, entry,
args := []string{
"--runtime",
"invoke",
"--method", item.Method.FullName,
"--config", filepath.Join(workdir, "config.json"),
"--secret-fd", "3",
}
args = append(args, secretArg...)
args = append(args,
"--metadata", metadataPath,
"--workdir", workdir,
"--service", item.Service.ID,
"--instance", item.Instance.ID,
)
cmd := exec.CommandContext(ctx, entry, args...)
cmd.Dir = workdir
cmd.ExtraFiles = []*os.File{secretFile}
if secretFile != nil {
cmd.ExtraFiles = []*os.File{secretFile}
}
cmd.Env = append(os.Environ(),
"OCTOBUS_SERVICE_ID="+item.Service.ID,
"OCTOBUS_INSTANCE_ID="+item.Instance.ID,
Expand Down Expand Up @@ -1658,6 +1663,14 @@ func secretReadFile(secret []byte) (*os.File, func(), error) {
return reader, closeFn, nil
}

func runtimeSecretArg(workdir string, secret []byte) ([]string, *os.File, func(), error) {
secretFile, closeFn, err := secretReadFile(secret)
if err != nil {
return nil, nil, nil, err
}
return []string{"--secret-fd", "3"}, secretFile, closeFn, nil
}

func (g *Gateway) validateOnDemandResponse(item store.ExposedMethod, raw []byte) error {
set, err := descriptors.Load(item.Service.DescriptorPath)
if err != nil {
Expand Down
48 changes: 39 additions & 9 deletions internal/supervisor/supervisor.go
Original file line number Diff line number Diff line change
Expand Up @@ -36,10 +36,19 @@ type Supervisor struct {
}

type processState struct {
cmd *exec.Cmd
done chan struct{}
attempt int
generation int64
cmd *exec.Cmd
done chan struct{}
attempt int
generation int64
cleanup func()
cleanupOnce sync.Once
}

func (p *processState) cleanupRuntimeSecret() {
if p == nil || p.cleanup == nil {
return
}
p.cleanupOnce.Do(p.cleanup)
}

type CreateInstanceRequest struct {
Expand Down Expand Up @@ -230,15 +239,20 @@ func (s *Supervisor) startWithAttempt(ctx context.Context, instanceID string, re
startErr = fmt.Errorf("runtime entry %q is not a regular file", svc.NodeEntry)
return startErr
}
secretFile, closeSecret, err := secretReadFile(inst.SecretJSON)
secretArg, secretFile, closeSecret, err := runtimeSecretArg(workdir, inst.SecretJSON)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

runtimeSecretArg 在两个 Go 包中定义了相同的未使用参数 workdir

internal/supervisor/supervisor.gointernal/protocol/gateway.go 中均新增了 runtimeSecretArg(workdir string, secret []byte) 函数,但函数体内完全没有使用 workdir 形参。该参数在当前实现中为死代码,增加了调用方的困惑,也暗示可能存在未完成的重构(例如计划将 secret 写入 workdir 下的临时文件,而非继续使用 memfd/pipe)。

Problem code:

Changed code at internal/supervisor/supervisor.go:242

Recommendation:
若短期内不需要基于 workdir 的 secret 文件策略,建议先从函数签名中移除 workdir 参数,保持接口简洁;若这是为后续功能预留的,应添加 TODO 注释说明,避免维护者困惑。也可以考虑将 runtimeSecretArg 提取到公共包中,消除 supervisor 与 gateway 之间的代码重复。

if err != nil {
startErr = err
return err
}
defer closeSecret()
cmd := exec.Command(entry, "--runtime", "serve", "--host", "127.0.0.1", "--port", fmt.Sprintf("%d", port), "--config", filepath.Join(workdir, "config.json"), "--secret-fd", "3", "--workdir", workdir, "--service", svc.ID, "--instance", instanceID)
cleanupSecret := closeSecret
args := []string{"--runtime", "serve", "--host", "127.0.0.1", "--port", fmt.Sprintf("%d", port), "--config", filepath.Join(workdir, "config.json")}
args = append(args, secretArg...)
args = append(args, "--workdir", workdir, "--service", svc.ID, "--instance", instanceID)
cmd := exec.Command(entry, args...)
cmd.Dir = workdir
cmd.ExtraFiles = []*os.File{secretFile}
if secretFile != nil {
cmd.ExtraFiles = []*os.File{secretFile}
}
cmd.Env = append(os.Environ(),
"OCTOBUS_SERVICE_ID="+svc.ID,
"OCTOBUS_INSTANCE_ID="+instanceID,
Expand All @@ -248,12 +262,14 @@ func (s *Supervisor) startWithAttempt(ctx context.Context, instanceID string, re
)
stdout, err := os.OpenFile(filepath.Join(workdir, "stdout.log"), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
if err != nil {
cleanupSecret()
startErr = err
return err
}
stderr, err := os.OpenFile(filepath.Join(workdir, "stderr.log"), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
if err != nil {
_ = stdout.Close()
cleanupSecret()
startErr = err
return err
}
Expand All @@ -265,12 +281,14 @@ func (s *Supervisor) startWithAttempt(ctx context.Context, instanceID string, re
if err := s.Store.UpsertInstance(ctx, inst); err != nil {
_ = stdout.Close()
_ = stderr.Close()
cleanupSecret()
startErr = err
return err
}
if err := cmd.Start(); err != nil {
_ = stdout.Close()
_ = stderr.Close()
cleanupSecret()
inst.Status = domain.StatusFailed
_ = s.Store.UpsertInstance(ctx, inst)
startErr = err
Expand All @@ -281,12 +299,13 @@ func (s *Supervisor) startWithAttempt(ctx context.Context, instanceID string, re
inst.Status = domain.StatusRunning
if err := s.Store.UpsertInstance(ctx, inst); err != nil {
_ = cmd.Process.Kill()
cleanupSecret()
startErr = err
return err
}
logger.Info("instance_started", "instance_id", instanceID, "pid", pid, "listen_addr", addr)
s.mu.Lock()
state := &processState{cmd: cmd, done: make(chan struct{}), attempt: restartAttempt, generation: generation}
state := &processState{cmd: cmd, done: make(chan struct{}), attempt: restartAttempt, generation: generation, cleanup: cleanupSecret}
s.procs[instanceID] = state
s.mu.Unlock()
if err := waitHealth(ctx, addr, 5*time.Second); err != nil {
Expand All @@ -313,6 +332,7 @@ func (s *Supervisor) cleanupFailedStart(instanceID string, state *processState,
_ = state.cmd.Process.Kill()
_ = state.cmd.Wait()
}
state.cleanupRuntimeSecret()
close(state.done)
_ = stdout.Close()
_ = stderr.Close()
Expand Down Expand Up @@ -414,6 +434,7 @@ func (s *Supervisor) stopProcess(ctx context.Context, inst domain.Instance, enab
case <-time.After(2 * time.Second):
}
}
state.cleanupRuntimeSecret()
}
if err := s.Store.UpsertInstance(ctx, inst); err != nil {
logger.Error("instance_stop_failed", "instance_id", inst.ID, "error", err)
Expand Down Expand Up @@ -523,10 +544,19 @@ func secretReadFile(secret []byte) (*os.File, func(), error) {
return reader, closeFn, nil
}

func runtimeSecretArg(workdir string, secret []byte) ([]string, *os.File, func(), error) {
secretFile, closeFn, err := secretReadFile(secret)
if err != nil {
return nil, nil, nil, err
}
return []string{"--secret-fd", "3"}, secretFile, closeFn, nil
}

func (s *Supervisor) wait(instanceID string, state *processState, stdout, stderr *os.File) {
err := state.cmd.Wait()
_ = stdout.Close()
_ = stderr.Close()
state.cleanupRuntimeSecret()
close(state.done)
s.mu.Lock()
current := s.procs[instanceID]
Expand Down
39 changes: 39 additions & 0 deletions internal/supervisor/supervisor_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1035,6 +1035,45 @@ func TestWaitMarksEnabledInstanceDegradedAfterProcessError(t *testing.T) {
}
}

func TestStopProcessCleansRuntimeSecretAfterWaitTimeout(t *testing.T) {
ctx := context.Background()
dataDir := filepath.Join(t.TempDir(), "data")
st, err := store.Open(filepath.Join(dataDir, "octobus.db"))
if err != nil {
t.Fatal(err)
}
defer st.Close()
if err := st.UpsertService(ctx, domain.Service{ID: "echo", Name: "Echo", PackageSource: "fixture", PackageArtifactPath: "pkg", PackageSHA256: "pkgsha", DescriptorPath: "desc", DescriptorSHA256: "descsha", DescriptorVersion: "descsha", NodeEntry: "entry"}); err != nil {
t.Fatal(err)
}
inst := domain.Instance{ID: "echo-test", ServiceID: "echo", Name: "Echo Test", Enabled: true, Status: domain.StatusRunning, NodeEntry: "entry", ConfigJSON: []byte(`{}`), SecretJSON: []byte(`{}`)}
if err := st.UpsertInstance(ctx, inst); err != nil {
t.Fatal(err)
}
sup := New(dataDir, st)
cmd := exec.Command("sh", "-c", "exit 0")
if err := cmd.Start(); err != nil {
t.Fatal(err)
}
cleanupCalls := 0
state := &processState{cmd: cmd, done: make(chan struct{}), cleanup: func() { cleanupCalls++ }}
sup.mu.Lock()
sup.procs["echo-test"] = state
sup.mu.Unlock()

if err := sup.stopProcess(ctx, inst, false); err != nil {
t.Fatal(err)
}
if cleanupCalls != 1 {
t.Fatalf("cleanup calls after stop timeout=%d want 1", cleanupCalls)
}
state.cleanupRuntimeSecret()
if cleanupCalls != 1 {
t.Fatalf("cleanup should be idempotent, calls=%d", cleanupCalls)
}
_ = cmd.Wait()
}

func TestWaitReturnsWhenInstanceDisabledOrStateReplaced(t *testing.T) {
ctx := context.Background()
dataDir := filepath.Join(t.TempDir(), "data")
Expand Down
12 changes: 10 additions & 2 deletions services/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,8 @@
"tencent-bh": "bin/tencent-bh.js",
"alibaba-sas": "bin/alibaba-sas.js",
"misp": "bin/misp.js",
"opencti": "bin/opencti.js"
"opencti": "bin/opencti.js",
"zhizhangyi-mbs": "zhizhangyi__mbs/bin/zhizhangyi-mbs.js"
},
"files": [
"bin/alibaba-cloud-simple-application-server-firewall.js",
Expand Down Expand Up @@ -173,7 +174,14 @@
"filigran__opencti",
"wangsu__label-ip",
"scripts",
"bin/octobus-tentacles.js"
"bin/octobus-tentacles.js",
"zhizhangyi__mbs/bin/zhizhangyi-mbs.js",
"zhizhangyi__mbs/config.schema.json",
"zhizhangyi__mbs/package.json",
"zhizhangyi__mbs/proto",
"zhizhangyi__mbs/secret.schema.json",
"zhizhangyi__mbs/service.json",
"zhizhangyi__mbs/src"
],
"scripts": {
"validate": "node scripts/validate-service-package.mjs",
Expand Down
7 changes: 7 additions & 0 deletions services/zhizhangyi__mbs/bin/zhizhangyi-mbs.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
#!/usr/bin/env node

import { runServiceMain } from "@chaitin-ai/octobus-sdk";

import { service } from "../src/service.js";

runServiceMain(service);
26 changes: 26 additions & 0 deletions services/zhizhangyi__mbs/config.schema.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"additionalProperties": true,
"properties": {
"endpoint": {
"type": "string",
"description": "MBS REST API base URL. Example: https://mbs.example.com:9074"
},
"baseUrl": {
"type": "string",
"description": "Legacy alias for endpoint."
},
"skipTlsVerify": {
"type": "boolean",
"default": false,
"description": "Skip TLS certificate verification. Enable only when the target MBS endpoint uses a trusted self-signed certificate."
},
"timeoutMs": {
"type": "integer",
"minimum": 1,
"default": 30000,
"description": "HTTP timeout in milliseconds."
}
}
}
20 changes: 20 additions & 0 deletions services/zhizhangyi__mbs/package.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
{
"name": "zhizhangyi-mbs",
"version": "0.0.0",
"private": true,
"type": "module",
"files": [
"bin/zhizhangyi-mbs.js",
"config.schema.json",
"secret.schema.json",
"service.json",
"proto",
"src"
],
"bin": {
"zhizhangyi-mbs": "bin/zhizhangyi-mbs.js"
},
"dependencies": {
"@chaitin-ai/octobus-sdk": "^0.5.0"
}
}
Loading