github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-19 09:52:49 +00:00
Code Issues Projects Releases Wiki Activity

Add stopCh to apiserver & context to kublet commands

Browse Source 
Remove SetupSignalContext call from the apiserver

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>
This commit is contained in:
Darren Shepherd 2019-08-26 15:55:35 -07:00 committed by Rafael Breno
parent c774e60758
commit dd1bd2ce00
9 changed files with 36 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
cmd/genkubedocs/gen_kube_docs.go
View File

@ -24,6 +24,7 @@ import (
"github.com/spf13/cobra/doc"
"github.com/spf13/pflag"
"k8s.io/apiserver/pkg/server"
"k8s.io/kubernetes/cmd/genutils"
apiservapp "k8s.io/kubernetes/cmd/kube-apiserver/app"
cmapp "k8s.io/kubernetes/cmd/kube-controller-manager/app"
@ -54,7 +55,7 @@ func main() {
switch module {
case "kube-apiserver":
// generate docs for kube-apiserver
apiserver := apiservapp.NewAPIServerCommand()
apiserver := apiservapp.NewAPIServerCommand(server.SetupSignalHandler())
doc.GenMarkdownTree(apiserver, outDir)
case "kube-controller-manager":
// generate docs for kube-controller-manager
@ -66,11 +67,11 @@ func main() {
doc.GenMarkdownTree(proxy, outDir)
case "kube-scheduler":
// generate docs for kube-scheduler
scheduler := schapp.NewSchedulerCommand()
scheduler := schapp.NewSchedulerCommand(server.SetupSignalHandler())
doc.GenMarkdownTree(scheduler, outDir)
case "kubelet":
// generate docs for kubelet
kubelet := kubeletapp.NewKubeletCommand()
kubelet := kubeletapp.NewKubeletCommand(server.SetupSignalContext())
doc.GenMarkdownTree(kubelet, outDir)
case "kubeadm":
// resets global flags created by kubelet or other commands e.g.

7
cmd/genman/gen_kube_man.go
View File

@ -26,6 +26,7 @@ import (
mangen "github.com/cpuguy83/go-md2man/v2/md2man"
"github.com/spf13/cobra"
"github.com/spf13/pflag"
"k8s.io/apiserver/pkg/server"
"k8s.io/cli-runtime/pkg/genericiooptions"
kubectlcmd "k8s.io/kubectl/pkg/cmd"
"k8s.io/kubernetes/cmd/genutils"
@ -62,7 +63,7 @@ func main() {
switch module {
case "kube-apiserver":
// generate manpage for kube-apiserver
apiserver := apiservapp.NewAPIServerCommand()
apiserver := apiservapp.NewAPIServerCommand(server.SetupSignalHandler())
genMarkdown(apiserver, "", outDir)
for _, c := range apiserver.Commands() {
genMarkdown(c, "kube-apiserver", outDir)
@ -83,14 +84,14 @@ func main() {
}
case "kube-scheduler":
// generate manpage for kube-scheduler
scheduler := schapp.NewSchedulerCommand()
scheduler := schapp.NewSchedulerCommand(server.SetupSignalHandler())
genMarkdown(scheduler, "", outDir)
for _, c := range scheduler.Commands() {
genMarkdown(c, "kube-scheduler", outDir)
}
case "kubelet":
// generate manpage for kubelet
kubelet := kubeletapp.NewKubeletCommand()
kubelet := kubeletapp.NewKubeletCommand(server.SetupSignalContext())
genMarkdown(kubelet, "", outDir)
for _, c := range kubelet.Commands() {
genMarkdown(c, "kubelet", outDir)

3
cmd/kube-apiserver/apiserver.go
View File

@ -22,6 +22,7 @@ import (
"os"
_ "time/tzdata" // for timeZone support in CronJob
"k8s.io/apiserver/pkg/server"
"k8s.io/component-base/cli"
_ "k8s.io/component-base/logs/json/register" // for JSON log format registration
_ "k8s.io/component-base/metrics/prometheus/clientgo" // load all the prometheus client-go plugins
@ -30,7 +31,7 @@ import (
)
func main() {
command := app.NewAPIServerCommand()
command := app.NewAPIServerCommand(server.SetupSignalHandler())
code := cli.Run(command)
os.Exit(code)
}

7
cmd/kube-apiserver/app/server.go
View File

@ -65,7 +65,7 @@ func init() {
}
// NewAPIServerCommand creates a *cobra.Command object with default parameters
func NewAPIServerCommand() *cobra.Command {
func NewAPIServerCommand(stopCh <-chan struct{}) *cobra.Command {
s := options.NewServerRunOptions()
ctx := genericapiserver.SetupSignalContext()
featureGate := s.GenericServerRunOptions.ComponentGlobalsRegistry.FeatureGateFor(basecompatibility.DefaultKubeComponent)
@ -110,7 +110,7 @@ cluster's shared state through which all other components interact.`,
}
// add feature enablement metrics
featureGate.(featuregate.MutableFeatureGate).AddMetrics()
return Run(ctx, completedOptions)
return Run(cmd.Context(), completedOptions, stopCh)
},
Args: func(cmd *cobra.Command, args []string) error {
for _, arg := range args {
@ -121,7 +121,6 @@ cluster's shared state through which all other components interact.`,
return nil
},
}
cmd.SetContext(ctx)
fs := cmd.Flags()
namedFlagSets := s.Flags()
@ -142,7 +141,7 @@ cluster's shared state through which all other components interact.`,
}
// Run runs the specified APIServer. This should never exit.
func Run(ctx context.Context, opts options.CompletedOptions) error {
func Run(ctx context.Context, opts options.CompletedOptions, stopCh <-chan struct{}) error {
// To help debugging, immediately log version
klog.Infof("Version: %+v", utilversion.Get())

12
cmd/kube-scheduler/app/server.go
View File

@ -34,7 +34,6 @@ import (
"k8s.io/apiserver/pkg/authorization/authorizer"
genericapifilters "k8s.io/apiserver/pkg/endpoints/filters"
apirequest "k8s.io/apiserver/pkg/endpoints/request"
"k8s.io/apiserver/pkg/server"
genericfilters "k8s.io/apiserver/pkg/server/filters"
"k8s.io/apiserver/pkg/server/healthz"
"k8s.io/apiserver/pkg/server/mux"
@ -86,7 +85,11 @@ func init() {
type Option func(runtime.Registry) error
// NewSchedulerCommand creates a *cobra.Command object with default parameters and registryOptions
func NewSchedulerCommand(registryOptions ...Option) *cobra.Command {
func NewSchedulerCommand(stopCh <-chan struct{}, registryOptions ...Option) *cobra.Command {
// explicitly register (if not already registered) the kube effective version and feature gate in DefaultComponentGlobalsRegistry,
// which will be used in NewOptions.
_, _ = s.De.DefaultComponentGlobalsRegistry.ComponentGlobalsOrRegister(
featuregate.DefaultKubeComponent, utilversion.DefaultBuildEffectiveVersion(), utilfeature.DefaultMutableFeatureGate)
opts := options.NewOptions()
cmd := &cobra.Command{
@ -104,7 +107,7 @@ for more information about scheduling and the kube-scheduler component.`,
return opts.ComponentGlobalsRegistry.Set()
},
RunE: func(cmd *cobra.Command, args []string) error {
return runCommand(cmd, opts, registryOptions...)
return runCommand(cmd, opts, stopCh, registryOptions...)
},
Args: func(cmd *cobra.Command, args []string) error {
for _, arg := range args {
@ -135,7 +138,7 @@ for more information about scheduling and the kube-scheduler component.`,
}
// runCommand runs the scheduler.
func runCommand(cmd *cobra.Command, opts *options.Options, registryOptions ...Option) error {
func runCommand(cmd *cobra.Command, opts *options.Options, stopCh <-chan struct{}, registryOptions ...Option) error {
verflag.PrintAndExitIfRequested()
fg := opts.ComponentGlobalsRegistry.FeatureGateFor(basecompatibility.DefaultKubeComponent)
// Activate logging as soon as possible, after that
@ -149,7 +152,6 @@ func runCommand(cmd *cobra.Command, opts *options.Options, registryOptions ...Op
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go func() {
stopCh := server.SetupSignalHandler()
<-stopCh
cancel()
}()

3
cmd/kube-scheduler/scheduler.go
View File

@ -19,6 +19,7 @@ package main
import (
"os"
"k8s.io/apiserver/pkg/server"
"k8s.io/component-base/cli"
_ "k8s.io/component-base/logs/json/register" // for JSON log format registration
_ "k8s.io/component-base/metrics/prometheus/clientgo"
@ -27,7 +28,7 @@ import (
)
func main() {
command := app.NewSchedulerCommand()
command := app.NewSchedulerCommand(server.SetupSignalHandler())
code := cli.Run(command)
os.Exit(code)
}

14
cmd/kubelet/app/server.go
View File

@ -63,7 +63,6 @@ import (
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/validation/field"
"k8s.io/apimachinery/pkg/util/wait"
genericapiserver "k8s.io/apiserver/pkg/server"
"k8s.io/apiserver/pkg/server/healthz"
utilfeature "k8s.io/apiserver/pkg/util/feature"
clientset "k8s.io/client-go/kubernetes"
@ -135,7 +134,7 @@ func init() {
}
// NewKubeletCommand creates a *cobra.Command object with default parameters
func NewKubeletCommand() *cobra.Command {
func NewKubeletCommand(ctx context.Context) *cobra.Command {
cleanFlagSet := pflag.NewFlagSet(server.ComponentKubelet, pflag.ContinueOnError)
cleanFlagSet.SetNormalizeFunc(cliflag.WordSepNormalizeFunc)
kubeletFlags := options.NewKubeletFlags()
@ -279,7 +278,6 @@ is checked every 20 seconds (also configurable with a flag).`,
if err := checkPermissions(); err != nil {
klog.ErrorS(err, "kubelet running with insufficient permissions")
}
// make the kubelet's config safe for logging
config := kubeletServer.KubeletConfiguration.DeepCopy()
for k := range config.StaticPodURLHeader {
@ -288,9 +286,6 @@ is checked every 20 seconds (also configurable with a flag).`,
// log the kubelet's config for inspection
klog.V(5).InfoS("KubeletConfiguration", "configuration", klog.Format(config))
// set up signal context for kubelet shutdown
ctx := genericapiserver.SetupSignalContext()
utilfeature.DefaultMutableFeatureGate.AddMetrics()
// run the kubelet
return Run(ctx, kubeletServer, kubeletDeps, utilfeature.DefaultFeatureGate)
@ -518,7 +513,8 @@ func UnsecuredDependencies(s *options.KubeletServer, featureGate featuregate.Fea
OSInterface: kubecontainer.RealOS{},
VolumePlugins: plugins,
DynamicPluginProber: GetDynamicPluginProber(s.VolumePluginDir, pluginRunner),
TLSOptions: tlsOptions}, nil
TLSOptions: tlsOptions,
}, nil
}
// Run runs the specified KubeletServer with the given Dependencies. This should never exit.
@ -884,7 +880,6 @@ func run(ctx context.Context, s *options.KubeletServer, kubeDeps *kubelet.Depend
kubeDeps.Recorder,
kubeDeps.KubeClient,
)
if err != nil {
return err
}
@ -1310,7 +1305,8 @@ func createAndInitKubelet(kubeServer *options.KubeletServer,
hostname string,
hostnameOverridden bool,
nodeName types.NodeName,
nodeIPs []net.IP) (k kubelet.Bootstrap, err error) {
nodeIPs []net.IP,
) (k kubelet.Bootstrap, err error) {
// TODO: block until all sources have delivered at least one update to the channel, or break the sync loop
// up into "per source" synchronizations

3
cmd/kubelet/kubelet.go
View File

@ -24,6 +24,7 @@ package main
import (
"os"
"k8s.io/apiserver/pkg/server"
"k8s.io/component-base/cli"
_ "k8s.io/component-base/logs/json/register" // for JSON log format registration
_ "k8s.io/component-base/metrics/prometheus/clientgo" // for client metric registration
@ -32,7 +33,7 @@ import (
)
func main() {
command := app.NewKubeletCommand()
command := app.NewKubeletCommand(server.SetupSignalContext())
code := cli.Run(command)
os.Exit(code)
}

8
test/e2e_node/services/apiserver.go
View File

@ -47,6 +47,7 @@ AwEHoUQDQgAEH6cuzP8XuD5wal6wf9M6xDljTOPLX2i8uIp/C/ASqiIGUeeKQtX0
// APIServer is a server which manages apiserver.
type APIServer struct {
storageConfig storagebackend.Config
stopCh chan struct{}
cancel func(error)
}
@ -54,6 +55,7 @@ type APIServer struct {
func NewAPIServer(storageConfig storagebackend.Config) *APIServer {
return &APIServer{
storageConfig: storageConfig,
stopCh: make(chan struct{}),
}
}
@ -112,7 +114,7 @@ func (a *APIServer) Start(ctx context.Context) error {
return
}
err = apiserver.Run(ctx, completedOptions)
err = apiserver.Run(ctx, completedOptions, a.stopCh)
if err != nil {
errCh <- fmt.Errorf("run apiserver error: %w", err)
return
@ -132,6 +134,10 @@ func (a *APIServer) Stop() error {
if a.cancel != nil {
a.cancel(errors.New("stopping API server"))
}
if a.stopCh != nil {
close(a.stopCh)
a.stopCh = nil
}
return nil
}