mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-07-24 04:06:03 +00:00
kubeadm: delete prepull ds
Signed-off-by: Xianglin Gao <xianglin.gxl@alibaba-inc.com>
This commit is contained in:
Xianglin Gao
parent
d8a513ef99
commit
6c6a702a99
@ -18,7 +18,6 @@ package upgrade
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/pkg/errors"
|
||||
"github.com/spf13/cobra"
|
||||
@ -27,7 +26,6 @@ import (
|
||||
"k8s.io/klog"
|
||||
kubeadmapi "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm"
|
||||
"k8s.io/kubernetes/cmd/kubeadm/app/cmd/options"
|
||||
"k8s.io/kubernetes/cmd/kubeadm/app/constants"
|
||||
"k8s.io/kubernetes/cmd/kubeadm/app/features"
|
||||
"k8s.io/kubernetes/cmd/kubeadm/app/phases/upgrade"
|
||||
kubeadmutil "k8s.io/kubernetes/cmd/kubeadm/app/util"
|
||||
@ -35,10 +33,6 @@ import (
|
||||
configutil "k8s.io/kubernetes/cmd/kubeadm/app/util/config"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultImagePullTimeout = 15 * time.Minute
|
||||
)
|
||||
|
||||
// applyFlags holds the information about the flags that can be passed to apply
|
||||
type applyFlags struct {
|
||||
*applyPlanFlags
|
||||
@ -48,7 +42,6 @@ type applyFlags struct {
|
||||
dryRun bool
|
||||
etcdUpgrade bool
|
||||
renewCerts bool
|
||||
imagePullTimeout time.Duration
|
||||
kustomizeDir string
|
||||
}
|
||||
|
||||
@ -60,10 +53,9 @@ func (f *applyFlags) sessionIsInteractive() bool {
|
||||
// NewCmdApply returns the cobra command for `kubeadm upgrade apply`
|
||||
func NewCmdApply(apf *applyPlanFlags) *cobra.Command {
|
||||
flags := &applyFlags{
|
||||
applyPlanFlags: apf,
|
||||
imagePullTimeout: defaultImagePullTimeout,
|
||||
etcdUpgrade: true,
|
||||
renewCerts: true,
|
||||
applyPlanFlags: apf,
|
||||
etcdUpgrade: true,
|
||||
renewCerts: true,
|
||||
}
|
||||
|
||||
cmd := &cobra.Command{
|
||||
@ -88,7 +80,6 @@ func NewCmdApply(apf *applyPlanFlags) *cobra.Command {
|
||||
cmd.Flags().BoolVar(&flags.dryRun, options.DryRun, flags.dryRun, "Do not change any state, just output what actions would be performed.")
|
||||
cmd.Flags().BoolVar(&flags.etcdUpgrade, "etcd-upgrade", flags.etcdUpgrade, "Perform the upgrade of etcd.")
|
||||
cmd.Flags().BoolVar(&flags.renewCerts, options.CertificateRenewal, flags.renewCerts, "Perform the renewal of certificates used by component changed during upgrades.")
|
||||
cmd.Flags().DurationVar(&flags.imagePullTimeout, "image-pull-timeout", flags.imagePullTimeout, "The maximum amount of time to wait for the control plane pods to be downloaded.")
|
||||
options.AddKustomizePodsFlag(cmd.Flags(), &flags.kustomizeDir)
|
||||
|
||||
return cmd
|
||||
@ -145,22 +136,7 @@ func runApply(flags *applyFlags, userVersion string) error {
|
||||
}
|
||||
}
|
||||
|
||||
// Set the timeout as flags.imagePullTimeout to ensure that Prepuller truly respects 'image-pull-timeout' flag
|
||||
waiter := getWaiter(flags.dryRun, client, flags.imagePullTimeout)
|
||||
|
||||
// Use a prepuller implementation based on creating DaemonSets
|
||||
// and block until all DaemonSets are ready; then we know for sure that all control plane images are cached locally
|
||||
klog.V(1).Infoln("[upgrade/apply] creating prepuller")
|
||||
prepuller := upgrade.NewDaemonSetPrepuller(client, waiter, &cfg.ClusterConfiguration)
|
||||
componentsToPrepull := constants.ControlPlaneComponents
|
||||
if cfg.Etcd.External == nil && flags.etcdUpgrade {
|
||||
componentsToPrepull = append(componentsToPrepull, constants.Etcd)
|
||||
}
|
||||
if err := upgrade.PrepullImagesInParallel(prepuller, flags.imagePullTimeout, componentsToPrepull); err != nil {
|
||||
return errors.Wrap(err, "[upgrade/prepull] Failed prepulled the images for the control plane components error")
|
||||
}
|
||||
|
||||
waiter = getWaiter(flags.dryRun, client, upgrade.UpgradeManifestTimeout)
|
||||
waiter := getWaiter(flags.dryRun, client, upgrade.UpgradeManifestTimeout)
|
||||
|
||||
// Now; perform the upgrade procedure
|
||||
klog.V(1).Infoln("[upgrade/apply] performing upgrade")
|
||||
|
||||
@ -187,8 +187,6 @@ const (
|
||||
TLSBootstrapRetryInterval = 5 * time.Second
|
||||
// PullImageRetry specifies how many times ContainerRuntime retries when pulling image failed
|
||||
PullImageRetry = 5
|
||||
// PrepullImagesInParallelTimeout specifies how long kubeadm should wait for prepulling images in parallel before timing out
|
||||
PrepullImagesInParallelTimeout = 10 * time.Second
|
||||
|
||||
// DefaultControlPlaneTimeout specifies the default control plane (actually API Server) timeout for use by kubeadm
|
||||
DefaultControlPlaneTimeout = 4 * time.Minute
|
||||
|
||||
@ -8,7 +8,6 @@ go_library(
|
||||
"policy.go",
|
||||
"postupgrade.go",
|
||||
"preflight.go",
|
||||
"prepull.go",
|
||||
"staticpods.go",
|
||||
"versiongetter.go",
|
||||
],
|
||||
@ -76,7 +75,6 @@ go_test(
|
||||
"compute_test.go",
|
||||
"policy_test.go",
|
||||
"postupgrade_test.go",
|
||||
"prepull_test.go",
|
||||
"staticpods_test.go",
|
||||
],
|
||||
embed = [":go_default_library"],
|
||||
|
||||
@ -1,213 +0,0 @@
|
||||
/*
|
||||
Copyright 2017 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 upgrade
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/pkg/errors"
|
||||
apps "k8s.io/api/apps/v1"
|
||||
v1 "k8s.io/api/core/v1"
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
clientset "k8s.io/client-go/kubernetes"
|
||||
kubeadmapi "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm"
|
||||
"k8s.io/kubernetes/cmd/kubeadm/app/constants"
|
||||
"k8s.io/kubernetes/cmd/kubeadm/app/images"
|
||||
"k8s.io/kubernetes/cmd/kubeadm/app/util/apiclient"
|
||||
utilpointer "k8s.io/utils/pointer"
|
||||
)
|
||||
|
||||
const (
|
||||
prepullPrefix = "upgrade-prepull-"
|
||||
)
|
||||
|
||||
// Prepuller defines an interface for performing a prepull operation in a create-wait-delete fashion in parallel
|
||||
type Prepuller interface {
|
||||
CreateFunc(string) error
|
||||
WaitFunc(string)
|
||||
DeleteFunc(string) error
|
||||
}
|
||||
|
||||
// DaemonSetPrepuller makes sure the control-plane images are available on all control-planes
|
||||
type DaemonSetPrepuller struct {
|
||||
client clientset.Interface
|
||||
cfg *kubeadmapi.ClusterConfiguration
|
||||
waiter apiclient.Waiter
|
||||
}
|
||||
|
||||
// NewDaemonSetPrepuller creates a new instance of the DaemonSetPrepuller struct
|
||||
func NewDaemonSetPrepuller(client clientset.Interface, waiter apiclient.Waiter, cfg *kubeadmapi.ClusterConfiguration) *DaemonSetPrepuller {
|
||||
return &DaemonSetPrepuller{
|
||||
client: client,
|
||||
cfg: cfg,
|
||||
waiter: waiter,
|
||||
}
|
||||
}
|
||||
|
||||
// CreateFunc creates a DaemonSet for making the image available on every relevant node
|
||||
func (d *DaemonSetPrepuller) CreateFunc(component string) error {
|
||||
var image string
|
||||
if component == constants.Etcd {
|
||||
image = images.GetEtcdImage(d.cfg)
|
||||
} else {
|
||||
image = images.GetKubernetesImage(component, d.cfg)
|
||||
}
|
||||
pauseImage := images.GetPauseImage(d.cfg)
|
||||
ds := buildPrePullDaemonSet(component, image, pauseImage)
|
||||
|
||||
// Create the DaemonSet in the API Server
|
||||
if err := apiclient.CreateOrUpdateDaemonSet(d.client, ds); err != nil {
|
||||
return errors.Wrapf(err, "unable to create a DaemonSet for prepulling the component %q", component)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// WaitFunc waits for all Pods in the specified DaemonSet to be in the Running state
|
||||
func (d *DaemonSetPrepuller) WaitFunc(component string) {
|
||||
fmt.Printf("[upgrade/prepull] Prepulling image for component %s.\n", component)
|
||||
d.waiter.WaitForPodsWithLabel("k8s-app=upgrade-prepull-" + component)
|
||||
}
|
||||
|
||||
// DeleteFunc deletes the DaemonSet used for making the image available on every relevant node
|
||||
func (d *DaemonSetPrepuller) DeleteFunc(component string) error {
|
||||
dsName := addPrepullPrefix(component)
|
||||
// TODO: The IsNotFound() check is required in cases where the DaemonSet is missing.
|
||||
// Investigate why this happens: https://github.com/kubernetes/kubeadm/issues/1700
|
||||
if err := apiclient.DeleteDaemonSetForeground(d.client, metav1.NamespaceSystem, dsName); err != nil && !apierrors.IsNotFound(err) {
|
||||
return errors.Wrapf(err, "unable to cleanup the DaemonSet used for prepulling %s", component)
|
||||
}
|
||||
fmt.Printf("[upgrade/prepull] Prepulled image for component %s.\n", component)
|
||||
return nil
|
||||
}
|
||||
|
||||
// PrepullImagesInParallel creates DaemonSets synchronously but waits in parallel for the images to pull
|
||||
func PrepullImagesInParallel(kubePrepuller Prepuller, timeout time.Duration, componentsToPrepull []string) error {
|
||||
fmt.Printf("[upgrade/prepull] Will prepull images for components %v\n", componentsToPrepull)
|
||||
|
||||
timeoutChan := time.After(timeout)
|
||||
|
||||
// Synchronously create the DaemonSets
|
||||
for _, component := range componentsToPrepull {
|
||||
if err := kubePrepuller.CreateFunc(component); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// Create a channel for streaming data from goroutines that run in parallel to a blocking for loop that cleans up
|
||||
prePulledChan := make(chan string, len(componentsToPrepull))
|
||||
for _, component := range componentsToPrepull {
|
||||
go func(c string) {
|
||||
// Wait as long as needed. This WaitFunc call should be blocking until completion
|
||||
kubePrepuller.WaitFunc(c)
|
||||
// When the task is done, go ahead and cleanup by sending the name to the channel
|
||||
prePulledChan <- c
|
||||
}(component)
|
||||
}
|
||||
|
||||
// This call blocks until all expected messages are received from the channel or errors out if timeoutChan fires.
|
||||
// For every successful wait, kubePrepuller.DeleteFunc is executed
|
||||
if err := waitForItemsFromChan(timeoutChan, prePulledChan, len(componentsToPrepull), kubePrepuller.DeleteFunc); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
fmt.Println("[upgrade/prepull] Successfully prepulled the images for all the control plane components")
|
||||
return nil
|
||||
}
|
||||
|
||||
// waitForItemsFromChan waits for n elements from stringChan with a timeout. For every item received from stringChan, cleanupFunc is executed
|
||||
func waitForItemsFromChan(timeoutChan <-chan time.Time, stringChan chan string, n int, cleanupFunc func(string) error) error {
|
||||
i := 0
|
||||
for {
|
||||
select {
|
||||
case <-timeoutChan:
|
||||
return errors.New("the prepull operation timed out")
|
||||
case result := <-stringChan:
|
||||
i++
|
||||
// If the cleanup function errors; error here as well
|
||||
if err := cleanupFunc(result); err != nil {
|
||||
return err
|
||||
}
|
||||
if i == n {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// addPrepullPrefix adds the prepull prefix for this functionality; can be used in names, labels, etc.
|
||||
func addPrepullPrefix(component string) string {
|
||||
return fmt.Sprintf("%s%s", prepullPrefix, component)
|
||||
}
|
||||
|
||||
// buildPrePullDaemonSet builds the DaemonSet that ensures the control plane image is available
|
||||
func buildPrePullDaemonSet(component, image, pauseImage string) *apps.DaemonSet {
|
||||
return &apps.DaemonSet{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: addPrepullPrefix(component),
|
||||
Namespace: metav1.NamespaceSystem,
|
||||
},
|
||||
Spec: apps.DaemonSetSpec{
|
||||
Selector: &metav1.LabelSelector{
|
||||
MatchLabels: map[string]string{
|
||||
"k8s-app": addPrepullPrefix(component),
|
||||
},
|
||||
},
|
||||
Template: v1.PodTemplateSpec{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Labels: map[string]string{
|
||||
"k8s-app": addPrepullPrefix(component),
|
||||
},
|
||||
},
|
||||
Spec: v1.PodSpec{
|
||||
// Use an init container to prepull the target component image.
|
||||
// Once the prepull completes, the "component --version" command is executed
|
||||
// to get an exit code of 0.
|
||||
// After the init container completes a regular container with "pause"
|
||||
// will start to get this Pod in Running state with a blocking container process.
|
||||
// Note that DaemonSet Pods can only use RestartPolicy of Always, so there has
|
||||
// to be a blocking process to achieve the Running state.
|
||||
InitContainers: []v1.Container{
|
||||
{
|
||||
Name: component,
|
||||
Image: image,
|
||||
Command: []string{component, "--version"},
|
||||
},
|
||||
},
|
||||
Containers: []v1.Container{
|
||||
{
|
||||
Name: "pause",
|
||||
Image: pauseImage,
|
||||
Command: []string{"/pause"},
|
||||
},
|
||||
},
|
||||
NodeSelector: map[string]string{
|
||||
constants.LabelNodeRoleMaster: "",
|
||||
},
|
||||
Tolerations: []v1.Toleration{constants.ControlPlaneToleration},
|
||||
TerminationGracePeriodSeconds: utilpointer.Int64Ptr(0),
|
||||
// Explicitly add a PodSecurityContext to allow these Pods to run as non-root.
|
||||
// This prevents restrictive PSPs from blocking the Pod creation.
|
||||
SecurityContext: &v1.PodSecurityContext{
|
||||
RunAsUser: utilpointer.Int64Ptr(999),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
@ -1,154 +0,0 @@
|
||||
/*
|
||||
Copyright 2017 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 upgrade
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/pkg/errors"
|
||||
|
||||
"k8s.io/kubernetes/cmd/kubeadm/app/constants"
|
||||
//"k8s.io/apimachinery/pkg/util/version"
|
||||
)
|
||||
|
||||
// failedCreatePrepuller is a fake prepuller that errors for kube-controller-manager in the CreateFunc call
|
||||
type failedCreatePrepuller struct{}
|
||||
|
||||
func NewFailedCreatePrepuller() Prepuller {
|
||||
return &failedCreatePrepuller{}
|
||||
}
|
||||
|
||||
func (p *failedCreatePrepuller) CreateFunc(component string) error {
|
||||
if component == "kube-controller-manager" {
|
||||
return errors.New("boo")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *failedCreatePrepuller) WaitFunc(component string) {}
|
||||
|
||||
func (p *failedCreatePrepuller) DeleteFunc(component string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// foreverWaitPrepuller is a fake prepuller that basically waits "forever" (10 mins, but longer than the 10sec timeout)
|
||||
type foreverWaitPrepuller struct{}
|
||||
|
||||
func NewForeverWaitPrepuller() Prepuller {
|
||||
return &foreverWaitPrepuller{}
|
||||
}
|
||||
|
||||
func (p *foreverWaitPrepuller) CreateFunc(component string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *foreverWaitPrepuller) WaitFunc(component string) {
|
||||
time.Sleep(10 * time.Minute)
|
||||
}
|
||||
|
||||
func (p *foreverWaitPrepuller) DeleteFunc(component string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// failedDeletePrepuller is a fake prepuller that errors for kube-scheduler in the DeleteFunc call
|
||||
type failedDeletePrepuller struct{}
|
||||
|
||||
func NewFailedDeletePrepuller() Prepuller {
|
||||
return &failedDeletePrepuller{}
|
||||
}
|
||||
|
||||
func (p *failedDeletePrepuller) CreateFunc(component string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *failedDeletePrepuller) WaitFunc(component string) {}
|
||||
|
||||
func (p *failedDeletePrepuller) DeleteFunc(component string) error {
|
||||
if component == "kube-scheduler" {
|
||||
return errors.New("boo")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// goodPrepuller is a fake prepuller that works as expected
|
||||
type goodPrepuller struct{}
|
||||
|
||||
func NewGoodPrepuller() Prepuller {
|
||||
return &goodPrepuller{}
|
||||
}
|
||||
|
||||
func (p *goodPrepuller) CreateFunc(component string) error {
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *goodPrepuller) WaitFunc(component string) {
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
}
|
||||
|
||||
func (p *goodPrepuller) DeleteFunc(component string) error {
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestPrepullImagesInParallel(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
p Prepuller
|
||||
timeout time.Duration
|
||||
expectedErr bool
|
||||
}{
|
||||
{
|
||||
name: "should error out; create failed",
|
||||
p: NewFailedCreatePrepuller(),
|
||||
timeout: constants.PrepullImagesInParallelTimeout,
|
||||
expectedErr: true,
|
||||
},
|
||||
{
|
||||
name: "should error out; timeout exceeded",
|
||||
p: NewForeverWaitPrepuller(),
|
||||
timeout: constants.PrepullImagesInParallelTimeout,
|
||||
expectedErr: true,
|
||||
},
|
||||
{
|
||||
name: "should error out; delete failed",
|
||||
p: NewFailedDeletePrepuller(),
|
||||
timeout: constants.PrepullImagesInParallelTimeout,
|
||||
expectedErr: true,
|
||||
},
|
||||
{
|
||||
name: "should work just fine",
|
||||
p: NewGoodPrepuller(),
|
||||
timeout: constants.PrepullImagesInParallelTimeout,
|
||||
expectedErr: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, rt := range tests {
|
||||
t.Run(rt.name, func(t *testing.T) {
|
||||
actualErr := PrepullImagesInParallel(rt.p, rt.timeout, append(constants.ControlPlaneComponents, constants.Etcd))
|
||||
if (actualErr != nil) != rt.expectedErr {
|
||||
t.Errorf(
|
||||
"failed TestPrepullImagesInParallel\n\texpected error: %t\n\tgot: %t",
|
||||
rt.expectedErr,
|
||||
(actualErr != nil),
|
||||
)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user