[defaults] Take ownership of default sources

- Add a command line parameter (-defaultsDir) to pass the folder where
defaults are stored
- Update the operator deployment to include the new parameter
- Move the default OperatorSources from the manifests folder to the
default folder
- Update the Dockerfile.rhel to include the default folder
- Add code to startup that ensures the defaults OperatorSources are
present with default spec after initialization
- Add code to the OperatorSource's deleted and succeeded reconciler to
ensure the default OperatorSources are present and match the default
spec

Author: aravindhp

Dockerfile

@@ -9,6 +9,7 @@ RUN useradd marketplace-operator
 USER marketplace-operator
 COPY --from=builder /go/src/github.com/operator-framework/operator-marketplace/build/_output/bin/marketplace-operator /usr/bin
 ADD manifests /manifests
+ADD defaults /defaults
 
 LABEL io.k8s.display-name="OpenShift Marketplace Operator" \
       io.k8s.description="This is a component of OpenShift Container Platform and manages the OpenShift Marketplace." \

Dockerfile.rhel7

@@ -9,6 +9,7 @@ RUN useradd marketplace-operator
 USER marketplace-operator
 COPY --from=builder /go/src/github.com/operator-framework/operator-marketplace/build/_output/bin/marketplace-operator /usr/bin
 ADD manifests /manifests
+ADD defaults /defaults
 
 LABEL io.k8s.display-name="OpenShift Marketplace Operator" \
       io.k8s.description="This is a component of OpenShift Container Platform and manages the OpenShift Marketplace." \
@@ -17,4 +18,4 @@ LABEL io.k8s.display-name="OpenShift Marketplace Operator" \
       maintainer="AOS Marketplace <aos-marketplace@redhat.com>"
 
 # entrypoint specified in operator.yaml as `marketplace-operator`
-CMD ["/usr/bin/marketplace-operator"]
+CMD ["/usr/bin/marketplace-operator"]

cmd/manager/main.go

@@ -14,6 +14,7 @@ import (
 	"github.com/operator-framework/operator-marketplace/pkg/apis"
 	"github.com/operator-framework/operator-marketplace/pkg/catalogsourceconfig"
 	"github.com/operator-framework/operator-marketplace/pkg/controller"
+	"github.com/operator-framework/operator-marketplace/pkg/defaults"
 	"github.com/operator-framework/operator-marketplace/pkg/operatorsource"
 	"github.com/operator-framework/operator-marketplace/pkg/proxy"
 	"github.com/operator-framework/operator-marketplace/pkg/registry"
@@ -22,8 +23,11 @@ import (
 	"github.com/operator-framework/operator-sdk/pkg/leader"
 	sdkVersion "github.com/operator-framework/operator-sdk/version"
 	log "github.com/sirupsen/logrus"
+	kruntime "k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
@@ -46,9 +50,11 @@ func printVersion() {
 func main() {
 	printVersion()
 
-	// Parse the command line arguments for the registry server image
+	// Parse the command line arguments
 	flag.StringVar(&registry.ServerImage, "registryServerImage",
 		registry.DefaultServerImage, "the image to use for creating the operator registry pod")
+	flag.StringVar(&defaults.Dir, "defaultsDir",
+		"", "the directory where the default OperatorSources are stored")
 	flag.Parse()
 
 	namespace, err := k8sutil.GetWatchNamespace()
@@ -127,6 +133,19 @@ func main() {
 	log.Info("Elected leader.")
 
 	log.Print("Starting the Cmd.")
+
+	// Populate the default OperatorSources tracker
+	err = defaults.PopulateTracker()
+	if err != nil {
+		exit(err)
+	}
+
+	// Handle the defaults
+	err = ensureDefaults(cfg, mgr.GetScheme())
+	if err != nil {
+		exit(err)
+	}
+
 	stopCh := signals.SetupSignalHandler()
 
 	// statusReportingDoneCh will be closed after the operator has successfully stopped reporting ClusterOperator status.
@@ -155,3 +174,23 @@ func exit(err error) {
 	log.Info("The operator exited gracefully, exit code 0")
 	os.Exit(0)
 }
+
+// ensureDefaults ensures that all the default OperatorSources are present on
+// the cluster
+func ensureDefaults(cfg *rest.Config, scheme *kruntime.Scheme) error {
+	// The default client serves read requests from the cache which only gets
+	// initialized after mgr.Start(). So we need to instantiate a new client
+	// for the defaults handler.
+	clientForDefaults, err := client.New(cfg, client.Options{Scheme: scheme})
+	if err != nil {
+		log.Errorf("Error initializing client for handling defaults - %v", err)
+		return err
+	}
+
+	err = defaults.New().EnsureAll(clientForDefaults)
+	if err != nil {
+		log.Errorf("[defaults] Error ensuring default OperatorSource(s) - %v", err)
+	}
+
+	return err
+}

…nifests/07_redhat_operatorsource.cr.yaml defaults/01_redhat_operators.cr.yamlmanifests/07_redhat_operatorsource.cr.yaml renamed to defaults/01_redhat_operators.cr.yaml manifests/07_redhat_operatorsource.cr.yaml renamed to defaults/01_redhat_operators.cr.yaml

@@ -41,6 +41,7 @@ spec:
           - marketplace-operator
           args:
           - -registryServerImage=quay.io/openshift/origin-operator-registry
+          - -defaultsDir=/defaults
           imagePullPolicy: Always
           livenessProbe:
             httpGet:

…ests/08_certified_operatorsource.cr.yaml defaults/02_certified_operators.yamlmanifests/08_certified_operatorsource.cr.yaml renamed to defaults/02_certified_operators.yaml manifests/08_certified_operatorsource.cr.yaml renamed to defaults/02_certified_operators.yaml

@@ -41,6 +41,7 @@ spec:
           - marketplace-operator
           args:
           - -registryServerImage=quay.io/openshift/origin-operator-registry
+          - -defaultsDir=/defaults
           imagePullPolicy: IfNotPresent
           livenessProbe:
             httpGet:

…ests/09_community_operatorsource.cr.yaml defaults/03_community_operators.yamlmanifests/09_community_operatorsource.cr.yaml renamed to defaults/03_community_operators.yaml manifests/09_community_operatorsource.cr.yaml renamed to defaults/03_community_operators.yaml

@@ -0,0 +1,157 @@
+package defaults
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+
+	"github.com/operator-framework/operator-marketplace/pkg/apis/operators/v1"
+	wrapper "github.com/operator-framework/operator-marketplace/pkg/client"
+	"github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/api/errors"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/apimachinery/pkg/util/yaml"
+)
+
+var (
+	// Dir is the directory where the default OperatorSource definitions are
+	// placed on disk. It will be empty if defaults are not required.
+	Dir string
+
+	// defaultsTracker is used to keep an in-memory record of default
+	// OperatorSources. It is a map of OperatorSource name to the OperatorSource
+	// definition in the defaults directory.
+	defaultsTracker = make(map[string]v1.OperatorSource)
+)
+
+// Defaults is the interface that can be used to ensure default OperatorSources
+// are always present on the cluster.
+type Defaults interface {
+	EnsureAll(client wrapper.Client) error
+	Ensure(client wrapper.Client, opsrcName string) error
+}
+
+type defaults struct {
+}
+
+// New returns a the singleton defaults
+func New() Defaults {
+	return &defaults{}
+}
+
+// Ensure checks if the given OperatorSource source is one of the
+// defaults and if it is, it ensures it is present on the cluster.
+func (d *defaults) Ensure(client wrapper.Client, opsrcName string) error {
+	opsrc, present := defaultsTracker[opsrcName]
+	if !present {
+		return nil
+	}
+
+	err := processOpSrc(client, opsrc)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// EnsureAll processes all the default OperatorSources and ensures they are
+// present on the cluster.
+func (d *defaults) EnsureAll(client wrapper.Client) error {
+	allErrors := []error{}
+	for name := range defaultsTracker {
+		err := d.Ensure(client, name)
+		if err != nil {
+			allErrors = append(allErrors, fmt.Errorf("Error handling %s - %v", name, err))
+		}
+	}
+	return utilerrors.NewAggregate(allErrors)
+}
+
+// PopulateTracker populates the defaultsTracker on initialization
+func PopulateTracker() error {
+	// Default directory has not been specified
+	if Dir == "" {
+		return nil
+	}
+
+	_, err := os.Stat(Dir)
+	if err != nil {
+		return err
+	}
+
+	fileInfos, err := ioutil.ReadDir(Dir)
+	if err != nil {
+		return err
+	}
+
+	for _, fileInfo := range fileInfos {
+		fileName := fileInfo.Name()
+		opsrc, err := getOpSrcDefinition(fileName)
+		if err != nil {
+			// Reinitialize the tracker as we hard error on even one failure
+			defaultsTracker = make(map[string]v1.OperatorSource)
+			return err
+		}
+		defaultsTracker[opsrc.Name] = *opsrc
+	}
+	return nil
+}
+
+// getOpSrcDefinition returns an OperatorSource definition from the given file
+// in the defaults directory. It only supports decoding OperatorSources. Any
+// other resource type will result in an error.
+func getOpSrcDefinition(fileName string) (*v1.OperatorSource, error) {
+	file, err := os.Open(filepath.Join(Dir, fileName))
+	if err != nil {
+		return nil, err
+	}
+
+	opsrc := &v1.OperatorSource{}
+	decoder := yaml.NewYAMLOrJSONDecoder(file, 1024)
+	err = decoder.Decode(opsrc)
+	if err != nil {
+		return nil, err
+	}
+	return opsrc, nil
+}
+
+// processOpSrc will ensure that the given OperatorSource exists on the cluster.
+func processOpSrc(client wrapper.Client, defaultOpsrc v1.OperatorSource) error {
+	// Get OperatorSource on the cluster
+	opsrcCluster := &v1.OperatorSource{}
+	err := client.Get(context.TODO(), wrapper.ObjectKey{
+		Name:      defaultOpsrc.Name,
+		Namespace: defaultOpsrc.Namespace},
+		opsrcCluster)
+	if err != nil && !errors.IsNotFound(err) {
+		return err
+	}
+
+	// Create if not present or is deleted
+	if errors.IsNotFound(err) || !opsrcCluster.ObjectMeta.DeletionTimestamp.IsZero() {
+		err = client.Create(context.TODO(), &defaultOpsrc)
+		if err != nil {
+			return err
+		}
+		logrus.Infof("[defaults] Creating OperatorSource %s", defaultOpsrc.Name)
+		return nil
+	}
+
+	if defaultOpsrc.Spec.IsEqual(&opsrcCluster.Spec) {
+		logrus.Infof("[defaults] OperatorSource %s default and on cluster specs are same", defaultOpsrc.Name)
+		return nil
+	}
+
+	// Update if the spec has changed
+	opsrcCluster.Spec = defaultOpsrc.Spec
+	err = client.Update(context.TODO(), opsrcCluster)
+	if err != nil {
+		return err
+	}
+	logrus.Infof("[defaults] Restoring OperatorSource %s", defaultOpsrc.Name)
+
+	return nil
+}