Merge pull request #2121 from brendandburns/standalone

Create a standalone k8s binary, capable of running a full cluster

cmd/kubernetes/kubernetes.go

@@ -0,0 +1,78 @@
+/*
+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.
+*/
+
+// A binary that is capable of running a complete, standalone kubernetes cluster.
+// Expects an etcd server is available, or on the path somewhere.
+// Does *not* currently setup the Kubernetes network model, that must be done ahead of time.
+// TODO: Setup the k8s network bridge as part of setup.
+package main
+
+import (
+	"flag"
+	"fmt"
+	"time"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/testapi"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/standalone"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+
+	"github.com/golang/glog"
+)
+
+var (
+	addr           = flag.String("addr", "127.0.0.1", "The address to use for the apiserver.")
+	port           = flag.Int("port", 8080, "The port for the apiserver to use.")
+	dockerEndpoint = flag.String("docker_endpoint", "", "If non-empty, use this for the docker endpoint to communicate with")
+	etcdServer     = flag.String("etcd_server", "http://localhost:4001", "If non-empty, path to the set of etcd server to use")
+	// TODO: Discover these by pinging the host machines, and rip out these flags.
+	nodeMilliCPU = flag.Int("node_milli_cpu", 1000, "The amount of MilliCPU provisioned on each node")
+	nodeMemory   = flag.Int("node_memory", 3*1024*1024*1024, "The amount of memory (in bytes) provisioned on each node")
+)
+
+func startComponents(etcdClient tools.EtcdClient, cl *client.Client, addr string, port int) {
+	machineList := []string{"localhost"}
+
+	standalone.RunApiServer(cl, etcdClient, addr, port)
+	standalone.RunScheduler(cl)
+	standalone.RunControllerManager(machineList, cl, *nodeMilliCPU, *nodeMemory)
+	standalone.RunKubelet(etcdClient, machineList[0], *dockerEndpoint)
+}
+
+func newApiClient(addr string, port int) *client.Client {
+	apiServerURL := fmt.Sprintf("http://%s:%d", addr, port)
+	cl := client.NewOrDie(&client.Config{Host: apiServerURL, Version: testapi.Version()})
+	cl.PollPeriod = time.Second * 1
+	cl.Sync = true
+	return cl
+}
+
+func main() {
+	flag.Parse()
+	util.InitLogs()
+	defer util.FlushLogs()
+
+	glog.Infof("Creating etcd client pointing to %v", *etcdServer)
+	etcdClient, err := tools.NewEtcdClientStartServerIfNecessary(*etcdServer)
+	if err != nil {
+		glog.Fatalf("Failed to connect to etcd: %v", err)
+	}
+	startComponents(etcdClient, newApiClient(*addr, *port), *addr, *port)
+	glog.Infof("Kubernetes API Server is up and running on http://%s:%d", *addr, *port)
+
+	select {}
+}

hack/lib/golang.sh

@@ -37,6 +37,7 @@ readonly KUBE_SERVER_PLATFORMS=(
 readonly KUBE_CLIENT_TARGETS=(
   cmd/kubecfg
   cmd/kubectl
+  cmd/kubernetes
 )
 readonly KUBE_CLIENT_BINARIES=("${KUBE_CLIENT_TARGETS[@]##*/}")
 

pkg/controller/replication_controller.go

@@ -126,7 +126,9 @@ func (rm *ReplicationManager) watchControllers(resourceVersion *string) {
 			// Sync even if this is a deletion event, to ensure that we leave
 			// it in the desired state.
 			glog.V(4).Infof("About to sync from watch: %v", rc.Name)
-			rm.syncHandler(*rc)
+			if err := rm.syncHandler(*rc); err != nil {
+				glog.Errorf("unexpected sync. error: %v", err)
+			}
 		}
 	}
 }

pkg/standalone/doc.go

@@ -0,0 +1,18 @@
+/*
+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

pkg/standalone/standalone.go

@@ -0,0 +1,150 @@
+/*
+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"
+	"os"
+	"time"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/apiserver"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
+	minionControllerPkg "github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider/controller"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/controller"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/config"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/master"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/resources"
+	"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/factory"
+
+	"github.com/fsouza/go-dockerclient"
+	"github.com/golang/glog"
+)
+
+const testRootDir = "/tmp/kubelet"
+
+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)
+}
+
+// Get a docker endpoint, either from the string passed in, or $DOCKER_HOST environment variables
+func GetDockerEndpoint(dockerEndpoint string) string {
+	var endpoint string
+	if len(dockerEndpoint) > 0 {
+		endpoint = dockerEndpoint
+	} else if len(os.Getenv("DOCKER_HOST")) > 0 {
+		endpoint = os.Getenv("DOCKER_HOST")
+	} else {
+		endpoint = "unix:///var/run/docker.sock"
+	}
+	glog.Infof("Connecting to docker on %s", endpoint)
+
+	return endpoint
+}
+
+// RunApiServer starts an API server in a go routine.
+func RunApiServer(cl *client.Client, etcdClient tools.EtcdClient, addr string, port int) {
+	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,
+		APIPrefix:         "/api",
+		Authorizer:        apiserver.NewAlwaysAllowAuthorizer(),
+
+		ReadWritePort: port,
+		ReadOnlyPort:  port,
+		PublicAddress: addr,
+	})
+	mux := http.NewServeMux()
+	apiserver.NewAPIGroup(m.API_v1beta1()).InstallREST(mux, "/api/v1beta1")
+	apiserver.NewAPIGroup(m.API_v1beta2()).InstallREST(mux, "/api/v1beta2")
+	apiserver.InstallSupport(mux)
+	handler.delegate = mux
+
+	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.ConfigFactory{cl}
+	schedulerConfig := schedulerConfigFactory.Create()
+	scheduler.New(schedulerConfig).Run()
+}
+
+// RunControllerManager starts a controller
+func RunControllerManager(machineList []string, cl *client.Client, nodeMilliCPU, nodeMemory int) {
+	nodeResources := &api.NodeResources{
+		Capacity: api.ResourceList{
+			resources.CPU:    util.NewIntOrStringFromInt(nodeMilliCPU),
+			resources.Memory: util.NewIntOrStringFromInt(nodeMemory),
+		},
+	}
+	minionController := minionControllerPkg.NewMinionController(nil, "", machineList, nodeResources, cl)
+	minionController.Run(10 * time.Second)
+
+	endpoints := service.NewEndpointController(cl)
+	go util.Forever(func() { endpoints.SyncServiceEndpoints() }, time.Second*10)
+
+	controllerManager := controller.NewReplicationManager(cl)
+	controllerManager.Run(10 * time.Second)
+}
+
+// RunKubelet starts a Kubelet talking to dockerEndpoint
+func RunKubelet(etcdClient tools.EtcdClient, hostname, dockerEndpoint string) {
+	dockerClient, err := docker.NewClient(GetDockerEndpoint(dockerEndpoint))
+	if err != nil {
+		glog.Fatal("Couldn't connect to docker.")
+	}
+
+	// Kubelet (localhost)
+	os.MkdirAll(testRootDir, 0750)
+	cfg1 := config.NewPodConfig(config.PodConfigNotificationSnapshotAndUpdates)
+	config.NewSourceEtcd(config.EtcdKeyForHost(hostname), etcdClient, cfg1.Channel("etcd"))
+	myKubelet := kubelet.NewIntegrationTestKubelet(hostname, testRootDir, dockerClient)
+	go util.Forever(func() { myKubelet.Run(cfg1.Updates()) }, 0)
+	go util.Forever(func() {
+		kubelet.ListenAndServeKubeletServer(myKubelet, cfg1.Channel("http"), net.ParseIP("127.0.0.1"), 10250, true)
+	}, 0)
+}

pkg/tools/etcd_tools.go

@@ -19,12 +19,18 @@ package tools
 import (
 	"errors"
 	"fmt"
+	"io/ioutil"
+	"net/http"
+	"os/exec"
 	"reflect"
 	"strconv"
+	"strings"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/conversion"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/coreos/go-etcd/etcd"
+
+	"github.com/golang/glog"
 )
 
 const (
@@ -360,3 +366,42 @@ func (h *EtcdHelper) AtomicUpdate(key string, ptrToType runtime.Object, tryUpdat
 		return err
 	}
 }
+
+func checkEtcd(host string) error {
+	response, err := http.Get(host + "/version")
+	if err != nil {
+		return err
+	}
+	defer response.Body.Close()
+	body, err := ioutil.ReadAll(response.Body)
+	if err != nil {
+		return err
+	}
+	if !strings.HasPrefix("etcd", string(body)) {
+		return fmt.Errorf("Unknown server: %s", string(body))
+	}
+	return nil
+}
+
+func startEtcd() (*exec.Cmd, error) {
+	cmd := exec.Command("etcd")
+	err := cmd.Start()
+	if err != nil {
+		return nil, err
+	}
+	return cmd, nil
+}
+
+func NewEtcdClientStartServerIfNecessary(server string) (EtcdClient, error) {
+	err := checkEtcd(server)
+	if err != nil {
+		glog.Infof("Failed to find etcd, attempting to start.")
+		_, err := startEtcd()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	servers := []string{server}
+	return etcd.NewClient(servers), nil
+}

plugin/pkg/scheduler/scheduler.go

@@ -22,6 +22,8 @@ import (
 	// TODO: move everything from pkg/scheduler into this package. Remove references from registry.
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/scheduler"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+
+	"github.com/golang/glog"
 )
 
 // Binder knows how to write a binding.
@@ -66,8 +68,10 @@ func (s *Scheduler) Run() {
 
 func (s *Scheduler) scheduleOne() {
 	pod := s.config.NextPod()
+	glog.V(3).Infof("Attempting to schedule: %v", pod)
 	dest, err := s.config.Algorithm.Schedule(*pod, s.config.MinionLister)
 	if err != nil {
+		glog.V(1).Infof("Failed to schedule: %v", pod)
 		record.Eventf(pod, string(api.PodPending), "failedScheduling", "Error scheduling: %v", err)
 		s.config.Error(pod, err)
 		return
@@ -78,6 +82,7 @@ func (s *Scheduler) scheduleOne() {
 		Host:       dest,
 	}
 	if err := s.config.Binder.Bind(b); err != nil {
+		glog.V(1).Infof("Failed to bind pod: %v", err)
 		record.Eventf(pod, string(api.PodPending), "failedScheduling", "Binding rejected: %v", err)
 		s.config.Error(pod, err)
 		return