Enable reading workflows from multiple namespaces in tink-server

internal/e2e/e2e_test.go

@@ -164,17 +164,17 @@ var _ = Describe("Tink API", func() {
 
 			// expected workflow name to context mapping
 			expectedWorkflows := map[string]*proto.WorkflowContext{
-				"wf1": {
-					WorkflowId:           "wf1",
+				"default/wf1": {
+					WorkflowId:           "default/wf1",
 					CurrentWorker:        "3c:ec:ef:4c:4f:54",
 					CurrentTask:          "os-installation",
 					CurrentAction:        "stream-image",
 					CurrentActionIndex:   0,
 					CurrentActionState:   proto.State_STATE_PENDING,
 					TotalNumberOfActions: 3,
 				},
-				"wf3": {
-					WorkflowId:           "wf3",
+				"default/wf3": {
+					WorkflowId:           "default/wf3",
 					CurrentWorker:        "3c:ec:ef:4c:4f:54",
 					CurrentTask:          "task-1",
 					CurrentAction:        "task-1-action-1",
@@ -193,7 +193,7 @@ var _ = Describe("Tink API", func() {
 				if !googleproto.Equal(want, got) {
 					fmt.Printf("Expected:\n\t%#v\nGot:\n\t%#v", want, got)
 				}
-				Expect(googleproto.Equal(want, got)).To(Equal(true), fmt.Sprintf("Didn't find expected context for %s", got.WorkflowId))
+				Expect(googleproto.Equal(want, got)).To(Equal(true), fmt.Sprintf("Not equal context as expected for %s", got.WorkflowId))
 
 				// Remove the key from the map
 				delete(expectedWorkflows, got.WorkflowId)

internal/server/kubernetes_api.go

@@ -43,15 +43,6 @@ func NewKubeBackedServer(logger logr.Logger, kubeconfig, apiserver, namespace st
 		return nil, err
 	}
 
-	namespace, _, err = ccfg.Namespace()
-	if err != nil {
-		return nil, err
-	}
-
-	if err != nil {
-		return nil, err
-	}
-
 	return NewKubeBackedServerFromREST(logger, cfg, namespace)
 }
 
@@ -61,8 +52,10 @@ func NewKubeBackedServerFromREST(logger logr.Logger, config *rest.Config, namesp
 	clstr, err := cluster.New(config, func(opts *cluster.Options) {
 		opts.Scheme = controller.DefaultScheme()
 		opts.Logger = zapr.NewLogger(zap.NewNop())
-		opts.Cache.DefaultNamespaces = map[string]cache.Config{
-			namespace: {},
+		if namespace != "" {
+			opts.Cache.DefaultNamespaces = map[string]cache.Config{
+				namespace: {},
+			}
 		}
 	})
 	if err != nil {
@@ -89,7 +82,6 @@ func NewKubeBackedServerFromREST(logger logr.Logger, config *rest.Config, namesp
 	return &KubernetesBackedServer{
 		logger:     logger,
 		ClientFunc: clstr.GetClient,
-		namespace:  namespace,
 		nowFunc:    time.Now,
 	}, nil
 }
@@ -98,7 +90,6 @@ func NewKubeBackedServerFromREST(logger logr.Logger, config *rest.Config, namesp
 type KubernetesBackedServer struct {
 	logger     logr.Logger
 	ClientFunc func() client.Client
-	namespace  string
 
 	nowFunc func() time.Time
 }

internal/server/kubernetes_api_test.go

@@ -421,7 +421,6 @@ func TestModifyWorkflowState(t *testing.T) {
 			server := &KubernetesBackedServer{
 				logger:     zapr.NewLogger(zap.Must(zap.NewDevelopment())),
 				ClientFunc: nil,
-				namespace:  "default",
 				nowFunc:    TestTime.Now,
 			}
 			gotErr := server.modifyWorkflowState(tc.inputWf, tc.inputWfContext)

internal/server/kubernetes_api_workflow.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"context"
+	"strings"
 
 	"github.com/pkg/errors"
 	"github.com/tinkerbell/tink/api/v1alpha1"
@@ -24,7 +25,7 @@ const (
 
 func getWorkflowContext(wf v1alpha1.Workflow) *proto.WorkflowContext {
 	return &proto.WorkflowContext{
-		WorkflowId:           wf.Name,
+		WorkflowId:           wf.Namespace + "/" + wf.Name,
 		CurrentWorker:        wf.GetCurrentWorker(),
 		CurrentTask:          wf.GetCurrentTask(),
 		CurrentAction:        wf.GetCurrentAction(),
@@ -52,9 +53,10 @@ func (s *KubernetesBackedServer) getCurrentAssignedNonTerminalWorkflowsForWorker
 	return wfs, nil
 }
 
-func (s *KubernetesBackedServer) getWorkflowByName(ctx context.Context, workflowID, namespace string) (*v1alpha1.Workflow, error) {
+func (s *KubernetesBackedServer) getWorkflowByName(ctx context.Context, workflowID string) (*v1alpha1.Workflow, error) {
+	workflowNamespace, workflowName, _ := strings.Cut(workflowID, "/")
 	wflw := &v1alpha1.Workflow{}
-	err := s.ClientFunc().Get(ctx, types.NamespacedName{Name: workflowID, Namespace: namespace}, wflw)
+	err := s.ClientFunc().Get(ctx, types.NamespacedName{Name: workflowName, Namespace: workflowNamespace}, wflw)
 	if err != nil {
 		s.logger.Error(err, "get client", "workflow", workflowID)
 		return nil, err
@@ -85,7 +87,7 @@ func (s *KubernetesBackedServer) GetWorkflowActions(ctx context.Context, req *pr
 	if wfID == "" {
 		return nil, status.Errorf(codes.InvalidArgument, errInvalidWorkflowID)
 	}
-	wf, err := s.getWorkflowByName(ctx, wfID, s.namespace)
+	wf, err := s.getWorkflowByName(ctx, wfID)
 	if err != nil {
 		return nil, err
 	}
@@ -189,7 +191,7 @@ func (s *KubernetesBackedServer) ReportActionStatus(ctx context.Context, req *pr
 	wfID := req.GetWorkflowId()
 	l := s.logger.WithValues("actionName", req.GetActionName(), "status", req.GetActionStatus(), "workflowID", req.GetWorkflowId(), "taskName", req.GetTaskName(), "worker", req.WorkerId)
 
-	wf, err := s.getWorkflowByName(ctx, wfID, s.namespace)
+	wf, err := s.getWorkflowByName(ctx, wfID)
 	if err != nil {
 		l.Error(err, "get workflow")
 		return nil, status.Errorf(codes.InvalidArgument, errInvalidWorkflowID)