refactor: clean up client code

server/auth/auth.go

@@ -16,6 +16,13 @@ import (
 	authorizationv1 "k8s.io/api/authorization/v1"
 )
 
+type key int
+
+const (
+	// ContextClientKey is the key used to identify the Client value in Context
+	ContextClientKey key = iota
+)
+
 func getBearerToken(ctx context.Context) (*string, bool) {
 	md, ok := metadata.FromIncomingContext(ctx)
 	if !ok {
@@ -58,7 +65,7 @@ func getClient(ctx context.Context, kubeConfig *v1.Config, db *v1.DB) (context.C
 		return nil, err
 	}
 
-	return context.WithValue(ctx, "kubeClient", client), nil
+	return context.WithValue(ctx, ContextClientKey, client), nil
 }
 
 func IsAuthorized(c *v1.Client, namespace, verb, group, resource, name string) (allowed bool, err error) {

server/auth_server.go

@@ -22,7 +22,7 @@ func (a *AuthServer) IsWorkspaceAuthenticated(ctx context.Context, request *api.
 	if ctx == nil {
 		return &empty.Empty{}, nil
 	}
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	md, ok := metadata.FromIncomingContext(ctx)
 	if !ok {
 		return &empty.Empty{}, errors.New("Error parsing headers.")
@@ -53,7 +53,7 @@ func (a *AuthServer) IsAuthorized(ctx context.Context, request *api.IsAuthorized
 		return res, status.Error(codes.Unauthenticated, "Unauthenticated.")
 	}
 	//User auth check
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	err = a.isValidToken(err, client)
 	if err != nil {
 		return nil, err
@@ -74,7 +74,7 @@ func (a *AuthServer) IsValidToken(ctx context.Context, req *api.IsValidTokenRequ
 		return nil, status.Error(codes.Unauthenticated, "Unauthenticated.")
 	}
 
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 
 	err = a.isValidToken(err, client)
 	if err != nil {

server/cron_workflow_server.go

@@ -45,7 +45,7 @@ func apiCronWorkflow(cwf *v1.CronWorkflow) (cronWorkflow *api.CronWorkflow) {
 }
 
 func (c *CronWorkflowServer) CreateCronWorkflow(ctx context.Context, req *api.CreateCronWorkflowRequest) (*api.CronWorkflow, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "create", "argoproj.io", "cronworkflows", "")
 	if err != nil || !allowed {
 		return nil, err
@@ -93,7 +93,7 @@ func (c *CronWorkflowServer) CreateCronWorkflow(ctx context.Context, req *api.Cr
 }
 
 func (c *CronWorkflowServer) UpdateCronWorkflow(ctx context.Context, req *api.UpdateCronWorkflowRequest) (*api.CronWorkflow, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "update", "argoproj.io", "cronworkflows", "")
 	if err != nil || !allowed {
 		return nil, err
@@ -142,7 +142,7 @@ func (c *CronWorkflowServer) UpdateCronWorkflow(ctx context.Context, req *api.Up
 }
 
 func (c *CronWorkflowServer) GetCronWorkflow(ctx context.Context, req *api.GetCronWorkflowRequest) (*api.CronWorkflow, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "get", "argoproj.io", "cronworkflows", req.Uid)
 	if err != nil || !allowed {
 		return nil, err
@@ -155,7 +155,7 @@ func (c *CronWorkflowServer) GetCronWorkflow(ctx context.Context, req *api.GetCr
 }
 
 func (c *CronWorkflowServer) ListCronWorkflows(ctx context.Context, req *api.ListCronWorkflowRequest) (*api.ListCronWorkflowsResponse, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "list", "argoproj.io", "cronworkflows", "")
 	if err != nil || !allowed {
 		return nil, err
@@ -186,7 +186,7 @@ func (c *CronWorkflowServer) ListCronWorkflows(ctx context.Context, req *api.Lis
 }
 
 func (c *CronWorkflowServer) DeleteCronWorkflow(ctx context.Context, req *api.DeleteCronWorkflowRequest) (*empty.Empty, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "delete", "argoproj.io", "cronworkflows", "")
 	if err != nil || !allowed {
 		return nil, err

server/label_server.go

@@ -56,7 +56,7 @@ func NewLabelServer() *LabelServer {
 func (s *LabelServer) GetLabels(ctx context.Context, req *api.GetLabelsRequest) (*api.GetLabelsResponse, error) {
 	argoResource := resourceIdentifierToArgoResource(req.Resource)
 
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "get", "argoproj.io", argoResource, "")
 	if err != nil || !allowed {
 		return nil, err
@@ -75,7 +75,7 @@ func (s *LabelServer) GetLabels(ctx context.Context, req *api.GetLabelsRequest)
 func (s *LabelServer) AddLabels(ctx context.Context, req *api.AddLabelsRequest) (*api.GetLabelsResponse, error) {
 	argoResource := resourceIdentifierToArgoResource(req.Resource)
 
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "create", "argoproj.io", argoResource, "")
 	if err != nil || !allowed {
 		return nil, err
@@ -99,7 +99,7 @@ func (s *LabelServer) AddLabels(ctx context.Context, req *api.AddLabelsRequest)
 func (s *LabelServer) ReplaceLabels(ctx context.Context, req *api.ReplaceLabelsRequest) (*api.GetLabelsResponse, error) {
 	argoResource := resourceIdentifierToArgoResource(req.Resource)
 
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "update", "argoproj.io", argoResource, "")
 	if err != nil || !allowed {
 		return nil, err
@@ -123,7 +123,7 @@ func (s *LabelServer) ReplaceLabels(ctx context.Context, req *api.ReplaceLabelsR
 func (s *LabelServer) DeleteLabel(ctx context.Context, req *api.DeleteLabelRequest) (*api.GetLabelsResponse, error) {
 	argoResource := resourceIdentifierToArgoResource(req.Resource)
 
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	// update verb here since we are not deleting the resource, but labels
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "update", "argoproj.io", argoResource, "")
 	if err != nil || !allowed {

server/namespace_server.go

@@ -25,7 +25,7 @@ func apiNamespace(ns *v1.Namespace) (namespace *api.Namespace) {
 }
 
 func (s *NamespaceServer) ListNamespaces(ctx context.Context, req *api.ListNamespacesRequest) (*api.ListNamespacesResponse, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, "", "list", "", "namespaces", "")
 	if err != nil || !allowed {
 		return nil, err
@@ -72,7 +72,7 @@ func (s *NamespaceServer) ListNamespaces(ctx context.Context, req *api.ListNames
 }
 
 func (s *NamespaceServer) CreateNamespace(ctx context.Context, createNamespace *api.CreateNamespaceRequest) (*api.Namespace, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, "", "create", "", "namespaces", "")
 	if err != nil || !allowed {
 		return nil, err

server/secret_server.go

@@ -23,7 +23,7 @@ func apiSecret(s *v1.Secret) *api.Secret {
 }
 
 func (s *SecretServer) CreateSecret(ctx context.Context, req *api.CreateSecretRequest) (*empty.Empty, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "create", "", "secrets", "")
 	if err != nil || !allowed {
 		return nil, err
@@ -40,7 +40,7 @@ func (s *SecretServer) CreateSecret(ctx context.Context, req *api.CreateSecretRe
 }
 
 func (s *SecretServer) SecretExists(ctx context.Context, req *api.SecretExistsRequest) (secretExists *api.SecretExistsResponse, err error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "get", "", "secrets", req.Name)
 	if err != nil || !allowed {
 		return nil, err
@@ -58,7 +58,7 @@ func (s *SecretServer) SecretExists(ctx context.Context, req *api.SecretExistsRe
 }
 
 func (s *SecretServer) GetSecret(ctx context.Context, req *api.GetSecretRequest) (*api.Secret, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "get", "", "secrets", req.Name)
 	if err != nil || !allowed {
 		return nil, err
@@ -72,7 +72,7 @@ func (s *SecretServer) GetSecret(ctx context.Context, req *api.GetSecretRequest)
 }
 
 func (s *SecretServer) ListSecrets(ctx context.Context, req *api.ListSecretsRequest) (*api.ListSecretsResponse, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "list", "", "secrets", "")
 	if err != nil || !allowed {
 		return nil, err
@@ -95,7 +95,7 @@ func (s *SecretServer) ListSecrets(ctx context.Context, req *api.ListSecretsRequ
 }
 
 func (s *SecretServer) DeleteSecret(ctx context.Context, req *api.DeleteSecretRequest) (deleted *api.DeleteSecretResponse, err error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "delete", "", "secrets", req.Name)
 	if err != nil || !allowed {
 		return nil, err
@@ -113,7 +113,7 @@ func (s *SecretServer) DeleteSecret(ctx context.Context, req *api.DeleteSecretRe
 }
 
 func (s *SecretServer) DeleteSecretKey(ctx context.Context, req *api.DeleteSecretKeyRequest) (deleted *api.DeleteSecretKeyResponse, err error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "delete", "", "secrets", req.SecretName)
 	if err != nil || !allowed {
 		return nil, err
@@ -139,7 +139,7 @@ func (s *SecretServer) DeleteSecretKey(ctx context.Context, req *api.DeleteSecre
 }
 
 func (s *SecretServer) AddSecretKeyValue(ctx context.Context, req *api.AddSecretKeyValueRequest) (updated *api.AddSecretKeyValueResponse, err error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "delete", "", "secrets", req.Secret.Name)
 	if err != nil || !allowed {
 		return nil, err
@@ -163,7 +163,7 @@ func (s *SecretServer) AddSecretKeyValue(ctx context.Context, req *api.AddSecret
 }
 
 func (s *SecretServer) UpdateSecretKeyValue(ctx context.Context, req *api.UpdateSecretKeyValueRequest) (updated *api.UpdateSecretKeyValueResponse, err error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "update", "", "secrets", req.Secret.Name)
 	if err != nil || !allowed {
 		return nil, err

server/server.go

@@ -1,5 +1,15 @@
 package server
 
+import (
+	"context"
+	v1 "github.com/onepanelio/core/pkg"
+	"github.com/onepanelio/core/server/auth"
+)
+
 const (
 	TimeLayout = "2006-01-02 15:04:05"
 )
+
+func getClient(ctx context.Context) *v1.Client {
+	return ctx.Value(auth.ContextClientKey).(*v1.Client)
+}

server/workflow_server.go

@@ -64,7 +64,7 @@ func apiWorkflowExecution(wf *v1.WorkflowExecution) (workflow *api.WorkflowExecu
 }
 
 func (s *WorkflowServer) CreateWorkflowExecution(ctx context.Context, req *api.CreateWorkflowExecutionRequest) (*api.WorkflowExecution, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "create", "argoproj.io", "workflows", "")
 	if err != nil || !allowed {
 		return nil, err
@@ -93,7 +93,7 @@ func (s *WorkflowServer) CreateWorkflowExecution(ctx context.Context, req *api.C
 }
 
 func (s *WorkflowServer) CloneWorkflowExecution(ctx context.Context, req *api.CloneWorkflowExecutionRequest) (*api.WorkflowExecution, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "create", "argoproj.io", "workflows", "")
 	if err != nil || !allowed {
 		return nil, err
@@ -108,7 +108,7 @@ func (s *WorkflowServer) CloneWorkflowExecution(ctx context.Context, req *api.Cl
 }
 
 func (s *WorkflowServer) AddWorkflowExecutionStatistics(ctx context.Context, req *api.AddWorkflowExecutionStatisticRequest) (*empty.Empty, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	phase := v1alpha1.NodeFailed
 	if req.Statistics.WorkflowStatus == "Succeeded" {
 		phase = v1alpha1.NodeSucceeded
@@ -133,7 +133,7 @@ func (s *WorkflowServer) AddWorkflowExecutionStatistics(ctx context.Context, req
 // instead pass in the cron workflow uid, we can load the cron workflow from db that way and get
 // all required data.
 func (s *WorkflowServer) CronStartWorkflowExecutionStatistic(ctx context.Context, req *api.CronStartWorkflowExecutionStatisticRequest) (*empty.Empty, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "get", "argoproj.io", "workflows", req.Uid)
 	if err != nil || !allowed {
 		return &empty.Empty{}, err
@@ -148,7 +148,7 @@ func (s *WorkflowServer) CronStartWorkflowExecutionStatistic(ctx context.Context
 }
 
 func (s *WorkflowServer) GetWorkflowExecution(ctx context.Context, req *api.GetWorkflowExecutionRequest) (*api.WorkflowExecution, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "get", "argoproj.io", "workflows", req.Uid)
 	if err != nil || !allowed {
 		return nil, err
@@ -171,7 +171,7 @@ func (s *WorkflowServer) GetWorkflowExecution(ctx context.Context, req *api.GetW
 }
 
 func (s *WorkflowServer) WatchWorkflowExecution(req *api.WatchWorkflowExecutionRequest, stream api.WorkflowService_WatchWorkflowExecutionServer) error {
-	client := stream.Context().Value("kubeClient").(*v1.Client)
+	client := getClient(stream.Context())
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "get", "argoproj.io", "workflows", req.Uid)
 	if err != nil || !allowed {
 		return err
@@ -195,7 +195,7 @@ func (s *WorkflowServer) WatchWorkflowExecution(req *api.WatchWorkflowExecutionR
 }
 
 func (s *WorkflowServer) GetWorkflowExecutionLogs(req *api.GetWorkflowExecutionLogsRequest, stream api.WorkflowService_GetWorkflowExecutionLogsServer) error {
-	client := stream.Context().Value("kubeClient").(*v1.Client)
+	client := getClient(stream.Context())
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "get", "argoproj.io", "workflows", req.Uid)
 	if err != nil || !allowed {
 		return err
@@ -225,7 +225,7 @@ func (s *WorkflowServer) GetWorkflowExecutionLogs(req *api.GetWorkflowExecutionL
 }
 
 func (s *WorkflowServer) GetWorkflowExecutionMetrics(ctx context.Context, req *api.GetWorkflowExecutionMetricsRequest) (*api.GetWorkflowExecutionMetricsResponse, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "get", "argoproj.io", "workflows", req.Uid)
 	if err != nil || !allowed {
 		return nil, err
@@ -252,7 +252,7 @@ func (s *WorkflowServer) GetWorkflowExecutionMetrics(ctx context.Context, req *a
 }
 
 func (s *WorkflowServer) ListWorkflowExecutions(ctx context.Context, req *api.ListWorkflowExecutionsRequest) (*api.ListWorkflowExecutionsResponse, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "list", "argoproj.io", "workflows", "")
 	if err != nil || !allowed {
 		return nil, err
@@ -284,7 +284,7 @@ func (s *WorkflowServer) ListWorkflowExecutions(ctx context.Context, req *api.Li
 }
 
 func (s *WorkflowServer) ResubmitWorkflowExecution(ctx context.Context, req *api.ResubmitWorkflowExecutionRequest) (*api.WorkflowExecution, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "create", "argoproj.io", "workflows", req.Uid)
 	if err != nil || !allowed {
 		return nil, err
@@ -299,7 +299,7 @@ func (s *WorkflowServer) ResubmitWorkflowExecution(ctx context.Context, req *api
 }
 
 func (s *WorkflowServer) TerminateWorkflowExecution(ctx context.Context, req *api.TerminateWorkflowExecutionRequest) (*empty.Empty, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "update", "argoproj.io", "workflows", "")
 	if err != nil || !allowed {
 		return nil, err
@@ -314,7 +314,7 @@ func (s *WorkflowServer) TerminateWorkflowExecution(ctx context.Context, req *ap
 }
 
 func (s *WorkflowServer) GetArtifact(ctx context.Context, req *api.GetArtifactRequest) (*api.ArtifactResponse, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "get", "argoproj.io", "workflows", req.Uid)
 	if err != nil || !allowed {
 		return nil, err
@@ -331,7 +331,7 @@ func (s *WorkflowServer) GetArtifact(ctx context.Context, req *api.GetArtifactRe
 }
 
 func (s *WorkflowServer) ListFiles(ctx context.Context, req *api.ListFilesRequest) (*api.ListFilesResponse, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "get", "argoproj.io", "workflows", req.Uid)
 	if err != nil || !allowed {
 		return nil, err
@@ -375,7 +375,7 @@ func (s *WorkflowServer) ListFiles(ctx context.Context, req *api.ListFilesReques
 }
 
 func (s *WorkflowServer) UpdateWorkflowExecutionStatus(ctx context.Context, req *api.UpdateWorkflowExecutionStatusRequest) (*empty.Empty, error) {
-	client := ctx.Value("kubeClient").(*v1.Client)
+	client := getClient(ctx)
 	allowed, err := auth.IsAuthorized(client, req.Namespace, "update", "argoproj.io", "workflows", req.Uid)
 	if err != nil || !allowed {
 		return &empty.Empty{}, err