using dynamic client

Author: Chao Xu

pkg/migrator/core.go

@@ -20,17 +20,24 @@ import (
 	"fmt"
 	"reflect"
 	"sync"
+	"time"
 
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
-	"k8s.io/client-go/rest"
+	"k8s.io/client-go/dynamic"
 )
 
 var metadataAccessor = meta.NewAccessor()
 
+const (
+	defaultChunkLimit  = 500
+	defaultConcurrency = 1
+)
+
 type progressTracker interface {
 	save(continueToken string) error
 	load() (continueToken string, err error)
@@ -47,55 +54,43 @@ func (d *dummyProgress) save(continueToken string) error {
 }
 
 type migrator struct {
-	resource        string
-	namespaceScoped bool
-	// Creator is responsible to configure a rest client. This includes
-	// creating a working serializer.
-	// We don't use dynamic client because it only supports json, which
-	// increases the memory usage of the apiserver.  When we support
-	// migrating CRD or aggregated apis, we need to plumb in a dynamic
-	// client as a backup.
-	client      rest.Interface
+	resource    schema.GroupVersionResource
+	client      dynamic.Interface
 	progress    progressTracker
 	concurrency int
 }
 
 // NewMigrator creates a migrator that can migrate a single resource type.
-func NewMigrator(resource string, namespaceScoped bool, client rest.Interface) *migrator {
+func NewMigrator(resource schema.GroupVersionResource, client dynamic.Interface) *migrator {
 	return &migrator{
-		resource:        resource,
-		namespaceScoped: namespaceScoped,
-		client:          client,
-		progress:        &dummyProgress{},
-		concurrency:     5,
+		resource:    resource,
+		client:      client,
+		progress:    &dummyProgress{},
+		concurrency: defaultConcurrency,
 	}
 }
 
-// get executes the GET request, but it does not decode the result.
-func (m *migrator) get(namespace, name string) rest.Result {
-	req := m.client.Get().
-		NamespaceIfScoped(namespace, m.namespaceScoped).
+func (m *migrator) get(namespace, name string) (*unstructured.Unstructured, error) {
+	// if namespace is empty, .Namespace(namespace) is ineffective.
+	return m.client.
 		Resource(m.resource).
-		Name(name)
-	return req.Do()
+		Namespace(namespace).
+		Get(name, metav1.GetOptions{})
 }
 
-// put executes the PUT request, but it does not decode the result.
-func (m *migrator) put(namespace, name string, data []byte) rest.Result {
-	req := m.client.Put().
-		NamespaceIfScoped(namespace, m.namespaceScoped).
+func (m *migrator) put(namespace, name string, obj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
+	// if namespace is empty, .Namespace(namespace) is ineffective.
+	return m.client.
 		Resource(m.resource).
-		Name(name).
-		Body(data)
-	return req.Do()
+		Namespace(namespace).
+		Update(obj)
 }
 
-func (m *migrator) list(options *metav1.ListOptions) (runtime.Object, error) {
-	req := m.client.Get().
-		NamespaceIfScoped(metav1.NamespaceAll, m.namespaceScoped).
+func (m *migrator) list(options metav1.ListOptions) (*unstructured.UnstructuredList, error) {
+	return m.client.
 		Resource(m.resource).
-		VersionedParams(options, metav1.ParameterCodec)
-	return req.Do().Get()
+		Namespace(metav1.NamespaceAll).
+		List(options)
 }
 
 // Run migrates all the instances of the resource type managed by the migrator.
@@ -106,13 +101,13 @@ func (m *migrator) Run() error {
 	}
 	for {
 		list, listError := m.list(
-			&metav1.ListOptions{
-				Limit:    500,
+			metav1.ListOptions{
+				Limit:    defaultChunkLimit,
 				Continue: continueToken,
 			},
 		)
 		if listError != nil && !errors.IsResourceExpired(listError) {
-			if canRetry(retriable(listError)) {
+			if canRetry(listError) {
 				continue
 			}
 			return listError
@@ -141,19 +136,15 @@ func (m *migrator) Run() error {
 	}
 }
 
-func (m *migrator) migrateList(l runtime.Object) error {
+func (m *migrator) migrateList(l *unstructured.UnstructuredList) error {
 	stop := make(chan struct{})
 	defer close(stop)
-	items, err := meta.ExtractList(l)
-	if err != nil {
-		return err
-	}
-	workc := make(chan runtime.Object)
+	workc := make(chan *unstructured.Unstructured)
 	go func() {
 		defer close(workc)
-		for _, item := range items {
+		for i := range l.Items {
 			select {
-			case workc <- item:
+			case workc <- &l.Items[i]:
 			case <-stop:
 				return
 			}
@@ -182,16 +173,10 @@ func (m *migrator) migrateList(l runtime.Object) error {
 	return utilerrors.NewAggregate(errors)
 }
 
-func (m *migrator) worker(stop <-chan struct{}, workc <-chan runtime.Object, errc chan<- error) {
+func (m *migrator) worker(stop <-chan struct{}, workc <-chan *unstructured.Unstructured, errc chan<- error) {
 	for item := range workc {
-		for {
-			err := m.try(item)
-			if err == nil {
-				break
-			}
-			if canRetry(err) {
-				continue
-			}
+		err := m.migrateOneItem(item)
+		if err != nil {
 			select {
 			case errc <- err:
 				continue
@@ -202,37 +187,54 @@ func (m *migrator) worker(stop <-chan struct{}, workc <-chan runtime.Object, err
 	}
 }
 
-// try tries to migrate the single object by GET and then PUT.
-func (m *migrator) try(work runtime.Object) error {
-	namespace, err := metadataAccessor.Namespace(work)
+func (m *migrator) migrateOneItem(item *unstructured.Unstructured) error {
+	namespace, err := metadataAccessor.Namespace(item)
 	if err != nil {
 		return err
 	}
-	name, err := metadataAccessor.Name(work)
+	name, err := metadataAccessor.Name(item)
 	if err != nil {
 		return err
 	}
-	get := m.get(namespace, name)
-	data, err := get.Raw()
-	if err != nil {
-		return retriable(get.Error())
+	getBeforePut := false
+	for {
+		getBeforePut, err = m.try(namespace, name, item, getBeforePut)
+		if err == nil || errors.IsNotFound(err) {
+			return nil
+		}
+		if canRetry(err) {
+			if seconds, delay := errors.SuggestsClientDelay(err); delay {
+				time.Sleep(time.Duration(seconds) * time.Second)
+			}
+			continue
+		}
+		// error is not retriable
+		return err
 	}
+}
 
-	put := m.put(namespace, name, data)
-	if err := put.Error(); err != nil {
-		return retriable(put.Error())
+// try tries to migrate the single object by PUT. It refreshes the object via
+// GET if "get" is true. If the PUT fails due to conflicts, or the GET fails,
+// the function requests the next try to GET the new object.
+func (m *migrator) try(namespace, name string, item *unstructured.Unstructured, get bool) (bool, error) {
+	var err error
+	if get {
+		item, err = m.get(namespace, name)
+		if err != nil {
+			return true, err
+		}
 	}
-
-	// TODO: The oc migrator tried to count if the update causes changes.
-	// To do so it needs to decode the PUT response and extract the
-	// resourceVersion.
+	_, err = m.put(namespace, name, item)
+	if err == nil {
+		return false, nil
+	}
+	return errors.IsConflict(err), err
 
 	// TODO: The oc admin uses a defer function to do bandwidth limiting
 	// after doing all operations. The rate limiter is marked as an alpha
 	// feature.  Is it better than the built-in qps limit in the REST
 	// client? Maybe it's necessary because not all resource types are of
 	// the same size?
-	return nil
 }
 
 // TODO: move this helper to "k8s.io/apimachinery/pkg/api/errors"