server.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329
  1. /*
  2. Copyright 2014 The Kubernetes Authors.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. */
  13. // Package app implements a Server object for running the scheduler.
  14. package app
  15. import (
  16. "context"
  17. "fmt"
  18. "io"
  19. "net/http"
  20. "os"
  21. goruntime "runtime"
  22. utilerrors "k8s.io/apimachinery/pkg/util/errors"
  23. utilruntime "k8s.io/apimachinery/pkg/util/runtime"
  24. "k8s.io/apiserver/pkg/authentication/authenticator"
  25. "k8s.io/apiserver/pkg/authorization/authorizer"
  26. genericapifilters "k8s.io/apiserver/pkg/endpoints/filters"
  27. apirequest "k8s.io/apiserver/pkg/endpoints/request"
  28. genericfilters "k8s.io/apiserver/pkg/server/filters"
  29. "k8s.io/apiserver/pkg/server/healthz"
  30. "k8s.io/apiserver/pkg/server/mux"
  31. "k8s.io/apiserver/pkg/server/routes"
  32. "k8s.io/apiserver/pkg/util/term"
  33. v1core "k8s.io/client-go/kubernetes/typed/core/v1"
  34. "k8s.io/client-go/tools/leaderelection"
  35. cliflag "k8s.io/component-base/cli/flag"
  36. "k8s.io/component-base/cli/globalflag"
  37. schedulerserverconfig "k8s.io/kubernetes/cmd/kube-scheduler/app/config"
  38. "k8s.io/kubernetes/cmd/kube-scheduler/app/options"
  39. "k8s.io/kubernetes/pkg/api/legacyscheme"
  40. "k8s.io/kubernetes/pkg/scheduler"
  41. "k8s.io/kubernetes/pkg/scheduler/algorithmprovider"
  42. kubeschedulerconfig "k8s.io/kubernetes/pkg/scheduler/apis/config"
  43. framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
  44. "k8s.io/kubernetes/pkg/scheduler/metrics"
  45. "k8s.io/kubernetes/pkg/util/configz"
  46. utilflag "k8s.io/kubernetes/pkg/util/flag"
  47. "k8s.io/kubernetes/pkg/version"
  48. "k8s.io/kubernetes/pkg/version/verflag"
  49. "github.com/prometheus/client_golang/prometheus"
  50. "github.com/spf13/cobra"
  51. "k8s.io/klog"
  52. )
  53. // NewSchedulerCommand creates a *cobra.Command object with default parameters
  54. func NewSchedulerCommand() *cobra.Command {
  55. opts, err := options.NewOptions()
  56. if err != nil {
  57. klog.Fatalf("unable to initialize command options: %v", err)
  58. }
  59. cmd := &cobra.Command{
  60. Use: "kube-scheduler",
  61. Long: `The Kubernetes scheduler is a policy-rich, topology-aware,
  62. workload-specific function that significantly impacts availability, performance,
  63. and capacity. The scheduler needs to take into account individual and collective
  64. resource requirements, quality of service requirements, hardware/software/policy
  65. constraints, affinity and anti-affinity specifications, data locality, inter-workload
  66. interference, deadlines, and so on. Workload-specific requirements will be exposed
  67. through the API as necessary.`,
  68. Run: func(cmd *cobra.Command, args []string) {
  69. if err := runCommand(cmd, args, opts); err != nil {
  70. fmt.Fprintf(os.Stderr, "%v\n", err)
  71. os.Exit(1)
  72. }
  73. },
  74. }
  75. fs := cmd.Flags()
  76. namedFlagSets := opts.Flags()
  77. verflag.AddFlags(namedFlagSets.FlagSet("global"))
  78. globalflag.AddGlobalFlags(namedFlagSets.FlagSet("global"), cmd.Name())
  79. for _, f := range namedFlagSets.FlagSets {
  80. fs.AddFlagSet(f)
  81. }
  82. usageFmt := "Usage:\n %s\n"
  83. cols, _, _ := term.TerminalSize(cmd.OutOrStdout())
  84. cmd.SetUsageFunc(func(cmd *cobra.Command) error {
  85. fmt.Fprintf(cmd.OutOrStderr(), usageFmt, cmd.UseLine())
  86. cliflag.PrintSections(cmd.OutOrStderr(), namedFlagSets, cols)
  87. return nil
  88. })
  89. cmd.SetHelpFunc(func(cmd *cobra.Command, args []string) {
  90. fmt.Fprintf(cmd.OutOrStdout(), "%s\n\n"+usageFmt, cmd.Long, cmd.UseLine())
  91. cliflag.PrintSections(cmd.OutOrStdout(), namedFlagSets, cols)
  92. })
  93. cmd.MarkFlagFilename("config", "yaml", "yml", "json")
  94. return cmd
  95. }
  96. // runCommand runs the scheduler.
  97. func runCommand(cmd *cobra.Command, args []string, opts *options.Options) error {
  98. verflag.PrintAndExitIfRequested()
  99. utilflag.PrintFlags(cmd.Flags())
  100. if len(args) != 0 {
  101. fmt.Fprint(os.Stderr, "arguments are not supported\n")
  102. }
  103. if errs := opts.Validate(); len(errs) > 0 {
  104. fmt.Fprintf(os.Stderr, "%v\n", utilerrors.NewAggregate(errs))
  105. os.Exit(1)
  106. }
  107. if len(opts.WriteConfigTo) > 0 {
  108. if err := options.WriteConfigFile(opts.WriteConfigTo, &opts.ComponentConfig); err != nil {
  109. fmt.Fprintf(os.Stderr, "%v\n", err)
  110. os.Exit(1)
  111. }
  112. klog.Infof("Wrote configuration to: %s\n", opts.WriteConfigTo)
  113. }
  114. c, err := opts.Config()
  115. if err != nil {
  116. fmt.Fprintf(os.Stderr, "%v\n", err)
  117. os.Exit(1)
  118. }
  119. stopCh := make(chan struct{})
  120. // Get the completed config
  121. cc := c.Complete()
  122. // To help debugging, immediately log version
  123. klog.Infof("Version: %+v", version.Get())
  124. // Apply algorithms based on feature gates.
  125. // TODO: make configurable?
  126. algorithmprovider.ApplyFeatureGates()
  127. // Configz registration.
  128. if cz, err := configz.New("componentconfig"); err == nil {
  129. cz.Set(cc.ComponentConfig)
  130. } else {
  131. return fmt.Errorf("unable to register configz: %s", err)
  132. }
  133. return Run(cc, stopCh)
  134. }
  135. // Run executes the scheduler based on the given configuration. It only return on error or when stopCh is closed.
  136. func Run(cc schedulerserverconfig.CompletedConfig, stopCh <-chan struct{}) error {
  137. // To help debugging, immediately log version
  138. klog.V(1).Infof("Starting Kubernetes Scheduler version %+v", version.Get())
  139. // Create the scheduler.
  140. sched, err := scheduler.New(cc.Client,
  141. cc.InformerFactory.Core().V1().Nodes(),
  142. cc.PodInformer,
  143. cc.InformerFactory.Core().V1().PersistentVolumes(),
  144. cc.InformerFactory.Core().V1().PersistentVolumeClaims(),
  145. cc.InformerFactory.Core().V1().ReplicationControllers(),
  146. cc.InformerFactory.Apps().V1().ReplicaSets(),
  147. cc.InformerFactory.Apps().V1().StatefulSets(),
  148. cc.InformerFactory.Core().V1().Services(),
  149. cc.InformerFactory.Policy().V1beta1().PodDisruptionBudgets(),
  150. cc.InformerFactory.Storage().V1().StorageClasses(),
  151. cc.Recorder,
  152. cc.ComponentConfig.AlgorithmSource,
  153. stopCh,
  154. framework.NewRegistry(),
  155. cc.ComponentConfig.Plugins,
  156. cc.ComponentConfig.PluginConfig,
  157. scheduler.WithName(cc.ComponentConfig.SchedulerName),
  158. scheduler.WithHardPodAffinitySymmetricWeight(cc.ComponentConfig.HardPodAffinitySymmetricWeight),
  159. scheduler.WithPreemptionDisabled(cc.ComponentConfig.DisablePreemption),
  160. scheduler.WithPercentageOfNodesToScore(cc.ComponentConfig.PercentageOfNodesToScore),
  161. scheduler.WithBindTimeoutSeconds(*cc.ComponentConfig.BindTimeoutSeconds))
  162. if err != nil {
  163. return err
  164. }
  165. // Prepare the event broadcaster.
  166. if cc.Broadcaster != nil && cc.EventClient != nil {
  167. cc.Broadcaster.StartLogging(klog.V(6).Infof)
  168. cc.Broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: cc.EventClient.Events("")})
  169. }
  170. // Setup healthz checks.
  171. var checks []healthz.HealthzChecker
  172. if cc.ComponentConfig.LeaderElection.LeaderElect {
  173. checks = append(checks, cc.LeaderElection.WatchDog)
  174. }
  175. // Start up the healthz server.
  176. if cc.InsecureServing != nil {
  177. separateMetrics := cc.InsecureMetricsServing != nil
  178. handler := buildHandlerChain(newHealthzHandler(&cc.ComponentConfig, separateMetrics, checks...), nil, nil)
  179. if err := cc.InsecureServing.Serve(handler, 0, stopCh); err != nil {
  180. return fmt.Errorf("failed to start healthz server: %v", err)
  181. }
  182. }
  183. if cc.InsecureMetricsServing != nil {
  184. handler := buildHandlerChain(newMetricsHandler(&cc.ComponentConfig), nil, nil)
  185. if err := cc.InsecureMetricsServing.Serve(handler, 0, stopCh); err != nil {
  186. return fmt.Errorf("failed to start metrics server: %v", err)
  187. }
  188. }
  189. if cc.SecureServing != nil {
  190. handler := buildHandlerChain(newHealthzHandler(&cc.ComponentConfig, false, checks...), cc.Authentication.Authenticator, cc.Authorization.Authorizer)
  191. // TODO: handle stoppedCh returned by c.SecureServing.Serve
  192. if _, err := cc.SecureServing.Serve(handler, 0, stopCh); err != nil {
  193. // fail early for secure handlers, removing the old error loop from above
  194. return fmt.Errorf("failed to start secure server: %v", err)
  195. }
  196. }
  197. // Start all informers.
  198. go cc.PodInformer.Informer().Run(stopCh)
  199. cc.InformerFactory.Start(stopCh)
  200. // Wait for all caches to sync before scheduling.
  201. cc.InformerFactory.WaitForCacheSync(stopCh)
  202. // Prepare a reusable runCommand function.
  203. run := func(ctx context.Context) {
  204. sched.Run()
  205. <-ctx.Done()
  206. }
  207. ctx, cancel := context.WithCancel(context.TODO()) // TODO once Run() accepts a context, it should be used here
  208. defer cancel()
  209. go func() {
  210. select {
  211. case <-stopCh:
  212. cancel()
  213. case <-ctx.Done():
  214. }
  215. }()
  216. // If leader election is enabled, runCommand via LeaderElector until done and exit.
  217. if cc.LeaderElection != nil {
  218. cc.LeaderElection.Callbacks = leaderelection.LeaderCallbacks{
  219. OnStartedLeading: run,
  220. OnStoppedLeading: func() {
  221. utilruntime.HandleError(fmt.Errorf("lost master"))
  222. },
  223. }
  224. leaderElector, err := leaderelection.NewLeaderElector(*cc.LeaderElection)
  225. if err != nil {
  226. return fmt.Errorf("couldn't create leader elector: %v", err)
  227. }
  228. leaderElector.Run(ctx)
  229. return fmt.Errorf("lost lease")
  230. }
  231. // Leader election is disabled, so runCommand inline until done.
  232. run(ctx)
  233. return fmt.Errorf("finished without leader elect")
  234. }
  235. // buildHandlerChain wraps the given handler with the standard filters.
  236. func buildHandlerChain(handler http.Handler, authn authenticator.Request, authz authorizer.Authorizer) http.Handler {
  237. requestInfoResolver := &apirequest.RequestInfoFactory{}
  238. failedHandler := genericapifilters.Unauthorized(legacyscheme.Codecs, false)
  239. handler = genericapifilters.WithRequestInfo(handler, requestInfoResolver)
  240. handler = genericapifilters.WithAuthorization(handler, authz, legacyscheme.Codecs)
  241. handler = genericapifilters.WithAuthentication(handler, authn, failedHandler, nil)
  242. handler = genericapifilters.WithRequestInfo(handler, requestInfoResolver)
  243. handler = genericfilters.WithPanicRecovery(handler)
  244. return handler
  245. }
  246. func installMetricHandler(pathRecorderMux *mux.PathRecorderMux) {
  247. configz.InstallHandler(pathRecorderMux)
  248. defaultMetricsHandler := prometheus.Handler().ServeHTTP
  249. pathRecorderMux.HandleFunc("/metrics", func(w http.ResponseWriter, req *http.Request) {
  250. if req.Method == "DELETE" {
  251. metrics.Reset()
  252. io.WriteString(w, "metrics reset\n")
  253. return
  254. }
  255. defaultMetricsHandler(w, req)
  256. })
  257. }
  258. // newMetricsHandler builds a metrics server from the config.
  259. func newMetricsHandler(config *kubeschedulerconfig.KubeSchedulerConfiguration) http.Handler {
  260. pathRecorderMux := mux.NewPathRecorderMux("kube-scheduler")
  261. installMetricHandler(pathRecorderMux)
  262. if config.EnableProfiling {
  263. routes.Profiling{}.Install(pathRecorderMux)
  264. if config.EnableContentionProfiling {
  265. goruntime.SetBlockProfileRate(1)
  266. }
  267. }
  268. return pathRecorderMux
  269. }
  270. // newHealthzHandler creates a healthz server from the config, and will also
  271. // embed the metrics handler if the healthz and metrics address configurations
  272. // are the same.
  273. func newHealthzHandler(config *kubeschedulerconfig.KubeSchedulerConfiguration, separateMetrics bool, checks ...healthz.HealthzChecker) http.Handler {
  274. pathRecorderMux := mux.NewPathRecorderMux("kube-scheduler")
  275. healthz.InstallHandler(pathRecorderMux, checks...)
  276. if !separateMetrics {
  277. installMetricHandler(pathRecorderMux)
  278. }
  279. if config.EnableProfiling {
  280. routes.Profiling{}.Install(pathRecorderMux)
  281. if config.EnableContentionProfiling {
  282. goruntime.SetBlockProfileRate(1)
  283. }
  284. }
  285. return pathRecorderMux
  286. }