Auto provision image stream

pkg/dockerregistry/middleware/repository/openshift.go

@@ -38,20 +38,21 @@ type repository struct {
 
 // newRepository returns a new repository middleware.
 func newRepository(repo distribution.Repository, options map[string]interface{}) (distribution.Repository, error) {
+	registryAddr := os.Getenv("REGISTRY_URL")
+	if len(registryAddr) == 0 {
+		return nil, errors.New("REGISTRY_URL is required")
+	}
 
 	registryClient, err := dockerregistry.NewRegistryOpenShiftClient()
 	if err != nil {
 		return nil, err
 	}
-	registryAddr := os.Getenv("REGISTRY_URL")
-	if len(registryAddr) == 0 {
-		return nil, errors.New("REGISTRY_URL is required")
-	}
 
 	nameParts := strings.SplitN(repo.Name(), "/", 2)
 	if len(nameParts) != 2 {
-		return nil, errors.New("Incorrect image stream name")
+		return nil, fmt.Errorf("Invalid repository name %q: it must be of the format <project>/<name>", repo.Name())
 	}
+
 	return &repository{
 		Repository:     repo,
 		registryClient: registryClient,
@@ -124,14 +125,13 @@ func (r *repository) GetByTag(ctx context.Context, tag string) (*manifest.Signed
 		// <repo>:<hex portion of digest>, so once we verify we got a 404 from
 		// getImageStreamTag, we construct a digest and attempt to get the
 		// imageStreamImage using that digest.
-
-		// TODO replace with kerrors.IsStatusError when it's rebased in
 		if err, ok := err.(*kerrors.StatusError); !ok {
 			log.Errorf("GetByTag: getImageStreamTag returned error: %s", err)
 			return nil, err
 		} else if err.ErrStatus.Code != http.StatusNotFound {
 			log.Errorf("GetByTag: getImageStreamTag returned non-404: %s", err)
 		}
+
 		// let's try to get by id
 		dgst, dgstErr := digest.ParseDigest("sha256:" + tag)
 		if dgstErr != nil {
@@ -173,7 +173,7 @@ func (r *repository) Put(ctx context.Context, manifest *manifest.SignedManifest)
 	}
 
 	// Upload to openshift
-	irm := imageapi.ImageStreamMapping{
+	ism := imageapi.ImageStreamMapping{
 		ObjectMeta: kapi.ObjectMeta{
 			Namespace: r.namespace,
 			Name:      r.name,
@@ -188,9 +188,42 @@ func (r *repository) Put(ctx context.Context, manifest *manifest.SignedManifest)
 		},
 	}
 
-	if err := r.registryClient.ImageStreamMappings(r.namespace).Create(&irm); err != nil {
-		log.Errorf("Error creating ImageStreamMapping: %s", err)
-		return err
+	if err := r.registryClient.ImageStreamMappings(r.namespace).Create(&ism); err != nil {
+		// if the error was that the image stream wasn't found, try to auto provision it
+		statusErr, ok := err.(*kerrors.StatusError)
+		if !ok {
+			log.Errorf("Error creating ImageStreamMapping: %s", err)
+			return err
+		}
+
+		status := statusErr.ErrStatus
+		if status.Code != http.StatusNotFound || status.Details.Kind != "imageStream" || status.Details.ID != r.name {
+			log.Errorf("Error creating ImageStreamMapping: %s", err)
+			return err
+		}
+
+		stream := imageapi.ImageStream{
+			ObjectMeta: kapi.ObjectMeta{
+				Name: r.name,
+			},
+		}
+
+		client, err := getUserOpenShiftClient(ctx)
+		if err != nil {
+			log.Errorf("Error creating user client to auto provision image stream: %s", err)
+			return statusErr
+		}
+
+		if _, err := client.ImageStreams(r.namespace).Create(&stream); err != nil {
+			log.Errorf("Error auto provisioning image stream: %s", err)
+			return statusErr
+		}
+
+		// try to create the ISM again
+		if err := r.registryClient.ImageStreamMappings(r.namespace).Create(&ism); err != nil {
+			log.Errorf("Error creating ImageStreamMapping: %s", err)
+			return err
+		}
 	}
 
 	// Grab each json signature and store them.
@@ -224,7 +257,8 @@ func (r *repository) getImageStream(ctx context.Context) (*imageapi.ImageStream,
 }
 
 // getImage retrieves the Image with digest `dgst`. This uses the registry's
-// credentials and should ONLY
+// credentials and should ONLY be called after verifying the user has access
+// to the image stream the imgae belongs to.
 func (r *repository) getImage(dgst digest.Digest) (*imageapi.Image, error) {
 	return r.registryClient.Images().Get(dgst.String())
 }

pkg/image/registry/imagestreammapping/rest_test.go

@@ -2,6 +2,7 @@ package imagestreammapping
 
 import (
 	"fmt"
+	"net/http"
 	"reflect"
 	"strings"
 	"testing"
@@ -142,7 +143,7 @@ func TestCreateErrorListingImageStreams(t *testing.T) {
 	}
 }
 
-func TestCreateImageStreamNotFound(t *testing.T) {
+func TestCreateImageStreamNotFoundWithDockerImageRepository(t *testing.T) {
 	fakeEtcdClient, _, storage := setup(t)
 	fakeEtcdClient.Data["/imageRepositories/default"] = tools.EtcdResponseWithError{
 		R: &etcd.Response{
@@ -166,6 +167,32 @@ func TestCreateImageStreamNotFound(t *testing.T) {
 	}
 }
 
+func TestCreateImageStreamNotFoundWithName(t *testing.T) {
+	fakeEtcdClient, _, storage := setup(t)
+	fakeEtcdClient.ExpectNotFoundGet("/imageRepositories/default/somerepo")
+
+	obj, err := storage.Create(kapi.NewDefaultContext(), validNewMappingWithName())
+	if obj != nil {
+		t.Errorf("Unexpected non-nil obj %#v", obj)
+	}
+	if err == nil {
+		t.Fatal("Unexpected nil err")
+	}
+	e, ok := err.(*errors.StatusError)
+	if !ok {
+		t.Fatalf("expected StatusError, got %#v", err)
+	}
+	if e, a := http.StatusNotFound, e.ErrStatus.Code; e != a {
+		t.Errorf("error status code: expected %d, got %d", e, a)
+	}
+	if e, a := "imageStream", e.ErrStatus.Details.Kind; e != a {
+		t.Errorf("error status details kind: expected %s, got %s", e, a)
+	}
+	if e, a := "somerepo", e.ErrStatus.Details.ID; e != a {
+		t.Errorf("error status details name: expected %s, got %s", e, a)
+	}
+}
+
 func TestCreateSuccessWithDockerImageRepository(t *testing.T) {
 	fakeEtcdClient, helper, storage := setup(t)
 

test/integration/v2_docker_registry_test.go

@@ -27,7 +27,7 @@ func init() {
 	testutil.RequireEtcd()
 }
 
-func signedManifest() ([]byte, digest.Digest, error) {
+func signedManifest(name string) ([]byte, digest.Digest, error) {
 	key, err := libtrust.GenerateECP256PrivateKey()
 	if err != nil {
 		return []byte{}, "", fmt.Errorf("error generating EC key: %s", err)
@@ -37,7 +37,7 @@ func signedManifest() ([]byte, digest.Digest, error) {
 		Versioned: manifest.Versioned{
 			SchemaVersion: 1,
 		},
-		Name:         "test-integration/test",
+		Name:         name,
 		Tag:          "latest",
 		Architecture: "amd64",
 		History: []manifest.History{
@@ -86,12 +86,12 @@ func TestV2RegistryGetTags(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error getting cluster admin client config: %v", err)
 	}
-	adminUser := "admin"
-	adminClient, err := testutil.CreateNewProject(clusterAdminClient, *clusterAdminClientConfig, testutil.Namespace(), adminUser)
+	user := "admin"
+	adminClient, err := testutil.CreateNewProject(clusterAdminClient, *clusterAdminClientConfig, testutil.Namespace(), user)
 	if err != nil {
 		t.Fatalf("error creating project: %v", err)
 	}
-	token, err := tokencmd.RequestToken(clusterAdminClientConfig, nil, adminUser, "password")
+	token, err := tokencmd.RequestToken(clusterAdminClientConfig, nil, user, "password")
 	if err != nil {
 		t.Fatalf("error requesting token: %v", err)
 	}
@@ -127,35 +127,20 @@ middleware:
 		t.Fatalf("error creating image stream: %s", err)
 	}
 
-	tags, err := getTags(stream.Name, adminUser, token)
+	tags, err := getTags(stream.Name, user, token)
 	if err != nil {
 		t.Fatal(err)
 	}
 	if len(tags) > 0 {
 		t.Fatalf("expected 0 tags, got: %#v", tags)
 	}
 
-	putUrl := fmt.Sprintf("http://127.0.0.1:5000/v2/%s/%s/manifests/%s", testutil.Namespace(), stream.Name, "latest")
-	signedManifest, dgst, err := signedManifest()
+	dgst, err := putManifest(stream.Name, user, token)
 	if err != nil {
 		t.Fatal(err)
 	}
-	req, err := http.NewRequest("PUT", putUrl, bytes.NewReader(signedManifest))
-	if err != nil {
-		t.Fatalf("error creating put request: %s", err)
-	}
-	req.SetBasicAuth(adminUser, token)
-	client := http.DefaultClient
-	resp, err := client.Do(req)
-	if err != nil {
-		t.Fatalf("error putting manifest: %s", err)
-	}
-	defer resp.Body.Close()
-	if resp.StatusCode != http.StatusAccepted {
-		t.Fatalf("unexpected put status code: %d", resp.StatusCode)
-	}
 
-	tags, err = getTags(stream.Name, adminUser, token)
+	tags, err = getTags(stream.Name, user, token)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -167,12 +152,12 @@ middleware:
 	}
 
 	url := fmt.Sprintf("http://127.0.0.1:5000/v2/%s/%s/manifests/%s", testutil.Namespace(), stream.Name, dgst.String())
-	req, err = http.NewRequest("GET", url, nil)
+	req, err := http.NewRequest("GET", url, nil)
 	if err != nil {
 		t.Fatalf("error creating request: %v", err)
 	}
-	req.SetBasicAuth(adminUser, token)
-	resp, err = client.Do(req)
+	req.SetBasicAuth(user, token)
+	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		t.Fatalf("error retrieving manifest from registry: %s", err)
 	}
@@ -185,7 +170,7 @@ middleware:
 	if err := json.Unmarshal(body, &retrievedManifest); err != nil {
 		t.Fatalf("error unmarshaling retrieved manifest")
 	}
-	if retrievedManifest.Name != "test-integration/test" {
+	if retrievedManifest.Name != fmt.Sprintf("%s/%s", testutil.Namespace(), stream.Name) {
 		t.Fatalf("unexpected manifest name: %s", retrievedManifest.Name)
 	}
 	if retrievedManifest.Tag != "latest" {
@@ -205,16 +190,72 @@ middleware:
 	if e, a := "foo", image.DockerImageMetadata.ID; e != a {
 		t.Errorf("image dockerImageMetadata.ID: expected %q, got %q", e, a)
 	}
+
+	// test auto provisioning
+	otherStream, err := adminClient.ImageStreams(testutil.Namespace()).Get("otherrepo")
+	t.Logf("otherStream=%#v, err=%v", otherStream, err)
+	if err == nil {
+		t.Fatalf("expected error getting otherrepo")
+	}
+
+	otherDigest, err := putManifest("otherrepo", user, token)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	otherStream, err = adminClient.ImageStreams(testutil.Namespace()).Get("otherrepo")
+	if err != nil {
+		t.Fatalf("unexpected error getting otherrepo: %s", err)
+	}
+	if otherStream == nil {
+		t.Fatalf("unexpected nil otherrepo")
+	}
+	if len(otherStream.Status.Tags) != 1 {
+		t.Errorf("expected 1 tag, got %#v", otherStream.Status.Tags)
+	}
+	history, ok := otherStream.Status.Tags["latest"]
+	if !ok {
+		t.Fatal("unable to find 'latest' tag")
+	}
+	if len(history.Items) != 1 {
+		t.Errorf("expected 1 tag event, got %#v", history.Items)
+	}
+	if e, a := otherDigest.String(), history.Items[0].Image; e != a {
+		t.Errorf("digest: expected %q, got %q", e, a)
+	}
+}
+
+func putManifest(name, user, token string) (digest.Digest, error) {
+	putUrl := fmt.Sprintf("http://127.0.0.1:5000/v2/%s/%s/manifests/%s", testutil.Namespace(), name, "latest")
+	signedManifest, dgst, err := signedManifest(fmt.Sprintf("%s/%s", testutil.Namespace(), name))
+	if err != nil {
+		return "", err
+	}
+	req, err := http.NewRequest("PUT", putUrl, bytes.NewReader(signedManifest))
+	if err != nil {
+		return "", fmt.Errorf("error creating put request: %s", err)
+	}
+	req.SetBasicAuth(user, token)
+	client := http.DefaultClient
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("error putting manifest: %s", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusAccepted {
+		return "", fmt.Errorf("unexpected put status code: %d", resp.StatusCode)
+	}
+	return dgst, nil
 }
 
-func getTags(streamName, adminUser, token string) ([]string, error) {
+func getTags(streamName, user, token string) ([]string, error) {
 	url := fmt.Sprintf("http://127.0.0.1:5000/v2/%s/%s/tags/list", testutil.Namespace(), streamName)
 	client := http.DefaultClient
 	req, err := http.NewRequest("GET", url, nil)
 	if err != nil {
 		return []string{}, fmt.Errorf("error creating request: %v", err)
 	}
-	req.SetBasicAuth(adminUser, token)
+	req.SetBasicAuth(user, token)
 	resp, err := client.Do(req)
 	if err != nil {
 		return []string{}, fmt.Errorf("error retrieving tags from registry: %s", err)