Merge pull request #11 from Random-Liu/use-patch-instead-of-update-status

Use patch instead of update status

Author: dchen1107

Godeps/Godeps.json

@@ -1,7 +1,7 @@
 all: push
 
 # See pod.yaml for the version currently running-- bump this ahead before rebuilding!
-TAG = 0.1
+TAG = v0.1
 
 PROJ = google_containers
 

Makefile

@@ -1,6 +1,6 @@
 {
 	"logPath": "/log/kern.log",
-	"bufferSize": 50,
+	"bufferSize": 10,
 	"rules": [
 		{
 			"type": "temporary",

config/kernel-monitor.json

@@ -14,7 +14,7 @@ spec:
         command:
         - /node-problem-detector
         - --kernel-monitor=/config/kernel-monitor.json
-        image: gcr.io/google_containers/node-problem-detector:0.1
+        image: gcr.io/google_containers/node-problem-detector:v0.1
         imagePullPolicy: Always
         env:
         # Config the host ip and port of apiserver.

node-problem-detector.yaml

@@ -125,7 +125,7 @@ func (c *conditionManager) sync() {
 	for i := range c.conditions {
 		conditions = append(conditions, problemutil.ConvertToAPICondition(c.conditions[i]))
 	}
-	if err := c.client.SetConditions(conditions, updateTimeout); err != nil {
+	if err := c.client.SetConditions(conditions); err != nil {
 		// The conditions will be updated again in future sync
 		glog.Errorf("failed to update node conditions: %v", err)
 		return

pkg/condition/manager.go

@@ -195,6 +195,8 @@ func defaultCondition() types.Condition {
 		Type:       KernelDeadlockCondition,
 		Status:     false,
 		Transition: time.Now(),
+		Reason:     "KernelHasNoDeadlock",
+		Message:    "kernel has no deadlock",
 	}
 }
 

pkg/kernelmonitor/kernel_monitor.go

@@ -17,38 +17,32 @@ limitations under the License.
 package problemclient
 
 import (
+	"encoding/json"
 	"fmt"
 	"os"
-	"time"
 
 	"k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/api/errors"
 	"k8s.io/kubernetes/pkg/api/unversioned"
-	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
-	unversionedcore "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/unversioned"
 	"k8s.io/kubernetes/pkg/client/record"
 	"k8s.io/kubernetes/pkg/client/restclient"
+	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/types"
 	"k8s.io/kubernetes/pkg/util"
-
-	"github.com/golang/glog"
 )
 
 // Client is the interface of problem client
 type Client interface {
 	// GetConditions get all specifiec conditions of current node.
 	GetConditions(conditionTypes []api.NodeConditionType) ([]*api.NodeCondition, error)
 	// SetConditions set or update conditions of current node.
-	// Notice that conditions with status api.ConditionFalse will be removed from the condition list, so that
-	// we'll only have useful conditions in the condition list.
-	SetConditions(conditions []api.NodeCondition, timeout time.Duration) error
+	SetConditions(conditions []api.NodeCondition) error
 	// Eventf reports the event.
 	Eventf(eventType string, source, reason, messageFmt string, args ...interface{})
 }
 
 type nodeProblemClient struct {
 	nodeName  string
-	client    clientset.Interface
+	client    *client.Client
 	clock     util.Clock
 	recorders map[string]record.EventRecorder
 	nodeRef   *api.ObjectReference
@@ -62,10 +56,7 @@ func NewClientOrDie() Client {
 		panic(err)
 	}
 	// TODO(random-liu): Set QPS Limit
-	c.client, err = clientset.NewForConfig(cfg)
-	if err != nil {
-		panic(err)
-	}
+	c.client = client.NewOrDie(cfg)
 	// TODO(random-liu): Get node name from cloud provider
 	c.nodeName, err = os.Hostname()
 	if err != nil {
@@ -77,7 +68,7 @@ func NewClientOrDie() Client {
 }
 
 func (c *nodeProblemClient) GetConditions(conditionTypes []api.NodeConditionType) ([]*api.NodeCondition, error) {
-	node, err := c.client.Core().Nodes().Get(c.nodeName)
+	node, err := c.client.Nodes().Get(c.nodeName)
 	if err != nil {
 		return nil, err
 	}
@@ -92,21 +83,16 @@ func (c *nodeProblemClient) GetConditions(conditionTypes []api.NodeConditionType
 	return conditions, nil
 }
 
-func (c *nodeProblemClient) SetConditions(newConditions []api.NodeCondition, timeout time.Duration) error {
+func (c *nodeProblemClient) SetConditions(newConditions []api.NodeCondition) error {
 	for i := range newConditions {
 		// Each time we update the conditions, we update the heart beat time
 		newConditions[i].LastHeartbeatTime = unversioned.NewTime(c.clock.Now())
 	}
-	return c.updateNodeCondition(func(conditions []api.NodeCondition) []api.NodeCondition {
-		for _, condition := range newConditions {
-			if condition.Status == api.ConditionFalse {
-				conditions = unsetCondition(condition.Type, conditions)
-			} else {
-				conditions = setCondition(condition, conditions)
-			}
-		}
-		return conditions
-	}, timeout)
+	patch, err := generatePatch(newConditions)
+	if err != nil {
+		return nil
+	}
+	return c.client.Patch(api.StrategicMergePatchType).Resource("nodes").Name(c.nodeName).SubResource("status").Body(patch).Do().Error()
 }
 
 func (c *nodeProblemClient) Eventf(eventType, source, reason, messageFmt string, args ...interface{}) {
@@ -119,60 +105,20 @@ func (c *nodeProblemClient) Eventf(eventType, source, reason, messageFmt string,
 	recorder.Eventf(c.nodeRef, eventType, reason, messageFmt, args...)
 }
 
-func unsetCondition(conditionType api.NodeConditionType, conditions []api.NodeCondition) []api.NodeCondition {
-	result := []api.NodeCondition{}
-	for _, condition := range conditions {
-		if condition.Type != conditionType {
-			result = append(result, condition)
-		}
-	}
-	return result
-}
-
-func setCondition(condition api.NodeCondition, conditions []api.NodeCondition) []api.NodeCondition {
-	found := false
-	for i := range conditions {
-		if conditions[i].Type == condition.Type {
-			target := &conditions[i]
-			*target = condition
-			found = true
-			break
-		}
-	}
-	if !found {
-		conditions = append(conditions, condition)
-	}
-	return conditions
-}
-
-func (c *nodeProblemClient) updateNodeCondition(updateFunc func([]api.NodeCondition) []api.NodeCondition, timeout time.Duration) error {
-	updateTime := c.clock.Now()
-	for {
-		node, err := c.client.Core().Nodes().Get(c.nodeName)
-		if err != nil {
-			return err
-		}
-		node.Status.Conditions = updateFunc(node.Status.Conditions)
-		_, err = c.client.Core().Nodes().UpdateStatus(node)
-		if err != nil {
-			if errors.IsConflict(err) {
-				glog.Warningf("Conflicting update node status for node %q, will retry soon: %v", c.nodeName, err)
-				if c.clock.Now().Sub(updateTime) >= timeout {
-					return timeoutError{node: c.nodeName, timeout: timeout}
-				}
-				continue
-			}
-			return err
-		}
-		return nil
+// generatePatch generates condition patch
+func generatePatch(conditions []api.NodeCondition) ([]byte, error) {
+	raw, err := json.Marshal(&conditions)
+	if err != nil {
+		return nil, err
 	}
+	return []byte(fmt.Sprintf(`{"status":{"conditions":%s}}`, raw)), nil
 }
 
 // getEventRecorder generates a recorder for specific node name and source.
-func getEventRecorder(c clientset.Interface, nodeName, source string) record.EventRecorder {
+func getEventRecorder(c *client.Client, nodeName, source string) record.EventRecorder {
 	eventBroadcaster := record.NewBroadcaster()
 	recorder := eventBroadcaster.NewRecorder(api.EventSource{Component: source, Host: nodeName})
-	eventBroadcaster.StartRecordingToSink(&unversionedcore.EventSinkImpl{Interface: c.Core().Events("")})
+	eventBroadcaster.StartRecordingToSink(c.Events(""))
 	return recorder
 }
 
@@ -184,19 +130,3 @@ func getNodeRef(nodeName string) *api.ObjectReference {
 		Namespace: "",
 	}
 }
-
-// timeoutError is the error returned by problem client when condition update timeout.
-type timeoutError struct {
-	node    string
-	timeout time.Duration
-}
-
-func (e timeoutError) Error() string {
-	return fmt.Sprintf("update condition for node %q timeout %s", e.node, e.timeout)
-}
-
-// IsErrTimeout checks whether a given error is timeout error.
-func IsErrTimeout(err error) bool {
-	_, ok := err.(timeoutError)
-	return ok
-}