diff --git a/Makefile b/Makefile index ca9d0cb3..e11f2ee9 100644 --- a/Makefile +++ b/Makefile @@ -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 diff --git a/config/kernel-monitor.json b/config/kernel-monitor.json index 9bb5527a..3d69ae20 100644 --- a/config/kernel-monitor.json +++ b/config/kernel-monitor.json @@ -1,6 +1,6 @@ { "logPath": "/log/kern.log", - "bufferSize": 50, + "bufferSize": 10, "rules": [ { "type": "temporary", diff --git a/node-problem-detector.yaml b/node-problem-detector.yaml index be97ae11..3f706529 100644 --- a/node-problem-detector.yaml +++ b/node-problem-detector.yaml @@ -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. diff --git a/pkg/condition/manager.go b/pkg/condition/manager.go index 80b08213..2846b9f8 100644 --- a/pkg/condition/manager.go +++ b/pkg/condition/manager.go @@ -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 diff --git a/pkg/kernelmonitor/kernel_monitor.go b/pkg/kernelmonitor/kernel_monitor.go index 3cd7756d..c749ba7e 100644 --- a/pkg/kernelmonitor/kernel_monitor.go +++ b/pkg/kernelmonitor/kernel_monitor.go @@ -195,6 +195,8 @@ func defaultCondition() types.Condition { Type: KernelDeadlockCondition, Status: false, Transition: time.Now(), + Reason: "KernelHasNoDeadlock", + Message: "kernel has no deadlock", } } diff --git a/pkg/problemclient/problem_client.go b/pkg/problemclient/problem_client.go index 59e3d1c1..943b1baf 100644 --- a/pkg/problemclient/problem_client.go +++ b/pkg/problemclient/problem_client.go @@ -17,21 +17,17 @@ 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 @@ -39,16 +35,14 @@ 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 -} diff --git a/pkg/problemclient/problem_client_test.go b/pkg/problemclient/problem_client_test.go index 5d51b17a..37a89e41 100644 --- a/pkg/problemclient/problem_client_test.go +++ b/pkg/problemclient/problem_client_test.go @@ -17,19 +17,17 @@ limitations under the License. package problemclient import ( + "encoding/json" "fmt" - "reflect" "testing" "time" "k8s.io/kubernetes/pkg/api" - "k8s.io/kubernetes/pkg/api/errors" "k8s.io/kubernetes/pkg/api/unversioned" - "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake" "k8s.io/kubernetes/pkg/client/record" - "k8s.io/kubernetes/pkg/client/testing/core" - "k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/util" + + "github.com/stretchr/testify/assert" ) const ( @@ -37,218 +35,49 @@ const ( testNode = "test-node" ) -func newFakeProblemClient(fakeClient *fake.Clientset) *nodeProblemClient { +func newFakeProblemClient() *nodeProblemClient { return &nodeProblemClient{ - nodeName: testNode, - client: fakeClient, + nodeName: testNode, + // There is no proper fake for *client.Client for now + // TODO(random-liu): Add test for SetConditions when we have good fake for *client.Client clock: &util.FakeClock{}, recorders: make(map[string]record.EventRecorder), nodeRef: getNodeRef(testNode), } } -func newFakeNode(conditions []api.NodeCondition) *api.Node { - node := &api.Node{} - node.Name = testNode - node.Status = api.NodeStatus{Conditions: conditions} - return node -} - -type action struct { - verb string - resource string - subresource string -} - -func TestSetConditions(t *testing.T) { +func TestGeneratePatch(t *testing.T) { now := time.Now() - expectedActions := []action{ + update := []api.NodeCondition{ { - verb: "get", - resource: "nodes", + Type: "TestType1", + Status: api.ConditionTrue, + LastTransitionTime: unversioned.NewTime(now), + Reason: "TestReason1", + Message: "TestMessage1", }, { - verb: "update", - resource: "nodes", - subresource: "status", + Type: "TestType2", + Status: api.ConditionFalse, + LastTransitionTime: unversioned.NewTime(now), + Reason: "TestReason2", + Message: "TestMessage2", }, } - for _, test := range []struct { - init []api.NodeCondition - update []api.NodeCondition - expected []api.NodeCondition - }{ - // Init condition with the same type should be override - { - init: []api.NodeCondition{ - { - Type: "TestType", - Status: api.ConditionTrue, - }, - }, - update: []api.NodeCondition{ - { - Type: "TestType", - Status: api.ConditionTrue, - LastTransitionTime: unversioned.NewTime(now), - Reason: "TestReason", - Message: "TestMessage", - }, - }, - expected: []api.NodeCondition{ - { - // LastHeartbeatTime should be updated in SetConditions - Type: "TestType", - Status: api.ConditionTrue, - LastHeartbeatTime: unversioned.NewTime(now), - LastTransitionTime: unversioned.NewTime(now), - Reason: "TestReason", - Message: "TestMessage", - }, - }, - }, - // Init condition with different type should be kept - { - init: []api.NodeCondition{ - { - Type: "InitType", - Status: api.ConditionTrue, - LastTransitionTime: unversioned.NewTime(now), - Reason: "InitReason", - Message: "InitMessage", - }, - }, - update: []api.NodeCondition{ - { - Type: "TestType", - Status: api.ConditionTrue, - LastTransitionTime: unversioned.NewTime(now), - Reason: "TestReason", - Message: "TestMessage", - }, - }, - expected: []api.NodeCondition{ - { - Type: "InitType", - Status: api.ConditionTrue, - LastTransitionTime: unversioned.NewTime(now), - Reason: "InitReason", - Message: "InitMessage", - }, - { - // LastHeartbeatTime should be updated in SetConditions - Type: "TestType", - Status: api.ConditionTrue, - LastHeartbeatTime: unversioned.NewTime(now), - LastTransitionTime: unversioned.NewTime(now), - Reason: "TestReason", - Message: "TestMessage", - }, - }, - }, - // Condition with false status should be removed - { - init: []api.NodeCondition{ - { - Type: "TestType", - Status: api.ConditionTrue, - LastHeartbeatTime: unversioned.NewTime(now), - LastTransitionTime: unversioned.NewTime(now), - Reason: "TestReason", - Message: "TestMessage", - }, - }, - update: []api.NodeCondition{ - { - Type: "TestType", - Status: api.ConditionFalse, - }, - }, - expected: []api.NodeCondition{}, - }, - } { - fakeClient := fake.NewSimpleClientset(newFakeNode(test.init)) - client := newFakeProblemClient(fakeClient) - clock := client.clock.(*util.FakeClock) - clock.SetTime(now) + raw, err := json.Marshal(&update) + assert.NoError(t, err) + expectedPatch := []byte(fmt.Sprintf(`{"status":{"conditions":%s}}`, raw)) - client.SetConditions(test.update, 10*time.Second) - - // The actions should match the expected actions - actions := fakeClient.Actions() - if len(expectedActions) != len(actions) { - t.Errorf("expected actions %+v, got %+v", expectedActions, fakeClient.Actions()) - continue - } - for i, a := range actions { - if !a.Matches(expectedActions[i].verb, expectedActions[i].resource) || a.GetSubresource() != expectedActions[i].subresource { - t.Errorf("expected action %+v, got %+v", expectedActions[i], a) - } - } - // The last action should be an update - a, ok := actions[len(actions)-1].(core.UpdateAction) - if !ok { - t.Errorf("expected the last action to be update, got %+v", actions[len(actions)-1]) - } - // The updated node conditions should match the expected conditions - node, ok := a.GetObject().(*api.Node) - if !ok { - t.Errorf("expected the update object to be node, got %+v", a.GetObject()) - } - if !api.Semantic.DeepEqual(test.expected, node.Status.Conditions) { - t.Errorf("expected conditions %+v, got %+v", test.expected, node.Status.Conditions) - } - } -} - -func TestSetConditionsError(t *testing.T) { - timeout := time.Duration(0) - node := newFakeNode([]api.NodeCondition{}) - for c, test := range []struct { - errMap map[string]error - expectedErr error - }{ - { - // Get error - errMap: map[string]error{"get": fmt.Errorf("get error")}, - expectedErr: fmt.Errorf("get error"), - }, - { - // Update error - errMap: map[string]error{"update": fmt.Errorf("update error")}, - expectedErr: fmt.Errorf("update error"), - }, - { - // Timeout error - errMap: map[string]error{ - "update": &errors.StatusError{ErrStatus: unversioned.Status{Reason: unversioned.StatusReasonConflict}}, - }, - expectedErr: timeoutError{node: testNode, timeout: timeout}, - }, - { - // No error - errMap: map[string]error{}, - expectedErr: nil, - }, - } { - fakeClient := &fake.Clientset{} - client := newFakeProblemClient(fakeClient) - fakeClient.AddReactor("get", "nodes", func(action core.Action) (bool, runtime.Object, error) { - return true, node, test.errMap["get"] - }) - fakeClient.AddReactor("update", "nodes", func(action core.Action) (bool, runtime.Object, error) { - return true, node, test.errMap["update"] - }) - err := client.SetConditions([]api.NodeCondition{}, timeout) - if !reflect.DeepEqual(err, test.expectedErr) { - t.Errorf("case %d: expected error %v, got %v", c+1, test.expectedErr, err) - } + patch, err := generatePatch(update) + assert.NoError(t, err) + if string(patch) != string(expectedPatch) { + t.Errorf("expected patch %q, got %q", expectedPatch, patch) } } func TestEvent(t *testing.T) { fakeRecorder := record.NewFakeRecorder(1) - client := newFakeProblemClient(&fake.Clientset{}) + client := newFakeProblemClient() client.recorders[testSource] = fakeRecorder client.Eventf(api.EventTypeWarning, testSource, "test reason", "test message") expected := fmt.Sprintf("%s %s %s", api.EventTypeWarning, "test reason", "test message")