Add appregistry-server binary

Add a new binary 'appregistry-server' that can be used to download
operator manifest(s) from a remote app registry and generate a
sqlite database to be served using grpc.

Author: tkashem

cmd/appregistry-server/main.go

@@ -0,0 +1,108 @@
+package main
+
+import (
+	"net"
+
+	"github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/reflection"
+
+	"github.com/operator-framework/operator-registry/pkg/api"
+	health "github.com/operator-framework/operator-registry/pkg/api/grpc_health_v1"
+	"github.com/operator-framework/operator-registry/pkg/appregistry"
+	"github.com/operator-framework/operator-registry/pkg/lib/log"
+	"github.com/operator-framework/operator-registry/pkg/server"
+)
+
+func main() {
+	var rootCmd = &cobra.Command{
+		Short: "appregistry-server",
+		Long:  `appregistry-server downloads operator manifest(s) from remote appregistry, builds a sqlite database containing these downloaded manifest(s) and serves a grpc API to query it`,
+
+		PreRunE: func(cmd *cobra.Command, args []string) error {
+			if debug, _ := cmd.Flags().GetBool("debug"); debug {
+				logrus.SetLevel(logrus.DebugLevel)
+			}
+			return nil
+		},
+
+		RunE: runCmdFunc,
+	}
+
+	rootCmd.Flags().Bool("debug", false, "enable debug logging")
+	rootCmd.Flags().StringP("kubeconfig", "k", "", "absolute path to kubeconfig file")
+	rootCmd.Flags().StringP("database", "d", "bundles.db", "name of db to output")
+	rootCmd.Flags().StringP("sources", "s", "", "comma separated list of OperatorSource object(s) {namespace}/{name}")
+	rootCmd.Flags().StringP("packages", "o", "", "comma separated list of package(s) to be downloaded from the specified operator source(s)")
+	rootCmd.Flags().StringP("port", "p", "50051", "port number to serve on")
+	rootCmd.Flags().StringP("termination-log", "t", "/dev/termination-log", "path to a container termination log file")
+	if err := rootCmd.Flags().MarkHidden("debug"); err != nil {
+		logrus.Panic(err.Error())
+	}
+
+	if err := rootCmd.Execute(); err != nil {
+		logrus.Panic(err.Error())
+	}
+}
+
+func runCmdFunc(cmd *cobra.Command, args []string) error {
+	// Immediately set up termination log
+	terminationLogPath, err := cmd.Flags().GetString("termination-log")
+	if err != nil {
+		return err
+	}
+	err = log.AddDefaultWriterHooks(terminationLogPath)
+	if err != nil {
+		return err
+	}
+	kubeconfig, err := cmd.Flags().GetString("kubeconfig")
+	if err != nil {
+		return err
+	}
+	port, err := cmd.Flags().GetString("port")
+	if err != nil {
+		return err
+	}
+	sources, err := cmd.Flags().GetString("sources")
+	if err != nil {
+		return err
+	}
+	packages, err := cmd.Flags().GetString("packages")
+	if err != nil {
+		return err
+	}
+	dbName, err := cmd.Flags().GetString("database")
+	if err != nil {
+		return err
+	}
+
+	logger := logrus.WithFields(logrus.Fields{"type": "appregistry", "port": port})
+
+	loader, err := appregistry.NewLoader(kubeconfig, logger)
+	if err != nil {
+		logger.Fatalf("error initializing - %v", err)
+	}
+
+	store, err := loader.Load(dbName, sources, packages)
+	if err != nil {
+		logger.Fatalf("error loading manifest from remote registry - %v", err)
+	}
+
+	lis, err := net.Listen("tcp", ":"+port)
+	if err != nil {
+		logger.Fatalf("failed to listen: %v", err)
+	}
+	s := grpc.NewServer()
+
+	api.RegisterRegistryServer(s, server.NewRegistryServer(store))
+	health.RegisterHealthServer(s, server.NewHealthServer())
+	reflection.Register(s)
+
+	logger.Info("serving registry")
+	if err := s.Serve(lis); err != nil {
+		logger.Fatalf("failed to serve: %v", err)
+	}
+
+	return nil
+}

go.mod

@@ -28,6 +28,7 @@ require (
 	github.com/modern-go/reflect2 v1.0.1 // indirect
 	github.com/operator-framework/go-appr v0.0.0-20180917210448-f2aef88446f2
 	github.com/operator-framework/operator-lifecycle-manager v0.0.0-20190125151539-1e295784b30a
+	github.com/operator-framework/operator-marketplace v0.0.0-20190212161948-a7ca81b96ad9
 	github.com/sirupsen/logrus v1.2.0
 	github.com/soheilhy/cmux v0.1.4 // indirect
 	github.com/spf13/cobra v0.0.3
@@ -42,4 +43,5 @@ require (
 	k8s.io/apiextensions-apiserver v0.0.0-20181204003618-e419c5771cdc
 	k8s.io/apimachinery v0.0.0-20190118094746-1525e4dadd2d
 	k8s.io/client-go v8.0.0+incompatible
+	sigs.k8s.io/controller-runtime v0.1.10 // indirect
 )

go.sum

@@ -137,6 +137,10 @@ github.com/operator-framework/operator-lifecycle-manager v0.0.0-20181023032605-e
 github.com/operator-framework/operator-lifecycle-manager v0.0.0-20190105193533-81104ffdc4fb/go.mod h1:XMyE4n2opUK4N6L45YGQkXXi8F9fD7XDYFv/CsS6V5I=
 github.com/operator-framework/operator-lifecycle-manager v0.0.0-20190125151539-1e295784b30a h1:gBTaGobFQZVx+M2ItXxJp7oxoc9dltkLMrkgyDE0Qfc=
 github.com/operator-framework/operator-lifecycle-manager v0.0.0-20190125151539-1e295784b30a/go.mod h1:vq6TTFvg6ti1Bn6ACsZneZTmjTsURgDD6tQtVDbEgsU=
+github.com/operator-framework/operator-marketplace v0.0.0-20190208230340-d06f7b349013 h1:SyVL/xPGF6ALLDOQNchwagHz7QvESdgiiDyNoIRnMxI=
+github.com/operator-framework/operator-marketplace v0.0.0-20190208230340-d06f7b349013/go.mod h1:msZSL8pXwzQjB+hU+awVrZQw94IwJi3sNZVD3NoESIs=
+github.com/operator-framework/operator-marketplace v0.0.0-20190212161948-a7ca81b96ad9 h1:VjGYvB+9cqsf0vgO7npB1bwAIslvLFqqL1ydX9ogCRM=
+github.com/operator-framework/operator-marketplace v0.0.0-20190212161948-a7ca81b96ad9/go.mod h1:msZSL8pXwzQjB+hU+awVrZQw94IwJi3sNZVD3NoESIs=
 github.com/operator-framework/operator-registry v1.0.1/go.mod h1:1xEdZjjUg2hPEd52LG3YQ0jtwiwEGdm98S1TH5P4RAA=
 github.com/operator-framework/operator-registry v1.0.4/go.mod h1:hve6YwcjM2nGVlscLtNsp9sIIBkNZo6jlJgzWw7vP9s=
 github.com/pborman/uuid v1.2.0/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k=
@@ -249,3 +253,5 @@ k8s.io/kube-openapi v0.0.0-20181031203759-72693cb1fadd h1:ggv/Vfza0i5xuhUZyYyxcc
 k8s.io/kube-openapi v0.0.0-20181031203759-72693cb1fadd/go.mod h1:BXM9ceUBTj2QnfH2MK1odQs778ajze1RxcmP6S8RVVc=
 k8s.io/kubernetes v1.11.7-beta.0.0.20181219023948-b875d52ea96d/go.mod h1:ocZa8+6APFNC2tX1DZASIbocyYT5jHzqFVsY5aoB7Jk=
 k8s.io/kubernetes v1.11.8-beta.0.0.20190124204751-3a10094374f2/go.mod h1:ocZa8+6APFNC2tX1DZASIbocyYT5jHzqFVsY5aoB7Jk=
+sigs.k8s.io/controller-runtime v0.1.10 h1:amLOmcekVdnsD1uIpmgRqfTbQWJ2qxvQkcdeFhcotn4=
+sigs.k8s.io/controller-runtime v0.1.10/go.mod h1:HFAYoOh6XMV+jKF1UjFwrknPbowfyHEHHRdJMf2jMX8=

pkg/appregistry/appregistry.go

@@ -0,0 +1,102 @@
+package appregistry
+
+import (
+	"fmt"
+
+	marketplace "github.com/operator-framework/operator-marketplace/pkg/client/clientset/versioned"
+	"github.com/operator-framework/operator-registry/pkg/sqlite"
+	"github.com/sirupsen/logrus"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+func NewLoader(kubeconfig string, logger *logrus.Entry) (*AppregistryLoader, error) {
+	client, err := NewClient(kubeconfig, logger)
+	if err != nil {
+		return nil, err
+	}
+
+	return &AppregistryLoader{
+		logger: logger,
+		input:  &inputParser{},
+		downloader: &downloader{
+			logger: logger,
+			client: client,
+		},
+		merger: &merger{
+			logger: logger,
+			parser: &manifestYAMLParser{},
+		},
+		loader: &dbLoader{
+			logger: logger,
+		},
+	}, nil
+}
+
+type AppregistryLoader struct {
+	logger     *logrus.Entry
+	input      *inputParser
+	downloader *downloader
+	merger     *merger
+	loader     *dbLoader
+}
+
+func (a *AppregistryLoader) Load(dbName string, csvSources string, csvPackages string) (store *sqlite.SQLQuerier, err error) {
+	a.logger.Infof("operator source(s) specified are - %s", csvSources)
+	a.logger.Infof("package(s) specified are - %s", csvPackages)
+
+	input, err := a.input.Parse(csvSources, csvPackages)
+	if err != nil {
+		return nil, err
+	}
+
+	a.logger.Infof("input sanitized - sources: %s, packages: %s", input.Sources, input.Packages)
+
+	rawManifests, err := a.downloader.Download(input)
+	if err != nil {
+		a.logger.Errorf("The following error occurred while downloading - %v", err)
+
+		if len(rawManifests) == 0 {
+			a.logger.Info("No package manifest downloaded")
+			return nil, err
+		}
+	}
+
+	a.logger.Infof("download complete - %d repositories have been downloaded", len(rawManifests))
+
+	data, err := a.merger.Merge(rawManifests)
+	if err != nil {
+		a.logger.Errorf("The following error occurred while processing manifest - %v", err)
+
+		if data == nil {
+			a.logger.Info("No operator manifest bundled")
+			return nil, err
+		}
+	}
+
+	a.logger.Info("all manifest(s) have been merged into one")
+	a.logger.Info("loading into sqlite database")
+
+	store, err = a.loader.LoadToSQLite(dbName, data)
+	return
+}
+
+func NewClient(kubeconfig string, logger *logrus.Entry) (clientset marketplace.Interface, err error) {
+	var config *rest.Config
+
+	if kubeconfig != "" {
+		logger.Infof("Loading kube client config from path %q", kubeconfig)
+		config, err = clientcmd.BuildConfigFromFlags("", kubeconfig)
+	} else {
+		logger.Infof("Using in-cluster kube client config")
+		config, err = rest.InClusterConfig()
+	}
+
+	if err != nil {
+		err = fmt.Errorf("Cannot load config for REST client: %v", err)
+		return
+	}
+
+	clientset, err = marketplace.NewForConfig(config)
+	return
+}

pkg/appregistry/crd.go

@@ -0,0 +1,44 @@
+package appregistry
+
+import (
+	"fmt"
+
+	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
+)
+
+// CRDKey contains metadata needed to uniquely identify a CRD.
+//
+// OLM uses CRDKey to uniquely identify a CustomResourceDefinition object. We
+// are following the same pattern to be consistent.
+type CRDKey struct {
+	Kind    string `json:"kind"`
+	Name    string `json:"name"`
+	Version string `json:"version"`
+}
+
+// String returns a string representation of this CRDKey object with Kind, Name
+// and Version concatenated.
+//
+// CRDKey is used as the key to map of CustomResourceDefinition object(s). This
+// function ensures that Kind, Name and Version are taken into account
+// to compute the key associated with a CustomResourceDefinition object.
+func (k CRDKey) String() string {
+	return fmt.Sprintf("%s/%s/%s", k.Kind, k.Name, k.Version)
+}
+
+// CustomResourceDefinition is a structured representation of custom resource
+// definition(s) specified in `customResourceDefinitions` section of an
+// operator manifest.
+type CustomResourceDefinition struct {
+	v1beta1.CustomResourceDefinition `json:",inline"`
+}
+
+// Key returns an instance of CRDKey which uniquely identifies a given
+// CustomResourceDefinition object.
+func (crd *CustomResourceDefinition) Key() CRDKey {
+	return CRDKey{
+		Kind:    crd.Spec.Names.Kind,
+		Name:    crd.GetName(),
+		Version: crd.Spec.Version,
+	}
+}

pkg/appregistry/csv.go

@@ -0,0 +1,94 @@
+package appregistry
+
+import (
+	"encoding/json"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	// Name of the section under which the list of owned and required list of
+	// CRD(s) is specified inside an operator manifest.
+	customResourceDefinitions = "customresourcedefinitions"
+
+	// The yaml attribute that points to the name of an older
+	// ClusterServiceVersion object that the current ClusterServiceVersion
+	// replaces.
+	replaces = "replaces"
+)
+
+// ClusterServiceVersion is a structured representation of cluster service
+// version object(s) specified inside the 'clusterServiceVersions' section of
+// an operator manifest.
+type ClusterServiceVersion struct {
+	// Type metadata.
+	metav1.TypeMeta `json:",inline"`
+
+	// Object metadata.
+	metav1.ObjectMeta `json:"metadata"`
+
+	// Spec is the raw representation of the 'spec' element of
+	// ClusterServiceVersion object. Since we are
+	// not interested in the content of spec we are not parsing it.
+	Spec json.RawMessage `json:"spec"`
+}
+
+// GetReplaces returns the name of the older ClusterServiceVersion object that
+// is replaced by this ClusterServiceVersion object.
+//
+// If not defined, the function returns an empty string.
+func (csv *ClusterServiceVersion) GetReplaces() (string, error) {
+	var objmap map[string]*json.RawMessage
+	if err := json.Unmarshal(csv.Spec, &objmap); err != nil {
+		return "", err
+	}
+
+	rawValue, ok := objmap[replaces]
+	if !ok || rawValue == nil {
+		return "", nil
+	}
+
+	var replaces string
+	if err := json.Unmarshal(*rawValue, &replaces); err != nil {
+		return "", err
+	}
+
+	return replaces, nil
+}
+
+// GetCustomResourceDefintions returns a list of owned and required
+// CustomResourceDefinition object(s) specified inside the
+// 'customresourcedefinitions' section of a ClusterServiceVersion 'spec'.
+//
+// owned represents the list of CRD(s) managed by this ClusterServiceVersion
+// object.
+// required represents the list of CRD(s) that this ClusterServiceVersion
+// object depends on.
+//
+// If owned or required is not defined in the spec then an empty list is
+// returned respectively.
+func (csv *ClusterServiceVersion) GetCustomResourceDefintions() (owned []*CRDKey, required []*CRDKey, err error) {
+	var objmap map[string]*json.RawMessage
+
+	if err = json.Unmarshal(csv.Spec, &objmap); err != nil {
+		return
+	}
+
+	rawValue, ok := objmap[customResourceDefinitions]
+	if !ok || rawValue == nil {
+		return
+	}
+
+	var definitions struct {
+		Owned    []*CRDKey `json:"owned"`
+		Required []*CRDKey `json:"required"`
+	}
+
+	if err = json.Unmarshal(*rawValue, &definitions); err != nil {
+		return
+	}
+
+	owned = definitions.Owned
+	required = definitions.Required
+	return
+}

pkg/appregistry/dbloader.go

@@ -0,0 +1,56 @@
+package appregistry
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/operator-framework/operator-registry/pkg/sqlite"
+	"github.com/sirupsen/logrus"
+)
+
+type dbLoader struct {
+	logger *logrus.Entry
+}
+
+// LoadToSQLite uses configMaploader to load the downloaded operator manifest(s)
+// into a sqllite database.
+func (l *dbLoader) LoadToSQLite(dbName string, manifest *RawOperatorManifestData) (store *sqlite.SQLQuerier, err error) {
+	l.logger.Infof("using configmap loader to build sqlite database")
+
+	data := map[string]string{
+		"customResourceDefinitions": manifest.CustomResourceDefinitions,
+		"clusterServiceVersions":    manifest.ClusterServiceVersions,
+		"packages":                  manifest.Packages,
+	}
+
+	sqlLoader, err := sqlite.NewSQLLiteLoader(dbName)
+	if err != nil {
+		return
+	}
+
+	configMapPopulator := sqlite.NewSQLLoaderForConfigMapData(l.logger, sqlLoader, data)
+	if err = configMapPopulator.Populate(); err != nil {
+		return
+	}
+
+	s, err := sqlite.NewSQLLiteQuerier(dbName)
+	if err != nil {
+		err = fmt.Errorf("failed to load db: %v", err)
+		return
+	}
+
+	// sanity check that the db is available.
+	tables, err := s.ListTables(context.TODO())
+	if err != nil {
+		err = fmt.Errorf("couldn't list tables in db, incorrect config: %v", err)
+		return
+	}
+
+	if len(tables) == 0 {
+		err = fmt.Errorf("no tables found in db")
+		return
+	}
+
+	store = s
+	return
+}

pkg/appregistry/downloader.go

@@ -0,0 +1,180 @@
+package appregistry
+
+import (
+	"fmt"
+
+	"github.com/operator-framework/operator-marketplace/pkg/apis/marketplace/v1alpha1"
+	marketplace "github.com/operator-framework/operator-marketplace/pkg/client/clientset/versioned"
+	"github.com/operator-framework/operator-registry/pkg/apprclient"
+	"github.com/sirupsen/logrus"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+)
+
+// downloadItem encapsulates the data that is needed to download a specific repository.
+type downloadItem struct {
+	// Repository points to the repository and the particular release that needs
+	// to be downloaded.
+	RepositoryMetadata *apprclient.RegistryMetadata
+
+	// Spec refers to the remote appregistry URL and remote registry namespace.
+	Spec *v1alpha1.OperatorSourceSpec
+}
+
+func (d *downloadItem) String() string {
+	return fmt.Sprintf("%s", d.RepositoryMetadata)
+}
+
+type downloader struct {
+	logger *logrus.Entry
+	client marketplace.Interface
+}
+
+// Download downloads manifest(s) associated with the specified package(s) from
+// the corresponding operator source(s).
+//
+// We take a best effort approach in downloading.
+//
+// If an OperatorSource is not found, we skip it and move on to the next source.
+func (d *downloader) Download(input *Input) (manifests []*apprclient.OperatorMetadata, err error) {
+	items, err := d.Prepare(input)
+	if err != nil {
+		d.logger.Errorf("the following error(s) occurred while preparing the download list: %v", err)
+
+		if len(items) == 0 {
+			d.logger.Infof("download list is empty, bailing out: %s", input.Packages)
+			return
+		}
+	}
+
+	d.logger.Infof("resolved the following packages: %s", items)
+
+	manifests, err = d.DownloadRepositories(items)
+
+	return
+}
+
+// Prepare prepares the list of repositories to download by resolving each
+// package specified to its corresponding operator source.
+//
+// If a package is specified more than once, the operator source that it
+// resolves to the first time is picked.
+//
+// We apply a best-effort approach here, if a package is can't be resolved we
+// log it and move on.
+func (d *downloader) Prepare(input *Input) (items []*downloadItem, err error) {
+	packageMap := input.PackagesToMap()
+	itemMap := map[string]*downloadItem{}
+	allErrors := []error{}
+
+	for _, source := range input.Sources {
+		if len(packageMap) == 0 {
+			// All specified package(s) have already been resolved.
+			break
+		}
+
+		spec, repositoryList, err := d.QuerySource(source)
+		if err != nil {
+			allErrors = append(allErrors, err)
+			d.logger.Infof("skipping operator source due to error: %s", source)
+
+			continue
+		}
+
+		for _, metadata := range repositoryList {
+			// Repository name has a one to one mapping to operator/package name.
+			// We use this as the key.
+			key := metadata.Name
+
+			if _, ok := packageMap[key]; ok {
+				// The package specified has been resolved to this repository
+				// name in remote registry.
+				itemMap[key] = &downloadItem{
+					RepositoryMetadata: metadata,
+					Spec:               spec,
+				}
+
+				// Remove the package specified since it has been resolved.
+				delete(packageMap, key)
+			}
+		}
+	}
+
+	// We might still have packages specified that have not been resolved.
+	if len(packageMap) > 0 {
+		d.logger.Infof("the following packages could not be resolved: %v", packageMap)
+	}
+
+	items = make([]*downloadItem, 0)
+	for _, v := range itemMap {
+		items = append(items, v)
+	}
+
+	err = utilerrors.NewAggregate(allErrors)
+	return
+}
+
+// DownloadRepositories iterates through each download item and downloads
+// operator manifest from the corresponding repository.
+func (d *downloader) DownloadRepositories(items []*downloadItem) (manifests []*apprclient.OperatorMetadata, err error) {
+	allErrors := []error{}
+
+	manifests = make([]*apprclient.OperatorMetadata, 0)
+	for _, item := range items {
+		endpoint := item.Spec.Endpoint
+
+		d.logger.Infof("downloading repository: %s from %s", item.RepositoryMetadata, endpoint)
+
+		factory := apprclient.NewClientFactory()
+
+		client, err := factory.New("appregistry", endpoint)
+		if err != nil {
+			allErrors = append(allErrors, err)
+			d.logger.Infof("skipping repository: %s", item.RepositoryMetadata)
+
+			continue
+		}
+
+		manifest, err := client.RetrieveOne(item.RepositoryMetadata.ID(), item.RepositoryMetadata.Release)
+		if err != nil {
+			allErrors = append(allErrors, err)
+			d.logger.Infof("skipping repository: %s", item.RepositoryMetadata)
+
+			continue
+		}
+
+		manifests = append(manifests, manifest)
+	}
+
+	err = utilerrors.NewAggregate(allErrors)
+	return
+}
+
+// QuerySource retrives the OperatorSource object specified by key. It queries
+// the registry namespace to list all the repositories associated with this
+// operator source.
+//
+// The function returns the spec ( associated with the OperatorSource object )
+// in the cluster and the list of repositories in remote registry associated
+// with it.
+func (d *downloader) QuerySource(key *types.NamespacedName) (spec *v1alpha1.OperatorSourceSpec, repositories []*apprclient.RegistryMetadata, err error) {
+	opsrc, err := d.client.MarketplaceV1alpha1().OperatorSources(key.Namespace).Get(key.Name, metav1.GetOptions{})
+	if err != nil {
+		return
+	}
+
+	factory := apprclient.NewClientFactory()
+	client, err := factory.New("appregistry", opsrc.Spec.Endpoint)
+	if err != nil {
+		return
+	}
+
+	repositories, err = client.ListPackages(opsrc.Spec.RegistryNamespace)
+	if err != nil {
+		return
+	}
+
+	spec = &opsrc.Spec
+	return
+}

pkg/appregistry/input.go

@@ -0,0 +1,103 @@
+package appregistry
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/types"
+)
+
+type Input struct {
+	// Sources is the set of namespaced name(s) of OperatorSource objects from
+	// which we need to pull packages.
+	Sources []*types.NamespacedName
+
+	// Packages is the set of package name(s) specified.
+	Packages []string
+}
+
+func (i *Input) PackagesToMap() map[string]bool {
+	packages := map[string]bool{}
+
+	for _, pkg := range i.Packages {
+		packages[pkg] = false
+	}
+
+	return packages
+}
+
+type inputParser struct {
+}
+
+// Parse parses the raw input provided, sanitizes it and returns an instance of
+// Input.
+//
+// csvSources is a comma separated list of namespaced name that specifies
+// the operator source(s), it is expected to comply to the following format -
+// {namespace}/{name},{namespace}/{name},
+//
+// csvPackages is a comma separated list of packages. It is expected to have
+// the following format.
+// etcd,prometheus,descheduler
+func (p *inputParser) Parse(csvSources string, csvPackages string) (*Input, error) {
+	sources, err := parseSources(csvSources)
+	if err != nil {
+		return nil, err
+	}
+
+	packages := sanitizePackageList(strings.Split(csvPackages, ","))
+
+	return &Input{
+		Sources:  sources,
+		Packages: packages,
+	}, nil
+}
+
+func parseSources(csvSources string) ([]*types.NamespacedName, error) {
+	values := strings.Split(csvSources, ",")
+	if len(values) == 0 {
+		return nil, errors.New(fmt.Sprintf("No OperatorSource(s) has been specified"))
+	}
+
+	names := make([]*types.NamespacedName, 0)
+	for _, v := range values {
+		name, err := split(v)
+		if err != nil {
+			return nil, err
+		}
+
+		names = append(names, name)
+	}
+
+	return names, nil
+}
+
+// sanitizePackageList sanitizes the set of package(s) specified. It removes
+// duplicates and ignores empty string.
+func sanitizePackageList(in []string) []string {
+	out := make([]string, 0)
+
+	inMap := map[string]bool{}
+	for _, item := range in {
+		if _, ok := inMap[item]; ok || item == "" {
+			continue
+		}
+
+		out = append(out, item)
+	}
+
+	return out
+}
+
+func split(sourceName string) (*types.NamespacedName, error) {
+	split := strings.Split(sourceName, "/")
+	if len(split) != 2 {
+		return nil, errors.New(fmt.Sprintf("OperatorSource name should be specified in this format {namespace}/{name}"))
+	}
+
+	return &types.NamespacedName{
+		Namespace: split[0],
+		Name:      split[1],
+	}, nil
+}

pkg/appregistry/merger.go

@@ -0,0 +1,50 @@
+package appregistry
+
+import (
+	"github.com/operator-framework/operator-registry/pkg/apprclient"
+	"github.com/sirupsen/logrus"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+)
+
+type merger struct {
+	logger *logrus.Entry
+	parser ManifestYAMLParser
+}
+
+// Merge merges a set of operator manifest(s) into one.
+//
+// For a given operator source we have N ( N >= 1 ) repositories within the
+// given registry namespace. It is required for each repository to contain
+// manifest for a single operator.
+//
+// Once downloaded we can use this function to merge manifest(s) from all
+// relevant repositories into an uber manifest.
+//
+// We assume that all CRD(s), CSV(s) and package(s) are globally unique.
+// Otherwise we will fail to load the uber manifest into sqlite.
+func (m *merger) Merge(rawManifests []*apprclient.OperatorMetadata) (manifests *RawOperatorManifestData, err error) {
+	allErrors := []error{}
+	merged := StructuredOperatorManifestData{}
+
+	for _, rawManifest := range rawManifests {
+		manifest, err := m.parser.Unmarshal(rawManifest.RawYAML)
+		if err != nil {
+			allErrors = append(allErrors, err)
+			m.logger.Infof("skipping repository due to parsing error - %s", rawManifest.RegistryMetadata)
+
+			continue
+		}
+
+		merged.Packages = append(merged.Packages, manifest.Packages...)
+		merged.CustomResourceDefinitions = append(merged.CustomResourceDefinitions, manifest.CustomResourceDefinitions...)
+		merged.ClusterServiceVersions = append(merged.ClusterServiceVersions, manifest.ClusterServiceVersions...)
+	}
+
+	manifests, err = m.parser.Marshal(&merged)
+	if err != nil {
+		allErrors = append(allErrors, err)
+	}
+
+	err = utilerrors.NewAggregate(allErrors)
+	return
+}

pkg/appregistry/parser.go

@@ -0,0 +1,181 @@
+package appregistry
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/ghodss/yaml"
+)
+
+// ManifestYAMLParser is an interface that is responsible for marshaling raw
+// operator manifest into structured representation and vice versa.
+type ManifestYAMLParser interface {
+	// Unmarshal unmarshals raw operator manifest YAML into structured
+	// representation.
+	//
+	// The function accepts raw yaml specified in rawYAML and converts it into
+	// an instance of StructuredOperatorManifestData.
+	Unmarshal(rawYAML []byte) (marshaled *StructuredOperatorManifestData, err error)
+
+	// Marshal marshals a structured representation of an operator manifest into
+	// raw YAML representation so that it can be used to create a configMap
+	// object for a catalog source in OLM.
+	//
+	// The function accepts a structured representation of operator manifest(s)
+	// specified in marshaled and returns a raw yaml representation of it.
+	Marshal(bundle *StructuredOperatorManifestData) (*RawOperatorManifestData, error)
+}
+
+// RawOperatorManifestData encapsulates the list of CRD(s), CSV(s) and
+// package(s) associated with a set of manifest(s).
+type RawOperatorManifestData struct {
+	// CustomResourceDefinitions is the set of custom resource definition(s)
+	// associated with this package manifest.
+	CustomResourceDefinitions string `yaml:"customResourceDefinitions"`
+
+	// ClusterServiceVersions is the set of cluster service version(s)
+	// associated with this package manifest.
+	ClusterServiceVersions string `yaml:"clusterServiceVersions"`
+
+	// Packages is the set of package(s) associated with this operator manifest.
+	Packages string `yaml:"packages"`
+}
+
+// StructuredOperatorManifestData is a structured representation of operator
+// manifest(s). An operator manifest is a YAML document with the following
+// sections:
+// - customResourceDefinitions
+// - clusterServiceVersions
+// - packages
+//
+// An operator manifest is unmarshaled into this type so that we can perform
+// certain operations like, but not limited to:
+// - Construct a new operator manifest object to be used by a CatalogSourceConfig
+//   by combining a set of existing operator manifest(s).
+// - Construct a new operator manifest object by extracting a certain
+//   operator/package from a a given operator manifest.
+type StructuredOperatorManifestData struct {
+	// CustomResourceDefinitions is the list of custom resource definition(s)
+	// associated with this operator manifest.
+	CustomResourceDefinitions []CustomResourceDefinition `json:"customResourceDefinitions"`
+
+	// ClusterServiceVersions is the list of cluster service version(s)
+	//associated with this operators manifest.
+	ClusterServiceVersions []ClusterServiceVersion `json:"clusterServiceVersions"`
+
+	// Packages is the list of package(s) associated with this operator manifest.
+	Packages []PackageManifest `json:"packages"`
+}
+
+// PackageManifest holds information about a package, which is a reference to
+// one (or more) channels under a single package.
+//
+// The following type has been copied as is from OLM.
+// See https://github.com/operator-framework/operator-lifecycle-manager/blob/724b209ccfff33b6208cc5d05283800d6661d441/pkg/controller/registry/types.go#L78:6.
+//
+// We use it to unmarshal 'packages' element of an operator manifest.
+type PackageManifest struct {
+	// PackageName is the name of the overall package, ala `etcd`.
+	PackageName string `json:"packageName"`
+
+	// Channels are the declared channels for the package,
+	// ala `stable` or `alpha`.
+	Channels []PackageChannel `json:"channels"`
+
+	// DefaultChannelName is, if specified, the name of the default channel for
+	// the package. The default channel will be installed if no other channel is
+	// explicitly given. If the package has a single channel, then that
+	// channel is implicitly the default.
+	DefaultChannelName string `json:"defaultChannel"`
+}
+
+// PackageChannel defines a single channel under a package, pointing to a
+// version of that package.
+//
+// The following type has been directly copied as is from OLM.
+// See https://github.com/operator-framework/operator-lifecycle-manager/blob/724b209ccfff33b6208cc5d05283800d6661d441/pkg/controller/registry/types.go#L105.
+//
+// We use it to unmarshal 'packages/package/channels' element of
+// an operator manifest.
+type PackageChannel struct {
+	// Name is the name of the channel, e.g. `alpha` or `stable`.
+	Name string `json:"name"`
+
+	// CurrentCSVName defines a reference to the CSV holding the version of
+	// this package currently for the channel.
+	CurrentCSVName string `json:"currentCSV"`
+}
+
+type manifestYAMLParser struct{}
+
+func (*manifestYAMLParser) Unmarshal(rawYAML []byte) (*StructuredOperatorManifestData, error) {
+	var manifestYAML struct {
+		Data RawOperatorManifestData `yaml:"data"`
+	}
+
+	if err := yaml.Unmarshal(rawYAML, &manifestYAML); err != nil {
+		return nil, fmt.Errorf("error parsing raw YAML : %s", err)
+	}
+
+	var crds []CustomResourceDefinition
+	var csvs []ClusterServiceVersion
+	var packages []PackageManifest
+	data := manifestYAML.Data
+
+	crdJSONRaw, err := yaml.YAMLToJSON([]byte(data.CustomResourceDefinitions))
+	if err != nil {
+		return nil, fmt.Errorf("error converting CRD list (YAML) to JSON : %s", err)
+	}
+	if err := json.Unmarshal(crdJSONRaw, &crds); err != nil {
+		return nil, fmt.Errorf("error parsing CRD list (JSON) : %s", err)
+	}
+
+	csvJSONRaw, err := yaml.YAMLToJSON([]byte(data.ClusterServiceVersions))
+	if err != nil {
+		return nil, fmt.Errorf("error converting CSV list (YAML) to JSON : %s", err)
+	}
+	if err := json.Unmarshal(csvJSONRaw, &csvs); err != nil {
+		return nil, fmt.Errorf("error parsing CSV list (JSON) : %s", err)
+	}
+
+	packageJSONRaw, err := yaml.YAMLToJSON([]byte(data.Packages))
+	if err != nil {
+		return nil, fmt.Errorf("error converting package list (JSON) to YAML : %s", err)
+	}
+	if err := json.Unmarshal(packageJSONRaw, &packages); err != nil {
+		return nil, fmt.Errorf("error parsing package list (JSON) : %s", err)
+	}
+
+	marshaled := &StructuredOperatorManifestData{
+		CustomResourceDefinitions: crds,
+		ClusterServiceVersions:    csvs,
+		Packages:                  packages,
+	}
+
+	return marshaled, nil
+}
+
+func (*manifestYAMLParser) Marshal(bundle *StructuredOperatorManifestData) (*RawOperatorManifestData, error) {
+	crdRaw, err := yaml.Marshal(bundle.CustomResourceDefinitions)
+	if err != nil {
+		return nil, fmt.Errorf("error marshaling CRD list into yaml : %s", err)
+	}
+
+	csvRaw, err := yaml.Marshal(bundle.ClusterServiceVersions)
+	if err != nil {
+		return nil, fmt.Errorf("error marshaling CSV list into YAML : %s", err)
+	}
+
+	packageRaw, err := yaml.Marshal(bundle.Packages)
+	if err != nil {
+		return nil, fmt.Errorf("error marshaling package list into YAML : %s", err)
+	}
+
+	data := &RawOperatorManifestData{
+		CustomResourceDefinitions: string(crdRaw),
+		ClusterServiceVersions:    string(csvRaw),
+		Packages:                  string(packageRaw),
+	}
+
+	return data, nil
+}

vendor/github.com/operator-framework/operator-marketplace/LICENSE

@@ -107,6 +107,12 @@ github.com/operator-framework/go-appr/appregistry/info
 github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/operators/v1alpha1
 github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry
 github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/operators
+# github.com/operator-framework/operator-marketplace v0.0.0-20190212161948-a7ca81b96ad9
+github.com/operator-framework/operator-marketplace/pkg/apis/marketplace/v1alpha1
+github.com/operator-framework/operator-marketplace/pkg/client/clientset/versioned
+github.com/operator-framework/operator-marketplace/pkg/client/clientset/versioned/typed/marketplace/v1alpha1
+github.com/operator-framework/operator-marketplace/pkg/client/clientset/versioned/scheme
+github.com/operator-framework/operator-marketplace/pkg/apis
 # github.com/peterbourgon/diskv v2.0.1+incompatible
 github.com/peterbourgon/diskv
 # github.com/pmezard/go-difflib v1.0.0
@@ -213,15 +219,17 @@ k8s.io/api/storage/v1
 k8s.io/api/storage/v1alpha1
 k8s.io/api/storage/v1beta1
 # k8s.io/apiextensions-apiserver v0.0.0-20181204003618-e419c5771cdc
-k8s.io/apiextensions-apiserver/pkg/apis/apiextensions
 k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1
+k8s.io/apiextensions-apiserver/pkg/apis/apiextensions
 k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/validation
 k8s.io/apiextensions-apiserver/pkg/apiserver/validation
 k8s.io/apiextensions-apiserver/pkg/features
 # k8s.io/apimachinery v0.0.0-20190118094746-1525e4dadd2d
 k8s.io/apimachinery/pkg/apis/meta/v1
 k8s.io/apimachinery/pkg/apis/meta/v1/unstructured
 k8s.io/apimachinery/pkg/util/yaml
+k8s.io/apimachinery/pkg/types
+k8s.io/apimachinery/pkg/util/errors
 k8s.io/apimachinery/pkg/runtime
 k8s.io/apimachinery/pkg/runtime/serializer
 k8s.io/apimachinery/pkg/api/validation
@@ -232,14 +240,12 @@ k8s.io/apimachinery/pkg/api/resource
 k8s.io/apimachinery/pkg/fields
 k8s.io/apimachinery/pkg/labels
 k8s.io/apimachinery/pkg/selection
-k8s.io/apimachinery/pkg/types
 k8s.io/apimachinery/pkg/util/intstr
 k8s.io/apimachinery/pkg/watch
 k8s.io/apimachinery/pkg/api/errors
 k8s.io/apimachinery/pkg/runtime/serializer/streaming
 k8s.io/apimachinery/pkg/util/net
 k8s.io/apimachinery/pkg/util/sets
-k8s.io/apimachinery/pkg/util/errors
 k8s.io/apimachinery/pkg/util/validation
 k8s.io/apimachinery/pkg/util/json
 k8s.io/apimachinery/pkg/util/runtime
@@ -318,3 +324,5 @@ k8s.io/client-go/util/connrotation
 k8s.io/client-go/tools/clientcmd/api/v1
 # k8s.io/kube-openapi v0.0.0-20181031203759-72693cb1fadd
 k8s.io/kube-openapi/pkg/util/proto
+# sigs.k8s.io/controller-runtime v0.1.10
+sigs.k8s.io/controller-runtime/pkg/runtime/scheme