Add Kubernetes Resource Model to Tink API

* Add e2e ginkgo tests

Signed-off-by: Micah Hausler <[email protected]>

.github/workflows/ci.yaml

@@ -41,6 +41,9 @@ jobs:
       - name: Generate
         run: nix-shell --run 'make generate'
 
+      - name: e2etest
+        run: make e2etest-setup
+
       - name: go test
         run: make test
 

.gitignore

@@ -23,3 +23,6 @@ out/
 
 .*.swp
 hack/tools
+
+# test worker files
+tests/worker

Makefile

@@ -11,8 +11,8 @@ crosscompile: $(crossbinaries) ## Build all binaries for Linux and all supported
 images: tink-cli-image tink-server-image tink-worker-image virtual-worker-image ## Build all docker images
 run: crosscompile run-stack ## Builds and runs the Tink stack (tink, db, cli) via docker-compose
 
-test: ## Run tests
-	go test -coverprofile=coverage.txt ./...
+test: e2etest-setup ## Run tests
+	source <(setup-envtest use  -p env) && go test -coverprofile=coverage.txt ./...
 
 verify: lint check-generated # Verify code style, is lint free, freshness ...
 	gofumpt -s -d .

cmd/tink-controller/main.go

@@ -13,6 +13,7 @@ import (
 	"github.com/tinkerbell/tink/pkg/controllers"
 	wfctrl "github.com/tinkerbell/tink/pkg/controllers/workflow"
 	"k8s.io/client-go/tools/clientcmd"
+	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
 )
 
 // version is set at build time.
@@ -58,12 +59,22 @@ func NewRootCommand(config *DaemonConfig, logger log.Logger) *cobra.Command {
 		RunE: func(cmd *cobra.Command, args []string) error {
 			logger.Info("starting controller version " + version)
 
-			config, err := clientcmd.BuildConfigFromFlags(config.K8sAPI, config.Kubeconfig)
+			ccfg := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
+				&clientcmd.ClientConfigLoadingRules{ExplicitPath: config.Kubeconfig},
+				&clientcmd.ConfigOverrides{ClusterInfo: clientcmdapi.Cluster{Server: config.K8sAPI}})
+
+			cfg, err := ccfg.ClientConfig()
 			if err != nil {
 				return err
 			}
 
-			manager, err := controllers.NewManager(config, controllers.GetControllerOptions())
+			namespace, _, err := ccfg.Namespace()
+			if err != nil {
+				return err
+			}
+			options := controllers.GetControllerOptions()
+			options.LeaderElectionNamespace = namespace
+			manager, err := controllers.NewManager(cfg, options)
 			if err != nil {
 				return err
 			}

cmd/tink-server/main.go

@@ -32,17 +32,20 @@ var version = "devel"
 // DaemonConfig represents all the values you can configure as part of the tink-server.
 // You can change the configuration via environment variable, or file, or command flags.
 type DaemonConfig struct {
-	Facility      string
-	PGDatabase    string
-	PGUSer        string
-	PGPassword    string
-	PGSSLMode     string
-	OnlyMigration bool
-	GRPCAuthority string
-	TLSCert       string
-	CertDir       string
-	HTTPAuthority string
-	TLS           bool
+	Facility                string
+	PGDatabase              string
+	PGUSer                  string
+	PGPassword              string
+	PGSSLMode               string
+	OnlyMigration           bool
+	GRPCAuthority           string
+	TLSCert                 string
+	CertDir                 string
+	HTTPAuthority           string
+	TLS                     bool
+	KubernetesResourceModel bool
+	KubeconfigPath          string
+	KubeAPI                 string
 }
 
 func (c *DaemonConfig) AddFlags(fs *pflag.FlagSet) {
@@ -57,6 +60,9 @@ func (c *DaemonConfig) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&c.CertDir, "cert-dir", "", "")
 	fs.StringVar(&c.HTTPAuthority, "http-authority", ":42114", "The address used to expose the HTTP server")
 	fs.BoolVar(&c.TLS, "tls", true, "Run in tls protected mode (disabling should only be done for development or if behind TLS terminating proxy)")
+	fs.BoolVar(&c.KubernetesResourceModel, "kubernetes-backend", false, "Feature flag to use the Kubernetes Resource Model")
+	fs.StringVar(&c.KubeconfigPath, "kubeconfig", "", "The path to the Kubeconfig. Only takes effect if `--kubernetes-backend=true`")
+	fs.StringVar(&c.KubeAPI, "kube-api", "", "The kubernetes API endpoitn. Only takes effect if `--kubernetes-backend=true`")
 }
 
 func (c *DaemonConfig) PopulateFromLegacyEnvVar() {
@@ -125,28 +131,12 @@ func NewRootCommand(config *DaemonConfig, logger log.Logger) *cobra.Command {
 			// figure this out in another PR
 			errCh := make(chan error, 2)
 
-			// TODO(gianarb): I moved this up because we need to be sure that both
-			// connection, the one used for the resources and the one used for
-			// listening to events and notification are coming in the same way.
-			// BUT we should be using the right flags
-			connInfo := fmt.Sprintf("dbname=%s user=%s password=%s sslmode=%s",
-				config.PGDatabase,
-				config.PGUSer,
-				config.PGPassword,
-				config.PGSSLMode,
-			)
-			database, err := internal.SetupPostgres(connInfo, config.OnlyMigration, logger)
-			if err != nil {
-				return err
-			}
-			if config.OnlyMigration {
-				return nil
-			}
-
 			var (
+				registrar   grpcserver.Registrar
 				grpcOpts    []grpc.ServerOption
 				certPEM     []byte
 				certModTime *time.Time
+				err         error
 			)
 			if config.TLS {
 				certsDir := os.Getenv("TINKERBELL_CERTS_DIR")
@@ -161,19 +151,44 @@ func NewRootCommand(config *DaemonConfig, logger log.Logger) *cobra.Command {
 				grpcOpts = append(grpcOpts, grpc.Creds(credentials.NewServerTLSFromCert(cert)))
 			}
 
-			tinkAPI, err := server.NewDBServer(
-				logger,
-				database,
-				server.WithCerts(*certModTime, certPEM),
-			)
-			if err != nil {
-				return err
-			}
+			if config.KubernetesResourceModel {
+				var err error
+				registrar, err = server.NewKubeBackedServer(logger, config.KubeconfigPath, config.KubeAPI)
+				if err != nil {
+					return err
+				}
+			} else {
+				// TODO(gianarb): I moved this up because we need to be sure that both
+				// connection, the one used for the resources and the one used for
+				// listening to events and notification are coming in the same way.
+				// BUT we should be using the right flags
+				connInfo := fmt.Sprintf("dbname=%s user=%s password=%s sslmode=%s",
+					config.PGDatabase,
+					config.PGUSer,
+					config.PGPassword,
+					config.PGSSLMode,
+				)
+				database, err := internal.SetupPostgres(connInfo, config.OnlyMigration, logger)
+				if err != nil {
+					return err
+				}
+				if config.OnlyMigration {
+					return nil
+				}
 
+				registrar, err = server.NewDBServer(
+					logger,
+					database,
+					server.WithCerts(*certModTime, certPEM),
+				)
+				if err != nil {
+					return err
+				}
+			}
 			// Start the gRPC server in the background
 			addr, err := grpcserver.SetupGRPC(
 				ctx,
-				tinkAPI,
+				registrar,
 				config.GRPCAuthority,
 				grpcOpts,
 				errCh)

cmd/tink-worker/worker/worker.go

@@ -207,11 +207,11 @@ func (w *Worker) execute(ctx context.Context, wfID string, action *pb.WorkflowAc
 	if err != nil {
 		return st, errors.Wrap(err, "wait container")
 	}
-
-	l.With("status", st.String()).Info("container removed")
+	// Removing the below line, its duplicated from above
+	// l.With("status", st.String()).Info("container removed")
 
 	if st == pb.State_STATE_SUCCESS {
-		l.With("status", st).Info("action container exited with success", st)
+		l.With("status", st).Info("action container exited with success")
 		return st, nil
 	}
 
@@ -299,15 +299,6 @@ func (w *Worker) ProcessWorkflowActions(ctx context.Context) error {
 					nextAction = actions.GetActionList()[wfContext.GetCurrentActionIndex()]
 					actionIndex = int(wfContext.GetCurrentActionIndex())
 				}
-				l := l.With(
-					"currentWorker", wfContext.GetCurrentWorker(),
-					"currentTask", wfContext.GetCurrentTask(),
-					"currentAction", wfContext.GetCurrentAction(),
-					"currentActionIndex", strconv.FormatInt(wfContext.GetCurrentActionIndex(), 10),
-					"currentActionState", wfContext.GetCurrentActionState(),
-					"totalNumberOfActions", wfContext.GetTotalNumberOfActions(),
-				)
-				l.Info("current context")
 				if nextAction.GetWorkerId() == w.workerID {
 					turn = true
 				}
@@ -338,12 +329,13 @@ func (w *Worker) ProcessWorkflowActions(ctx context.Context) error {
 						os.Exit(1)
 					}
 				}
-				l.Info("starting with action")
+				l.Info("starting action")
 			}
 
 			for turn {
 				action := actions.GetActionList()[actionIndex]
-				l := l.With("actionName", action.GetName(),
+				l := l.With(
+					"actionName", action.GetName(),
 					"taskName", action.GetTaskName(),
 				)
 				ctx := context.WithValue(ctx, loggingContextKey, &l)
@@ -361,7 +353,7 @@ func (w *Worker) ProcessWorkflowActions(ctx context.Context) error {
 					if err != nil {
 						exitWithGrpcError(err, l)
 					}
-					l.With("duration", strconv.FormatInt(actionStatus.Seconds, 10)).Info("sent action status")
+					l.With("status", actionStatus.ActionStatus, "duration", strconv.FormatInt(actionStatus.Seconds, 10)).Info("sent action status")
 				}
 
 				// get workflow data
@@ -445,6 +437,7 @@ func (w *Worker) reportActionStatus(ctx context.Context, actionStatus *pb.Workfl
 		"workerID", actionStatus.GetWorkerId(),
 		"actionName", actionStatus.GetActionName(),
 		"taskName", actionStatus.GetTaskName(),
+		"status", actionStatus.ActionStatus,
 	)
 	var err error
 	for r := 1; r <= w.retries; r++ {

cmd/virtual-worker/cmd/root.go

@@ -64,7 +64,11 @@ func NewRootCommand(version string, logger log.Logger) *cobra.Command {
 				return err
 			}
 			rClient := pb.NewWorkflowServiceClient(conn)
-
+			logger.With(
+				"worker", workerID,
+				"server", conn.Target(),
+				"state", conn.GetState(),
+			).Info("Connected to server")
 			containerManager := worker.NewFakeContainerManager(logger, 1000, 2000)
 			logCapturer := worker.NewEmptyLogCapturer()
 
@@ -76,6 +80,7 @@ func NewRootCommand(version string, logger log.Logger) *cobra.Command {
 				logger,
 				tinkWorker.WithMaxFileSize(maxFileSize),
 				tinkWorker.WithRetries(retryInterval, retries),
+				tinkWorker.WithDataDir("./worker"),
 				tinkWorker.WithLogCapture(captureActionLogs))
 
 			err = w.ProcessWorkflowActions(ctx)

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/docker/distribution v2.8.1+incompatible
 	github.com/docker/docker v20.10.7+incompatible
 	github.com/equinix-labs/otel-init-go v0.0.1
-	github.com/go-logr/zapr v0.4.0
+	github.com/go-logr/zapr v1.2.0
 	github.com/go-openapi/strfmt v0.19.3 // indirect
 	github.com/golang/protobuf v1.5.2
 	github.com/google/go-cmp v0.5.6
@@ -19,6 +19,8 @@ require (
 	github.com/lib/pq v1.10.1
 	github.com/matryer/moq v0.2.3
 	github.com/moby/term v0.0.0-20210619224110-3f7ff695adc6 // indirect
+	github.com/onsi/ginkgo/v2 v2.1.3
+	github.com/onsi/gomega v1.18.1
 	github.com/opencontainers/image-spec v1.0.2
 	github.com/packethost/pkg v0.0.0-20200903155310-0433e0605550
 	github.com/pkg/errors v0.9.1
@@ -37,12 +39,14 @@ require (
 	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0
 	google.golang.org/protobuf v1.27.1
 	gopkg.in/yaml.v2 v2.4.0
-	k8s.io/apimachinery v0.22.2
-	k8s.io/client-go v0.22.2
+	k8s.io/apimachinery v0.23.0
+	k8s.io/client-go v0.23.0
 	knative.dev/pkg v0.0.0-20211119170723-a99300deff34
 	mvdan.cc/gofumpt v0.1.1
-	sigs.k8s.io/controller-runtime v0.10.1
-	sigs.k8s.io/controller-tools v0.7.0
+	sigs.k8s.io/controller-runtime v0.11.1
+	sigs.k8s.io/controller-runtime/tools/setup-envtest v0.0.0-20220304125252-9ee63fc65a97
+	sigs.k8s.io/controller-tools v0.8.0
+	sigs.k8s.io/yaml v1.3.0
 )
 
 replace github.com/stormcat24/protodep => github.com/ackintosh/protodep v0.0.0-20200728152107-abf8eb579d6c