Reconcile OperatorSource type

Gopkg.toml

@@ -60,3 +60,7 @@ required = [
 [[constraint]]
   name = "github.com/operator-framework/operator-lifecycle-manager"
   version = "0.7.1"
+
+[[constraint]]  
+  branch = "master"
+  name = "github.com/operator-framework/go-appr"

cmd/marketplace-operator/main.go

@@ -31,10 +31,12 @@ func main() {
 	if err != nil {
 		logrus.Fatalf("failed to get watch namespace: %v", err)
 	}
-	resyncPeriod := time.Duration(5) * time.Second
-	logrus.Infof("Watching %s, %s, %s", resource, catalogSourceConfigKind, namespace)
+
 	// No reason to resync until we implement updating of CatalogSourceConfig CRs
-	sdk.Watch(resource, catalogSourceConfigKind, namespace, 0)
+	resyncPeriod := time.Duration(0) * time.Second
+
+	logrus.Infof("Watching %s, %s, %s", resource, catalogSourceConfigKind, namespace)
+	sdk.Watch(resource, catalogSourceConfigKind, namespace, resyncPeriod)
 
 	operatorSourceKind := "OperatorSource"
 	logrus.Infof("Watching %s, %s, %s, %d", resource, operatorSourceKind, namespace, resyncPeriod)

deploy/operatorsource.cr.yaml

@@ -1,4 +1,7 @@
 apiVersion: "marketplace.redhat.com/v1alpha1"
 kind: "OperatorSource"
 metadata:
-  name: "operatorsource-example"
+  name: "localhost"
+spec:
+  type: appregistry
+  endpoint: "http://localhost:5000/cnr"

pkg/apis/marketplace/v1alpha1/register.go

@@ -17,7 +17,9 @@ var (
 	SchemeBuilder = runtime.NewSchemeBuilder(addKnownTypes)
 	AddToScheme   = SchemeBuilder.AddToScheme
 	// SchemeGroupVersion is the group version used to register these objects.
-	SchemeGroupVersion = schema.GroupVersion{Group: groupName, Version: version}
+	SchemeGroupVersion      = schema.GroupVersion{Group: groupName, Version: version}
+	OperatorSourceKind      = "OperatorSource"
+	CatalogSourceConfigKind = "CatalogSourceConfig"
 )
 
 func init() {

pkg/apis/marketplace/v1alpha1/types.go

@@ -54,8 +54,13 @@ type OperatorSource struct {
 }
 
 type OperatorSourceSpec struct {
-	// Fill me
+	// Type of operator source
+	Type string `json:"type"`
+
+	// Endpoint points to the URL from where operator manifests can be fetched
+	Endpoint string `json:"endpoint"`
 }
+
 type OperatorSourceStatus struct {
 	// Fill me
 }

pkg/appregistry/adapter.go

@@ -0,0 +1,71 @@
+package appregistry
+
+import (
+	"bytes"
+
+	appr "github.com/operator-framework/go-appr/appregistry"
+	appr_blobs "github.com/operator-framework/go-appr/appregistry/blobs"
+	appr_package "github.com/operator-framework/go-appr/appregistry/package_appr"
+	appr_models "github.com/operator-framework/go-appr/models"
+)
+
+const (
+	mediaType string = "helm"
+)
+
+// This interface (internal to this package) encapsulates nitty gritty details of go-appr client bindings
+type apprApiAdapter interface {
+	// ListPackages returns a list of package(s) available to the user
+	ListPackages() (appr_models.Packages, error)
+
+	// GetPackageMetadata returns metadata associated with a given package
+	GetPackageMetadata(namespace string, repository string, release string) (*appr_models.Package, error)
+
+	// DownloadOperatorManifest downloads the blob associated with a given digest that directly corresponds to a package release
+	DownloadOperatorManifest(namespace string, repository string, digest string) ([]byte, error)
+}
+
+type apprApiAdapterImpl struct {
+	client *appr.Appregistry
+}
+
+func (a *apprApiAdapterImpl) ListPackages() (appr_models.Packages, error) {
+	params := appr_package.NewListPackagesParams()
+
+	packages, err := a.client.PackageAppr.ListPackages(params)
+	if err != nil {
+		return nil, err
+	}
+
+	return packages.Payload, nil
+}
+
+func (a *apprApiAdapterImpl) GetPackageMetadata(namespace string, repository string, release string) (*appr_models.Package, error) {
+	params := appr_package.NewShowPackageParams().
+		WithNamespace(namespace).
+		WithPackage(repository).
+		WithRelease(release).
+		WithMediaType(mediaType)
+
+	pkg, err := a.client.PackageAppr.ShowPackage(params)
+	if err != nil {
+		return nil, err
+	}
+
+	return pkg.Payload, nil
+}
+
+func (a *apprApiAdapterImpl) DownloadOperatorManifest(namespace string, repository string, digest string) ([]byte, error) {
+	params := appr_blobs.NewPullBlobParams().
+		WithNamespace(namespace).
+		WithPackage(repository).
+		WithDigest(digest)
+
+	writer := &bytes.Buffer{}
+	_, err := a.client.Blobs.PullBlob(params, writer)
+	if err != nil {
+		return nil, err
+	}
+
+	return writer.Bytes(), nil
+}

pkg/appregistry/appregistry.go

@@ -0,0 +1,48 @@
+package appregistry
+
+import (
+	"net/url"
+
+	"github.com/go-openapi/runtime"
+	httptransport "github.com/go-openapi/runtime/client"
+	"github.com/go-openapi/strfmt"
+	apprclient "github.com/operator-framework/go-appr/appregistry"
+)
+
+// NewClientFactory return a factory which can be used to instantiate a new appregistry client
+func NewClientFactory() ClientFactory {
+	return &factory{}
+}
+
+type ClientFactory interface {
+	// New returns a new instance of appregistry Client from given source and type
+	New(sourceType, source string) (Client, error)
+}
+
+// Client exposes the functionality of app registry server
+type Client interface {
+	// RetrieveAll retrieves all visible packages from the given source
+	RetrieveAll() ([]*OperatorMetadata, error)
+
+	// RetrieveOneretrieves a given package from the source
+	RetrieveOne(name, release string) (*OperatorMetadata, error)
+}
+
+type factory struct{}
+
+func (f *factory) New(sourceType, source string) (Client, error) {
+	u, err := url.Parse(source)
+	if err != nil {
+		return nil, err
+	}
+
+	transport := httptransport.New(u.Host, u.Path, []string{u.Scheme})
+	transport.Consumers["application/x-gzip"] = runtime.ByteStreamConsumer()
+	c := apprclient.New(transport, strfmt.Default)
+
+	return &client{
+		adapter:      &apprApiAdapterImpl{client: c},
+		decoder:      &blobDecoderImpl{},
+		unmarshaller: &blobUnmarshallerImpl{},
+	}, nil
+}

pkg/appregistry/appregistry_test.go

@@ -0,0 +1,22 @@
+package appregistry_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/operator-framework/operator-marketplace/pkg/appregistry"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRetrieveAll(t *testing.T) {
+	factory := appregistry.NewClientFactory()
+
+	client, err := factory.New("appregistry", "http://localhost:5000/cnr")
+	require.NoError(t, err)
+
+	packages, err := client.RetrieveAll()
+
+	assert.NoError(t, err)
+	assert.NotNil(t, packages)
+}

pkg/appregistry/client.go

@@ -0,0 +1,105 @@
+package appregistry
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+)
+
+// OperatorMetadata encapsulates operator metadata and manifest assocated with a package
+type OperatorMetadata struct {
+	// Namespace is the namespace in app registry server under which the package is hosted.
+	Namespace string
+
+	// Repository is the repository name for the specified package in app registry
+	Repository string
+
+	// Release represents the release or version number of the given package
+	Release string
+
+	// Digest is the sha256 hash value that uniquely corresponds to the blob associated with the release
+	Digest string
+
+	// Manifest encapsulates operator manifest
+	Manifest *Manifest
+}
+
+func (om *OperatorMetadata) ID() string {
+	return fmt.Sprintf("%s/%s", om.Namespace, om.Repository)
+}
+
+type client struct {
+	adapter      apprApiAdapter
+	decoder      blobDecoder
+	unmarshaller blobUnmarshaller
+}
+
+func (c *client) RetrieveAll() ([]*OperatorMetadata, error) {
+	packages, err := c.adapter.ListPackages()
+	if err != nil {
+		return nil, err
+	}
+
+	list := make([]*OperatorMetadata, len(packages))
+	for i, pkg := range packages {
+		manifest, err := c.RetrieveOne(pkg.Name, pkg.Default)
+		if err != nil {
+			return nil, err
+		}
+
+		list[i] = manifest
+	}
+
+	return list, nil
+}
+
+func (c *client) RetrieveOne(name, release string) (*OperatorMetadata, error) {
+	namespace, repository, err := split(name)
+	if err != nil {
+		return nil, err
+	}
+
+	metadata, err := c.adapter.GetPackageMetadata(namespace, repository, release)
+	if err != nil {
+		return nil, err
+	}
+
+	digest := metadata.Content.Digest
+	blob, err := c.adapter.DownloadOperatorManifest(namespace, repository, digest)
+	if err != nil {
+		return nil, err
+	}
+
+	decoded, err := c.decoder.Decode(blob)
+	if err != nil {
+		return nil, err
+	}
+
+	manifest, err := c.unmarshaller.Unmarshal(decoded)
+	if err != nil {
+		return nil, err
+	}
+
+	om := &OperatorMetadata{
+		Namespace:  namespace,
+		Repository: repository,
+		Release:    release,
+		Manifest:   manifest,
+		Digest:     digest,
+	}
+
+	return om, nil
+}
+
+func split(name string) (namespace string, repository string, err error) {
+	// we expect package name to comply to this format - {namespace}/{repository}
+	split := strings.Split(name, "/")
+	if len(split) != 2 {
+		return "", "", errors.New(fmt.Sprintf("package name should be specified in this format {namespace}/{repository}"))
+	}
+
+	namespace = split[0]
+	repository = split[1]
+
+	return namespace, repository, nil
+}

pkg/appregistry/client_test.go

@@ -0,0 +1,61 @@
+package appregistry
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/golang/mock/gomock"
+	appr_models "github.com/operator-framework/go-appr/models"
+)
+
+func TestRetrieveOne_PackageExists_SuccessExpected(t *testing.T) {
+	controller := gomock.NewController(t)
+	defer controller.Finish()
+
+	adapter := NewMockapprApiAdapter(controller)
+	decoder := NewMockblobDecoder(controller)
+	unmarshaller := NewMockblobUnmarshaller(controller)
+
+	client := client{
+		adapter:      adapter,
+		decoder:      decoder,
+		unmarshaller: unmarshaller,
+	}
+
+	namespace := "redhat"
+	repository := "foo"
+	release := "1.0"
+	digest := "abcdefgh"
+
+	pkg := &appr_models.Package{Content: &appr_models.OciDescriptor{
+		Digest: digest,
+	}}
+	adapter.EXPECT().GetPackageMetadata(namespace, repository, release).Return(pkg, nil).Times(1)
+
+	blobExpected := []byte{'e', 'n', 'c', 'o', 'd', 'e', 'd'}
+	adapter.EXPECT().DownloadOperatorManifest(namespace, repository, digest).Return(blobExpected, nil).Times(1)
+
+	decodedExpected := []byte{'d', 'e', 'c', 'o', 'd', 'e', 'd'}
+	decoder.EXPECT().Decode(blobExpected).Return(decodedExpected, nil).Times(1)
+
+	manifestExpected := &Manifest{
+		Publisher: "redhat",
+		Data: Data{
+			CRDs:     "my crds",
+			CSVs:     "my csvs",
+			Packages: "my packages",
+		},
+	}
+	unmarshaller.EXPECT().Unmarshal(decodedExpected).Return(manifestExpected, nil)
+
+	metadata, err := client.RetrieveOne(fmt.Sprintf("%s/%s", namespace, repository), release)
+
+	assert.NoError(t, err)
+	assert.Equal(t, namespace, metadata.Namespace)
+	assert.Equal(t, repository, metadata.Repository)
+	assert.Equal(t, release, metadata.Release)
+	assert.Equal(t, digest, metadata.Digest)
+	assert.Equal(t, manifestExpected, metadata.Manifest)
+}