capmanagers.go

Documentation: k8s.io/apimachinery/pkg/util/managedfields/internal

     1  /*
     2  Copyright 2019 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 internal
    18  
    19  import (
    20  	"fmt"
    21  	"sort"
    22  
    23  	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    24  	"k8s.io/apimachinery/pkg/runtime"
    25  	"sigs.k8s.io/structured-merge-diff/v4/fieldpath"
    26  )
    27  
    28  type capManagersManager struct {
    29  	fieldManager          Manager
    30  	maxUpdateManagers     int
    31  	oldUpdatesManagerName string
    32  }
    33  
    34  var _ Manager = &capManagersManager{}
    35  
    36  // NewCapManagersManager creates a new wrapped FieldManager which ensures that the number of managers from updates
    37  // does not exceed maxUpdateManagers, by merging some of the oldest entries on each update.
    38  func NewCapManagersManager(fieldManager Manager, maxUpdateManagers int) Manager {
    39  	return &capManagersManager{
    40  		fieldManager:          fieldManager,
    41  		maxUpdateManagers:     maxUpdateManagers,
    42  		oldUpdatesManagerName: "ancient-changes",
    43  	}
    44  }
    45  
    46  // Update implements Manager.
    47  func (f *capManagersManager) Update(liveObj, newObj runtime.Object, managed Managed, manager string) (runtime.Object, Managed, error) {
    48  	object, managed, err := f.fieldManager.Update(liveObj, newObj, managed, manager)
    49  	if err != nil {
    50  		return object, managed, err
    51  	}
    52  	if managed, err = f.capUpdateManagers(managed); err != nil {
    53  		return nil, nil, fmt.Errorf("failed to cap update managers: %v", err)
    54  	}
    55  	return object, managed, nil
    56  }
    57  
    58  // Apply implements Manager.
    59  func (f *capManagersManager) Apply(liveObj, appliedObj runtime.Object, managed Managed, fieldManager string, force bool) (runtime.Object, Managed, error) {
    60  	return f.fieldManager.Apply(liveObj, appliedObj, managed, fieldManager, force)
    61  }
    62  
    63  // capUpdateManagers merges a number of the oldest update entries into versioned buckets,
    64  // such that the number of entries from updates does not exceed f.maxUpdateManagers.
    65  func (f *capManagersManager) capUpdateManagers(managed Managed) (newManaged Managed, err error) {
    66  	// Gather all entries from updates
    67  	updaters := []string{}
    68  	for manager, fields := range managed.Fields() {
    69  		if !fields.Applied() {
    70  			updaters = append(updaters, manager)
    71  		}
    72  	}
    73  	if len(updaters) <= f.maxUpdateManagers {
    74  		return managed, nil
    75  	}
    76  
    77  	// If we have more than the maximum, sort the update entries by time, oldest first.
    78  	sort.Slice(updaters, func(i, j int) bool {
    79  		iTime, jTime, iSeconds, jSeconds := managed.Times()[updaters[i]], managed.Times()[updaters[j]], int64(0), int64(0)
    80  		if iTime != nil {
    81  			iSeconds = iTime.Unix()
    82  		}
    83  		if jTime != nil {
    84  			jSeconds = jTime.Unix()
    85  		}
    86  		if iSeconds != jSeconds {
    87  			return iSeconds < jSeconds
    88  		}
    89  		return updaters[i] < updaters[j]
    90  	})
    91  
    92  	// Merge the oldest updaters with versioned bucket managers until the number of updaters is under the cap
    93  	versionToFirstManager := map[string]string{}
    94  	for i, length := 0, len(updaters); i < len(updaters) && length > f.maxUpdateManagers; i++ {
    95  		manager := updaters[i]
    96  		vs := managed.Fields()[manager]
    97  		time := managed.Times()[manager]
    98  		version := string(vs.APIVersion())
    99  
   100  		// Create a new manager identifier for the versioned bucket entry.
   101  		// The version for this manager comes from the version of the update being merged into the bucket.
   102  		bucket, err := BuildManagerIdentifier(&metav1.ManagedFieldsEntry{
   103  			Manager:    f.oldUpdatesManagerName,
   104  			Operation:  metav1.ManagedFieldsOperationUpdate,
   105  			APIVersion: version,
   106  		})
   107  		if err != nil {
   108  			return managed, fmt.Errorf("failed to create bucket manager for version %v: %v", version, err)
   109  		}
   110  
   111  		// Merge the fieldets if this is not the first time the version was seen.
   112  		// Otherwise just record the manager name in versionToFirstManager
   113  		if first, ok := versionToFirstManager[version]; ok {
   114  			// If the bucket doesn't exists yet, create one.
   115  			if _, ok := managed.Fields()[bucket]; !ok {
   116  				s := managed.Fields()[first]
   117  				delete(managed.Fields(), first)
   118  				managed.Fields()[bucket] = s
   119  			}
   120  
   121  			managed.Fields()[bucket] = fieldpath.NewVersionedSet(vs.Set().Union(managed.Fields()[bucket].Set()), vs.APIVersion(), vs.Applied())
   122  			delete(managed.Fields(), manager)
   123  			length--
   124  
   125  			// Use the time from the update being merged into the bucket, since it is more recent.
   126  			managed.Times()[bucket] = time
   127  		} else {
   128  			versionToFirstManager[version] = manager
   129  		}
   130  	}
   131  
   132  	return managed, nil
   133  }
   134
View as plain text