github/kubernetes
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Rolling updater enhancements

Browse Source 
Add an UpdateAcceptor interface to the rolling updater which supports
injecting code to validate the first replica during scale-up. If the
replica is not accepted, the deployment fails. This facilitates canary
checking so that many broken replicas aren't rolled out during an update.

Make the rolling update scale amount configurable as a percent of the replica
count; a negative value changes the scale direction to down/up to support
in-place deployments.
This commit is contained in:
Dan Mace
2015-06-18 16:29:28 -04:00
parent 7645513d2a
commit d044208b38
3 changed files with 546 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

194
pkg/kubectl/rolling_updater.go
View File

@@ -20,6 +20,7 @@ import (
goerrors "errors"
"fmt"
"io"
"math"
"strconv"
"strings"
"time"
@@ -39,8 +40,14 @@ type RollingUpdater struct {
c RollingUpdaterClient
// Namespace for resources
ns string
// scaleAndWait scales a controller and returns its updated state.
scaleAndWait scaleAndWait
}
// scaleAndWait scales rc and returns its updated state. This typedef is to
// abstract away the use of a Scaler to ease testing.
type scaleAndWait func(rc *api.ReplicationController, retry *RetryParams, wait *RetryParams) (*api.ReplicationController, error)
// RollingUpdaterConfig is the configuration for a rolling deployment process.
type RollingUpdaterConfig struct {
// Out is a writer for progress output.
@@ -60,6 +67,16 @@ type RollingUpdaterConfig struct {
// CleanupPolicy defines the cleanup action to take after the deployment is
// complete.
CleanupPolicy RollingUpdaterCleanupPolicy
// UpdateAcceptor is given a chance to accept the new controller after each
// scale-up operation. If the controller is accepted, updates continue; if
// the controller is rejected, the update will fail immediately.
UpdateAcceptor UpdateAcceptor
// UpdatePercent is optional; if specified, the amount of replicas scaled up
// and down each interval will be computed as a percentage of the desired
// replicas for the new RC. If UpdatePercent is nil, one replica will be
// scaled up and down each interval. If UpdatePercent is negative, the order
// of scaling will be down/up instead of up/down.
UpdatePercent *int
}
// RollingUpdaterCleanupPolicy is a cleanup action to take after the
@@ -76,6 +93,26 @@ const (
RenameRollingUpdateCleanupPolicy RollingUpdaterCleanupPolicy = "Rename"
)
// UpdateAcceptor is given a chance to accept or reject the new controller
// during a deployment each time the controller is scaled up.
//
// After the successful scale-up of the controller, the controller is given to
// the UpdateAcceptor. If the UpdateAcceptor rejects the controller, the
// deployment is stopped with an error.
type UpdateAcceptor interface {
// Accept returns nil if the controller is okay, otherwise returns an error.
Accept(*api.ReplicationController) error
}
// AlwaysAccept is an UpdateAcceptor which always accepts the controller.
type AlwaysAccept struct{}
// Accept implements UpdateAcceptor.
func (a *AlwaysAccept) Accept(*api.ReplicationController) error { return nil }
// DefaultUpdaterAcceptor always accepts controllers.
var DefaultUpdateAcceptor UpdateAcceptor = &AlwaysAccept{}
func LoadExistingNextReplicationController(c *client.Client, namespace, newName string) (*api.ReplicationController, error) {
if len(newName) == 0 {
return nil, nil
@@ -121,10 +158,12 @@ func CreateNewControllerFromCurrentController(c *client.Client, namespace, oldNa
}
// NewRollingUpdater creates a RollingUpdater from a client
func NewRollingUpdater(namespace string, c RollingUpdaterClient) *RollingUpdater {
func NewRollingUpdater(namespace string, client RollingUpdaterClient) *RollingUpdater {
return &RollingUpdater{
c,
namespace,
c: client,
ns: namespace,
// Use a real scaleAndWait implementation.
scaleAndWait: scalerScaleAndWait(client, namespace),
}
}
@@ -175,6 +214,21 @@ func UpdateExistingReplicationController(c client.Interface, oldRc *api.Replicat
}
}
// scalerScaleAndWait returns a scaleAndWait function which scales a
// controller using a Scaler and a real client.
func scalerScaleAndWait(client RollingUpdaterClient, namespace string) scaleAndWait {
scaler, err := ScalerFor("ReplicationController", client)
return func(rc *api.ReplicationController, retry *RetryParams, wait *RetryParams) (*api.ReplicationController, error) {
if err != nil {
return nil, fmt.Errorf("Couldn't make scaler: %s", err)
}
if err := scaler.Scale(rc.Namespace, rc.Name, uint(rc.Spec.Replicas), &ScalePrecondition{-1, ""}, retry, wait); err != nil {
return nil, err
}
return client.GetReplicationController(namespace, rc.ObjectMeta.Name)
}
}
const MaxRetries = 3
func AddDeploymentKeyToReplicationController(oldRc *api.ReplicationController, client client.Interface, deploymentKey, deploymentValue, namespace string, out io.Writer) (*api.ReplicationController, error) {
@@ -297,10 +351,14 @@ func FindSourceController(r RollingUpdaterClient, namespace, name string) (*api.
}
// Update all pods for a ReplicationController (oldRc) by creating a new
// controller (newRc) with 0 replicas, and synchronously scaling oldRc,newRc
// by 1 until oldRc has 0 replicas and newRc has the original # of desired
// controller (newRc) with 0 replicas, and synchronously scaling oldRc and
// newRc until oldRc has 0 replicas and newRc has the original # of desired
// replicas. Cleanup occurs based on a RollingUpdaterCleanupPolicy.
//
// The scaling amount each interval is either 1 or based on a percent of the
// desired replicas. If a percentage is used and the percentage is negative,
// the scaling order is inverted to down/up instead of the default up/down.
//
// If an update from newRc to oldRc is already in progress, we attempt to
// drive it to completion. If an error occurs at any step of the update, the
// error will be returned.
@@ -353,51 +411,50 @@ func (r *RollingUpdater) Update(config *RollingUpdaterConfig) error {
}
}
// +1, -1 on oldRc, newRc until newRc has desired number of replicas or oldRc has 0 replicas
for newRc.Spec.Replicas < desired && oldRc.Spec.Replicas != 0 {
newRc.Spec.Replicas += 1
oldRc.Spec.Replicas -= 1
fmt.Printf("At beginning of loop: %s replicas: %d, %s replicas: %d\n",
oldName, oldRc.Spec.Replicas,
newName, newRc.Spec.Replicas)
fmt.Fprintf(out, "Updating %s replicas: %d, %s replicas: %d\n",
oldName, oldRc.Spec.Replicas,
newName, newRc.Spec.Replicas)
// Compute the scale amount based on a percentage of the new desired count.
// A negative percentage indicates our scale direction should be down-first.
scaleAmount := 1
skipFirstUp := false
if config.UpdatePercent != nil {
scaleAmount = int(math.Ceil(float64(desired) * (math.Abs(float64(*config.UpdatePercent)) / 100)))
if *config.UpdatePercent < 0 {
skipFirstUp = true
}
}
// Helpful output about what we're about to do.
direction := "up"
if skipFirstUp {
direction = "down"
}
fmt.Fprintf(out, "Scaling up %s from %d to %d, scaling down %s from %d to 0 (scale %s first by %d each interval)\n",
newRc.Name, newRc.Spec.Replicas, desired, oldRc.Name, oldRc.Spec.Replicas, direction, scaleAmount)
newRc, err = r.scaleAndWait(newRc, retry, waitForReplicas)
// Scale newRc and oldRc until newRc has the desired number of replicas and
// oldRc has 0 replicas.
for newRc.Spec.Replicas != desired || oldRc.Spec.Replicas != 0 {
// Choose up/down vs. down/up scaling direction.
if !skipFirstUp {
scaledRc, err := r.scaleUp(newRc, oldRc, desired, scaleAmount, retry, waitForReplicas, out, config)
if err != nil {
return err
}
newRc = scaledRc
time.Sleep(updatePeriod)
skipFirstUp = true
}
scaledRc, err := r.scaleDown(newRc, oldRc, desired, scaleAmount, retry, waitForReplicas, out, config)
if err != nil {
return err
}
rc = scaledRc
time.Sleep(updatePeriod)
oldRc, err = r.scaleAndWait(oldRc, retry, waitForReplicas)
if err != nil {
return err
}
fmt.Printf("At end of loop: %s replicas: %d, %s replicas: %d\n",
oldName, oldRc.Spec.Replicas,
newName, newRc.Spec.Replicas)
}
// delete remaining replicas on oldRc
if oldRc.Spec.Replicas != 0 {
fmt.Fprintf(out, "Stopping %s replicas: %d -> %d\n",
oldName, oldRc.Spec.Replicas, 0)
oldRc.Spec.Replicas = 0
oldRc, err = r.scaleAndWait(oldRc, retry, waitForReplicas)
// oldRc, err = r.scaleAndWait(oldRc, interval, timeout)
if err != nil {
return err
}
}
// add remaining replicas on newRc
if newRc.Spec.Replicas != desired {
fmt.Fprintf(out, "Scaling %s replicas: %d -> %d\n",
newName, newRc.Spec.Replicas, desired)
newRc.Spec.Replicas = desired
newRc, err = r.scaleAndWait(newRc, retry, waitForReplicas)
scaledRc, err = r.scaleUp(newRc, oldRc, desired, scaleAmount, retry, waitForReplicas, out, config)
if err != nil {
return err
}
newRc = scaledRc
}
// Clean up annotations
if newRc, err = r.c.GetReplicationController(r.ns, newName); err != nil {
return err
@@ -429,6 +486,50 @@ func (r *RollingUpdater) Update(config *RollingUpdaterConfig) error {
}
}
// scaleUp scales up newRc to desired by scaleAmount. It accounts for
// fencepost conditions. If newRc is already scaled to desired, scaleUp does
// nothing. If the oldRc is already scaled to 0, newRc is scaled to desired
// immediately regardless of scale count.
func (r *RollingUpdater) scaleUp(newRc, oldRc *api.ReplicationController, desired, scaleAmount int, retry, wait *RetryParams, out io.Writer, config *RollingUpdaterConfig) (*api.ReplicationController, error) {
if newRc.Spec.Replicas == desired {
return newRc, nil
}
newRc.Spec.Replicas += scaleAmount
if newRc.Spec.Replicas > desired || oldRc.Spec.Replicas == 0 {
newRc.Spec.Replicas = desired
}
fmt.Fprintf(out, "Scaling %s up to %d\n", newRc.Name, newRc.Spec.Replicas)
scaledRc, err := r.scaleAndWait(newRc, retry, wait)
if err != nil {
return nil, err
}
err = config.UpdateAcceptor.Accept(scaledRc)
if err != nil {
return nil, fmt.Errorf("update rejected for %s: %v", scaledRc.Name, err)
}
return scaledRc, nil
}
// scaleDown scales down oldRc to 0 by scaleAmount. It accounts for fencepost
// conditions. If oldRc is already scaled to 0, scaleDown does nothing. If
// newRc is already scaled to desired, oldRc is scaled to 0 immediately
// regardless of scaleAmount.
func (r *RollingUpdater) scaleDown(newRc, oldRc *api.ReplicationController, desired, scaleAmount int, retry, wait *RetryParams, out io.Writer, config *RollingUpdaterConfig) (*api.ReplicationController, error) {
if oldRc.Spec.Replicas == 0 {
return oldRc, nil
}
oldRc.Spec.Replicas -= scaleAmount
if oldRc.Spec.Replicas < 0 || newRc.Spec.Replicas == desired {
oldRc.Spec.Replicas = 0
}
fmt.Fprintf(out, "Scaling %s down to %d\n", oldRc.Name, oldRc.Spec.Replicas)
scaledRc, err := r.scaleAndWait(oldRc, retry, wait)
if err != nil {
return nil, err
}
return scaledRc, nil
}
func (r *RollingUpdater) getExistingNewRc(sourceId, name string) (rc *api.ReplicationController, existing bool, err error) {
if rc, err = r.c.GetReplicationController(r.ns, name); err == nil {
existing = true
@@ -444,17 +545,6 @@ func (r *RollingUpdater) getExistingNewRc(sourceId, name string) (rc *api.Replic
return
}
func (r *RollingUpdater) scaleAndWait(rc *api.ReplicationController, retry *RetryParams, wait *RetryParams) (*api.ReplicationController, error) {
scaler, err := ScalerFor("ReplicationController", r.c)
if err != nil {
return nil, err
}
if err := scaler.Scale(rc.Namespace, rc.Name, uint(rc.Spec.Replicas), &ScalePrecondition{-1, ""}, retry, wait); err != nil {
return nil, err
}
return r.c.GetReplicationController(r.ns, rc.ObjectMeta.Name)
}
func (r *RollingUpdater) updateAndWait(rc *api.ReplicationController, interval, timeout time.Duration) (*api.ReplicationController, error) {
rc, err := r.c.UpdateReplicationController(r.ns, rc)
if err != nil {