Enable reading workflows from multiple namespaces in tink-server (#969)

## Description

Removes the default setting where not specifying the kubernetes-namespace defaults it to whatever namespace the controller is running in, and changes the workflowID to namespace/name so it can find the workflows

## Why is this needed

tinkerbell/cluster-api-provider-tinkerbell#385

With this change, you can create hardware and workflow resources in different namespaces.

Fixes: #

## How Has This Been Tested?
We have a cluster-api setup where we're adding some bare metal nodes to a cluster.
With this change, the workflows that previously only worked from the tink-system namespace now also work from a different namespace.  I also tested the old working setup and that still works as well.
The change is minimal, so it shouldn't impact much.
I haven't tested if the --kube-namespace setting would restrict it to one namespace again.


## How are existing users impacted? What migration steps/scripts do we need?

No migration steps are needed, unless users have multiple instances of tink-server running in different namespaces, or have another reason why they specifically don't want resources in a different namespace to be picked up.

This could probably be avoided by having the helm chart add the kube-namespace argument to the deployment and have it pull the value from the downward api somehow, but it seems to me that having it default to looking at all namespaces would be preferrable for most users.

## Checklist:

I have:

- [ ] updated the documentation and/or roadmap (if required)
- [ ] added unit or e2e tests
- [ ] provided instructions on how to upgrade

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)