github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-08-03 17:30:00 +00:00
Code Issues Projects Releases Wiki Activity

Make scheduler service variables private

Browse Source
This commit is contained in:
Dr. Stefan Schimanski 2015-10-26 21:29:32 -05:00
parent 0bd1666d9b
commit 18fbc1fe57
3 changed files with 263 additions and 263 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
contrib/mesos/pkg/scheduler/service/publish.go
View File

@ -42,7 +42,7 @@ func (m *SchedulerServer) newServiceWriter(stop <-chan struct{}) func() {
glog.Errorf("Can't create scheduler service: %v", err) glog.Errorf("Can't create scheduler service: %v", err)
} }
if err := m.setEndpoints(SCHEDULER_SERVICE_NAME, net.IP(m.Address), m.Port); err != nil { if err := m.setEndpoints(SCHEDULER_SERVICE_NAME, net.IP(m.address), m.port); err != nil {
glog.Errorf("Can't create scheduler endpoints: %v", err) glog.Errorf("Can't create scheduler endpoints: %v", err)
} }
@ -76,8 +76,8 @@ func (m *SchedulerServer) createSchedulerServiceIfNeeded(serviceName string, ser
SessionAffinity: api.ServiceAffinityNone, SessionAffinity: api.ServiceAffinityNone,
}, },
} }
if m.ServiceAddress != nil { if m.serviceAddress != nil {
svc.Spec.ClusterIP = m.ServiceAddress.String() svc.Spec.ClusterIP = m.serviceAddress.String()
} }
_, err := m.client.Services(api.NamespaceValue(ctx)).Create(svc) _, err := m.client.Services(api.NamespaceValue(ctx)).Create(svc)
if err != nil && errors.IsAlreadyExists(err) { if err != nil && errors.IsAlreadyExists(err) {

516
contrib/mesos/pkg/scheduler/service/service.go
View File

@ -87,78 +87,78 @@ const (
) )
type SchedulerServer struct { type SchedulerServer struct {
Port int port int
Address net.IP address net.IP
EnableProfiling bool enableProfiling bool
AuthPath string authPath string
APIServerList []string apiServerList []string
EtcdServerList []string etcdServerList []string
EtcdConfigFile string etcdConfigFile string
AllowPrivileged bool allowPrivileged bool
ExecutorPath string executorPath string
ProxyPath string proxyPath string
MesosMaster string mesosMaster string
MesosUser string mesosUser string
MesosRole string mesosRole string
MesosAuthPrincipal string mesosAuthPrincipal string
MesosAuthSecretFile string mesosAuthSecretFile string
MesosCgroupPrefix string mesosCgroupPrefix string
MesosExecutorCPUs mresource.CPUShares mesosExecutorCPUs mresource.CPUShares
MesosExecutorMem mresource.MegaBytes mesosExecutorMem mresource.MegaBytes
Checkpoint bool checkpoint bool
FailoverTimeout float64 failoverTimeout float64
ExecutorLogV int executorLogV int
ExecutorBindall bool executorBindall bool
ExecutorSuicideTimeout time.Duration executorSuicideTimeout time.Duration
LaunchGracePeriod time.Duration launchGracePeriod time.Duration
RunProxy bool runProxy bool
ProxyBindall bool proxyBindall bool
ProxyLogV int proxyLogV int
MinionPathOverride string minionPathOverride string
MinionLogMaxSize resource.Quantity minionLogMaxSize resource.Quantity
MinionLogMaxBackups int minionLogMaxBackups int
MinionLogMaxAgeInDays int minionLogMaxAgeInDays int
MesosAuthProvider string mesosAuthProvider string
DriverPort uint driverPort uint
HostnameOverride string hostnameOverride string
ReconcileInterval int64 reconcileInterval int64
ReconcileCooldown time.Duration reconcileCooldown time.Duration
DefaultContainerCPULimit mresource.CPUShares defaultContainerCPULimit mresource.CPUShares
DefaultContainerMemLimit mresource.MegaBytes defaultContainerMemLimit mresource.MegaBytes
SchedulerConfigFileName string schedulerConfigFileName string
Graceful bool graceful bool
FrameworkName string frameworkName string
FrameworkWebURI string frameworkWebURI string
HA bool ha bool
AdvertisedAddress string advertisedAddress string
ServiceAddress net.IP serviceAddress net.IP
HADomain string haDomain string
KMPath string kmPath string
ClusterDNS net.IP clusterDNS net.IP
ClusterDomain string clusterDomain string
KubeletRootDirectory string kubeletRootDirectory string
KubeletDockerEndpoint string kubeletDockerEndpoint string
KubeletPodInfraContainerImage string kubeletPodInfraContainerImage string
KubeletCadvisorPort uint kubeletCadvisorPort uint
KubeletHostNetworkSources string kubeletHostNetworkSources string
KubeletSyncFrequency time.Duration kubeletSyncFrequency time.Duration
KubeletNetworkPluginName string kubeletNetworkPluginName string
StaticPodsConfigPath string staticPodsConfigPath string
DockerCfgPath string dockerCfgPath string
ContainPodResources bool containPodResources bool
AccountForPodResources bool accountForPodResources bool
nodeRelistPeriod time.Duration nodeRelistPeriod time.Duration
SandboxOverlay string sandboxOverlay string
executable string // path to the binary running this service executable string // path to the binary running this service
client *client.Client client *client.Client
driver bindings.SchedulerDriver driver bindings.SchedulerDriver
driverMutex sync.RWMutex driverMutex sync.RWMutex
mux *http.ServeMux mux *http.ServeMux
} }
// useful for unit testing specific funcs // useful for unit testing specific funcs
@ -171,36 +171,36 @@ type schedulerProcessInterface interface {
// NewSchedulerServer creates a new SchedulerServer with default parameters // NewSchedulerServer creates a new SchedulerServer with default parameters
func NewSchedulerServer() *SchedulerServer { func NewSchedulerServer() *SchedulerServer {
s := SchedulerServer{ s := SchedulerServer{
Port: ports.SchedulerPort, port: ports.SchedulerPort,
Address: net.ParseIP("127.0.0.1"), address: net.ParseIP("127.0.0.1"),
FailoverTimeout: time.Duration((1 << 62) - 1).Seconds(), failoverTimeout: time.Duration((1 << 62) - 1).Seconds(),
RunProxy: true, runProxy: true,
ExecutorSuicideTimeout: execcfg.DefaultSuicideTimeout, executorSuicideTimeout: execcfg.DefaultSuicideTimeout,
LaunchGracePeriod: execcfg.DefaultLaunchGracePeriod, launchGracePeriod: execcfg.DefaultLaunchGracePeriod,
DefaultContainerCPULimit: mresource.DefaultDefaultContainerCPULimit, defaultContainerCPULimit: mresource.DefaultDefaultContainerCPULimit,
DefaultContainerMemLimit: mresource.DefaultDefaultContainerMemLimit, defaultContainerMemLimit: mresource.DefaultDefaultContainerMemLimit,
MinionLogMaxSize: minioncfg.DefaultLogMaxSize(), minionLogMaxSize: minioncfg.DefaultLogMaxSize(),
MinionLogMaxBackups: minioncfg.DefaultLogMaxBackups, minionLogMaxBackups: minioncfg.DefaultLogMaxBackups,
MinionLogMaxAgeInDays: minioncfg.DefaultLogMaxAgeInDays, minionLogMaxAgeInDays: minioncfg.DefaultLogMaxAgeInDays,
MesosAuthProvider: sasl.ProviderName, mesosAuthProvider: sasl.ProviderName,
MesosCgroupPrefix: minioncfg.DefaultCgroupPrefix, mesosCgroupPrefix: minioncfg.DefaultCgroupPrefix,
MesosMaster: defaultMesosMaster, mesosMaster: defaultMesosMaster,
MesosUser: defaultMesosUser, mesosUser: defaultMesosUser,
MesosExecutorCPUs: defaultExecutorCPUs, mesosExecutorCPUs: defaultExecutorCPUs,
MesosExecutorMem: defaultExecutorMem, mesosExecutorMem: defaultExecutorMem,
ReconcileInterval: defaultReconcileInterval, reconcileInterval: defaultReconcileInterval,
ReconcileCooldown: defaultReconcileCooldown, reconcileCooldown: defaultReconcileCooldown,
Checkpoint: true, checkpoint: true,
FrameworkName: defaultFrameworkName, frameworkName: defaultFrameworkName,
HA: false, ha: false,
mux: http.NewServeMux(), mux: http.NewServeMux(),
KubeletCadvisorPort: 4194, // copied from github.com/GoogleCloudPlatform/kubernetes/blob/release-0.14/cmd/kubelet/app/server.go kubeletCadvisorPort: 4194, // copied from github.com/GoogleCloudPlatform/kubernetes/blob/release-0.14/cmd/kubelet/app/server.go
KubeletSyncFrequency: 10 * time.Second, kubeletSyncFrequency: 10 * time.Second,
ContainPodResources: true, containPodResources: true,
AccountForPodResources: true, accountForPodResources: true,
nodeRelistPeriod: defaultNodeRelistPeriod, nodeRelistPeriod: defaultNodeRelistPeriod,
} }
// cache this for later use. also useful in case the original binary gets deleted, e.g. // cache this for later use. also useful in case the original binary gets deleted, e.g.
@ -209,76 +209,76 @@ func NewSchedulerServer() *SchedulerServer {
log.Fatalf("failed to determine path to currently running executable: %v", err) log.Fatalf("failed to determine path to currently running executable: %v", err)
} else { } else {
s.executable = filename s.executable = filename
s.KMPath = filename s.kmPath = filename
} }
return &s return &s
} }
func (s *SchedulerServer) addCoreFlags(fs *pflag.FlagSet) { func (s *SchedulerServer) addCoreFlags(fs *pflag.FlagSet) {
fs.IntVar(&s.Port, "port", s.Port, "The port that the scheduler's http service runs on") fs.IntVar(&s.port, "port", s.port, "The port that the scheduler's http service runs on")
fs.IPVar(&s.Address, "address", s.Address, "The IP address to serve on (set to 0.0.0.0 for all interfaces)") fs.IPVar(&s.address, "address", s.address, "The IP address to serve on (set to 0.0.0.0 for all interfaces)")
fs.BoolVar(&s.EnableProfiling, "profiling", s.EnableProfiling, "Enable profiling via web interface host:port/debug/pprof/") fs.BoolVar(&s.enableProfiling, "profiling", s.enableProfiling, "Enable profiling via web interface host:port/debug/pprof/")
fs.StringSliceVar(&s.APIServerList, "api-servers", s.APIServerList, "List of Kubernetes API servers for publishing events, and reading pods and services. (ip:port), comma separated.") fs.StringSliceVar(&s.apiServerList, "api-servers", s.apiServerList, "List of Kubernetes API servers for publishing events, and reading pods and services. (ip:port), comma separated.")
fs.StringVar(&s.AuthPath, "auth-path", s.AuthPath, "Path to .kubernetes_auth file, specifying how to authenticate to API server.") fs.StringVar(&s.authPath, "auth-path", s.authPath, "Path to .kubernetes_auth file, specifying how to authenticate to API server.")
fs.StringSliceVar(&s.EtcdServerList, "etcd-servers", s.EtcdServerList, "List of etcd servers to watch (http://ip:port), comma separated. Mutually exclusive with --etcd-config") fs.StringSliceVar(&s.etcdServerList, "etcd-servers", s.etcdServerList, "List of etcd servers to watch (http://ip:port), comma separated. Mutually exclusive with --etcd-config")
fs.StringVar(&s.EtcdConfigFile, "etcd-config", s.EtcdConfigFile, "The config file for the etcd client. Mutually exclusive with --etcd-servers.") fs.StringVar(&s.etcdConfigFile, "etcd-config", s.etcdConfigFile, "The config file for the etcd client. Mutually exclusive with --etcd-servers.")
fs.BoolVar(&s.AllowPrivileged, "allow-privileged", s.AllowPrivileged, "If true, allow privileged containers.") fs.BoolVar(&s.allowPrivileged, "allow-privileged", s.allowPrivileged, "If true, allow privileged containers.")
fs.StringVar(&s.ClusterDomain, "cluster-domain", s.ClusterDomain, "Domain for this cluster. If set, kubelet will configure all containers to search this domain in addition to the host's search domains") fs.StringVar(&s.clusterDomain, "cluster-domain", s.clusterDomain, "Domain for this cluster. If set, kubelet will configure all containers to search this domain in addition to the host's search domains")
fs.IPVar(&s.ClusterDNS, "cluster-dns", s.ClusterDNS, "IP address for a cluster DNS server. If set, kubelet will configure all containers to use this for DNS resolution in addition to the host's DNS servers") fs.IPVar(&s.clusterDNS, "cluster-dns", s.clusterDNS, "IP address for a cluster DNS server. If set, kubelet will configure all containers to use this for DNS resolution in addition to the host's DNS servers")
fs.StringVar(&s.StaticPodsConfigPath, "static-pods-config", s.StaticPodsConfigPath, "Path for specification of static pods. Path should point to dir containing the staticPods configuration files. Defaults to none.") fs.StringVar(&s.staticPodsConfigPath, "static-pods-config", s.staticPodsConfigPath, "Path for specification of static pods. Path should point to dir containing the staticPods configuration files. Defaults to none.")
fs.StringVar(&s.MesosMaster, "mesos-master", s.MesosMaster, "Location of the Mesos master. The format is a comma-delimited list of of hosts like zk://host1:port,host2:port/mesos. If using ZooKeeper, pay particular attention to the leading zk:// and trailing /mesos! If not using ZooKeeper, standard URLs like http://localhost are also acceptable.") fs.StringVar(&s.mesosMaster, "mesos-master", s.mesosMaster, "Location of the Mesos master. The format is a comma-delimited list of of hosts like zk://host1:port,host2:port/mesos. If using ZooKeeper, pay particular attention to the leading zk:// and trailing /mesos! If not using ZooKeeper, standard URLs like http://localhost are also acceptable.")
fs.StringVar(&s.MesosUser, "mesos-user", s.MesosUser, "Mesos user for this framework, defaults to root.") fs.StringVar(&s.mesosUser, "mesos-user", s.mesosUser, "Mesos user for this framework, defaults to root.")
fs.StringVar(&s.MesosRole, "mesos-role", s.MesosRole, "Mesos role for this framework, defaults to none.") fs.StringVar(&s.mesosRole, "mesos-role", s.mesosRole, "Mesos role for this framework, defaults to none.")
fs.StringVar(&s.MesosAuthPrincipal, "mesos-authentication-principal", s.MesosAuthPrincipal, "Mesos authentication principal.") fs.StringVar(&s.mesosAuthPrincipal, "mesos-authentication-principal", s.mesosAuthPrincipal, "Mesos authentication principal.")
fs.StringVar(&s.MesosAuthSecretFile, "mesos-authentication-secret-file", s.MesosAuthSecretFile, "Mesos authentication secret file.") fs.StringVar(&s.mesosAuthSecretFile, "mesos-authentication-secret-file", s.mesosAuthSecretFile, "Mesos authentication secret file.")
fs.StringVar(&s.MesosAuthProvider, "mesos-authentication-provider", s.MesosAuthProvider, fmt.Sprintf("Authentication provider to use, default is SASL that supports mechanisms: %+v", mech.ListSupported())) fs.StringVar(&s.mesosAuthProvider, "mesos-authentication-provider", s.mesosAuthProvider, fmt.Sprintf("Authentication provider to use, default is SASL that supports mechanisms: %+v", mech.ListSupported()))
fs.StringVar(&s.DockerCfgPath, "dockercfg-path", s.DockerCfgPath, "Path to a dockercfg file that will be used by the docker instance of the minions.") fs.StringVar(&s.dockerCfgPath, "dockercfg-path", s.dockerCfgPath, "Path to a dockercfg file that will be used by the docker instance of the minions.")
fs.StringVar(&s.MesosCgroupPrefix, "mesos-cgroup-prefix", s.MesosCgroupPrefix, "The cgroup prefix concatenated with MESOS_DIRECTORY must give the executor cgroup set by Mesos") fs.StringVar(&s.mesosCgroupPrefix, "mesos-cgroup-prefix", s.mesosCgroupPrefix, "The cgroup prefix concatenated with MESOS_DIRECTORY must give the executor cgroup set by Mesos")
fs.Var(&s.MesosExecutorCPUs, "mesos-executor-cpus", "Initial CPU shares to allocate for each Mesos executor container.") fs.Var(&s.mesosExecutorCPUs, "mesos-executor-cpus", "Initial CPU shares to allocate for each Mesos executor container.")
fs.Var(&s.MesosExecutorMem, "mesos-executor-mem", "Initial memory (MB) to allocate for each Mesos executor container.") fs.Var(&s.mesosExecutorMem, "mesos-executor-mem", "Initial memory (MB) to allocate for each Mesos executor container.")
fs.BoolVar(&s.Checkpoint, "checkpoint", s.Checkpoint, "Enable/disable checkpointing for the kubernetes-mesos framework.") fs.BoolVar(&s.checkpoint, "checkpoint", s.checkpoint, "Enable/disable checkpointing for the kubernetes-mesos framework.")
fs.Float64Var(&s.FailoverTimeout, "failover-timeout", s.FailoverTimeout, fmt.Sprintf("Framework failover timeout, in sec.")) fs.Float64Var(&s.failoverTimeout, "failover-timeout", s.failoverTimeout, fmt.Sprintf("Framework failover timeout, in sec."))
fs.UintVar(&s.DriverPort, "driver-port", s.DriverPort, "Port that the Mesos scheduler driver process should listen on.") fs.UintVar(&s.driverPort, "driver-port", s.driverPort, "Port that the Mesos scheduler driver process should listen on.")
fs.StringVar(&s.HostnameOverride, "hostname-override", s.HostnameOverride, "If non-empty, will use this string as identification instead of the actual hostname.") fs.StringVar(&s.hostnameOverride, "hostname-override", s.hostnameOverride, "If non-empty, will use this string as identification instead of the actual hostname.")
fs.Int64Var(&s.ReconcileInterval, "reconcile-interval", s.ReconcileInterval, "Interval at which to execute task reconciliation, in sec. Zero disables.") fs.Int64Var(&s.reconcileInterval, "reconcile-interval", s.reconcileInterval, "Interval at which to execute task reconciliation, in sec. Zero disables.")
fs.DurationVar(&s.ReconcileCooldown, "reconcile-cooldown", s.ReconcileCooldown, "Minimum rest period between task reconciliation operations.") fs.DurationVar(&s.reconcileCooldown, "reconcile-cooldown", s.reconcileCooldown, "Minimum rest period between task reconciliation operations.")
fs.StringVar(&s.SchedulerConfigFileName, "scheduler-config", s.SchedulerConfigFileName, "An ini-style configuration file with low-level scheduler settings.") fs.StringVar(&s.schedulerConfigFileName, "scheduler-config", s.schedulerConfigFileName, "An ini-style configuration file with low-level scheduler settings.")
fs.BoolVar(&s.Graceful, "graceful", s.Graceful, "Indicator of a graceful failover, intended for internal use only.") fs.BoolVar(&s.graceful, "graceful", s.graceful, "Indicator of a graceful failover, intended for internal use only.")
fs.BoolVar(&s.HA, "ha", s.HA, "Run the scheduler in high availability mode with leader election. All peers should be configured exactly the same.") fs.BoolVar(&s.ha, "ha", s.ha, "Run the scheduler in high availability mode with leader election. All peers should be configured exactly the same.")
fs.StringVar(&s.FrameworkName, "framework-name", s.FrameworkName, "The framework name to register with Mesos.") fs.StringVar(&s.frameworkName, "framework-name", s.frameworkName, "The framework name to register with Mesos.")
fs.StringVar(&s.FrameworkWebURI, "framework-weburi", s.FrameworkWebURI, "A URI that points to a web-based interface for interacting with the framework.") fs.StringVar(&s.frameworkWebURI, "framework-weburi", s.frameworkWebURI, "A URI that points to a web-based interface for interacting with the framework.")
fs.StringVar(&s.AdvertisedAddress, "advertised-address", s.AdvertisedAddress, "host:port address that is advertised to clients. May be used to construct artifact download URIs.") fs.StringVar(&s.advertisedAddress, "advertised-address", s.advertisedAddress, "host:port address that is advertised to clients. May be used to construct artifact download URIs.")
fs.IPVar(&s.ServiceAddress, "service-address", s.ServiceAddress, "The service portal IP address that the scheduler should register with (if unset, chooses randomly)") fs.IPVar(&s.serviceAddress, "service-address", s.serviceAddress, "The service portal IP address that the scheduler should register with (if unset, chooses randomly)")
fs.Var(&s.DefaultContainerCPULimit, "default-container-cpu-limit", "Containers without a CPU resource limit are admitted this much CPU shares") fs.Var(&s.defaultContainerCPULimit, "default-container-cpu-limit", "Containers without a CPU resource limit are admitted this much CPU shares")
fs.Var(&s.DefaultContainerMemLimit, "default-container-mem-limit", "Containers without a memory resource limit are admitted this much amount of memory in MB") fs.Var(&s.defaultContainerMemLimit, "default-container-mem-limit", "Containers without a memory resource limit are admitted this much amount of memory in MB")
fs.BoolVar(&s.ContainPodResources, "contain-pod-resources", s.ContainPodResources, "Reparent pod containers into mesos cgroups; disable if you're having strange mesos/docker/systemd interactions.") fs.BoolVar(&s.containPodResources, "contain-pod-resources", s.containPodResources, "Reparent pod containers into mesos cgroups; disable if you're having strange mesos/docker/systemd interactions.")
fs.BoolVar(&s.AccountForPodResources, "account-for-pod-resources", s.AccountForPodResources, "Allocate pod CPU and memory resources from offers (Default: true)") fs.BoolVar(&s.accountForPodResources, "account-for-pod-resources", s.accountForPodResources, "Allocate pod CPU and memory resources from offers (Default: true)")
fs.DurationVar(&s.nodeRelistPeriod, "node-monitor-period", s.nodeRelistPeriod, "Period between relisting of all nodes from the apiserver.") fs.DurationVar(&s.nodeRelistPeriod, "node-monitor-period", s.nodeRelistPeriod, "Period between relisting of all nodes from the apiserver.")
fs.IntVar(&s.ExecutorLogV, "executor-logv", s.ExecutorLogV, "Logging verbosity of spawned minion and executor processes.") fs.IntVar(&s.executorLogV, "executor-logv", s.executorLogV, "Logging verbosity of spawned minion and executor processes.")
fs.BoolVar(&s.ExecutorBindall, "executor-bindall", s.ExecutorBindall, "When true will set -address of the executor to 0.0.0.0.") fs.BoolVar(&s.executorBindall, "executor-bindall", s.executorBindall, "When true will set -address of the executor to 0.0.0.0.")
fs.DurationVar(&s.ExecutorSuicideTimeout, "executor-suicide-timeout", s.ExecutorSuicideTimeout, "Executor self-terminates after this period of inactivity. Zero disables suicide watch.") fs.DurationVar(&s.executorSuicideTimeout, "executor-suicide-timeout", s.executorSuicideTimeout, "Executor self-terminates after this period of inactivity. Zero disables suicide watch.")
fs.DurationVar(&s.LaunchGracePeriod, "mesos-launch-grace-period", s.LaunchGracePeriod, "Launch grace period after which launching tasks will be cancelled. Zero disables launch cancellation.") fs.DurationVar(&s.launchGracePeriod, "mesos-launch-grace-period", s.launchGracePeriod, "Launch grace period after which launching tasks will be cancelled. Zero disables launch cancellation.")
fs.StringVar(&s.SandboxOverlay, "mesos-sandbox-overlay", s.SandboxOverlay, "Path to an archive (tar.gz, tar.bz2 or zip) extracted into the sandbox.") fs.StringVar(&s.sandboxOverlay, "mesos-sandbox-overlay", s.sandboxOverlay, "Path to an archive (tar.gz, tar.bz2 or zip) extracted into the sandbox.")
fs.BoolVar(&s.ProxyBindall, "proxy-bindall", s.ProxyBindall, "When true pass -proxy-bindall to the executor.") fs.BoolVar(&s.proxyBindall, "proxy-bindall", s.proxyBindall, "When true pass -proxy-bindall to the executor.")
fs.BoolVar(&s.RunProxy, "run-proxy", s.RunProxy, "Run the kube-proxy as a side process of the executor.") fs.BoolVar(&s.runProxy, "run-proxy", s.runProxy, "Run the kube-proxy as a side process of the executor.")
fs.IntVar(&s.ProxyLogV, "proxy-logv", s.ProxyLogV, "Logging verbosity of spawned minion proxy processes.") fs.IntVar(&s.proxyLogV, "proxy-logv", s.proxyLogV, "Logging verbosity of spawned minion proxy processes.")
fs.StringVar(&s.MinionPathOverride, "minion-path-override", s.MinionPathOverride, "Override the PATH in the environment of the minion sub-processes.") fs.StringVar(&s.minionPathOverride, "minion-path-override", s.minionPathOverride, "Override the PATH in the environment of the minion sub-processes.")
fs.Var(resource.NewQuantityFlagValue(&s.MinionLogMaxSize), "minion-max-log-size", "Maximum log file size for the executor and proxy before rotation") fs.Var(resource.NewQuantityFlagValue(&s.minionLogMaxSize), "minion-max-log-size", "Maximum log file size for the executor and proxy before rotation")
fs.IntVar(&s.MinionLogMaxAgeInDays, "minion-max-log-age", s.MinionLogMaxAgeInDays, "Maximum log file age of the executor and proxy in days") fs.IntVar(&s.minionLogMaxAgeInDays, "minion-max-log-age", s.minionLogMaxAgeInDays, "Maximum log file age of the executor and proxy in days")
fs.IntVar(&s.MinionLogMaxBackups, "minion-max-log-backups", s.MinionLogMaxBackups, "Maximum log file backups of the executor and proxy to keep after rotation") fs.IntVar(&s.minionLogMaxBackups, "minion-max-log-backups", s.minionLogMaxBackups, "Maximum log file backups of the executor and proxy to keep after rotation")
fs.StringVar(&s.KubeletRootDirectory, "kubelet-root-dir", s.KubeletRootDirectory, "Directory path for managing kubelet files (volume mounts,etc). Defaults to executor sandbox.") fs.StringVar(&s.kubeletRootDirectory, "kubelet-root-dir", s.kubeletRootDirectory, "Directory path for managing kubelet files (volume mounts,etc). Defaults to executor sandbox.")
fs.StringVar(&s.KubeletDockerEndpoint, "kubelet-docker-endpoint", s.KubeletDockerEndpoint, "If non-empty, kubelet will use this for the docker endpoint to communicate with.") fs.StringVar(&s.kubeletDockerEndpoint, "kubelet-docker-endpoint", s.kubeletDockerEndpoint, "If non-empty, kubelet will use this for the docker endpoint to communicate with.")
fs.StringVar(&s.KubeletPodInfraContainerImage, "kubelet-pod-infra-container-image", s.KubeletPodInfraContainerImage, "The image whose network/ipc namespaces containers in each pod will use.") fs.StringVar(&s.kubeletPodInfraContainerImage, "kubelet-pod-infra-container-image", s.kubeletPodInfraContainerImage, "The image whose network/ipc namespaces containers in each pod will use.")
fs.UintVar(&s.KubeletCadvisorPort, "kubelet-cadvisor-port", s.KubeletCadvisorPort, "The port of the kubelet's local cAdvisor endpoint") fs.UintVar(&s.kubeletCadvisorPort, "kubelet-cadvisor-port", s.kubeletCadvisorPort, "The port of the kubelet's local cAdvisor endpoint")
fs.StringVar(&s.KubeletHostNetworkSources, "kubelet-host-network-sources", s.KubeletHostNetworkSources, "Comma-separated list of sources from which the Kubelet allows pods to use of host network. For all sources use \"*\" [default=\"file\"]") fs.StringVar(&s.kubeletHostNetworkSources, "kubelet-host-network-sources", s.kubeletHostNetworkSources, "Comma-separated list of sources from which the Kubelet allows pods to use of host network. For all sources use \"*\" [default=\"file\"]")
fs.DurationVar(&s.KubeletSyncFrequency, "kubelet-sync-frequency", s.KubeletSyncFrequency, "Max period between synchronizing running containers and config") fs.DurationVar(&s.kubeletSyncFrequency, "kubelet-sync-frequency", s.kubeletSyncFrequency, "Max period between synchronizing running containers and config")
fs.StringVar(&s.KubeletNetworkPluginName, "kubelet-network-plugin", s.KubeletNetworkPluginName, "<Warning: Alpha feature> The name of the network plugin to be invoked for various events in kubelet/pod lifecycle") fs.StringVar(&s.kubeletNetworkPluginName, "kubelet-network-plugin", s.kubeletNetworkPluginName, "<Warning: Alpha feature> The name of the network plugin to be invoked for various events in kubelet/pod lifecycle")
//TODO(jdef) support this flag once we have a better handle on mesos-dns and k8s DNS integration //TODO(jdef) support this flag once we have a better handle on mesos-dns and k8s DNS integration
//fs.StringVar(&s.HADomain, "ha-domain", s.HADomain, "Domain of the HA scheduler service, only used in HA mode. If specified may be used to construct artifact download URIs.") //fs.StringVar(&s.HADomain, "ha-domain", s.HADomain, "Domain of the HA scheduler service, only used in HA mode. If specified may be used to construct artifact download URIs.")
@ -286,12 +286,12 @@ func (s *SchedulerServer) addCoreFlags(fs *pflag.FlagSet) {
func (s *SchedulerServer) AddStandaloneFlags(fs *pflag.FlagSet) { func (s *SchedulerServer) AddStandaloneFlags(fs *pflag.FlagSet) {
s.addCoreFlags(fs) s.addCoreFlags(fs)
fs.StringVar(&s.ExecutorPath, "executor-path", s.ExecutorPath, "Location of the kubernetes executor executable") fs.StringVar(&s.executorPath, "executor-path", s.executorPath, "Location of the kubernetes executor executable")
} }
func (s *SchedulerServer) AddHyperkubeFlags(fs *pflag.FlagSet) { func (s *SchedulerServer) AddHyperkubeFlags(fs *pflag.FlagSet) {
s.addCoreFlags(fs) s.addCoreFlags(fs)
fs.StringVar(&s.KMPath, "km-path", s.KMPath, "Location of the km executable, may be a URI or an absolute file path.") fs.StringVar(&s.kmPath, "km-path", s.kmPath, "Location of the km executable, may be a URI or an absolute file path.")
} }
// returns (downloadURI, basename(path)) // returns (downloadURI, basename(path))
@ -311,12 +311,12 @@ func (s *SchedulerServer) serveFrameworkArtifactWithFilename(path string, filena
serveFile("/"+filename, path) serveFile("/"+filename, path)
hostURI := "" hostURI := ""
if s.AdvertisedAddress != "" { if s.advertisedAddress != "" {
hostURI = fmt.Sprintf("http://%s/%s", s.AdvertisedAddress, filename) hostURI = fmt.Sprintf("http://%s/%s", s.advertisedAddress, filename)
} else if s.HA && s.HADomain != "" { } else if s.ha && s.haDomain != "" {
hostURI = fmt.Sprintf("http://%s.%s:%d/%s", SCHEDULER_SERVICE_NAME, s.HADomain, ports.SchedulerPort, filename) hostURI = fmt.Sprintf("http://%s.%s:%d/%s", SCHEDULER_SERVICE_NAME, s.haDomain, ports.SchedulerPort, filename)
} else { } else {
hostURI = fmt.Sprintf("http://%s:%d/%s", s.Address.String(), s.Port, filename) hostURI = fmt.Sprintf("http://%s:%d/%s", s.address.String(), s.port, filename)
} }
log.V(2).Infof("Hosting artifact '%s' at '%s'", filename, hostURI) log.V(2).Infof("Hosting artifact '%s' at '%s'", filename, hostURI)
@ -328,21 +328,21 @@ func (s *SchedulerServer) prepareExecutorInfo(hks hyperkube.Interface) (*mesos.E
Shell: proto.Bool(false), Shell: proto.Bool(false),
} }
if s.ExecutorPath != "" { if s.executorPath != "" {
uri, executorCmd := s.serveFrameworkArtifact(s.ExecutorPath) uri, executorCmd := s.serveFrameworkArtifact(s.executorPath)
ci.Uris = append(ci.Uris, &mesos.CommandInfo_URI{Value: proto.String(uri), Executable: proto.Bool(true)}) ci.Uris = append(ci.Uris, &mesos.CommandInfo_URI{Value: proto.String(uri), Executable: proto.Bool(true)})
ci.Value = proto.String(fmt.Sprintf("./%s", executorCmd)) ci.Value = proto.String(fmt.Sprintf("./%s", executorCmd))
} else if !hks.FindServer(hyperkube.CommandMinion) { } else if !hks.FindServer(hyperkube.CommandMinion) {
return nil, nil, fmt.Errorf("either run this scheduler via km or else --executor-path is required") return nil, nil, fmt.Errorf("either run this scheduler via km or else --executor-path is required")
} else { } else {
if strings.Index(s.KMPath, "://") > 0 { if strings.Index(s.kmPath, "://") > 0 {
// URI could point directly to executable, e.g. hdfs:///km // URI could point directly to executable, e.g. hdfs:///km
// or else indirectly, e.g. http://acmestorage/tarball.tgz // or else indirectly, e.g. http://acmestorage/tarball.tgz
// so we assume that for this case the command will always "km" // so we assume that for this case the command will always "km"
ci.Uris = append(ci.Uris, &mesos.CommandInfo_URI{Value: proto.String(s.KMPath), Executable: proto.Bool(true)}) ci.Uris = append(ci.Uris, &mesos.CommandInfo_URI{Value: proto.String(s.kmPath), Executable: proto.Bool(true)})
ci.Value = proto.String("./km") // TODO(jdef) extract constant ci.Value = proto.String("./km") // TODO(jdef) extract constant
} else if s.KMPath != "" { } else if s.kmPath != "" {
uri, kmCmd := s.serveFrameworkArtifact(s.KMPath) uri, kmCmd := s.serveFrameworkArtifact(s.kmPath)
ci.Uris = append(ci.Uris, &mesos.CommandInfo_URI{Value: proto.String(uri), Executable: proto.Bool(true)}) ci.Uris = append(ci.Uris, &mesos.CommandInfo_URI{Value: proto.String(uri), Executable: proto.Bool(true)})
ci.Value = proto.String(fmt.Sprintf("./%s", kmCmd)) ci.Value = proto.String(fmt.Sprintf("./%s", kmCmd))
} else { } else {
@ -352,55 +352,55 @@ func (s *SchedulerServer) prepareExecutorInfo(hks hyperkube.Interface) (*mesos.E
} }
ci.Arguments = append(ci.Arguments, hyperkube.CommandMinion) ci.Arguments = append(ci.Arguments, hyperkube.CommandMinion)
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--run-proxy=%v", s.RunProxy)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--run-proxy=%v", s.runProxy))
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--proxy-bindall=%v", s.ProxyBindall)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--proxy-bindall=%v", s.proxyBindall))
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--proxy-logv=%d", s.ProxyLogV)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--proxy-logv=%d", s.proxyLogV))
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--path-override=%s", s.MinionPathOverride)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--path-override=%s", s.minionPathOverride))
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--max-log-size=%v", s.MinionLogMaxSize.String())) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--max-log-size=%v", s.minionLogMaxSize.String()))
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--max-log-backups=%d", s.MinionLogMaxBackups)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--max-log-backups=%d", s.minionLogMaxBackups))
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--max-log-age=%d", s.MinionLogMaxAgeInDays)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--max-log-age=%d", s.minionLogMaxAgeInDays))
} }
if s.SandboxOverlay != "" { if s.sandboxOverlay != "" {
if _, err := os.Stat(s.SandboxOverlay); os.IsNotExist(err) { if _, err := os.Stat(s.sandboxOverlay); os.IsNotExist(err) {
log.Fatalf("Sandbox overlay archive not found: %s", s.SandboxOverlay) log.Fatalf("Sandbox overlay archive not found: %s", s.sandboxOverlay)
} }
uri, _ := s.serveFrameworkArtifact(s.SandboxOverlay) uri, _ := s.serveFrameworkArtifact(s.sandboxOverlay)
ci.Uris = append(ci.Uris, &mesos.CommandInfo_URI{Value: proto.String(uri), Executable: proto.Bool(false), Extract: proto.Bool(true)}) ci.Uris = append(ci.Uris, &mesos.CommandInfo_URI{Value: proto.String(uri), Executable: proto.Bool(false), Extract: proto.Bool(true)})
} }
if s.DockerCfgPath != "" { if s.dockerCfgPath != "" {
uri := s.serveFrameworkArtifactWithFilename(s.DockerCfgPath, ".dockercfg") uri := s.serveFrameworkArtifactWithFilename(s.dockerCfgPath, ".dockercfg")
ci.Uris = append(ci.Uris, &mesos.CommandInfo_URI{Value: proto.String(uri), Executable: proto.Bool(false), Extract: proto.Bool(false)}) ci.Uris = append(ci.Uris, &mesos.CommandInfo_URI{Value: proto.String(uri), Executable: proto.Bool(false), Extract: proto.Bool(false)})
} }
//TODO(jdef): provide some way (env var?) for users to customize executor config //TODO(jdef): provide some way (env var?) for users to customize executor config
//TODO(jdef): set -address to 127.0.0.1 if `address` is 127.0.0.1 //TODO(jdef): set -address to 127.0.0.1 if `address` is 127.0.0.1
apiServerArgs := strings.Join(s.APIServerList, ",") apiServerArgs := strings.Join(s.apiServerList, ",")
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--api-servers=%s", apiServerArgs)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--api-servers=%s", apiServerArgs))
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--v=%d", s.ExecutorLogV)) // this also applies to the minion ci.Arguments = append(ci.Arguments, fmt.Sprintf("--v=%d", s.executorLogV)) // this also applies to the minion
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--allow-privileged=%t", s.AllowPrivileged)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--allow-privileged=%t", s.allowPrivileged))
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--suicide-timeout=%v", s.ExecutorSuicideTimeout)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--suicide-timeout=%v", s.executorSuicideTimeout))
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--mesos-launch-grace-period=%v", s.LaunchGracePeriod)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--mesos-launch-grace-period=%v", s.launchGracePeriod))
if s.ExecutorBindall { if s.executorBindall {
//TODO(jdef) determine whether hostname-override is really needed for bindall because //TODO(jdef) determine whether hostname-override is really needed for bindall because
//it conflicts with kubelet node status checks/updates //it conflicts with kubelet node status checks/updates
//ci.Arguments = append(ci.Arguments, "--hostname-override=0.0.0.0") //ci.Arguments = append(ci.Arguments, "--hostname-override=0.0.0.0")
ci.Arguments = append(ci.Arguments, "--address=0.0.0.0") ci.Arguments = append(ci.Arguments, "--address=0.0.0.0")
} }
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--mesos-cgroup-prefix=%v", s.MesosCgroupPrefix)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--mesos-cgroup-prefix=%v", s.mesosCgroupPrefix))
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--cadvisor-port=%v", s.KubeletCadvisorPort)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--cadvisor-port=%v", s.kubeletCadvisorPort))
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--sync-frequency=%v", s.KubeletSyncFrequency)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--sync-frequency=%v", s.kubeletSyncFrequency))
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--contain-pod-resources=%t", s.ContainPodResources)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--contain-pod-resources=%t", s.containPodResources))
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--enable-debugging-handlers=%t", s.EnableProfiling)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--enable-debugging-handlers=%t", s.enableProfiling))
if s.AuthPath != "" { if s.authPath != "" {
//TODO(jdef) should probably support non-local files, e.g. hdfs:///some/config/file //TODO(jdef) should probably support non-local files, e.g. hdfs:///some/config/file
uri, basename := s.serveFrameworkArtifact(s.AuthPath) uri, basename := s.serveFrameworkArtifact(s.authPath)
ci.Uris = append(ci.Uris, &mesos.CommandInfo_URI{Value: proto.String(uri)}) ci.Uris = append(ci.Uris, &mesos.CommandInfo_URI{Value: proto.String(uri)})
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--auth-path=%s", basename)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--auth-path=%s", basename))
} }
@ -409,15 +409,15 @@ func (s *SchedulerServer) prepareExecutorInfo(hks hyperkube.Interface) (*mesos.E
ci.Arguments = append(ci.Arguments, fmt.Sprintf("--%s=%s", name, value)) ci.Arguments = append(ci.Arguments, fmt.Sprintf("--%s=%s", name, value))
} }
} }
if s.ClusterDNS != nil { if s.clusterDNS != nil {
appendOptional("cluster-dns", s.ClusterDNS.String()) appendOptional("cluster-dns", s.clusterDNS.String())
} }
appendOptional("cluster-domain", s.ClusterDomain) appendOptional("cluster-domain", s.clusterDomain)
appendOptional("root-dir", s.KubeletRootDirectory) appendOptional("root-dir", s.kubeletRootDirectory)
appendOptional("docker-endpoint", s.KubeletDockerEndpoint) appendOptional("docker-endpoint", s.kubeletDockerEndpoint)
appendOptional("pod-infra-container-image", s.KubeletPodInfraContainerImage) appendOptional("pod-infra-container-image", s.kubeletPodInfraContainerImage)
appendOptional("host-network-sources", s.KubeletHostNetworkSources) appendOptional("host-network-sources", s.kubeletHostNetworkSources)
appendOptional("network-plugin", s.KubeletNetworkPluginName) appendOptional("network-plugin", s.kubeletNetworkPluginName)
log.V(1).Infof("prepared executor command %q with args '%+v'", ci.GetValue(), ci.Arguments) log.V(1).Infof("prepared executor command %q with args '%+v'", ci.GetValue(), ci.Arguments)
@ -430,8 +430,8 @@ func (s *SchedulerServer) prepareExecutorInfo(hks hyperkube.Interface) (*mesos.E
// Check for staticPods // Check for staticPods
var staticPodCPUs, staticPodMem float64 var staticPodCPUs, staticPodMem float64
if s.StaticPodsConfigPath != "" { if s.staticPodsConfigPath != "" {
bs, paths, err := archive.ZipDir(s.StaticPodsConfigPath) bs, paths, err := archive.ZipDir(s.staticPodsConfigPath)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@ -452,8 +452,8 @@ func (s *SchedulerServer) prepareExecutorInfo(hks hyperkube.Interface) (*mesos.E
} }
// TODO(sttts): allow unlimited static pods as well and patch in the default resource limits // TODO(sttts): allow unlimited static pods as well and patch in the default resource limits
unlimitedCPU := mresource.LimitPodCPU(&pod, s.DefaultContainerCPULimit) unlimitedCPU := mresource.LimitPodCPU(&pod, s.defaultContainerCPULimit)
unlimitedMem := mresource.LimitPodMem(&pod, s.DefaultContainerMemLimit) unlimitedMem := mresource.LimitPodMem(&pod, s.defaultContainerMemLimit)
if unlimitedCPU { if unlimitedCPU {
return nil, nil, fmt.Errorf("found static pod without limit on cpu resources: %v", podPath) return nil, nil, fmt.Errorf("found static pod without limit on cpu resources: %v", podPath)
} }
@ -474,8 +474,8 @@ func (s *SchedulerServer) prepareExecutorInfo(hks hyperkube.Interface) (*mesos.E
} }
execInfo.Resources = []*mesos.Resource{ execInfo.Resources = []*mesos.Resource{
mutil.NewScalarResource("cpus", float64(s.MesosExecutorCPUs)+staticPodCPUs), mutil.NewScalarResource("cpus", float64(s.mesosExecutorCPUs)+staticPodCPUs),
mutil.NewScalarResource("mem", float64(s.MesosExecutorMem)+staticPodMem), mutil.NewScalarResource("mem", float64(s.mesosExecutorMem)+staticPodMem),
} }
// calculate ExecutorInfo hash to be used for validating compatibility // calculate ExecutorInfo hash to be used for validating compatibility
@ -490,7 +490,7 @@ func (s *SchedulerServer) prepareExecutorInfo(hks hyperkube.Interface) (*mesos.E
// TODO(jdef): hacked from kubelet/server/server.go // TODO(jdef): hacked from kubelet/server/server.go
// TODO(k8s): replace this with clientcmd // TODO(k8s): replace this with clientcmd
func (s *SchedulerServer) createAPIServerClient() (*client.Client, error) { func (s *SchedulerServer) createAPIServerClient() (*client.Client, error) {
authInfo, err := clientauth.LoadFromFile(s.AuthPath) authInfo, err := clientauth.LoadFromFile(s.authPath)
if err != nil { if err != nil {
log.Warningf("Could not load kubernetes auth path: %v. Continuing with defaults.", err) log.Warningf("Could not load kubernetes auth path: %v. Continuing with defaults.", err)
} }
@ -502,14 +502,14 @@ func (s *SchedulerServer) createAPIServerClient() (*client.Client, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
if len(s.APIServerList) < 1 { if len(s.apiServerList) < 1 {
return nil, fmt.Errorf("no api servers specified") return nil, fmt.Errorf("no api servers specified")
} }
// TODO: adapt Kube client to support LB over several servers // TODO: adapt Kube client to support LB over several servers
if len(s.APIServerList) > 1 { if len(s.apiServerList) > 1 {
log.Infof("Multiple api servers specified. Picking first one") log.Infof("Multiple api servers specified. Picking first one")
} }
clientConfig.Host = s.APIServerList[0] clientConfig.Host = s.apiServerList[0]
c, err := client.New(&clientConfig) c, err := client.New(&clientConfig)
if err != nil { if err != nil {
return nil, err return nil, err
@ -532,8 +532,8 @@ func (s *SchedulerServer) getDriver() (driver bindings.SchedulerDriver) {
func (s *SchedulerServer) Run(hks hyperkube.Interface, _ []string) error { func (s *SchedulerServer) Run(hks hyperkube.Interface, _ []string) error {
// get scheduler low-level config // get scheduler low-level config
sc := schedcfg.CreateDefaultConfig() sc := schedcfg.CreateDefaultConfig()
if s.SchedulerConfigFileName != "" { if s.schedulerConfigFileName != "" {
f, err := os.Open(s.SchedulerConfigFileName) f, err := os.Open(s.schedulerConfigFileName)
if err != nil { if err != nil {
log.Fatalf("Cannot open scheduler config file: %v", err) log.Fatalf("Cannot open scheduler config file: %v", err)
} }
@ -546,18 +546,18 @@ func (s *SchedulerServer) Run(hks hyperkube.Interface, _ []string) error {
schedulerProcess, driverFactory, etcdClient, eid := s.bootstrap(hks, sc) schedulerProcess, driverFactory, etcdClient, eid := s.bootstrap(hks, sc)
if s.EnableProfiling { if s.enableProfiling {
profile.InstallHandler(s.mux) profile.InstallHandler(s.mux)
} }
go runtime.Until(func() { go runtime.Until(func() {
log.V(1).Info("Starting HTTP interface") log.V(1).Info("Starting HTTP interface")
log.Error(http.ListenAndServe(net.JoinHostPort(s.Address.String(), strconv.Itoa(s.Port)), s.mux)) log.Error(http.ListenAndServe(net.JoinHostPort(s.address.String(), strconv.Itoa(s.port)), s.mux))
}, sc.HttpBindInterval.Duration, schedulerProcess.Terminal()) }, sc.HttpBindInterval.Duration, schedulerProcess.Terminal())
if s.HA { if s.ha {
validation := ha.ValidationFunc(validateLeadershipTransition) validation := ha.ValidationFunc(validateLeadershipTransition)
srv := ha.NewCandidate(schedulerProcess, driverFactory, validation) srv := ha.NewCandidate(schedulerProcess, driverFactory, validation)
path := fmt.Sprintf(meta.DefaultElectionFormat, s.FrameworkName) path := fmt.Sprintf(meta.DefaultElectionFormat, s.frameworkName)
sid := uid.New(eid.Group(), "").String() sid := uid.New(eid.Group(), "").String()
log.Infof("registering for election at %v with id %v", path, sid) log.Infof("registering for election at %v with id %v", path, sid)
go election.Notify(election.NewEtcdMasterElector(etcdClient), path, sid, srv, nil) go election.Notify(election.NewEtcdMasterElector(etcdClient), path, sid, srv, nil)
@ -596,7 +596,7 @@ func (s *SchedulerServer) awaitFailover(schedulerProcess schedulerProcessInterfa
case <-schedulerProcess.Failover(): case <-schedulerProcess.Failover():
err = doFailover() err = doFailover()
default: default:
if s.HA { if s.ha {
err = fmt.Errorf("ha scheduler exiting instead of failing over") err = fmt.Errorf("ha scheduler exiting instead of failing over")
} else { } else {
log.Infof("exiting scheduler") log.Infof("exiting scheduler")
@ -638,22 +638,22 @@ func newEtcd(etcdConfigFile string, etcdServerList []string) (client tools.EtcdC
func (s *SchedulerServer) bootstrap(hks hyperkube.Interface, sc *schedcfg.Config) (*ha.SchedulerProcess, ha.DriverFactory, tools.EtcdClient, *uid.UID) { func (s *SchedulerServer) bootstrap(hks hyperkube.Interface, sc *schedcfg.Config) (*ha.SchedulerProcess, ha.DriverFactory, tools.EtcdClient, *uid.UID) {
s.FrameworkName = strings.TrimSpace(s.FrameworkName) s.frameworkName = strings.TrimSpace(s.frameworkName)
if s.FrameworkName == "" { if s.frameworkName == "" {
log.Fatalf("framework-name must be a non-empty string") log.Fatalf("framework-name must be a non-empty string")
} }
s.FrameworkWebURI = strings.TrimSpace(s.FrameworkWebURI) s.frameworkWebURI = strings.TrimSpace(s.frameworkWebURI)
metrics.Register() metrics.Register()
runtime.Register() runtime.Register()
s.mux.Handle("/metrics", prometheus.Handler()) s.mux.Handle("/metrics", prometheus.Handler())
healthz.InstallHandler(s.mux) healthz.InstallHandler(s.mux)
if (s.EtcdConfigFile != "" && len(s.EtcdServerList) != 0) || (s.EtcdConfigFile == "" && len(s.EtcdServerList) == 0) { if (s.etcdConfigFile != "" && len(s.etcdServerList) != 0) || (s.etcdConfigFile == "" && len(s.etcdServerList) == 0) {
log.Fatalf("specify either --etcd-servers or --etcd-config") log.Fatalf("specify either --etcd-servers or --etcd-config")
} }
if len(s.APIServerList) < 1 { if len(s.apiServerList) < 1 {
log.Fatal("No api servers specified.") log.Fatal("No api servers specified.")
} }
@ -663,9 +663,9 @@ func (s *SchedulerServer) bootstrap(hks hyperkube.Interface, sc *schedcfg.Config
} }
s.client = client s.client = client
if s.ReconcileCooldown < defaultReconcileCooldown { if s.reconcileCooldown < defaultReconcileCooldown {
s.ReconcileCooldown = defaultReconcileCooldown s.reconcileCooldown = defaultReconcileCooldown
log.Warningf("user-specified reconcile cooldown too small, defaulting to %v", s.ReconcileCooldown) log.Warningf("user-specified reconcile cooldown too small, defaulting to %v", s.reconcileCooldown)
} }
executor, eid, err := s.prepareExecutorInfo(hks) executor, eid, err := s.prepareExecutorInfo(hks)
@ -677,24 +677,24 @@ func (s *SchedulerServer) bootstrap(hks hyperkube.Interface, sc *schedcfg.Config
// (1) the generic config store is available for the FrameworkId storage // (1) the generic config store is available for the FrameworkId storage
// (2) the generic master election is provided by the apiserver // (2) the generic master election is provided by the apiserver
// Compare docs/proposals/high-availability.md // Compare docs/proposals/high-availability.md
etcdClient, err := newEtcd(s.EtcdConfigFile, s.EtcdServerList) etcdClient, err := newEtcd(s.etcdConfigFile, s.etcdServerList)
if err != nil { if err != nil {
log.Fatalf("misconfigured etcd: %v", err) log.Fatalf("misconfigured etcd: %v", err)
} }
as := podschedulers.NewAllocationStrategy( as := podschedulers.NewAllocationStrategy(
podtask.NewDefaultPredicate( podtask.NewDefaultPredicate(
s.DefaultContainerCPULimit, s.defaultContainerCPULimit,
s.DefaultContainerMemLimit, s.defaultContainerMemLimit,
), ),
podtask.NewDefaultProcurement( podtask.NewDefaultProcurement(
s.DefaultContainerCPULimit, s.defaultContainerCPULimit,
s.DefaultContainerMemLimit, s.defaultContainerMemLimit,
), ),
) )
// downgrade allocation strategy if user disables "account-for-pod-resources" // downgrade allocation strategy if user disables "account-for-pod-resources"
if !s.AccountForPodResources { if !s.accountForPodResources {
as = podschedulers.NewAllocationStrategy( as = podschedulers.NewAllocationStrategy(
podtask.DefaultMinimalPredicate, podtask.DefaultMinimalPredicate,
podtask.DefaultMinimalProcurement) podtask.DefaultMinimalProcurement)
@ -723,20 +723,20 @@ func (s *SchedulerServer) bootstrap(hks hyperkube.Interface, sc *schedcfg.Config
Executor: executor, Executor: executor,
PodScheduler: fcfs, PodScheduler: fcfs,
Client: client, Client: client,
FailoverTimeout: s.FailoverTimeout, FailoverTimeout: s.failoverTimeout,
ReconcileInterval: s.ReconcileInterval, ReconcileInterval: s.reconcileInterval,
ReconcileCooldown: s.ReconcileCooldown, ReconcileCooldown: s.reconcileCooldown,
LookupNode: lookupNode, LookupNode: lookupNode,
StoreFrameworkId: func(id string) { StoreFrameworkId: func(id string) {
// TODO(jdef): port FrameworkId store to generic Kubernetes config store as soon as available // TODO(jdef): port FrameworkId store to generic Kubernetes config store as soon as available
_, err := etcdClient.Set(meta.FrameworkIDKey, id, uint64(s.FailoverTimeout)) _, err := etcdClient.Set(meta.FrameworkIDKey, id, uint64(s.failoverTimeout))
if err != nil { if err != nil {
log.Errorf("failed to renew frameworkId TTL: %v", err) log.Errorf("failed to renew frameworkId TTL: %v", err)
} }
}, },
}) })
masterUri := s.MesosMaster masterUri := s.mesosMaster
info, cred, err := s.buildFrameworkInfo() info, cred, err := s.buildFrameworkInfo()
if err != nil { if err != nil {
log.Fatalf("Misconfigured mesos framework: %v", err) log.Fatalf("Misconfigured mesos framework: %v", err)
@ -748,12 +748,12 @@ func (s *SchedulerServer) bootstrap(hks hyperkube.Interface, sc *schedcfg.Config
Framework: info, Framework: info,
Master: masterUri, Master: masterUri,
Credential: cred, Credential: cred,
BindingAddress: s.Address, BindingAddress: s.address,
BindingPort: uint16(s.DriverPort), BindingPort: uint16(s.driverPort),
HostnameOverride: s.HostnameOverride, HostnameOverride: s.hostnameOverride,
WithAuthContext: func(ctx context.Context) context.Context { WithAuthContext: func(ctx context.Context) context.Context {
ctx = auth.WithLoginProvider(ctx, s.MesosAuthProvider) ctx = auth.WithLoginProvider(ctx, s.mesosAuthProvider)
ctx = sasl.WithBindingAddress(ctx, s.Address) ctx = sasl.WithBindingAddress(ctx, s.address)
return ctx return ctx
}, },
} }
@ -813,14 +813,14 @@ func (s *SchedulerServer) failover(driver bindings.SchedulerDriver, hks hyperkub
args = append(args, fmt.Sprintf("--%s=%s", flag.Name, flag.Value.String())) args = append(args, fmt.Sprintf("--%s=%s", flag.Name, flag.Value.String()))
} }
}) })
if !s.Graceful { if !s.graceful {
args = append(args, "--graceful") args = append(args, "--graceful")
} }
if len(s.APIServerList) > 0 { if len(s.apiServerList) > 0 {
args = append(args, "--api-servers="+strings.Join(s.APIServerList, ",")) args = append(args, "--api-servers="+strings.Join(s.apiServerList, ","))
} }
if len(s.EtcdServerList) > 0 { if len(s.etcdServerList) > 0 {
args = append(args, "--etcd-servers="+strings.Join(s.EtcdServerList, ",")) args = append(args, "--etcd-servers="+strings.Join(s.etcdServerList, ","))
} }
args = append(args, flags.Args()...) args = append(args, flags.Args()...)
@ -853,30 +853,30 @@ func (s *SchedulerServer) buildFrameworkInfo() (info *mesos.FrameworkInfo, cred
} }
log.V(2).Infof("Framework configured with mesos user %v", username) log.V(2).Infof("Framework configured with mesos user %v", username)
info = &mesos.FrameworkInfo{ info = &mesos.FrameworkInfo{
Name: proto.String(s.FrameworkName), Name: proto.String(s.frameworkName),
User: proto.String(username), User: proto.String(username),
Checkpoint: proto.Bool(s.Checkpoint), Checkpoint: proto.Bool(s.checkpoint),
} }
if s.FrameworkWebURI != "" { if s.frameworkWebURI != "" {
info.WebuiUrl = proto.String(s.FrameworkWebURI) info.WebuiUrl = proto.String(s.frameworkWebURI)
} }
if s.FailoverTimeout > 0 { if s.failoverTimeout > 0 {
info.FailoverTimeout = proto.Float64(s.FailoverTimeout) info.FailoverTimeout = proto.Float64(s.failoverTimeout)
} }
if s.MesosRole != "" { if s.mesosRole != "" {
info.Role = proto.String(s.MesosRole) info.Role = proto.String(s.mesosRole)
} }
if s.MesosAuthPrincipal != "" { if s.mesosAuthPrincipal != "" {
info.Principal = proto.String(s.MesosAuthPrincipal) info.Principal = proto.String(s.mesosAuthPrincipal)
if s.MesosAuthSecretFile == "" { if s.mesosAuthSecretFile == "" {
return nil, nil, errors.New("authentication principal specified without the required credentials file") return nil, nil, errors.New("authentication principal specified without the required credentials file")
} }
secret, err := ioutil.ReadFile(s.MesosAuthSecretFile) secret, err := ioutil.ReadFile(s.mesosAuthSecretFile)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
cred = &mesos.Credential{ cred = &mesos.Credential{
Principal: proto.String(s.MesosAuthPrincipal), Principal: proto.String(s.mesosAuthPrincipal),
Secret: secret, Secret: secret,
} }
} }
@ -884,7 +884,7 @@ func (s *SchedulerServer) buildFrameworkInfo() (info *mesos.FrameworkInfo, cred
} }
func (s *SchedulerServer) fetchFrameworkID(client tools.EtcdClient) (*mesos.FrameworkID, error) { func (s *SchedulerServer) fetchFrameworkID(client tools.EtcdClient) (*mesos.FrameworkID, error) {
if s.FailoverTimeout > 0 { if s.failoverTimeout > 0 {
if response, err := client.Get(meta.FrameworkIDKey, false, false); err != nil { if response, err := client.Get(meta.FrameworkIDKey, false, false); err != nil {
if !etcdstorage.IsEtcdNotFound(err) { if !etcdstorage.IsEtcdNotFound(err) {
return nil, fmt.Errorf("unexpected failure attempting to load framework ID from etcd: %v", err) return nil, fmt.Errorf("unexpected failure attempting to load framework ID from etcd: %v", err)
@ -907,7 +907,7 @@ func (s *SchedulerServer) fetchFrameworkID(client tools.EtcdClient) (*mesos.Fram
} }
func (s *SchedulerServer) getUsername() (username string, err error) { func (s *SchedulerServer) getUsername() (username string, err error) {
username = s.MesosUser username = s.mesosUser
if username == "" { if username == "" {
if u, err := user.Current(); err == nil { if u, err := user.Current(); err == nil {
username = u.Username username = u.Username

4
contrib/mesos/pkg/scheduler/service/service_test.go
View File

@ -121,8 +121,8 @@ func Test_DefaultResourceLimits(t *testing.T) {
assert := assert.New(t) assert := assert.New(t)
s := NewSchedulerServer() s := NewSchedulerServer()
assert.Equal(s.DefaultContainerCPULimit, mresource.DefaultDefaultContainerCPULimit) assert.Equal(s.defaultContainerCPULimit, mresource.DefaultDefaultContainerCPULimit)
assert.Equal(s.DefaultContainerMemLimit, mresource.DefaultDefaultContainerMemLimit) assert.Equal(s.defaultContainerMemLimit, mresource.DefaultDefaultContainerMemLimit)
} }
func Test_StaticPods(t *testing.T) { func Test_StaticPods(t *testing.T) {