operator: Make configurator image configurable

Users would like to use our operator in closed environments where
public docker hub is not available. This change give a way to change
image, tag and imagePullPolicy via flags.

src/go/k8s/controllers/redpanda/cluster_controller.go

@@ -37,14 +37,15 @@ import (
 var (
 	errNonexistentLastObservesState = errors.New("expecting to have statefulset LastObservedState set but it's nil")
 	errNodePortMissing              = errors.New("the node port is missing from the service")
+	errInvalidImagePullPolicy       = errors.New("invalid image pull policy")
 )
 
 // ClusterReconciler reconciles a Cluster object
 type ClusterReconciler struct {
 	client.Client
-	Log             logr.Logger
-	configuratorTag string
-	Scheme          *runtime.Scheme
+	Log                  logr.Logger
+	configuratorSettings resources.ConfiguratorSettings
+	Scheme               *runtime.Scheme
 }
 
 //+kubebuilder:rbac:groups=redpanda.vectorized.io,resources=clusters,verbs=get;list;watch;create;update;patch;delete
@@ -160,7 +161,7 @@ func (r *ClusterReconciler) Reconcile(
 		pki.AdminAPIClientCert(),
 		pki.PandaproxyAPINodeCert(),
 		sa.Key().Name,
-		r.configuratorTag,
+		r.configuratorSettings,
 		log)
 	toApply := []resources.Reconciler{
 		headlessSvc,
@@ -231,13 +232,28 @@ func (r *ClusterReconciler) Reconcile(
 
 // SetupWithManager sets up the controller with the Manager.
 func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	if err := validateImagePullPolicy(r.configuratorSettings.ImagePullPolicy); err != nil {
+		return fmt.Errorf("invalid image pull policy \"%s\": %w", r.configuratorSettings.ImagePullPolicy, err)
+	}
+
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&redpandav1alpha1.Cluster{}).
 		Owns(&appsv1.StatefulSet{}).
 		Owns(&corev1.Service{}).
 		Complete(r)
 }
 
+func validateImagePullPolicy(imagePullPolicy corev1.PullPolicy) error {
+	switch imagePullPolicy {
+	case corev1.PullAlways:
+	case corev1.PullIfNotPresent:
+	case corev1.PullNever:
+	default:
+		return fmt.Errorf("available image pull policy: \"%s\", \"%s\" or \"%s\": %w", corev1.PullAlways, corev1.PullIfNotPresent, corev1.PullNever, errInvalidImagePullPolicy)
+	}
+	return nil
+}
+
 func (r *ClusterReconciler) reportStatus(
 	ctx context.Context,
 	redpandaCluster *redpandav1alpha1.Cluster,
@@ -310,11 +326,11 @@ func statusShouldBeUpdated(
 		status.Replicas != readyReplicas
 }
 
-// WithConfiguratorTag set the configuratorTag
-func (r *ClusterReconciler) WithConfiguratorTag(
-	configuratorTag string,
+// WithConfiguratorSettings set the configuratorTag
+func (r *ClusterReconciler) WithConfiguratorSettings(
+	configuratorSettings resources.ConfiguratorSettings,
 ) *ClusterReconciler {
-	r.configuratorTag = configuratorTag
+	r.configuratorSettings = configuratorSettings
 	return r
 }
 

src/go/k8s/controllers/redpanda/cluster_controller_test.go

@@ -14,7 +14,9 @@ import (
 	"time"
 
 	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/ginkgo/extensions/table"
 	. "github.com/onsi/gomega"
+	types2 "github.com/onsi/gomega/types"
 	"github.com/vectorizedio/redpanda/src/go/k8s/apis/redpanda/v1alpha1"
 	"github.com/vectorizedio/redpanda/src/go/k8s/controllers/redpanda"
 	res "github.com/vectorizedio/redpanda/src/go/k8s/pkg/resources"
@@ -321,6 +323,30 @@ var _ = Describe("RedPandaCluster controller", func() {
 
 		})
 	})
+
+	DescribeTable("Image pull policy tests table", func(imagePullPolicy string, matcher types2.GomegaMatcher) {
+		k8sManager, err := ctrl.NewManager(cfg, ctrl.Options{
+			Scheme:             scheme.Scheme,
+			MetricsBindAddress: "0",
+		})
+		Expect(err).NotTo(HaveOccurred())
+
+		r := &redpanda.ClusterReconciler{
+			Client: fake.NewClientBuilder().Build(),
+			Log:    ctrl.Log,
+			Scheme: scheme.Scheme,
+		}
+
+		Expect(r.WithConfiguratorSettings(res.ConfiguratorSettings{
+			ImagePullPolicy: corev1.PullPolicy(imagePullPolicy),
+		}).SetupWithManager(k8sManager)).To(matcher)
+
+	},
+		Entry("Always image pull policy", "Always", Succeed()),
+		Entry("IfNotPresent image pull policy", "IfNotPresent", Succeed()),
+		Entry("Never image pull policy", "Never", Succeed()),
+		Entry("Empty image pull policy", "", Not(Succeed())),
+		Entry("Random image pull policy", "asdvasd", Not(Succeed())))
 })
 
 func findPort(ports []corev1.ServicePort, name string) int32 {

src/go/k8s/controllers/redpanda/suite_test.go

@@ -20,7 +20,9 @@ import (
 	"github.com/onsi/gomega/gexec"
 	redpandav1alpha1 "github.com/vectorizedio/redpanda/src/go/k8s/apis/redpanda/v1alpha1"
 	redpandacontrollers "github.com/vectorizedio/redpanda/src/go/k8s/controllers/redpanda"
+	"github.com/vectorizedio/redpanda/src/go/k8s/pkg/resources"
 	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
@@ -34,6 +36,7 @@ import (
 
 var k8sClient client.Client
 var testEnv *envtest.Environment
+var cfg *rest.Config
 
 func TestAPIs(t *testing.T) {
 	RegisterFailHandler(Fail)
@@ -51,7 +54,8 @@ var _ = BeforeSuite(func(done Done) {
 		CRDDirectoryPaths: []string{filepath.Join("..", "..", "config", "crd", "bases")},
 	}
 
-	cfg, err := testEnv.Start()
+	var err error
+	cfg, err = testEnv.Start()
 	Expect(err).NotTo(HaveOccurred())
 	Expect(cfg).NotTo(BeNil())
 
@@ -73,6 +77,10 @@ var _ = BeforeSuite(func(done Done) {
 		Client: k8sManager.GetClient(),
 		Log:    ctrl.Log.WithName("controllers").WithName("core").WithName("RedpandaCluster"),
 		Scheme: k8sManager.GetScheme(),
+	}).WithConfiguratorSettings(resources.ConfiguratorSettings{
+		ConfiguratorBaseImage: "vectorized/configurator",
+		ConfiguratorTag:       "latest",
+		ImagePullPolicy:       "Always",
 	}).SetupWithManager(k8sManager)
 	Expect(err).ToNot(HaveOccurred())
 

src/go/k8s/main.go

@@ -15,6 +15,8 @@ import (
 	cmapiv1 "github.com/jetstack/cert-manager/pkg/apis/certmanager/v1"
 	redpandav1alpha1 "github.com/vectorizedio/redpanda/src/go/k8s/apis/redpanda/v1alpha1"
 	redpandacontrollers "github.com/vectorizedio/redpanda/src/go/k8s/controllers/redpanda"
+	"github.com/vectorizedio/redpanda/src/go/k8s/pkg/resources"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
@@ -24,6 +26,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 )
 
+const (
+	defaultConfiguratorContainerImage = "vectorized/configurator"
+)
+
 var (
 	scheme   = runtime.NewScheme()
 	setupLog = ctrl.Log.WithName("setup")
@@ -39,11 +45,13 @@ func init() {
 
 func main() {
 	var (
-		metricsAddr          string
-		enableLeaderElection bool
-		probeAddr            string
-		webhookEnabled       bool
-		configuratorTag      string
+		metricsAddr                 string
+		enableLeaderElection        bool
+		probeAddr                   string
+		webhookEnabled              bool
+		configuratorBaseImage       string
+		configuratorTag             string
+		configuratorImagePullPolicy string
 	)
 
 	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
@@ -52,7 +60,9 @@ func main() {
 		"Enable leader election for controller manager. "+
 			"Enabling this will ensure there is only one active controller manager.")
 	flag.BoolVar(&webhookEnabled, "webhook-enabled", false, "Enable webhook Manager")
+	flag.StringVar(&configuratorBaseImage, "configurator-base-image", defaultConfiguratorContainerImage, "Set the configurator base image")
 	flag.StringVar(&configuratorTag, "configurator-tag", "latest", "Set the configurator tag")
+	flag.StringVar(&configuratorImagePullPolicy, "configurator-image-pull-policy", "Always", "Set the configurator image pull policy")
 
 	opts := zap.Options{
 		Development: true,
@@ -77,11 +87,17 @@ func main() {
 		os.Exit(1)
 	}
 
+	configurator := resources.ConfiguratorSettings{
+		ConfiguratorBaseImage: configuratorBaseImage,
+		ConfiguratorTag:       configuratorTag,
+		ImagePullPolicy:       corev1.PullPolicy(configuratorImagePullPolicy),
+	}
+
 	if err = (&redpandacontrollers.ClusterReconciler{
 		Client: mgr.GetClient(),
 		Log:    ctrl.Log.WithName("controllers").WithName("redpanda").WithName("Cluster"),
 		Scheme: mgr.GetScheme(),
-	}).WithConfiguratorTag(configuratorTag).SetupWithManager(mgr); err != nil {
+	}).WithConfiguratorSettings(configurator).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "Unable to create controller", "controller", "Cluster")
 		os.Exit(1)
 	}

src/go/k8s/pkg/resources/resource_integration_test.go

@@ -78,7 +78,11 @@ func TestEnsure_StatefulSet(t *testing.T) {
 		types.NamespacedName{},
 		types.NamespacedName{},
 		"",
-		"latest",
+		res.ConfiguratorSettings{
+			ConfiguratorBaseImage: "vectorized/configurator",
+			ConfiguratorTag:       "latest",
+			ImagePullPolicy:       "Always",
+		},
 		ctrl.Log.WithName("test"))
 
 	err := sts.Ensure(context.Background())

src/go/k8s/pkg/resources/statefulset.go

@@ -36,9 +36,8 @@ var _ Resource = &StatefulSetResource{}
 var errNodePortMissing = errors.New("the node port is missing from the service")
 
 const (
-	redpandaContainerName      = "redpanda"
-	configuratorContainerName  = "redpanda-configurator"
-	configuratorContainerImage = "vectorized/configurator"
+	redpandaContainerName     = "redpanda"
+	configuratorContainerName = "redpanda-configurator"
 
 	userID  = 101
 	groupID = 101
@@ -52,6 +51,14 @@ const (
 	defaultDatadirCapacity = "100Gi"
 )
 
+// ConfiguratorSettings holds settings related to configurator container and deployment
+// strategy
+type ConfiguratorSettings struct {
+	ConfiguratorBaseImage string
+	ConfiguratorTag       string
+	ImagePullPolicy       corev1.PullPolicy
+}
+
 // StatefulSetResource is part of the reconciliation of redpanda.vectorized.io CRD
 // focusing on the management of redpanda cluster
 type StatefulSetResource struct {
@@ -69,7 +76,7 @@ type StatefulSetResource struct {
 	adminAPIClientCertSecretKey    types.NamespacedName // TODO this is unused, can be removed
 	pandaproxyAPINodeCertSecretKey types.NamespacedName
 	serviceAccountName             string
-	configuratorTag                string
+	configuratorSettings           ConfiguratorSettings
 	logger                         logr.Logger
 
 	LastObservedState *appsv1.StatefulSet
@@ -90,7 +97,7 @@ func NewStatefulSet(
 	adminAPIClientCertSecretKey types.NamespacedName,
 	pandaproxyAPINodeCertSecretKey types.NamespacedName,
 	serviceAccountName string,
-	configuratorTag string,
+	configuratorSettings ConfiguratorSettings,
 	logger logr.Logger,
 ) *StatefulSetResource {
 	return &StatefulSetResource{
@@ -108,7 +115,7 @@ func NewStatefulSet(
 		adminAPIClientCertSecretKey,
 		pandaproxyAPINodeCertSecretKey,
 		serviceAccountName,
-		configuratorTag,
+		configuratorSettings,
 		logger.WithValues("Kind", statefulSetKind()),
 		nil,
 	}
@@ -288,7 +295,7 @@ func (r *StatefulSetResource) obj() (k8sclient.Object, error) {
 						{
 							Name:            configuratorContainerName,
 							Image:           r.fullConfiguratorImage(),
-							ImagePullPolicy: corev1.PullIfNotPresent,
+							ImagePullPolicy: r.configuratorSettings.ImagePullPolicy,
 							Env: append([]corev1.EnvVar{
 								{
 									Name:  "SERVICE_FQDN",
@@ -728,5 +735,5 @@ func statefulSetKind() string {
 }
 
 func (r *StatefulSetResource) fullConfiguratorImage() string {
-	return fmt.Sprintf("%s:%s", configuratorContainerImage, r.configuratorTag)
+	return fmt.Sprintf("%s:%s", r.configuratorSettings.ConfiguratorBaseImage, r.configuratorSettings.ConfiguratorTag)
 }

src/go/k8s/pkg/resources/statefulset_test.go

@@ -88,7 +88,11 @@ func TestEnsure(t *testing.T) {
 			types.NamespacedName{},
 			types.NamespacedName{},
 			"",
-			"latest",
+			res.ConfiguratorSettings{
+				ConfiguratorBaseImage: "vectorized/configurator",
+				ConfiguratorTag:       "latest",
+				ImagePullPolicy:       "Always",
+			},
 			ctrl.Log.WithName("test"))
 
 		err = sts.Ensure(context.Background())