Skip to content

Commit

Permalink
Add DataImportCron CronJobs Proxy support (#2455)
Browse files Browse the repository at this point in the history 
* Add DataImportCron CronJobs Proxy support

Signed-off-by: Arnon Gilboa <[email protected]>

* CR fixes

Signed-off-by: Arnon Gilboa <[email protected]>

* Add cronjob env utest

Signed-off-by: Arnon Gilboa <[email protected]>

Signed-off-by: Arnon Gilboa <[email protected]>
  • Loading branch information
@arnongilboa
arnongilboa authored Nov 3, 2022
1 parent b08fd8e commit eef0358
Show file tree
Hide file tree
Showing 11 changed files with 335 additions and 141 deletions.
1 change: 1 addition & 0 deletions pkg/controller/BUILD.bazel
View file
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,7 @@ go_test(
"//vendor/k8s.io/client-go/tools/cache:go_default_library",
"//vendor/k8s.io/client-go/tools/record:go_default_library",
"//vendor/k8s.io/cluster-bootstrap/token/api:go_default_library",
"//vendor/k8s.io/utils/pointer:go_default_library",
"//vendor/kubevirt.io/controller-lifecycle-operator-sdk/api:go_default_library",
"//vendor/sigs.k8s.io/controller-runtime/pkg/client:go_default_library",
"//vendor/sigs.k8s.io/controller-runtime/pkg/client/fake:go_default_library",
Expand Down
64 changes: 33 additions & 31 deletions pkg/controller/dataimportcron-controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ import (
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/validation"
"k8s.io/client-go/tools/record"
"k8s.io/utils/pointer"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/event"
Expand Down Expand Up @@ -771,14 +772,25 @@ func (r *DataImportCronReconciler) newCronJob(cron *cdiv1.DataImportCron) (*batc
TerminationMessagePolicy: corev1.TerminationMessageReadFile,
}

volumes := []corev1.Volume{}
hasCertConfigMap := regSource.CertConfigMap != nil && *regSource.CertConfigMap != ""
if hasCertConfigMap {
vm := corev1.VolumeMount{
Name: CertVolName,
MountPath: common.ImporterCertDir,
}
container.VolumeMounts = []corev1.VolumeMount{vm}
container.VolumeMounts = append(container.VolumeMounts, vm)
container.Command = append(container.Command, "-certdir", common.ImporterCertDir)
volumes = append(volumes, createConfigMapVolume(CertVolName, *regSource.CertConfigMap))
}

if volName, _ := GetImportProxyConfig(cdiConfig, common.ImportProxyConfigMapName); volName != "" {
vm := corev1.VolumeMount{
Name: ProxyCertVolName,
MountPath: common.ImporterProxyCertDir,
}
container.VolumeMounts = append(container.VolumeMounts, vm)
volumes = append(volumes, createConfigMapVolume(ProxyCertVolName, volName))
}

if regSource.SecretRef != nil && *regSource.SecretRef != "" {
Expand Down Expand Up @@ -808,20 +820,25 @@ func (r *DataImportCronReconciler) newCronJob(cron *cdiv1.DataImportCron) (*batc
)
}

addEnvVar := func(varName, value string) {
container.Env = append(container.Env, corev1.EnvVar{Name: varName, Value: value})
}

if insecureTLS {
container.Env = append(container.Env,
corev1.EnvVar{
Name: common.InsecureTLSVar,
Value: "true",
},
)
addEnvVar(common.InsecureTLSVar, "true")
}

successfulJobsHistoryLimit := int32(0)
failedJobsHistoryLimit := int32(0)
ttlSecondsAfterFinished := int32(0)
backoffLimit := int32(2)
gracePeriodSeconds := int64(0)
addEnvVarFromImportProxyConfig := func(varName string) {
if value, err := GetImportProxyConfig(cdiConfig, varName); err == nil {
addEnvVar(varName, value)
} else {
r.log.Info("Missing", varName, err.Error())
}
}

addEnvVarFromImportProxyConfig(common.ImportProxyHTTP)
addEnvVarFromImportProxyConfig(common.ImportProxyHTTPS)
addEnvVarFromImportProxyConfig(common.ImportProxyNoProxy)

cronJobName := GetCronJobName(cron)
cronJob := &batchv1.CronJob{
Expand All @@ -832,39 +849,24 @@ func (r *DataImportCronReconciler) newCronJob(cron *cdiv1.DataImportCron) (*batc
Spec: batchv1.CronJobSpec{
Schedule: cron.Spec.Schedule,
ConcurrencyPolicy: batchv1.ForbidConcurrent,
SuccessfulJobsHistoryLimit: &successfulJobsHistoryLimit,
FailedJobsHistoryLimit: &failedJobsHistoryLimit,
SuccessfulJobsHistoryLimit: pointer.Int32(1),
JobTemplate: batchv1.JobTemplateSpec{
Spec: batchv1.JobSpec{
Template: corev1.PodTemplateSpec{
Spec: corev1.PodSpec{
RestartPolicy: corev1.RestartPolicyNever,
TerminationGracePeriodSeconds: &gracePeriodSeconds,
TerminationGracePeriodSeconds: pointer.Int64(0),
Containers: []corev1.Container{container},
ServiceAccountName: common.CronJobServiceAccountName,
Volumes: volumes,
},
},
TTLSecondsAfterFinished: &ttlSecondsAfterFinished,
BackoffLimit: &backoffLimit,
BackoffLimit: pointer.Int32(2),
},
},
},
}

if hasCertConfigMap {
vol := corev1.Volume{
Name: CertVolName,
VolumeSource: corev1.VolumeSource{
ConfigMap: &corev1.ConfigMapVolumeSource{
LocalObjectReference: corev1.LocalObjectReference{
Name: *regSource.CertConfigMap,
},
},
},
}
cronJob.Spec.JobTemplate.Spec.Template.Spec.Volumes = []corev1.Volume{vol}
}

if err := r.setJobCommon(cron, cronJob); err != nil {
return nil, err
}
Expand Down
89 changes: 85 additions & 4 deletions pkg/controller/dataimportcron-controller_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ import (
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/tools/record"
"k8s.io/utils/pointer"

"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
Expand All @@ -50,8 +51,12 @@ import (
)

var (
cronLog = logf.Log.WithName("data-import-cron-controller-test")
cronName = "test-cron"
cronLog = logf.Log.WithName("data-import-cron-controller-test")
cronName = "test-cron"
httpProxy = "test-http-proxy"
httpsProxy = "test-https-proxy"
noProxy = "test-no-proxy"
trustedCAProxy = "test-trusted-ca-proxy"
)

const (
Expand Down Expand Up @@ -194,14 +199,49 @@ var _ = Describe("All DataImportCron Tests", func() {
})

It("Should create and delete CronJob if DataImportCron is created and deleted", func() {
cdiConfig := MakeEmptyCDIConfigSpec(common.ConfigName)
cdiConfig.Status.ImportProxy = &cdiv1.ImportProxy{
HTTPProxy: &httpProxy,
HTTPSProxy: &httpsProxy,
NoProxy: &noProxy,
TrustedCAProxy: &trustedCAProxy,
}

cron = newDataImportCron(cronName)
reconciler = createDataImportCronReconciler(cron)
reconciler = createDataImportCronReconcilerWithoutConfig(cron, cdiConfig)
_, err := reconciler.Reconcile(context.TODO(), cronReq)
Expect(err).ToNot(HaveOccurred())

cronjob := &batchv1.CronJob{}
err = reconciler.client.Get(context.TODO(), cronJobKey(cron), cronjob)
Expect(err).ToNot(HaveOccurred())
containers := cronjob.Spec.JobTemplate.Spec.Template.Spec.Containers
Expect(containers).To(HaveLen(1))

env := containers[0].Env
Expect(getEnvVar(env, common.ImportProxyHTTP)).To(Equal(httpProxy))
Expect(getEnvVar(env, common.ImportProxyHTTPS)).To(Equal(httpsProxy))
Expect(getEnvVar(env, common.ImportProxyNoProxy)).To(Equal(noProxy))

volMounts := containers[0].VolumeMounts
Expect(volMounts).To(HaveLen(1))
Expect(volMounts[0]).To(Equal(corev1.VolumeMount{
Name: ProxyCertVolName,
MountPath: common.ImporterProxyCertDir,
}))

volumes := cronjob.Spec.JobTemplate.Spec.Template.Spec.Volumes
Expect(volumes).To(HaveLen(1))
Expect(volumes[0]).To(Equal(corev1.Volume{
Name: ProxyCertVolName,
VolumeSource: corev1.VolumeSource{
ConfigMap: &corev1.ConfigMapVolumeSource{
LocalObjectReference: corev1.LocalObjectReference{
Name: trustedCAProxy,
},
},
},
}))

err = reconciler.client.Get(context.TODO(), cronKey, cron)
Expect(err).ToNot(HaveOccurred())
Expand All @@ -218,6 +258,32 @@ var _ = Describe("All DataImportCron Tests", func() {
Expect(err).To(HaveOccurred())
})

It("Should verify CronJob container env variables are empty and no extra volume is set when proxy is not configured", func() {
cron = newDataImportCron(cronName)
reconciler = createDataImportCronReconciler(cron)
_, err := reconciler.Reconcile(context.TODO(), cronReq)
Expect(err).ToNot(HaveOccurred())

cronjob := &batchv1.CronJob{}
err = reconciler.client.Get(context.TODO(), cronJobKey(cron), cronjob)
Expect(err).ToNot(HaveOccurred())

Expect(cronjob.Spec.SuccessfulJobsHistoryLimit).To(Equal(pointer.Int32(1)))
Expect(cronjob.Spec.FailedJobsHistoryLimit).To(BeNil())

jobTemplateSpec := cronjob.Spec.JobTemplate.Spec.Template.Spec
containers := jobTemplateSpec.Containers
Expect(containers).To(HaveLen(1))

env := containers[0].Env
Expect(getEnvVar(env, common.ImportProxyHTTP)).To(BeEmpty())
Expect(getEnvVar(env, common.ImportProxyHTTPS)).To(BeEmpty())
Expect(getEnvVar(env, common.ImportProxyNoProxy)).To(BeEmpty())

Expect(containers[0].VolumeMounts).To(HaveLen(0))
Expect(jobTemplateSpec.Volumes).To(HaveLen(0))
})

It("Should update CronJob on reconcile", func() {
cron = newDataImportCron(cronName)
reconciler = createDataImportCronReconciler(cron)
Expand Down Expand Up @@ -562,8 +628,14 @@ var _ = Describe("untagURL", func() {

func createDataImportCronReconciler(objects ...runtime.Object) *DataImportCronReconciler {
cdiConfig := MakeEmptyCDIConfigSpec(common.ConfigName)
objs := []runtime.Object{cdiConfig}
objs = append(objs, objects...)
return createDataImportCronReconcilerWithoutConfig(objs...)
}

func createDataImportCronReconcilerWithoutConfig(objects ...runtime.Object) *DataImportCronReconciler {
crd := &extv1.CustomResourceDefinition{ObjectMeta: metav1.ObjectMeta{Name: "dataimportcrons.cdi.kubevirt.io"}}
objs := []runtime.Object{cdiConfig, crd}
objs := []runtime.Object{crd}
objs = append(objs, objects...)

s := scheme.Scheme
Expand Down Expand Up @@ -662,3 +734,12 @@ func verifyConditionState(condType string, condState cdiv1.ConditionState, desir
Expect(condState.Status).To(Equal(desiredStatus))
Expect(condState.Reason).To(Equal(desiredReason))
}

func getEnvVar(env []corev1.EnvVar, name string) string {
for _, envVar := range env {
if envVar.Name == name {
return envVar.Value
}
}
return ""
}
18 changes: 3 additions & 15 deletions pkg/controller/import-controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1171,20 +1171,8 @@ func makeImporterPodSpec(args *importerPodArgs) *corev1.Pod {
Name: CertVolName,
MountPath: common.ImporterCertDir,
}

vol := corev1.Volume{
Name: CertVolName,
VolumeSource: corev1.VolumeSource{
ConfigMap: &corev1.ConfigMapVolumeSource{
LocalObjectReference: corev1.LocalObjectReference{
Name: args.podEnvVar.certConfigMap,
},
},
},
}

pod.Spec.Containers[0].VolumeMounts = append(pod.Spec.Containers[0].VolumeMounts, vm)
pod.Spec.Volumes = append(pod.Spec.Volumes, vol)
pod.Spec.Volumes = append(pod.Spec.Volumes, createConfigMapVolume(CertVolName, args.podEnvVar.certConfigMap))
}

if args.podEnvVar.certConfigMapProxy != "" {
Expand All @@ -1193,7 +1181,7 @@ func makeImporterPodSpec(args *importerPodArgs) *corev1.Pod {
MountPath: common.ImporterProxyCertDir,
}
pod.Spec.Containers[0].VolumeMounts = append(pod.Spec.Containers[0].VolumeMounts, vm)
pod.Spec.Volumes = append(pod.Spec.Volumes, createProxyConfigMapVolume(ProxyCertVolName, args.podEnvVar.certConfigMapProxy))
pod.Spec.Volumes = append(pod.Spec.Volumes, createConfigMapVolume(ProxyCertVolName, args.podEnvVar.certConfigMapProxy))
}

for index, header := range args.podEnvVar.secretExtraHeaders {
Expand Down Expand Up @@ -1259,7 +1247,7 @@ func makeImporterContainerSpec(image, verbose, pullPolicy string) *corev1.Contai
}
}

func createProxyConfigMapVolume(certVolName, objRef string) corev1.Volume {
func createConfigMapVolume(certVolName, objRef string) corev1.Volume {
return corev1.Volume{
Name: certVolName,
VolumeSource: corev1.VolumeSource{
Expand Down
51 changes: 24 additions & 27 deletions pkg/importer/registry-datasource.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@ limitations under the License.
package importer

import (
"fmt"
"io/ioutil"
"net/url"
"os"
Expand Down Expand Up @@ -54,7 +53,7 @@ type RegistryDataSource struct {

// NewRegistryDataSource creates a new instance of the Registry Data Source.
func NewRegistryDataSource(endpoint, accessKey, secKey, certDir string, insecureTLS bool) *RegistryDataSource {
allCertDir, err := createCertificateDir(certDir)
allCertDir, err := CreateCertificateDir(certDir)
if err != nil {
klog.Infof("Error creating allCertDir %v", err)
if allCertDir != "/" {
Expand Down Expand Up @@ -164,42 +163,40 @@ func getImageFileName(dir string) (string, error) {
return filename, nil
}

func createCertificateDir(registryCertDir string) (string, error) {
// CreateCertificateDir creates a common certificate dir
func CreateCertificateDir(registryCertDir string) (string, error) {
allCerts := "/tmp/all_certs"
err := os.MkdirAll(allCerts, 0700)
if err != nil {
if err := os.MkdirAll(allCerts, 0700); err != nil {
return allCerts, err
}

klog.Info("Copying proxy certs")
directory, err := os.Open(common.ImporterProxyCertDir)
if err := collectCerts(common.ImporterProxyCertDir, allCerts, "proxy-"); err != nil {
return allCerts, err
}
klog.Info("Copying registry certs")
if err := collectCerts(registryCertDir, allCerts, ""); err != nil {
return allCerts, err
}
return allCerts, nil
}

func collectCerts(certDir, targetDir, targetPrefix string) error {
directory, err := os.Open(certDir)
if err != nil {
return "", err
return err
}
objects, err := directory.Readdir(-1)
if err != nil {
return "", err
return err
}
for _, obj := range objects {
if strings.HasSuffix(obj.Name(), ".crt") {
err = util.LinkFile(filepath.Join(common.ImporterProxyCertDir, obj.Name()), filepath.Join(allCerts, fmt.Sprintf("proxy-%s", obj.Name())))
if err != nil {
return allCerts, err
}
if !strings.HasSuffix(obj.Name(), ".crt") {
continue
}
}
klog.Info("Copying registry certs")
directory, _ = os.Open(registryCertDir)
objects, err = directory.Readdir(-1)
if err != nil {
return "", err
}
for _, obj := range objects {
if strings.HasSuffix(obj.Name(), ".crt") {
err = util.LinkFile(filepath.Join(registryCertDir, obj.Name()), filepath.Join(allCerts, obj.Name()))
if err != nil {
return allCerts, err
}
if err := util.LinkFile(filepath.Join(certDir, obj.Name()), filepath.Join(targetDir, targetPrefix+obj.Name())); err != nil {
return err
}
}
return allCerts, nil
return nil
}
Loading

0 comments on commit eef0358

Please sign in to comment.