Gangplank/ocp: use per-function kubernetes client

The use of a package level client for Kuberntes restricts running
Gangplank in in a cluster. To make testing easer, func `k8sClient()` has
been renamed to `k8sInClusterClient()`, which is called by code that
requires an incluster client.

The goal of this refactor is allow for:
- mocking the Kube API's for testing
- make testing of a buildapv1.Build object easier
- allow for `cosa remote` to directly build worker pods without going
  through and intermediary pod
- reduce the package scope variable use

Signed-off-by: Ben Howard <[email protected]>

Author: Ben Howard

gangplank/cmd/ci.go

@@ -32,7 +32,7 @@ func init() {
 // useful.
 func runCI(c *cobra.Command, args []string) {
 	defer cancel()
-	m, err := ocp.NewCIBuilder(ctx, cosaOverrideImage, serviceAccount, specFile)
+	m, err := ocp.NewCIBuilder(ctx, true, cosaOverrideImage, serviceAccount, specFile)
 	if err != nil {
 		log.Fatalf("failed to define CI builder: %v", err)
 	}

gangplank/ocp/bc.go

@@ -74,13 +74,8 @@ func newBC() (*buildConfig, error) {
 		}
 	}
 
-	// Init the API Client for k8s itself
-	// The API client requires that the pod/buildconfig include a service account.
-	k8sAPIErr := k8sAPIClient()
-	if k8sAPIErr != nil && k8sAPIErr != ErrNotInCluster {
-		log.Errorf("Failed to initalized Kubernetes in cluster API client: %v", k8sAPIErr)
-		return nil, k8sAPIErr
-	}
+	// Open the Kubernetes Client
+	ac, pn, kubeErr := k8sInClusterClient()
 
 	// Init the OpenShift Build API Client.
 	buildAPIErr := ocpBuildClient()
@@ -89,6 +84,10 @@ func newBC() (*buildConfig, error) {
 		return nil, buildAPIErr
 	}
 
+	if kubeErr != nil && buildAPIErr != nil {
+		return nil, ErrInvalidOCPMode
+	}
+
 	// Query Kubernetes to find out what this pods network identity is.
 	// TODO: remove this CI exception once we have a kubernetes mock
 	if !forceNotInCluster {
@@ -102,7 +101,7 @@ func newBC() (*buildConfig, error) {
 			v.HostIP = apiBuild.Annotations[fmt.Sprintf(ciAnnotation, "IP")]
 		} else {
 			log.Info("Querying for pod ID")
-			hIP, err := getPodIP(v.HostPod)
+			hIP, err := getPodIP(ac, pn, v.HostPod)
 			if err != nil {
 				log.Errorf("Failed to determine buildconfig's pod")
 			}
@@ -117,11 +116,6 @@ func newBC() (*buildConfig, error) {
 		}).Info("found build.openshift.io/buildconfig identity")
 	}
 
-	// Build requires either a Build API Client or Kuberneres Cluster client.
-	if k8sAPIErr != nil && buildAPIErr != nil {
-		return nil, ErrInvalidOCPMode
-	}
-
 	if _, err := os.Stat(cosaSrvDir); os.IsNotExist(err) {
 		return nil, fmt.Errorf("Context dir %q does not exist", cosaSrvDir)
 	}
@@ -158,6 +152,11 @@ func (bc *buildConfig) Exec(ctx context.Context) error {
 		return err
 	}
 
+	ac, pn, err := k8sInClusterClient()
+	if err != nil {
+		return fmt.Errorf("failed create a kubernetes client: %w", err)
+	}
+
 	// Define, but do not start minio.
 	m := newMinioServer()
 	m.dir = cosaSrvDir
@@ -297,7 +296,12 @@ binary build interface.`)
 		}
 
 		index := n + 1
-		if err := createWorkerPod(ctx, index, eVars); err != nil {
+		cpod, err := NewCosaPodder(ctx, apiBuild, ac, pn, index)
+		if err != nil {
+			log.Errorf("FAILED TO CREATE POD DEFINITION: %v", err)
+			continue
+		}
+		if err := cpod.WorkerRunner(eVars); err != nil {
 			log.Errorf("FAILED stage: %v", err)
 		}
 	}

gangplank/ocp/bc_ci_test.go

@@ -19,7 +19,7 @@ func init() {
 
 func TestNoEnv(t *testing.T) {
 	if _, err := newBC(); err != ErrInvalidOCPMode {
-		t.Errorf("failed to raise: %v", ErrInvalidOCPMode)
+		t.Errorf("failed to raise error\n   want: %v\n    got: %v", ErrInvalidOCPMode, err)
 	}
 }
 

gangplank/ocp/ci.go

@@ -31,9 +31,18 @@ import (
 */
 
 type ci struct {
-	jobSpecFile string
-	bc          *buildConfig
-	ciBuild     string
+	apibuild *buildapiv1.Build
+	bc       *buildConfig
+	js       *spec.JobSpec
+
+	pod *v1.Pod
+
+	hostname         string
+	image            string
+	ipaddr           string
+	jobSpecFile      string
+	projectNamespace string
+	serviceAccount   string
 }
 
 // CIBuilder is the manual/unbounded Build interface.
@@ -57,16 +66,14 @@ const (
 
 func (c *ci) Exec(ctx context.Context) error {
 	log.Info("Executing unbounded builder")
+	if c.bc == nil {
+		return errors.New("buildconfig is nil")
+	}
 	return c.bc.Exec(ctx)
 }
 
 // NewCIBuilder returns a CIBuilder ready for execution.
-func NewCIBuilder(ctx context.Context, image, serviceAccount, jsF string) (CIBuilder, error) {
-	// Require a Kubernetes Service Account Client
-	if err := k8sAPIClient(); err != nil {
-		return nil, fmt.Errorf("failed create a kubernetes client: %w", err)
-	}
-
+func NewCIBuilder(ctx context.Context, inCluster bool, image, serviceAccount, jsF string) (CIBuilder, error) {
 	// Directly inject the jobspec
 	js, err := spec.JobSpecFromFile(jsF)
 	if err != nil {
@@ -75,56 +82,71 @@ func NewCIBuilder(ctx context.Context, image, serviceAccount, jsF string) (CIBui
 	if js.Recipe.GitURL == "" {
 		return nil, errors.New("JobSpec does inclue a Git Recipe")
 	}
-	os.Setenv("SOURCE_REF", js.Recipe.GitRef)
-	os.Setenv("SOURCE_URI", js.Recipe.GitURL)
 
-	log.WithFields(log.Fields{
-		"override image": image,
-		"source ref":     js.Recipe.GitRef,
-		"source uri":     js.Recipe.GitURL,
-	}).Info("jobspec defined source")
+	c := &ci{
+		js:             &js,
+		jobSpecFile:    jsF,
+		serviceAccount: serviceAccount,
+		image:          image,
+	}
 
-	// Get the ci API
-	ciBuild, err := ciAPIBuild(&js, image, serviceAccount)
+	// TODO: implement out-of-cluster
+	if err := c.setInCluster(); err != nil {
+		return nil, fmt.Errorf("failed setting incluster options: %v", err)
+	}
+
+	// Generate the build.openshift.io/v1 object
+	if err := c.generateAPIBuild(); err != nil {
+		return nil, fmt.Errorf("failed to generate api build: %v", err)
+	}
+	ciBuild, err := c.encodeAPIBuild()
 	if err != nil {
-		return nil, fmt.Errorf("failed to create a ci API build object")
+		return nil, fmt.Errorf("failed to encode apibuild: %v", err)
 	}
-	os.Setenv("BUILD", ciBuild)
 
 	// Create the buildConfig object
+	os.Setenv("BUILD", ciBuild)
+	os.Setenv("SOURCE_REF", js.Recipe.GitRef)
+	os.Setenv("SOURCE_URI", js.Recipe.GitURL)
 	bc, err := newBC()
 	if err != nil {
 		return nil, err
 	}
 	bc.JobSpec = js
 	bc.JobSpecFile = jsF
+	c.bc = bc
 
-	c := &ci{
-		jobSpecFile: jsF,
-		bc:          bc,
-		ciBuild:     ciBuild,
-	}
 	return c, nil
 }
 
-func ciAPIBuild(js *spec.JobSpec, image, serviceAccount string) (string, error) {
+func (c *ci) setInCluster() error {
 	// Dig deep and query find out what Kubernetes thinks this pod
 	// Discover where this running
 	hostname, ok := os.LookupEnv("HOSTNAME")
 	if !ok {
-		return "", errors.New("Unable to find hostname")
+		return errors.New("Unable to find hostname")
+	}
+	c.hostname = hostname
+
+	// Open the Kubernetes Client
+	ac, pn, err := k8sInClusterClient()
+	if err != nil {
+		return fmt.Errorf("failed create a kubernetes client: %w", err)
 	}
+	c.projectNamespace = pn
 
-	myIP, err := getPodIP(hostname)
+	myIP, err := getPodIP(ac, pn, hostname)
 	if err != nil {
-		return "", fmt.Errorf("failed to query my hostname: %w", err)
+		return fmt.Errorf("failed to query my hostname: %w", err)
 	}
+	c.ipaddr = myIP
 
 	// Discover where this running
-	myPod, err := apiClient.Pods(projectNamespace).Get(hostname, metav1.GetOptions{})
+	myPod, err := ac.CoreV1().Pods(pn).Get(hostname, metav1.GetOptions{})
 	if err != nil {
-		return "", err
+		return err
 	}
+	c.pod = myPod
 
 	// find the running pod this is running on.
 	var myContainer *v1.Container = nil
@@ -139,24 +161,19 @@ func ciAPIBuild(js *spec.JobSpec, image, serviceAccount string) (string, error)
 	}
 
 	// allow both the service account and the image to be overriden
-	if serviceAccount == "" {
-		serviceAccount = myPod.Spec.ServiceAccountName
+	if c.serviceAccount == "" {
+		c.serviceAccount = myPod.Spec.ServiceAccountName
 	}
-	if image == "" {
-		image = myContainer.Image
+	if c.image == "" {
+		c.image = myContainer.Image
 	}
-	if serviceAccount == "" || image == "" {
-		return "", errors.New("serviceAccount and image must be defined by running pod or via overrides")
+	if c.serviceAccount == "" || c.image == "" {
+		return errors.New("serviceAccount and image must be defined by running pod or via overrides")
 	}
+	return nil
+}
 
-	l := log.WithFields(log.Fields{
-		"ip":             myIP,
-		"image":          image,
-		"serviceAccount": serviceAccount,
-		"host":           myContainer.Name,
-	})
-	l.Info("identified pod")
-
+func (c *ci) generateAPIBuild() error {
 	// Create just _enough_ of the OpenShift BuildConfig spec
 	// Create a "ci" build.openshift.io/v1 specification.
 	ciBuildNumber := time.Now().Format("20060102150405")
@@ -167,7 +184,7 @@ func ciAPIBuild(js *spec.JobSpec, image, serviceAccount string) (string, error)
 		// ciRunnerTag is tested for to determine if this is
 		// a buildconfig or a faked one
 		ciRunnerTag:                     "true",
-		fmt.Sprintf(ciAnnotation, "IP"): myIP,
+		fmt.Sprintf(ciAnnotation, "IP"): c.ipaddr,
 		// Required Labels
 		buildapiv1.BuildConfigAnnotation: "ci-cosa-bc",
 		buildapiv1.BuildNumberAnnotation: ciBuildNumber,
@@ -180,25 +197,33 @@ func ciAPIBuild(js *spec.JobSpec, image, serviceAccount string) (string, error)
 
 	// Populate the Spec
 	a.Spec = buildapiv1.BuildSpec{}
-	a.Spec.ServiceAccount = myPod.Spec.ServiceAccountName
+	a.Spec.ServiceAccount = c.serviceAccount
 	a.Spec.Strategy = buildapiv1.BuildStrategy{}
 	a.Spec.Strategy.CustomStrategy = new(buildapiv1.CustomBuildStrategy)
 	a.Spec.Strategy.CustomStrategy.From = corev1.ObjectReference{
-		Name: image,
+		Name: c.image,
 	}
 	a.Spec.Source = buildapiv1.BuildSource{
 		ContextDir: cosaSrvDir,
 		Git: &buildapiv1.GitBuildSource{
-			Ref: js.Recipe.GitRef,
-			URI: js.Recipe.GitURL,
+			Ref: c.js.Recipe.GitRef,
+			URI: c.js.Recipe.GitURL,
 		},
 	}
 
-	// Render the ci buildapiv1 object to a JSON object.
-	// JSON is the messaginging interface for Kubernetes.
+	c.apibuild = &a
+	return nil
+}
+
+// encodeAPIBuilder the ci buildapiv1 object to a JSON object.
+// JSON is the messaginging interface for Kubernetes.
+func (c *ci) encodeAPIBuild() (string, error) {
+	if c.apibuild == nil {
+		return "", errors.New("apibuild is not defined yet")
+	}
 	aW := bytes.NewBuffer([]byte{})
 	s := json.NewYAMLSerializer(json.DefaultMetaFactory, buildScheme, buildScheme)
-	if err := s.Encode(&a, aW); err != nil {
+	if err := s.Encode(c.apibuild, aW); err != nil {
 		return "", err
 	}
 	d, err := ioutil.ReadAll(aW)

gangplank/ocp/cosa-pod.go

@@ -22,6 +22,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/kubernetes"
 )
 
 const (
@@ -89,48 +90,88 @@ var (
 	}
 )
 
-// setPodDefaults checks the Kubernetes version to determine if
-// we're on OCP 3.11 (v1.11) or 4.2(v1.15) or later.
-// https://docs.openshift.com/container-platform/4.2/release_notes/ocp-4-2-release-notes.html#ocp-4-2-about-this-release
-func setPodDefaults() error {
-	vi, err := apiClientSet.DiscoveryClient.ServerVersion()
+// cosaPod is a COSA pod
+type cosaPod struct {
+	apiBuild     *buildapiv1.Build
+	apiClientSet *kubernetes.Clientset
+	project      string
+
+	ocpInitCommand  []string
+	ocpRequirements v1.ResourceList
+	ocpSecContext   *v1.SecurityContext
+	volumes         []v1.Volume
+	volumeMounts    []v1.VolumeMount
+
+	ctx   context.Context
+	index int
+	pod   *v1.Pod
+}
+
+// CosaPodder create COSA capable pods.
+type CosaPodder interface {
+	WorkerRunner(envVar []v1.EnvVar) error
+}
+
+// a cosaPod is a CosaPodder
+var _ = CosaPodder(&cosaPod{})
+
+// NewCosaPodder creates a CosaPodder
+func NewCosaPodder(
+	ctx context.Context,
+	apiBuild *buildapiv1.Build,
+	apiClientSet *kubernetes.Clientset,
+	project string,
+	index int) (CosaPodder, error) {
+
+	cp := &cosaPod{
+		apiBuild:     apiBuild,
+		apiClientSet: apiClientSet,
+		project:      project,
+		ctx:          ctx,
+		index:        index,
+
+		// Set defaults for OCP4.x
+		ocpRequirements: ocpRequirements,
+		ocpSecContext:   ocpSecContext,
+		ocpInitCommand:  ocpInitCommand,
+
+		volumes:      volumes,
+		volumeMounts: volumeMounts,
+	}
+
+	vi, err := cp.apiClientSet.DiscoveryClient.ServerVersion()
 	if err != nil {
-		return fmt.Errorf("failed to query the kubernetes version: %w", err)
+		return nil, fmt.Errorf("failed to query the kubernetes version: %w", err)
 	}
 
 	minor, err := strconv.Atoi(strings.TrimRight(vi.Minor, "+"))
 	log.Infof("Kubernetes version of cluster is %s %s.%d", vi.String(), vi.Major, minor)
 	if err != nil {
-		return fmt.Errorf("failed to detect OCP cluster version: %v", err)
+		return nil, fmt.Errorf("failed to detect OCP cluster version: %v", err)
 	}
 	if minor >= 15 {
 		log.Info("Detected OpenShift 4.x cluster")
-		return nil
+		return cp, nil
 	}
 
 	log.Infof("Creating container with Openshift v3.x defaults")
-	ocpRequirements = ocp3Requirements
-	ocpInitCommand = ocp3InitCommand
-	ocpSecContext = ocp3SecContext
-	return nil
+	cp.ocpRequirements = ocp3Requirements
+	cp.ocpSecContext = ocp3SecContext
+	cp.ocpInitCommand = ocp3InitCommand
+
+	return cp, nil
 }
 
 func ptrInt(i int64) *int64 { return &i }
 
 func ptrBool(b bool) *bool { return &b }
 
-// Create creates a pod and ensures that the pod is cleaned up when the
-// process finishes
-// Inspired by https://github.com/kubernetes/client-go/blob/master/examples/create-update-delete-deployment/main.go
-func createWorkerPod(ctx context.Context, index int, eVars []v1.EnvVar) error {
-	if err := setPodDefaults(); err != nil {
-		log.WithField("err", err).Error("assuming OpenShift version v4.x")
-	}
-
+// WorkerRunner runs a worker pod and watches until finished
+func (cp *cosaPod) WorkerRunner(envVars []v1.EnvVar) error {
 	podName := fmt.Sprintf("%s-%s-worker-%d",
-		apiBuild.Annotations[buildapiv1.BuildConfigAnnotation],
-		apiBuild.Annotations[buildapiv1.BuildNumberAnnotation],
-		index,
+		cp.apiBuild.Annotations[buildapiv1.BuildConfigAnnotation],
+		cp.apiBuild.Annotations[buildapiv1.BuildNumberAnnotation],
+		cp.index,
 	)
 	log.Infof("Creating pod %s", podName)
 
@@ -144,10 +185,10 @@ func createWorkerPod(ctx context.Context, index int, eVars []v1.EnvVar) error {
 			"/usr/bin/gangplank",
 			"builder",
 		},
-		Env:             eVars,
+		Env:             envVars,
 		WorkingDir:      "/srv",
-		VolumeMounts:    volumeMounts,
-		SecurityContext: ocpSecContext,
+		VolumeMounts:    cp.volumeMounts,
+		SecurityContext: cp.ocpSecContext,
 		Resources: v1.ResourceRequirements{
 			Limits:   ocpRequirements,
 			Requests: ocpRequirements,
@@ -187,15 +228,16 @@ func createWorkerPod(ctx context.Context, index int, eVars []v1.EnvVar) error {
 		},
 	}
 
-	resp, err := apiClient.Pods(projectNamespace).Create(req)
+	ac := cp.apiClientSet.CoreV1()
+	resp, err := ac.Pods(cp.project).Create(req)
 	if err != nil {
 		return fmt.Errorf("failed to create pod %s: %w", podName, err)
 	}
-
 	log.Infof("Pod created: %s", podName)
+	cp.pod = req
 
 	status := resp.Status
-	w, err := apiClient.Pods(projectNamespace).Watch(
+	w, err := ac.Pods(cp.project).Watch(
 		metav1.ListOptions{
 			Watch:           true,
 			ResourceVersion: resp.ResourceVersion,
@@ -213,7 +255,7 @@ func createWorkerPod(ctx context.Context, index int, eVars []v1.EnvVar) error {
 	// ender is our clean-up that kill our pods
 	ender := func() {
 		l.Infof("terminating")
-		if err := apiClient.Pods(projectNamespace).Delete(podName, &metav1.DeleteOptions{}); err != nil {
+		if err := ac.Pods(cp.project).Delete(podName, &metav1.DeleteOptions{}); err != nil {
 			l.WithField("err", err).Error("Failed delete on pod, yolo.")
 		}
 	}
@@ -244,7 +286,7 @@ func createWorkerPod(ctx context.Context, index int, eVars []v1.EnvVar) error {
 				return nil
 			case v1.PodRunning:
 				l.Infof("Pod successfully completed")
-				if err := streamPodLogs(&logStarted, req); err != nil {
+				if err := cp.streamPodLogs(&logStarted, req); err != nil {
 					l.WithField("err", err).Error("failed to open logging")
 				}
 			case v1.PodFailed:
@@ -261,7 +303,7 @@ func createWorkerPod(ctx context.Context, index int, eVars []v1.EnvVar) error {
 		case <-sigs:
 			ender()
 			return errors.New("Termination requested")
-		case <-ctx.Done():
+		case <-cp.ctx.Done():
 			return nil
 		}
 	}
@@ -271,15 +313,15 @@ func createWorkerPod(ctx context.Context, index int, eVars []v1.EnvVar) error {
 // pods are responsible for their work, but not for their logs.
 // To make streamPodLogs thread safe and non-blocking, it expects
 // a pointer to a bool. If that pointer is nil or true, then we return
-func streamPodLogs(logging *bool, pod *v1.Pod) error {
+func (cp *cosaPod) streamPodLogs(logging *bool, pod *v1.Pod) error {
 	if logging != nil && *logging {
 		return nil
 	}
 	*logging = true
 	podLogOpts := v1.PodLogOptions{
 		Follow: true,
 	}
-	req := apiClient.Pods(projectNamespace).GetLogs(pod.Name, &podLogOpts)
+	req := cp.apiClientSet.CoreV1().Pods(cp.project).GetLogs(pod.Name, &podLogOpts)
 	podLogs, err := req.Stream()
 	if err != nil {
 		return err

gangplank/ocp/k8s.go

@@ -11,64 +11,52 @@ import (
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes"
-	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
 )
 
 const clusterNamespaceFile = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
 
 var (
-	// apiClient is v1 Client Interface for interacting Kubernetes
-	apiClient corev1.CoreV1Interface
-
-	// apiClientSet is a generic Kubernetes Client Set
-	apiClientSet *kubernetes.Clientset
-
-	// projectNamespace is the current namespace
-	projectNamespace string
-
 	// forceNotInCluster is used for testing. This is set to
 	// true for when testing is run with `-tag ci`
 	forceNotInCluster = false
 )
 
-// k8sAPIClient opens an in-cluster Kubernetes API client.
+// k8sInClusterClient opens an in-cluster Kubernetes API client.
 // The running pod must have a service account defined in the PodSpec.
-func k8sAPIClient() error {
+func k8sInClusterClient() (*kubernetes.Clientset, string, error) {
 	_, kport := os.LookupEnv("KUBERNETES_SERVICE_PORT")
 	_, khost := os.LookupEnv("KUBERNETES_SERVICE_HOST")
 	if !khost || !kport || forceNotInCluster {
-		return ErrNotInCluster
+		return nil, "", ErrNotInCluster
 	}
 
 	// creates the in-cluster config
 	cc, err := rest.InClusterConfig()
 	if err != nil {
-		return err
+		return nil, "", err
 	}
 
 	// creates the clientset
 	nc, err := kubernetes.NewForConfig(cc)
 	if err != nil {
-		return err
+		return nil, "", err
 	}
-	apiClient = nc.CoreV1()
-	apiClientSet = nc
 
 	pname, err := ioutil.ReadFile(clusterNamespaceFile)
 	if err != nil {
-		return fmt.Errorf("Failed determining the current namespace: %v", err)
+		return nil, "", fmt.Errorf("Failed determining the current namespace: %v", err)
 	}
-	projectNamespace = string(pname)
+	pn := string(pname)
 
-	log.Infof("Current project/namespace is %s", projectNamespace)
-	return nil
+	log.Infof("Current project/namespace is %s", pn)
+	return nc, pn, nil
 }
 
 // getPodiP returns the IP of a pod. getPodIP blocks pending until the podIP
 // is recieved.
-func getPodIP(podName string) (string, error) {
-	w, err := apiClient.Pods(projectNamespace).Watch(
+func getPodIP(cs *kubernetes.Clientset, podNamespace, podName string) (string, error) {
+	w, err := cs.CoreV1().Pods(podNamespace).Watch(
 		metav1.ListOptions{
 			Watch:         true,
 			FieldSelector: fields.Set{"metadata.name": podName}.AsSelector().String(),

gangplank/ocp/sa_secrets.go

@@ -9,6 +9,7 @@ import (
 
 	log "github.com/sirupsen/logrus"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 )
 
 /*
@@ -155,15 +156,15 @@ func (sm *secretMap) writeSecretFiles(toDir, name string, d map[string][]byte, r
 // 'coreos-assembler.coreos.com/secret=k' and then maps the secret
 // automatically in. "k" must be in the "known" secrets type to be mapped
 // automatically.
-func kubernetesSecretsSetup(toDir string) ([]string, error) {
+func kubernetesSecretsSetup(ac *kubernetes.Clientset, ns, toDir string) ([]string, error) {
 	lo := metav1.ListOptions{
 		LabelSelector: secretLabelName,
 		Limit:         100,
 	}
 
 	var ret []string
 
-	secrets, err := apiClient.Secrets(projectNamespace).List(lo)
+	secrets, err := ac.CoreV1().Secrets(ns).List(lo)
 	if err != nil {
 		return ret, err
 	}

gangplank/ocp/worker.go

@@ -49,12 +49,6 @@ func newWorkSpec(ctx context.Context) (*workSpec, error) {
 	if err := ws.Unmarshal(r); err != nil {
 		return nil, err
 	}
-
-	// Require a Kubernetes Service Account Client
-	if err := k8sAPIClient(); err != nil {
-		return nil, fmt.Errorf("failed create a kubernetes client: %w", err)
-	}
-
 	if _, err := os.Stat(cosaSrvDir); os.IsNotExist(err) {
 		return nil, fmt.Errorf("Context dir %q does not exist", cosaSrvDir)
 	}
@@ -85,12 +79,17 @@ func (ws *workSpec) Marshal() ([]byte, error) {
 
 // Exec executes the work spec tasks.
 func (ws *workSpec) Exec(ctx context.Context) error {
+	ac, pn, err := k8sInClusterClient()
+	if err != nil {
+		return fmt.Errorf("failed create a kubernetes client: %w", err)
+	}
+
 	// Workers always will use /srv
 	if err := os.Chdir(cosaSrvDir); err != nil {
 		return fmt.Errorf("unable to switch to %s: %w", cosaSrvDir, err)
 	}
 
-	ks, err := kubernetesSecretsSetup(cosaSrvDir)
+	ks, err := kubernetesSecretsSetup(ac, pn, cosaSrvDir)
 	if err != nil {
 		log.Errorf("Failed to setup Service Account Secrets: %v", err)
 	}