mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-07-29 22:46:12 +00:00
Merge pull request #4031 from jbeda/hyperkube-kubelet
Convert kubelet and proxy to hyperkube
This commit is contained in:
Brian Grant
commit
2903fce2b8
@ -23,7 +23,9 @@ import (
|
||||
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/controllermanager"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/hyperkube"
|
||||
kubelet "github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/server"
|
||||
apiserver "github.com/GoogleCloudPlatform/kubernetes/pkg/master/server"
|
||||
proxy "github.com/GoogleCloudPlatform/kubernetes/pkg/proxy/server"
|
||||
sched "github.com/GoogleCloudPlatform/kubernetes/plugin/pkg/scheduler/server"
|
||||
)
|
||||
|
||||
@ -36,6 +38,8 @@ func main() {
|
||||
hk.AddServer(apiserver.NewHyperkubeServer())
|
||||
hk.AddServer(controllermanager.NewHyperkubeServer())
|
||||
hk.AddServer(sched.NewHyperkubeServer())
|
||||
hk.AddServer(kubelet.NewHyperkubeServer())
|
||||
hk.AddServer(proxy.NewHyperkubeServer())
|
||||
|
||||
hk.RunToExit(os.Args)
|
||||
}
|
||||
|
||||
@ -40,16 +40,17 @@ import (
|
||||
nodeControllerPkg "github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider/controller"
|
||||
replicationControllerPkg "github.com/GoogleCloudPlatform/kubernetes/pkg/controller"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/dockertools"
|
||||
kubeletServer "github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/server"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume/empty_dir"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/master"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/probe"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/service"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/standalone"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/wait"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/plugin/pkg/admission/admit"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/plugin/pkg/scheduler"
|
||||
_ "github.com/GoogleCloudPlatform/kubernetes/plugin/pkg/scheduler/algorithmprovider"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/plugin/pkg/scheduler/factory"
|
||||
|
||||
"github.com/coreos/go-etcd/etcd"
|
||||
@ -202,13 +203,13 @@ func startComponents(manifestURL string) (apiServerURL string) {
|
||||
// Kubelet (localhost)
|
||||
testRootDir := makeTempDirOrDie("kubelet_integ_1.")
|
||||
glog.Infof("Using %s as root dir for kubelet #1", testRootDir)
|
||||
standalone.SimpleRunKubelet(cl, nil, &fakeDocker1, machineList[0], testRootDir, manifestURL, "127.0.0.1", 10250, api.NamespaceDefault, empty_dir.ProbeVolumePlugins())
|
||||
kubeletServer.SimpleRunKubelet(cl, nil, &fakeDocker1, machineList[0], testRootDir, manifestURL, "127.0.0.1", 10250, api.NamespaceDefault, empty_dir.ProbeVolumePlugins())
|
||||
// Kubelet (machine)
|
||||
// Create a second kubelet so that the guestbook example's two redis slaves both
|
||||
// have a place they can schedule.
|
||||
testRootDir = makeTempDirOrDie("kubelet_integ_2.")
|
||||
glog.Infof("Using %s as root dir for kubelet #2", testRootDir)
|
||||
standalone.SimpleRunKubelet(cl, nil, &fakeDocker2, machineList[1], testRootDir, "", "127.0.0.1", 10251, api.NamespaceDefault, empty_dir.ProbeVolumePlugins())
|
||||
kubeletServer.SimpleRunKubelet(cl, nil, &fakeDocker2, machineList[1], testRootDir, "", "127.0.0.1", 10251, api.NamespaceDefault, empty_dir.ProbeVolumePlugins())
|
||||
|
||||
return apiServer.URL
|
||||
}
|
||||
|
||||
@ -17,126 +17,22 @@ limitations under the License.
|
||||
package main
|
||||
|
||||
import (
|
||||
"net"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
|
||||
_ "github.com/GoogleCloudPlatform/kubernetes/pkg/healthz"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/proxy"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/proxy/config"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/proxy/server"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/exec"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/iptables"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/version/verflag"
|
||||
|
||||
"github.com/coreos/go-etcd/etcd"
|
||||
"github.com/golang/glog"
|
||||
flag "github.com/spf13/pflag"
|
||||
"github.com/spf13/pflag"
|
||||
)
|
||||
|
||||
var (
|
||||
etcdServerList util.StringList
|
||||
etcdConfigFile = flag.String("etcd_config", "", "The config file for the etcd client. Mutually exclusive with -etcd_servers")
|
||||
bindAddress = util.IP(net.ParseIP("0.0.0.0"))
|
||||
clientConfig = &client.Config{}
|
||||
healthz_port = flag.Int("healthz_port", 10249, "The port to bind the health check server. Use 0 to disable.")
|
||||
oomScoreAdj = flag.Int("oom_score_adj", -899, "The oom_score_adj value for kube-proxy process. Values must be within the range [-1000, 1000]")
|
||||
)
|
||||
|
||||
func init() {
|
||||
client.BindClientConfigFlags(flag.CommandLine, clientConfig)
|
||||
flag.Var(&etcdServerList, "etcd_servers", "List of etcd servers to watch (http://ip:port), comma separated (optional). Mutually exclusive with -etcd_config")
|
||||
flag.Var(&bindAddress, "bind_address", "The IP address for the proxy server to serve on (set to 0.0.0.0 for all interfaces)")
|
||||
}
|
||||
|
||||
func main() {
|
||||
s := server.NewProxyServer()
|
||||
s.AddFlags(pflag.CommandLine)
|
||||
|
||||
util.InitFlags()
|
||||
util.InitLogs()
|
||||
defer util.FlushLogs()
|
||||
|
||||
if err := util.ApplyOomScoreAdj(*oomScoreAdj); err != nil {
|
||||
glog.Info(err)
|
||||
}
|
||||
|
||||
verflag.PrintAndExitIfRequested()
|
||||
|
||||
serviceConfig := config.NewServiceConfig()
|
||||
endpointsConfig := config.NewEndpointsConfig()
|
||||
|
||||
protocol := iptables.ProtocolIpv4
|
||||
if net.IP(bindAddress).To4() == nil {
|
||||
protocol = iptables.ProtocolIpv6
|
||||
}
|
||||
loadBalancer := proxy.NewLoadBalancerRR()
|
||||
proxier := proxy.NewProxier(loadBalancer, net.IP(bindAddress), iptables.New(exec.New(), protocol))
|
||||
if proxier == nil {
|
||||
glog.Fatalf("failed to create proxier, aborting")
|
||||
}
|
||||
// Wire proxier to handle changes to services
|
||||
serviceConfig.RegisterHandler(proxier)
|
||||
// And wire loadBalancer to handle changes to endpoints to services
|
||||
endpointsConfig.RegisterHandler(loadBalancer)
|
||||
|
||||
// Note: RegisterHandler() calls need to happen before creation of Sources because sources
|
||||
// only notify on changes, and the initial update (on process start) may be lost if no handlers
|
||||
// are registered yet.
|
||||
|
||||
// define api config source
|
||||
if clientConfig.Host != "" {
|
||||
glog.Infof("Using api calls to get config %v", clientConfig.Host)
|
||||
client, err := client.New(clientConfig)
|
||||
if err != nil {
|
||||
glog.Fatalf("Invalid API configuration: %v", err)
|
||||
}
|
||||
config.NewSourceAPI(
|
||||
client.Services(api.NamespaceAll),
|
||||
client.Endpoints(api.NamespaceAll),
|
||||
30*time.Second,
|
||||
serviceConfig.Channel("api"),
|
||||
endpointsConfig.Channel("api"),
|
||||
)
|
||||
} else {
|
||||
|
||||
var etcdClient *etcd.Client
|
||||
|
||||
// Set up etcd client
|
||||
if len(etcdServerList) > 0 {
|
||||
// Set up logger for etcd client
|
||||
etcd.SetLogger(util.NewLogger("etcd "))
|
||||
etcdClient = etcd.NewClient(etcdServerList)
|
||||
} else if *etcdConfigFile != "" {
|
||||
// Set up logger for etcd client
|
||||
etcd.SetLogger(util.NewLogger("etcd "))
|
||||
var err error
|
||||
etcdClient, err = etcd.NewClientFromFile(*etcdConfigFile)
|
||||
|
||||
if err != nil {
|
||||
glog.Fatalf("Error with etcd config file: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Create a configuration source that handles configuration from etcd.
|
||||
if etcdClient != nil {
|
||||
glog.Infof("Using etcd servers %v", etcdClient.GetCluster())
|
||||
|
||||
config.NewConfigSourceEtcd(etcdClient,
|
||||
serviceConfig.Channel("etcd"),
|
||||
endpointsConfig.Channel("etcd"))
|
||||
}
|
||||
}
|
||||
|
||||
if *healthz_port > 0 {
|
||||
go util.Forever(func() {
|
||||
err := http.ListenAndServe(bindAddress.String()+":"+strconv.Itoa(*healthz_port), nil)
|
||||
if err != nil {
|
||||
glog.Errorf("Starting health server failed: %v", err)
|
||||
}
|
||||
}, 5*time.Second)
|
||||
}
|
||||
|
||||
// Just loop forever for now...
|
||||
proxier.SyncLoop()
|
||||
s.Run(pflag.CommandLine.Args())
|
||||
}
|
||||
|
||||
@ -21,145 +21,22 @@ limitations under the License.
|
||||
package main
|
||||
|
||||
import (
|
||||
"math/rand"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
"github.com/GoogleCloudPlatform/kubernetes/cmd/kubelet/app"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/credentialprovider"
|
||||
_ "github.com/GoogleCloudPlatform/kubernetes/pkg/healthz"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/master/ports"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/standalone"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/server"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/version/verflag"
|
||||
|
||||
"github.com/golang/glog"
|
||||
flag "github.com/spf13/pflag"
|
||||
"github.com/spf13/pflag"
|
||||
)
|
||||
|
||||
const defaultRootDir = "/var/lib/kubelet"
|
||||
|
||||
var (
|
||||
config = flag.String("config", "", "Path to the config file or directory of files")
|
||||
syncFrequency = flag.Duration("sync_frequency", 10*time.Second, "Max period between synchronizing running containers and config")
|
||||
fileCheckFrequency = flag.Duration("file_check_frequency", 20*time.Second, "Duration between checking config files for new data")
|
||||
httpCheckFrequency = flag.Duration("http_check_frequency", 20*time.Second, "Duration between checking http for new data")
|
||||
manifestURL = flag.String("manifest_url", "", "URL for accessing the container manifest")
|
||||
enableServer = flag.Bool("enable_server", true, "Enable the info server")
|
||||
address = util.IP(net.ParseIP("127.0.0.1"))
|
||||
port = flag.Uint("port", ports.KubeletPort, "The port for the info server to serve on")
|
||||
hostnameOverride = flag.String("hostname_override", "", "If non-empty, will use this string as identification instead of the actual hostname.")
|
||||
podInfraContainerImage = flag.String("pod_infra_container_image", kubelet.PodInfraContainerImage, "The image whose network/ipc namespaces containers in each pod will use.")
|
||||
dockerEndpoint = flag.String("docker_endpoint", "", "If non-empty, use this for the docker endpoint to communicate with")
|
||||
etcdServerList util.StringList
|
||||
etcdConfigFile = flag.String("etcd_config", "", "The config file for the etcd client. Mutually exclusive with -etcd_servers")
|
||||
rootDirectory = flag.String("root_dir", defaultRootDir, "Directory path for managing kubelet files (volume mounts,etc).")
|
||||
allowPrivileged = flag.Bool("allow_privileged", false, "If true, allow containers to request privileged mode. [default=false]")
|
||||
registryPullQPS = flag.Float64("registry_qps", 0.0, "If > 0, limit registry pull QPS to this value. If 0, unlimited. [default=0.0]")
|
||||
registryBurst = flag.Int("registry_burst", 10, "Maximum size of a bursty pulls, temporarily allows pulls to burst to this number, while still not exceeding registry_qps. Only used if --registry_qps > 0")
|
||||
runonce = flag.Bool("runonce", false, "If true, exit after spawning pods from local manifests or remote urls. Exclusive with --etcd_servers, --api_servers, and --enable-server")
|
||||
enableDebuggingHandlers = flag.Bool("enable_debugging_handlers", true, "Enables server endpoints for log collection and local running of containers and commands")
|
||||
minimumGCAge = flag.Duration("minimum_container_ttl_duration", 1*time.Minute, "Minimum age for a finished container before it is garbage collected. Examples: '300ms', '10s' or '2h45m'")
|
||||
maxContainerCount = flag.Int("maximum_dead_containers_per_container", 5, "Maximum number of old instances of a container to retain per container. Each container takes up some disk space. Default: 5.")
|
||||
authPath = flag.String("auth_path", "", "Path to .kubernetes_auth file, specifying how to authenticate to API server.")
|
||||
cAdvisorPort = flag.Uint("cadvisor_port", 4194, "The port of the localhost cAdvisor endpoint")
|
||||
oomScoreAdj = flag.Int("oom_score_adj", -900, "The oom_score_adj value for kubelet process. Values must be within the range [-1000, 1000]")
|
||||
apiServerList util.StringList
|
||||
clusterDomain = flag.String("cluster_domain", "", "Domain for this cluster. If set, kubelet will configure all containers to search this domain in addition to the host's search domains")
|
||||
masterServiceNamespace = flag.String("master_service_namespace", api.NamespaceDefault, "The namespace from which the kubernetes master services should be injected into pods")
|
||||
clusterDNS = util.IP(nil)
|
||||
reallyCrashForTesting = flag.Bool("really_crash_for_testing", false, "If true, crash with panics more often.")
|
||||
)
|
||||
|
||||
func init() {
|
||||
flag.Var(&etcdServerList, "etcd_servers", "List of etcd servers to watch (http://ip:port), comma separated. Mutually exclusive with -etcd_config")
|
||||
flag.Var(&address, "address", "The IP address for the info server to serve on (set to 0.0.0.0 for all interfaces)")
|
||||
flag.Var(&apiServerList, "api_servers", "List of Kubernetes API servers for publishing events, and reading pods and services. (ip:port), comma separated.")
|
||||
flag.Var(&clusterDNS, "cluster_dns", "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")
|
||||
}
|
||||
|
||||
func setupRunOnce() {
|
||||
if *runonce {
|
||||
// Don't use remote (etcd or apiserver) sources
|
||||
if len(etcdServerList) > 0 {
|
||||
glog.Fatalf("invalid option: --runonce and --etcd_servers are mutually exclusive")
|
||||
}
|
||||
if len(apiServerList) > 0 {
|
||||
glog.Fatalf("invalid option: --runonce and --api_servers are mutually exclusive")
|
||||
}
|
||||
if *enableServer {
|
||||
glog.Infof("--runonce is set, disabling server")
|
||||
*enableServer = false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func main() {
|
||||
s := server.NewKubeletServer()
|
||||
s.AddFlags(pflag.CommandLine)
|
||||
|
||||
util.InitFlags()
|
||||
util.InitLogs()
|
||||
util.ReallyCrash = *reallyCrashForTesting
|
||||
defer util.FlushLogs()
|
||||
rand.Seed(time.Now().UTC().UnixNano())
|
||||
|
||||
verflag.PrintAndExitIfRequested()
|
||||
|
||||
// Cluster creation scripts support both kubernetes versions that 1) support kublet watching
|
||||
// apiserver for pods, and 2) ones that don't. So they ca set both --etcd_servers and
|
||||
// --api_servers. The current code will ignore the --etcd_servers flag, while older kubelet
|
||||
// code will use the --etd_servers flag for pods, and use --api_servers for event publising.
|
||||
//
|
||||
// TODO(erictune): convert all cloud provider scripts and Google Container Engine to
|
||||
// use only --api_servers, then delete --etcd_servers flag and the resulting dead code.
|
||||
if len(etcdServerList) > 0 && len(apiServerList) > 0 {
|
||||
glog.Infof("Both --etcd_servers and --api_servers are set. Not using etcd source.")
|
||||
etcdServerList = util.StringList{}
|
||||
}
|
||||
|
||||
setupRunOnce()
|
||||
|
||||
if err := util.ApplyOomScoreAdj(*oomScoreAdj); err != nil {
|
||||
glog.Info(err)
|
||||
}
|
||||
|
||||
client, err := standalone.GetAPIServerClient(*authPath, apiServerList)
|
||||
if err != nil && len(apiServerList) > 0 {
|
||||
glog.Warningf("No API client: %v", err)
|
||||
}
|
||||
|
||||
credentialprovider.SetPreferredDockercfgPath(*rootDirectory)
|
||||
|
||||
kcfg := standalone.KubeletConfig{
|
||||
Address: address,
|
||||
AllowPrivileged: *allowPrivileged,
|
||||
HostnameOverride: *hostnameOverride,
|
||||
RootDirectory: *rootDirectory,
|
||||
ConfigFile: *config,
|
||||
ManifestURL: *manifestURL,
|
||||
FileCheckFrequency: *fileCheckFrequency,
|
||||
HttpCheckFrequency: *httpCheckFrequency,
|
||||
PodInfraContainerImage: *podInfraContainerImage,
|
||||
SyncFrequency: *syncFrequency,
|
||||
RegistryPullQPS: *registryPullQPS,
|
||||
RegistryBurst: *registryBurst,
|
||||
MinimumGCAge: *minimumGCAge,
|
||||
MaxContainerCount: *maxContainerCount,
|
||||
ClusterDomain: *clusterDomain,
|
||||
ClusterDNS: clusterDNS,
|
||||
Runonce: *runonce,
|
||||
Port: *port,
|
||||
CAdvisorPort: *cAdvisorPort,
|
||||
EnableServer: *enableServer,
|
||||
EnableDebuggingHandlers: *enableDebuggingHandlers,
|
||||
DockerClient: util.ConnectToDockerOrDie(*dockerEndpoint),
|
||||
KubeClient: client,
|
||||
EtcdClient: kubelet.EtcdClientOrDie(etcdServerList, *etcdConfigFile),
|
||||
MasterServiceNamespace: *masterServiceNamespace,
|
||||
VolumePlugins: app.ProbeVolumePlugins(),
|
||||
}
|
||||
|
||||
standalone.RunKubelet(&kcfg)
|
||||
// runs forever
|
||||
select {}
|
||||
s.Run(pflag.CommandLine.Args())
|
||||
}
|
||||
|
||||
@ -23,14 +23,25 @@ package main
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
kubeletapp "github.com/GoogleCloudPlatform/kubernetes/cmd/kubelet/app"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/resource"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/testapi"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/apiserver"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/standalone"
|
||||
nodeControllerPkg "github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider/controller"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/controller"
|
||||
kubeletServer "github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/server"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/master"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/master/ports"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/service"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/plugin/pkg/scheduler"
|
||||
_ "github.com/GoogleCloudPlatform/kubernetes/plugin/pkg/scheduler/algorithmprovider"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/plugin/pkg/scheduler/factory"
|
||||
|
||||
"github.com/golang/glog"
|
||||
flag "github.com/spf13/pflag"
|
||||
@ -47,15 +58,89 @@ var (
|
||||
masterServiceNamespace = flag.String("master_service_namespace", api.NamespaceDefault, "The namespace from which the kubernetes master services should be injected into pods")
|
||||
)
|
||||
|
||||
type delegateHandler struct {
|
||||
delegate http.Handler
|
||||
}
|
||||
|
||||
func (h *delegateHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
||||
if h.delegate != nil {
|
||||
h.delegate.ServeHTTP(w, req)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
}
|
||||
|
||||
// RunApiServer starts an API server in a go routine.
|
||||
func runApiServer(cl *client.Client, etcdClient tools.EtcdClient, addr net.IP, port int, masterServiceNamespace string) {
|
||||
handler := delegateHandler{}
|
||||
|
||||
helper, err := master.NewEtcdHelper(etcdClient, "")
|
||||
if err != nil {
|
||||
glog.Fatalf("Unable to get etcd helper: %v", err)
|
||||
}
|
||||
|
||||
// Create a master and install handlers into mux.
|
||||
m := master.New(&master.Config{
|
||||
Client: cl,
|
||||
EtcdHelper: helper,
|
||||
KubeletClient: &client.HTTPKubeletClient{
|
||||
Client: http.DefaultClient,
|
||||
Port: 10250,
|
||||
},
|
||||
EnableLogsSupport: false,
|
||||
EnableSwaggerSupport: true,
|
||||
APIPrefix: "/api",
|
||||
Authorizer: apiserver.NewAlwaysAllowAuthorizer(),
|
||||
|
||||
ReadWritePort: port,
|
||||
ReadOnlyPort: port,
|
||||
PublicAddress: addr,
|
||||
MasterServiceNamespace: masterServiceNamespace,
|
||||
})
|
||||
handler.delegate = m.InsecureHandler
|
||||
|
||||
go http.ListenAndServe(fmt.Sprintf("%s:%d", addr, port), &handler)
|
||||
}
|
||||
|
||||
// RunScheduler starts up a scheduler in it's own goroutine
|
||||
func runScheduler(cl *client.Client) {
|
||||
// Scheduler
|
||||
schedulerConfigFactory := factory.NewConfigFactory(cl)
|
||||
schedulerConfig, err := schedulerConfigFactory.Create()
|
||||
if err != nil {
|
||||
glog.Fatalf("Couldn't create scheduler config: %v", err)
|
||||
}
|
||||
scheduler.New(schedulerConfig).Run()
|
||||
}
|
||||
|
||||
// RunControllerManager starts a controller
|
||||
func runControllerManager(machineList []string, cl *client.Client, nodeMilliCPU, nodeMemory int64) {
|
||||
nodeResources := &api.NodeResources{
|
||||
Capacity: api.ResourceList{
|
||||
api.ResourceCPU: *resource.NewMilliQuantity(nodeMilliCPU, resource.DecimalSI),
|
||||
api.ResourceMemory: *resource.NewQuantity(nodeMemory, resource.BinarySI),
|
||||
},
|
||||
}
|
||||
kubeClient := &client.HTTPKubeletClient{Client: http.DefaultClient, Port: ports.KubeletPort}
|
||||
nodeController := nodeControllerPkg.NewNodeController(nil, "", machineList, nodeResources, cl, kubeClient)
|
||||
nodeController.Run(10*time.Second, 10)
|
||||
|
||||
endpoints := service.NewEndpointController(cl)
|
||||
go util.Forever(func() { endpoints.SyncServiceEndpoints() }, time.Second*10)
|
||||
|
||||
controllerManager := controller.NewReplicationManager(cl)
|
||||
controllerManager.Run(10 * time.Second)
|
||||
}
|
||||
|
||||
func startComponents(etcdClient tools.EtcdClient, cl *client.Client, addr net.IP, port int) {
|
||||
machineList := []string{"localhost"}
|
||||
|
||||
standalone.RunApiServer(cl, etcdClient, addr, port, *masterServiceNamespace)
|
||||
standalone.RunScheduler(cl)
|
||||
standalone.RunControllerManager(machineList, cl, *nodeMilliCPU, *nodeMemory)
|
||||
runApiServer(cl, etcdClient, addr, port, *masterServiceNamespace)
|
||||
runScheduler(cl)
|
||||
runControllerManager(machineList, cl, *nodeMilliCPU, *nodeMemory)
|
||||
|
||||
dockerClient := util.ConnectToDockerOrDie(*dockerEndpoint)
|
||||
standalone.SimpleRunKubelet(cl, nil, dockerClient, machineList[0], "/tmp/kubernetes", "", "127.0.0.1", 10250, *masterServiceNamespace, kubeletapp.ProbeVolumePlugins())
|
||||
kubeletServer.SimpleRunKubelet(cl, nil, dockerClient, machineList[0], "/tmp/kubernetes", "", "127.0.0.1", 10250, *masterServiceNamespace, kubeletServer.ProbeVolumePlugins())
|
||||
}
|
||||
|
||||
func newApiClient(addr net.IP, port int) *client.Client {
|
||||
|
||||
@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package app
|
||||
package server
|
||||
|
||||
// This file exists to force the desired plugin implementations to be linked.
|
||||
import (
|
||||
@ -28,6 +28,7 @@ import (
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume/host_path"
|
||||
)
|
||||
|
||||
// ProbeVolumePlugins collects all volume plugins into an easy to use list.
|
||||
func ProbeVolumePlugins() []volume.Plugin {
|
||||
allPlugins := []volume.Plugin{}
|
||||
|
||||
430
pkg/kubelet/server/server.go
Normal file
430
pkg/kubelet/server/server.go
Normal file
@ -0,0 +1,430 @@
|
||||
/*
|
||||
Copyright 2014 Google Inc. All rights reserved.
|
||||
|
||||
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 server makes it easy to create a kubelet server for various contexts.
|
||||
package server
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/clientauth"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/credentialprovider"
|
||||
_ "github.com/GoogleCloudPlatform/kubernetes/pkg/healthz"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/hyperkube"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/config"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/dockertools"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/master/ports"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
|
||||
|
||||
"github.com/golang/glog"
|
||||
"github.com/spf13/pflag"
|
||||
)
|
||||
|
||||
const defaultRootDir = "/var/lib/kubelet"
|
||||
|
||||
// KubeletServer encapsulates all of the parameters necessary for starting up
|
||||
// a kubelet. These can either be set via command line or directly.
|
||||
type KubeletServer struct {
|
||||
Config string
|
||||
SyncFrequency time.Duration
|
||||
FileCheckFrequency time.Duration
|
||||
HTTPCheckFrequency time.Duration
|
||||
ManifestURL string
|
||||
EnableServer bool
|
||||
Address util.IP
|
||||
Port uint
|
||||
HostnameOverride string
|
||||
PodInfraContainerImage string
|
||||
DockerEndpoint string
|
||||
EtcdServerList util.StringList
|
||||
EtcdConfigFile string
|
||||
RootDirectory string
|
||||
AllowPrivileged bool
|
||||
RegistryPullQPS float64
|
||||
RegistryBurst int
|
||||
RunOnce bool
|
||||
EnableDebuggingHandlers bool
|
||||
MinimumGCAge time.Duration
|
||||
MaxContainerCount int
|
||||
AuthPath string
|
||||
CAdvisorPort uint
|
||||
OOMScoreAdj int
|
||||
APIServerList util.StringList
|
||||
ClusterDomain string
|
||||
MasterServiceNamespace string
|
||||
ClusterDNS util.IP
|
||||
ReallyCrashForTesting bool
|
||||
}
|
||||
|
||||
// NewKubeletServer will create a new KubeletServer with default values.
|
||||
func NewKubeletServer() *KubeletServer {
|
||||
return &KubeletServer{
|
||||
SyncFrequency: 10 * time.Second,
|
||||
FileCheckFrequency: 20 * time.Second,
|
||||
HTTPCheckFrequency: 20 * time.Second,
|
||||
EnableServer: true,
|
||||
Address: util.IP(net.ParseIP("127.0.0.1")),
|
||||
Port: ports.KubeletPort,
|
||||
PodInfraContainerImage: kubelet.PodInfraContainerImage,
|
||||
RootDirectory: defaultRootDir,
|
||||
RegistryBurst: 10,
|
||||
EnableDebuggingHandlers: true,
|
||||
MinimumGCAge: 1 * time.Minute,
|
||||
MaxContainerCount: 5,
|
||||
CAdvisorPort: 4194,
|
||||
OOMScoreAdj: -900,
|
||||
MasterServiceNamespace: api.NamespaceDefault,
|
||||
}
|
||||
}
|
||||
|
||||
// NewHyperkubeServer creates a new hyperkube Server object that includes the
|
||||
// description and flags.
|
||||
func NewHyperkubeServer() *hyperkube.Server {
|
||||
s := NewKubeletServer()
|
||||
hks := hyperkube.Server{
|
||||
SimpleUsage: "kubelet",
|
||||
Long: `The kubelet binary is responsible for maintaining a set of containers on a
|
||||
particular node. It syncs data from a variety of sources including a
|
||||
Kubernetes API server, an etcd cluster, HTTP endpoint or local file. It then
|
||||
queries Docker to see what is currently running. It synchronizes the
|
||||
configuration data, with the running set of containers by starting or stopping
|
||||
Docker containers.`,
|
||||
Run: func(_ *hyperkube.Server, args []string) error {
|
||||
return s.Run(args)
|
||||
},
|
||||
}
|
||||
s.AddFlags(hks.Flags())
|
||||
return &hks
|
||||
}
|
||||
|
||||
// AddFlags adds flags for a specific KubeletServer to the specified FlagSet
|
||||
func (s *KubeletServer) AddFlags(fs *pflag.FlagSet) {
|
||||
fs.StringVar(&s.Config, "config", s.Config, "Path to the config file or directory of files")
|
||||
fs.DurationVar(&s.SyncFrequency, "sync_frequency", s.SyncFrequency, "Max period between synchronizing running containers and config")
|
||||
fs.DurationVar(&s.FileCheckFrequency, "file_check_frequency", s.FileCheckFrequency, "Duration between checking config files for new data")
|
||||
fs.DurationVar(&s.HTTPCheckFrequency, "http_check_frequency", s.HTTPCheckFrequency, "Duration between checking http for new data")
|
||||
fs.StringVar(&s.ManifestURL, "manifest_url", s.ManifestURL, "URL for accessing the container manifest")
|
||||
fs.BoolVar(&s.EnableServer, "enable_server", s.EnableServer, "Enable the info server")
|
||||
fs.Var(&s.Address, "address", "The IP address for the info server to serve on (set to 0.0.0.0 for all interfaces)")
|
||||
fs.UintVar(&s.Port, "port", s.Port, "The port for the info server to serve 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.PodInfraContainerImage, "pod_infra_container_image", s.PodInfraContainerImage, "The image whose network/ipc namespaces containers in each pod will use.")
|
||||
fs.StringVar(&s.DockerEndpoint, "docker_endpoint", s.DockerEndpoint, "If non-empty, use this for the docker endpoint to communicate with")
|
||||
fs.Var(&s.EtcdServerList, "etcd_servers", "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.RootDirectory, "root_dir", s.RootDirectory, "Directory path for managing kubelet files (volume mounts,etc).")
|
||||
fs.BoolVar(&s.AllowPrivileged, "allow_privileged", s.AllowPrivileged, "If true, allow containers to request privileged mode. [default=false]")
|
||||
fs.Float64Var(&s.RegistryPullQPS, "registry_qps", s.RegistryPullQPS, "If > 0, limit registry pull QPS to this value. If 0, unlimited. [default=0.0]")
|
||||
fs.IntVar(&s.RegistryBurst, "registry_burst", s.RegistryBurst, "Maximum size of a bursty pulls, temporarily allows pulls to burst to this number, while still not exceeding registry_qps. Only used if --registry_qps > 0")
|
||||
fs.BoolVar(&s.RunOnce, "runonce", s.RunOnce, "If true, exit after spawning pods from local manifests or remote urls. Exclusive with --etcd_servers, --api_servers, and --enable-server")
|
||||
fs.BoolVar(&s.EnableDebuggingHandlers, "enable_debugging_handlers", s.EnableDebuggingHandlers, "Enables server endpoints for log collection and local running of containers and commands")
|
||||
fs.DurationVar(&s.MinimumGCAge, "minimum_container_ttl_duration", s.MinimumGCAge, "Minimum age for a finished container before it is garbage collected. Examples: '300ms', '10s' or '2h45m'")
|
||||
fs.IntVar(&s.MaxContainerCount, "maximum_dead_containers_per_container", s.MaxContainerCount, "Maximum number of old instances of a container to retain per container. Each container takes up some disk space. Default: 5.")
|
||||
fs.StringVar(&s.AuthPath, "auth_path", s.AuthPath, "Path to .kubernetes_auth file, specifying how to authenticate to API server.")
|
||||
fs.UintVar(&s.CAdvisorPort, "cadvisor_port", s.CAdvisorPort, "The port of the localhost cAdvisor endpoint")
|
||||
fs.IntVar(&s.OOMScoreAdj, "oom_score_adj", s.OOMScoreAdj, "The oom_score_adj value for kubelet process. Values must be within the range [-1000, 1000]")
|
||||
fs.Var(&s.APIServerList, "api_servers", "List of Kubernetes API servers for publishing events, and reading pods and services. (ip:port), comma separated.")
|
||||
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.MasterServiceNamespace, "master_service_namespace", s.MasterServiceNamespace, "The namespace from which the kubernetes master services should be injected into pods")
|
||||
fs.Var(&s.ClusterDNS, "cluster_dns", "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.BoolVar(&s.ReallyCrashForTesting, "really_crash_for_testing", s.ReallyCrashForTesting, "If true, crash with panics more often.")
|
||||
}
|
||||
|
||||
// Run runs the specified KubeletServer. This should never exit.
|
||||
func (s *KubeletServer) Run(_ []string) error {
|
||||
util.ReallyCrash = s.ReallyCrashForTesting
|
||||
rand.Seed(time.Now().UTC().UnixNano())
|
||||
|
||||
// Cluster creation scripts support both kubernetes versions that 1)
|
||||
// support kublet watching apiserver for pods, and 2) ones that don't. So
|
||||
// they can set both --etcd_servers and --api_servers. The current code
|
||||
// will ignore the --etcd_servers flag, while older kubelet code will use
|
||||
// the --etcd_servers flag for pods, and use --api_servers for event
|
||||
// publising.
|
||||
//
|
||||
// TODO(erictune): convert all cloud provider scripts and Google Container Engine to
|
||||
// use only --api_servers, then delete --etcd_servers flag and the resulting dead code.
|
||||
if len(s.EtcdServerList) > 0 && len(s.APIServerList) > 0 {
|
||||
glog.Infof("Both --etcd_servers and --api_servers are set. Not using etcd source.")
|
||||
s.EtcdServerList = util.StringList{}
|
||||
}
|
||||
|
||||
if err := util.ApplyOomScoreAdj(s.OOMScoreAdj); err != nil {
|
||||
glog.Info(err)
|
||||
}
|
||||
|
||||
client, err := s.createAPIServerClient()
|
||||
if err != nil && len(s.APIServerList) > 0 {
|
||||
glog.Warningf("No API client: %v", err)
|
||||
}
|
||||
|
||||
credentialprovider.SetPreferredDockercfgPath(s.RootDirectory)
|
||||
|
||||
kcfg := KubeletConfig{
|
||||
Address: s.Address,
|
||||
AllowPrivileged: s.AllowPrivileged,
|
||||
HostnameOverride: s.HostnameOverride,
|
||||
RootDirectory: s.RootDirectory,
|
||||
ConfigFile: s.Config,
|
||||
ManifestURL: s.ManifestURL,
|
||||
FileCheckFrequency: s.FileCheckFrequency,
|
||||
HTTPCheckFrequency: s.HTTPCheckFrequency,
|
||||
PodInfraContainerImage: s.PodInfraContainerImage,
|
||||
SyncFrequency: s.SyncFrequency,
|
||||
RegistryPullQPS: s.RegistryPullQPS,
|
||||
RegistryBurst: s.RegistryBurst,
|
||||
MinimumGCAge: s.MinimumGCAge,
|
||||
MaxContainerCount: s.MaxContainerCount,
|
||||
ClusterDomain: s.ClusterDomain,
|
||||
ClusterDNS: s.ClusterDNS,
|
||||
Runonce: s.RunOnce,
|
||||
Port: s.Port,
|
||||
CAdvisorPort: s.CAdvisorPort,
|
||||
EnableServer: s.EnableServer,
|
||||
EnableDebuggingHandlers: s.EnableDebuggingHandlers,
|
||||
DockerClient: util.ConnectToDockerOrDie(s.DockerEndpoint),
|
||||
KubeClient: client,
|
||||
EtcdClient: kubelet.EtcdClientOrDie(s.EtcdServerList, s.EtcdConfigFile),
|
||||
MasterServiceNamespace: s.MasterServiceNamespace,
|
||||
VolumePlugins: ProbeVolumePlugins(),
|
||||
}
|
||||
|
||||
RunKubelet(&kcfg)
|
||||
|
||||
// runs forever
|
||||
select {}
|
||||
|
||||
}
|
||||
|
||||
func (s *KubeletServer) setupRunOnce() {
|
||||
if s.RunOnce {
|
||||
// Don't use remote (etcd or apiserver) sources
|
||||
if len(s.EtcdServerList) > 0 {
|
||||
glog.Fatalf("invalid option: --runonce and --etcd_servers are mutually exclusive")
|
||||
}
|
||||
if len(s.APIServerList) > 0 {
|
||||
glog.Fatalf("invalid option: --runonce and --api_servers are mutually exclusive")
|
||||
}
|
||||
if s.EnableServer {
|
||||
glog.Infof("--runonce is set, disabling server")
|
||||
s.EnableServer = false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: replace this with clientcmd
|
||||
func (s *KubeletServer) createAPIServerClient() (*client.Client, error) {
|
||||
authInfo, err := clientauth.LoadFromFile(s.AuthPath)
|
||||
if err != nil {
|
||||
glog.Warningf("Could not load kubernetes auth path: %v. Continuing with defaults.", err)
|
||||
}
|
||||
if authInfo == nil {
|
||||
// authInfo didn't load correctly - continue with defaults.
|
||||
authInfo = &clientauth.Info{}
|
||||
}
|
||||
clientConfig, err := authInfo.MergeWithConfig(client.Config{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(s.APIServerList) < 1 {
|
||||
return nil, fmt.Errorf("no api servers specified")
|
||||
}
|
||||
// TODO: adapt Kube client to support LB over several servers
|
||||
if len(s.APIServerList) > 1 {
|
||||
glog.Infof("Multiple api servers specified. Picking first one")
|
||||
}
|
||||
clientConfig.Host = s.APIServerList[0]
|
||||
c, err := client.New(&clientConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// SimpleRunKubelet is a simple way to start a Kubelet talking to dockerEndpoint, using an etcdClient.
|
||||
// Under the hood it calls RunKubelet (below)
|
||||
func SimpleRunKubelet(client *client.Client,
|
||||
etcdClient tools.EtcdClient,
|
||||
dockerClient dockertools.DockerInterface,
|
||||
hostname, rootDir, manifestURL, address string,
|
||||
port uint,
|
||||
masterServiceNamespace string,
|
||||
volumePlugins []volume.Plugin) {
|
||||
kcfg := KubeletConfig{
|
||||
KubeClient: client,
|
||||
EtcdClient: etcdClient,
|
||||
DockerClient: dockerClient,
|
||||
HostnameOverride: hostname,
|
||||
RootDirectory: rootDir,
|
||||
ManifestURL: manifestURL,
|
||||
PodInfraContainerImage: kubelet.PodInfraContainerImage,
|
||||
Port: port,
|
||||
Address: util.IP(net.ParseIP(address)),
|
||||
EnableServer: true,
|
||||
EnableDebuggingHandlers: true,
|
||||
SyncFrequency: 3 * time.Second,
|
||||
MinimumGCAge: 10 * time.Second,
|
||||
MaxContainerCount: 5,
|
||||
MasterServiceNamespace: masterServiceNamespace,
|
||||
VolumePlugins: volumePlugins,
|
||||
}
|
||||
RunKubelet(&kcfg)
|
||||
}
|
||||
|
||||
// RunKubelet is responsible for setting up and running a kubelet. It is used in three different applications:
|
||||
// 1 Integration tests
|
||||
// 2 Kubelet binary
|
||||
// 3 Standalone 'kubernetes' binary
|
||||
// Eventually, #2 will be replaced with instances of #3
|
||||
func RunKubelet(kcfg *KubeletConfig) {
|
||||
kcfg.Hostname = util.GetHostname(kcfg.HostnameOverride)
|
||||
if kcfg.KubeClient != nil {
|
||||
kubelet.SetupEventSending(kcfg.KubeClient, kcfg.Hostname)
|
||||
} else {
|
||||
glog.Infof("No api server defined - no events will be sent.")
|
||||
}
|
||||
kubelet.SetupLogging()
|
||||
kubelet.SetupCapabilities(kcfg.AllowPrivileged)
|
||||
|
||||
credentialprovider.SetPreferredDockercfgPath(kcfg.RootDirectory)
|
||||
|
||||
podCfg := makePodSourceConfig(kcfg)
|
||||
k, err := createAndInitKubelet(kcfg, podCfg)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to create kubelet: %s", err)
|
||||
return
|
||||
}
|
||||
// process pods and exit.
|
||||
if kcfg.Runonce {
|
||||
if _, err := k.RunOnce(podCfg.Updates()); err != nil {
|
||||
glog.Errorf("--runonce failed: %v", err)
|
||||
}
|
||||
} else {
|
||||
startKubelet(k, podCfg, kcfg)
|
||||
}
|
||||
}
|
||||
|
||||
func startKubelet(k *kubelet.Kubelet, podCfg *config.PodConfig, kc *KubeletConfig) {
|
||||
// start the kubelet
|
||||
go util.Forever(func() { k.Run(podCfg.Updates()) }, 0)
|
||||
|
||||
// start the kubelet server
|
||||
if kc.EnableServer {
|
||||
go util.Forever(func() {
|
||||
kubelet.ListenAndServeKubeletServer(k, net.IP(kc.Address), kc.Port, kc.EnableDebuggingHandlers)
|
||||
}, 0)
|
||||
}
|
||||
}
|
||||
|
||||
func makePodSourceConfig(kc *KubeletConfig) *config.PodConfig {
|
||||
// source of all configuration
|
||||
cfg := config.NewPodConfig(config.PodConfigNotificationSnapshotAndUpdates)
|
||||
|
||||
// define file config source
|
||||
if kc.ConfigFile != "" {
|
||||
glog.Infof("Adding manifest file: %v", kc.ConfigFile)
|
||||
config.NewSourceFile(kc.ConfigFile, kc.FileCheckFrequency, cfg.Channel(kubelet.FileSource))
|
||||
}
|
||||
|
||||
// define url config source
|
||||
if kc.ManifestURL != "" {
|
||||
glog.Infof("Adding manifest url: %v", kc.ManifestURL)
|
||||
config.NewSourceURL(kc.ManifestURL, kc.HTTPCheckFrequency, cfg.Channel(kubelet.HTTPSource))
|
||||
}
|
||||
if kc.EtcdClient != nil {
|
||||
glog.Infof("Watching for etcd configs at %v", kc.EtcdClient.GetCluster())
|
||||
config.NewSourceEtcd(config.EtcdKeyForHost(kc.Hostname), kc.EtcdClient, cfg.Channel(kubelet.EtcdSource))
|
||||
}
|
||||
if kc.KubeClient != nil {
|
||||
glog.Infof("Watching apiserver")
|
||||
config.NewSourceApiserver(kc.KubeClient, kc.Hostname, cfg.Channel(kubelet.ApiserverSource))
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
// KubeletConfig is all of the parameters necessary for running a kubelet.
|
||||
// TODO: This should probably be merged with KubeletServer. The extra object is a consequence of refactoring.
|
||||
type KubeletConfig struct {
|
||||
EtcdClient tools.EtcdClient
|
||||
KubeClient *client.Client
|
||||
DockerClient dockertools.DockerInterface
|
||||
CAdvisorPort uint
|
||||
Address util.IP
|
||||
AllowPrivileged bool
|
||||
HostnameOverride string
|
||||
RootDirectory string
|
||||
ConfigFile string
|
||||
ManifestURL string
|
||||
FileCheckFrequency time.Duration
|
||||
HTTPCheckFrequency time.Duration
|
||||
Hostname string
|
||||
PodInfraContainerImage string
|
||||
SyncFrequency time.Duration
|
||||
RegistryPullQPS float64
|
||||
RegistryBurst int
|
||||
MinimumGCAge time.Duration
|
||||
MaxContainerCount int
|
||||
ClusterDomain string
|
||||
ClusterDNS util.IP
|
||||
EnableServer bool
|
||||
EnableDebuggingHandlers bool
|
||||
Port uint
|
||||
Runonce bool
|
||||
MasterServiceNamespace string
|
||||
VolumePlugins []volume.Plugin
|
||||
}
|
||||
|
||||
func createAndInitKubelet(kc *KubeletConfig, pc *config.PodConfig) (*kubelet.Kubelet, 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
|
||||
|
||||
k, err := kubelet.NewMainKubelet(
|
||||
kc.Hostname,
|
||||
kc.DockerClient,
|
||||
kc.EtcdClient,
|
||||
kc.KubeClient,
|
||||
kc.RootDirectory,
|
||||
kc.PodInfraContainerImage,
|
||||
kc.SyncFrequency,
|
||||
float32(kc.RegistryPullQPS),
|
||||
kc.RegistryBurst,
|
||||
kc.MinimumGCAge,
|
||||
kc.MaxContainerCount,
|
||||
pc.IsSourceSeen,
|
||||
kc.ClusterDomain,
|
||||
net.IP(kc.ClusterDNS),
|
||||
kc.MasterServiceNamespace,
|
||||
kc.VolumePlugins)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
k.BirthCry()
|
||||
|
||||
go k.GarbageCollectLoop()
|
||||
go kubelet.MonitorCAdvisor(k, kc.CAdvisorPort)
|
||||
|
||||
return k, nil
|
||||
}
|
||||
173
pkg/proxy/server/server.go
Normal file
173
pkg/proxy/server/server.go
Normal file
@ -0,0 +1,173 @@
|
||||
/*
|
||||
Copyright 2014 Google Inc. All rights reserved.
|
||||
|
||||
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 server does all of the work necessary to configure and run a
|
||||
// Kubernetes proxy process.
|
||||
package server
|
||||
|
||||
import (
|
||||
"net"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
|
||||
_ "github.com/GoogleCloudPlatform/kubernetes/pkg/healthz"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/hyperkube"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/proxy"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/proxy/config"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/exec"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/iptables"
|
||||
|
||||
"github.com/coreos/go-etcd/etcd"
|
||||
"github.com/golang/glog"
|
||||
"github.com/spf13/pflag"
|
||||
)
|
||||
|
||||
// ProxyServer contains configures and runs a Kubernetes proxy server
|
||||
type ProxyServer struct {
|
||||
EtcdServerList util.StringList
|
||||
EtcdConfigFile string
|
||||
BindAddress util.IP
|
||||
ClientConfig client.Config
|
||||
HealthzPort int
|
||||
OOMScoreAdj int
|
||||
}
|
||||
|
||||
// NewProxyServer creates a new ProxyServer object with default parameters
|
||||
func NewProxyServer() *ProxyServer {
|
||||
return &ProxyServer{
|
||||
BindAddress: util.IP(net.ParseIP("0.0.0.0")),
|
||||
HealthzPort: 10249,
|
||||
OOMScoreAdj: -899,
|
||||
}
|
||||
}
|
||||
|
||||
// NewHyperkubeServer creates a new hyperkube Server object that includes the
|
||||
// description and flags.
|
||||
func NewHyperkubeServer() *hyperkube.Server {
|
||||
s := NewProxyServer()
|
||||
|
||||
hks := hyperkube.Server{
|
||||
SimpleUsage: "proxy",
|
||||
Long: `The Kubernetes proxy server is responsible for taking traffic directed at
|
||||
services and forwarding it to the appropriate pods. It generally runs on
|
||||
nodes next to the Kubelet and proxies traffic from local pods to remote pods.
|
||||
It is also used when handling incoming external traffic.`,
|
||||
Run: func(_ *hyperkube.Server, args []string) error {
|
||||
return s.Run(args)
|
||||
},
|
||||
}
|
||||
s.AddFlags(hks.Flags())
|
||||
return &hks
|
||||
}
|
||||
|
||||
// AddFlags adds flags for a specific ProxyServer to the specified FlagSet
|
||||
func (s *ProxyServer) AddFlags(fs *pflag.FlagSet) {
|
||||
fs.StringVar(&s.EtcdConfigFile, "etcd_config", s.EtcdConfigFile, "The config file for the etcd client. Mutually exclusive with -etcd_servers")
|
||||
fs.Var(&s.EtcdServerList, "etcd_servers", "List of etcd servers to watch (http://ip:port), comma separated (optional). Mutually exclusive with -etcd_config")
|
||||
fs.Var(&s.BindAddress, "bind_address", "The IP address for the proxy server to serve on (set to 0.0.0.0 for all interfaces)")
|
||||
client.BindClientConfigFlags(fs, &s.ClientConfig)
|
||||
fs.IntVar(&s.HealthzPort, "healthz_port", s.HealthzPort, "The port to bind the health check server. Use 0 to disable.")
|
||||
fs.IntVar(&s.OOMScoreAdj, "oom_score_adj", s.OOMScoreAdj, "The oom_score_adj value for kube-proxy process. Values must be within the range [-1000, 1000]")
|
||||
}
|
||||
|
||||
// Run runs the specified ProxyServer. This should never exit.
|
||||
func (s *ProxyServer) Run(_ []string) error {
|
||||
if err := util.ApplyOomScoreAdj(s.OOMScoreAdj); err != nil {
|
||||
glog.Info(err)
|
||||
}
|
||||
|
||||
serviceConfig := config.NewServiceConfig()
|
||||
endpointsConfig := config.NewEndpointsConfig()
|
||||
|
||||
protocol := iptables.ProtocolIpv4
|
||||
if net.IP(s.BindAddress).To4() == nil {
|
||||
protocol = iptables.ProtocolIpv6
|
||||
}
|
||||
loadBalancer := proxy.NewLoadBalancerRR()
|
||||
proxier := proxy.NewProxier(loadBalancer, net.IP(s.BindAddress), iptables.New(exec.New(), protocol))
|
||||
if proxier == nil {
|
||||
glog.Fatalf("failed to create proxier, aborting")
|
||||
}
|
||||
|
||||
// Wire proxier to handle changes to services
|
||||
serviceConfig.RegisterHandler(proxier)
|
||||
// And wire loadBalancer to handle changes to endpoints to services
|
||||
endpointsConfig.RegisterHandler(loadBalancer)
|
||||
|
||||
// Note: RegisterHandler() calls need to happen before creation of Sources because sources
|
||||
// only notify on changes, and the initial update (on process start) may be lost if no handlers
|
||||
// are registered yet.
|
||||
|
||||
// define api config source
|
||||
if s.ClientConfig.Host != "" {
|
||||
glog.Infof("Using API calls to get config %v", s.ClientConfig.Host)
|
||||
client, err := client.New(&s.ClientConfig)
|
||||
if err != nil {
|
||||
glog.Fatalf("Invalid API configuration: %v", err)
|
||||
}
|
||||
config.NewSourceAPI(
|
||||
client.Services(api.NamespaceAll),
|
||||
client.Endpoints(api.NamespaceAll),
|
||||
30*time.Second,
|
||||
serviceConfig.Channel("api"),
|
||||
endpointsConfig.Channel("api"),
|
||||
)
|
||||
} else {
|
||||
var etcdClient *etcd.Client
|
||||
|
||||
// Set up etcd client
|
||||
if len(s.EtcdServerList) > 0 {
|
||||
// Set up logger for etcd client
|
||||
etcd.SetLogger(util.NewLogger("etcd "))
|
||||
etcdClient = etcd.NewClient(s.EtcdServerList)
|
||||
} else if s.EtcdConfigFile != "" {
|
||||
// Set up logger for etcd client
|
||||
etcd.SetLogger(util.NewLogger("etcd "))
|
||||
var err error
|
||||
etcdClient, err = etcd.NewClientFromFile(s.EtcdConfigFile)
|
||||
|
||||
if err != nil {
|
||||
glog.Fatalf("Error with etcd config file: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Create a configuration source that handles configuration from etcd.
|
||||
if etcdClient != nil {
|
||||
glog.Infof("Using etcd servers %v", etcdClient.GetCluster())
|
||||
|
||||
config.NewConfigSourceEtcd(etcdClient,
|
||||
serviceConfig.Channel("etcd"),
|
||||
endpointsConfig.Channel("etcd"))
|
||||
}
|
||||
}
|
||||
|
||||
if s.HealthzPort > 0 {
|
||||
go util.Forever(func() {
|
||||
err := http.ListenAndServe(s.BindAddress.String()+":"+strconv.Itoa(s.HealthzPort), nil)
|
||||
if err != nil {
|
||||
glog.Errorf("Starting health server failed: %v", err)
|
||||
}
|
||||
}, 5*time.Second)
|
||||
}
|
||||
|
||||
// Just loop forever for now...
|
||||
proxier.SyncLoop()
|
||||
return nil
|
||||
}
|
||||
@ -1,18 +0,0 @@
|
||||
/*
|
||||
Copyright 2014 Google Inc. All rights reserved.
|
||||
|
||||
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 standalone has utilities for running different Kubernetes binaries in a single binary.
|
||||
package standalone
|
||||
@ -1,317 +0,0 @@
|
||||
/*
|
||||
Copyright 2014 Google Inc. All rights reserved.
|
||||
|
||||
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 standalone
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/resource"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/apiserver"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
|
||||
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/clientauth"
|
||||
nodeControllerPkg "github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider/controller"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/controller"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/credentialprovider"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/config"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/dockertools"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/master"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/master/ports"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/service"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/plugin/pkg/scheduler"
|
||||
_ "github.com/GoogleCloudPlatform/kubernetes/plugin/pkg/scheduler/algorithmprovider"
|
||||
"github.com/GoogleCloudPlatform/kubernetes/plugin/pkg/scheduler/factory"
|
||||
|
||||
"github.com/golang/glog"
|
||||
)
|
||||
|
||||
type delegateHandler struct {
|
||||
delegate http.Handler
|
||||
}
|
||||
|
||||
func (h *delegateHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
||||
if h.delegate != nil {
|
||||
h.delegate.ServeHTTP(w, req)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
}
|
||||
|
||||
// TODO: replace this with clientcmd
|
||||
func GetAPIServerClient(authPath string, apiServerList util.StringList) (*client.Client, error) {
|
||||
authInfo, err := clientauth.LoadFromFile(authPath)
|
||||
if err != nil {
|
||||
glog.Warningf("Could not load kubernetes auth path: %v. Continuing with defaults.", err)
|
||||
}
|
||||
if authInfo == nil {
|
||||
// authInfo didn't load correctly - continue with defaults.
|
||||
authInfo = &clientauth.Info{}
|
||||
}
|
||||
clientConfig, err := authInfo.MergeWithConfig(client.Config{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(apiServerList) < 1 {
|
||||
return nil, fmt.Errorf("no api servers specified.")
|
||||
}
|
||||
// TODO: adapt Kube client to support LB over several servers
|
||||
if len(apiServerList) > 1 {
|
||||
glog.Infof("Multiple api servers specified. Picking first one")
|
||||
}
|
||||
clientConfig.Host = apiServerList[0]
|
||||
c, err := client.New(&clientConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// RunApiServer starts an API server in a go routine.
|
||||
func RunApiServer(cl *client.Client, etcdClient tools.EtcdClient, addr net.IP, port int, masterServiceNamespace string) {
|
||||
handler := delegateHandler{}
|
||||
|
||||
helper, err := master.NewEtcdHelper(etcdClient, "")
|
||||
if err != nil {
|
||||
glog.Fatalf("Unable to get etcd helper: %v", err)
|
||||
}
|
||||
|
||||
// Create a master and install handlers into mux.
|
||||
m := master.New(&master.Config{
|
||||
Client: cl,
|
||||
EtcdHelper: helper,
|
||||
KubeletClient: &client.HTTPKubeletClient{
|
||||
Client: http.DefaultClient,
|
||||
Port: 10250,
|
||||
},
|
||||
EnableLogsSupport: false,
|
||||
EnableSwaggerSupport: true,
|
||||
EnableIndex: true,
|
||||
APIPrefix: "/api",
|
||||
Authorizer: apiserver.NewAlwaysAllowAuthorizer(),
|
||||
|
||||
ReadWritePort: port,
|
||||
ReadOnlyPort: port,
|
||||
PublicAddress: addr,
|
||||
MasterServiceNamespace: masterServiceNamespace,
|
||||
})
|
||||
handler.delegate = m.InsecureHandler
|
||||
|
||||
go http.ListenAndServe(fmt.Sprintf("%s:%d", addr, port), &handler)
|
||||
}
|
||||
|
||||
// RunScheduler starts up a scheduler in it's own goroutine
|
||||
func RunScheduler(cl *client.Client) {
|
||||
// Scheduler
|
||||
schedulerConfigFactory := factory.NewConfigFactory(cl)
|
||||
schedulerConfig, err := schedulerConfigFactory.Create()
|
||||
if err != nil {
|
||||
glog.Fatalf("Couldn't create scheduler config: %v", err)
|
||||
}
|
||||
scheduler.New(schedulerConfig).Run()
|
||||
}
|
||||
|
||||
// RunControllerManager starts a controller
|
||||
func RunControllerManager(machineList []string, cl *client.Client, nodeMilliCPU, nodeMemory int64) {
|
||||
nodeResources := &api.NodeResources{
|
||||
Capacity: api.ResourceList{
|
||||
api.ResourceCPU: *resource.NewMilliQuantity(nodeMilliCPU, resource.DecimalSI),
|
||||
api.ResourceMemory: *resource.NewQuantity(nodeMemory, resource.BinarySI),
|
||||
},
|
||||
}
|
||||
kubeClient := &client.HTTPKubeletClient{Client: http.DefaultClient, Port: ports.KubeletPort}
|
||||
nodeController := nodeControllerPkg.NewNodeController(nil, "", machineList, nodeResources, cl, kubeClient)
|
||||
nodeController.Run(10*time.Second, 10)
|
||||
|
||||
endpoints := service.NewEndpointController(cl)
|
||||
go util.Forever(func() { endpoints.SyncServiceEndpoints() }, time.Second*10)
|
||||
|
||||
controllerManager := controller.NewReplicationManager(cl)
|
||||
controllerManager.Run(10 * time.Second)
|
||||
}
|
||||
|
||||
// SimpleRunKubelet is a simple way to start a Kubelet talking to dockerEndpoint, using an etcdClient.
|
||||
// Under the hood it calls RunKubelet (below)
|
||||
func SimpleRunKubelet(client *client.Client,
|
||||
etcdClient tools.EtcdClient,
|
||||
dockerClient dockertools.DockerInterface,
|
||||
hostname, rootDir, manifestURL, address string,
|
||||
port uint,
|
||||
masterServiceNamespace string,
|
||||
volumePlugins []volume.Plugin) {
|
||||
kcfg := KubeletConfig{
|
||||
KubeClient: client,
|
||||
EtcdClient: etcdClient,
|
||||
DockerClient: dockerClient,
|
||||
HostnameOverride: hostname,
|
||||
RootDirectory: rootDir,
|
||||
ManifestURL: manifestURL,
|
||||
PodInfraContainerImage: kubelet.PodInfraContainerImage,
|
||||
Port: port,
|
||||
Address: util.IP(net.ParseIP(address)),
|
||||
EnableServer: true,
|
||||
EnableDebuggingHandlers: true,
|
||||
SyncFrequency: 3 * time.Second,
|
||||
MinimumGCAge: 10 * time.Second,
|
||||
MaxContainerCount: 5,
|
||||
MasterServiceNamespace: masterServiceNamespace,
|
||||
VolumePlugins: volumePlugins,
|
||||
}
|
||||
RunKubelet(&kcfg)
|
||||
}
|
||||
|
||||
// RunKubelet is responsible for setting up and running a kubelet. It is used in three different applications:
|
||||
// 1 Integration tests
|
||||
// 2 Kubelet binary
|
||||
// 3 Standalone 'kubernetes' binary
|
||||
// Eventually, #2 will be replaced with instances of #3
|
||||
func RunKubelet(kcfg *KubeletConfig) {
|
||||
kcfg.Hostname = util.GetHostname(kcfg.HostnameOverride)
|
||||
if kcfg.KubeClient != nil {
|
||||
kubelet.SetupEventSending(kcfg.KubeClient, kcfg.Hostname)
|
||||
} else {
|
||||
glog.Infof("No api server defined - no events will be sent.")
|
||||
}
|
||||
kubelet.SetupLogging()
|
||||
kubelet.SetupCapabilities(kcfg.AllowPrivileged)
|
||||
|
||||
credentialprovider.SetPreferredDockercfgPath(kcfg.RootDirectory)
|
||||
|
||||
cfg := makePodSourceConfig(kcfg)
|
||||
k, err := createAndInitKubelet(kcfg, cfg)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to create kubelet: %s", err)
|
||||
return
|
||||
}
|
||||
// process pods and exit.
|
||||
if kcfg.Runonce {
|
||||
if _, err := k.RunOnce(cfg.Updates()); err != nil {
|
||||
glog.Errorf("--runonce failed: %v", err)
|
||||
}
|
||||
} else {
|
||||
startKubelet(k, cfg, kcfg)
|
||||
}
|
||||
}
|
||||
|
||||
func startKubelet(k *kubelet.Kubelet, cfg *config.PodConfig, kc *KubeletConfig) {
|
||||
// start the kubelet
|
||||
go util.Forever(func() { k.Run(cfg.Updates()) }, 0)
|
||||
|
||||
// start the kubelet server
|
||||
if kc.EnableServer {
|
||||
go util.Forever(func() {
|
||||
kubelet.ListenAndServeKubeletServer(k, net.IP(kc.Address), kc.Port, kc.EnableDebuggingHandlers)
|
||||
}, 0)
|
||||
}
|
||||
}
|
||||
|
||||
func makePodSourceConfig(kc *KubeletConfig) *config.PodConfig {
|
||||
// source of all configuration
|
||||
cfg := config.NewPodConfig(config.PodConfigNotificationSnapshotAndUpdates)
|
||||
|
||||
// define file config source
|
||||
if kc.ConfigFile != "" {
|
||||
glog.Infof("Adding manifest file: %v", kc.ConfigFile)
|
||||
config.NewSourceFile(kc.ConfigFile, kc.FileCheckFrequency, cfg.Channel(kubelet.FileSource))
|
||||
}
|
||||
|
||||
// define url config source
|
||||
if kc.ManifestURL != "" {
|
||||
glog.Infof("Adding manifest url: %v", kc.ManifestURL)
|
||||
config.NewSourceURL(kc.ManifestURL, kc.HttpCheckFrequency, cfg.Channel(kubelet.HTTPSource))
|
||||
}
|
||||
if kc.EtcdClient != nil {
|
||||
glog.Infof("Watching for etcd configs at %v", kc.EtcdClient.GetCluster())
|
||||
config.NewSourceEtcd(config.EtcdKeyForHost(kc.Hostname), kc.EtcdClient, cfg.Channel(kubelet.EtcdSource))
|
||||
}
|
||||
if kc.KubeClient != nil {
|
||||
glog.Infof("Watching apiserver")
|
||||
config.NewSourceApiserver(kc.KubeClient, kc.Hostname, cfg.Channel(kubelet.ApiserverSource))
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
type KubeletConfig struct {
|
||||
EtcdClient tools.EtcdClient
|
||||
KubeClient *client.Client
|
||||
DockerClient dockertools.DockerInterface
|
||||
CAdvisorPort uint
|
||||
Address util.IP
|
||||
AllowPrivileged bool
|
||||
HostnameOverride string
|
||||
RootDirectory string
|
||||
ConfigFile string
|
||||
ManifestURL string
|
||||
FileCheckFrequency time.Duration
|
||||
HttpCheckFrequency time.Duration
|
||||
Hostname string
|
||||
PodInfraContainerImage string
|
||||
SyncFrequency time.Duration
|
||||
RegistryPullQPS float64
|
||||
RegistryBurst int
|
||||
MinimumGCAge time.Duration
|
||||
MaxContainerCount int
|
||||
ClusterDomain string
|
||||
ClusterDNS util.IP
|
||||
EnableServer bool
|
||||
EnableDebuggingHandlers bool
|
||||
Port uint
|
||||
Runonce bool
|
||||
MasterServiceNamespace string
|
||||
VolumePlugins []volume.Plugin
|
||||
}
|
||||
|
||||
func createAndInitKubelet(kc *KubeletConfig, pc *config.PodConfig) (*kubelet.Kubelet, 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
|
||||
|
||||
k, err := kubelet.NewMainKubelet(
|
||||
kc.Hostname,
|
||||
kc.DockerClient,
|
||||
kc.EtcdClient,
|
||||
kc.KubeClient,
|
||||
kc.RootDirectory,
|
||||
kc.PodInfraContainerImage,
|
||||
kc.SyncFrequency,
|
||||
float32(kc.RegistryPullQPS),
|
||||
kc.RegistryBurst,
|
||||
kc.MinimumGCAge,
|
||||
kc.MaxContainerCount,
|
||||
pc.IsSourceSeen,
|
||||
kc.ClusterDomain,
|
||||
net.IP(kc.ClusterDNS),
|
||||
kc.MasterServiceNamespace,
|
||||
kc.VolumePlugins)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
k.BirthCry()
|
||||
|
||||
go k.GarbageCollectLoop()
|
||||
go kubelet.MonitorCAdvisor(k, kc.CAdvisorPort)
|
||||
|
||||
return k, nil
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user