Fix issues with oc adm migrate authorization

This change handles a number of bugs with the migrate authorization
command:

1. Rework MigrateAuthorizationOptions.checkParity to be a
MigrateActionFunc instead of a MigrateVisitFunc.  This allows us to
take advantage of migrate's default retry handling.
2. Add proper retry logic to checkParity, and have all check* funcs
return the appropriate TemporaryError based on the situation.  This
coupled with migrate's existing retry logic makes the command
resilient against common errors such as the deletion of resources.
3. Remove the binding of the standard migrate flags from migrate
authorization.  This command supports no parameters, and exposing
the standard migrate parameters allows the user to accidentally
break how the command runs.
4. Fix GroupVersion constants used for discovery based gating.  They
were incorrectly set to the internal version instead of v1.  This
would cause the policy based gating to always think that the server
did not support policy objects.
5. Force RBAC client to use v1beta1 since that is the only version
supported by a 3.6 server.  This allows you to use a 3.9 client
against a 3.6 server.
6. Remove rate limiting from the RBAC client to fix BZ 1513139.
Only a cluster admin can interact with RBAC resources on a 3.6
server, so this will quickly error out if run by a non-privileged
user.

Bug 1513139

Signed-off-by: Monis Khan <[email protected]>

Author: enj

pkg/oc/admin/migrate/authorization/authorization.go

@@ -5,9 +5,10 @@ import (
 	"fmt"
 	"io"
 
+	"k8s.io/api/rbac/v1beta1"
+	kerrs "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/client-go/util/flowcontrol"
 	rbacinternalversion "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/rbac/internalversion"
 	"k8s.io/kubernetes/pkg/kubectl/cmd/templates"
 	kcmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
@@ -39,7 +40,8 @@ var (
 
 		No resources are mutated.`)
 
-	errOutOfSync = errors.New("is not in sync with RBAC")
+	// errOutOfSync is retriable since it could be caused by the controller lagging behind
+	errOutOfSync = migrate.ErrRetriable{errors.New("is not in sync with RBAC")}
 )
 
 type MigrateAuthorizationOptions struct {
@@ -54,6 +56,7 @@ func NewCmdMigrateAuthorization(name, fullName string, f *clientcmd.Factory, in
 			Out:           out,
 			ErrOut:        errout,
 			AllNamespaces: true,
+			Confirm:       true, // force our save function to always run (it is read only)
 			Include: []string{
 				"clusterroles.authorization.openshift.io",
 				"roles.authorization.openshift.io",
@@ -73,8 +76,6 @@ func NewCmdMigrateAuthorization(name, fullName string, f *clientcmd.Factory, in
 		},
 		Deprecated: fmt.Sprintf("will not work against 3.7 servers"),
 	}
-	options.ResourceOptions.Bind(cmd)
-
 	return cmd
 }
 
@@ -83,20 +84,40 @@ func (o *MigrateAuthorizationOptions) Complete(name string, f *clientcmd.Factory
 		return fmt.Errorf("%s takes no positional arguments", name)
 	}
 
+	o.ResourceOptions.SaveFn = o.checkParity
 	if err := o.ResourceOptions.Complete(f, c); err != nil {
 		return err
 	}
 
-	kclient, err := f.ClientSet()
+	discovery, err := f.DiscoveryClient()
 	if err != nil {
 		return err
 	}
 
-	if err := clientcmd.LegacyPolicyResourceGate(kclient.Discovery()); err != nil {
+	if err := clientcmd.LegacyPolicyResourceGate(discovery); err != nil {
+		return err
+	}
+
+	config, err := f.ClientConfig()
+	if err != nil {
 		return err
 	}
 
-	o.rbac = kclient.Rbac()
+	// do not rate limit this client because it has to scan all RBAC data across the cluster
+	// this is safe because only a cluster admin will have the ability to read these objects
+	configShallowCopy := *config
+	configShallowCopy.RateLimiter = flowcontrol.NewFakeAlwaysRateLimiter()
+
+	// This command is only compatible with a 3.6 server, which only supported RBAC v1beta1
+	// Thus we must force that GV otherwise the client will default to v1
+	gv := v1beta1.SchemeGroupVersion
+	configShallowCopy.GroupVersion = &gv
+
+	rbac, err := rbacinternalversion.NewForConfig(&configShallowCopy)
+	if err != nil {
+		return err
+	}
+	o.rbac = rbac
 
 	return nil
 }
@@ -106,130 +127,159 @@ func (o MigrateAuthorizationOptions) Validate() error {
 }
 
 func (o MigrateAuthorizationOptions) Run() error {
-	return o.ResourceOptions.Visitor().Visit(func(info *resource.Info) (migrate.Reporter, error) {
-		return o.checkParity(info.Object)
-	})
+	// we lie and say this object has changed so our save function will run
+	return o.ResourceOptions.Visitor().Visit(migrate.AlwaysRequiresMigration)
 }
 
 // checkParity confirms that Openshift authorization objects are in sync with Kubernetes RBAC
 // and returns an error if they are out of sync or if it encounters a conversion error
-func (o *MigrateAuthorizationOptions) checkParity(obj runtime.Object) (migrate.Reporter, error) {
-	var errlist []error
-	switch t := obj.(type) {
+func (o *MigrateAuthorizationOptions) checkParity(info *resource.Info, _ migrate.Reporter) error {
+	var err migrate.TemporaryError
+
+	switch t := info.Object.(type) {
 	case *authorizationapi.ClusterRole:
-		errlist = append(errlist, o.checkClusterRole(t)...)
+		err = o.checkClusterRole(t)
 	case *authorizationapi.Role:
-		errlist = append(errlist, o.checkRole(t)...)
+		err = o.checkRole(t)
 	case *authorizationapi.ClusterRoleBinding:
-		errlist = append(errlist, o.checkClusterRoleBinding(t)...)
+		err = o.checkClusterRoleBinding(t)
 	case *authorizationapi.RoleBinding:
-		errlist = append(errlist, o.checkRoleBinding(t)...)
+		err = o.checkRoleBinding(t)
 	default:
-		return nil, nil // indicate that we ignored the object
+		// this should never happen unless o.Include or the server is broken
+		return fmt.Errorf("impossible type %T for checkParity info=%#v object=%#v", t, info, t)
+	}
+
+	// We encountered no error, so this object is in sync.
+	if err == nil {
+		// we only perform read operations so we return this error to signal that we did not change anything
+		return migrate.ErrUnchanged
+	}
+
+	// At this point we know that we have some non-nil TemporaryError.
+	// If it has the possibility of being transient, we need to sync ourselves with the current state of the object.
+	if err.Temporary() {
+		// The most likely cause is that an authorization object was deleted after we initially fetched it,
+		// and the controller deleted the associated RBAC object, which caused our RBAC GET to fail.
+		// We can confirm this by refetching the authorization object.
+		refreshErr := info.Get()
+		if refreshErr != nil {
+			// Our refresh GET for this authorization object failed.
+			// The default logic for migration errors is appropriate in this case (refreshErr is most likely a NotFound).
+			return migrate.DefaultRetriable(info, refreshErr)
+		}
+		// We had no refreshErr, but encountered some other possibly transient error.
+		// No special handling is required in this case, we just pass it through below.
 	}
-	return migrate.NotChanged, utilerrors.NewAggregate(errlist) // we only perform read operations
+
+	// All of the check* funcs return an appropriate TemporaryError based on the failure,
+	// so we can pass that through to the default migration logic which will retry as needed.
+	return err
 }
 
-func (o *MigrateAuthorizationOptions) checkClusterRole(originClusterRole *authorizationapi.ClusterRole) []error {
-	var errlist []error
+// handleRBACGetError signals for a retry on NotFound (handles deletion and sync lag)
+// and ServerTimeout (handles heavy load against the server).
+func handleRBACGetError(err error) migrate.TemporaryError {
+	switch {
+	case kerrs.IsNotFound(err), kerrs.IsServerTimeout(err):
+		return migrate.ErrRetriable{err}
+	default:
+		return migrate.ErrNotRetriable{err}
+	}
+}
 
+func (o *MigrateAuthorizationOptions) checkClusterRole(originClusterRole *authorizationapi.ClusterRole) migrate.TemporaryError {
 	// convert the origin role to a rbac role
 	convertedClusterRole, err := util.ConvertToRBACClusterRole(originClusterRole)
 	if err != nil {
-		errlist = append(errlist, err)
+		// conversion errors should basically never happen, so we do not attempt to retry on those
+		return migrate.ErrNotRetriable{err}
 	}
 
 	// try to get the equivalent rbac role from the api
 	rbacClusterRole, err := o.rbac.ClusterRoles().Get(originClusterRole.Name, v1.GetOptions{})
 	if err != nil {
-		errlist = append(errlist, err)
+		// it is possible that the controller has not synced this yet
+		return handleRBACGetError(err)
 	}
 
-	// compare the results if there have been no errors so far
-	if len(errlist) == 0 {
-		// if they are not equal, something has gone wrong and the two objects are not in sync
-		if util.PrepareForUpdateClusterRole(convertedClusterRole, rbacClusterRole) {
-			errlist = append(errlist, errOutOfSync)
-		}
+	// if they are not equal, something has gone wrong and the two objects are not in sync
+	if util.PrepareForUpdateClusterRole(convertedClusterRole, rbacClusterRole) {
+		// we retry on this since it could be caused by the controller lagging behind
+		return errOutOfSync
 	}
 
-	return errlist
+	return nil
 }
 
-func (o *MigrateAuthorizationOptions) checkRole(originRole *authorizationapi.Role) []error {
-	var errlist []error
-
+func (o *MigrateAuthorizationOptions) checkRole(originRole *authorizationapi.Role) migrate.TemporaryError {
 	// convert the origin role to a rbac role
 	convertedRole, err := util.ConvertToRBACRole(originRole)
 	if err != nil {
-		errlist = append(errlist, err)
+		// conversion errors should basically never happen, so we do not attempt to retry on those
+		return migrate.ErrNotRetriable{err}
 	}
 
 	// try to get the equivalent rbac role from the api
 	rbacRole, err := o.rbac.Roles(originRole.Namespace).Get(originRole.Name, v1.GetOptions{})
 	if err != nil {
-		errlist = append(errlist, err)
+		// it is possible that the controller has not synced this yet
+		return handleRBACGetError(err)
 	}
 
-	// compare the results if there have been no errors so far
-	if len(errlist) == 0 {
-		// if they are not equal, something has gone wrong and the two objects are not in sync
-		if util.PrepareForUpdateRole(convertedRole, rbacRole) {
-			errlist = append(errlist, errOutOfSync)
-		}
+	// if they are not equal, something has gone wrong and the two objects are not in sync
+	if util.PrepareForUpdateRole(convertedRole, rbacRole) {
+		// we retry on this since it could be caused by the controller lagging behind
+		return errOutOfSync
 	}
 
-	return errlist
+	return nil
 }
 
-func (o *MigrateAuthorizationOptions) checkClusterRoleBinding(originRoleBinding *authorizationapi.ClusterRoleBinding) []error {
-	var errlist []error
-
+func (o *MigrateAuthorizationOptions) checkClusterRoleBinding(originRoleBinding *authorizationapi.ClusterRoleBinding) migrate.TemporaryError {
 	// convert the origin role binding to a rbac role binding
 	convertedRoleBinding, err := util.ConvertToRBACClusterRoleBinding(originRoleBinding)
 	if err != nil {
-		errlist = append(errlist, err)
+		// conversion errors should basically never happen, so we do not attempt to retry on those
+		return migrate.ErrNotRetriable{err}
 	}
 
 	// try to get the equivalent rbac role binding from the api
 	rbacRoleBinding, err := o.rbac.ClusterRoleBindings().Get(originRoleBinding.Name, v1.GetOptions{})
 	if err != nil {
-		errlist = append(errlist, err)
+		// it is possible that the controller has not synced this yet
+		return handleRBACGetError(err)
 	}
 
-	// compare the results if there have been no errors so far
-	if len(errlist) == 0 {
-		// if they are not equal, something has gone wrong and the two objects are not in sync
-		if util.PrepareForUpdateClusterRoleBinding(convertedRoleBinding, rbacRoleBinding) {
-			errlist = append(errlist, errOutOfSync)
-		}
+	// if they are not equal, something has gone wrong and the two objects are not in sync
+	if util.PrepareForUpdateClusterRoleBinding(convertedRoleBinding, rbacRoleBinding) {
+		// we retry on this since it could be caused by the controller lagging behind
+		return errOutOfSync
 	}
 
-	return errlist
+	return nil
 }
 
-func (o *MigrateAuthorizationOptions) checkRoleBinding(originRoleBinding *authorizationapi.RoleBinding) []error {
-	var errlist []error
-
+func (o *MigrateAuthorizationOptions) checkRoleBinding(originRoleBinding *authorizationapi.RoleBinding) migrate.TemporaryError {
 	// convert the origin role binding to a rbac role binding
 	convertedRoleBinding, err := util.ConvertToRBACRoleBinding(originRoleBinding)
 	if err != nil {
-		errlist = append(errlist, err)
+		// conversion errors should basically never happen, so we do not attempt to retry on those
+		return migrate.ErrNotRetriable{err}
 	}
 
 	// try to get the equivalent rbac role binding from the api
 	rbacRoleBinding, err := o.rbac.RoleBindings(originRoleBinding.Namespace).Get(originRoleBinding.Name, v1.GetOptions{})
 	if err != nil {
-		errlist = append(errlist, err)
+		// it is possible that the controller has not synced this yet
+		return handleRBACGetError(err)
 	}
 
-	// compare the results if there have been no errors so far
-	if len(errlist) == 0 {
-		// if they are not equal, something has gone wrong and the two objects are not in sync
-		if util.PrepareForUpdateRoleBinding(convertedRoleBinding, rbacRoleBinding) {
-			errlist = append(errlist, errOutOfSync)
-		}
+	// if they are not equal, something has gone wrong and the two objects are not in sync
+	if util.PrepareForUpdateRoleBinding(convertedRoleBinding, rbacRoleBinding) {
+		// we retry on this since it could be caused by the controller lagging behind
+		return errOutOfSync
 	}
 
-	return errlist
+	return nil
 }

pkg/oc/admin/migrate/migrator.go

@@ -54,9 +54,6 @@ func timeStampNow() string {
 	return time.Now().Format("0102 15:04:05.000000")
 }
 
-// NotChanged is a Reporter returned by operations that are guaranteed to be read-only
-var NotChanged = ReporterBool(false)
-
 // ResourceOptions assists in performing migrations on any object that
 // can be retrieved via the API.
 type ResourceOptions struct {
@@ -102,7 +99,12 @@ func (o *ResourceOptions) Bind(c *cobra.Command) {
 }
 
 func (o *ResourceOptions) Complete(f *clientcmd.Factory, c *cobra.Command) error {
-	o.Output = kcmdutil.GetFlagString(c, "output")
+	// oc adm migrate authorization does not support printing and takes no parameters, so we ignore the error here
+	var flagErr error
+	o.Output, flagErr = c.Flags().GetString("output")
+	if flagErr != nil {
+		glog.V(4).Infof("Error getting output flag: %v", flagErr)
+	}
 	switch {
 	case len(o.Output) > 0:
 		printer, err := f.PrinterForOptions(kcmdutil.ExtractCmdPrintOptions(c, false))
@@ -362,23 +364,27 @@ var ErrUnchanged = fmt.Errorf("migration was not necessary")
 // both status and spec must be changed).
 var ErrRecalculate = fmt.Errorf("recalculate migration")
 
+// MigrateError is an exported alias to error to allow external packages to use ErrRetriable and ErrNotRetriable
+type MigrateError error
+
 // ErrRetriable is a wrapper for an error that a migrator may use to indicate the
 // specific error can be retried.
 type ErrRetriable struct {
-	error
+	MigrateError
 }
 
 func (ErrRetriable) Temporary() bool { return true }
 
 // ErrNotRetriable is a wrapper for an error that a migrator may use to indicate the
 // specific error cannot be retried.
 type ErrNotRetriable struct {
-	error
+	MigrateError
 }
 
 func (ErrNotRetriable) Temporary() bool { return false }
 
-type temporary interface {
+type TemporaryError interface {
+	error
 	// Temporary should return true if this is a temporary error
 	Temporary() bool
 }
@@ -490,7 +496,7 @@ func (t *migrateTracker) try(info *resource.Info, retries int) (attemptResult, e
 
 // canRetry returns true if the provided error indicates a retry is possible.
 func canRetry(err error) bool {
-	if temp, ok := err.(temporary); ok && temp.Temporary() {
+	if temp, ok := err.(TemporaryError); ok && temp.Temporary() {
 		return true
 	}
 	return err == ErrRecalculate

pkg/oc/cli/cmd/create/policy_binding.go

@@ -72,12 +72,12 @@ func (o *CreatePolicyBindingOptions) Complete(cmd *cobra.Command, f *clientcmd.F
 	}
 	o.BindingNamespace = namespace
 
-	kclient, err := f.ClientSet()
+	discovery, err := f.DiscoveryClient()
 	if err != nil {
 		return err
 	}
 
-	if err := clientcmd.LegacyPolicyResourceGate(kclient.Discovery()); err != nil {
+	if err := clientcmd.LegacyPolicyResourceGate(discovery); err != nil {
 		return err
 	}
 	client, err := f.OpenshiftInternalAuthorizationClient()

pkg/oc/cli/util/clientcmd/gating.go

@@ -8,7 +8,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/discovery"
 
-	"github.com/openshift/origin/pkg/authorization/apis/authorization"
+	authorization "github.com/openshift/api/authorization/v1"
 )
 
 // LegacyPolicyResourceGate returns err if the server does not support the set of legacy policy objects (< 3.7)