azurepathfix: give job required env vars

pkg/operator/azurepathfixcontroller.go

@@ -11,38 +11,60 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	batchv1informers "k8s.io/client-go/informers/batch/v1"
+	corev1informers "k8s.io/client-go/informers/core/v1"
 	batchv1client "k8s.io/client-go/kubernetes/typed/batch/v1"
 	batchv1listers "k8s.io/client-go/listers/batch/v1"
+	corev1listers "k8s.io/client-go/listers/core/v1"
+	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 
 	operatorv1 "github.com/openshift/api/operator/v1"
+	configv1informers "github.com/openshift/client-go/config/informers/externalversions/config/v1"
+	configlisters "github.com/openshift/client-go/config/listers/config/v1"
+	imageregistryv1informers "github.com/openshift/client-go/imageregistry/informers/externalversions/imageregistry/v1"
+	imageregistryv1listers "github.com/openshift/client-go/imageregistry/listers/imageregistry/v1"
 	"github.com/openshift/library-go/pkg/operator/v1helpers"
 
 	"github.com/openshift/cluster-image-registry-operator/pkg/defaults"
 	"github.com/openshift/cluster-image-registry-operator/pkg/resource"
 )
 
 type AzurePathFixController struct {
-	batchClient    batchv1client.BatchV1Interface
-	operatorClient v1helpers.OperatorClient
-	jobLister      batchv1listers.JobNamespaceLister
+	batchClient               batchv1client.BatchV1Interface
+	operatorClient            v1helpers.OperatorClient
+	jobLister                 batchv1listers.JobNamespaceLister
+	imageRegistryConfigLister imageregistryv1listers.ConfigLister
+	secretLister              corev1listers.SecretNamespaceLister
+	infrastructureLister      configlisters.InfrastructureLister
+	proxyLister               configlisters.ProxyLister
+	kubeconfig                *restclient.Config
 
 	cachesToSync []cache.InformerSynced
 	queue        workqueue.RateLimitingInterface
 }
 
 func NewAzurePathFixController(
+	kubeconfig *restclient.Config,
 	batchClient batchv1client.BatchV1Interface,
 	operatorClient v1helpers.OperatorClient,
 	jobInformer batchv1informers.JobInformer,
+	imageRegistryConfigInformer imageregistryv1informers.ConfigInformer,
+	infrastructureInformer configv1informers.InfrastructureInformer,
+	secretInformer corev1informers.SecretInformer,
+	proxyInformer configv1informers.ProxyInformer,
 ) (*AzurePathFixController, error) {
 	c := &AzurePathFixController{
-		batchClient:    batchClient,
-		operatorClient: operatorClient,
-		jobLister:      jobInformer.Lister().Jobs(defaults.ImageRegistryOperatorNamespace),
-		queue:          workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "AzurePathFixController"),
+		batchClient:               batchClient,
+		operatorClient:            operatorClient,
+		jobLister:                 jobInformer.Lister().Jobs(defaults.ImageRegistryOperatorNamespace),
+		imageRegistryConfigLister: imageRegistryConfigInformer.Lister(),
+		infrastructureLister:      infrastructureInformer.Lister(),
+		secretLister:              secretInformer.Lister().Secrets(defaults.ImageRegistryOperatorNamespace),
+		proxyLister:               proxyInformer.Lister(),
+		kubeconfig:                kubeconfig,
+		queue:                     workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "AzurePathFixController"),
 	}
 
 	if _, err := jobInformer.Informer().AddEventHandler(c.eventHandler()); err != nil {
@@ -99,7 +121,19 @@ func (c *AzurePathFixController) processNextWorkItem() bool {
 
 func (c *AzurePathFixController) sync() error {
 	ctx := context.TODO()
-	gen := resource.NewGeneratorAzurePathFixJob(c.jobLister, c.batchClient)
+	imageRegistryConfig, err := c.imageRegistryConfigLister.Get("cluster")
+	if err != nil {
+		return err
+	}
+	gen := resource.NewGeneratorAzurePathFixJob(
+		c.jobLister,
+		c.batchClient,
+		c.secretLister,
+		c.infrastructureLister,
+		c.proxyLister,
+		imageRegistryConfig,
+		c.kubeconfig,
+	)
 
 	availableCondition := operatorv1.OperatorCondition{
 		Type:   "AzurePathFixAvailable",

pkg/operator/starter.go

@@ -169,9 +169,14 @@ func RunOperator(ctx context.Context, kubeconfig *restclient.Config) error {
 	}
 
 	azurePathFixController, err := NewAzurePathFixController(
+		kubeconfig,
 		kubeClient.BatchV1(),
 		configOperatorClient,
 		kubeInformers.Batch().V1().Jobs(),
+		imageregistryInformers.Imageregistry().V1().Configs(),
+		configInformers.Config().V1().Infrastructures(),
+		kubeInformers.Core().V1().Secrets(),
+		configInformers.Config().V1().Proxies(),
 	)
 	if err != nil {
 		return err

pkg/resource/azurepathfixjob.go

@@ -2,30 +2,57 @@ package resource
 
 import (
 	"context"
+	"fmt"
 	"os"
 
 	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
 	kcorev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	batchset "k8s.io/client-go/kubernetes/typed/batch/v1"
 	batchlisters "k8s.io/client-go/listers/batch/v1"
+	corev1listers "k8s.io/client-go/listers/core/v1"
+	restclient "k8s.io/client-go/rest"
 
+	configapiv1 "github.com/openshift/api/config/v1"
+	imageregistryv1 "github.com/openshift/api/imageregistry/v1"
+	configlisters "github.com/openshift/client-go/config/listers/config/v1"
 	"github.com/openshift/cluster-image-registry-operator/pkg/defaults"
+	"github.com/openshift/cluster-image-registry-operator/pkg/storage/azure"
 )
 
 var _ Mutator = &generatorAzurePathFixJob{}
 
 type generatorAzurePathFixJob struct {
-	lister batchlisters.JobNamespaceLister
-	client batchset.BatchV1Interface
+	lister               batchlisters.JobNamespaceLister
+	secretLister         corev1listers.SecretNamespaceLister
+	infrastructureLister configlisters.InfrastructureLister
+	proxyLister          configlisters.ProxyLister
+	client               batchset.BatchV1Interface
+	cr                   *imageregistryv1.Config
+	kubeconfig           *restclient.Config
 }
 
-func NewGeneratorAzurePathFixJob(lister batchlisters.JobNamespaceLister, client batchset.BatchV1Interface) *generatorAzurePathFixJob {
+func NewGeneratorAzurePathFixJob(
+	lister batchlisters.JobNamespaceLister,
+	client batchset.BatchV1Interface,
+	secretLister corev1listers.SecretNamespaceLister,
+	infrastructureLister configlisters.InfrastructureLister,
+	proxyLister configlisters.ProxyLister,
+	cr *imageregistryv1.Config,
+	kubeconfig *restclient.Config,
+) *generatorAzurePathFixJob {
 	return &generatorAzurePathFixJob{
-		lister: lister,
-		client: client,
+		lister:               lister,
+		client:               client,
+		cr:                   cr,
+		infrastructureLister: infrastructureLister,
+		secretLister:         secretLister,
+		proxyLister:          proxyLister,
+		kubeconfig:           kubeconfig,
 	}
 }
 
@@ -42,6 +69,48 @@ func (gapfj *generatorAzurePathFixJob) GetName() string {
 }
 
 func (gapfj *generatorAzurePathFixJob) expected() (runtime.Object, error) {
+	azureCfg, err := azure.GetConfig(gapfj.secretLister, gapfj.infrastructureLister)
+	if err != nil {
+		return nil, err
+	}
+	clusterProxy, err := gapfj.proxyLister.Get(defaults.ClusterProxyResourceName)
+	if errors.IsNotFound(err) {
+		clusterProxy = &configapiv1.Proxy{}
+	} else if err != nil {
+		// TODO: should we report Degraded?
+		return nil, fmt.Errorf("unable to get cluster proxy configuration: %v", err)
+	}
+
+	envs := []corev1.EnvVar{
+		{Name: "AZURE_STORAGE_ACCOUNT_NAME", Value: gapfj.cr.Spec.Storage.Azure.AccountName}, // use cr.Status ?
+		{Name: "AZURE_CONTAINER_NAME", Value: gapfj.cr.Spec.Storage.Azure.Container},
+		{Name: "AZURE_CLIENT_ID", Value: azureCfg.ClientID},
+		{Name: "AZURE_TENANT_ID", Value: azureCfg.TenantID},
+		{Name: "AZURE_CLIENT_SECRET", Value: azureCfg.ClientSecret},
+		{Name: "AZURE_FEDERATED_TOKEN_FILE", Value: azureCfg.FederatedTokenFile},
+	}
+	if len(gapfj.cr.Spec.Storage.Azure.CloudName) > 0 {
+		envs = append(envs, corev1.EnvVar{Name: "AZURE_ENVIRONMENT", Value: gapfj.cr.Spec.Storage.Azure.CloudName})
+	}
+
+	if gapfj.cr.Spec.Proxy.HTTP != "" {
+		envs = append(envs, corev1.EnvVar{Name: "HTTP_PROXY", Value: gapfj.cr.Spec.Proxy.HTTP})
+	} else if clusterProxy.Status.HTTPProxy != "" {
+		envs = append(envs, corev1.EnvVar{Name: "HTTP_PROXY", Value: clusterProxy.Status.HTTPProxy})
+	}
+
+	if gapfj.cr.Spec.Proxy.HTTPS != "" {
+		envs = append(envs, corev1.EnvVar{Name: "HTTPS_PROXY", Value: gapfj.cr.Spec.Proxy.HTTPS})
+	} else if clusterProxy.Status.HTTPSProxy != "" {
+		envs = append(envs, corev1.EnvVar{Name: "HTTPS_PROXY", Value: clusterProxy.Status.HTTPSProxy})
+	}
+
+	if gapfj.cr.Spec.Proxy.NoProxy != "" {
+		envs = append(envs, corev1.EnvVar{Name: "NO_PROXY", Value: gapfj.cr.Spec.Proxy.NoProxy})
+	} else if clusterProxy.Status.NoProxy != "" {
+		envs = append(envs, corev1.EnvVar{Name: "NO_PROXY", Value: clusterProxy.Status.NoProxy})
+	}
+
 	backoffLimit := int32(0)
 	cj := &batchv1.Job{
 		ObjectMeta: metav1.ObjectMeta{
@@ -65,11 +134,12 @@ func (gapfj *generatorAzurePathFixJob) expected() (runtime.Object, error) {
 								},
 							},
 							TerminationMessagePolicy: kcorev1.TerminationMessageFallbackToLogsOnError,
+							Env:                      envs,
 							Name:                     gapfj.GetName(),
 							Command:                  []string{"/bin/sh"},
 							Args: []string{
 								"-c",
-								"sleep 60",
+								"/usr/bin/move-blobs",
 							},
 						},
 					},