github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-29 14:37:00 +00:00
Code Issues Projects Releases Wiki Activity

Scale up early when deployment creates new Replica set

Browse Source
This commit is contained in:
Janet Kuo 2016-02-04 18:05:38 -08:00
parent 3616b4bfec
commit e9262df456
3 changed files with 76 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
pkg/controller/deployment/deployment_controller.go
View File

@ -586,7 +586,7 @@ func (dc *DeploymentController) getAllReplicaSets(deployment extensions.Deployme
maxOldV := maxRevision(allOldRSs)
// Get new replica set with the updated revision number
newRS, err := dc.getNewReplicaSet(deployment, maxOldV)
newRS, err := dc.getNewReplicaSet(deployment, maxOldV, oldRSs)
if err != nil {
return nil, nil, err
}
@ -648,7 +648,7 @@ func (dc *DeploymentController) getOldReplicaSets(deployment extensions.Deployme
// Returns a replica set that matches the intent of the given deployment.
// It creates a new replica set if required.
// The revision of the new replica set will be updated to maxOldRevision + 1
func (dc *DeploymentController) getNewReplicaSet(deployment extensions.Deployment, maxOldRevision int64) (*extensions.ReplicaSet, error) {
func (dc *DeploymentController) getNewReplicaSet(deployment extensions.Deployment, maxOldRevision int64, oldRSs []*extensions.ReplicaSet) (*extensions.ReplicaSet, error) {
// Calculate revision number for this new replica set
newRevision := strconv.FormatInt(maxOldRevision+1, 10)
@ -701,14 +701,22 @@ func (dc *DeploymentController) getNewReplicaSet(deployment extensions.Deploymen
}
// Set new replica set's annotation
setNewReplicaSetAnnotations(&deployment, &newRS, newRevision)
allRSs := append(oldRSs, &newRS)
newReplicasCount, err := deploymentutil.NewRSNewReplicas(&deployment, allRSs, &newRS)
if err != nil {
return nil, err
}
newRS.Spec.Replicas = newReplicasCount
createdRS, err := dc.client.Extensions().ReplicaSets(namespace).Create(&newRS)
if err != nil {
dc.rsExpectations.DeleteExpectations(dKey)
return nil, fmt.Errorf("error creating replica set: %v", err)
}
if newReplicasCount > 0 {
dc.eventRecorder.Eventf(&deployment, api.EventTypeNormal, "ScalingReplicaSet", "Scaled %s replica set %s to %d", "up", createdRS.Name, newReplicasCount)
}
err = dc.updateDeploymentRevision(deployment, newRevision)
return createdRS, err
return createdRS, dc.updateDeploymentRevision(deployment, newRevision)
}
// setNewReplicaSetAnnotations sets new replica set's annotations appropriately by updating its revision and
@ -752,9 +760,12 @@ func (dc *DeploymentController) updateDeploymentRevision(deployment extensions.D
if deployment.Annotations == nil {
deployment.Annotations = make(map[string]string)
}
deployment.Annotations[deploymentutil.RevisionAnnotation] = revision
_, err := dc.updateDeployment(&deployment)
return err
if deployment.Annotations[deploymentutil.RevisionAnnotation] != revision {
deployment.Annotations[deploymentutil.RevisionAnnotation] = revision
_, err := dc.updateDeployment(&deployment)
return err
}
return nil
}
func (dc *DeploymentController) reconcileNewReplicaSet(allRSs []*extensions.ReplicaSet, newRS *extensions.ReplicaSet, deployment extensions.Deployment) (bool, error) {
@ -764,29 +775,15 @@ func (dc *DeploymentController) reconcileNewReplicaSet(allRSs []*extensions.Repl
}
if newRS.Spec.Replicas > deployment.Spec.Replicas {
// Scale down.
_, err := dc.scaleReplicaSetAndRecordEvent(newRS, deployment.Spec.Replicas, deployment)
return true, err
scaled, _, err := dc.scaleReplicaSetAndRecordEvent(newRS, deployment.Spec.Replicas, deployment)
return scaled, err
}
// Check if we can scale up.
maxSurge, err := intstrutil.GetValueFromIntOrPercent(&deployment.Spec.Strategy.RollingUpdate.MaxSurge, deployment.Spec.Replicas)
newReplicasCount, err := deploymentutil.NewRSNewReplicas(&deployment, allRSs, newRS)
if err != nil {
return false, err
}
// Find the total number of pods
currentPodCount := deploymentutil.GetReplicaCountForReplicaSets(allRSs)
maxTotalPods := deployment.Spec.Replicas + maxSurge
if currentPodCount >= maxTotalPods {
// Cannot scale up.
return false, nil
}
// Scale up.
scaleUpCount := maxTotalPods - currentPodCount
// Do not exceed the number of desired replicas.
scaleUpCount = integer.IntMin(scaleUpCount, deployment.Spec.Replicas-newRS.Spec.Replicas)
newReplicasCount := newRS.Spec.Replicas + scaleUpCount
_, err = dc.scaleReplicaSetAndRecordEvent(newRS, newReplicasCount, deployment)
return true, err
scaled, _, err := dc.scaleReplicaSetAndRecordEvent(newRS, newReplicasCount, deployment)
return scaled, err
}
// Set expectationsCheck to false to bypass expectations check when testing
@ -903,7 +900,7 @@ func (dc *DeploymentController) cleanupUnhealthyReplicas(oldRSs []*extensions.Re
scaledDownCount := integer.IntMin(maxCleanupCount-totalScaledDown, targetRS.Spec.Replicas-readyPodCount)
newReplicasCount := targetRS.Spec.Replicas - scaledDownCount
_, err = dc.scaleReplicaSetAndRecordEvent(targetRS, newReplicasCount, deployment)
_, _, err = dc.scaleReplicaSetAndRecordEvent(targetRS, newReplicasCount, deployment)
if err != nil {
return totalScaledDown, err
}
@ -949,7 +946,7 @@ func (dc *DeploymentController) scaleDownOldReplicaSetsForRollingUpdate(allRSs [
// Scale down.
scaleDownCount := integer.IntMin(targetRS.Spec.Replicas, totalScaleDownCount-totalScaledDown)
newReplicasCount := targetRS.Spec.Replicas - scaleDownCount
_, err = dc.scaleReplicaSetAndRecordEvent(targetRS, newReplicasCount, deployment)
_, _, err = dc.scaleReplicaSetAndRecordEvent(targetRS, newReplicasCount, deployment)
if err != nil {
return totalScaledDown, err
}
@ -968,23 +965,21 @@ func (dc *DeploymentController) scaleDownOldReplicaSetsForRecreate(oldRSs []*ext
if rs.Spec.Replicas == 0 {
continue
}
_, err := dc.scaleReplicaSetAndRecordEvent(rs, 0, deployment)
scaledRS, _, err := dc.scaleReplicaSetAndRecordEvent(rs, 0, deployment)
if err != nil {
return false, err
}
scaled = true
if scaledRS {
scaled = true
}
}
return scaled, nil
}
// scaleUpNewReplicaSetForRecreate scales up new replica set when deployment strategy is "Recreate"
func (dc *DeploymentController) scaleUpNewReplicaSetForRecreate(newRS *extensions.ReplicaSet, deployment extensions.Deployment) (bool, error) {
if newRS.Spec.Replicas == deployment.Spec.Replicas {
// Scaling not required.
return false, nil
}
_, err := dc.scaleReplicaSetAndRecordEvent(newRS, deployment.Spec.Replicas, deployment)
return true, err
scaled, _, err := dc.scaleReplicaSetAndRecordEvent(newRS, deployment.Spec.Replicas, deployment)
return scaled, err
}
func (dc *DeploymentController) cleanupOldReplicaSets(oldRSs []*extensions.ReplicaSet, deployment extensions.Deployment) error {
@ -1042,7 +1037,11 @@ func (dc *DeploymentController) calculateStatus(allRSs []*extensions.ReplicaSet,
return
}
func (dc *DeploymentController) scaleReplicaSetAndRecordEvent(rs *extensions.ReplicaSet, newScale int, deployment extensions.Deployment) (*extensions.ReplicaSet, error) {
func (dc *DeploymentController) scaleReplicaSetAndRecordEvent(rs *extensions.ReplicaSet, newScale int, deployment extensions.Deployment) (bool, *extensions.ReplicaSet, error) {
// No need to scale
if rs.Spec.Replicas == newScale {
return false, rs, nil
}
scalingOperation := "down"
if rs.Spec.Replicas < newScale {
scalingOperation = "up"
@ -1051,7 +1050,7 @@ func (dc *DeploymentController) scaleReplicaSetAndRecordEvent(rs *extensions.Rep
if err == nil {
dc.eventRecorder.Eventf(&deployment, api.EventTypeNormal, "ScalingReplicaSet", "Scaled %s replica set %s to %d", scalingOperation, rs.Name, newScale)
}
return newRS, err
return true, newRS, err
}
func (dc *DeploymentController) scaleReplicaSet(rs *extensions.ReplicaSet, newScale int) (*extensions.ReplicaSet, error) {

37
pkg/util/deployment/deployment.go
View File

@ -26,6 +26,8 @@ import (
"k8s.io/kubernetes/pkg/apis/extensions"
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
"k8s.io/kubernetes/pkg/labels"
"k8s.io/kubernetes/pkg/util/integer"
intstrutil "k8s.io/kubernetes/pkg/util/intstr"
labelsutil "k8s.io/kubernetes/pkg/util/labels"
podutil "k8s.io/kubernetes/pkg/util/pod"
)
@ -233,3 +235,38 @@ func Revision(rs *extensions.ReplicaSet) (int64, error) {
}
return strconv.ParseInt(v, 10, 64)
}
func IsRollingUpdate(deployment *extensions.Deployment) bool {
return deployment.Spec.Strategy.Type == extensions.RollingUpdateDeploymentStrategyType
}
// NewRSNewReplicas calculates the number of replicas a deployment's new RS should have.
// When one of the followings is true, we're rolling out the deployment; otherwise, we're scaling it.
// 1) The new RS is saturated: newRS's replicas == deployment's replicas
// 2) Max number of pods allowed is reached: deployment's replicas + maxSurge == all RSs' replicas
func NewRSNewReplicas(deployment *extensions.Deployment, allRSs []*extensions.ReplicaSet, newRS *extensions.ReplicaSet) (int, error) {
switch deployment.Spec.Strategy.Type {
case extensions.RollingUpdateDeploymentStrategyType:
// Check if we can scale up.
maxSurge, err := intstrutil.GetValueFromIntOrPercent(&deployment.Spec.Strategy.RollingUpdate.MaxSurge, deployment.Spec.Replicas)
if err != nil {
return 0, err
}
// Find the total number of pods
currentPodCount := GetReplicaCountForReplicaSets(allRSs)
maxTotalPods := deployment.Spec.Replicas + maxSurge
if currentPodCount >= maxTotalPods {
// Cannot scale up.
return newRS.Spec.Replicas, nil
}
// Scale up.
scaleUpCount := maxTotalPods - currentPodCount
// Do not exceed the number of desired replicas.
scaleUpCount = integer.IntMin(scaleUpCount, deployment.Spec.Replicas-newRS.Spec.Replicas)
return newRS.Spec.Replicas + scaleUpCount, nil
case extensions.RecreateDeploymentStrategyType:
return deployment.Spec.Replicas, nil
default:
return 0, fmt.Errorf("deployment type %v isn't supported", deployment.Spec.Strategy.Type)
}
}

4
test/e2e/deployment.go
View File

@ -386,8 +386,8 @@ func testRecreateDeployment(f *Framework) {
newRS, err := deploymentutil.GetNewReplicaSet(*deployment, c)
Expect(err).NotTo(HaveOccurred())
Expect(newRS).NotTo(Equal(nil))
Expect(events.Items[0].Message).Should(Equal(fmt.Sprintf("Scaled down replica set %s to 0", rsName)))
Expect(events.Items[1].Message).Should(Equal(fmt.Sprintf("Scaled up replica set %s to 3", newRS.Name)))
Expect(events.Items[0].Message).Should(Equal(fmt.Sprintf("Scaled up replica set %s to 3", newRS.Name)))
Expect(events.Items[1].Message).Should(Equal(fmt.Sprintf("Scaled down replica set %s to 0", rsName)))
// Check if it's updated to revision 1 correctly
checkDeploymentRevision(c, ns, deploymentName, "1", "redis", "redis")