From 39a1ba609468ec83360eccc39507b2d093a694c7 Mon Sep 17 00:00:00 2001 From: edi-oai Date: Wed, 22 Jul 2026 01:28:44 +0100 Subject: [PATCH 1/2] Support dynamically updating Softnet policy --- .golangci.yml | 1 + go.mod | 5 +- go.sum | 10 +- internal/controller/api_vms.go | 24 ++- internal/tests/spec_update_test.go | 153 ++++++++++++++++- internal/worker/option.go | 7 + internal/worker/runtime/runtime.go | 1 + internal/worker/runtime/synthetic.go | 1 + internal/worker/runtime/tart.go | 3 +- internal/worker/runtime/vetu.go | 1 + internal/worker/vmmanager/base/cmd.go | 22 ++- internal/worker/vmmanager/stop_test.go | 2 +- .../worker/vmmanager/synthetic/synthetic.go | 4 + internal/worker/vmmanager/tart/cmd.go | 29 +++- .../worker/vmmanager/tart/softnet_control.go | 158 ++++++++++++++++++ internal/worker/vmmanager/tart/tart.go | 78 ++++++--- internal/worker/vmmanager/vetu/vetu.go | 5 + internal/worker/vmmanager/vmmanager.go | 1 + internal/worker/worker.go | 43 ++++- pkg/resource/v1/v1.go | 11 ++ 20 files changed, 526 insertions(+), 33 deletions(-) create mode 100644 internal/worker/vmmanager/tart/softnet_control.go diff --git a/.golangci.yml b/.golangci.yml index 4fc517d6..ab98339d 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -72,6 +72,7 @@ linters: # It's OK to not initialize some struct fields - exhaustruct + - exhaustruct_v5 # We'll control the variable name length ourselves - varnamelen diff --git a/go.mod b/go.mod index f6f1b917..03c81a65 100644 --- a/go.mod +++ b/go.mod @@ -44,8 +44,10 @@ require ( go.uber.org/zap v1.27.1 golang.org/x/crypto v0.49.0 golang.org/x/exp v0.0.0-20250218142911-aa4b98e5adaa + golang.org/x/exp/jsonrpc2 v0.0.0-20260718201538-764159d718ef golang.org/x/net v0.52.0 golang.org/x/sync v0.20.0 + golang.org/x/sys v0.46.0 golang.org/x/term v0.41.0 golang.org/x/text v0.35.0 google.golang.org/grpc v1.79.3 @@ -143,6 +145,7 @@ require ( go.uber.org/multierr v1.11.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/arch v0.24.0 // indirect - golang.org/x/sys v0.42.0 // indirect + golang.org/x/exp/event v0.0.0-20260611194520-c48552f49976 // indirect + golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 // indirect ) diff --git a/go.sum b/go.sum index 0edf211d..98e9609d 100644 --- a/go.sum +++ b/go.sum @@ -397,6 +397,10 @@ golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtC golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20250218142911-aa4b98e5adaa h1:t2QcU6V556bFjYgu4L6C+6VrCPyJZ+eyRsABUPs1mz4= golang.org/x/exp v0.0.0-20250218142911-aa4b98e5adaa/go.mod h1:BHOTPb3L19zxehTsLoJXVaTktb06DFgmdW6Wb9s8jqk= +golang.org/x/exp/event v0.0.0-20260611194520-c48552f49976 h1:aMc3mP8qjwSEdljznT5CYeyvpoFP7e3VCNqZZwUf3F0= +golang.org/x/exp/event v0.0.0-20260611194520-c48552f49976/go.mod h1:cV3mj8TmLcC/Zki//yCMYw/nSC1BpHamZNkMSU7lgPw= +golang.org/x/exp/jsonrpc2 v0.0.0-20260718201538-764159d718ef h1:VmBLplHGs0Tp6iOiqQMfK/Ku2m5mAMeFJAFzywb6NwI= +golang.org/x/exp/jsonrpc2 v0.0.0-20260718201538-764159d718ef/go.mod h1:it3Sp7J2OmdyHxl5q3GMU4d2QZsFQGglbij9kqQVVaI= golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE= golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU= golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= @@ -450,8 +454,8 @@ golang.org/x/sys v0.0.0-20221010170243-090e33056c14/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= -golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= +golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= @@ -480,6 +484,8 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da h1:noIWHXmPHxILtqtCOPIhSt0ABwskkZKjD3bXGnZGpNY= +golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da/go.mod h1:NDW/Ps6MPRej6fsCIbMTohpP40sJ/P/vI1MoTEGwX90= gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM= diff --git a/internal/controller/api_vms.go b/internal/controller/api_vms.go index 92afd840..f0dc9a37 100644 --- a/internal/controller/api_vms.go +++ b/internal/controller/api_vms.go @@ -1,6 +1,7 @@ package controller import ( + "context" "encoding/base64" "encoding/json" "errors" @@ -16,6 +17,7 @@ import ( "github.com/cirruslabs/orchard/internal/worker/ondiskname" "github.com/cirruslabs/orchard/pkg/client" "github.com/cirruslabs/orchard/pkg/resource/v1" + "github.com/cirruslabs/orchard/rpc" "github.com/gin-gonic/gin" "github.com/google/go-cmp/cmp" "github.com/google/uuid" @@ -163,8 +165,12 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder } name := ctx.Param("name") + var affectedWorker string + + response := controller.storeUpdate(func(txn storepkg.Transaction) responder.Responder { + // Re-initialize to guard against potential transaction retry + affectedWorker = "" - return controller.storeUpdate(func(txn storepkg.Transaction) responder.Responder { dbVM, err := txn.GetVM(name) if err != nil { return responder.Error(err) @@ -231,8 +237,24 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder return responder.Code(http.StatusInternalServerError) } + affectedWorker = dbVM.Worker + return responder.JSON(http.StatusOK, dbVM) }) + + if affectedWorker != "" { + notifyContext, cancel := context.WithTimeout(ctx.Request.Context(), time.Second) + defer cancel() + + if err := controller.workerNotifier.Notify(notifyContext, affectedWorker, &rpc.WatchInstruction{ + Action: &rpc.WatchInstruction_SyncVmsAction{}, + }); err != nil { + controller.logger.Warnf("failed to reactively sync updated VM %s on worker %s: %v", + name, affectedWorker, err) + } + } + + return response } func (controller *Controller) updateVMState(ctx *gin.Context) responder.Responder { diff --git a/internal/tests/spec_update_test.go b/internal/tests/spec_update_test.go index 9a61c505..95bac22a 100644 --- a/internal/tests/spec_update_test.go +++ b/internal/tests/spec_update_test.go @@ -3,13 +3,17 @@ package tests import ( "context" "fmt" + "net" + "os" "runtime" "testing" "time" + commandssh "github.com/cirruslabs/orchard/internal/command/ssh" "github.com/cirruslabs/orchard/internal/imageconstant" "github.com/cirruslabs/orchard/internal/tests/devcontroller" "github.com/cirruslabs/orchard/internal/tests/wait" + "github.com/cirruslabs/orchard/internal/worker" "github.com/cirruslabs/orchard/internal/worker/ondiskname" "github.com/cirruslabs/orchard/internal/worker/vmmanager" "github.com/cirruslabs/orchard/internal/worker/vmmanager/tart" @@ -18,6 +22,7 @@ import ( "github.com/shirou/gopsutil/v4/process" "github.com/stretchr/testify/require" "go.uber.org/zap" + "golang.org/x/crypto/ssh" ) func TestSpecUpdateSoftnet(t *testing.T) { @@ -25,7 +30,17 @@ func TestSpecUpdateSoftnet(t *testing.T) { t.Skip("Softnet is only supported on macOS with Tart") } - devClient, _, _ := devcontroller.StartIntegrationTestEnvironment(t) + if os.Getenv("ORCHARD_SKIP_SOFTNET_TESTS") != "" { + t.Skip("softnet tests require root") + } + + devClient, _, _ := devcontroller.StartIntegrationTestEnvironmentWithAdditionalOpts( + t, + false, + nil, + false, + []worker.Option{worker.WithSoftnetPolicyUpdates(false)}, + ) // Create a VM vmName := "test" @@ -92,7 +107,17 @@ func TestSpecUpdateSoftnetSuspendable(t *testing.T) { t.Skip("Softnet is only supported on macOS with Tart") } - devClient, _, _ := devcontroller.StartIntegrationTestEnvironment(t) + if os.Getenv("ORCHARD_SKIP_SOFTNET_TESTS") != "" { + t.Skip("softnet tests require root") + } + + devClient, _, _ := devcontroller.StartIntegrationTestEnvironmentWithAdditionalOpts( + t, + false, + nil, + false, + []worker.Option{worker.WithSoftnetPolicyUpdates(false)}, + ) // Create a suspendable VM with Softnet enabled vmName := "test" @@ -159,6 +184,130 @@ func TestSpecUpdateSoftnetSuspendable(t *testing.T) { require.True(t, sliceContainsAnotherSlice(tartRunCmdline, []string{"--net-softnet-block", "0.0.0.0/0"})) } +//nolint:gosec,modernize,perfsprint,staticcheck // preserve the original integration test +func TestSpecUpdateSoftnetPolicy(t *testing.T) { + if runtime.GOOS != "darwin" { + t.Skip("Softnet is only supported on macOS with Tart") + } + + if os.Getenv("ORCHARD_SKIP_SOFTNET_TESTS") != "" { + t.Skip("softnet tests require root") + } + + devClient, _, _ := devcontroller.StartIntegrationTestEnvironmentWithAdditionalOpts( + t, + false, + nil, + false, + []worker.Option{worker.WithSoftnetPolicyUpdates(true)}, + ) + + // Create a VM with Softnet enabled + vmName := "test" + + err := devClient.VMs().Create(t.Context(), &v1.VM{ + Meta: v1.Meta{ + Name: vmName, + }, + Image: imageconstant.DefaultMacosImage, + CPU: 4, + Memory: 8 * 1024, + Headless: true, + VMSpec: v1.VMSpec{ + NetSoftnet: true, + }, + }) + require.NoError(t, err) + + // Wait for the VM to start + var vm *v1.VM + + require.True(t, wait.Wait(2*time.Minute, func() bool { + vm, err = devClient.VMs().Get(context.Background(), vmName) + require.NoError(t, err) + + t.Logf("Waiting for the VM to start. Current status: %s", vm.Status) + + return vm.Status == v1.VMStatusRunning + }), "failed to start a VM") + + // Ensure that the VM is using "--net-softnet" + tartVMName := ondiskname.New(vmName, vm.UID, vm.RestartCount).String() + + tartRunCmdline, err := tartRunProcessCmdline(tartVMName) + require.NoError(t, err) + require.Contains(t, tartRunCmdline, "--net-softnet") + + // Connect to the VM over SSH + var netConn net.Conn + + require.True(t, wait.Wait(2*time.Minute, func() bool { + netConn, err = devClient.VMs().PortForward(t.Context(), vmName, 22, 120) + if err != nil { + t.Logf("Waiting for SSH to become available: %v", err) + } + + return err == nil + }), "failed to connect to the VM over SSH") + defer netConn.Close() + + username, password := commandssh.ChooseUsernameAndPassword(t.Context(), devClient, vmName, "", "") + + sshConn, chans, reqs, err := ssh.NewClientConn(netConn, "", &ssh.ClientConfig{ + User: username, + Auth: []ssh.AuthMethod{ssh.Password(password)}, + HostKeyCallback: ssh.InsecureIgnoreHostKey(), + }) + require.NoError(t, err) + + sshClient := ssh.NewClient(sshConn, chans, reqs) + defer sshClient.Close() + + curl := func(address string) error { + session, err := sshClient.NewSession() + require.NoError(t, err) + defer session.Close() + + return session.Run(fmt.Sprintf( + "/usr/bin/curl -4 -k -sS -o /dev/null --connect-timeout 5 --max-time 10 https://%s", + address, + )) + } + + // Ensure that the address is reachable before blocking it + require.NoError(t, curl("1.1.1.1")) + + // Update the Softnet policy + restartCountBeforePolicyUpdate := vm.RestartCount + vm.NetSoftnetBlock = []string{"1.1.1.1/32"} + + vm, err = devClient.VMs().Update(t.Context(), *vm) + require.NoError(t, err) + require.EqualValues(t, 1, vm.Generation) + require.EqualValues(t, 0, vm.ObservedGeneration) + + require.True(t, wait.Wait(2*time.Minute, func() bool { + vm, err = devClient.VMs().Get(context.Background(), vmName) + require.NoError(t, err) + + t.Logf("Waiting for the VM's observed generation to be updated...") + + return vm.ObservedGeneration == 1 + }), "failed to wait for the VM's observed generation to be updated") + + // Ensure that the policy was updated without restarting the VM + require.Equal(t, restartCountBeforePolicyUpdate, vm.RestartCount) + require.Equal(t, tartVMName, vm.TartName) + + updatedTartRunCmdline, err := tartRunProcessCmdline(tartVMName) + require.NoError(t, err) + require.Equal(t, tartRunCmdline, updatedTartRunCmdline) + + // Ensure that the new policy is applied without disrupting other traffic + require.Error(t, curl("1.1.1.1")) + require.NoError(t, curl("1.0.0.1")) +} + func TestSpecUpdatePowerStateSuspend(t *testing.T) { if runtime.GOOS != "darwin" { t.Skip("VM suspension is only supported on macOS with Tart") diff --git a/internal/worker/option.go b/internal/worker/option.go index f419bdd6..89b0f39e 100644 --- a/internal/worker/option.go +++ b/internal/worker/option.go @@ -4,6 +4,7 @@ import ( "github.com/cirruslabs/orchard/internal/dialer" "github.com/cirruslabs/orchard/internal/worker/runtime" v1 "github.com/cirruslabs/orchard/pkg/resource/v1" + "github.com/samber/mo" "go.uber.org/zap" ) @@ -52,6 +53,12 @@ func WithSynthetic() Option { } } +func WithSoftnetPolicyUpdates(enabled bool) Option { + return func(worker *Worker) { + worker.softnetPolicyUpdates = mo.Some(enabled) + } +} + func WithLogger(logger *zap.Logger) Option { return func(worker *Worker) { worker.logger = logger.Sugar() diff --git a/internal/worker/runtime/runtime.go b/internal/worker/runtime/runtime.go index 77acd6bd..4bf7815a 100644 --- a/internal/worker/runtime/runtime.go +++ b/internal/worker/runtime/runtime.go @@ -19,6 +19,7 @@ type Runtime interface { eventStreamer *client.EventStreamer, vmPullTimeHistogram metric.Float64Histogram, dialer dialer.Dialer, + softnetPolicyUpdates bool, logger *zap.SugaredLogger, ) vmmanager.VM ListVMs(ctx context.Context, logger *zap.SugaredLogger) ([]vmmanager.VMInfo, error) diff --git a/internal/worker/runtime/synthetic.go b/internal/worker/runtime/synthetic.go index 94584f61..421c3ae0 100644 --- a/internal/worker/runtime/synthetic.go +++ b/internal/worker/runtime/synthetic.go @@ -37,6 +37,7 @@ func (synthetic *Synthetic) NewVM( eventStreamer *client.EventStreamer, vmPullTimeHistogram metric.Float64Histogram, _ dialer.Dialer, + _ bool, logger *zap.SugaredLogger, ) vmmanager.VM { return syntheticpkg.NewVM(vmResource, eventStreamer, vmPullTimeHistogram, logger) diff --git a/internal/worker/runtime/tart.go b/internal/worker/runtime/tart.go index f1522066..35673ba4 100644 --- a/internal/worker/runtime/tart.go +++ b/internal/worker/runtime/tart.go @@ -31,9 +31,10 @@ func (tart *Tart) NewVM( eventStreamer *client.EventStreamer, vmPullTimeHistogram metric.Float64Histogram, dialer dialer.Dialer, + softnetPolicyUpdates bool, logger *zap.SugaredLogger, ) vmmanager.VM { - return tartpkg.NewVM(vmResource, eventStreamer, vmPullTimeHistogram, dialer, logger) + return tartpkg.NewVM(vmResource, eventStreamer, vmPullTimeHistogram, dialer, softnetPolicyUpdates, logger) } func (tart *Tart) ListVMs(ctx context.Context, logger *zap.SugaredLogger) ([]vmmanager.VMInfo, error) { diff --git a/internal/worker/runtime/vetu.go b/internal/worker/runtime/vetu.go index 84884d79..d8349f16 100644 --- a/internal/worker/runtime/vetu.go +++ b/internal/worker/runtime/vetu.go @@ -31,6 +31,7 @@ func (vetu *Vetu) NewVM( eventStreamer *client.EventStreamer, vmPullTimeHistogram metric.Float64Histogram, dialer dialer.Dialer, + _ bool, logger *zap.SugaredLogger, ) vmmanager.VM { return vetupkg.NewVM(vmResource, eventStreamer, vmPullTimeHistogram, dialer, logger) diff --git a/internal/worker/vmmanager/base/cmd.go b/internal/worker/vmmanager/base/cmd.go index 837eb3d7..a95e499b 100644 --- a/internal/worker/vmmanager/base/cmd.go +++ b/internal/worker/vmmanager/base/cmd.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "os" "os/exec" "strings" @@ -18,16 +19,35 @@ func Cmd( logger *zap.SugaredLogger, commandName string, args ...string, +) (string, string, error) { + return CmdWithExtraFiles(ctx, logger, commandName, nil, args...) +} + +func CmdWithExtraFiles( + ctx context.Context, + logger *zap.SugaredLogger, + commandName string, + extraFiles []*os.File, + args ...string, ) (string, string, error) { cmd := exec.CommandContext(ctx, commandName, args...) + cmd.ExtraFiles = extraFiles + var stdout, stderr bytes.Buffer cmd.Stdout = &stdout cmd.Stderr = &stderr logger.Debugf("running '%s %s'", commandName, strings.Join(args, " ")) - err := cmd.Run() + err := cmd.Start() + for _, file := range extraFiles { + _ = file.Close() + } + if err == nil { + err = cmd.Wait() + } + if err != nil { if errors.Is(err, exec.ErrNotFound) { return "", "", fmt.Errorf("%s command not found in PATH, make sure %s is installed: %w", diff --git a/internal/worker/vmmanager/stop_test.go b/internal/worker/vmmanager/stop_test.go index bfb825d8..3ffb04fa 100644 --- a/internal/worker/vmmanager/stop_test.go +++ b/internal/worker/vmmanager/stop_test.go @@ -26,7 +26,7 @@ func TestStopWaitsForCommandAndRun(t *testing.T) { new func(v1.VM) vmmanager.VM }{ {name: "tart", new: func(resource v1.VM) vmmanager.VM { - return tart.NewVM(resource, nil, nil, nil, zap.NewNop().Sugar()) + return tart.NewVM(resource, nil, nil, nil, false, zap.NewNop().Sugar()) }}, {name: "vetu", new: func(resource v1.VM) vmmanager.VM { return vetu.NewVM(resource, nil, nil, nil, zap.NewNop().Sugar()) diff --git a/internal/worker/vmmanager/synthetic/synthetic.go b/internal/worker/vmmanager/synthetic/synthetic.go index ea82286e..c53b111c 100644 --- a/internal/worker/vmmanager/synthetic/synthetic.go +++ b/internal/worker/vmmanager/synthetic/synthetic.go @@ -92,6 +92,10 @@ func (vm *VM) SetResource(vmResource v1.VM) { vm.resource.ObservedGeneration = vmResource.Generation } +func (vm *VM) UpdateSoftnetPolicy(context.Context, []string, []string) error { + return nil +} + func (vm *VM) OnDiskName() ondiskname.OnDiskName { return vm.onDiskName } diff --git a/internal/worker/vmmanager/tart/cmd.go b/internal/worker/vmmanager/tart/cmd.go index e493f434..c41673e2 100644 --- a/internal/worker/vmmanager/tart/cmd.go +++ b/internal/worker/vmmanager/tart/cmd.go @@ -3,16 +3,43 @@ package tart import ( "context" "encoding/json" + "fmt" + "os" + "strings" "github.com/cirruslabs/orchard/internal/worker/vmmanager" "github.com/cirruslabs/orchard/internal/worker/vmmanager/base" + goversion "github.com/hashicorp/go-version" "go.uber.org/zap" ) const tartCommandName = "tart" func Tart(ctx context.Context, logger *zap.SugaredLogger, args ...string) (string, string, error) { - return base.Cmd(ctx, logger, tartCommandName, args...) + return TartWithExtraFiles(ctx, logger, nil, args...) +} + +func Version(ctx context.Context, logger *zap.SugaredLogger) (*goversion.Version, error) { + stdout, _, err := Tart(ctx, logger, "--version") + if err != nil { + return nil, err + } + + tartVersion, err := goversion.NewSemver(strings.TrimSpace(stdout)) + if err != nil { + return nil, fmt.Errorf("failed to parse Tart version: %w", err) + } + + return tartVersion, nil +} + +func TartWithExtraFiles( + ctx context.Context, + logger *zap.SugaredLogger, + extraFiles []*os.File, + args ...string, +) (string, string, error) { + return base.CmdWithExtraFiles(ctx, logger, tartCommandName, extraFiles, args...) } func List(ctx context.Context, logger *zap.SugaredLogger) ([]vmmanager.VMInfo, error) { diff --git a/internal/worker/vmmanager/tart/softnet_control.go b/internal/worker/vmmanager/tart/softnet_control.go new file mode 100644 index 00000000..ac01d200 --- /dev/null +++ b/internal/worker/vmmanager/tart/softnet_control.go @@ -0,0 +1,158 @@ +//nolint:err113,perfsprint,staticcheck // preserve the original Softnet control implementation +package tart + +import ( + "context" + "fmt" + "io" + "net" + "os" + "time" + + "golang.org/x/exp/jsonrpc2" + "golang.org/x/sys/unix" +) + +const ( + softnetControlFD = 3 + softnetControlTimeout = 5 * time.Second +) + +type softnetPolicyControl struct { + conn *jsonrpc2.Connection +} + +func (control *softnetPolicyControl) close() { + _ = control.conn.Close() +} + +type softnetPolicyTransport struct { + net.Conn +} + +func (transport *softnetPolicyTransport) Dial(context.Context) (io.ReadWriteCloser, error) { + return transport, nil +} + +func (transport *softnetPolicyTransport) Write(data []byte) (int, error) { + n, err := transport.Conn.Write(data) + if err != nil { + return n, err + } + if n != len(data) { + return n, io.ErrShortWrite + } + + _, err = io.WriteString(transport.Conn, "\n") + return n, err +} + +type softnetPolicyParams struct { + Allow []string `json:"allow"` + Block []string `json:"block"` +} + +type softnetPolicyResult struct { + Allow []string `json:"allow"` + Block []string `json:"block"` +} + +func newSoftnetPolicyControl() (*softnetPolicyControl, *os.File, error) { + fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM, 0) + if err != nil { + return nil, nil, err + } + + unix.CloseOnExec(fds[0]) + unix.CloseOnExec(fds[1]) + + workerFile := os.NewFile(uintptr(fds[0]), "orchard-softnet-control") + tartFile := os.NewFile(uintptr(fds[1]), "tart-softnet-control") + + conn, err := net.FileConn(workerFile) + _ = workerFile.Close() + if err != nil { + _ = tartFile.Close() + return nil, nil, err + } + + rpcConn, err := jsonrpc2.Dial( + context.Background(), + &softnetPolicyTransport{Conn: conn}, + jsonrpc2.ConnectionOptions{ + Framer: jsonrpc2.RawFramer(), + }, + ) + if err != nil { + _ = conn.Close() + _ = tartFile.Close() + return nil, nil, err + } + + return &softnetPolicyControl{conn: rpcConn}, tartFile, nil +} + +func (control *softnetPolicyControl) setPolicy( + ctx context.Context, + allow []string, + block []string, +) error { + if allow == nil { + allow = []string{} + } + if block == nil { + block = []string{} + } + + if _, hasDeadline := ctx.Deadline(); !hasDeadline { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, softnetControlTimeout) + defer cancel() + } + + var result *softnetPolicyResult + call := control.conn.Call(ctx, "softnet.policy.set", softnetPolicyParams{ + Allow: allow, + Block: block, + }) + if err := call.Await(ctx, &result); err != nil { + return fmt.Errorf("failed to update Softnet policy: %w", err) + } + + if result == nil { + return fmt.Errorf("invalid Softnet policy response: missing result") + } + + return nil +} + +func (vm *VM) installSoftnetPolicyControl(control *softnetPolicyControl) { + vm.softnetControlMtx.Lock() + defer vm.softnetControlMtx.Unlock() + + if vm.softnetControl != nil { + vm.softnetControl.close() + } + vm.softnetControl = control +} + +func (vm *VM) removeSoftnetPolicyControl(control *softnetPolicyControl) { + vm.softnetControlMtx.Lock() + defer vm.softnetControlMtx.Unlock() + + if vm.softnetControl == control { + control.close() + vm.softnetControl = nil + } +} + +func (vm *VM) UpdateSoftnetPolicy(ctx context.Context, allow []string, block []string) error { + vm.softnetControlMtx.Lock() + defer vm.softnetControlMtx.Unlock() + + if vm.softnetControl == nil { + return fmt.Errorf("Softnet policy control is unavailable") + } + + return vm.softnetControl.setPolicy(ctx, allow, block) +} diff --git a/internal/worker/vmmanager/tart/tart.go b/internal/worker/vmmanager/tart/tart.go index b655c47a..8657aab5 100644 --- a/internal/worker/vmmanager/tart/tart.go +++ b/internal/worker/vmmanager/tart/tart.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "os" "os/exec" "strconv" "strings" @@ -24,9 +25,10 @@ import ( const tartDeleteExitCodeNotFound = 2 type VM struct { - onDiskName ondiskname.OnDiskName - resource v1.VM - logger *zap.SugaredLogger + onDiskName ondiskname.OnDiskName + resource v1.VM + resourceMtx sync.RWMutex + logger *zap.SugaredLogger // Image FQN feature, see https://github.com/cirruslabs/orchard/issues/164 imageFQN atomic.Pointer[string] @@ -41,6 +43,10 @@ type VM struct { dialer dialer.Dialer + softnetPolicyUpdates bool + softnetControl *softnetPolicyControl + softnetControlMtx sync.Mutex + *base.VM } @@ -49,6 +55,7 @@ func NewVM( eventStreamer *client.EventStreamer, vmPullTimeHistogram metric.Float64Histogram, dialer dialer.Dialer, + softnetPolicyUpdates bool, logger *zap.SugaredLogger, ) *VM { vmContext, vmContextCancel := context.WithCancel(context.Background()) @@ -67,7 +74,8 @@ func NewVM( wg: &sync.WaitGroup{}, - dialer: dialer, + dialer: dialer, + softnetPolicyUpdates: softnetPolicyUpdates, VM: base.NewVM(logger), } @@ -123,10 +131,16 @@ func NewVM( } func (vm *VM) Resource() v1.VM { + vm.resourceMtx.RLock() + defer vm.resourceMtx.RUnlock() + return vm.resource } func (vm *VM) SetResource(vmResource v1.VM) { + vm.resourceMtx.Lock() + defer vm.resourceMtx.Unlock() + vm.resource = vmResource vm.resource.ObservedGeneration = vmResource.Generation } @@ -282,62 +296,80 @@ func (vm *VM) cloneAndConfigure(ctx context.Context) error { return nil } +//nolint:contextcheck,perfsprint,staticcheck // preserve the original launch expressions and context ownership func (vm *VM) run(ctx context.Context, eventStreamer *client.EventStreamer) { // Stop owns Stopping until both its command and this goroutine finish. defer vm.ConditionsSet().RemoveAll(v1.ConditionTypeRunning, v1.ConditionTypeSuspending) + resource := vm.Resource() + // Launch the startup script goroutine as close as possible // to the VM startup (below) to avoid "tart ip" timing out - if vm.resource.StartupScript != nil { + if resource.StartupScript != nil { vm.SetStatusMessage("VM started, running startup script...") - go vm.RunScript(vm.ctx, vm.resource.Username, vm.resource.Password, vm.resource.StartupScript, + go vm.RunScript(vm.ctx, resource.Username, resource.Password, resource.StartupScript, eventStreamer, vm.dialer, vm.IP) } else { vm.SetStatusMessage("VM started") } + var extraFiles []*os.File var runArgs = []string{"run"} - if vm.resource.NetSoftnetDeprecated || vm.resource.NetSoftnet { + if resource.VMSpec.SoftnetEnabled() { runArgs = append(runArgs, "--net-softnet") + + if vm.softnetPolicyUpdates { + ourFile, tartFile, err := newSoftnetPolicyControl() + if err != nil { + vm.SetErr(fmt.Errorf("failed to create Softnet policy control channel: %w", err)) + return + } + + vm.installSoftnetPolicyControl(ourFile) + defer vm.removeSoftnetPolicyControl(ourFile) + + extraFiles = append(extraFiles, tartFile) + runArgs = append(runArgs, fmt.Sprintf("--net-softnet-control-fd=%d", softnetControlFD)) + } } - if len(vm.resource.NetSoftnetAllow) != 0 { - runArgs = append(runArgs, "--net-softnet-allow", strings.Join(vm.resource.NetSoftnetAllow, ",")) + if len(resource.NetSoftnetAllow) != 0 { + runArgs = append(runArgs, "--net-softnet-allow", strings.Join(resource.NetSoftnetAllow, ",")) } - if len(vm.resource.NetSoftnetBlock) != 0 { - runArgs = append(runArgs, "--net-softnet-block", strings.Join(vm.resource.NetSoftnetBlock, ",")) + if len(resource.NetSoftnetBlock) != 0 { + runArgs = append(runArgs, "--net-softnet-block", strings.Join(resource.NetSoftnetBlock, ",")) } - if vm.resource.NetBridged != "" { - runArgs = append(runArgs, fmt.Sprintf("--net-bridged=%s", vm.resource.NetBridged)) + if resource.NetBridged != "" { + runArgs = append(runArgs, fmt.Sprintf("--net-bridged=%s", resource.NetBridged)) } - if vm.resource.Headless { + if resource.Headless { runArgs = append(runArgs, "--no-graphics") } - if vm.resource.Nested { + if resource.Nested { runArgs = append(runArgs, "--nested") } - if vm.resource.NoAudio { + if resource.NoAudio { runArgs = append(runArgs, "--no-audio") } - if vm.resource.NoClipboard { + if resource.NoClipboard { runArgs = append(runArgs, "--no-clipboard") } - if vm.resource.Suspendable { + if resource.Suspendable { runArgs = append(runArgs, "--suspendable") } - for _, hostDir := range vm.resource.HostDirs { + for _, hostDir := range resource.HostDirs { runArgs = append(runArgs, fmt.Sprintf("--dir=%s", hostDir.String())) } runArgs = append(runArgs, vm.id()) - _, _, err := Tart(ctx, vm.logger, runArgs...) + _, _, err := TartWithExtraFiles(ctx, vm.logger, extraFiles, runArgs...) if err != nil { select { case <-vm.ctx.Done(): @@ -360,9 +392,11 @@ func (vm *VM) run(ctx context.Context, eventStreamer *client.EventStreamer) { } func (vm *VM) IP(ctx context.Context) (string, error) { + resource := vm.Resource() + // Bridged networking is problematic, so try with // the agent resolver first using a small timeout - if vm.resource.NetBridged != "" { + if resource.NetBridged != "" { stdout, _, err := Tart(ctx, vm.logger, "ip", "--wait", "5", "--resolver", "agent", vm.id()) if err == nil { @@ -372,7 +406,7 @@ func (vm *VM) IP(ctx context.Context) (string, error) { args := []string{"ip", "--wait", "60"} - if vm.resource.NetBridged != "" { + if resource.NetBridged != "" { args = append(args, "--resolver", "arp") } diff --git a/internal/worker/vmmanager/vetu/vetu.go b/internal/worker/vmmanager/vetu/vetu.go index d5f79cff..59542368 100644 --- a/internal/worker/vmmanager/vetu/vetu.go +++ b/internal/worker/vmmanager/vetu/vetu.go @@ -2,6 +2,7 @@ package vetu import ( "context" + "errors" "fmt" "strconv" "strings" @@ -127,6 +128,10 @@ func (vm *VM) SetResource(vmResource v1.VM) { vm.resource.ObservedGeneration = vmResource.Generation } +func (vm *VM) UpdateSoftnetPolicy(context.Context, []string, []string) error { + return errors.ErrUnsupported +} + func (vm *VM) OnDiskName() ondiskname.OnDiskName { return vm.onDiskName } diff --git a/internal/worker/vmmanager/vmmanager.go b/internal/worker/vmmanager/vmmanager.go index 9f2c030c..fdf9d936 100644 --- a/internal/worker/vmmanager/vmmanager.go +++ b/internal/worker/vmmanager/vmmanager.go @@ -12,6 +12,7 @@ import ( type VM interface { Resource() v1.VM SetResource(vmResource v1.VM) + UpdateSoftnetPolicy(ctx context.Context, allow []string, block []string) error OnDiskName() ondiskname.OnDiskName ImageFQN() *string Status() v1.VMStatus diff --git a/internal/worker/worker.go b/internal/worker/worker.go index fda971e4..78cae8b6 100644 --- a/internal/worker/worker.go +++ b/internal/worker/worker.go @@ -18,12 +18,15 @@ import ( "github.com/cirruslabs/orchard/internal/worker/platform" "github.com/cirruslabs/orchard/internal/worker/runtime" "github.com/cirruslabs/orchard/internal/worker/vmmanager" + "github.com/cirruslabs/orchard/internal/worker/vmmanager/tart" "github.com/cirruslabs/orchard/pkg/client" v1 "github.com/cirruslabs/orchard/pkg/resource/v1" "github.com/cirruslabs/orchard/rpc" mapset "github.com/deckarep/golang-set/v2" "github.com/dustin/go-humanize" + "github.com/google/go-cmp/cmp" "github.com/hashicorp/go-multierror" + goversion "github.com/hashicorp/go-version" "github.com/samber/lo" "github.com/samber/mo" "github.com/shirou/gopsutil/v4/cpu" @@ -43,6 +46,8 @@ const ( rpcWatchReconnectMultiplier = 2 rpcWatchHealthyInterval = time.Second onDiskVMSyncTimeout = 30 * time.Second + + tartVersionSoftnetPolicyUpdates = "2.34.0" ) var ( @@ -66,6 +71,8 @@ type Worker struct { runtime runtime.Runtime + softnetPolicyUpdates mo.Option[bool] + vmPullTimeHistogram metric.Float64Histogram dialer dialer.Dialer @@ -147,6 +154,17 @@ func New(client *client.Client, opts ...Option) (*Worker, error) { worker.logger = zap.NewNop().Sugar() } + if worker.softnetPolicyUpdates.IsAbsent() && + worker.runtime.ID() == v1.RuntimeTart && !worker.runtime.Synthetic() { + tartVersion, err := tart.Version(context.Background(), worker.logger) + if err != nil { + worker.logger.Warnf("failed to check whether Tart supports Softnet policy updates: %v", err) + } else { + minimumVersion := goversion.Must(goversion.NewSemver(tartVersionSoftnetPolicyUpdates)) + worker.softnetPolicyUpdates = mo.Some(tartVersion.GreaterThanOrEqual(minimumVersion)) + } + } + return worker, nil } @@ -572,6 +590,28 @@ func (worker *Worker) syncVMs( return err } case ActionMonitorRunning: + currentVMResource := vm.Resource() + + if worker.softnetPolicyUpdates.OrElse(false) && + currentVMResource.SoftnetEnabled() && vmResource.SoftnetEnabled() && + currentVMResource.SoftnetPolicyChanged(vmResource.VMSpec) { + if err := vm.UpdateSoftnetPolicy(ctx, + vmResource.NetSoftnetAllow, vmResource.NetSoftnetBlock); err != nil { + worker.logger.Warnf("failed to update Softnet policy in-place, "+ + "falling back to restart: %v", err) + } else { + currentVMResource.NetSoftnetAllow = vmResource.NetSoftnetAllow + currentVMResource.NetSoftnetBlock = vmResource.NetSoftnetBlock + + // Advance the generation only if no other spec changes remain + if cmp.Equal(currentVMResource.VMSpec, vmResource.VMSpec) { + currentVMResource = *vmResource + } + + vm.SetResource(currentVMResource) + } + } + if err := worker.monitorRunningVM(ctx, vmResource, vm, updateVM); err != nil { return err } @@ -834,7 +874,8 @@ func (worker *Worker) deleteVM(vm vmmanager.VM) error { func (worker *Worker) createVM(odn ondiskname.OnDiskName, vmResource v1.VM) { eventStreamer := worker.client.VMs().StreamEvents(vmResource.Name) - vm := worker.runtime.NewVM(vmResource, eventStreamer, worker.vmPullTimeHistogram, worker.dialer, worker.logger) + vm := worker.runtime.NewVM(vmResource, eventStreamer, worker.vmPullTimeHistogram, + worker.dialer, worker.softnetPolicyUpdates.OrElse(false), worker.logger) worker.vmm.Put(odn, vm) } diff --git a/pkg/resource/v1/v1.go b/pkg/resource/v1/v1.go index ac2bd1fc..017d6b08 100644 --- a/pkg/resource/v1/v1.go +++ b/pkg/resource/v1/v1.go @@ -3,6 +3,7 @@ package v1 import ( "encoding/json" "fmt" + "slices" "time" ) @@ -193,6 +194,16 @@ type VMSpec struct { PowerState PowerState `json:"powerState,omitempty"` } +func (vm VMSpec) SoftnetEnabled() bool { + return vm.NetSoftnetDeprecated || vm.NetSoftnet || + len(vm.NetSoftnetAllow) != 0 || len(vm.NetSoftnetBlock) != 0 +} + +func (vm VMSpec) SoftnetPolicyChanged(other VMSpec) bool { + return !slices.Equal(vm.NetSoftnetAllow, other.NetSoftnetAllow) || + !slices.Equal(vm.NetSoftnetBlock, other.NetSoftnetBlock) +} + type VMSpecReadOnly struct { LocalName string `json:"localName,omitempty"` From 29a78cdc266c70c96118f204fedee56c80665088 Mon Sep 17 00:00:00 2001 From: edi-oai Date: Thu, 6 Aug 2026 16:33:46 +0100 Subject: [PATCH 2/2] Use cmpopts.EquateEmpty() when comparing two VMSpecs --- internal/controller/api_vms.go | 3 +-- internal/worker/worker.go | 3 +-- pkg/resource/v1/cmp_test.go | 10 ++++++++++ pkg/resource/v1/v1.go | 8 ++++++++ 4 files changed, 20 insertions(+), 4 deletions(-) diff --git a/internal/controller/api_vms.go b/internal/controller/api_vms.go index f0dc9a37..32eb7469 100644 --- a/internal/controller/api_vms.go +++ b/internal/controller/api_vms.go @@ -19,7 +19,6 @@ import ( "github.com/cirruslabs/orchard/pkg/resource/v1" "github.com/cirruslabs/orchard/rpc" "github.com/gin-gonic/gin" - "github.com/google/go-cmp/cmp" "github.com/google/uuid" "github.com/samber/lo" ) @@ -222,7 +221,7 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder "transition: only suspendable VMs can be suspended")) } - if cmp.Equal(dbVM.VMSpec, userVM.VMSpec) { + if dbVM.SemanticallyEqual(userVM.VMSpec) { // Nothing was changed return responder.JSON(http.StatusOK, dbVM) } diff --git a/internal/worker/worker.go b/internal/worker/worker.go index 78cae8b6..9b4fcb32 100644 --- a/internal/worker/worker.go +++ b/internal/worker/worker.go @@ -24,7 +24,6 @@ import ( "github.com/cirruslabs/orchard/rpc" mapset "github.com/deckarep/golang-set/v2" "github.com/dustin/go-humanize" - "github.com/google/go-cmp/cmp" "github.com/hashicorp/go-multierror" goversion "github.com/hashicorp/go-version" "github.com/samber/lo" @@ -604,7 +603,7 @@ func (worker *Worker) syncVMs( currentVMResource.NetSoftnetBlock = vmResource.NetSoftnetBlock // Advance the generation only if no other spec changes remain - if cmp.Equal(currentVMResource.VMSpec, vmResource.VMSpec) { + if currentVMResource.SemanticallyEqual(vmResource.VMSpec) { currentVMResource = *vmResource } diff --git a/pkg/resource/v1/cmp_test.go b/pkg/resource/v1/cmp_test.go index e8094df5..bb853170 100644 --- a/pkg/resource/v1/cmp_test.go +++ b/pkg/resource/v1/cmp_test.go @@ -5,6 +5,7 @@ import ( v1 "github.com/cirruslabs/orchard/pkg/resource/v1" "github.com/google/go-cmp/cmp" + "github.com/stretchr/testify/require" ) // TestVM ensures that v1.VM and its embedded structs can be compared @@ -12,3 +13,12 @@ import ( func TestVM(t *testing.T) { cmp.Equal(v1.VM{}, v1.VM{}) } + +func TestVMSpecSemanticallyEqualEquatesEmptySlices(t *testing.T) { + nilSlicesSpec := v1.VMSpec{} + emptySlicesSpec := v1.VMSpec{ + NetSoftnetAllow: []string{}, + NetSoftnetBlock: []string{}, + } + require.True(t, nilSlicesSpec.SemanticallyEqual(emptySlicesSpec)) +} diff --git a/pkg/resource/v1/v1.go b/pkg/resource/v1/v1.go index 017d6b08..b1134e32 100644 --- a/pkg/resource/v1/v1.go +++ b/pkg/resource/v1/v1.go @@ -5,6 +5,9 @@ import ( "fmt" "slices" "time" + + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" ) // Meta is a common set of fields that apply to all resources managed by the Controller. @@ -194,6 +197,11 @@ type VMSpec struct { PowerState PowerState `json:"powerState,omitempty"` } +func (vm VMSpec) SemanticallyEqual(other VMSpec) bool { + // Treat omitted and explicitly empty collections as the same VM specification + return cmp.Equal(vm, other, cmpopts.EquateEmpty()) +} + func (vm VMSpec) SoftnetEnabled() bool { return vm.NetSoftnetDeprecated || vm.NetSoftnet || len(vm.NetSoftnetAllow) != 0 || len(vm.NetSoftnetBlock) != 0