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)
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