From 0582108ea65b7425e4283e3670102c3ce621e1f1 Mon Sep 17 00:00:00 2001 From: Fedor Korotkov Date: Mon, 13 Mar 2023 08:04:17 -0400 Subject: [PATCH] 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 * Use POST * Make newEventKey private * Append events in batches * Lint issues * Private `scopePrefix` --------- Co-authored-by: Nikolay Edigaryev --- api/openapi.yaml | 45 ++++++++ go.mod | 2 + go.sum | 4 + internal/command/dev/dev.go | 58 ++++++---- internal/controller/api.go | 6 + internal/controller/api_vms.go | 57 ++++++++- .../controller/store/badger/badger_events.go | 98 ++++++++++++++++ internal/controller/store/store.go | 4 + internal/orchardhome/orchardhome.go | 3 + internal/tests/integration_test.go | 109 ++++++++++++++++++ internal/worker/worker.go | 58 ++++++---- pkg/resource/v1/v1.go | 6 + 12 files changed, 405 insertions(+), 45 deletions(-) create mode 100644 internal/controller/store/badger/badger_events.go create mode 100644 internal/tests/integration_test.go diff --git a/api/openapi.yaml b/api/openapi.yaml index 143672f..b97d257 100644 --- a/api/openapi.yaml +++ b/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 diff --git a/go.mod b/go.mod index f072804..df46a47 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 6c7de66..ac760ad 100644 --- a/go.sum +++ b/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= diff --git a/internal/command/dev/dev.go b/internal/command/dev/dev.go index 4fd15e2..bc213e5 100644 --- a/internal/command/dev/dev.go +++ b/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) - if err != nil { - return err - } + devController, devWorker, err := CreateDevControllerAndWorker(devDataDirPath) - 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 +} diff --git a/internal/controller/api.go b/internal/controller/api.go index c854509..e185212 100644 --- a/internal/controller/api.go +++ b/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 } diff --git a/internal/controller/api_vms.go b/internal/controller/api_vms.go index c9b0522..6340bb5 100644 --- a/internal/controller/api_vms.go +++ b/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) + }) +} diff --git a/internal/controller/store/badger/badger_events.go b/internal/controller/store/badger/badger_events.go new file mode 100644 index 0000000..7237dea --- /dev/null +++ b/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 +} diff --git a/internal/controller/store/store.go b/internal/controller/store/store.go index cc456e1..2c7a385 100644 --- a/internal/controller/store/store.go +++ b/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) } diff --git a/internal/orchardhome/orchardhome.go b/internal/orchardhome/orchardhome.go index 50a840f..9e543d7 100644 --- a/internal/orchardhome/orchardhome.go +++ b/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") diff --git a/internal/tests/integration_test.go b/internal/tests/integration_test.go new file mode 100644 index 0000000..722d0bc --- /dev/null +++ b/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 +} diff --git a/internal/worker/worker.go b/internal/worker/worker.go index 5884f5a..3b81966 100644 --- a/internal/worker/worker.go +++ b/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 +} diff --git a/pkg/resource/v1/v1.go b/pkg/resource/v1/v1.go index 56abd27..2d8c10d 100644 --- a/pkg/resource/v1/v1.go +++ b/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 }