feat: add control loop framework for microvm reconciliation

Makefile

@@ -77,13 +77,13 @@ lint: $(GOLANGCI_LINT) $(BUF) ## Lint
 
 .PHONY: test
 test: ## Run unit tests
-	go test -v ./...
+	go test -v -race ./...
 
 .PHONY: test-int
 test-int: $(OUT_DIR) ## Run tests (including intengration tests)
 	CTR_ROOT_DIR=$(OUT_DIR)/containerd
 	mkdir -p $(CTR_ROOT_DIR) 
-	sudo go test -v -count=1 ./...
+	sudo go test -v -race -count=1 ./...
 	sudo rm -rf $(CTR_ROOT_DIR) 
 
 .PHONY: test-e2e

cmd/dev-helper/main.go

@@ -5,7 +5,9 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"time"
 
+	"github.com/containerd/containerd/namespaces"
 	"github.com/sirupsen/logrus"
 
 	_ "github.com/containerd/containerd/api/events"
@@ -16,6 +18,8 @@ import (
 	"github.com/weaveworks/reignite/core/models"
 	"github.com/weaveworks/reignite/core/ports"
 	"github.com/weaveworks/reignite/infrastructure/containerd"
+	"github.com/weaveworks/reignite/infrastructure/controllers"
+	"github.com/weaveworks/reignite/pkg/defaults"
 	rlog "github.com/weaveworks/reignite/pkg/log"
 )
 
@@ -34,7 +38,7 @@ func main() {
 	ctx, cancel := context.WithCancel(context.Background())
 
 	rlog.Configure(&rlog.Config{
-		Verbosity: 0,
+		Verbosity: 10,
 		Format:    "text",
 		Output:    "stderr",
 	})
@@ -46,13 +50,18 @@ func main() {
 	logger.Info("starting containerd event listener")
 	go eventListener(ctx, socketPath, logger)
 
-	logger.Infof("Press [enter] to write vmspec to using containerd repo")
-	fmt.Scanln()
-	repoTest(ctx, socketPath, logger)
+	logger.Infof("Press [enter] to run controller test")
+	//fmt.Scanln()
+	go controllerTest(ctx, socketPath, logger)
+	go publishPeriodicEvents(ctx, socketPath, logger)
 
-	logger.Infof("Press [enter] to get image %s", imageName)
-	fmt.Scanln()
-	imageServiceTest(ctx, socketPath, logger)
+	//logger.Infof("Press [enter] to write vmspec to using containerd repo")
+	//fmt.Scanln()
+	//repoTest(ctx, socketPath, logger)
+
+	//logger.Infof("Press [enter] to get image %s", imageName)
+	//fmt.Scanln()
+	//imageServiceTest(ctx, socketPath, logger)
 
 	logger.Info("Press [enter] to exit")
 	fmt.Scanln()
@@ -104,7 +113,7 @@ func repoTest(ctx context.Context, socketPath string, logger *logrus.Entry) {
 	repo := containerd.NewMicroVMRepoWithClient(client)
 
 	vmSpec := getTestSpec()
-	logger.Infof("saving microvm spec %s/%s", vmSpec.Namespace, vmSpec.ID)
+	logger.Infof("saving microvm spec %s", vmSpec.ID)
 
 	_, err = repo.Save(ctx, vmSpec)
 	if err != nil {
@@ -141,6 +150,57 @@ func imageServiceTest(ctx context.Context, socketPath string, logger *logrus.Ent
 	logger.Infof("mounted image %s to %s (type %s)", imageName, mountPoint[0].Source, mountPoint[0].Type)
 }
 
+func publishPeriodicEvents(ctx context.Context, socketPath string, logger *logrus.Entry) {
+	client, err := ctr.New(socketPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	namespaceCtx := namespaces.WithNamespace(ctx, defaults.ContainerdNamespace)
+
+	tickCh := time.Tick(10 * time.Second)
+
+	for {
+		select {
+		case <-ctx.Done():
+			logger.Info("Cancelled, exiting publisher")
+			return
+		case <-tickCh:
+			logger.Info("publishing event")
+			createEvt := &events.MicroVMSpecCreated{
+				ID:        "vm1",
+				Namespace: "testns",
+			}
+
+			err = client.EventService().Publish(namespaceCtx, defaults.TopicMicroVMEvents, createEvt)
+			if err != nil {
+				logger.Error(err)
+			}
+
+		}
+	}
+}
+
+func controllerTest(ctx context.Context, socketPath string, logger *logrus.Entry) {
+	em, err := containerd.NewEventService(&containerd.Config{
+		SocketPath: socketPath,
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	app := &testApp{
+		logger: logger,
+	}
+
+	controller := controllers.New(em, app)
+
+	logger.Info("running controller")
+	controller.Run(ctx, 1)
+	logger.Info("controller not running")
+
+}
+
 func eventListener(ctx context.Context, socketPath string, logger *logrus.Entry) {
 	cfg := &containerd.Config{
 		SocketPath: socketPath,
@@ -167,9 +227,9 @@ func eventListener(ctx context.Context, socketPath string, logger *logrus.Entry)
 }
 
 func getTestSpec() *models.MicroVM {
+	vmid, _ := models.NewVMID(vmName, vmNamespace)
 	return &models.MicroVM{
-		ID:        vmName,
-		Namespace: vmNamespace,
+		ID: *vmid,
 		Spec: models.MicroVMSpec{
 			MemoryInMb: 2048,
 			VCPU:       4,
@@ -207,3 +267,13 @@ func getTestSpec() *models.MicroVM {
 		},
 	}
 }
+
+type testApp struct {
+	logger *logrus.Entry
+}
+
+func (t *testApp) ReconcileMicroVM(ctx context.Context, id, namespace string) error {
+	t.logger.Infof("received request to reconcile %s/%s", id, namespace)
+
+	return nil
+}

core/application/app.go

@@ -7,7 +7,7 @@ import "github.com/weaveworks/reignite/core/ports"
 type App interface {
 	ports.MicroVMCommandUseCases
 	ports.MicroVMQueryUseCases
-	// ports.ReconcileMicroVMsUseCase
+	ports.ReconcileMicroVMsUseCase
 }
 
 func New(repo ports.MicroVMRepository, eventSvc ports.EventService, idSvc ports.IDService, mvmProvider ports.MicroVMProvider) App {

core/application/app_test.go

@@ -37,17 +37,17 @@ func TestApp_CreateMicroVM(t *testing.T) {
 				rm.Get(
 					gomock.AssignableToTypeOf(context.Background()),
 					gomock.Eq("id1234"),
-					gomock.Eq(defaults.ContainerdNamespace),
+					gomock.Eq(defaults.MicroVMNamespace),
 				).Return(
 					nil,
 					nil,
 				)
 
 				rm.Save(
 					gomock.AssignableToTypeOf(context.Background()),
-					gomock.Eq(createTestSpec("id1234", defaults.ContainerdNamespace)),
+					gomock.Eq(createTestSpec("id1234", defaults.MicroVMNamespace)),
 				).Return(
-					createTestSpec("id1234", defaults.ContainerdNamespace),
+					createTestSpec("id1234", defaults.MicroVMNamespace),
 					nil,
 				)
 
@@ -56,7 +56,7 @@ func TestApp_CreateMicroVM(t *testing.T) {
 					gomock.Eq(defaults.TopicMicroVMEvents),
 					gomock.Eq(&events.MicroVMSpecCreated{
 						ID:        "id1234",
-						Namespace: defaults.ContainerdNamespace,
+						Namespace: defaults.MicroVMNamespace,
 					}),
 				)
 			},
@@ -155,14 +155,6 @@ func TestApp_UpdateMicroVM(t *testing.T) {
 			specToUpdate: createTestSpec("", ""),
 			expectError:  true,
 			expect: func(rm *mock.MockMicroVMRepositoryMockRecorder, em *mock.MockEventServiceMockRecorder, im *mock.MockIDServiceMockRecorder, pm *mock.MockMicroVMProviderMockRecorder) {
-				rm.Get(
-					gomock.AssignableToTypeOf(context.Background()),
-					gomock.Eq(""),
-					gomock.Eq(""),
-				).Return(
-					nil,
-					nil,
-				)
 			},
 		},
 		{
@@ -317,10 +309,17 @@ func TestApp_DeleteMicroVM(t *testing.T) {
 	}
 }
 
-func createTestSpec(id, namespace string) *models.MicroVM {
+func createTestSpec(name, ns string) *models.MicroVM {
+	var vmid *models.VMID
+
+	if name == "" && ns == "" {
+		vmid = &models.VMID{}
+	} else {
+		vmid, _ = models.NewVMID(name, ns)
+	}
+
 	return &models.MicroVM{
-		ID:        id,
-		Namespace: namespace,
+		ID: *vmid,
 		Spec: models.MicroVMSpec{
 			VCPU:       2,
 			MemoryInMb: 2048,

core/application/commands.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/weaveworks/reignite/api/events"
+	coreerrs "github.com/weaveworks/reignite/core/errors"
 	"github.com/weaveworks/reignite/core/models"
 	"github.com/weaveworks/reignite/pkg/defaults"
 	"github.com/weaveworks/reignite/pkg/log"
@@ -15,28 +16,29 @@ func (a *app) CreateMicroVM(ctx context.Context, mvm *models.MicroVM) (*models.M
 	logger.Trace("creating microvm")
 
 	if mvm == nil {
-		return nil, errSpecRequired
+		return nil, coreerrs.ErrSpecRequired
 	}
 
-	if mvm.ID == "" {
-		newID, err := a.idSvc.GenerateRandom()
+	if mvm.ID.IsEmpty() {
+		name, err := a.idSvc.GenerateRandom()
 		if err != nil {
-			return nil, fmt.Errorf("generating random id for microvm: %w", err)
+			return nil, fmt.Errorf("generating random name for microvm: %w", err)
 		}
-		mvm.ID = newID
-	}
-	if mvm.Namespace == "" {
-		mvm.Namespace = defaults.ContainerdNamespace // TODO: not sure this is correct
+		vmid, err := models.NewVMID(name, defaults.MicroVMNamespace)
+		if err != nil {
+			return nil, fmt.Errorf("creating vmid: %w", err)
+		}
+		mvm.ID = *vmid
 	}
 
-	foundMvm, err := a.repo.Get(ctx, mvm.ID, mvm.Namespace)
+	foundMvm, err := a.repo.Get(ctx, mvm.ID.Name(), mvm.ID.Namespace())
 	if err != nil {
 		return nil, fmt.Errorf("checking to see if spec exists: %w", err)
 	}
 	if foundMvm != nil {
 		return nil, errSpecAlreadyExists{
-			name:      mvm.ID,
-			namespace: mvm.Namespace,
+			name:      mvm.ID.Name(),
+			namespace: mvm.ID.Namespace(),
 		}
 	}
 
@@ -48,8 +50,8 @@ func (a *app) CreateMicroVM(ctx context.Context, mvm *models.MicroVM) (*models.M
 	}
 
 	if err := a.eventSvc.Publish(ctx, defaults.TopicMicroVMEvents, &events.MicroVMSpecCreated{
-		ID:        mvm.ID,
-		Namespace: mvm.Namespace,
+		ID:        mvm.ID.Name(),
+		Namespace: mvm.ID.Namespace(),
 	}); err != nil {
 		return nil, fmt.Errorf("publishing microvm created event: %w", err)
 	}
@@ -62,31 +64,34 @@ func (a *app) UpdateMicroVM(ctx context.Context, mvm *models.MicroVM) (*models.M
 	logger.Trace("updating microvm")
 
 	if mvm == nil {
-		return nil, errSpecRequired
+		return nil, coreerrs.ErrSpecRequired
+	}
+	if mvm.ID.IsEmpty() {
+		return nil, coreerrs.ErrVMIDRequired
 	}
 
-	foundMvm, err := a.repo.Get(ctx, mvm.ID, mvm.Namespace)
+	foundMvm, err := a.repo.Get(ctx, mvm.ID.Name(), mvm.ID.Namespace())
 	if err != nil {
 		return nil, fmt.Errorf("checking to see if spec exists: %w", err)
 	}
 	if foundMvm == nil {
 		return nil, errSpecNotFound{
-			name:      mvm.ID,
-			namespace: mvm.Namespace,
+			name:      mvm.ID.Name(),
+			namespace: mvm.ID.Namespace(),
 		}
 	}
 
 	// TODO: validate incoming spec
-	// TODO: check if update is valide (i.e. compare existing to requested update)
+	// TODO: check if update is valid (i.e. compare existing to requested update)
 
 	updatedMVM, err := a.repo.Save(ctx, mvm)
 	if err != nil {
 		return nil, fmt.Errorf("updating microvm spec: %w", err)
 	}
 
 	if err := a.eventSvc.Publish(ctx, defaults.TopicMicroVMEvents, &events.MicroVMSpecUpdated{
-		ID:        mvm.ID,
-		Namespace: mvm.Namespace,
+		ID:        mvm.ID.Name(),
+		Namespace: mvm.ID.Namespace(),
 	}); err != nil {
 		return nil, fmt.Errorf("publishing microvm updated event: %w", err)
 	}

core/application/errors.go

@@ -6,8 +6,8 @@ import (
 )
 
 var (
-	errSpecRequired = errors.New("microvm spec is required")
-	errIDRequired   = errors.New("microvm id is required")
+	errIDRequired    = errors.New("microvm id is required")
+	errNotImplemeted = errors.New("not implemented")
 )
 
 type errSpecAlreadyExists struct {

core/application/query.go

@@ -7,9 +7,9 @@ import (
 )
 
 func (a *app) GetMicroVM(ctx context.Context, id, namespace string) (*models.MicroVM, error) {
-	return nil, nil
+	return nil, errNotImplemeted
 }
 
 func (a *app) GetAllMicroVM(ctx context.Context, namespace string) ([]*models.MicroVM, error) {
-	return nil, nil
+	return nil, errNotImplemeted
 }

core/application/reconcile.go

@@ -0,0 +1,7 @@
+package application
+
+import "context"
+
+func (a *app) ReconcileMicroVM(ctx context.Context, id, namespace string) error {
+	return errNotImplemeted
+}