chore: Update to latest vsphere pkg

cmd/server/main.go

@@ -8,12 +8,10 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
-	"strconv"
 	"sync"
 	"syscall"
 	"time"
 
-	ce "github.com/cloudevents/sdk-go/v2"
 	"github.com/embano1/memlog"
 	"github.com/embano1/vsphere/event"
 	"github.com/embano1/vsphere/logger"
@@ -23,10 +21,6 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
-const (
-	eventFormat = "vmware.vsphere.%s.v0"
-)
-
 var pollInterval = time.Second // poll vcenter events
 
 func main() {
@@ -56,7 +50,7 @@ func main() {
 	ctx, cancel := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM)
 	defer cancel()
 
-	srv, err := newServer(ctx, env.Port)
+	srv, err := newServer(ctx, fmt.Sprintf("0.0.0.0:%d", env.Port))
 	if err != nil {
 		l.Fatal("could not create server", zap.Error(err))
 	}
@@ -130,23 +124,17 @@ func run(ctx context.Context, srv *server) error {
 						}
 					})
 
-					cevent := ce.NewEvent()
-
-					d := event.GetDetails(e)
-					cevent.SetSource(source)
-					cevent.SetID(strconv.Itoa(int(id)))
-					cevent.SetType(fmt.Sprintf(eventFormat, d.Type))
-					cevent.SetTime(e.GetEvent().CreatedTime)
-					if err := cevent.SetData(ce.ApplicationJSON, e); err != nil {
-						l.Error("set cloudevent data", zap.Error(err), zap.Any("event", e))
-						return fmt.Errorf("set cloudevent data: %v", err)
+					details := event.GetDetails(e)
+					cevent, err := event.ToCloudEvent(source, e, map[string]string{"eventclass": details.Class})
+					if err != nil {
+						l.Error("convert vsphere event to cloudevent", zap.Error(err), zap.Any("event", e))
+						return fmt.Errorf("convert vsphere event to cloudevent: %w", err)
 					}
-					cevent.SetExtension("eventclass", d.Class)
 
 					b, err := json.Marshal(cevent)
 					if err != nil {
 						l.Error("marshal cloudevent to JSON", zap.Error(err), zap.String("event", cevent.String()))
-						return fmt.Errorf("marshal cloudevent to JSON: %v", err)
+						return fmt.Errorf("marshal cloudevent to JSON: %w", err)
 					}
 
 					offset, err := srv.log.Write(egCtx, b)
@@ -164,7 +152,7 @@ func run(ctx context.Context, srv *server) error {
 	})
 
 	eg.Go(func() error {
-		l.Info("starting http listener", zap.Int("port", env.Port))
+		l.Info("starting http listener", zap.String("address", srv.http.Addr))
 		if err := srv.http.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
 			return fmt.Errorf("serve http: %w", err)
 		}

cmd/server/main_test.go

@@ -2,15 +2,19 @@ package main
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
 	"net/http"
+	"net/http/httptest"
 	"os"
 	"path/filepath"
 	"testing"
 	"time"
 
-	"github.com/embano1/memlog"
+	ce "github.com/cloudevents/sdk-go/v2"
 	"github.com/embano1/vsphere/client"
 	"github.com/embano1/vsphere/logger"
+	"github.com/julienschmidt/httprouter"
 	"github.com/vmware/govmomi/simulator"
 	_ "github.com/vmware/govmomi/vapi/simulator"
 	"github.com/vmware/govmomi/vim25"
@@ -36,7 +40,7 @@ func Test_run(t *testing.T) {
 			ctx = logger.Set(ctx, zaptest.NewLogger(t))
 
 			pollInterval = time.Millisecond * 10
-			ctx, cancel := context.WithTimeout(ctx, time.Second)
+			ctx, cancel := context.WithCancel(ctx)
 			defer cancel()
 
 			t.Setenv("VCENTER_URL", vimclient.URL().String())
@@ -46,23 +50,46 @@ func Test_run(t *testing.T) {
 			vc, err := client.New(ctx)
 			assert.NilError(t, err)
 
-			log, err := memlog.New(ctx)
+			const address = "127.0.0.1:8080"
+			srv, err := newServer(ctx, address)
 			assert.NilError(t, err)
 
-			srv := server{
-				http: &http.Server{
-					Addr: "127.0.0.1:0",
-				},
-				vc:  vc,
-				log: log,
-			}
+			srv.vc = vc
 
-			err = run(ctx, &srv)
-			assert.ErrorType(t, err, context.DeadlineExceeded)
+			runErrCh := make(chan error)
+			go func() {
+				runErrCh <- run(ctx, srv)
+			}()
 
-			// check the server (log) received events from VC
-			_, latest := log.Range(context.TODO())
-			assert.Assert(t, latest != -1)
+			// give server time to initialize event stream log
+			time.Sleep(time.Second)
+
+			wanteventID := "20"
+			rec := httptest.NewRecorder()
+			req := httptest.NewRequest(
+				http.MethodGet,
+				fmt.Sprintf("http://%s/api/v1/events/%s", address, wanteventID),
+				nil,
+			)
+
+			h := srv.getEvent(ctx)
+			h(rec, req, httprouter.Params{{
+				Key:   "id",
+				Value: wanteventID,
+			}})
+
+			assert.Equal(t, rec.Code, http.StatusOK)
+
+			var gotevent ce.Event
+			err = json.NewDecoder(rec.Body).Decode(&gotevent)
+			assert.NilError(t, err)
+			assert.NilError(t, gotevent.Validate())
+			assert.Equal(t, gotevent.ID(), wanteventID)
+
+			// stop server
+			cancel()
+			err = <-runErrCh
+			assert.ErrorContains(t, err, "context canceled")
 
 			return nil
 		})

cmd/server/server.go

@@ -50,7 +50,7 @@ type envConfig struct {
 	Debug       bool          `envconfig:"DEBUG" default:"false"`
 }
 
-func newServer(ctx context.Context, port int) (*server, error) {
+func newServer(ctx context.Context, address string) (*server, error) {
 	var srv server
 	vc, err := client.New(ctx)
 	if err != nil {
@@ -63,9 +63,8 @@ func newServer(ctx context.Context, port int) (*server, error) {
 	router.GET(apiPath+"/events/:id", srv.getEvent(ctx))
 	router.GET(apiPath+"/range", srv.getRange(ctx))
 
-	bindAddr := fmt.Sprintf("0.0.0.0:%d", port)
 	h := http.Server{
-		Addr:         bindAddr,
+		Addr:         address,
 		Handler:      router,
 		ReadTimeout:  readTimeout,
 		WriteTimeout: streamTimeout,

go.mod

@@ -5,7 +5,7 @@ go 1.17
 require (
 	github.com/cloudevents/sdk-go/v2 v2.10.1
 	github.com/embano1/memlog v0.4.3
-	github.com/embano1/vsphere v0.2.3
+	github.com/embano1/vsphere v0.2.4
 	github.com/google/go-cmp v0.5.8
 	github.com/google/uuid v1.3.0
 	github.com/julienschmidt/httprouter v1.3.0

go.sum

@@ -9,8 +9,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/embano1/memlog v0.4.3 h1:riVTumM3e3/qBrsKYsml7sSJYvCm6z4V4+4C3hRk0aA=
 github.com/embano1/memlog v0.4.3/go.mod h1:yDNCaHBS5MqgHjF5SRlCibLaGqSrcC9DjG3AomLj+bk=
-github.com/embano1/vsphere v0.2.3 h1:zFqG85PZsPyu1EVPq6WHwtmyVxkM+NUD4F5bwl419vA=
-github.com/embano1/vsphere v0.2.3/go.mod h1:+xavYhmRnFo2wHDGC//Ht/WsJVwqUmXzBXZNY+1YHaQ=
+github.com/embano1/vsphere v0.2.4 h1:h19uT62wxZkVOFBpvpFG98D8Fm/7h5iptm3j77ZFD+w=
+github.com/embano1/vsphere v0.2.4/go.mod h1:A1M+PubKxIb2FquVeC7qNZn0H2sK0H/YXiPa7LHxs00=
 github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=