public_git/orchard
1
0
Fork 0
mirror of https://github.com/cirruslabs/orchard.git synced 2026-10-09 08:16:07 +02:00
Code Issues Packages Projects Releases Wiki Activity

Support dynamically updating Softnet policy (#477)

Browse Source 
* Support dynamically updating Softnet policy

* Use cmpopts.EquateEmpty() when comparing two VMSpecs
This commit is contained in:
edi-oai
2026-09-02 14:19:29 +01:00
committed by GitHub
parent f0bdf3a151
commit 62acf3c265
21 changed files with 544 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.golangci.yml
+1
View File
@@ -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
go.mod
+4 -1
View File
@@ -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
)
go.sum
+8 -2
View File
@@ -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=
internal/controller/api_vms.go
+24 -3
View File
@@ -1,6 +1,7 @@
package controller
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
@@ -16,8 +17,8 @@ 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"
"github.com/samber/lo"
)
@@ -163,8 +164,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)
@@ -216,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)
}
@@ -231,8 +236,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 {
internal/tests/spec_update_test.go
+151 -2
View File
@@ -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")
internal/worker/option.go
+7
View File
@@ -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()
internal/worker/runtime/runtime.go
+1
View File
@@ -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)
internal/worker/runtime/synthetic.go
+1
View File
@@ -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)
internal/worker/runtime/tart.go
+2 -1
View File
@@ -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) {
internal/worker/runtime/vetu.go
+1
View File
@@ -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)
internal/worker/vmmanager/base/cmd.go
+21 -1
View File
@@ -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",
internal/worker/vmmanager/stop_test.go
+1 -1
View File
@@ -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())
internal/worker/vmmanager/synthetic/synthetic.go
+4
View File
@@ -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
}
internal/worker/vmmanager/tart/cmd.go
+28 -1
View File
@@ -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) {
internal/worker/vmmanager/tart/softnet_control.go
+158
View File
@@ -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)
}
internal/worker/vmmanager/tart/tart.go
+56 -22
View File
@@ -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")
}
internal/worker/vmmanager/vetu/vetu.go
+5
View File
@@ -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
}
internal/worker/vmmanager/vmmanager.go
+1
View File
@@ -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
internal/worker/worker.go
+41 -1
View File
@@ -18,12 +18,14 @@ 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/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 +45,8 @@ const (
rpcWatchReconnectMultiplier = 2
rpcWatchHealthyInterval = time.Second
onDiskVMSyncTimeout = 30 * time.Second
tartVersionSoftnetPolicyUpdates = "2.34.0"
)
var (
@@ -66,6 +70,8 @@ type Worker struct {
runtime runtime.Runtime
softnetPolicyUpdates mo.Option[bool]
vmPullTimeHistogram metric.Float64Histogram
dialer dialer.Dialer
@@ -147,6 +153,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 +589,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 currentVMResource.SemanticallyEqual(vmResource.VMSpec) {
currentVMResource = *vmResource
}
vm.SetResource(currentVMResource)
}
}
if err := worker.monitorRunningVM(ctx, vmResource, vm, updateVM); err != nil {
return err
}
@@ -834,7 +873,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)
}
pkg/resource/v1/cmp_test.go
+10
View File
@@ -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))
}
pkg/resource/v1/v1.go
+19
View File
@@ -3,7 +3,11 @@ package v1
import (
"encoding/json"
"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.
@@ -193,6 +197,21 @@ 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
}
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"`