feat: implement control loop framework

The adds the control loop that is used to kick off microvm
reconciliation. The actual reconciliation is empty and will be
implemented in later changes.

This change also introduces a value object `VMID` that i sused to
represent the identity of a microvm.

Signed-off-by: Richard Case <[email protected]>

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) ReconcileMicroVMs(ctx context.Context, id, namespace string) error {
+	t.logger.Infof("received request to reconcole %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,10 @@ func TestApp_DeleteMicroVM(t *testing.T) {
 	}
 }
 
-func createTestSpec(id, namespace string) *models.MicroVM {
+func createTestSpec(name, ns string) *models.MicroVM {
+	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 == nil {
+		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 == nil {
+		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) ReconcileMicroVMs(ctx context.Context, id, namespace string) error {
+	return errNotImplemeted
+}