Events Entity (#28)

* Generic Events

We can try to use these generic events for script execution and storing of the output logs in events with `log` kind.

* Lint issues

* Cleanup events upon VM deletion

* Basic integration test

* Run an actual VM in tests

* Apply suggestions from code review

Co-authored-by: Nikolay Edigaryev <edigaryev@gmail.com>

* Use POST

* Make newEventKey private

* Append events in batches

* Lint issues

* Private `scopePrefix`

---------

Co-authored-by: Nikolay Edigaryev <edigaryev@gmail.com>

api/openapi.yaml

@@ -193,6 +193,33 @@ paths:
           description: VM resource was successfully deleted
         '404':
           description: VM resource with the given name doesn't exist
+  /vms/{name}/events:
+    get:
+      summary: "Retrieve events for a given VM"
+      tags:
+        - vms
+      responses:
+        '200':
+          description: OK
+          content:
+            application/json:
+              schema:
+                $ref: '#components/schemas/Events'
+        '404':
+          description: VM resource with the given name doesn't exist
+    post:
+      summary: "Add new events to a VM"
+      tags:
+        - vms
+      responses:
+        '200':
+          description: Events were successfully added
+          content:
+            application/json:
+              schema:
+                $ref: '#components/schemas/Events'
+        '404':
+          description: VM resource with the given name doesn't exist
 components:
   schemas:
     Worker:
@@ -209,6 +236,24 @@ components:
         name:
           type: string
           description: VM name
+    Events:
+      title: Events
+      type: object
+      items:
+        $ref: '#components/schemas/Event'
+    Event:
+      title: Generic Resource Event
+      type: object
+      properties:
+        kind:
+          type: string
+          description: Kind of the event
+        payload:
+          type: string
+          description: Payload of the event
+        timestamp:
+          type: integer
+          description: Unix timestamp of the event
     ServiceAccount:
       title: Service Account
       type: object

go.mod

@@ -36,6 +36,8 @@ require (
 	github.com/golang/snappy v0.0.3 // indirect
 	github.com/google/flatbuffers v1.12.1 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
+	github.com/hashicorp/errwrap v1.0.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/inconshreveable/mousetrap v1.0.1 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.12.3 // indirect

go.sum

@@ -81,6 +81,10 @@ github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gosuri/uitable v0.0.4 h1:IG2xLKRvErL3uhY6e1BylFzG+aJiwQviDDTfOKeKTpY=
 github.com/gosuri/uitable v0.0.4/go.mod h1:tKR86bXuXPZazfOTG1FIzvjIdXzd0mo4Vtn16vt0PJo=
+github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
+github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
+github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/inconshreveable/mousetrap v1.0.1 h1:U3uMjPSQEBMNp1lFxmllqCPM6P5u/Xq7Pgzkat/bFNc=

internal/command/dev/dev.go

@@ -26,17 +26,6 @@ func NewCommand() *cobra.Command {
 }
 
 func runDev(cmd *cobra.Command, args []string) error {
-	// Initialize the logger
-	logger, err := zap.NewDevelopment()
-	if err != nil {
-		return err
-	}
-	defer func() {
-		if syncErr := logger.Sync(); syncErr != nil && err == nil {
-			err = syncErr
-		}
-	}()
-
 	if !filepath.IsAbs(devDataDirPath) {
 		pwd, err := os.Getwd()
 		if err != nil {
@@ -45,18 +34,8 @@ func runDev(cmd *cobra.Command, args []string) error {
 		devDataDirPath = path.Join(pwd, devDataDirPath)
 	}
 
-	dataDir, err := controller.NewDataDir(devDataDirPath)
+	devController, devWorker, err := CreateDevControllerAndWorker(devDataDirPath)
-	if err != nil {
-		return err
-	}
 
-	controller, err := controller.New(controller.WithDataDir(dataDir),
-		controller.WithInsecureAuthDisabled(), controller.WithLogger(logger))
-	if err != nil {
-		return err
-	}
-
-	worker, err := worker.New(worker.WithDataDirPath(devDataDirPath), worker.WithLogger(logger))
 	if err != nil {
 		return err
 	}
@@ -64,16 +43,47 @@ func runDev(cmd *cobra.Command, args []string) error {
 	errChan := make(chan error, 2)
 
 	go func() {
-		if err := controller.Run(cmd.Context()); err != nil {
+		if err := devController.Run(cmd.Context()); err != nil {
 			errChan <- err
 		}
 	}()
 
 	go func() {
-		if err := worker.Run(cmd.Context()); err != nil {
+		if err := devWorker.Run(cmd.Context()); err != nil {
 			errChan <- err
 		}
 	}()
 
 	return <-errChan
 }
+
+func CreateDevControllerAndWorker(devDataDirPath string) (*controller.Controller, *worker.Worker, error) {
+	// Initialize the logger
+	logger, err := zap.NewDevelopment()
+	if err != nil {
+		return nil, nil, err
+	}
+	defer func() {
+		if syncErr := logger.Sync(); syncErr != nil && err == nil {
+			err = syncErr
+		}
+	}()
+
+	dataDir, err := controller.NewDataDir(devDataDirPath)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	devController, err := controller.New(controller.WithDataDir(dataDir),
+		controller.WithInsecureAuthDisabled(), controller.WithLogger(logger))
+	if err != nil {
+		return nil, nil, err
+	}
+
+	devWorker, err := worker.New(worker.WithDataDirPath(devDataDirPath), worker.WithLogger(logger))
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return devController, devWorker, nil
+}

internal/controller/api.go

@@ -77,6 +77,12 @@ func (controller *Controller) initAPI() *gin.Engine {
 	v1.DELETE("/vms/:name", func(c *gin.Context) {
 		controller.deleteVM(c).Respond(c)
 	})
+	v1.GET("/vms/:name/events", func(c *gin.Context) {
+		controller.listVMEvents(c).Respond(c)
+	})
+	v1.POST("/vms/:name/events", func(c *gin.Context) {
+		controller.appendVMEvents(c).Respond(c)
+	})
 
 	return ginEngine
 }

internal/controller/api_vms.go

@@ -122,10 +122,65 @@ func (controller *Controller) deleteVM(ctx *gin.Context) responder.Responder {
 	name := ctx.Param("name")
 
 	return controller.storeUpdate(func(txn storepkg.Transaction) responder.Responder {
-		if err := txn.DeleteVM(name); err != nil {
+		vm, err := txn.GetVM(name)
+		if err != nil {
+			return responder.Error(err)
+		}
+		err = txn.DeleteVM(name)
+		if err != nil {
+			return responder.Error(err)
+		}
+		err = txn.DeleteEvents("vms", vm.UID)
+		if err != nil {
 			return responder.Error(err)
 		}
 
 		return responder.Code(http.StatusOK)
 	})
 }
+func (controller *Controller) appendVMEvents(ctx *gin.Context) responder.Responder {
+	if !controller.authorize(ctx, v1.ServiceAccountRoleComputeWrite) {
+		return responder.Code(http.StatusUnauthorized)
+	}
+
+	var events []v1.Event
+
+	if err := ctx.ShouldBindJSON(&events); err != nil {
+		return responder.Code(http.StatusBadRequest)
+	}
+
+	name := ctx.Param("name")
+
+	return controller.storeUpdate(func(txn storepkg.Transaction) responder.Responder {
+		vm, err := txn.GetVM(name)
+		if err != nil {
+			return responder.Error(err)
+		}
+		if err := txn.AppendEvents(events, "vms", vm.UID); err != nil {
+			return responder.Error(err)
+		}
+
+		return responder.Code(http.StatusOK)
+	})
+}
+
+func (controller *Controller) listVMEvents(ctx *gin.Context) responder.Responder {
+	if !controller.authorize(ctx, v1.ServiceAccountRoleComputeRead) {
+		return responder.Code(http.StatusUnauthorized)
+	}
+
+	name := ctx.Param("name")
+
+	return controller.storeView(func(txn storepkg.Transaction) responder.Responder {
+		vm, err := txn.GetVM(name)
+		if err != nil {
+			return responder.Error(err)
+		}
+		events, err := txn.ListEvents("vms", vm.UID)
+		if err != nil {
+			return responder.Error(err)
+		}
+
+		return responder.JSON(http.StatusOK, events)
+	})
+}

internal/controller/store/badger/badger_events.go

@@ -0,0 +1,98 @@
+package badger
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/dgraph-io/badger/v3"
+	"math/rand"
+	"path"
+)
+
+const SpaceEvents = "/events"
+
+func scopePrefix(scope []string) []byte {
+	keyParts := []string{SpaceEvents}
+	keyParts = append(keyParts, scope...)
+	return []byte(path.Join(keyParts...))
+}
+func (txn *Transaction) AppendEvents(events []v1.Event, scope ...string) (err error) {
+	defer func() {
+		err = mapErr(err)
+	}()
+
+	for index, event := range events {
+		valueBytes, err := json.Marshal(event)
+		if err != nil {
+			return err
+		}
+		//nolint:gosec
+		eventUID := fmt.Sprintf("/%d-%d-%d",
+			event.Timestamp,
+			index,         // to preserve order in case a batch of events has some events with the same timestamp
+			rand.Uint32(), // extra cautions to avoid collisions in case several batches of events are appended at the same time
+		)
+
+		eventKey := scopePrefix(scope)
+		eventKey = append(eventKey, []byte(eventUID)...)
+
+		err = txn.badgerTxn.Set(eventKey, valueBytes)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (txn *Transaction) ListEvents(scope ...string) (result []v1.Event, err error) {
+	defer func() {
+		err = mapErr(err)
+	}()
+
+	it := txn.badgerTxn.NewIterator(badger.IteratorOptions{
+		Prefix: scopePrefix(scope),
+	})
+	defer it.Close()
+
+	for it.Rewind(); it.Valid(); it.Next() {
+		item := it.Item()
+
+		eventBytes, err := item.ValueCopy(nil)
+		if err != nil {
+			return nil, err
+		}
+
+		var event v1.Event
+
+		if err := json.Unmarshal(eventBytes, &event); err != nil {
+			return nil, err
+		}
+
+		result = append(result, event)
+	}
+
+	return result, nil
+}
+
+func (txn *Transaction) DeleteEvents(scope ...string) (err error) {
+	defer func() {
+		err = mapErr(err)
+	}()
+
+	it := txn.badgerTxn.NewIterator(badger.IteratorOptions{
+		Prefix:         scopePrefix(scope),
+		AllVersions:    false,
+		PrefetchValues: false, // only need keys
+	})
+	defer it.Close()
+
+	for it.Rewind(); it.Valid(); it.Next() {
+		keyToDelete := it.Item().KeyCopy(nil)
+		err := txn.badgerTxn.Delete(keyToDelete)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

internal/controller/store/store.go

@@ -22,4 +22,8 @@ type Transaction interface {
 	SetServiceAccount(serviceAccount *v1.ServiceAccount) (err error)
 	DeleteServiceAccount(name string) (err error)
 	ListServiceAccounts() (result []*v1.ServiceAccount, err error)
+
+	AppendEvents(event []v1.Event, scope ...string) (err error)
+	ListEvents(scope ...string) (result []v1.Event, err error)
+	DeleteEvents(scope ...string) (err error)
 }

internal/orchardhome/orchardhome.go

@@ -15,6 +15,9 @@ func Path() (string, error) {
 		return "", fmt.Errorf("%w: failed to retrieve current user's home directory %v",
 			ErrFailed, err)
 	}
+	if orchardHome := os.Getenv("ORCHARD_HOME"); orchardHome != "" {
+		homeDir = orchardHome
+	}
 
 	orchardDir := filepath.Join(homeDir, ".orchard")
 

internal/tests/integration_test.go

@@ -0,0 +1,109 @@
+package tests_test
+
+import (
+	"context"
+	"errors"
+	"github.com/cirruslabs/orchard/internal/command/dev"
+	"github.com/cirruslabs/orchard/pkg/client"
+	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/stretchr/testify/assert"
+	"net/http"
+	"testing"
+	"time"
+)
+
+func TestSingleVM(t *testing.T) {
+	devClient := StartIntegrationTestEnvironment(t)
+
+	workers, err := devClient.Workers().List(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Equal(t, 1, len(workers))
+	err = devClient.VMs().Create(context.Background(), &v1.VM{
+		Meta: v1.Meta{
+			Name: "test-vm",
+		},
+		Image:    "ghcr.io/cirruslabs/macos-ventura-base:latest",
+		CPU:      4,
+		Memory:   8 * 1024,
+		Softnet:  false,
+		Headless: true,
+		Status:   v1.VMStatusPending,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.True(t, Wait(2*time.Minute, func() bool {
+		vm, err := devClient.VMs().Get(context.Background(), "test-vm")
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Logf("Waiting for the VM to start. Current status: %s", vm.Status)
+		return vm.Status == v1.VMStatusRunning
+	}), "failed to start a VM")
+	stoppingVM, err := devClient.VMs().Stop(context.Background(), "test-vm")
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Equal(t, v1.VMStatusStopping, stoppingVM.Status)
+	assert.True(t, Wait(2*time.Minute, func() bool {
+		vm, err := devClient.VMs().Get(context.Background(), "test-vm")
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Logf("Waiting for the VM to stop. Current status: %s", vm.Status)
+		return vm.Status == v1.VMStatusStopped
+	}), "failed to stop a VM")
+}
+
+func Wait(duration time.Duration, condition func() bool) bool {
+	ctx, cancel := context.WithTimeout(context.Background(), duration)
+	defer cancel()
+	for {
+		if condition() {
+			// all good
+			return true
+		}
+		select {
+		case <-ctx.Done():
+			return false
+		case <-time.After(5 * time.Second):
+			// try again
+			continue
+		}
+	}
+}
+
+func StartIntegrationTestEnvironment(t *testing.T) *client.Client {
+	t.Setenv("ORCHARD_HOME", t.TempDir())
+	devController, devWorker, err := dev.CreateDevControllerAndWorker(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		_ = devWorker.DeleteAllVMs()
+	})
+	devContext, cancelDevFunc := context.WithCancel(context.Background())
+	t.Cleanup(cancelDevFunc)
+	go func() {
+		err := devController.Run(devContext)
+		if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, http.ErrServerClosed) {
+			t.Errorf("dev controller failed: %v", err)
+		}
+	}()
+	go func() {
+		err := devWorker.Run(devContext)
+		if err != nil && !errors.Is(err, context.Canceled) {
+			t.Errorf("dev worker failed: %v", err)
+		}
+	}()
+
+	time.Sleep(5 * time.Second)
+
+	devClient, err := client.New()
+	if err != nil {
+		t.Fatal(err)
+	}
+	return devClient
+}

internal/worker/worker.go

@@ -7,6 +7,7 @@ import (
 	"github.com/cirruslabs/orchard/internal/worker/vmmanager"
 	"github.com/cirruslabs/orchard/pkg/client"
 	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/hashicorp/go-multierror"
 	"go.uber.org/zap"
 	"os"
 	"time"
@@ -162,7 +163,26 @@ func (worker *Worker) syncVMs(ctx context.Context) error {
 
 	worker.logger.Infof("syncing %d VMs...", len(remoteVMs))
 
-	// first try to sync local VMs with the remote ones
+	// check if need to stop any of the VMs
+	for _, vmResource := range remoteVMs {
+		if vmResource.Status == v1.VMStatusStopping && worker.vmm.Exists(vmResource) {
+			if err := worker.stopVM(vmResource); err != nil {
+				return err
+			}
+		}
+	}
+
+	// then, handle pending VMs first
+	for _, vmResource := range remoteVMs {
+		// handle pending VMs
+		if vmResource.Status == v1.VMStatusPending && !worker.vmm.Exists(vmResource) {
+			if err := worker.createVM(vmResource); err != nil {
+				return err
+			}
+		}
+	}
+
+	// lastly, try to sync local VMs with the remote ones
 	for _, vm := range worker.vmm.List() {
 		remoteVM, ok := remoteVMs[vm.Resource.UID]
 		if !ok {
@@ -179,25 +199,6 @@ func (worker *Worker) syncVMs(ctx context.Context) error {
 		}
 	}
 
-	// check if need to stop any of the VMs
-	for _, vmResource := range remoteVMs {
-		if vmResource.Status == v1.VMStatusStopping && worker.vmm.Exists(vmResource) {
-			if err := worker.stopVM(vmResource); err != nil {
-				return err
-			}
-		}
-	}
-
-	// finally, handle pending VMs first
-	for _, vmResource := range remoteVMs {
-		// handle pending VMs
-		if vmResource.Status == v1.VMStatusPending && !worker.vmm.Exists(vmResource) {
-			if err := worker.createVM(vmResource); err != nil {
-				return err
-			}
-		}
-	}
-
 	return nil
 }
 
@@ -249,3 +250,20 @@ func (worker *Worker) stopVM(vmResource v1.VM) error {
 	// Stop VM locally
 	return worker.vmm.Stop(vmResource)
 }
+
+func (worker *Worker) DeleteAllVMs() error {
+	var result error
+	for _, vm := range worker.vmm.List() {
+		err := vm.Stop()
+		if err != nil {
+			result = multierror.Append(result, err)
+		}
+	}
+	for _, vm := range worker.vmm.List() {
+		err := vm.Delete()
+		if err != nil {
+			result = multierror.Append(result, err)
+		}
+	}
+	return result
+}

pkg/resource/v1/v1.go

@@ -52,6 +52,12 @@ type VM struct {
 	Meta
 }
 
+type Event struct {
+	Kind      string `json:"kind"`
+	Timestamp uint64 `json:"timestamp"`
+	Payload   string `json:"payload"`
+}
+
 func (vm VM) TerminalState() bool {
 	return vm.Status == VMStatusStopped || vm.Status == VMStatusFailed
 }