scale_test.go

Documentation: k8s.io/kubernetes/test/integration/apiserver/apply

     1  /*
     2  Copyright 2021 The Kubernetes Authors.
     3  
     4  Licensed under the Apache License, Version 2.0 (the "License");
     5  you may not use this file except in compliance with the License.
     6  You may obtain a copy of the License at
     7  
     8      http://www.apache.org/licenses/LICENSE-2.0
     9  
    10  Unless required by applicable law or agreed to in writing, software
    11  distributed under the License is distributed on an "AS IS" BASIS,
    12  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13  See the License for the specific language governing permissions and
    14  limitations under the License.
    15  */
    16  
    17  package apiserver
    18  
    19  import (
    20  	"context"
    21  	"encoding/json"
    22  	"fmt"
    23  	"path"
    24  	"strings"
    25  	"testing"
    26  
    27  	apierrors "k8s.io/apimachinery/pkg/api/errors"
    28  	"k8s.io/apimachinery/pkg/api/meta"
    29  	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
    30  	"k8s.io/apimachinery/pkg/runtime/schema"
    31  	"k8s.io/apimachinery/pkg/types"
    32  	"k8s.io/apimachinery/pkg/util/managedfields"
    33  	clientset "k8s.io/client-go/kubernetes"
    34  	"k8s.io/client-go/kubernetes/scheme"
    35  	deploymentstorage "k8s.io/kubernetes/pkg/registry/apps/deployment/storage"
    36  	replicasetstorage "k8s.io/kubernetes/pkg/registry/apps/replicaset/storage"
    37  	statefulsetstorage "k8s.io/kubernetes/pkg/registry/apps/statefulset/storage"
    38  	replicationcontrollerstorage "k8s.io/kubernetes/pkg/registry/core/replicationcontroller/storage"
    39  )
    40  
    41  type scaleTest struct {
    42  	kind     string
    43  	resource string
    44  	path     string
    45  	validObj string
    46  }
    47  
    48  func TestScaleAllResources(t *testing.T) {
    49  	client, closeFn := setup(t)
    50  	defer closeFn()
    51  
    52  	tests := []scaleTest{
    53  		{
    54  			kind:     "Deployment",
    55  			resource: "deployments",
    56  			path:     "/apis/apps/v1",
    57  			validObj: validAppsV1("Deployment"),
    58  		},
    59  		{
    60  			kind:     "StatefulSet",
    61  			resource: "statefulsets",
    62  			path:     "/apis/apps/v1",
    63  			validObj: validAppsV1("StatefulSet"),
    64  		},
    65  		{
    66  			kind:     "ReplicaSet",
    67  			resource: "replicasets",
    68  			path:     "/apis/apps/v1",
    69  			validObj: validAppsV1("ReplicaSet"),
    70  		},
    71  		{
    72  			kind:     "ReplicationController",
    73  			resource: "replicationcontrollers",
    74  			path:     "/api/v1",
    75  			validObj: validV1ReplicationController(),
    76  		},
    77  	}
    78  
    79  	for _, test := range tests {
    80  		t.Run(test.kind, func(t *testing.T) {
    81  			validObject := []byte(test.validObj)
    82  
    83  			// Create the object
    84  			_, err := client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
    85  				AbsPath(test.path).
    86  				Namespace("default").
    87  				Resource(test.resource).
    88  				Name("test").
    89  				Param("fieldManager", "apply_test").
    90  				Body(validObject).
    91  				Do(context.TODO()).Get()
    92  			if err != nil {
    93  				t.Fatalf("Failed to create object using apply: %v", err)
    94  			}
    95  			obj := retrieveObject(t, client, test.path, test.resource)
    96  			assertReplicasValue(t, obj, 1)
    97  			assertReplicasOwnership(t, obj, "apply_test")
    98  
    99  			// Call scale subresource to update replicas
   100  			_, err = client.CoreV1().RESTClient().
   101  				Patch(types.MergePatchType).
   102  				AbsPath(test.path).
   103  				Namespace("default").
   104  				Resource(test.resource).
   105  				Name("test").
   106  				SubResource("scale").
   107  				Param("fieldManager", "scale_test").
   108  				Body([]byte(`{"spec":{"replicas": 5}}`)).
   109  				Do(context.TODO()).Get()
   110  			if err != nil {
   111  				t.Fatalf("Failed to scale object: %v", err)
   112  			}
   113  			obj = retrieveObject(t, client, test.path, test.resource)
   114  			assertReplicasValue(t, obj, 5)
   115  			assertReplicasOwnership(t, obj, "scale_test")
   116  
   117  			// Re-apply the original object, it should fail with conflict because replicas have changed
   118  			_, err = client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
   119  				AbsPath(test.path).
   120  				Namespace("default").
   121  				Resource(test.resource).
   122  				Name("test").
   123  				Param("fieldManager", "apply_test").
   124  				Body(validObject).
   125  				Do(context.TODO()).Get()
   126  			if !apierrors.IsConflict(err) {
   127  				t.Fatalf("Expected conflict when re-applying the original object, but got: %v", err)
   128  			}
   129  
   130  			// Re-apply forcing the changes should succeed
   131  			_, err = client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
   132  				AbsPath(test.path).
   133  				Namespace("default").
   134  				Resource(test.resource).
   135  				Name("test").
   136  				Param("fieldManager", "apply_test").
   137  				Param("force", "true").
   138  				Body(validObject).
   139  				Do(context.TODO()).Get()
   140  			if err != nil {
   141  				t.Fatalf("Error force-updating: %v", err)
   142  			}
   143  			obj = retrieveObject(t, client, test.path, test.resource)
   144  			assertReplicasValue(t, obj, 1)
   145  			assertReplicasOwnership(t, obj, "apply_test")
   146  
   147  			// Run "Apply" with a scale object with a different number of replicas. It should generate a conflict.
   148  			_, err = client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
   149  				AbsPath(test.path).
   150  				Namespace("default").
   151  				Resource(test.resource).
   152  				SubResource("scale").
   153  				Name("test").
   154  				Param("fieldManager", "apply_scale").
   155  				Body([]byte(`{"kind":"Scale","apiVersion":"autoscaling/v1","metadata":{"name":"test","namespace":"default"},"spec":{"replicas":17}}`)).
   156  				Do(context.TODO()).Get()
   157  			if !apierrors.IsConflict(err) {
   158  				t.Fatalf("Expected conflict error but got: %v", err)
   159  			}
   160  			if !strings.Contains(err.Error(), "apply_test") {
   161  				t.Fatalf("Expected conflict with `apply_test` manager when but got: %v", err)
   162  			}
   163  
   164  			// Same as before but force. Only the new manager should own .spec.replicas
   165  			_, err = client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
   166  				AbsPath(test.path).
   167  				Namespace("default").
   168  				Resource(test.resource).
   169  				SubResource("scale").
   170  				Name("test").
   171  				Param("fieldManager", "apply_scale").
   172  				Param("force", "true").
   173  				Body([]byte(`{"kind":"Scale","apiVersion":"autoscaling/v1","metadata":{"name":"test","namespace":"default"},"spec":{"replicas":17}}`)).
   174  				Do(context.TODO()).Get()
   175  			if err != nil {
   176  				t.Fatalf("Error updating object by applying scale and forcing: %v ", err)
   177  			}
   178  			obj = retrieveObject(t, client, test.path, test.resource)
   179  			assertReplicasValue(t, obj, 17)
   180  			assertReplicasOwnership(t, obj, "apply_scale")
   181  
   182  			// Replace scale object
   183  			_, err = client.CoreV1().RESTClient().Put().
   184  				AbsPath(test.path).
   185  				Namespace("default").
   186  				Resource(test.resource).
   187  				SubResource("scale").
   188  				Name("test").
   189  				Param("fieldManager", "replace_test").
   190  				Body([]byte(`{"kind":"Scale","apiVersion":"autoscaling/v1","metadata":{"name":"test","namespace":"default"},"spec":{"replicas":7}}`)).
   191  				Do(context.TODO()).Get()
   192  			if err != nil {
   193  				t.Fatalf("Error replacing object: %v", err)
   194  			}
   195  			obj = retrieveObject(t, client, test.path, test.resource)
   196  			assertReplicasValue(t, obj, 7)
   197  			assertReplicasOwnership(t, obj, "replace_test")
   198  
   199  			// Apply the same number of replicas, both managers should own the field
   200  			_, err = client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
   201  				AbsPath(test.path).
   202  				Namespace("default").
   203  				Resource(test.resource).
   204  				SubResource("scale").
   205  				Name("test").
   206  				Param("fieldManager", "co_owning_test").
   207  				Body([]byte(`{"kind":"Scale","apiVersion":"autoscaling/v1","metadata":{"name":"test","namespace":"default"},"spec":{"replicas":7}}`)).
   208  				Do(context.TODO()).Get()
   209  			if err != nil {
   210  				t.Fatalf("Error updating object: %v", err)
   211  			}
   212  			obj = retrieveObject(t, client, test.path, test.resource)
   213  			assertReplicasValue(t, obj, 7)
   214  			assertReplicasOwnership(t, obj, "replace_test", "co_owning_test")
   215  
   216  			// Scaling again should make this manager the only owner of replicas
   217  			_, err = client.CoreV1().RESTClient().Patch(types.MergePatchType).
   218  				AbsPath(test.path).
   219  				Namespace("default").
   220  				Resource(test.resource).
   221  				SubResource("scale").
   222  				Name("test").
   223  				Param("fieldManager", "scale_test").
   224  				Body([]byte(`{"spec":{"replicas": 5}}`)).
   225  				Do(context.TODO()).Get()
   226  			if err != nil {
   227  				t.Fatalf("Error scaling object: %v", err)
   228  			}
   229  			obj = retrieveObject(t, client, test.path, test.resource)
   230  			assertReplicasValue(t, obj, 5)
   231  			assertReplicasOwnership(t, obj, "scale_test")
   232  		})
   233  	}
   234  }
   235  
   236  func TestScaleUpdateOnlyStatus(t *testing.T) {
   237  	client, closeFn := setup(t)
   238  	defer closeFn()
   239  
   240  	resource := "deployments"
   241  	path := "/apis/apps/v1"
   242  	validObject := []byte(validAppsV1("Deployment"))
   243  
   244  	// Create the object
   245  	_, err := client.CoreV1().RESTClient().Patch(types.ApplyPatchType).
   246  		AbsPath(path).
   247  		Namespace("default").
   248  		Resource(resource).
   249  		Name("test").
   250  		Param("fieldManager", "apply_test").
   251  		Body(validObject).
   252  		Do(context.TODO()).Get()
   253  	if err != nil {
   254  		t.Fatalf("Failed to create object using apply: %v", err)
   255  	}
   256  	obj := retrieveObject(t, client, path, resource)
   257  	assertReplicasValue(t, obj, 1)
   258  	assertReplicasOwnership(t, obj, "apply_test")
   259  
   260  	// Call scale subresource to update replicas
   261  	_, err = client.CoreV1().RESTClient().
   262  		Patch(types.MergePatchType).
   263  		AbsPath(path).
   264  		Namespace("default").
   265  		Resource(resource).
   266  		Name("test").
   267  		SubResource("scale").
   268  		Param("fieldManager", "scale_test").
   269  		Body([]byte(`{"status":{"replicas": 42}}`)).
   270  		Do(context.TODO()).Get()
   271  	if err != nil {
   272  		t.Fatalf("Failed to scale object: %v", err)
   273  	}
   274  	obj = retrieveObject(t, client, path, resource)
   275  	assertReplicasValue(t, obj, 1)
   276  	assertReplicasOwnership(t, obj, "apply_test")
   277  }
   278  
   279  func TestAllKnownVersionsAreInMappings(t *testing.T) {
   280  	cases := []struct {
   281  		groupKind schema.GroupKind
   282  		mappings  managedfields.ResourcePathMappings
   283  	}{
   284  		{
   285  			groupKind: schema.GroupKind{Group: "apps", Kind: "ReplicaSet"},
   286  			mappings:  replicasetstorage.ReplicasPathMappings(),
   287  		},
   288  		{
   289  			groupKind: schema.GroupKind{Group: "apps", Kind: "StatefulSet"},
   290  			mappings:  statefulsetstorage.ReplicasPathMappings(),
   291  		},
   292  		{
   293  			groupKind: schema.GroupKind{Group: "apps", Kind: "Deployment"},
   294  			mappings:  deploymentstorage.ReplicasPathMappings(),
   295  		},
   296  		{
   297  			groupKind: schema.GroupKind{Group: "", Kind: "ReplicationController"},
   298  			mappings:  replicationcontrollerstorage.ReplicasPathMappings(),
   299  		},
   300  	}
   301  	for _, c := range cases {
   302  		knownVersions := scheme.Scheme.VersionsForGroupKind(c.groupKind)
   303  		for _, version := range knownVersions {
   304  			if _, ok := c.mappings[version.String()]; !ok {
   305  				t.Errorf("missing version %v for %v mappings", version, c.groupKind)
   306  			}
   307  		}
   308  
   309  		if len(knownVersions) != len(c.mappings) {
   310  			t.Errorf("%v mappings has extra items: %v vs %v", c.groupKind, c.mappings, knownVersions)
   311  		}
   312  	}
   313  }
   314  
   315  func validAppsV1(kind string) string {
   316  	return fmt.Sprintf(`{
   317  	    "apiVersion": "apps/v1",
   318  	    "kind": "%s",
   319  	    "metadata": {
   320  	      "name": "test"
   321  	    },
   322  	    "spec": {
   323  	      "replicas": 1,
   324  	      "selector": {
   325  	        "matchLabels": {
   326  	           "app": "nginx"
   327  	        }
   328  	      },
   329  	      "template": {
   330  	        "metadata": {
   331  	          "labels": {
   332  	            "app": "nginx"
   333  	          }
   334  	        },
   335  	        "spec": {
   336  	          "containers": [{
   337  	            "name":  "nginx",
   338  	            "image": "nginx:latest"
   339  	          }]
   340  	        }
   341  	      }
   342  	    }
   343  	  }`, kind)
   344  }
   345  
   346  func validV1ReplicationController() string {
   347  	return `{
   348  	    "apiVersion": "v1",
   349  	    "kind": "ReplicationController",
   350  	    "metadata": {
   351  	      "name": "test"
   352  	    },
   353  	    "spec": {
   354  	      "replicas": 1,
   355  	      "selector": {
   356            "app": "nginx"
   357  	      },
   358  	      "template": {
   359  	        "metadata": {
   360  	          "labels": {
   361  	            "app": "nginx"
   362  	          }
   363  	        },
   364  	        "spec": {
   365  	          "containers": [{
   366  	            "name":  "nginx",
   367  	            "image": "nginx:latest"
   368  	          }]
   369  	        }
   370  	      }
   371  	    }
   372  	  }`
   373  }
   374  
   375  func retrieveObject(t *testing.T, client clientset.Interface, prefix, resource string) *unstructured.Unstructured {
   376  	t.Helper()
   377  
   378  	urlPath := path.Join(prefix, "namespaces", "default", resource, "test")
   379  	bytes, err := client.CoreV1().RESTClient().Get().AbsPath(urlPath).DoRaw(context.TODO())
   380  	if err != nil {
   381  		t.Fatalf("Failed to retrieve object: %v", err)
   382  	}
   383  	obj := &unstructured.Unstructured{}
   384  	if err := json.Unmarshal(bytes, obj); err != nil {
   385  		t.Fatalf("Error unmarshalling the retrieved object: %v", err)
   386  	}
   387  	return obj
   388  }
   389  
   390  func assertReplicasValue(t *testing.T, obj *unstructured.Unstructured, value int) {
   391  	actualValue, found, err := unstructured.NestedInt64(obj.Object, "spec", "replicas")
   392  
   393  	if err != nil {
   394  		t.Fatalf("Error when retrieving replicas field: %v", err)
   395  	}
   396  	if !found {
   397  		t.Fatalf("Replicas field not found")
   398  	}
   399  
   400  	if int(actualValue) != value {
   401  		t.Fatalf("Expected replicas field value to be %d but got %d", value, actualValue)
   402  	}
   403  }
   404  
   405  func assertReplicasOwnership(t *testing.T, obj *unstructured.Unstructured, fieldManagers ...string) {
   406  	t.Helper()
   407  
   408  	accessor, err := meta.Accessor(obj)
   409  	if err != nil {
   410  		t.Fatalf("Failed to get meta accessor for object: %v", err)
   411  	}
   412  
   413  	seen := make(map[string]bool)
   414  	for _, m := range fieldManagers {
   415  		seen[m] = false
   416  	}
   417  
   418  	for _, managedField := range accessor.GetManagedFields() {
   419  		var entryJSON map[string]interface{}
   420  		if err := json.Unmarshal(managedField.FieldsV1.Raw, &entryJSON); err != nil {
   421  			t.Fatalf("failed to read into json")
   422  		}
   423  
   424  		spec, ok := entryJSON["f:spec"].(map[string]interface{})
   425  		if !ok {
   426  			// continue with the next managedField, as we this field does not hold the spec entry
   427  			continue
   428  		}
   429  
   430  		if _, ok := spec["f:replicas"]; !ok {
   431  			// continue with the next managedField, as we this field does not hold the spec.replicas entry
   432  			continue
   433  		}
   434  
   435  		// check if the manager is one of the ones we expect
   436  		if _, ok := seen[managedField.Manager]; !ok {
   437  			t.Fatalf("Unexpected field manager, found %q, expected to be in: %v", managedField.Manager, seen)
   438  		}
   439  
   440  		seen[managedField.Manager] = true
   441  	}
   442  
   443  	var missingManagers []string
   444  	for manager, managerSeen := range seen {
   445  		if !managerSeen {
   446  			missingManagers = append(missingManagers, manager)
   447  		}
   448  	}
   449  	if len(missingManagers) > 0 {
   450  		t.Fatalf("replicas fields should be owned by %v", missingManagers)
   451  	}
   452  }
   453
View as plain text