Skip to content

wrapper goroutines with safego #670

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Pangjiping wants to merge 3 commits into
base: main
Choose a base branch
from
Open

wrapper goroutines with safego #670

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
822392a
chore(execd): extract safego to internal common package
Pangjiping Apr 8, 2026
f4a6852
feat(egress): wrapper goroutines with safego package
Pangjiping Apr 8, 2026
10a3f10
feat(execd): wrapper goroutines with safego package
Pangjiping Apr 8, 2026
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
3 changes: 3 additions & 0 deletions components/egress/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ require (
github.com/stretchr/testify v1.11.1
go.opentelemetry.io/otel v1.41.0
go.opentelemetry.io/otel/metric v1.41.0
go.uber.org/automaxprocs v1.6.0
golang.org/x/sys v0.41.0
)

Expand Down Expand Up @@ -38,6 +39,8 @@ require (
google.golang.org/grpc v1.79.3 // indirect
google.golang.org/protobuf v1.36.11 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
k8s.io/apimachinery v0.34.2 // indirect
k8s.io/klog/v2 v2.130.1 // indirect
)

replace github.com/alibaba/opensandbox/internal => ../internal
8 changes: 8 additions & 0 deletions components/egress/go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@ github.com/miekg/dns v1.1.61 h1:nLxbwF3XxhwVSm8g9Dghm9MHPaUZuqhPiGL+675ZmEs=
github.com/miekg/dns v1.1.61/go.mod h1:mnAarhS3nWaW+NVP2wTkYVIZyHNJ098SJZUki3eykwQ=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/prashantv/gostub v1.1.0 h1:BTyx3RfQjRHnUWaGF9oQos79AlQ5k8WNktv7VGvVH4g=
github.com/prashantv/gostub v1.1.0/go.mod h1:A5zLQHz7ieHGG7is6LLXLz7I8+3LZzsrV0P1IAHhP5U=
github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
Expand All @@ -45,6 +47,8 @@ go.opentelemetry.io/otel/trace v1.41.0 h1:Vbk2co6bhj8L59ZJ6/xFTskY+tGAbOnCtQGVVa
go.opentelemetry.io/otel/trace v1.41.0/go.mod h1:U1NU4ULCoxeDKc09yCWdWe+3QoyweJcISEVa1RBzOis=
go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A=
go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4=
go.uber.org/automaxprocs v1.6.0 h1:O3y2/QNTOdbF+e/dpXNNW7Rx2hZ4sTIPyybbxyNqTUs=
go.uber.org/automaxprocs v1.6.0/go.mod h1:ifeIMSnPZuznNm6jmdzmU3/bfk01Fe2fotchwEFJ8r8=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
Expand Down Expand Up @@ -78,3 +82,7 @@ gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntN
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
k8s.io/apimachinery v0.34.2 h1:zQ12Uk3eMHPxrsbUJgNF8bTauTVR2WgqJsTmwTE/NW4=
k8s.io/apimachinery v0.34.2/go.mod h1:/GwIlEcWuTX9zKIg2mbw0LRFIsXwrfoVxn+ef0X13lw=
k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk=
k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
5 changes: 5 additions & 0 deletions components/egress/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,9 @@ import (
"syscall"
"time"

_ "github.com/alibaba/opensandbox/internal/safego"
_ "go.uber.org/automaxprocs/maxprocs"

"github.com/alibaba/opensandbox/egress/pkg/constants"
"github.com/alibaba/opensandbox/egress/pkg/dnsproxy"
"github.com/alibaba/opensandbox/egress/pkg/events"
Expand All @@ -31,6 +34,7 @@ import (
"github.com/alibaba/opensandbox/egress/pkg/policy"
"github.com/alibaba/opensandbox/egress/pkg/telemetry"
slogger "github.com/alibaba/opensandbox/internal/logger"
"github.com/alibaba/opensandbox/internal/safego"
"github.com/alibaba/opensandbox/internal/version"
)

Expand Down Expand Up @@ -119,6 +123,7 @@ func withLogger(ctx context.Context) context.Context {
base = base.With(extra...)
}
logger := base.Named("opensandbox.egress")
safego.InitPanicLogger(ctx, logger)
return log.WithLogger(ctx, logger)
}

Expand Down
7 changes: 4 additions & 3 deletions components/egress/pkg/dnsproxy/proxy.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ import (
"github.com/alibaba/opensandbox/egress/pkg/policy"
"github.com/alibaba/opensandbox/egress/pkg/telemetry"
slogger "github.com/alibaba/opensandbox/internal/logger"
"github.com/alibaba/opensandbox/internal/safego"
)

const defaultListenAddr = "127.0.0.1:15353"
Expand Down Expand Up @@ -122,11 +123,11 @@ func (p *Proxy) Start(ctx context.Context) error {
errCh := make(chan error, len(p.servers))
for _, srv := range p.servers {
s := srv
go func() {
safego.Go(func() {
if err := s.ListenAndServe(); err != nil {
errCh <- err
}
}()
})
}

timer := time.NewTimer(200 * time.Millisecond)
Expand All @@ -138,7 +139,7 @@ func (p *Proxy) Start(ctx context.Context) error {
// listeners bound; start upstream probes only after DNS servers are up
}

go p.runUpstreamProbes(ctx)
safego.Go(func() { p.runUpstreamProbes(ctx) })

return nil
}
Expand Down
8 changes: 6 additions & 2 deletions components/egress/pkg/dnsproxy/upstream.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@ import (

"github.com/miekg/dns"

"github.com/alibaba/opensandbox/internal/safego"

"github.com/alibaba/opensandbox/egress/pkg/constants"
"github.com/alibaba/opensandbox/egress/pkg/log"
)
Expand Down Expand Up @@ -97,10 +99,12 @@ func (p *Proxy) probeUpstreams() {
var wg sync.WaitGroup
for i := range all {
wg.Add(1)
go func(idx int, addr string) {
idx := i
addr := all[i]
safego.Go(func() {
defer wg.Done()
healthy[idx] = p.probeOneUpstream(addr, timeout)
}(i, all[i])
})
}
wg.Wait()

Expand Down
5 changes: 3 additions & 2 deletions components/egress/pkg/events/broadcaster.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import (
"time"

"github.com/alibaba/opensandbox/egress/pkg/log"
"github.com/alibaba/opensandbox/internal/safego"
)

const defaultQueueSize = 128
Expand Down Expand Up @@ -76,7 +77,7 @@ func (b *Broadcaster) AddSubscriber(sub Subscriber) {
b.subscribers = append(b.subscribers, ch)
b.mu.Unlock()

go func() {
safego.Go(func() {
for {
select {
case <-b.ctx.Done():
Expand All @@ -88,7 +89,7 @@ func (b *Broadcaster) AddSubscriber(sub Subscriber) {
sub.HandleBlocked(b.ctx, ev)
}
}
}()
})
}

// Publish sends an event to all subscribers; drops and logs when a subscriber queue is full.
Expand Down
9 changes: 5 additions & 4 deletions components/egress/policy_server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import (
"github.com/alibaba/opensandbox/egress/pkg/log"
"github.com/alibaba/opensandbox/egress/pkg/nftables"
"github.com/alibaba/opensandbox/egress/pkg/policy"
"github.com/alibaba/opensandbox/internal/safego"
)

type policyUpdater interface {
Expand Down Expand Up @@ -81,22 +82,22 @@ func startPolicyServer(proxy policyUpdater, nft nftApplier, enforcementMode stri
handler.server = srv

errCh := make(chan error, 1)
go func() {
safego.Go(func() {
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
errCh <- err
}
}()
})

select {
case err := <-errCh:
return nil, err
case <-time.After(200 * time.Millisecond):
// assume healthy start; keep logging future errors
go func() {
safego.Go(func() {
if err := <-errCh; err != nil {
log.Errorf("policy server error: %v", err)
}
}()
})
return srv, nil
}
}
Expand Down
2 changes: 1 addition & 1 deletion components/execd/DEVELOPMENT.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,7 +159,7 @@ logs.Debug("received event: type=%s", eventType)
Always use `safego.Go` to prevent panics:

```go
import "github.com/alibaba/opensandbox/execd/pkg/util/safego"
import "github.com/alibaba/opensandbox/internal/safego"

safego.Go(func() {
processInBackground()
Expand Down
6 changes: 3 additions & 3 deletions components/execd/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ require (
github.com/yusufpapurcu/wmi v1.2.4 // indirect
go.uber.org/multierr v1.11.0 // indirect
go.uber.org/zap v1.27.1 // indirect
go.yaml.in/yaml/v2 v2.4.2 // indirect
go.yaml.in/yaml/v2 v2.4.3 // indirect
golang.org/x/arch v0.8.0 // indirect
golang.org/x/crypto v0.48.0 // indirect
golang.org/x/net v0.50.0 // indirect
Expand All @@ -63,8 +63,8 @@ require (
gopkg.in/inf.v0 v0.9.1 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
k8s.io/klog/v2 v2.130.1 // indirect
k8s.io/utils v0.0.0-20250604170112-4c0f3b243397 // indirect
sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 // indirect
k8s.io/utils v0.0.0-20251002143259-bc988d571ff4 // indirect
sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect
sigs.k8s.io/randfill v1.0.0 // indirect
sigs.k8s.io/structured-merge-diff/v6 v6.3.0 // indirect
)
Expand Down
12 changes: 6 additions & 6 deletions components/execd/go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,8 +125,8 @@ go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
go.uber.org/zap v1.27.1 h1:08RqriUEv8+ArZRYSTXy1LeBScaMpVSTBhCeaZYfMYc=
go.uber.org/zap v1.27.1/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
go.yaml.in/yaml/v2 v2.4.3 h1:6gvOSjQoTB3vt1l+CU+tSyi/HOjfOjRLJ4YwYZGwRO0=
go.yaml.in/yaml/v2 v2.4.3/go.mod h1:zSxWcmIDjOzPXpjlTTbAsKokqkDNAVtZO0WOMiT90s8=
golang.org/x/arch v0.0.0-20210923205945-b76863e36670/go.mod h1:5om86z9Hs0C8fWVUuoMHwpExlXzs5Tkyp9hOrfG7pp8=
golang.org/x/arch v0.8.0 h1:3wRIsP3pM4yUptoR96otTUOXI367OS0+c9eeRi9doIc=
golang.org/x/arch v0.8.0/go.mod h1:FEVrYAQjsQXMVJ1nsMoVVXPZg6p2JE2mx8psSWTDQys=
Expand Down Expand Up @@ -182,12 +182,12 @@ k8s.io/client-go v0.34.2 h1:Co6XiknN+uUZqiddlfAjT68184/37PS4QAzYvQvDR8M=
k8s.io/client-go v0.34.2/go.mod h1:2VYDl1XXJsdcAxw7BenFslRQX28Dxz91U9MWKjX97fE=
k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk=
k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
k8s.io/utils v0.0.0-20250604170112-4c0f3b243397 h1:hwvWFiBzdWw1FhfY1FooPn3kzWuJ8tmbZBHi4zVsl1Y=
k8s.io/utils v0.0.0-20250604170112-4c0f3b243397/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
k8s.io/utils v0.0.0-20251002143259-bc988d571ff4 h1:SjGebBtkBqHFOli+05xYbK8YF1Dzkbzn+gDM4X9T4Ck=
k8s.io/utils v0.0.0-20251002143259-bc988d571ff4/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
nullprogram.com/x/optparse v1.0.0/go.mod h1:KdyPE+Igbe0jQUrVfMqDMeJQIJZEuyV7pjYmp6pbG50=
rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4=
sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 h1:gBQPwqORJ8d8/YNZWEjoZs7npUVDpVXUUOFfW6CgAqE=
sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg=
sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 h1:IpInykpT6ceI+QxKBbEflcR5EXP7sU1kvOlxwZh5txg=
sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg=
sigs.k8s.io/randfill v1.0.0 h1:JfjMILfT8A6RbawdsK2JXGBR5AQVfd+9TbzrlneTyrU=
sigs.k8s.io/randfill v1.0.0/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY=
sigs.k8s.io/structured-merge-diff/v6 v6.3.0 h1:jTijUJbW353oVOd9oTlifJqOGEkUw2jB/fXCbTiQEco=
Expand Down
2 changes: 1 addition & 1 deletion components/execd/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,12 +20,12 @@ import (

"github.com/alibaba/opensandbox/internal/version"

_ "github.com/alibaba/opensandbox/internal/safego"
_ "go.uber.org/automaxprocs/maxprocs"

"github.com/alibaba/opensandbox/execd/pkg/clone3compat"
"github.com/alibaba/opensandbox/execd/pkg/flag"
"github.com/alibaba/opensandbox/execd/pkg/log"
_ "github.com/alibaba/opensandbox/execd/pkg/util/safego"
"github.com/alibaba/opensandbox/execd/pkg/web"
"github.com/alibaba/opensandbox/execd/pkg/web/controller"
)
Expand Down
5 changes: 3 additions & 2 deletions components/execd/pkg/jupyter/execute/execute.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"sync"
"time"

"github.com/alibaba/opensandbox/internal/safego"
"github.com/google/uuid"
"github.com/gorilla/websocket"

Expand Down Expand Up @@ -90,7 +91,7 @@ func (c *Client) Connect(wsURL string) error {
c.registerDefaultHandlers()

// Start message receiving goroutine
go c.receiveMessages()
safego.Go(func() { c.receiveMessages() })

return nil
}
Expand Down Expand Up @@ -286,7 +287,7 @@ func (c *Client) handleExecutionStatus(msg *Message, state *streamExecutionState
return
}
state.executeDone = true
go c.finalizeExecution(state, resultChan)
safego.Go(func() { c.finalizeExecution(state, resultChan) })
}

func (c *Client) finalizeExecution(state *streamExecutionState, resultChan chan *ExecutionResult) {
Expand Down
3 changes: 3 additions & 0 deletions components/execd/pkg/log/log.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,11 @@
package log

import (
"context"
"os"

slogger "github.com/alibaba/opensandbox/internal/logger"
"github.com/alibaba/opensandbox/internal/safego"
)

const logFileEnvKey = "EXECD_LOG_FILE"
Expand All @@ -28,6 +30,7 @@ var current slogger.Logger
// Legacy levels: 0/1/2=fatal, 3=error, 4=warn, 5/6=info, 7+=debug.
func Init(level int) {
current = newLogger(mapLevel(level))
safego.InitPanicLogger(context.Background(), current)
}

func mapLevel(level int) string {
Expand Down
13 changes: 7 additions & 6 deletions components/execd/pkg/runtime/bash_session_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ import (
"github.com/stretchr/testify/require"

"github.com/alibaba/opensandbox/execd/pkg/jupyter/execute"
"github.com/alibaba/opensandbox/internal/safego"
)

func TestBashSession_NonZeroExitEmitsError(t *testing.T) {
Expand Down Expand Up @@ -519,9 +520,9 @@ func TestBashSession_CloseKillsRunningProcess(t *testing.T) {
Timeout: 60 * time.Second,
Hooks: ExecuteResultHook{},
}
go func() {
safego.Go(func() {
runDone <- session.run(context.Background(), req)
}()
})

// Give the child process time to start.
time.Sleep(200 * time.Millisecond)
Expand Down Expand Up @@ -557,9 +558,9 @@ func TestBashSession_DeleteBashSessionKillsRunningProcess(t *testing.T) {
Timeout: 60 * time.Second,
Hooks: ExecuteResultHook{},
}
go func() {
safego.Go(func() {
runDone <- c.RunInBashSession(context.Background(), req)
}()
})

time.Sleep(200 * time.Millisecond)

Expand All @@ -585,10 +586,10 @@ func TestBashSession_CloseWithNoActiveRun(t *testing.T) {
require.NoError(t, session.start())

done := make(chan struct{}, 1)
go func() {
safego.Go(func() {
_ = session.close()
done <- struct{}{}
}()
})

select {
case <-done:
Expand Down
7 changes: 4 additions & 3 deletions components/execd/pkg/runtime/command.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,9 +30,10 @@ import (
"syscall"
"time"

"github.com/alibaba/opensandbox/internal/safego"

"github.com/alibaba/opensandbox/execd/pkg/jupyter/execute"
"github.com/alibaba/opensandbox/execd/pkg/log"
"github.com/alibaba/opensandbox/execd/pkg/util/safego"
)

// getShell returns the preferred shell, falling back to sh if bash is not available.
Expand Down Expand Up @@ -157,7 +158,7 @@ func (c *Controller) runCommand(ctx context.Context, request *ExecuteCodeRequest
c.storeCommandKernel(session, kernel)
request.Hooks.OnExecuteInit(session)

go func() {
safego.Go(func() {
for {
select {
case <-ctx.Done():
Expand All @@ -172,7 +173,7 @@ func (c *Controller) runCommand(ctx context.Context, request *ExecuteCodeRequest
}
}
}
}()
})

err = cmd.Wait()
close(done)
Expand Down
2 changes: 1 addition & 1 deletion components/execd/pkg/runtime/command_windows.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import (

"github.com/alibaba/opensandbox/execd/pkg/jupyter/execute"
"github.com/alibaba/opensandbox/execd/pkg/log"
"github.com/alibaba/opensandbox/execd/pkg/util/safego"
"github.com/alibaba/opensandbox/internal/safego"
)

// runCommand executes shell commands and streams their output on Windows.
Expand Down
Loading
Loading