engine: double-write cmds to apiserver (#4297)

internal/controllers/client_builder.go

@@ -8,19 +8,24 @@ import (
 )
 
 type ClientBuilder struct {
-	cluster.ClientBuilder
+	delegate cluster.ClientBuilder
 	deferred *DeferredClient
 }
 
 func NewClientBuilder(deferred *DeferredClient) cluster.ClientBuilder {
-	return &ClientBuilder{
-		ClientBuilder: cluster.NewClientBuilder(),
-		deferred:      deferred,
+	return ClientBuilder{
+		delegate: cluster.NewClientBuilder(),
+		deferred: deferred,
 	}
 }
 
-func (b *ClientBuilder) Build(cache cache.Cache, config *rest.Config, options client.Options) (client.Client, error) {
-	c, err := b.ClientBuilder.Build(cache, config, options)
+func (b ClientBuilder) WithUncached(objs ...client.Object) cluster.ClientBuilder {
+	b.delegate = b.delegate.WithUncached(objs...)
+	return b
+}
+
+func (b ClientBuilder) Build(cache cache.Cache, config *rest.Config, options client.Options) (client.Client, error) {
+	c, err := b.delegate.Build(cache, config, options)
 	if err != nil {
 		return nil, err
 	}

internal/engine/local/controller.go

@@ -12,6 +12,7 @@ import (
 	"github.com/tilt-dev/probe/pkg/probe"
 	"github.com/tilt-dev/probe/pkg/prober"
 	"k8s.io/apimachinery/pkg/api/equality"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
@@ -113,7 +114,7 @@ func (c *Controller) reconcile(ctx context.Context, st store.RStore, name types.
 	proc.spec = cmd.Spec
 	ctx, proc.cancelFunc = context.WithCancel(ctx)
 
-	c.resetStatus(st, name, cmd)
+	c.resetStatus(ctx, st, name, cmd)
 
 	statusCh := make(chan statusAndMetadata)
 
@@ -131,7 +132,7 @@ func (c *Controller) reconcile(ctx context.Context, st store.RStore, name types.
 			resultLoggerFunc)
 		if err != nil {
 			logger.Get(ctx).Errorf("Invalid readiness probe: %v", err)
-			c.updateStatus(st, name, func(status *CmdStatus) {
+			c.updateStatus(ctx, st, name, func(status *CmdStatus) {
 				*status = CmdStatus{
 					Terminated: &CmdStateTerminated{
 						ExitCode: 1,
@@ -171,7 +172,7 @@ func (c *Controller) processReadinessProbeStatusChange(ctx context.Context, st s
 		ready := status == prober.Success || status == prober.Warning
 		if existingReady != ready {
 			existingReady = ready
-			c.updateStatus(st, name, func(status *CmdStatus) { status.Ready = ready })
+			c.updateStatus(ctx, st, name, func(status *CmdStatus) { status.Ready = ready })
 		}
 	}
 }
@@ -212,7 +213,7 @@ func processReadinessProbeResultLogger(ctx context.Context, stillHasSameProcNum
 	}
 }
 
-func (c *Controller) resetStatus(st store.RStore, name types.NamespacedName, cmd *Cmd) {
+func (c *Controller) resetStatus(ctx context.Context, st store.RStore, name types.NamespacedName, cmd *Cmd) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
@@ -222,6 +223,12 @@ func (c *Controller) resetStatus(st store.RStore, name types.NamespacedName, cmd
 	}
 	c.updateCmds[name] = updated
 
+	err := c.client.Status().Update(ctx, updated)
+	if err != nil && !apierrors.IsNotFound(err) {
+		st.Dispatch(store.NewErrorAction(fmt.Errorf("syncing to apiserver: %v", err)))
+		return
+	}
+
 	st.Dispatch(NewCmdUpdateStatusAction(updated))
 }
 
@@ -230,7 +237,7 @@ func (c *Controller) resetStatus(st store.RStore, name types.NamespacedName, cmd
 // In a real K8s controller, this would be a queue to make sure we don't miss updates.
 //
 // update() -> a pure function that applies a delta to the status object.
-func (c *Controller) updateStatus(st store.RStore, name types.NamespacedName, update func(status *CmdStatus)) {
+func (c *Controller) updateStatus(ctx context.Context, st store.RStore, name types.NamespacedName, update func(status *CmdStatus)) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
@@ -240,6 +247,13 @@ func (c *Controller) updateStatus(st store.RStore, name types.NamespacedName, up
 	}
 
 	update(&cmd.Status)
+
+	err := c.client.Status().Update(ctx, cmd)
+	if err != nil && !apierrors.IsNotFound(err) {
+		st.Dispatch(store.NewErrorAction(fmt.Errorf("syncing to apiserver: %v", err)))
+		return
+	}
+
 	st.Dispatch(NewCmdUpdateStatusAction(cmd))
 }
 
@@ -260,7 +274,7 @@ func (c *Controller) processStatuses(
 		}
 
 		if sm.status == Error || sm.status == Done {
-			c.updateStatus(st, name, func(status *CmdStatus) {
+			c.updateStatus(ctx, st, name, func(status *CmdStatus) {
 				status.Waiting = nil
 				status.Running = nil
 				status.Terminated = &CmdStateTerminated{
@@ -278,7 +292,7 @@ func (c *Controller) processStatuses(
 				})
 			}
 
-			c.updateStatus(st, name, func(status *CmdStatus) {
+			c.updateStatus(ctx, st, name, func(status *CmdStatus) {
 				status.Waiting = nil
 				status.Running = &CmdStateRunning{
 					PID:       int32(sm.pid),

internal/engine/local/controller_test.go

@@ -2,6 +2,7 @@ package local
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"os"
 	"strings"
@@ -10,6 +11,9 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
+	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/tilt-dev/tilt/internal/controllers/fake"
 	"github.com/tilt-dev/tilt/internal/store"
@@ -27,7 +31,7 @@ func TestNoop(t *testing.T) {
 	defer f.teardown()
 
 	f.step()
-	assert.Equal(t, 0, f.st.CmdCount())
+	f.assertCmdCount(0)
 }
 
 func TestUpdate(t *testing.T) {
@@ -44,9 +48,7 @@ func TestUpdate(t *testing.T) {
 	t2 := time.Unix(2, 0)
 	f.resource("foo", "false", ".", t2)
 	f.step()
-	f.assertCmdMatches("foo-serve-1", func(cmd *Cmd) bool {
-		return cmd.DeletionTimestamp != nil
-	})
+	f.assertCmdDeleted("foo-serve-1")
 
 	f.step()
 	f.assertCmdMatches("foo-serve-2", func(cmd *Cmd) bool {
@@ -56,7 +58,7 @@ func TestUpdate(t *testing.T) {
 	f.fe.RequireNoKnownProcess(t, "true")
 	f.assertLogMessage("foo", "Starting cmd false")
 	f.assertLogMessage("foo", "cmd true canceled")
-	assert.Equal(t, 2, f.st.CmdCount())
+	f.assertCmdCount(1)
 }
 
 func TestServe(t *testing.T) {
@@ -201,7 +203,13 @@ func (s *testStore) Cmd(name string) *Cmd {
 func (s *testStore) CmdCount() int {
 	st := s.RLockState()
 	defer s.RUnlockState()
-	return len(st.Cmds)
+	count := 0
+	for _, cmd := range st.Cmds {
+		if cmd.DeletionTimestamp != nil {
+			count++
+		}
+	}
+	return count
 }
 
 func (s *testStore) Dispatch(action store.Action) {
@@ -211,6 +219,9 @@ func (s *testStore) Dispatch(action store.Action) {
 	defer s.UnlockMutableState()
 
 	switch action := action.(type) {
+	case store.ErrorAction:
+		panic(fmt.Sprintf("no error action allowed: %s", action.Error))
+
 	case store.LogAction:
 		_, _ = s.out.Write(action.Message())
 
@@ -234,6 +245,7 @@ type fixture struct {
 	fe     *FakeExecer
 	fpm    *FakeProberManager
 	sc     *ServerController
+	client ctrlclient.Client
 	c      *Controller
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -250,7 +262,7 @@ func newFixture(t *testing.T) *fixture {
 	fe := NewFakeExecer()
 	fpm := NewFakeProberManager()
 	fc := fake.NewTiltClient()
-	sc := NewServerController()
+	sc := NewServerController(fc)
 	c := NewController(fe, fpm, fc)
 
 	return &fixture{
@@ -263,6 +275,7 @@ func newFixture(t *testing.T) *fixture {
 		c:      c,
 		ctx:    ctx,
 		cancel: cancel,
+		client: fc,
 	}
 }
 
@@ -335,4 +348,30 @@ func (f *fixture) assertCmdMatches(name string, matcher func(cmd *Cmd) bool) {
 		}
 		return matcher(cmd)
 	}, timeout, interval)
+
+	var cmd Cmd
+	err := f.client.Get(f.ctx, types.NamespacedName{Name: name}, &cmd)
+	require.NoError(f.t, err)
+	assert.True(f.t, matcher(&cmd))
+}
+
+func (f *fixture) assertCmdDeleted(name string) {
+	assert.Eventually(f.t, func() bool {
+		cmd := f.st.Cmd(name)
+		return cmd == nil || cmd.DeletionTimestamp != nil
+	}, timeout, interval)
+
+	var cmd Cmd
+	err := f.client.Get(f.ctx, types.NamespacedName{Name: name}, &cmd)
+	assert.Error(f.t, err)
+	assert.True(f.t, apierrors.IsNotFound(err))
+}
+
+func (f *fixture) assertCmdCount(count int) {
+	assert.Equal(f.t, count, f.st.CmdCount())
+
+	var list CmdList
+	err := f.client.List(f.ctx, &list)
+	require.NoError(f.t, err)
+	assert.Equal(f.t, count, len(list.Items))
 }

internal/engine/local/servercontroller.go

@@ -6,12 +6,17 @@ import (
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/equality"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/tilt-dev/tilt/internal/store"
 	"github.com/tilt-dev/tilt/pkg/apis/core/v1alpha1"
 )
 
+const AnnotationOwnerName = "tilt.dev/owner-name"
+const AnnotationOwnerKind = "tilt.dev/owner-kind"
+
 // A controller that reads the Tilt data model and creates new Cmd objects.
 //
 // Reads the Cmd Status.
@@ -27,17 +32,19 @@ type ServerController struct {
 	createdCmds        map[string]*Cmd
 	createdTriggerTime map[string]time.Time
 	deletingCmds       map[string]bool
+	client             ctrlclient.Client
 
 	cmdCount int
 }
 
 var _ store.Subscriber = &ServerController{}
 
-func NewServerController() *ServerController {
+func NewServerController(client ctrlclient.Client) *ServerController {
 	return &ServerController{
 		createdCmds:        make(map[string]*Cmd),
 		createdTriggerTime: make(map[string]time.Time),
 		deletingCmds:       make(map[string]bool),
+		client:             client,
 	}
 }
 
@@ -61,13 +68,13 @@ func (c *ServerController) determineServers(ctx context.Context, st store.RStore
 	// Simulates controller-runtime's notion of Owns().
 	ownedCmds := make(map[string]*Cmd)
 	for _, cmd := range state.Cmds {
-		for _, ref := range cmd.ObjectMeta.OwnerReferences {
-			if ref.Kind != "CmdServer" {
-				continue
-			}
-
-			ownedCmds[ref.Name] = cmd
+		ownerName := cmd.Annotations[AnnotationOwnerName]
+		ownerKind := cmd.Annotations[AnnotationOwnerKind]
+		if ownerKind != "CmdServer" {
+			continue
 		}
+
+		ownedCmds[ownerName] = cmd
 	}
 
 	var r []CmdServer
@@ -138,6 +145,13 @@ func (c *ServerController) reconcile(ctx context.Context, server CmdServer, st s
 	if isCreated && server.Status.CmdStatus.Terminated == nil {
 		if !c.deletingCmds[name] {
 			c.deletingCmds[name] = true
+
+			err := c.client.Delete(ctx, created)
+			if err != nil && !apierrors.IsNotFound(err) {
+				st.Dispatch(store.NewErrorAction(fmt.Errorf("syncing to apiserver: %v", err)))
+				return
+			}
+
 			st.Dispatch(CmdDeleteAction{Name: server.Status.CmdName})
 		}
 		return
@@ -154,13 +168,12 @@ func (c *ServerController) reconcile(ctx context.Context, server CmdServer, st s
 	cmd := &Cmd{
 		ObjectMeta: ObjectMeta{
 			Name: cmdName,
-			OwnerReferences: []metav1.OwnerReference{
-				{
-					Name: name,
-					Kind: "CmdServer",
-				},
-			},
 			Annotations: map[string]string{
+				// TODO(nick): This should be an owner reference once CmdServer is a
+				// full-fledged type.
+				AnnotationOwnerName: name,
+				AnnotationOwnerKind: "CmdServer",
+
 				v1alpha1.AnnotationSpanID: string(spanID),
 			},
 			Labels: map[string]string{
@@ -170,6 +183,13 @@ func (c *ServerController) reconcile(ctx context.Context, server CmdServer, st s
 		Spec: cmdSpec,
 	}
 	c.createdCmds[name] = cmd
+
+	err := c.client.Create(ctx, cmd)
+	if err != nil && !apierrors.IsNotFound(err) {
+		st.Dispatch(store.NewErrorAction(fmt.Errorf("syncing to apiserver: %v", err)))
+		return
+	}
+
 	st.Dispatch(CmdCreateAction{Cmd: cmd})
 }
 

internal/engine/local/types.go

@@ -7,6 +7,7 @@ import (
 )
 
 type Cmd = v1alpha1.Cmd
+type CmdList = v1alpha1.CmdList
 type CmdStatus = v1alpha1.CmdStatus
 type CmdSpec = v1alpha1.CmdSpec
 type CmdStateWaiting = v1alpha1.CmdStateWaiting

internal/engine/upper_test.go

@@ -3774,7 +3774,7 @@ func newTestFixture(t *testing.T) *testFixture {
 	fc := fake.NewTiltClient()
 	fcb := fake.NewClientBuilder(fc)
 	lc := local.NewController(fe, fpm, fc)
-	lsc := local.NewServerController()
+	lsc := local.NewServerController(fc)
 	ts := hud.NewTerminalStream(hud.NewIncrementalPrinter(log), st)
 	tp := prompt.NewTerminalPrompt(ta, prompt.TTYOpen, prompt.BrowserOpen,
 		log, "localhost", model.WebURL{})