diff --git a/test/integration/scheduler_perf/create.go b/test/integration/scheduler_perf/create.go index 15d0973bbf3..d44816ec8b7 100644 --- a/test/integration/scheduler_perf/create.go +++ b/test/integration/scheduler_perf/create.go @@ -104,22 +104,13 @@ func (c *createAny) create(tCtx ktesting.TContext, env map[string]any) { // Not caching the discovery result isn't very efficient, but good enough when // createAny isn't done often. - discoveryCache := memory.NewMemCacheClient(tCtx.Client().Discovery()) - restMapper := restmapper.NewDeferredDiscoveryRESTMapper(discoveryCache) - gv, err := schema.ParseGroupVersion(obj.GetAPIVersion()) + mapping, err := restMappingFromUnstructuredObj(tCtx, obj) if err != nil { - tCtx.Fatalf("%s: extract group+version from object %q: %v", c.TemplatePath, klog.KObj(obj), err) + tCtx.Fatalf("%s: %v", c.TemplatePath, err) } - gk := schema.GroupKind{Group: gv.Group, Kind: obj.GetKind()} + resourceClient := tCtx.Dynamic().Resource(mapping.Resource) create := func() error { - mapping, err := restMapper.RESTMapping(gk, gv.Version) - if err != nil { - // Cached mapping might be stale, refresh on next try. - restMapper.Reset() - return fmt.Errorf("map %q to resource: %v", gk, err) - } - resourceClient := tCtx.Dynamic().Resource(mapping.Resource) options := metav1.CreateOptions{ // If the YAML input is invalid, then we want the // apiserver to tell us via an error. This can @@ -129,12 +120,12 @@ func (c *createAny) create(tCtx ktesting.TContext, env map[string]any) { } if c.Namespace != "" { if mapping.Scope.Name() != meta.RESTScopeNameNamespace { - return fmt.Errorf("namespace %q set for %q, but %q has scope %q", c.Namespace, c.TemplatePath, gk, mapping.Scope.Name()) + return fmt.Errorf("namespace %q set for %q, but %q has scope %q", c.Namespace, c.TemplatePath, mapping.GroupVersionKind, mapping.Scope.Name()) } _, err = resourceClient.Namespace(c.Namespace).Create(tCtx, obj, options) } else { if mapping.Scope.Name() != meta.RESTScopeNameRoot { - return fmt.Errorf("namespace not set for %q, but %q has scope %q", c.TemplatePath, gk, mapping.Scope.Name()) + return fmt.Errorf("namespace not set for %q, but %q has scope %q", c.TemplatePath, mapping.GroupVersionKind, mapping.Scope.Name()) } _, err = resourceClient.Create(tCtx, obj, options) } @@ -175,3 +166,21 @@ func getSpecFromTextTemplateFile(path string, env map[string]any, spec interface return yaml.UnmarshalStrict(buffer.Bytes(), spec) } + +func restMappingFromUnstructuredObj(tCtx ktesting.TContext, obj *unstructured.Unstructured) (*meta.RESTMapping, error) { + discoveryCache := memory.NewMemCacheClient(tCtx.Client().Discovery()) + restMapper := restmapper.NewDeferredDiscoveryRESTMapper(discoveryCache) + gv, err := schema.ParseGroupVersion(obj.GetAPIVersion()) + if err != nil { + return nil, fmt.Errorf("extract group+version from object %q: %w", klog.KObj(obj), err) + } + gk := schema.GroupKind{Group: gv.Group, Kind: obj.GetKind()} + + mapping, err := restMapper.RESTMapping(gk, gv.Version) + if err != nil { + // Cached mapping might be stale, refresh on next try. + restMapper.Reset() + return nil, fmt.Errorf("failed mapping %q to resource: %w", gk, err) + } + return mapping, nil +} diff --git a/test/integration/scheduler_perf/scheduler_perf.go b/test/integration/scheduler_perf/scheduler_perf.go index 071916c460b..5b97d69275d 100644 --- a/test/integration/scheduler_perf/scheduler_perf.go +++ b/test/integration/scheduler_perf/scheduler_perf.go @@ -86,6 +86,7 @@ const ( createResourceClaimsOpcode operationCode = "createResourceClaims" createResourceDriverOpcode operationCode = "createResourceDriver" churnOpcode operationCode = "churn" + updateAnyOpcode operationCode = "updateAny" barrierOpcode operationCode = "barrier" sleepOpcode operationCode = "sleep" startCollectingMetricsOpcode operationCode = "startCollectingMetrics" @@ -437,6 +438,7 @@ func (op *op) UnmarshalJSON(b []byte) error { createResourceClaimsOpcode: &createResourceClaimsOp{}, createResourceDriverOpcode: &createResourceDriverOp{}, churnOpcode: &churnOp{}, + updateAnyOpcode: &updateAny{}, barrierOpcode: &barrierOp{}, sleepOpcode: &sleepOp{}, startCollectingMetricsOpcode: &startCollectingMetricsOp{}, diff --git a/test/integration/scheduler_perf/update.go b/test/integration/scheduler_perf/update.go new file mode 100644 index 00000000000..ceae7868ce0 --- /dev/null +++ b/test/integration/scheduler_perf/update.go @@ -0,0 +1,162 @@ +/* +Copyright 2024 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 benchmark + +import ( + "fmt" + "time" + + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/kubernetes/test/utils/ktesting" + "k8s.io/utils/ptr" +) + +// updateAny defines an op where some object gets updated from a YAML file. +// The nameset can be specified. +type updateAny struct { + // Must match updateAny. + Opcode operationCode + // Namespace the object should be updated in. Must be empty for cluster-scoped objects. + Namespace string + // Path to spec file describing the object to update. + // This will be processed with text/template. + // .Index will be in the range [0, Count-1] when updating + // more than one object. .Count is the total number of objects. + TemplatePath string + // Count determines how many objects get updated. Defaults to 1 if unset. + Count *int + // Template parameter for Count. + CountParam string + // Number of objects to be updated per second. + // If set to 0, all objects are updated at once. + // Optional + UpdatePerSecond int + // Internal field of the struct used for caching the mapping. + cachedMapping *meta.RESTMapping +} + +var _ runnableOp = &updateAny{} + +func (c *updateAny) isValid(allowParameterization bool) error { + if c.TemplatePath == "" { + return fmt.Errorf("TemplatePath must be set") + } + if c.UpdatePerSecond < 0 { + return fmt.Errorf("invalid UpdatePerSecond=%d; should be non-negative", c.UpdatePerSecond) + } + // The namespace can only be checked during later because we don't know yet + // whether the object is namespaced or cluster-scoped. + return nil +} + +func (c *updateAny) collectsMetrics() bool { + return false +} + +func (c updateAny) patchParams(w *workload) (realOp, error) { + if c.CountParam != "" { + count, err := w.Params.get(c.CountParam[1:]) + if err != nil { + return nil, err + } + c.Count = ptr.To(count) + } + c.cachedMapping = nil + return &c, c.isValid(false) +} + +func (c *updateAny) requiredNamespaces() []string { + if c.Namespace == "" { + return nil + } + return []string{c.Namespace} +} + +func (c *updateAny) run(tCtx ktesting.TContext) { + count := 1 + if c.Count != nil { + count = *c.Count + } + + if c.UpdatePerSecond == 0 { + for index := 0; index < count; index++ { + err := c.update(tCtx, map[string]any{"Index": index, "Count": count}) + if err != nil { + tCtx.Fatalf("Failed to update object: %w", err) + } + } + return + } + + ticker := time.NewTicker(time.Second / time.Duration(c.UpdatePerSecond)) + defer ticker.Stop() + for index := 0; index < count; index++ { + select { + case <-ticker.C: + err := c.update(tCtx, map[string]any{"Index": index, "Count": count}) + if err != nil { + tCtx.Fatalf("Failed to update object: %w", err) + } + case <-tCtx.Done(): + return + } + } +} + +func (c *updateAny) update(tCtx ktesting.TContext, env map[string]any) error { + var obj *unstructured.Unstructured + if err := getSpecFromTextTemplateFile(c.TemplatePath, env, &obj); err != nil { + return fmt.Errorf("%s: parsing failed: %w", c.TemplatePath, err) + } + + if c.cachedMapping == nil { + mapping, err := restMappingFromUnstructuredObj(tCtx, obj) + if err != nil { + return err + } + c.cachedMapping = mapping + } + resourceClient := tCtx.Dynamic().Resource(c.cachedMapping.Resource) + + options := metav1.UpdateOptions{ + // If the YAML input is invalid, then we want the + // apiserver to tell us via an error. This can + // happen because decoding into an unstructured object + // doesn't validate. + FieldValidation: "Strict", + } + if c.Namespace != "" { + if c.cachedMapping.Scope.Name() != meta.RESTScopeNameNamespace { + return fmt.Errorf("namespace %q set for %q, but %q has scope %q", c.Namespace, c.TemplatePath, c.cachedMapping.GroupVersionKind, c.cachedMapping.Scope.Name()) + } + _, err := resourceClient.Namespace(c.Namespace).Update(tCtx, obj, options) + if err != nil { + return fmt.Errorf("failed to update object in namespace %q: %w", c.Namespace, err) + } + return nil + } + if c.cachedMapping.Scope.Name() != meta.RESTScopeNameRoot { + return fmt.Errorf("namespace not set for %q, but %q has scope %q", c.TemplatePath, c.cachedMapping.GroupVersionKind, c.cachedMapping.Scope.Name()) + } + _, err := resourceClient.Update(tCtx, obj, options) + if err != nil { + return fmt.Errorf("failed to update object: %w", err) + } + return nil +}