Merge pull request #16821 from miminar/image-pruning-dereference-istags

Automatic merge from submit-queue.

image-pruning: dereference ImageStreamTags

Create strong references to images for each pod/bc/dc/etc that uses `<host>/<repo>:tag` reference.

Resolves [bz#1498604](https://bugzilla.redhat.com/show_bug.cgi?id=1498604) and https://bugzilla.redhat.com/show_bug.cgi?id=1386917



Images can manually removed using `oc delete`. Image stream tags having references to these images become obsolete - we may delete them.

To be sure that we don't remove reference to image that has just been created (and we don't know about it), make sure to honor `--keep-younger-than`.

Author: openshift-merge-robot

pkg/apps/graph/nodes/nodes.go

@@ -3,13 +3,69 @@ package nodes
 import (
 	"github.com/gonum/graph"
 
+	kapisext "k8s.io/kubernetes/pkg/apis/extensions"
+
 	osgraph "github.com/openshift/origin/pkg/api/graph"
 	kubegraph "github.com/openshift/origin/pkg/api/kubegraph/nodes"
-	depoyapi "github.com/openshift/origin/pkg/apps/apis/apps"
+	deployapi "github.com/openshift/origin/pkg/apps/apis/apps"
 )
 
+// EnsureDaemonSetNode adds the provided daemon set to the graph if it does not exist
+func EnsureDaemonSetNode(g osgraph.MutableUniqueGraph, ds *kapisext.DaemonSet) *DaemonSetNode {
+	dsName := DaemonSetNodeName(ds)
+	dsNode := osgraph.EnsureUnique(
+		g,
+		dsName,
+		func(node osgraph.Node) graph.Node {
+			return &DaemonSetNode{Node: node, DaemonSet: ds, IsFound: true}
+		},
+	).(*DaemonSetNode)
+
+	podTemplateSpecNode := kubegraph.EnsurePodTemplateSpecNode(g, &ds.Spec.Template, ds.Namespace, dsName)
+	g.AddEdge(dsNode, podTemplateSpecNode, osgraph.ContainsEdgeKind)
+
+	return dsNode
+}
+
+func FindOrCreateSyntheticDaemonSetNode(g osgraph.MutableUniqueGraph, ds *kapisext.DaemonSet) *DaemonSetNode {
+	return osgraph.EnsureUnique(
+		g,
+		DaemonSetNodeName(ds),
+		func(node osgraph.Node) graph.Node {
+			return &DaemonSetNode{Node: node, DaemonSet: ds, IsFound: false}
+		},
+	).(*DaemonSetNode)
+}
+
+// EnsureDeploymentNode adds the provided upstream deployment to the graph if it does not exist
+func EnsureDeploymentNode(g osgraph.MutableUniqueGraph, deployment *kapisext.Deployment) *DeploymentNode {
+	deploymentName := DeploymentNodeName(deployment)
+	deploymentNode := osgraph.EnsureUnique(
+		g,
+		deploymentName,
+		func(node osgraph.Node) graph.Node {
+			return &DeploymentNode{Node: node, Deployment: deployment, IsFound: true}
+		},
+	).(*DeploymentNode)
+
+	podTemplateSpecNode := kubegraph.EnsurePodTemplateSpecNode(g, &deployment.Spec.Template, deployment.Namespace, deploymentName)
+	g.AddEdge(deploymentNode, podTemplateSpecNode, osgraph.ContainsEdgeKind)
+
+	return deploymentNode
+}
+
+func FindOrCreateSyntheticDeploymentNode(g osgraph.MutableUniqueGraph, deployment *kapisext.Deployment) *DeploymentNode {
+	return osgraph.EnsureUnique(
+		g,
+		DeploymentNodeName(deployment),
+		func(node osgraph.Node) graph.Node {
+			return &DeploymentNode{Node: node, Deployment: deployment, IsFound: false}
+		},
+	).(*DeploymentNode)
+}
+
 // EnsureDeploymentConfigNode adds the provided deployment config to the graph if it does not exist
-func EnsureDeploymentConfigNode(g osgraph.MutableUniqueGraph, dc *depoyapi.DeploymentConfig) *DeploymentConfigNode {
+func EnsureDeploymentConfigNode(g osgraph.MutableUniqueGraph, dc *deployapi.DeploymentConfig) *DeploymentConfigNode {
 	dcName := DeploymentConfigNodeName(dc)
 	dcNode := osgraph.EnsureUnique(
 		g,
@@ -27,7 +83,7 @@ func EnsureDeploymentConfigNode(g osgraph.MutableUniqueGraph, dc *depoyapi.Deplo
 	return dcNode
 }
 
-func FindOrCreateSyntheticDeploymentConfigNode(g osgraph.MutableUniqueGraph, dc *depoyapi.DeploymentConfig) *DeploymentConfigNode {
+func FindOrCreateSyntheticDeploymentConfigNode(g osgraph.MutableUniqueGraph, dc *deployapi.DeploymentConfig) *DeploymentConfigNode {
 	return osgraph.EnsureUnique(
 		g,
 		DeploymentConfigNodeName(dc),
@@ -36,3 +92,30 @@ func FindOrCreateSyntheticDeploymentConfigNode(g osgraph.MutableUniqueGraph, dc
 		},
 	).(*DeploymentConfigNode)
 }
+
+// EnsureReplicaSetNode adds the provided replica set to the graph if it does not exist
+func EnsureReplicaSetNode(g osgraph.MutableUniqueGraph, rs *kapisext.ReplicaSet) *ReplicaSetNode {
+	rsName := ReplicaSetNodeName(rs)
+	rsNode := osgraph.EnsureUnique(
+		g,
+		rsName,
+		func(node osgraph.Node) graph.Node {
+			return &ReplicaSetNode{Node: node, ReplicaSet: rs, IsFound: true}
+		},
+	).(*ReplicaSetNode)
+
+	podTemplateSpecNode := kubegraph.EnsurePodTemplateSpecNode(g, &rs.Spec.Template, rs.Namespace, rsName)
+	g.AddEdge(rsNode, podTemplateSpecNode, osgraph.ContainsEdgeKind)
+
+	return rsNode
+}
+
+func FindOrCreateSyntheticReplicaSetNode(g osgraph.MutableUniqueGraph, rs *kapisext.ReplicaSet) *ReplicaSetNode {
+	return osgraph.EnsureUnique(
+		g,
+		ReplicaSetNodeName(rs),
+		func(node osgraph.Node) graph.Node {
+			return &ReplicaSetNode{Node: node, ReplicaSet: rs, IsFound: false}
+		},
+	).(*ReplicaSetNode)
+}

pkg/apps/graph/nodes/types.go

@@ -3,14 +3,73 @@ package nodes
 import (
 	"reflect"
 
+	kapisext "k8s.io/kubernetes/pkg/apis/extensions"
+
 	osgraph "github.com/openshift/origin/pkg/api/graph"
 	deployapi "github.com/openshift/origin/pkg/apps/apis/apps"
 )
 
 var (
+	DaemonSetNodeKind        = reflect.TypeOf(kapisext.DaemonSet{}).Name()
+	DeploymentNodeKind       = reflect.TypeOf(kapisext.Deployment{}).Name()
 	DeploymentConfigNodeKind = reflect.TypeOf(deployapi.DeploymentConfig{}).Name()
+	ReplicaSetNodeKind       = reflect.TypeOf(kapisext.ReplicaSet{}).Name()
 )
 
+func DaemonSetNodeName(o *kapisext.DaemonSet) osgraph.UniqueName {
+	return osgraph.GetUniqueRuntimeObjectNodeName(DaemonSetNodeKind, o)
+}
+
+type DaemonSetNode struct {
+	osgraph.Node
+	DaemonSet *kapisext.DaemonSet
+
+	IsFound bool
+}
+
+func (n DaemonSetNode) Found() bool {
+	return n.IsFound
+}
+
+func (n DaemonSetNode) Object() interface{} {
+	return n.DaemonSet
+}
+
+func (n DaemonSetNode) String() string {
+	return string(DaemonSetNodeName(n.DaemonSet))
+}
+
+func (*DaemonSetNode) Kind() string {
+	return DaemonSetNodeKind
+}
+
+func DeploymentNodeName(o *kapisext.Deployment) osgraph.UniqueName {
+	return osgraph.GetUniqueRuntimeObjectNodeName(DeploymentNodeKind, o)
+}
+
+type DeploymentNode struct {
+	osgraph.Node
+	Deployment *kapisext.Deployment
+
+	IsFound bool
+}
+
+func (n DeploymentNode) Found() bool {
+	return n.IsFound
+}
+
+func (n DeploymentNode) Object() interface{} {
+	return n.Deployment
+}
+
+func (n DeploymentNode) String() string {
+	return string(DeploymentNodeName(n.Deployment))
+}
+
+func (*DeploymentNode) Kind() string {
+	return DeploymentNodeKind
+}
+
 func DeploymentConfigNodeName(o *deployapi.DeploymentConfig) osgraph.UniqueName {
 	return osgraph.GetUniqueRuntimeObjectNodeName(DeploymentConfigNodeKind, o)
 }
@@ -37,3 +96,30 @@ func (n DeploymentConfigNode) String() string {
 func (*DeploymentConfigNode) Kind() string {
 	return DeploymentConfigNodeKind
 }
+
+func ReplicaSetNodeName(o *kapisext.ReplicaSet) osgraph.UniqueName {
+	return osgraph.GetUniqueRuntimeObjectNodeName(ReplicaSetNodeKind, o)
+}
+
+type ReplicaSetNode struct {
+	osgraph.Node
+	ReplicaSet *kapisext.ReplicaSet
+
+	IsFound bool
+}
+
+func (n ReplicaSetNode) Found() bool {
+	return n.IsFound
+}
+
+func (n ReplicaSetNode) Object() interface{} {
+	return n.ReplicaSet
+}
+
+func (n ReplicaSetNode) String() string {
+	return string(ReplicaSetNodeName(n.ReplicaSet))
+}
+
+func (*ReplicaSetNode) Kind() string {
+	return ReplicaSetNodeKind
+}

pkg/cmd/server/bootstrappolicy/policy.go

@@ -551,6 +551,9 @@ func GetOpenshiftBootstrapClusterRoles() []rbac.ClusterRole {
 				rbac.NewRule("list").Groups(kapiGroup).Resources("limitranges").RuleOrDie(),
 				rbac.NewRule("get", "list").Groups(buildGroup, legacyBuildGroup).Resources("buildconfigs", "builds").RuleOrDie(),
 				rbac.NewRule("get", "list").Groups(deployGroup, legacyDeployGroup).Resources("deploymentconfigs").RuleOrDie(),
+				rbac.NewRule("get", "list").Groups(extensionsGroup).Resources("daemonsets").RuleOrDie(),
+				rbac.NewRule("get", "list").Groups(extensionsGroup).Resources("deployments").RuleOrDie(),
+				rbac.NewRule("get", "list").Groups(extensionsGroup).Resources("replicasets").RuleOrDie(),
 
 				rbac.NewRule("delete").Groups(imageGroup, legacyImageGroup).Resources("images").RuleOrDie(),
 				rbac.NewRule("get", "list").Groups(imageGroup, legacyImageGroup).Resources("images", "imagestreams").RuleOrDie(),

pkg/image/graph/nodes/nodes.go

@@ -36,8 +36,12 @@ func EnsureAllImageStreamTagNodes(g osgraph.MutableUniqueGraph, is *imageapi.Ima
 	return ret
 }
 
-func FindImage(g osgraph.MutableUniqueGraph, imageName string) graph.Node {
-	return g.Find(ImageNodeName(&imageapi.Image{ObjectMeta: metav1.ObjectMeta{Name: imageName}}))
+func FindImage(g osgraph.MutableUniqueGraph, imageName string) *ImageNode {
+	n := g.Find(ImageNodeName(&imageapi.Image{ObjectMeta: metav1.ObjectMeta{Name: imageName}}))
+	if imageNode, ok := n.(*ImageNode); ok {
+		return imageNode
+	}
+	return nil
 }
 
 // EnsureDockerRepositoryNode adds the named Docker repository tag reference to the graph if it does

pkg/image/prune/helper.go

@@ -7,6 +7,8 @@ import (
 	"sort"
 	"strings"
 
+	kapi "k8s.io/kubernetes/pkg/api"
+
 	"github.com/docker/distribution/registry/api/errcode"
 	"github.com/golang/glog"
 
@@ -206,3 +208,60 @@ func TryProtocolsWithRegistryURL(registry string, allowInsecure bool, action fun
 type retryPath struct{ err error }
 
 func (rp *retryPath) Error() string { return rp.err.Error() }
+
+// ErrBadReference denotes an invalid reference to image, imagestreamtag or imagestreamimage stored in a
+// particular object. The object is identified by kind, namespace and name.
+type ErrBadReference struct {
+	kind       string
+	namespace  string
+	name       string
+	targetKind string
+	reference  string
+	reason     string
+}
+
+func newErrBadReferenceToImage(reference string, obj *kapi.ObjectReference, reason string) error {
+	kind := "<UnknownType>"
+	namespace := ""
+	name := "<unknown-name>"
+	if obj != nil {
+		kind = obj.Kind
+		namespace = obj.Namespace
+		name = obj.Name
+	}
+
+	return &ErrBadReference{
+		kind:      kind,
+		namespace: namespace,
+		name:      name,
+		reference: reference,
+		reason:    reason,
+	}
+}
+
+func newErrBadReferenceTo(targetKind, reference string, obj *kapi.ObjectReference, reason string) error {
+	return &ErrBadReference{
+		kind:       obj.Kind,
+		namespace:  obj.Namespace,
+		name:       obj.Name,
+		targetKind: targetKind,
+		reference:  reference,
+		reason:     reason,
+	}
+}
+
+func (e *ErrBadReference) Error() string {
+	return e.String()
+}
+
+func (e *ErrBadReference) String() string {
+	name := e.name
+	if len(e.namespace) > 0 {
+		name = e.namespace + "/" + name
+	}
+	targetKind := "docker image"
+	if len(e.targetKind) > 0 {
+		targetKind = e.targetKind
+	}
+	return fmt.Sprintf("%s[%s]: invalid %s reference %q: %s", e.kind, name, targetKind, e.reference, e.reason)
+}

pkg/image/prune/prune.go

@@ -2,6 +2,7 @@ package prune
 
 import (
 	"crypto/x509"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -15,9 +16,14 @@ import (
 	"time"
 
 	"github.com/spf13/cobra"
+
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	kutilerrors "k8s.io/apimachinery/pkg/util/errors"
 	knet "k8s.io/apimachinery/pkg/util/net"
+	discovery "k8s.io/client-go/discovery"
 	restclient "k8s.io/client-go/rest"
+	kclientcmd "k8s.io/client-go/tools/clientcmd"
 	kapi "k8s.io/kubernetes/pkg/api"
 	kclientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	"k8s.io/kubernetes/pkg/kubectl/cmd/templates"
@@ -32,6 +38,7 @@ import (
 	"github.com/openshift/origin/pkg/image/prune"
 	oserrors "github.com/openshift/origin/pkg/util/errors"
 	"github.com/openshift/origin/pkg/util/netutils"
+	"github.com/openshift/origin/pkg/version"
 )
 
 // PruneImagesRecommendedName is the recommended command name
@@ -72,8 +79,8 @@ var (
 			--insecure-skip-tls-verify or allowed for insecure connection)`)
 
 	imagesExample = templates.Examples(`
-	  # See, what the prune command would delete if only images more than an hour old and obsoleted
-	  # by 3 newer revisions under the same tag were considered.
+	  # See, what the prune command would delete if only images and their referrers were more than an hour old
+	  # and obsoleted by 3 newer revisions under the same tag were considered.
 	  %[1]s %[2]s --keep-tag-revisions=3 --keep-younger-than=60m
 
 	  # To actually perform the prune operation, the confirm flag must be appended
@@ -111,12 +118,15 @@ type PruneImagesOptions struct {
 	Namespace           string
 	ForceInsecure       bool
 
-	ClientConfig *restclient.Config
-	AppsClient   appsclient.AppsInterface
-	BuildClient  buildclient.BuildInterface
-	ImageClient  imageclient.ImageInterface
-	KubeClient   kclientset.Interface
-	Out          io.Writer
+	ClientConfig    *restclient.Config
+	AppsClient      appsclient.AppsInterface
+	BuildClient     buildclient.BuildInterface
+	ImageClient     imageclient.ImageInterface
+	DiscoveryClient discovery.DiscoveryInterface
+	KubeClient      kclientset.Interface
+	Timeout         time.Duration
+	Out             io.Writer
+	ErrOut          io.Writer
 }
 
 // NewCmdPruneImages implements the OpenShift cli prune images command.
@@ -146,7 +156,7 @@ func NewCmdPruneImages(f *clientcmd.Factory, parentName, name string, out io.Wri
 
 	cmd.Flags().BoolVar(&opts.Confirm, "confirm", opts.Confirm, "If true, specify that image pruning should proceed. Defaults to false, displaying what would be deleted but not actually deleting anything. Requires a valid route to the integrated Docker registry (see --registry-url).")
 	cmd.Flags().BoolVar(opts.AllImages, "all", *opts.AllImages, "Include images that were imported from external registries as candidates for pruning.  If pruned, all the mirrored objects associated with them will also be removed from the integrated registry.")
-	cmd.Flags().DurationVar(opts.KeepYoungerThan, "keep-younger-than", *opts.KeepYoungerThan, "Specify the minimum age of an image for it to be considered a candidate for pruning.")
+	cmd.Flags().DurationVar(opts.KeepYoungerThan, "keep-younger-than", *opts.KeepYoungerThan, "Specify the minimum age of an image and its referrers for it to be considered a candidate for pruning.")
 	cmd.Flags().IntVar(opts.KeepTagRevisions, "keep-tag-revisions", *opts.KeepTagRevisions, "Specify the number of image revisions for a tag in an image stream that will be preserved.")
 	cmd.Flags().BoolVar(opts.PruneOverSizeLimit, "prune-over-size-limit", *opts.PruneOverSizeLimit, "Specify if images which are exceeding LimitRanges (see 'openshift.io/Image'), specified in the same namespace, should be considered for pruning. This flag cannot be combined with --keep-younger-than nor --keep-tag-revisions.")
 	cmd.Flags().StringVar(&opts.CABundle, "certificate-authority", opts.CABundle, "The path to a certificate authority bundle to use when communicating with the managed Docker registries. Defaults to the certificate authority data from the current user's config file. It cannot be used together with --force-insecure.")
@@ -183,13 +193,13 @@ func (o *PruneImagesOptions) Complete(f *clientcmd.Factory, cmd *cobra.Command,
 		}
 	}
 	o.Out = out
+	o.ErrOut = os.Stderr
 
 	clientConfig, err := f.ClientConfig()
 	if err != nil {
 		return err
 	}
 	o.ClientConfig = clientConfig
-
 	appsClient, buildClient, imageClient, kubeClient, err := getClients(f)
 	if err != nil {
 		return err
@@ -198,6 +208,12 @@ func (o *PruneImagesOptions) Complete(f *clientcmd.Factory, cmd *cobra.Command,
 	o.BuildClient = buildClient
 	o.ImageClient = imageClient
 	o.KubeClient = kubeClient
+	o.DiscoveryClient = kubeClient.Discovery()
+
+	o.Timeout = clientConfig.Timeout
+	if o.Timeout == 0 {
+		o.Timeout = time.Duration(10 * time.Second)
+	}
 
 	return nil
 }
@@ -261,11 +277,38 @@ func (o PruneImagesOptions) Run() error {
 		return err
 	}
 
+	allDSs, err := o.KubeClient.Extensions().DaemonSets(o.Namespace).List(metav1.ListOptions{})
+	if err != nil {
+		// TODO: remove in future (3.9) release
+		if !kerrors.IsForbidden(err) {
+			return err
+		}
+		fmt.Fprintf(o.ErrOut, "Failed to list daemonsets: %v\n - * Make sure to update clusterRoleBindings.\n", err)
+	}
+
+	allDeployments, err := o.KubeClient.Extensions().Deployments(o.Namespace).List(metav1.ListOptions{})
+	if err != nil {
+		// TODO: remove in future (3.9) release
+		if !kerrors.IsForbidden(err) {
+			return err
+		}
+		fmt.Fprintf(o.ErrOut, "Failed to list deployments: %v\n - * Make sure to update clusterRoleBindings.\n", err)
+	}
+
 	allDCs, err := o.AppsClient.DeploymentConfigs(o.Namespace).List(metav1.ListOptions{})
 	if err != nil {
 		return err
 	}
 
+	allRSs, err := o.KubeClient.Extensions().ReplicaSets(o.Namespace).List(metav1.ListOptions{})
+	if err != nil {
+		// TODO: remove in future (3.9) release
+		if !kerrors.IsForbidden(err) {
+			return err
+		}
+		fmt.Fprintf(o.ErrOut, "Failed to list replicasets: %v\n - * Make sure to update clusterRoleBindings.\n", err)
+	}
+
 	limitRangesList, err := o.KubeClient.Core().LimitRanges(o.Namespace).List(metav1.ListOptions{})
 	if err != nil {
 		return err
@@ -332,7 +375,10 @@ func (o PruneImagesOptions) Run() error {
 		RCs:                allRCs,
 		BCs:                allBCs,
 		Builds:             allBuilds,
+		DSs:                allDSs,
+		Deployments:        allDeployments,
 		DCs:                allDCs,
+		RSs:                allRSs,
 		LimitRanges:        limitRangesMap,
 		DryRun:             o.Confirm == false,
 		RegistryClient:     registryClient,
@@ -341,19 +387,20 @@ func (o PruneImagesOptions) Run() error {
 	if o.Namespace != metav1.NamespaceAll {
 		options.Namespace = o.Namespace
 	}
-	pruner, err := prune.NewPruner(options)
-	if err != nil {
-		return err
+	pruner, errs := prune.NewPruner(options)
+	if errs != nil {
+		o.printGraphBuildErrors(errs)
+		return fmt.Errorf("failed to build graph - no changes made")
 	}
 
 	w := tabwriter.NewWriter(o.Out, 10, 4, 3, ' ', 0)
 	defer w.Flush()
 
-	imageDeleter := &describingImageDeleter{w: w}
-	imageStreamDeleter := &describingImageStreamDeleter{w: w}
-	layerLinkDeleter := &describingLayerLinkDeleter{w: w}
-	blobDeleter := &describingBlobDeleter{w: w}
-	manifestDeleter := &describingManifestDeleter{w: w}
+	imageDeleter := &describingImageDeleter{w: w, errOut: o.ErrOut}
+	imageStreamDeleter := &describingImageStreamDeleter{w: w, errOut: o.ErrOut}
+	layerLinkDeleter := &describingLayerLinkDeleter{w: w, errOut: o.ErrOut}
+	blobDeleter := &describingBlobDeleter{w: w, errOut: o.ErrOut}
+	manifestDeleter := &describingManifestDeleter{w: w, errOut: o.ErrOut}
 
 	if o.Confirm {
 		imageDeleter.delegate = prune.NewImageDeleter(o.ImageClient)
@@ -362,18 +409,49 @@ func (o PruneImagesOptions) Run() error {
 		blobDeleter.delegate = prune.NewBlobDeleter()
 		manifestDeleter.delegate = prune.NewManifestDeleter()
 	} else {
-		fmt.Fprintln(os.Stderr, "Dry run enabled - no modifications will be made. Add --confirm to remove images")
+		fmt.Fprintln(o.ErrOut, "Dry run enabled - no modifications will be made. Add --confirm to remove images")
 	}
 
 	return pruner.Prune(imageDeleter, imageStreamDeleter, layerLinkDeleter, blobDeleter, manifestDeleter)
 }
 
+func (o *PruneImagesOptions) printGraphBuildErrors(errs kutilerrors.Aggregate) {
+	refErrors := []error{}
+
+	fmt.Fprintf(o.ErrOut, "Failed to build graph!\n")
+
+	for _, err := range errs.Errors() {
+		if _, ok := err.(*prune.ErrBadReference); ok {
+			refErrors = append(refErrors, err)
+		} else {
+			fmt.Fprintf(o.ErrOut, "%v\n", err)
+		}
+	}
+
+	if len(refErrors) > 0 {
+		clientVersion, masterVersion, err := getClientAndMasterVersions(o.DiscoveryClient, o.Timeout)
+		if err != nil {
+			fmt.Fprintf(o.ErrOut, "Failed to get master API version: %v\n", err)
+		}
+		fmt.Fprintf(o.ErrOut, "\nThe following objects have invalid references:\n\n")
+		for _, err := range refErrors {
+			fmt.Fprintf(o.ErrOut, "  %s\n", err)
+		}
+		fmt.Fprintf(o.ErrOut, "\nEither fix the references or delete the objects to make the pruner proceed.\n")
+
+		if masterVersion != nil && (clientVersion.Major != masterVersion.Major || clientVersion.Minor != masterVersion.Minor) {
+			fmt.Fprintf(o.ErrOut, "Client version (%s) doesn't match master (%s), which may allow for different image references. Try to re-run this binary with the same version.\n", clientVersion, masterVersion)
+		}
+	}
+}
+
 // describingImageStreamDeleter prints information about each image stream update.
 // If a delegate exists, its DeleteImageStream function is invoked prior to returning.
 type describingImageStreamDeleter struct {
 	w             io.Writer
 	delegate      prune.ImageStreamDeleter
 	headerPrinted bool
+	errOut        io.Writer
 }
 
 var _ prune.ImageStreamDeleter = &describingImageStreamDeleter{}
@@ -389,7 +467,7 @@ func (p *describingImageStreamDeleter) UpdateImageStream(stream *imageapi.ImageS
 
 	updatedStream, err := p.delegate.UpdateImageStream(stream)
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "error updating image stream %s/%s to remove image references: %v\n", stream.Namespace, stream.Name, err)
+		fmt.Fprintf(p.errOut, "error updating image stream %s/%s to remove image references: %v\n", stream.Namespace, stream.Name, err)
 	}
 
 	return updatedStream, err
@@ -416,6 +494,7 @@ type describingImageDeleter struct {
 	w             io.Writer
 	delegate      prune.ImageDeleter
 	headerPrinted bool
+	errOut        io.Writer
 }
 
 var _ prune.ImageDeleter = &describingImageDeleter{}
@@ -435,7 +514,7 @@ func (p *describingImageDeleter) DeleteImage(image *imageapi.Image) error {
 
 	err := p.delegate.DeleteImage(image)
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "error deleting image %s from server: %v\n", image.Name, err)
+		fmt.Fprintf(p.errOut, "error deleting image %s from server: %v\n", image.Name, err)
 	}
 
 	return err
@@ -447,6 +526,7 @@ type describingLayerLinkDeleter struct {
 	w             io.Writer
 	delegate      prune.LayerLinkDeleter
 	headerPrinted bool
+	errOut        io.Writer
 }
 
 var _ prune.LayerLinkDeleter = &describingLayerLinkDeleter{}
@@ -466,7 +546,7 @@ func (p *describingLayerLinkDeleter) DeleteLayerLink(registryClient *http.Client
 
 	err := p.delegate.DeleteLayerLink(registryClient, registryURL, repo, name)
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "error deleting repository %s layer link %s from the registry: %v\n", repo, name, err)
+		fmt.Fprintf(p.errOut, "error deleting repository %s layer link %s from the registry: %v\n", repo, name, err)
 	}
 
 	return err
@@ -478,6 +558,7 @@ type describingBlobDeleter struct {
 	w             io.Writer
 	delegate      prune.BlobDeleter
 	headerPrinted bool
+	errOut        io.Writer
 }
 
 var _ prune.BlobDeleter = &describingBlobDeleter{}
@@ -497,7 +578,7 @@ func (p *describingBlobDeleter) DeleteBlob(registryClient *http.Client, registry
 
 	err := p.delegate.DeleteBlob(registryClient, registryURL, layer)
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "error deleting blob %s from the registry: %v\n", layer, err)
+		fmt.Fprintf(p.errOut, "error deleting blob %s from the registry: %v\n", layer, err)
 	}
 
 	return err
@@ -510,6 +591,7 @@ type describingManifestDeleter struct {
 	w             io.Writer
 	delegate      prune.ManifestDeleter
 	headerPrinted bool
+	errOut        io.Writer
 }
 
 var _ prune.ManifestDeleter = &describingManifestDeleter{}
@@ -529,7 +611,7 @@ func (p *describingManifestDeleter) DeleteManifest(registryClient *http.Client,
 
 	err := p.delegate.DeleteManifest(registryClient, registryURL, repo, manifest)
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "error deleting manifest %s from repository %s: %v\n", manifest, repo, err)
+		fmt.Fprintf(p.errOut, "error deleting manifest %s from repository %s: %v\n", manifest, repo, err)
 	}
 
 	return err
@@ -643,3 +725,48 @@ func getRegistryClient(clientConfig *restclient.Config, registryCABundle string,
 		Transport: wrappedTransport,
 	}, nil
 }
+
+// getClientAndMasterVersions returns version info for client and master binaries. If it takes too long to get
+// a response from the master, timeout error is returned.
+func getClientAndMasterVersions(client discovery.DiscoveryInterface, timeout time.Duration) (clientVersion, masterVersion *version.Info, err error) {
+	done := make(chan error)
+
+	go func() {
+		defer close(done)
+
+		ocVersionBody, err := client.RESTClient().Get().AbsPath("/version/openshift").Do().Raw()
+		switch {
+		case err == nil:
+			var ocServerInfo version.Info
+			err = json.Unmarshal(ocVersionBody, &ocServerInfo)
+			if err != nil && len(ocVersionBody) > 0 {
+				done <- err
+				return
+			}
+			masterVersion = &ocServerInfo
+
+		case kerrors.IsNotFound(err) || kerrors.IsUnauthorized(err) || kerrors.IsForbidden(err):
+		default:
+			done <- err
+			return
+		}
+	}()
+
+	select {
+	case err, closed := <-done:
+		if strings.HasSuffix(fmt.Sprintf("%v", err), "connection refused") || clientcmd.IsConfigurationMissing(err) || kclientcmd.IsConfigurationInvalid(err) {
+			return nil, nil, err
+		}
+		if closed && err != nil {
+			return nil, nil, err
+		}
+	// do not block error printing if the master is busy
+	case <-time.After(timeout):
+		return nil, nil, fmt.Errorf("error: server took too long to respond with version information.")
+	}
+
+	v := version.Get()
+	clientVersion = &v
+
+	return
+}

pkg/image/prune/prune_test.go

@@ -1,17 +1,39 @@
 package prune
 
 import (
+	"bytes"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
 	"io/ioutil"
+	"net/http"
+	"os"
+	"strings"
 	"testing"
 
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/sets"
+	fakediscovery "k8s.io/client-go/discovery/fake"
+	"k8s.io/client-go/kubernetes/scheme"
+	restclient "k8s.io/client-go/rest"
+	kapi "k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake"
+	"k8s.io/kubernetes/staging/src/k8s.io/apimachinery/pkg/util/diff"
+
+	restfake "k8s.io/client-go/rest/fake"
 
 	appsclient "github.com/openshift/origin/pkg/apps/generated/internalclientset/fake"
 	buildclient "github.com/openshift/origin/pkg/build/generated/internalclientset/fake"
 	imageclient "github.com/openshift/origin/pkg/image/generated/internalclientset/fake"
+	"github.com/openshift/origin/pkg/image/prune/testutil"
+	"github.com/openshift/origin/pkg/version"
 )
 
+var logLevel = flag.Int("loglevel", 0, "")
+
 func TestImagePruneNamespaced(t *testing.T) {
+	flag.Lookup("v").Value.Set(fmt.Sprint(*logLevel))
 	kFake := fake.NewSimpleClientset()
 	imageFake := imageclient.NewSimpleClientset()
 	opts := &PruneImagesOptions{
@@ -22,6 +44,7 @@ func TestImagePruneNamespaced(t *testing.T) {
 		ImageClient: imageFake.Image(),
 		KubeClient:  kFake,
 		Out:         ioutil.Discard,
+		ErrOut:      os.Stderr,
 	}
 
 	if err := opts.Run(); err != nil {
@@ -46,3 +69,119 @@ func TestImagePruneNamespaced(t *testing.T) {
 		}
 	}
 }
+
+func TestImagePruneErrOnBadReference(t *testing.T) {
+	flag.Lookup("v").Value.Set(fmt.Sprint(*logLevel))
+	podBad := testutil.Pod("foo", "pod1", kapi.PodRunning, "invalid image reference")
+	podGood := testutil.Pod("foo", "pod2", kapi.PodRunning, "example.com/foo/bar@sha256:0000000000000000000000000000000000000000000000000000000000000000")
+	dep := testutil.Deployment("foo", "dep1", "do not blame me")
+	bcBad := testutil.BC("foo", "bc1", "source", "ImageStreamImage", "foo", "bar:invalid-digest")
+
+	kFake := fake.NewSimpleClientset(&podBad, &podGood, &dep)
+	imageFake := imageclient.NewSimpleClientset()
+	fakeDiscovery := &fakeVersionDiscovery{
+		masterVersion: version.Get(),
+	}
+
+	switch d := kFake.Discovery().(type) {
+	case *fakediscovery.FakeDiscovery:
+		fakeDiscovery.FakeDiscovery = d
+	default:
+		t.Fatalf("unexpected discovery type: %T != %T", d, &fakediscovery.FakeDiscovery{})
+	}
+
+	errBuf := bytes.NewBuffer(make([]byte, 0, 4096))
+	opts := &PruneImagesOptions{
+		AppsClient:      appsclient.NewSimpleClientset().Apps(),
+		BuildClient:     buildclient.NewSimpleClientset(&bcBad).Build(),
+		ImageClient:     imageFake.Image(),
+		KubeClient:      kFake,
+		DiscoveryClient: fakeDiscovery,
+		Out:             ioutil.Discard,
+		ErrOut:          errBuf,
+	}
+
+	verifyOutput := func(out string, expectClientVersionMismatch bool) {
+		t.Logf("pruner error output: %s\n", out)
+
+		badRefErrors := sets.NewString()
+		for _, l := range strings.Split(out, "\n") {
+			if strings.HasPrefix(l, "  ") {
+				badRefErrors.Insert(l[2:])
+			}
+		}
+		expBadRefErrors := sets.NewString(
+			`Pod[foo/pod1]: invalid docker image reference "invalid image reference": invalid reference format`,
+			`BuildConfig[foo/bc1]: invalid ImageStreamImage reference "bar:invalid-digest": expected exactly one @ in the isimage name "bar:invalid-digest"`,
+			`Deployment[foo/dep1]: invalid docker image reference "do not blame me": invalid reference format`)
+
+		if a, e := badRefErrors, expBadRefErrors; !a.Equal(e) {
+			t.Fatalf("got unexpected invalid reference errors: %s", diff.ObjectDiff(a, e))
+		}
+
+		if expectClientVersionMismatch {
+			if msg := "client version"; !strings.Contains(strings.ToLower(out), msg) {
+				t.Errorf("expected message %q is not contained in the output", msg)
+			}
+		} else {
+			for _, msg := range []string{"failed to get master api version", "client version"} {
+				if strings.Contains(strings.ToLower(out), msg) {
+					t.Errorf("got unexpected message %q in the output", msg)
+				}
+			}
+		}
+	}
+
+	err := opts.Run()
+	if err == nil {
+		t.Fatal("Unexpected non-error")
+	}
+
+	t.Logf("pruner error: %s\n", err)
+	verifyOutput(errBuf.String(), false)
+
+	t.Logf("bump master version and try again")
+	fakeDiscovery.masterVersion.Minor += "1"
+	errBuf.Reset()
+	err = opts.Run()
+	if err == nil {
+		t.Fatal("Unexpected non-error")
+	}
+
+	t.Logf("pruner error: %s\n", err)
+	verifyOutput(errBuf.String(), true)
+}
+
+type fakeVersionDiscovery struct {
+	*fakediscovery.FakeDiscovery
+	masterVersion version.Info
+}
+
+func (f *fakeVersionDiscovery) RESTClient() restclient.Interface {
+	return &restfake.RESTClient{
+		APIRegistry:          kapi.Registry,
+		NegotiatedSerializer: scheme.Codecs,
+		Client: restfake.CreateHTTPClient(func(req *http.Request) (*http.Response, error) {
+			if req.URL.Path != "/version/openshift" {
+				return &http.Response{
+					StatusCode: http.StatusNotFound,
+				}, nil
+			}
+			header := http.Header{}
+			header.Set("Content-Type", runtime.ContentTypeJSON)
+			return &http.Response{
+				StatusCode: http.StatusOK,
+				Header:     header,
+				Body:       objBody(&f.masterVersion),
+			}, nil
+		}),
+	}
+}
+
+func objBody(object interface{}) io.ReadCloser {
+	output, err := json.MarshalIndent(object, "", "")
+	if err != nil {
+		panic(err)
+	}
+	return ioutil.NopCloser(bytes.NewReader([]byte(output)))
+}

pkg/image/prune/testutil/util.go

@@ -74,13 +74,12 @@ func addImageStreamsToGraph(g graph.Graph, streams *imageapi.ImageStreamList) {
 		for tag, history := range stream.Status.Tags {
 			for i := range history.Items {
 				image := history.Items[i]
-				n := imagegraph.FindImage(g, image.Image)
-				if n == nil {
+				imageNode := imagegraph.FindImage(g, image.Image)
+				if imageNode == nil {
 					glog.V(2).Infof("Unable to find image %q in graph (from tag=%q, dockerImageReference=%s)",
 						history.Items[i].Image, tag, image.DockerImageReference)
 					continue
 				}
-				imageNode := n.(*imagegraph.ImageNode)
 				glog.V(4).Infof("Adding edge from %q to %q", imageStreamNode.UniqueName(), imageNode.UniqueName())
 				edgeKind := ImageStreamImageEdgeKind
 				if i > 1 {

pkg/oc/admin/prune/images.go

@@ -1871,6 +1871,27 @@ items:
     verbs:
     - get
     - list
+  - apiGroups:
+    - extensions
+    resources:
+    - daemonsets
+    verbs:
+    - get
+    - list
+  - apiGroups:
+    - extensions
+    resources:
+    - deployments
+    verbs:
+    - get
+    - list
+  - apiGroups:
+    - extensions
+    resources:
+    - replicasets
+    verbs:
+    - get
+    - list
   - apiGroups:
     - ""
     - image.openshift.io

pkg/oc/admin/prune/images_test.go

@@ -2041,6 +2041,30 @@ items:
     verbs:
     - get
     - list
+  - apiGroups:
+    - extensions
+    attributeRestrictions: null
+    resources:
+    - daemonsets
+    verbs:
+    - get
+    - list
+  - apiGroups:
+    - extensions
+    attributeRestrictions: null
+    resources:
+    - deployments
+    verbs:
+    - get
+    - list
+  - apiGroups:
+    - extensions
+    attributeRestrictions: null
+    resources:
+    - replicasets
+    verbs:
+    - get
+    - list
   - apiGroups:
     - ""
     - image.openshift.io