...
▽

Source file src/k8s.io/kubernetes/pkg/controller/deployment/util/deployment_util.go

Documentation: k8s.io/kubernetes/pkg/controller/deployment/util 

     1  /*
     2  Copyright 2016 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 util
    18  
    19  import (
    20  	"context"
    21  	"fmt"
    22  	"math"
    23  	"sort"
    24  	"strconv"
    25  	"strings"
    26  	"time"
    27  
    28  	apps "k8s.io/api/apps/v1"
    29  	v1 "k8s.io/api/core/v1"
    30  	apiequality "k8s.io/apimachinery/pkg/api/equality"
    31  	"k8s.io/apimachinery/pkg/api/meta"
    32  	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    33  	"k8s.io/apimachinery/pkg/labels"
    34  	"k8s.io/apimachinery/pkg/runtime"
    35  	"k8s.io/apimachinery/pkg/types"
    36  	intstrutil "k8s.io/apimachinery/pkg/util/intstr"
    37  	"k8s.io/apimachinery/pkg/util/wait"
    38  	appsclient "k8s.io/client-go/kubernetes/typed/apps/v1"
    39  	appslisters "k8s.io/client-go/listers/apps/v1"
    40  	"k8s.io/klog/v2"
    41  	"k8s.io/kubernetes/pkg/controller"
    42  	labelsutil "k8s.io/kubernetes/pkg/util/labels"
    43  	"k8s.io/utils/integer"
    44  )
    45  
    46  const (
    47  	// RevisionAnnotation is the revision annotation of a deployment's replica sets which records its rollout sequence
    48  	RevisionAnnotation = "deployment.kubernetes.io/revision"
    49  	// RevisionHistoryAnnotation maintains the history of all old revisions that a replica set has served for a deployment.
    50  	RevisionHistoryAnnotation = "deployment.kubernetes.io/revision-history"
    51  	// DesiredReplicasAnnotation is the desired replicas for a deployment recorded as an annotation
    52  	// in its replica sets. Helps in separating scaling events from the rollout process and for
    53  	// determining if the new replica set for a deployment is really saturated.
    54  	DesiredReplicasAnnotation = "deployment.kubernetes.io/desired-replicas"
    55  	// MaxReplicasAnnotation is the maximum replicas a deployment can have at a given point, which
    56  	// is deployment.spec.replicas + maxSurge. Used by the underlying replica sets to estimate their
    57  	// proportions in case the deployment has surge replicas.
    58  	MaxReplicasAnnotation = "deployment.kubernetes.io/max-replicas"
    59  
    60  	// RollbackRevisionNotFound is not found rollback event reason
    61  	RollbackRevisionNotFound = "DeploymentRollbackRevisionNotFound"
    62  	// RollbackTemplateUnchanged is the template unchanged rollback event reason
    63  	RollbackTemplateUnchanged = "DeploymentRollbackTemplateUnchanged"
    64  	// RollbackDone is the done rollback event reason
    65  	RollbackDone = "DeploymentRollback"
    66  
    67  	// Reasons for deployment conditions
    68  	//
    69  	// Progressing:
    70  
    71  	// ReplicaSetUpdatedReason is added in a deployment when one of its replica sets is updated as part
    72  	// of the rollout process.
    73  	ReplicaSetUpdatedReason = "ReplicaSetUpdated"
    74  	// FailedRSCreateReason is added in a deployment when it cannot create a new replica set.
    75  	FailedRSCreateReason = "ReplicaSetCreateError"
    76  	// NewReplicaSetReason is added in a deployment when it creates a new replica set.
    77  	NewReplicaSetReason = "NewReplicaSetCreated"
    78  	// FoundNewRSReason is added in a deployment when it adopts an existing replica set.
    79  	FoundNewRSReason = "FoundNewReplicaSet"
    80  	// NewRSAvailableReason is added in a deployment when its newest replica set is made available
    81  	// ie. the number of new pods that have passed readiness checks and run for at least minReadySeconds
    82  	// is at least the minimum available pods that need to run for the deployment.
    83  	NewRSAvailableReason = "NewReplicaSetAvailable"
    84  	// TimedOutReason is added in a deployment when its newest replica set fails to show any progress
    85  	// within the given deadline (progressDeadlineSeconds).
    86  	TimedOutReason = "ProgressDeadlineExceeded"
    87  	// PausedDeployReason is added in a deployment when it is paused. Lack of progress shouldn't be
    88  	// estimated once a deployment is paused.
    89  	PausedDeployReason = "DeploymentPaused"
    90  	// ResumedDeployReason is added in a deployment when it is resumed. Useful for not failing accidentally
    91  	// deployments that paused amidst a rollout and are bounded by a deadline.
    92  	ResumedDeployReason = "DeploymentResumed"
    93  	//
    94  	// Available:
    95  
    96  	// MinimumReplicasAvailable is added in a deployment when it has its minimum replicas required available.
    97  	MinimumReplicasAvailable = "MinimumReplicasAvailable"
    98  	// MinimumReplicasUnavailable is added in a deployment when it doesn't have the minimum required replicas
    99  	// available.
   100  	MinimumReplicasUnavailable = "MinimumReplicasUnavailable"
   101  )
   102  
   103  // NewDeploymentCondition creates a new deployment condition.
   104  func NewDeploymentCondition(condType apps.DeploymentConditionType, status v1.ConditionStatus, reason, message string) *apps.DeploymentCondition {
   105  	return &apps.DeploymentCondition{
   106  		Type:               condType,
   107  		Status:             status,
   108  		LastUpdateTime:     metav1.Now(),
   109  		LastTransitionTime: metav1.Now(),
   110  		Reason:             reason,
   111  		Message:            message,
   112  	}
   113  }
   114  
   115  // GetDeploymentCondition returns the condition with the provided type.
   116  func GetDeploymentCondition(status apps.DeploymentStatus, condType apps.DeploymentConditionType) *apps.DeploymentCondition {
   117  	for i := range status.Conditions {
   118  		c := status.Conditions[i]
   119  		if c.Type == condType {
   120  			return &c
   121  		}
   122  	}
   123  	return nil
   124  }
   125  
   126  // SetDeploymentCondition updates the deployment to include the provided condition. If the condition that
   127  // we are about to add already exists and has the same status and reason then we are not going to update.
   128  func SetDeploymentCondition(status *apps.DeploymentStatus, condition apps.DeploymentCondition) {
   129  	currentCond := GetDeploymentCondition(*status, condition.Type)
   130  	if currentCond != nil && currentCond.Status == condition.Status && currentCond.Reason == condition.Reason {
   131  		return
   132  	}
   133  	// Do not update lastTransitionTime if the status of the condition doesn't change.
   134  	if currentCond != nil && currentCond.Status == condition.Status {
   135  		condition.LastTransitionTime = currentCond.LastTransitionTime
   136  	}
   137  	newConditions := filterOutCondition(status.Conditions, condition.Type)
   138  	status.Conditions = append(newConditions, condition)
   139  }
   140  
   141  // RemoveDeploymentCondition removes the deployment condition with the provided type.
   142  func RemoveDeploymentCondition(status *apps.DeploymentStatus, condType apps.DeploymentConditionType) {
   143  	status.Conditions = filterOutCondition(status.Conditions, condType)
   144  }
   145  
   146  // filterOutCondition returns a new slice of deployment conditions without conditions with the provided type.
   147  func filterOutCondition(conditions []apps.DeploymentCondition, condType apps.DeploymentConditionType) []apps.DeploymentCondition {
   148  	var newConditions []apps.DeploymentCondition
   149  	for _, c := range conditions {
   150  		if c.Type == condType {
   151  			continue
   152  		}
   153  		newConditions = append(newConditions, c)
   154  	}
   155  	return newConditions
   156  }
   157  
   158  // ReplicaSetToDeploymentCondition converts a replica set condition into a deployment condition.
   159  // Useful for promoting replica set failure conditions into deployments.
   160  func ReplicaSetToDeploymentCondition(cond apps.ReplicaSetCondition) apps.DeploymentCondition {
   161  	return apps.DeploymentCondition{
   162  		Type:               apps.DeploymentConditionType(cond.Type),
   163  		Status:             cond.Status,
   164  		LastTransitionTime: cond.LastTransitionTime,
   165  		LastUpdateTime:     cond.LastTransitionTime,
   166  		Reason:             cond.Reason,
   167  		Message:            cond.Message,
   168  	}
   169  }
   170  
   171  // SetDeploymentRevision updates the revision for a deployment.
   172  func SetDeploymentRevision(deployment *apps.Deployment, revision string) bool {
   173  	updated := false
   174  
   175  	if deployment.Annotations == nil {
   176  		deployment.Annotations = make(map[string]string)
   177  	}
   178  	if deployment.Annotations[RevisionAnnotation] != revision {
   179  		deployment.Annotations[RevisionAnnotation] = revision
   180  		updated = true
   181  	}
   182  
   183  	return updated
   184  }
   185  
   186  // MaxRevision finds the highest revision in the replica sets
   187  func MaxRevision(logger klog.Logger, allRSs []*apps.ReplicaSet) int64 {
   188  	max := int64(0)
   189  	for _, rs := range allRSs {
   190  		if v, err := Revision(rs); err != nil {
   191  			// Skip the replica sets when it failed to parse their revision information
   192  			logger.V(4).Info("Couldn't parse revision for replica set, deployment controller will skip it when reconciling revisions", "replicaSet", klog.KObj(rs), "err", err)
   193  		} else if v > max {
   194  			max = v
   195  		}
   196  	}
   197  	return max
   198  }
   199  
   200  // LastRevision finds the second max revision number in all replica sets (the last revision)
   201  func LastRevision(logger klog.Logger, allRSs []*apps.ReplicaSet) int64 {
   202  	max, secMax := int64(0), int64(0)
   203  	for _, rs := range allRSs {
   204  		if v, err := Revision(rs); err != nil {
   205  			// Skip the replica sets when it failed to parse their revision information
   206  			logger.V(4).Info("Couldn't parse revision for replica set, deployment controller will skip it when reconciling revisions", "replicaSet", klog.KObj(rs), "err", err)
   207  		} else if v >= max {
   208  			secMax = max
   209  			max = v
   210  		} else if v > secMax {
   211  			secMax = v
   212  		}
   213  	}
   214  	return secMax
   215  }
   216  
   217  // Revision returns the revision number of the input object.
   218  func Revision(obj runtime.Object) (int64, error) {
   219  	acc, err := meta.Accessor(obj)
   220  	if err != nil {
   221  		return 0, err
   222  	}
   223  	v, ok := acc.GetAnnotations()[RevisionAnnotation]
   224  	if !ok {
   225  		return 0, nil
   226  	}
   227  	return strconv.ParseInt(v, 10, 64)
   228  }
   229  
   230  // SetNewReplicaSetAnnotations sets new replica set's annotations appropriately by updating its revision and
   231  // copying required deployment annotations to it; it returns true if replica set's annotation is changed.
   232  func SetNewReplicaSetAnnotations(ctx context.Context, deployment *apps.Deployment, newRS *apps.ReplicaSet, newRevision string, exists bool, revHistoryLimitInChars int) bool {
   233  	logger := klog.FromContext(ctx)
   234  	// First, copy deployment's annotations (except for apply and revision annotations)
   235  	annotationChanged := copyDeploymentAnnotationsToReplicaSet(deployment, newRS)
   236  	// Then, update replica set's revision annotation
   237  	if newRS.Annotations == nil {
   238  		newRS.Annotations = make(map[string]string)
   239  	}
   240  	oldRevision, ok := newRS.Annotations[RevisionAnnotation]
   241  	// The newRS's revision should be the greatest among all RSes. Usually, its revision number is newRevision (the max revision number
   242  	// of all old RSes + 1). However, it's possible that some of the old RSes are deleted after the newRS revision being updated, and
   243  	// newRevision becomes smaller than newRS's revision. We should only update newRS revision when it's smaller than newRevision.
   244  
   245  	oldRevisionInt, err := strconv.ParseInt(oldRevision, 10, 64)
   246  	if err != nil {
   247  		if oldRevision != "" {
   248  			logger.Info("Updating replica set revision OldRevision not int", "err", err)
   249  			return false
   250  		}
   251  		//If the RS annotation is empty then initialise it to 0
   252  		oldRevisionInt = 0
   253  	}
   254  	newRevisionInt, err := strconv.ParseInt(newRevision, 10, 64)
   255  	if err != nil {
   256  		logger.Info("Updating replica set revision NewRevision not int", "err", err)
   257  		return false
   258  	}
   259  	if oldRevisionInt < newRevisionInt {
   260  		newRS.Annotations[RevisionAnnotation] = newRevision
   261  		annotationChanged = true
   262  		logger.V(4).Info("Updating replica set revision", "replicaSet", klog.KObj(newRS), "newRevision", newRevision)
   263  	}
   264  	// If a revision annotation already existed and this replica set was updated with a new revision
   265  	// then that means we are rolling back to this replica set. We need to preserve the old revisions
   266  	// for historical information.
   267  	if ok && oldRevisionInt < newRevisionInt {
   268  		revisionHistoryAnnotation := newRS.Annotations[RevisionHistoryAnnotation]
   269  		oldRevisions := strings.Split(revisionHistoryAnnotation, ",")
   270  		if len(oldRevisions[0]) == 0 {
   271  			newRS.Annotations[RevisionHistoryAnnotation] = oldRevision
   272  		} else {
   273  			totalLen := len(revisionHistoryAnnotation) + len(oldRevision) + 1
   274  			// index for the starting position in oldRevisions
   275  			start := 0
   276  			for totalLen > revHistoryLimitInChars && start < len(oldRevisions) {
   277  				totalLen = totalLen - len(oldRevisions[start]) - 1
   278  				start++
   279  			}
   280  			if totalLen <= revHistoryLimitInChars {
   281  				oldRevisions = append(oldRevisions[start:], oldRevision)
   282  				newRS.Annotations[RevisionHistoryAnnotation] = strings.Join(oldRevisions, ",")
   283  			} else {
   284  				logger.Info("Not appending revision due to revision history length limit reached", "revisionHistoryLimit", revHistoryLimitInChars)
   285  			}
   286  		}
   287  	}
   288  	// If the new replica set is about to be created, we need to add replica annotations to it.
   289  	if !exists && SetReplicasAnnotations(newRS, *(deployment.Spec.Replicas), *(deployment.Spec.Replicas)+MaxSurge(*deployment)) {
   290  		annotationChanged = true
   291  	}
   292  	return annotationChanged
   293  }
   294  
   295  var annotationsToSkip = map[string]bool{
   296  	v1.LastAppliedConfigAnnotation: true,
   297  	RevisionAnnotation:             true,
   298  	RevisionHistoryAnnotation:      true,
   299  	DesiredReplicasAnnotation:      true,
   300  	MaxReplicasAnnotation:          true,
   301  	apps.DeprecatedRollbackTo:      true,
   302  }
   303  
   304  // skipCopyAnnotation returns true if we should skip copying the annotation with the given annotation key
   305  // TODO: How to decide which annotations should / should not be copied?
   306  //
   307  // See https://github.com/kubernetes/kubernetes/pull/20035#issuecomment-179558615
   308  func skipCopyAnnotation(key string) bool {
   309  	return annotationsToSkip[key]
   310  }
   311  
   312  // copyDeploymentAnnotationsToReplicaSet copies deployment's annotations to replica set's annotations,
   313  // and returns true if replica set's annotation is changed.
   314  // Note that apply and revision annotations are not copied.
   315  func copyDeploymentAnnotationsToReplicaSet(deployment *apps.Deployment, rs *apps.ReplicaSet) bool {
   316  	rsAnnotationsChanged := false
   317  	if rs.Annotations == nil {
   318  		rs.Annotations = make(map[string]string)
   319  	}
   320  	for k, v := range deployment.Annotations {
   321  		// newRS revision is updated automatically in getNewReplicaSet, and the deployment's revision number is then updated
   322  		// by copying its newRS revision number. We should not copy deployment's revision to its newRS, since the update of
   323  		// deployment revision number may fail (revision becomes stale) and the revision number in newRS is more reliable.
   324  		if _, exist := rs.Annotations[k]; skipCopyAnnotation(k) || (exist && rs.Annotations[k] == v) {
   325  			continue
   326  		}
   327  		rs.Annotations[k] = v
   328  		rsAnnotationsChanged = true
   329  	}
   330  	return rsAnnotationsChanged
   331  }
   332  
   333  // SetDeploymentAnnotationsTo sets deployment's annotations as given RS's annotations.
   334  // This action should be done if and only if the deployment is rolling back to this rs.
   335  // Note that apply and revision annotations are not changed.
   336  func SetDeploymentAnnotationsTo(deployment *apps.Deployment, rollbackToRS *apps.ReplicaSet) {
   337  	deployment.Annotations = getSkippedAnnotations(deployment.Annotations)
   338  	for k, v := range rollbackToRS.Annotations {
   339  		if !skipCopyAnnotation(k) {
   340  			deployment.Annotations[k] = v
   341  		}
   342  	}
   343  }
   344  
   345  func getSkippedAnnotations(annotations map[string]string) map[string]string {
   346  	skippedAnnotations := make(map[string]string)
   347  	for k, v := range annotations {
   348  		if skipCopyAnnotation(k) {
   349  			skippedAnnotations[k] = v
   350  		}
   351  	}
   352  	return skippedAnnotations
   353  }
   354  
   355  // FindActiveOrLatest returns the only active or the latest replica set in case there is at most one active
   356  // replica set. If there are more active replica sets, then we should proportionally scale them.
   357  func FindActiveOrLatest(newRS *apps.ReplicaSet, oldRSs []*apps.ReplicaSet) *apps.ReplicaSet {
   358  	if newRS == nil && len(oldRSs) == 0 {
   359  		return nil
   360  	}
   361  
   362  	sort.Sort(sort.Reverse(controller.ReplicaSetsByCreationTimestamp(oldRSs)))
   363  	allRSs := controller.FilterActiveReplicaSets(append(oldRSs, newRS))
   364  
   365  	switch len(allRSs) {
   366  	case 0:
   367  		// If there is no active replica set then we should return the newest.
   368  		if newRS != nil {
   369  			return newRS
   370  		}
   371  		return oldRSs[0]
   372  	case 1:
   373  		return allRSs[0]
   374  	default:
   375  		return nil
   376  	}
   377  }
   378  
   379  // GetDesiredReplicasAnnotation returns the number of desired replicas
   380  func GetDesiredReplicasAnnotation(logger klog.Logger, rs *apps.ReplicaSet) (int32, bool) {
   381  	return getIntFromAnnotation(logger, rs, DesiredReplicasAnnotation)
   382  }
   383  
   384  func getMaxReplicasAnnotation(logger klog.Logger, rs *apps.ReplicaSet) (int32, bool) {
   385  	return getIntFromAnnotation(logger, rs, MaxReplicasAnnotation)
   386  }
   387  
   388  func getIntFromAnnotation(logger klog.Logger, rs *apps.ReplicaSet, annotationKey string) (int32, bool) {
   389  	annotationValue, ok := rs.Annotations[annotationKey]
   390  	if !ok {
   391  		return int32(0), false
   392  	}
   393  	intValue, err := strconv.Atoi(annotationValue)
   394  	if err != nil {
   395  		logger.V(2).Info("Could not convert the value with annotation key for the replica set", "annotationValue", annotationValue, "annotationKey", annotationKey, "replicaSet", klog.KObj(rs))
   396  		return int32(0), false
   397  	}
   398  	return int32(intValue), true
   399  }
   400  
   401  // SetReplicasAnnotations sets the desiredReplicas and maxReplicas into the annotations
   402  func SetReplicasAnnotations(rs *apps.ReplicaSet, desiredReplicas, maxReplicas int32) bool {
   403  	updated := false
   404  	if rs.Annotations == nil {
   405  		rs.Annotations = make(map[string]string)
   406  	}
   407  	desiredString := fmt.Sprintf("%d", desiredReplicas)
   408  	if hasString := rs.Annotations[DesiredReplicasAnnotation]; hasString != desiredString {
   409  		rs.Annotations[DesiredReplicasAnnotation] = desiredString
   410  		updated = true
   411  	}
   412  	maxString := fmt.Sprintf("%d", maxReplicas)
   413  	if hasString := rs.Annotations[MaxReplicasAnnotation]; hasString != maxString {
   414  		rs.Annotations[MaxReplicasAnnotation] = maxString
   415  		updated = true
   416  	}
   417  	return updated
   418  }
   419  
   420  // ReplicasAnnotationsNeedUpdate return true if ReplicasAnnotations need to be updated
   421  func ReplicasAnnotationsNeedUpdate(rs *apps.ReplicaSet, desiredReplicas, maxReplicas int32) bool {
   422  	if rs.Annotations == nil {
   423  		return true
   424  	}
   425  	desiredString := fmt.Sprintf("%d", desiredReplicas)
   426  	if hasString := rs.Annotations[DesiredReplicasAnnotation]; hasString != desiredString {
   427  		return true
   428  	}
   429  	maxString := fmt.Sprintf("%d", maxReplicas)
   430  	if hasString := rs.Annotations[MaxReplicasAnnotation]; hasString != maxString {
   431  		return true
   432  	}
   433  	return false
   434  }
   435  
   436  // MaxUnavailable returns the maximum unavailable pods a rolling deployment can take.
   437  func MaxUnavailable(deployment apps.Deployment) int32 {
   438  	if !IsRollingUpdate(&deployment) || *(deployment.Spec.Replicas) == 0 {
   439  		return int32(0)
   440  	}
   441  	// Error caught by validation
   442  	_, maxUnavailable, _ := ResolveFenceposts(deployment.Spec.Strategy.RollingUpdate.MaxSurge, deployment.Spec.Strategy.RollingUpdate.MaxUnavailable, *(deployment.Spec.Replicas))
   443  	if maxUnavailable > *deployment.Spec.Replicas {
   444  		return *deployment.Spec.Replicas
   445  	}
   446  	return maxUnavailable
   447  }
   448  
   449  // MinAvailable returns the minimum available pods of a given deployment
   450  func MinAvailable(deployment *apps.Deployment) int32 {
   451  	if !IsRollingUpdate(deployment) {
   452  		return int32(0)
   453  	}
   454  	return *(deployment.Spec.Replicas) - MaxUnavailable(*deployment)
   455  }
   456  
   457  // MaxSurge returns the maximum surge pods a rolling deployment can take.
   458  func MaxSurge(deployment apps.Deployment) int32 {
   459  	if !IsRollingUpdate(&deployment) {
   460  		return int32(0)
   461  	}
   462  	// Error caught by validation
   463  	maxSurge, _, _ := ResolveFenceposts(deployment.Spec.Strategy.RollingUpdate.MaxSurge, deployment.Spec.Strategy.RollingUpdate.MaxUnavailable, *(deployment.Spec.Replicas))
   464  	return maxSurge
   465  }
   466  
   467  // GetProportion will estimate the proportion for the provided replica set using 1. the current size
   468  // of the parent deployment, 2. the replica count that needs be added on the replica sets of the
   469  // deployment, and 3. the total replicas added in the replica sets of the deployment so far.
   470  func GetProportion(logger klog.Logger, rs *apps.ReplicaSet, d apps.Deployment, deploymentReplicasToAdd, deploymentReplicasAdded int32) int32 {
   471  	if rs == nil || *(rs.Spec.Replicas) == 0 || deploymentReplicasToAdd == 0 || deploymentReplicasToAdd == deploymentReplicasAdded {
   472  		return int32(0)
   473  	}
   474  
   475  	rsFraction := getReplicaSetFraction(logger, *rs, d)
   476  	allowed := deploymentReplicasToAdd - deploymentReplicasAdded
   477  
   478  	if deploymentReplicasToAdd > 0 {
   479  		// Use the minimum between the replica set fraction and the maximum allowed replicas
   480  		// when scaling up. This way we ensure we will not scale up more than the allowed
   481  		// replicas we can add.
   482  		return min(rsFraction, allowed)
   483  	}
   484  	// Use the maximum between the replica set fraction and the maximum allowed replicas
   485  	// when scaling down. This way we ensure we will not scale down more than the allowed
   486  	// replicas we can remove.
   487  	return max(rsFraction, allowed)
   488  }
   489  
   490  // getReplicaSetFraction estimates the fraction of replicas a replica set can have in
   491  // 1. a scaling event during a rollout or 2. when scaling a paused deployment.
   492  func getReplicaSetFraction(logger klog.Logger, rs apps.ReplicaSet, d apps.Deployment) int32 {
   493  	// If we are scaling down to zero then the fraction of this replica set is its whole size (negative)
   494  	if *(d.Spec.Replicas) == int32(0) {
   495  		return -*(rs.Spec.Replicas)
   496  	}
   497  
   498  	deploymentReplicas := *(d.Spec.Replicas) + MaxSurge(d)
   499  	annotatedReplicas, ok := getMaxReplicasAnnotation(logger, &rs)
   500  	if !ok {
   501  		// If we cannot find the annotation then fallback to the current deployment size. Note that this
   502  		// will not be an accurate proportion estimation in case other replica sets have different values
   503  		// which means that the deployment was scaled at some point but we at least will stay in limits
   504  		// due to the min-max comparisons in getProportion.
   505  		annotatedReplicas = d.Status.Replicas
   506  	}
   507  
   508  	// We should never proportionally scale up from zero which means rs.spec.replicas and annotatedReplicas
   509  	// will never be zero here.
   510  	newRSsize := (float64(*(rs.Spec.Replicas) * deploymentReplicas)) / float64(annotatedReplicas)
   511  	return integer.RoundToInt32(newRSsize) - *(rs.Spec.Replicas)
   512  }
   513  
   514  // RsListFromClient returns an rsListFunc that wraps the given client.
   515  func RsListFromClient(c appsclient.AppsV1Interface) RsListFunc {
   516  	return func(namespace string, options metav1.ListOptions) ([]*apps.ReplicaSet, error) {
   517  		rsList, err := c.ReplicaSets(namespace).List(context.TODO(), options)
   518  		if err != nil {
   519  			return nil, err
   520  		}
   521  		var ret []*apps.ReplicaSet
   522  		for i := range rsList.Items {
   523  			ret = append(ret, &rsList.Items[i])
   524  		}
   525  		return ret, err
   526  	}
   527  }
   528  
   529  // TODO: switch RsListFunc and podListFunc to full namespacers
   530  
   531  // RsListFunc returns the ReplicaSet from the ReplicaSet namespace and the List metav1.ListOptions.
   532  type RsListFunc func(string, metav1.ListOptions) ([]*apps.ReplicaSet, error)
   533  
   534  // podListFunc returns the PodList from the Pod namespace and the List metav1.ListOptions.
   535  type podListFunc func(string, metav1.ListOptions) (*v1.PodList, error)
   536  
   537  // ListReplicaSets returns a slice of RSes the given deployment targets.
   538  // Note that this does NOT attempt to reconcile ControllerRef (adopt/orphan),
   539  // because only the controller itself should do that.
   540  // However, it does filter out anything whose ControllerRef doesn't match.
   541  func ListReplicaSets(deployment *apps.Deployment, getRSList RsListFunc) ([]*apps.ReplicaSet, error) {
   542  	// TODO: Right now we list replica sets by their labels. We should list them by selector, i.e. the replica set's selector
   543  	//       should be a superset of the deployment's selector, see https://github.com/kubernetes/kubernetes/issues/19830.
   544  	namespace := deployment.Namespace
   545  	selector, err := metav1.LabelSelectorAsSelector(deployment.Spec.Selector)
   546  	if err != nil {
   547  		return nil, err
   548  	}
   549  	options := metav1.ListOptions{LabelSelector: selector.String()}
   550  	all, err := getRSList(namespace, options)
   551  	if err != nil {
   552  		return nil, err
   553  	}
   554  	// Only include those whose ControllerRef matches the Deployment.
   555  	owned := make([]*apps.ReplicaSet, 0, len(all))
   556  	for _, rs := range all {
   557  		if metav1.IsControlledBy(rs, deployment) {
   558  			owned = append(owned, rs)
   559  		}
   560  	}
   561  	return owned, nil
   562  }
   563  
   564  // ListPods returns a list of pods the given deployment targets.
   565  // This needs a list of ReplicaSets for the Deployment,
   566  // which can be found with ListReplicaSets().
   567  // Note that this does NOT attempt to reconcile ControllerRef (adopt/orphan),
   568  // because only the controller itself should do that.
   569  // However, it does filter out anything whose ControllerRef doesn't match.
   570  func ListPods(deployment *apps.Deployment, rsList []*apps.ReplicaSet, getPodList podListFunc) (*v1.PodList, error) {
   571  	namespace := deployment.Namespace
   572  	selector, err := metav1.LabelSelectorAsSelector(deployment.Spec.Selector)
   573  	if err != nil {
   574  		return nil, err
   575  	}
   576  	options := metav1.ListOptions{LabelSelector: selector.String()}
   577  	all, err := getPodList(namespace, options)
   578  	if err != nil {
   579  		return all, err
   580  	}
   581  	// Only include those whose ControllerRef points to a ReplicaSet that is in
   582  	// turn owned by this Deployment.
   583  	rsMap := make(map[types.UID]bool, len(rsList))
   584  	for _, rs := range rsList {
   585  		rsMap[rs.UID] = true
   586  	}
   587  	owned := &v1.PodList{Items: make([]v1.Pod, 0, len(all.Items))}
   588  	for i := range all.Items {
   589  		pod := &all.Items[i]
   590  		controllerRef := metav1.GetControllerOf(pod)
   591  		if controllerRef != nil && rsMap[controllerRef.UID] {
   592  			owned.Items = append(owned.Items, *pod)
   593  		}
   594  	}
   595  	return owned, nil
   596  }
   597  
   598  // EqualIgnoreHash returns true if two given podTemplateSpec are equal, ignoring the diff in value of Labels[pod-template-hash]
   599  // We ignore pod-template-hash because:
   600  //  1. The hash result would be different upon podTemplateSpec API changes
   601  //     (e.g. the addition of a new field will cause the hash code to change)
   602  //  2. The deployment template won't have hash labels
   603  func EqualIgnoreHash(template1, template2 *v1.PodTemplateSpec) bool {
   604  	t1Copy := template1.DeepCopy()
   605  	t2Copy := template2.DeepCopy()
   606  	// Remove hash labels from template.Labels before comparing
   607  	delete(t1Copy.Labels, apps.DefaultDeploymentUniqueLabelKey)
   608  	delete(t2Copy.Labels, apps.DefaultDeploymentUniqueLabelKey)
   609  	return apiequality.Semantic.DeepEqual(t1Copy, t2Copy)
   610  }
   611  
   612  // FindNewReplicaSet returns the new RS this given deployment targets (the one with the same pod template).
   613  func FindNewReplicaSet(deployment *apps.Deployment, rsList []*apps.ReplicaSet) *apps.ReplicaSet {
   614  	sort.Sort(controller.ReplicaSetsByCreationTimestamp(rsList))
   615  	for i := range rsList {
   616  		if EqualIgnoreHash(&rsList[i].Spec.Template, &deployment.Spec.Template) {
   617  			// In rare cases, such as after cluster upgrades, Deployment may end up with
   618  			// having more than one new ReplicaSets that have the same template as its template,
   619  			// see https://github.com/kubernetes/kubernetes/issues/40415
   620  			// We deterministically choose the oldest new ReplicaSet.
   621  			return rsList[i]
   622  		}
   623  	}
   624  	// new ReplicaSet does not exist.
   625  	return nil
   626  }
   627  
   628  // FindOldReplicaSets returns the old replica sets targeted by the given Deployment, with the given slice of RSes.
   629  // Note that the first set of old replica sets doesn't include the ones with no pods, and the second set of old replica sets include all old replica sets.
   630  func FindOldReplicaSets(deployment *apps.Deployment, rsList []*apps.ReplicaSet) ([]*apps.ReplicaSet, []*apps.ReplicaSet) {
   631  	var requiredRSs []*apps.ReplicaSet
   632  	var allRSs []*apps.ReplicaSet
   633  	newRS := FindNewReplicaSet(deployment, rsList)
   634  	for _, rs := range rsList {
   635  		// Filter out new replica set
   636  		if newRS != nil && rs.UID == newRS.UID {
   637  			continue
   638  		}
   639  		allRSs = append(allRSs, rs)
   640  		if *(rs.Spec.Replicas) != 0 {
   641  			requiredRSs = append(requiredRSs, rs)
   642  		}
   643  	}
   644  	return requiredRSs, allRSs
   645  }
   646  
   647  // SetFromReplicaSetTemplate sets the desired PodTemplateSpec from a replica set template to the given deployment.
   648  func SetFromReplicaSetTemplate(deployment *apps.Deployment, template v1.PodTemplateSpec) *apps.Deployment {
   649  	deployment.Spec.Template.ObjectMeta = template.ObjectMeta
   650  	deployment.Spec.Template.Spec = template.Spec
   651  	deployment.Spec.Template.ObjectMeta.Labels = labelsutil.CloneAndRemoveLabel(
   652  		deployment.Spec.Template.ObjectMeta.Labels,
   653  		apps.DefaultDeploymentUniqueLabelKey)
   654  	return deployment
   655  }
   656  
   657  // GetReplicaCountForReplicaSets returns the sum of Replicas of the given replica sets.
   658  func GetReplicaCountForReplicaSets(replicaSets []*apps.ReplicaSet) int32 {
   659  	totalReplicas := int32(0)
   660  	for _, rs := range replicaSets {
   661  		if rs != nil {
   662  			totalReplicas += *(rs.Spec.Replicas)
   663  		}
   664  	}
   665  	return totalReplicas
   666  }
   667  
   668  // GetActualReplicaCountForReplicaSets returns the sum of actual replicas of the given replica sets.
   669  func GetActualReplicaCountForReplicaSets(replicaSets []*apps.ReplicaSet) int32 {
   670  	totalActualReplicas := int32(0)
   671  	for _, rs := range replicaSets {
   672  		if rs != nil {
   673  			totalActualReplicas += rs.Status.Replicas
   674  		}
   675  	}
   676  	return totalActualReplicas
   677  }
   678  
   679  // GetReadyReplicaCountForReplicaSets returns the number of ready pods corresponding to the given replica sets.
   680  func GetReadyReplicaCountForReplicaSets(replicaSets []*apps.ReplicaSet) int32 {
   681  	totalReadyReplicas := int32(0)
   682  	for _, rs := range replicaSets {
   683  		if rs != nil {
   684  			totalReadyReplicas += rs.Status.ReadyReplicas
   685  		}
   686  	}
   687  	return totalReadyReplicas
   688  }
   689  
   690  // GetAvailableReplicaCountForReplicaSets returns the number of available pods corresponding to the given replica sets.
   691  func GetAvailableReplicaCountForReplicaSets(replicaSets []*apps.ReplicaSet) int32 {
   692  	totalAvailableReplicas := int32(0)
   693  	for _, rs := range replicaSets {
   694  		if rs != nil {
   695  			totalAvailableReplicas += rs.Status.AvailableReplicas
   696  		}
   697  	}
   698  	return totalAvailableReplicas
   699  }
   700  
   701  // IsRollingUpdate returns true if the strategy type is a rolling update.
   702  func IsRollingUpdate(deployment *apps.Deployment) bool {
   703  	return deployment.Spec.Strategy.Type == apps.RollingUpdateDeploymentStrategyType
   704  }
   705  
   706  // DeploymentComplete considers a deployment to be complete once all of its desired replicas
   707  // are updated and available, and no old pods are running.
   708  func DeploymentComplete(deployment *apps.Deployment, newStatus *apps.DeploymentStatus) bool {
   709  	return newStatus.UpdatedReplicas == *(deployment.Spec.Replicas) &&
   710  		newStatus.Replicas == *(deployment.Spec.Replicas) &&
   711  		newStatus.AvailableReplicas == *(deployment.Spec.Replicas) &&
   712  		newStatus.ObservedGeneration >= deployment.Generation
   713  }
   714  
   715  // DeploymentProgressing reports progress for a deployment. Progress is estimated by comparing the
   716  // current with the new status of the deployment that the controller is observing. More specifically,
   717  // when new pods are scaled up or become ready or available, or old pods are scaled down, then we
   718  // consider the deployment is progressing.
   719  func DeploymentProgressing(deployment *apps.Deployment, newStatus *apps.DeploymentStatus) bool {
   720  	oldStatus := deployment.Status
   721  
   722  	// Old replicas that need to be scaled down
   723  	oldStatusOldReplicas := oldStatus.Replicas - oldStatus.UpdatedReplicas
   724  	newStatusOldReplicas := newStatus.Replicas - newStatus.UpdatedReplicas
   725  
   726  	return (newStatus.UpdatedReplicas > oldStatus.UpdatedReplicas) ||
   727  		(newStatusOldReplicas < oldStatusOldReplicas) ||
   728  		newStatus.ReadyReplicas > deployment.Status.ReadyReplicas ||
   729  		newStatus.AvailableReplicas > deployment.Status.AvailableReplicas
   730  }
   731  
   732  // used for unit testing
   733  var nowFn = func() time.Time { return time.Now() }
   734  
   735  // DeploymentTimedOut considers a deployment to have timed out once its condition that reports progress
   736  // is older than progressDeadlineSeconds or a Progressing condition with a TimedOutReason reason already
   737  // exists.
   738  func DeploymentTimedOut(ctx context.Context, deployment *apps.Deployment, newStatus *apps.DeploymentStatus) bool {
   739  	if !HasProgressDeadline(deployment) {
   740  		return false
   741  	}
   742  
   743  	// Look for the Progressing condition. If it doesn't exist, we have no base to estimate progress.
   744  	// If it's already set with a TimedOutReason reason, we have already timed out, no need to check
   745  	// again.
   746  	condition := GetDeploymentCondition(*newStatus, apps.DeploymentProgressing)
   747  	if condition == nil {
   748  		return false
   749  	}
   750  	// If the previous condition has been a successful rollout then we shouldn't try to
   751  	// estimate any progress. Scenario:
   752  	//
   753  	// * progressDeadlineSeconds is smaller than the difference between now and the time
   754  	//   the last rollout finished in the past.
   755  	// * the creation of a new ReplicaSet triggers a resync of the Deployment prior to the
   756  	//   cached copy of the Deployment getting updated with the status.condition that indicates
   757  	//   the creation of the new ReplicaSet.
   758  	//
   759  	// The Deployment will be resynced and eventually its Progressing condition will catch
   760  	// up with the state of the world.
   761  	if condition.Reason == NewRSAvailableReason {
   762  		return false
   763  	}
   764  	if condition.Reason == TimedOutReason {
   765  		return true
   766  	}
   767  	logger := klog.FromContext(ctx)
   768  	// Look at the difference in seconds between now and the last time we reported any
   769  	// progress or tried to create a replica set, or resumed a paused deployment and
   770  	// compare against progressDeadlineSeconds.
   771  	from := condition.LastUpdateTime
   772  	now := nowFn()
   773  	delta := time.Duration(*deployment.Spec.ProgressDeadlineSeconds) * time.Second
   774  	timedOut := from.Add(delta).Before(now)
   775  
   776  	logger.V(4).Info("Deployment timed out from last progress check", "deployment", klog.KObj(deployment), "timeout", timedOut, "from", from, "now", now)
   777  	return timedOut
   778  }
   779  
   780  // NewRSNewReplicas calculates the number of replicas a deployment's new RS should have.
   781  // When one of the followings is true, we're rolling out the deployment; otherwise, we're scaling it.
   782  // 1) The new RS is saturated: newRS's replicas == deployment's replicas
   783  // 2) Max number of pods allowed is reached: deployment's replicas + maxSurge == all RSs' replicas
   784  func NewRSNewReplicas(deployment *apps.Deployment, allRSs []*apps.ReplicaSet, newRS *apps.ReplicaSet) (int32, error) {
   785  	switch deployment.Spec.Strategy.Type {
   786  	case apps.RollingUpdateDeploymentStrategyType:
   787  		// Check if we can scale up.
   788  		maxSurge, err := intstrutil.GetScaledValueFromIntOrPercent(deployment.Spec.Strategy.RollingUpdate.MaxSurge, int(*(deployment.Spec.Replicas)), true)
   789  		if err != nil {
   790  			return 0, err
   791  		}
   792  		// Find the total number of pods
   793  		currentPodCount := GetReplicaCountForReplicaSets(allRSs)
   794  		maxTotalPods := *(deployment.Spec.Replicas) + int32(maxSurge)
   795  		if currentPodCount >= maxTotalPods {
   796  			// Cannot scale up.
   797  			return *(newRS.Spec.Replicas), nil
   798  		}
   799  		// Scale up.
   800  		scaleUpCount := maxTotalPods - currentPodCount
   801  		// Do not exceed the number of desired replicas.
   802  		scaleUpCount = min(scaleUpCount, *(deployment.Spec.Replicas)-*(newRS.Spec.Replicas))
   803  		return *(newRS.Spec.Replicas) + scaleUpCount, nil
   804  	case apps.RecreateDeploymentStrategyType:
   805  		return *(deployment.Spec.Replicas), nil
   806  	default:
   807  		return 0, fmt.Errorf("deployment type %v isn't supported", deployment.Spec.Strategy.Type)
   808  	}
   809  }
   810  
   811  // IsSaturated checks if the new replica set is saturated by comparing its size with its deployment size.
   812  // Both the deployment and the replica set have to believe this replica set can own all of the desired
   813  // replicas in the deployment and the annotation helps in achieving that. All pods of the ReplicaSet
   814  // need to be available.
   815  func IsSaturated(deployment *apps.Deployment, rs *apps.ReplicaSet) bool {
   816  	if rs == nil {
   817  		return false
   818  	}
   819  	desiredString := rs.Annotations[DesiredReplicasAnnotation]
   820  	desired, err := strconv.Atoi(desiredString)
   821  	if err != nil {
   822  		return false
   823  	}
   824  	return *(rs.Spec.Replicas) == *(deployment.Spec.Replicas) &&
   825  		int32(desired) == *(deployment.Spec.Replicas) &&
   826  		rs.Status.AvailableReplicas == *(deployment.Spec.Replicas)
   827  }
   828  
   829  // WaitForObservedDeployment polls for deployment to be updated so that deployment.Status.ObservedGeneration >= desiredGeneration.
   830  // Returns error if polling timesout.
   831  func WaitForObservedDeployment(getDeploymentFunc func() (*apps.Deployment, error), desiredGeneration int64, interval, timeout time.Duration) error {
   832  	// TODO: This should take clientset.Interface when all code is updated to use clientset. Keeping it this way allows the function to be used by callers who have client.Interface.
   833  	return wait.PollImmediate(interval, timeout, func() (bool, error) {
   834  		deployment, err := getDeploymentFunc()
   835  		if err != nil {
   836  			return false, err
   837  		}
   838  		return deployment.Status.ObservedGeneration >= desiredGeneration, nil
   839  	})
   840  }
   841  
   842  // ResolveFenceposts resolves both maxSurge and maxUnavailable. This needs to happen in one
   843  // step. For example:
   844  //
   845  // 2 desired, max unavailable 1%, surge 0% - should scale old(-1), then new(+1), then old(-1), then new(+1)
   846  // 1 desired, max unavailable 1%, surge 0% - should scale old(-1), then new(+1)
   847  // 2 desired, max unavailable 25%, surge 1% - should scale new(+1), then old(-1), then new(+1), then old(-1)
   848  // 1 desired, max unavailable 25%, surge 1% - should scale new(+1), then old(-1)
   849  // 2 desired, max unavailable 0%, surge 1% - should scale new(+1), then old(-1), then new(+1), then old(-1)
   850  // 1 desired, max unavailable 0%, surge 1% - should scale new(+1), then old(-1)
   851  func ResolveFenceposts(maxSurge, maxUnavailable *intstrutil.IntOrString, desired int32) (int32, int32, error) {
   852  	surge, err := intstrutil.GetScaledValueFromIntOrPercent(intstrutil.ValueOrDefault(maxSurge, intstrutil.FromInt32(0)), int(desired), true)
   853  	if err != nil {
   854  		return 0, 0, err
   855  	}
   856  	unavailable, err := intstrutil.GetScaledValueFromIntOrPercent(intstrutil.ValueOrDefault(maxUnavailable, intstrutil.FromInt32(0)), int(desired), false)
   857  	if err != nil {
   858  		return 0, 0, err
   859  	}
   860  
   861  	if surge == 0 && unavailable == 0 {
   862  		// Validation should never allow the user to explicitly use zero values for both maxSurge
   863  		// maxUnavailable. Due to rounding down maxUnavailable though, it may resolve to zero.
   864  		// If both fenceposts resolve to zero, then we should set maxUnavailable to 1 on the
   865  		// theory that surge might not work due to quota.
   866  		unavailable = 1
   867  	}
   868  
   869  	return int32(surge), int32(unavailable), nil
   870  }
   871  
   872  // HasProgressDeadline checks if the Deployment d is expected to surface the reason
   873  // "ProgressDeadlineExceeded" when the Deployment progress takes longer than expected time.
   874  func HasProgressDeadline(d *apps.Deployment) bool {
   875  	return d.Spec.ProgressDeadlineSeconds != nil && *d.Spec.ProgressDeadlineSeconds != math.MaxInt32
   876  }
   877  
   878  // HasRevisionHistoryLimit checks if the Deployment d is expected to keep a specified number of
   879  // old replicaSets. These replicaSets are mainly kept with the purpose of rollback.
   880  // The RevisionHistoryLimit can start from 0 (no retained replicasSet). When set to math.MaxInt32,
   881  // the Deployment will keep all revisions.
   882  func HasRevisionHistoryLimit(d *apps.Deployment) bool {
   883  	return d.Spec.RevisionHistoryLimit != nil && *d.Spec.RevisionHistoryLimit != math.MaxInt32
   884  }
   885  
   886  // GetDeploymentsForReplicaSet returns a list of Deployments that potentially
   887  // match a ReplicaSet. Only the one specified in the ReplicaSet's ControllerRef
   888  // will actually manage it.
   889  // Returns an error only if no matching Deployments are found.
   890  func GetDeploymentsForReplicaSet(deploymentLister appslisters.DeploymentLister, rs *apps.ReplicaSet) ([]*apps.Deployment, error) {
   891  	if len(rs.Labels) == 0 {
   892  		return nil, fmt.Errorf("no deployments found for ReplicaSet %v because it has no labels", rs.Name)
   893  	}
   894  
   895  	// TODO: MODIFY THIS METHOD so that it checks for the podTemplateSpecHash label
   896  	dList, err := deploymentLister.Deployments(rs.Namespace).List(labels.Everything())
   897  	if err != nil {
   898  		return nil, err
   899  	}
   900  
   901  	var deployments []*apps.Deployment
   902  	for _, d := range dList {
   903  		selector, err := metav1.LabelSelectorAsSelector(d.Spec.Selector)
   904  		if err != nil {
   905  			// This object has an invalid selector, it does not match the replicaset
   906  			continue
   907  		}
   908  		// If a deployment with a nil or empty selector creeps in, it should match nothing, not everything.
   909  		if selector.Empty() || !selector.Matches(labels.Set(rs.Labels)) {
   910  			continue
   911  		}
   912  		deployments = append(deployments, d)
   913  	}
   914  
   915  	if len(deployments) == 0 {
   916  		return nil, fmt.Errorf("could not find deployments set for ReplicaSet %s in namespace %s with labels: %v", rs.Name, rs.Namespace, rs.Labels)
   917  	}
   918  
   919  	return deployments, nil
   920  }
   921  
   922  // ReplicaSetsByRevision sorts a list of ReplicaSet by revision, using their creation timestamp or name as a tie breaker.
   923  // By using the creation timestamp, this sorts from old to new replica sets.
   924  type ReplicaSetsByRevision []*apps.ReplicaSet
   925  
   926  func (o ReplicaSetsByRevision) Len() int      { return len(o) }
   927  func (o ReplicaSetsByRevision) Swap(i, j int) { o[i], o[j] = o[j], o[i] }
   928  func (o ReplicaSetsByRevision) Less(i, j int) bool {
   929  	revision1, err1 := Revision(o[i])
   930  	revision2, err2 := Revision(o[j])
   931  	if err1 != nil || err2 != nil || revision1 == revision2 {
   932  		return controller.ReplicaSetsByCreationTimestamp(o).Less(i, j)
   933  	}
   934  	return revision1 < revision2
   935  }
   936  

View as plain text