Sync node status from node controller to master.

cmd/integration/integration.go

@@ -184,8 +184,8 @@ func startComponents(manifestURL string) (apiServerURL string) {
 	controllerManager.Run(10 * time.Minute)
 
 	nodeResources := &api.NodeResources{}
-	nodeController := nodeControllerPkg.NewNodeController(nil, "", machineList, nodeResources, cl)
-	nodeController.Run(10 * time.Second)
+	nodeController := nodeControllerPkg.NewNodeController(nil, "", machineList, nodeResources, cl, fakeKubeletClient{})
+	nodeController.Run(10*time.Second, 10)
 
 	// Kubelet (localhost)
 	testRootDir := makeTempDirOrDie("kubelet_integ_1.")

cmd/kube-apiserver/apiserver.go

@@ -184,7 +184,6 @@ func main() {
 		Client:                 client,
 		Cloud:                  cloud,
 		EtcdHelper:             helper,
-		HealthCheckMinions:     *healthCheckMinions,
 		EventTTL:               *eventTTL,
 		KubeletClient:          kubeletClient,
 		PortalNet:              &n,

cmd/kube-controller-manager/controller-manager.go

@@ -53,18 +53,22 @@ var (
 	nodeSyncPeriod  = flag.Duration("node_sync_period", 10*time.Second, ""+
 		"The period for syncing nodes from cloudprovider. Longer periods will result in "+
 		"fewer calls to cloud provider, but may delay addition of new nodes to cluster.")
+	resourceQuotaSyncPeriod = flag.Duration("resource_quota_sync_period", 10*time.Second, "The period for syncing quota usage status in the system")
+	registerRetryCount      = flag.Int("register_retry_count", 10, ""+
+		"The number of retries for initial node registration.  Retry interval equals node_sync_period.")
 	machineList util.StringList
 	// TODO: Discover these by pinging the host machines, and rip out these flags.
 	// TODO: in the meantime, use resource.QuantityFlag() instead of these
-	nodeMilliCPU            = flag.Int64("node_milli_cpu", 1000, "The amount of MilliCPU provisioned on each node")
-	nodeMemory              = resource.QuantityFlag("node_memory", "3Gi", "The amount of memory (in bytes) provisioned on each node")
-	resourceQuotaSyncPeriod = flag.Duration("resource_quota_sync_period", 10*time.Second, "The period for syncing quota usage status in the system")
+	nodeMilliCPU  = flag.Int64("node_milli_cpu", 1000, "The amount of MilliCPU provisioned on each node")
+	nodeMemory    = resource.QuantityFlag("node_memory", "3Gi", "The amount of memory (in bytes) provisioned on each node")
+	kubeletConfig = client.KubeletConfig{Port: ports.KubeletPort, EnableHttps: false}
 )
 
 func init() {
 	flag.Var(&address, "address", "The IP address to serve on (set to 0.0.0.0 for all interfaces)")
 	flag.Var(&machineList, "machines", "List of machines to schedule onto, comma separated.")
 	client.BindClientConfigFlags(flag.CommandLine, clientConfig)
+	client.BindKubeletClientConfigFlags(flag.CommandLine, &kubeletConfig)
 }
 
 func verifyMinionFlags() {
@@ -104,6 +108,10 @@ func main() {
 	controllerManager := replicationControllerPkg.NewReplicationManager(kubeClient)
 	controllerManager.Run(10 * time.Second)
 
+	kubeletClient, err := client.NewKubeletClient(&kubeletConfig)
+	if err != nil {
+		glog.Fatalf("Failure to start kubelet client: %v", err)
+	}
 	cloud := cloudprovider.InitCloudProvider(*cloudProvider, *cloudConfigFile)
 	nodeResources := &api.NodeResources{
 		Capacity: api.ResourceList{
@@ -111,8 +119,8 @@ func main() {
 			api.ResourceMemory: *nodeMemory,
 		},
 	}
-	nodeController := nodeControllerPkg.NewNodeController(cloud, *minionRegexp, machineList, nodeResources, kubeClient)
-	nodeController.Run(*nodeSyncPeriod)
+	nodeController := nodeControllerPkg.NewNodeController(cloud, *minionRegexp, machineList, nodeResources, kubeClient, kubeletClient)
+	nodeController.Run(*nodeSyncPeriod, *registerRetryCount)
 
 	resourceQuotaManager := resourcequota.NewResourceQuotaManager(kubeClient)
 	resourceQuotaManager.Run(*resourceQuotaSyncPeriod)

pkg/api/validation/validation.go

@@ -673,9 +673,11 @@ func ValidateMinionUpdate(oldMinion *api.Node, minion *api.Node) errs.Validation
 	allErrs := errs.ValidationErrorList{}
 	allErrs = append(allErrs, ValidateObjectMetaUpdate(&oldMinion.ObjectMeta, &minion.ObjectMeta).Prefix("metadata")...)
 
-	if !api.Semantic.DeepEqual(minion.Status, api.NodeStatus{}) {
-		allErrs = append(allErrs, errs.NewFieldInvalid("status", minion.Status, "status must be empty"))
-	}
+	// TODO: Enable the code once we have better api object.status update model. Currently,
+	// anyone can update node status.
+	// if !api.Semantic.DeepEqual(minion.Status, api.NodeStatus{}) {
+	// 	allErrs = append(allErrs, errs.NewFieldInvalid("status", minion.Status, "status must be empty"))
+	// }
 
 	// TODO: move reset function to its own location
 	// Ignore metadata changes now that they have been tested

pkg/api/validation/validation_test.go

@@ -1481,20 +1481,6 @@ func TestValidateMinionUpdate(t *testing.T) {
 				Name:   "foo",
 				Labels: map[string]string{"bar": "foo"},
 			},
-		}, api.Node{
-			ObjectMeta: api.ObjectMeta{
-				Name:   "foo",
-				Labels: map[string]string{"bar": "fooobaz"},
-			},
-			Status: api.NodeStatus{
-				HostIP: "1.2.3.4",
-			},
-		}, false},
-		{api.Node{
-			ObjectMeta: api.ObjectMeta{
-				Name:   "foo",
-				Labels: map[string]string{"bar": "foo"},
-			},
 			Status: api.NodeStatus{
 				HostIP: "1.2.3.4",
 			},

pkg/master/master.go

@@ -68,7 +68,6 @@ type Config struct {
 	Client                 *client.Client
 	Cloud                  cloudprovider.Interface
 	EtcdHelper             tools.EtcdHelper
-	HealthCheckMinions     bool
 	EventTTL               time.Duration
 	MinionRegexp           string
 	KubeletClient          client.KubeletClient
@@ -235,7 +234,7 @@ func setDefaults(c *Config) {
 //   any unhandled paths to "Handler".
 func New(c *Config) *Master {
 	setDefaults(c)
-	minionRegistry := makeMinionRegistry(c)
+	minionRegistry := etcd.NewRegistry(c.EtcdHelper, nil)
 	serviceRegistry := etcd.NewRegistry(c.EtcdHelper, nil)
 	boundPodFactory := &pod.BasicBoundPodFactory{
 		ServiceRegistry:        serviceRegistry,
@@ -330,15 +329,6 @@ func logStackOnRecover(panicReason interface{}, httpWriter http.ResponseWriter)
 	glog.Errorln(buffer.String())
 }
 
-func makeMinionRegistry(c *Config) minion.Registry {
-	var minionRegistry minion.Registry = etcd.NewRegistry(c.EtcdHelper, nil)
-	// TODO: plumb in nodeIPCache here
-	if c.HealthCheckMinions {
-		minionRegistry = minion.NewHealthyRegistry(minionRegistry, c.KubeletClient, util.RealClock{}, 20*time.Second)
-	}
-	return minionRegistry
-}
-
 // init initializes master.
 func (m *Master) init(c *Config) {
 	var userContexts = handlers.NewUserRequestContext()

pkg/probe/http/http.go

@@ -45,8 +45,8 @@ type HTTPGetInterface interface {
 }
 
 // DoHTTPProbe checks if a GET request to the url succeeds.
-// If the HTTP response code is successful (i.e. 400 > code >= 200), it returns Healthy.
-// If the HTTP response code is unsuccessful or HTTP communication fails, it returns Unhealthy.
+// If the HTTP response code is successful (i.e. 400 > code >= 200), it returns Success.
+// If the HTTP response code is unsuccessful or HTTP communication fails, it returns Failure.
 // This is exported because some other packages may want to do direct HTTP probes.
 func DoHTTPProbe(url string, client HTTPGetInterface) (probe.Status, error) {
 	res, err := client.Get(url)

pkg/probe/tcp/tcp.go

@@ -36,8 +36,8 @@ func (pr TCPProber) Probe(host string, port int) (probe.Status, error) {
 }
 
 // DoTCPProbe checks that a TCP socket to the address can be opened.
-// If the socket can be opened, it returns Healthy.
-// If the socket fails to open, it returns Unhealthy.
+// If the socket can be opened, it returns Success
+// If the socket fails to open, it returns Failure.
 // This is exported because some other packages may want to do direct TCP probes.
 func DoTCPProbe(addr string) (probe.Status, error) {
 	conn, err := net.Dial("tcp", addr)

pkg/registry/minion/healthy_registry.go

-/*
-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 minion
-
-import (
-	"sync"
-	"time"
-
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/probe"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
-
-	"github.com/golang/glog"
-)
-
-type HealthyRegistry struct {
-	delegate Registry
-	client   client.KubeletHealthChecker
-	cache    util.TimeCache
-}
-
-func NewHealthyRegistry(delegate Registry, client client.KubeletHealthChecker, clock util.Clock, ttl time.Duration) Registry {
-	h := &HealthyRegistry{
-		delegate: delegate,
-		client:   client,
-	}
-	h.cache = util.NewTimeCache(clock, ttl, h.doCheck)
-	return h
-}
-
-func (r *HealthyRegistry) GetMinion(ctx api.Context, minionID string) (*api.Node, error) {
-	minion, err := r.delegate.GetMinion(ctx, minionID)
-	if err != nil {
-		return nil, err
-	}
-	return r.checkMinion(minion), nil
-}
-
-func (r *HealthyRegistry) DeleteMinion(ctx api.Context, minionID string) error {
-	return r.delegate.DeleteMinion(ctx, minionID)
-}
-
-func (r *HealthyRegistry) CreateMinion(ctx api.Context, minion *api.Node) error {
-	return r.delegate.CreateMinion(ctx, minion)
-}
-
-func (r *HealthyRegistry) UpdateMinion(ctx api.Context, minion *api.Node) error {
-	return r.delegate.UpdateMinion(ctx, minion)
-}
-
-func (r *HealthyRegistry) ListMinions(ctx api.Context) (currentMinions *api.NodeList, err error) {
-	list, err := r.delegate.ListMinions(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	// In case the cache is empty, health check in parallel instead of serially.
-	var wg sync.WaitGroup
-	wg.Add(len(list.Items))
-	for i := range list.Items {
-		go func(i int) {
-			list.Items[i] = *r.checkMinion(&list.Items[i])
-			wg.Done()
-		}(i)
-	}
-	wg.Wait()
-	return list, nil
-}
-
-func (r *HealthyRegistry) WatchMinions(ctx api.Context, label, field labels.Selector, resourceVersion string) (watch.Interface, error) {
-	w, err := r.delegate.WatchMinions(ctx, label, field, resourceVersion)
-	if err != nil {
-		return nil, err
-	}
-	return watch.Filter(w, watch.FilterFunc(func(in watch.Event) (watch.Event, bool) {
-		if node, ok := in.Object.(*api.Node); ok && node != nil {
-			in.Object = r.checkMinion(node)
-		}
-		return in, true
-	})), nil
-}
-
-func (r *HealthyRegistry) checkMinion(node *api.Node) *api.Node {
-	condition := r.cache.Get(node.Name).(api.NodeConditionStatus)
-	// TODO: distinguish other conditions like Reachable/Live, and begin storing this
-	// data on nodes directly via sync loops.
-	node.Status.Conditions = append(node.Status.Conditions, api.NodeCondition{
-		Kind:   api.NodeReady,
-		Status: condition,
-	})
-	return node
-}
-
-// This is called to fill the cache.
-func (r *HealthyRegistry) doCheck(key string) util.T {
-	var nodeStatus api.NodeConditionStatus
-	switch status, err := r.client.HealthCheck(key); {
-	case err != nil:
-		glog.V(2).Infof("HealthyRegistry: node %q health check error: %v", key, err)
-		nodeStatus = api.ConditionUnknown
-	case status == probe.Failure:
-		nodeStatus = api.ConditionNone
-	default:
-		nodeStatus = api.ConditionFull
-	}
-	glog.V(3).Infof("HealthyRegistry: node %q status was %q", key, nodeStatus)
-	return nodeStatus
-}

pkg/registry/minion/healthy_registry_test.go

-/*
-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 minion
-
-import (
-	"reflect"
-	"testing"
-	"time"
-
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/probe"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/registrytest"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
-)
-
-type alwaysYes struct{}
-
-func (alwaysYes) HealthCheck(host string) (probe.Status, error) {
-	return probe.Success, nil
-}
-
-func TestBasicDelegation(t *testing.T) {
-	ctx := api.NewContext()
-	mockMinionRegistry := registrytest.NewMinionRegistry([]string{"m1", "m2", "m3"}, api.NodeResources{})
-	healthy := NewHealthyRegistry(
-		mockMinionRegistry,
-		alwaysYes{},
-		&util.FakeClock{},
-		60*time.Second,
-	)
-	list, err := healthy.ListMinions(ctx)
-	if err != nil {
-		t.Errorf("unexpected error: %v", err)
-	}
-	if !reflect.DeepEqual(list, &mockMinionRegistry.Minions) {
-		t.Errorf("Expected %v, Got %v", mockMinionRegistry.Minions, list)
-	}
-	err = healthy.CreateMinion(ctx, &api.Node{
-		ObjectMeta: api.ObjectMeta{Name: "foo"},
-	})
-	if err != nil {
-		t.Errorf("unexpected error: %v", err)
-	}
-	minion, err := healthy.GetMinion(ctx, "m1")
-	if err != nil {
-		t.Errorf("unexpected error: %v", err)
-	}
-	if minion == nil {
-		t.Errorf("Unexpected absence of 'm1'")
-	}
-	minion, err = healthy.GetMinion(ctx, "m5")
-	if err == nil {
-		t.Errorf("unexpected non-error")
-	}
-	if minion != nil {
-		t.Errorf("Unexpected presence of 'm5'")
-	}
-}
-
-type notMinion struct {
-	minion string
-}
-
-func (n *notMinion) HealthCheck(host string) (probe.Status, error) {
-	if host != n.minion {
-		return probe.Success, nil
-	} else {
-		return probe.Failure, nil
-	}
-}
-
-func TestFiltering(t *testing.T) {
-	ctx := api.NewContext()
-	mockMinionRegistry := registrytest.NewMinionRegistry([]string{"m1", "m2", "m3"}, api.NodeResources{})
-	healthy := NewHealthyRegistry(
-		mockMinionRegistry,
-		&notMinion{minion: "m1"},
-		&util.FakeClock{},
-		60*time.Second,
-	)
-	expected := []string{"m1", "m2", "m3"}
-	list, err := healthy.ListMinions(ctx)
-	if err != nil {
-		t.Errorf("unexpected error: %v", err)
-	}
-	expectedMinions := registrytest.MakeMinionList(expected, api.NodeResources{})
-	expectedMinions.Items[0].Status.Conditions = []api.NodeCondition{{Kind: api.NodeReady, Status: api.ConditionNone}}
-	expectedMinions.Items[1].Status.Conditions = []api.NodeCondition{{Kind: api.NodeReady, Status: api.ConditionFull}}
-	expectedMinions.Items[2].Status.Conditions = []api.NodeCondition{{Kind: api.NodeReady, Status: api.ConditionFull}}
-	if !reflect.DeepEqual(list, expectedMinions) {
-		t.Errorf("Expected %v, Got %v", expected, list)
-	}
-	minion, err := healthy.GetMinion(ctx, "m1")
-	if err != nil {
-		t.Errorf("unexpected error: %v", err)
-	}
-	if minion == nil {
-		t.Errorf("Unexpected empty 'm1'")
-	}
-}

pkg/registry/minion/rest_test.go

@@ -18,13 +18,11 @@ package minion
 
 import (
 	"testing"
-	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/registrytest"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 )
 
 func TestMinionRegistryREST(t *testing.T) {
@@ -89,57 +87,6 @@ func TestMinionRegistryREST(t *testing.T) {
 	}
 }
 
-func TestMinionRegistryHealthCheck(t *testing.T) {
-	minionRegistry := registrytest.NewMinionRegistry([]string{}, api.NodeResources{})
-	minionHealthRegistry := NewHealthyRegistry(
-		minionRegistry,
-		&notMinion{minion: "m1"},
-		&util.FakeClock{},
-		60*time.Second,
-	)
-
-	ms := NewREST(minionHealthRegistry)
-	ctx := api.NewContext()
-
-	c, err := ms.Create(ctx, &api.Node{ObjectMeta: api.ObjectMeta{Name: "m1"}})
-	if err != nil {
-		t.Fatalf("insert failed: %v", err)
-	}
-	result := <-c
-	if m, ok := result.Object.(*api.Node); !ok || m.Name != "m1" {
-		t.Errorf("insert return value was weird: %#v", result)
-	}
-	if _, err := ms.Get(ctx, "m1"); err != nil {
-		t.Errorf("node is unhealthy, expect no error: %v", err)
-	}
-}
-
-func contains(nodes *api.NodeList, nodeID string) bool {
-	for _, node := range nodes.Items {
-		if node.Name == nodeID {
-			return true
-		}
-	}
-	return false
-}
-
-func TestMinionRegistryInvalidUpdate(t *testing.T) {
-	storage := NewREST(registrytest.NewMinionRegistry([]string{"foo", "bar"}, api.NodeResources{}))
-	ctx := api.NewContext()
-	obj, err := storage.Get(ctx, "foo")
-	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
-	}
-	minion, ok := obj.(*api.Node)
-	if !ok {
-		t.Fatalf("Object is not a minion: %#v", obj)
-	}
-	minion.Status.HostIP = "1.2.3.4"
-	if _, err = storage.Update(ctx, minion); err == nil {
-		t.Error("Unexpected non-error.")
-	}
-}
-
 func TestMinionRegistryValidUpdate(t *testing.T) {
 	storage := NewREST(registrytest.NewMinionRegistry([]string{"foo", "bar"}, api.NodeResources{}))
 	ctx := api.NewContext()
@@ -192,3 +139,12 @@ func TestMinionRegistryValidatesCreate(t *testing.T) {
 		}
 	}
 }
+
+func contains(nodes *api.NodeList, nodeID string) bool {
+	for _, node := range nodes.Items {
+		if node.Name == nodeID {
+			return true
+		}
+	}
+	return false
+}

pkg/standalone/standalone.go

@@ -36,6 +36,7 @@ import (
 	"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"
@@ -134,8 +135,9 @@ func RunControllerManager(machineList []string, cl *client.Client, nodeMilliCPU,
 			api.ResourceMemory: *resource.NewQuantity(nodeMemory, resource.BinarySI),
 		},
 	}
-	nodeController := nodeControllerPkg.NewNodeController(nil, "", machineList, nodeResources, cl)
-	nodeController.Run(10 * time.Second)
+	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)

plugin/pkg/scheduler/factory/factory.go

@@ -205,9 +205,8 @@ func (factory *ConfigFactory) pollMinions() (cache.Enumerator, error) {
 				nodes.Items = append(nodes.Items, node)
 			}
 		} else {
-			// If no condition is set, either node health check is disabled (master
-			// flag "healthCheckMinions" is set to false), or we get unknown condition.
-			// In such cases, we add nodes unconditionally.
+			// If no condition is set, we get unknown node condition. In such cases,
+			// we add nodes unconditionally.
 			nodes.Items = append(nodes.Items, node)
 		}
 	}

test/integration/auth_test.go

@@ -203,7 +203,7 @@ func getTestRequests() []struct {
 		// Normal methods on services
 		{"GET", "/api/v1beta1/services", "", code200},
 		{"POST", "/api/v1beta1/services" + timeoutFlag, aService, code200},
-		{"PUT", "/api/v1beta1/services/a" + timeoutFlag, aService, code409}, // TODO: GET and put back server-provided fields to avoid a 422
+		{"PUT", "/api/v1beta1/services/a" + timeoutFlag, aService, code409}, // See #2115 about why 409
 		{"GET", "/api/v1beta1/services", "", code200},
 		{"GET", "/api/v1beta1/services/a", "", code200},
 		{"DELETE", "/api/v1beta1/services/a" + timeoutFlag, "", code200},
@@ -227,7 +227,7 @@ func getTestRequests() []struct {
 		// Normal methods on minions
 		{"GET", "/api/v1beta1/minions", "", code200},
 		{"POST", "/api/v1beta1/minions" + timeoutFlag, aMinion, code200},
-		{"PUT", "/api/v1beta1/minions/a" + timeoutFlag, aMinion, code422}, // TODO: GET and put back server-provided fields to avoid a 422
+		{"PUT", "/api/v1beta1/minions/a" + timeoutFlag, aMinion, code409}, // See #2115 about why 409
 		{"GET", "/api/v1beta1/minions", "", code200},
 		{"GET", "/api/v1beta1/minions/a", "", code200},
 		{"DELETE", "/api/v1beta1/minions/a" + timeoutFlag, "", code200},