Merge pull request #70876 from chuckha/idempotency

cmd/kubeadm/app/util/apiclient/BUILD

@@ -62,14 +62,17 @@ go_test(
     name = "go_default_test",
     srcs = [
         "dryrunclient_test.go",
+        "idempotency_test.go",
         "init_dryrun_test.go",
     ],
     embed = [":go_default_library"],
     deps = [
+        "//pkg/kubelet/apis:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/api/rbac/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//staging/src/k8s.io/client-go/kubernetes/fake:go_default_library",
         "//staging/src/k8s.io/client-go/testing:go_default_library",
     ],
 )

cmd/kubeadm/app/util/apiclient/idempotency.go

@@ -194,13 +194,16 @@ func CreateOrUpdateClusterRoleBinding(client clientset.Interface, clusterRoleBin
 	return nil
 }
 
-// PatchNode tries to patch a node using the following client, executing patchFn for the actual mutating logic
-func PatchNode(client clientset.Interface, nodeName string, patchFn func(*v1.Node)) error {
-	// Loop on every false return. Return with an error if raised. Exit successfully if true is returned.
-	return wait.Poll(constants.APICallRetryInterval, constants.PatchNodeTimeout, func() (bool, error) {
+// PatchNodeOnce executes patchFn on the node object found by the node name.
+// This is a condition function meant to be used with wait.Poll. false, nil
+// implies it is safe to try again, an error indicates no more tries should be
+// made and true indicates success.
+func PatchNodeOnce(client clientset.Interface, nodeName string, patchFn func(*v1.Node)) func() (bool, error) {
+	return func() (bool, error) {
 		// First get the node object
 		n, err := client.CoreV1().Nodes().Get(nodeName, metav1.GetOptions{})
 		if err != nil {
+			// TODO this should only be for timeouts
 			return false, nil
 		}
 
@@ -212,7 +215,7 @@ func PatchNode(client clientset.Interface, nodeName string, patchFn func(*v1.Nod
 
 		oldData, err := json.Marshal(n)
 		if err != nil {
-			return false, err
+			return false, errors.Wrapf(err, "failed to marshal unmodified node %q into JSON", n.Name)
 		}
 
 		// Execute the mutating function
@@ -220,22 +223,32 @@ func PatchNode(client clientset.Interface, nodeName string, patchFn func(*v1.Nod
 
 		newData, err := json.Marshal(n)
 		if err != nil {
-			return false, err
+			return false, errors.Wrapf(err, "failed to marshal modified node %q into JSON", n.Name)
 		}
 
 		patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, v1.Node{})
 		if err != nil {
-			return false, err
+			return false, errors.Wrap(err, "failed to create two way merge patch")
 		}
 
 		if _, err := client.CoreV1().Nodes().Patch(n.Name, types.StrategicMergePatchType, patchBytes); err != nil {
+			// TODO also check for timeouts
 			if apierrors.IsConflict(err) {
 				fmt.Println("[patchnode] Temporarily unable to update node metadata due to conflict (will retry)")
 				return false, nil
 			}
-			return false, err
+			return false, errors.Wrapf(err, "error patching node %q through apiserver", n.Name)
 		}
 
 		return true, nil
-	})
+	}
+}
+
+// PatchNode tries to patch a node using patchFn for the actual mutating logic.
+// Retries are provided by the wait package.
+func PatchNode(client clientset.Interface, nodeName string, patchFn func(*v1.Node)) error {
+	// wait.Poll will rerun the condition function every interval function if
+	// the function returns false. If the condition function returns an error
+	// then the retries end and the error is returned.
+	return wait.Poll(constants.APICallRetryInterval, constants.PatchNodeTimeout, PatchNodeOnce(client, nodeName, patchFn))
 }

cmd/kubeadm/app/util/apiclient/idempotency_test.go

+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package apiclient_test
+
+import (
+	"testing"
+
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/kubernetes/cmd/kubeadm/app/util/apiclient"
+	kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
+)
+
+func TestPatchNodeNonErrorCases(t *testing.T) {
+	testcases := []struct {
+		name       string
+		lookupName string
+		node       v1.Node
+		success    bool
+	}{
+		{
+			name:       "simple update",
+			lookupName: "testnode",
+			node: v1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:   "testnode",
+					Labels: map[string]string{kubeletapis.LabelHostname: ""},
+				},
+			},
+			success: true,
+		},
+		{
+			name:       "node does not exist",
+			lookupName: "whale",
+			success:    false,
+		},
+		{
+			name:       "node not labelled yet",
+			lookupName: "robin",
+			node: v1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "robin",
+				},
+			},
+			success: false,
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			client := fake.NewSimpleClientset()
+			_, err := client.Core().Nodes().Create(&tc.node)
+			if err != nil {
+				t.Fatalf("failed to create node to fake client: %v", err)
+			}
+			conditionFunction := apiclient.PatchNodeOnce(client, tc.lookupName, func(node *v1.Node) {
+				node.Name = "testNewNode"
+			})
+			success, err := conditionFunction()
+			if err != nil {
+				t.Fatalf("did not expect error: %v", err)
+			}
+			if success != tc.success {
+				t.Fatalf("expected %v got %v", tc.success, success)
+			}
+		})
+	}
+}