@@ -70,7 +70,7 @@ func (autoscalerStrategy) AllowCreateOnUpdate() bool {

 // PrepareForUpdate clears fields that are not allowed to be set by end users on update.

 func (autoscalerStrategy) PrepareForUpdate(obj, old runtime.Object) {

 	newHPA := obj.(*extensions.HorizontalPodAutoscaler)

-	oldHPA := obj.(*extensions.HorizontalPodAutoscaler)

+	oldHPA := old.(*extensions.HorizontalPodAutoscaler)

 	// Update is not allowed to set status

 	newHPA.Status = oldHPA.Status

 }

@@ -64,7 +64,7 @@ func (persistentvolumeStrategy) AllowCreateOnUpdate() bool {

 // PrepareForUpdate sets the Status fields which is not allowed to be set by an end user updating a PV

 func (persistentvolumeStrategy) PrepareForUpdate(obj, old runtime.Object) {

 	newPv := obj.(*api.PersistentVolume)

-	oldPv := obj.(*api.PersistentVolume)

+	oldPv := old.(*api.PersistentVolume)

 	newPv.Status = oldPv.Status

 }

@@ -86,7 +86,7 @@ var StatusStrategy = persistentvolumeStatusStrategy{Strategy}

 // PrepareForUpdate sets the Spec field which is not allowed to be changed when updating a PV's Status

 func (persistentvolumeStatusStrategy) PrepareForUpdate(obj, old runtime.Object) {

 	newPv := obj.(*api.PersistentVolume)

-	oldPv := obj.(*api.PersistentVolume)

+	oldPv := old.(*api.PersistentVolume)

 	newPv.Spec = oldPv.Spec

 }

@@ -64,7 +64,7 @@ func (persistentvolumeclaimStrategy) AllowCreateOnUpdate() bool {

 // PrepareForUpdate sets the Status field which is not allowed to be set by end users on update

 func (persistentvolumeclaimStrategy) PrepareForUpdate(obj, old runtime.Object) {

 	newPvc := obj.(*api.PersistentVolumeClaim)

-	oldPvc := obj.(*api.PersistentVolumeClaim)

+	oldPvc := old.(*api.PersistentVolumeClaim)

 	newPvc.Status = oldPvc.Status

 }

@@ -86,7 +86,7 @@ var StatusStrategy = persistentvolumeclaimStatusStrategy{Strategy}

 // PrepareForUpdate sets the Spec field which is not allowed to be changed when updating a PV's Status

 func (persistentvolumeclaimStatusStrategy) PrepareForUpdate(obj, old runtime.Object) {

 	newPv := obj.(*api.PersistentVolumeClaim)

-	oldPv := obj.(*api.PersistentVolumeClaim)

+	oldPv := old.(*api.PersistentVolumeClaim)

 	newPv.Spec = oldPv.Spec

 }

@@ -18,6 +18,7 @@ package storage

 import (

 	"fmt"

+	"net/http"

 	"reflect"

 	"strconv"

 	"strings"

@@ -25,8 +26,10 @@ import (

 	"time"

 	"k8s.io/kubernetes/pkg/api"

+	"k8s.io/kubernetes/pkg/api/errors"

 	"k8s.io/kubernetes/pkg/api/meta"

 	"k8s.io/kubernetes/pkg/api/rest"

+	"k8s.io/kubernetes/pkg/api/unversioned"

 	"k8s.io/kubernetes/pkg/client/cache"

 	"k8s.io/kubernetes/pkg/conversion"

 	"k8s.io/kubernetes/pkg/runtime"

@@ -264,7 +267,10 @@ func (c *Cacher) Watch(ctx context.Context, key string, resourceVersion string,

 	defer c.watchCache.RUnlock()

 	initEvents, err := c.watchCache.GetAllEventsSinceThreadUnsafe(watchRV)

 	if err != nil {

-		return nil, err

+		// To match the uncached watch implementation, once we have passed authn/authz/admission,

+		// and successfully parsed a resource version, other errors must fail with a watch event of type ERROR,

+		// rather than a directly returned error.

+		return newErrWatcher(err), nil

 	}

 	c.Lock()

@@ -460,6 +466,46 @@ func (lw *cacherListerWatcher) Watch(options api.ListOptions) (watch.Interface,

 	return lw.storage.WatchList(context.TODO(), lw.resourcePrefix, options.ResourceVersion, Everything)

 }

+// cacherWatch implements watch.Interface to return a single error

+type errWatcher struct {

+	result chan watch.Event

+}

+

+func newErrWatcher(err error) *errWatcher {

+	// Create an error event

+	errEvent := watch.Event{Type: watch.Error}

+	switch err := err.(type) {

+	case runtime.Object:

+		errEvent.Object = err

+	case *errors.StatusError:

+		errEvent.Object = &err.ErrStatus

+	default:

+		errEvent.Object = &unversioned.Status{

+			Status:  unversioned.StatusFailure,

+			Message: err.Error(),

+			Reason:  unversioned.StatusReasonInternalError,

+			Code:    http.StatusInternalServerError,

+		}

+	}

+

+	// Create a watcher with room for a single event, populate it, and close the channel

+	watcher := &errWatcher{result: make(chan watch.Event, 1)}

+	watcher.result <- errEvent

+	close(watcher.result)

+

+	return watcher

+}

+

+// Implements watch.Interface.

+func (c *errWatcher) ResultChan() <-chan watch.Event {

+	return c.result

+}

+

+// Implements watch.Interface.

+func (c *errWatcher) Stop() {

+	// no-op

+}

+

 // cacherWatch implements watch.Interface

 type cacheWatcher struct {

 	sync.Mutex

@@ -24,6 +24,7 @@ import (

 	"time"

 	"k8s.io/kubernetes/pkg/api"

+	"k8s.io/kubernetes/pkg/api/errors"

 	"k8s.io/kubernetes/pkg/api/meta"

 	"k8s.io/kubernetes/pkg/api/testapi"

 	apitesting "k8s.io/kubernetes/pkg/api/testing"

@@ -225,11 +226,15 @@ func TestWatch(t *testing.T) {

 	verifyWatchEvent(t, watcher, watch.Added, podFoo)

 	verifyWatchEvent(t, watcher, watch.Modified, podFooPrime)

-	// Check whether we get too-old error.

-	_, err = cacher.Watch(context.TODO(), "pods/ns/foo", "1", storage.Everything)

-	if err == nil {

-		t.Errorf("Expected 'error too old' error")

+	// Check whether we get too-old error via the watch channel

+	tooOldWatcher, err := cacher.Watch(context.TODO(), "pods/ns/foo", "1", storage.Everything)

+	if err != nil {

+		t.Fatalf("Expected no direct error, got %v", err)

 	}

+	defer tooOldWatcher.Stop()

+	// Ensure we get a "Gone" error

+	expectedGoneError := errors.NewGone("").(*errors.StatusError).ErrStatus

+	verifyWatchEvent(t, tooOldWatcher, watch.Error, &expectedGoneError)

 	initialWatcher, err := cacher.Watch(context.TODO(), "pods/ns/foo", fooCreated.ResourceVersion, storage.Everything)

 	if err != nil {

@@ -173,7 +173,7 @@ func importImages(ctx gocontext.Context, retriever RepositoryRetriever, isi *api

 			}

 			for _, index := range tags[j] {

 				if tag.Err != nil {

-					setImageImportStatus(isi, index, tag.Err)

+					setImageImportStatus(isi, index, tag.Name, tag.Err)

 					continue

 				}

 				copied := *tag.Image

@@ -194,7 +194,7 @@ func importImages(ctx gocontext.Context, retriever RepositoryRetriever, isi *api

 			}

 			for _, index := range ids[j] {

 				if digest.Err != nil {

-					setImageImportStatus(isi, index, digest.Err)

+					setImageImportStatus(isi, index, "", digest.Err)

 					continue

 				}

 				image := &isi.Status.Images[index]

@@ -267,6 +267,7 @@ func importFromRepository(ctx gocontext.Context, retriever RepositoryRetriever,

 	status.Status.Status = unversioned.StatusSuccess

 	status.Images = make([]api.ImageImportStatus, len(repo.Tags))

 	for i, tag := range repo.Tags {

+		status.Images[i].Tag = tag.Name

 		if tag.Err != nil {

 			failures++

 			status.Images[i].Status = imageImportStatus(tag.Err, "", "repository")

@@ -277,7 +278,6 @@ func importFromRepository(ctx gocontext.Context, retriever RepositoryRetriever,

 		copied := *tag.Image

 		ref.Tag, ref.ID = tag.Name, copied.Name

 		copied.DockerImageReference = ref.MostSpecific().Exact()

-		status.Images[i].Tag = tag.Name

 		status.Images[i].Image = &copied

 	}

 	if failures > 0 {

@@ -598,7 +598,8 @@ func imageImportStatus(err error, kind, position string) unversioned.Status {

 	}

 }

-func setImageImportStatus(images *api.ImageStreamImport, i int, err error) {

+func setImageImportStatus(images *api.ImageStreamImport, i int, tag string, err error) {

+	images.Status.Images[i].Tag = tag

 	images.Status.Images[i].Status = imageImportStatus(err, "", "")

 }

@@ -149,6 +149,12 @@ func TestImport(t *testing.T) {

 				if status := isi.Status.Images[3].Status; status.Status != "" {

 					t.Errorf("unexpected status: %#v", isi.Status.Images[3].Status)

 				}

+				expectedTags := []string{"latest", "", "", ""}

+				for i, image := range isi.Status.Images {

+					if image.Tag != expectedTags[i] {

+						t.Errorf("unexpected tag of status %d (%s != %s)", i, image.Tag, expectedTags[i])

+					}

+				}

 			},

 		},

 		{

@@ -186,6 +192,7 @@ func TestImport(t *testing.T) {

 				if len(isi.Status.Images) != 2 {

 					t.Errorf("unexpected number of images: %#v", isi.Status.Repository.Images)

 				}

+				expectedTags := []string{"", "tag"}

 				for i, image := range isi.Status.Images {

 					if image.Status.Status != unversioned.StatusSuccess {

 						t.Errorf("unexpected status %d: %#v", i, image.Status)

@@ -198,6 +205,9 @@ func TestImport(t *testing.T) {

 					if image.Image.DockerImageReference != "test@sha256:958608f8ecc1dc62c93b6c610f3a834dae4220c9642e6e8b4e0f2b3ad7cbd238" {

 						t.Errorf("unexpected ref %d: %#v", i, image.Image.DockerImageReference)

 					}

+					if image.Tag != expectedTags[i] {

+						t.Errorf("unexpected tag of status %d (%s != %s)", i, image.Tag, expectedTags[i])

+					}

 				}

 			},

 		},

@@ -222,10 +232,14 @@ func TestImport(t *testing.T) {

 				if len(isi.Status.Repository.Images) != 5 {

 					t.Errorf("unexpected number of images: %#v", isi.Status.Repository.Images)

 				}

+				expectedTags := []string{"3", "v2", "v1", "3.1", "abc"}

 				for i, image := range isi.Status.Repository.Images {

 					if image.Status.Status != unversioned.StatusFailure || image.Status.Message != "Internal error occurred: no such tag" {

 						t.Errorf("unexpected status %d: %#v", i, isi.Status.Repository.Images)

 					}

+					if image.Tag != expectedTags[i] {

+						t.Errorf("unexpected tag of status %d (%s != %s)", i, image.Tag, expectedTags[i])

+					}

 				}

 			},

 		},

@@ -177,6 +177,10 @@ func (r *REST) Create(ctx kapi.Context, obj runtime.Object) (runtime.Object, err

 	if spec := isi.Spec.Repository; spec != nil {

 		for i, status := range isi.Status.Repository.Images {

+			if checkImportFailure(status, stream, status.Tag, nextGeneration, now) {

+				continue

+			}

+

 			image := status.Image

 			ref, err := api.ParseDockerImageReference(image.DockerImageReference)

 			if err != nil {

@@ -196,10 +200,6 @@ func (r *REST) Create(ctx kapi.Context, obj runtime.Object) (runtime.Object, err

 			// we've imported a set of tags, ensure spec tag will point to this for later imports

 			from.ID, from.Tag = "", tag

-			if checkImportFailure(status, stream, tag, nextGeneration, now) {

-				continue

-			}

-

 			if updated, ok := r.importSuccessful(ctx, image, stream, tag, from.Exact(), nextGeneration, now, spec.ImportPolicy, importedImages, updatedImages); ok {

 				isi.Status.Repository.Images[i].Image = updated

 			}

@@ -278,6 +278,17 @@ func checkImportFailure(status api.ImageImportStatus, stream *api.ImageStream, t

 		LastTransitionTime: now,

 	}

+

+	if tag == "" {

+		if len(status.Tag) > 0 {

+			tag = status.Tag

+		} else if status.Image != nil {

+			if ref, err := api.ParseDockerImageReference(status.Image.DockerImageReference); err == nil {

+				tag = ref.Tag

+			}

+		}

+	}

+

 	if !api.HasTagCondition(stream, tag, condition) {

 		api.SetTagConditions(stream, tag, condition)

 		if tagRef, ok := stream.Spec.Tags[tag]; ok {

@@ -13,6 +13,7 @@ import (

 	"testing"

 	"time"

+	"github.com/docker/distribution/registry/api/errcode"

 	gocontext "golang.org/x/net/context"

 	kapi "k8s.io/kubernetes/pkg/api"

@@ -135,16 +136,26 @@ func TestImageStreamImport(t *testing.T) {

 	}

 }

-func mockRegistryHandler(t *testing.T, count *int) http.Handler {

+// mockRegistryHandler returns a registry mock handler with several repositories. requireAuth causes handler

+// to return unauthorized and request basic authentication header if not given. count is increased each

+// time the handler is invoked. There are three repositories:

+//  - test/image with phpManifest

+//  - test/image2 with etcdManifest

+//  - test/image3 with tags: v1, v2 and latest

+//    - the first points to etcdManifest

+//    - the others cause handler to return unknown error

+func mockRegistryHandler(t *testing.T, requireAuth bool, count *int) http.Handler {

 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

 		(*count)++

 		t.Logf("%d got %s %s", *count, r.Method, r.URL.Path)

 		w.Header().Set("Docker-Distribution-API-Version", "registry/2.0")

-		if len(r.Header.Get("Authorization")) == 0 {

-			w.Header().Set("WWW-Authenticate", "BASIC")

-			w.WriteHeader(http.StatusUnauthorized)

-			return

+		if requireAuth {

+			if len(r.Header.Get("Authorization")) == 0 {

+				w.Header().Set("WWW-Authenticate", "BASIC")

+				w.WriteHeader(http.StatusUnauthorized)

+				return

+			}

 		}

 		switch r.URL.Path {

@@ -154,6 +165,12 @@ func mockRegistryHandler(t *testing.T, count *int) http.Handler {

 			w.Write([]byte(phpManifest))

 		case "/v2/test/image2/manifests/" + etcdDigest:

 			w.Write([]byte(etcdManifest))

+		case "/v2/test/image3/tags/list":

+			w.Write([]byte("{\"name\": \"test/image3\", \"tags\": [\"latest\", \"v1\", \"v2\"]}"))

+		case "/v2/test/image3/manifests/latest", "/v2/test/image3/manifests/v2", "/v2/test/image3/manifests/" + danglingDigest:

+			errcode.ServeJSON(w, errcode.ErrorCodeUnknown)

+		case "/v2/test/image3/manifests/v1", "/v2/test/image3/manifests/" + etcdDigest:

+			w.Write([]byte(etcdManifest))

 		default:

 			t.Fatalf("unexpected request %s: %#v", r.URL.Path, r)

 		}

@@ -164,7 +181,7 @@ func TestImageStreamImportAuthenticated(t *testing.T) {

 	testutil.RequireEtcd(t)

 	// start regular HTTP servers

 	count := 0

-	server := httptest.NewServer(mockRegistryHandler(t, &count))

+	server := httptest.NewServer(mockRegistryHandler(t, true, &count))

 	server2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

 		w.Header().Set("Docker-Distribution-API-Version", "registry/2.0")

 		if len(r.Header.Get("Authorization")) == 0 {

@@ -177,7 +194,7 @@ func TestImageStreamImportAuthenticated(t *testing.T) {

 	// start a TLS server

 	count2 := 0

-	server3 := httptest.NewTLSServer(mockRegistryHandler(t, &count2))

+	server3 := httptest.NewTLSServer(mockRegistryHandler(t, true, &count2))

 	url1, _ := url.Parse(server.URL)

 	url2, _ := url.Parse(server2.URL)

@@ -324,6 +341,105 @@ func TestImageStreamImportAuthenticated(t *testing.T) {

 	}

 }

+// Verifies that individual errors for particular tags are handled properly when pulling all tags from a

+// repository.

+func TestImageStreamImportTagsFromRepository(t *testing.T) {

+	testutil.RequireEtcd(t)

+	// start regular HTTP servers

+	count := 0

+	server := httptest.NewServer(mockRegistryHandler(t, false, &count))

+

+	url, _ := url.Parse(server.URL)

+

+	// start a master

+	_, clusterAdminKubeConfig, err := testserver.StartTestMaster()

+	if err != nil {

+		t.Fatalf("unexpected error: %v", err)

+	}

+	/*

+		_, err := testutil.GetClusterAdminKubeClient(clusterAdminKubeConfig)

+		if err != nil {

+			t.Fatalf("unexpected error: %v", err)

+		}

+	*/

+	c, err := testutil.GetClusterAdminClient(clusterAdminKubeConfig)

+	if err != nil {

+		t.Fatalf("unexpected error: %v", err)

+	}

+	err = testutil.CreateNamespace(clusterAdminKubeConfig, testutil.Namespace())

+	if err != nil {

+		t.Fatalf("unexpected error: %v", err)

+	}

+

+	importSpec := &api.ImageStreamImport{

+		ObjectMeta: kapi.ObjectMeta{Name: "test"},

+		Spec: api.ImageStreamImportSpec{

+			Import: true,

+			Repository: &api.RepositoryImportSpec{

+				From:            kapi.ObjectReference{Kind: "DockerImage", Name: url.Host + "/test/image3"},

+				ImportPolicy:    api.TagImportPolicy{Insecure: true},

+				IncludeManifest: true,

+			},

+		},

+	}

+

+	// import expecting regular image to pass

+	isi, err := c.ImageStreams(testutil.Namespace()).Import(importSpec)

+	if err != nil {

+		t.Fatal(err)

+	}

+	if len(isi.Status.Images) != 0 {

+		t.Errorf("imported unexpected number of images (%d != 0)", len(isi.Status.Images))

+	}

+	if isi.Status.Repository == nil {

+		t.Fatalf("exported non-nil repository status")

+	}

+	if len(isi.Status.Repository.Images) != 3 {

+		t.Fatalf("imported unexpected number of tags (%d != 3)", len(isi.Status.Repository.Images))

+	}

+	for i, image := range isi.Status.Repository.Images {

+		switch i {

+		case 2:

+			if image.Status.Status != unversioned.StatusSuccess {

+				t.Errorf("import of image %d did not succeed: %#v", i, image.Status)

+			}

+			if image.Tag != "v1" {

+				t.Errorf("unexpected tag at position %d (%s != v1)", i, image.Tag)

+			}

+			if image.Image == nil {

+				t.Fatalf("expected image to be set")

+			}

+			if image.Image.DockerImageReference != url.Host+"/test/image3@"+etcdDigest {

+				t.Errorf("unexpected DockerImageReference (%s != %s)", image.Image.DockerImageReference, url.Host+"/test/image3@"+etcdDigest)

+			}

+			if image.Image.Name != etcdDigest {

+				t.Errorf("expected etcd digest as a name of the image (%s != %s)", image.Image.Name, etcdDigest)

+			}

+		default:

+			if image.Status.Status != unversioned.StatusFailure || image.Status.Reason != unversioned.StatusReasonInternalError {

+				t.Fatalf("import of image %d did not report internal server error: %#v", i, image.Status)

+			}

+			expectedTags := []string{"latest", "v2"}[i]

+			if image.Tag != expectedTags {

+				t.Errorf("unexpected tag at position %d (%s != %s)", i, image.Tag, expectedTags[i])

+			}

+		}

+	}

+

+	is, err := c.ImageStreams(testutil.Namespace()).Get("test")

+	if err != nil {

+		t.Fatal(err)

+	}

+	tagEvent := api.LatestTaggedImage(is, "v1")

+	if tagEvent == nil {

+		t.Fatalf("no image tagged for v1: %#v", is)

+	}

+

+	if tagEvent == nil || tagEvent.Image != etcdDigest || tagEvent.DockerImageReference != url.Host+"/test/image3@"+etcdDigest {

+		t.Fatalf("expected the etcd image to be tagged: %#v", tagEvent)

+	}

+}

+

 // Verifies that the import scheduler fetches an image repeatedly (every 1s as per the default

 // test controller interval), updates the image stream only when there are changes, and if an

 // error occurs writes the error only once (instead of every interval)

@@ -861,3 +977,5 @@ const phpManifest = `{

       }

    ]

 }`

+

+const danglingDigest = `sha256:f374c0d9b59e6fdf9f8922d59e946b05fbeabaed70b0639d7b6b524f3299e87b`