Merge pull request #5005 from xiang90/clu_storage

membership: update attr in membership pkg

etcdserver/membership/cluster.go

@@ -307,21 +307,21 @@ func (c *RaftCluster) RemoveMember(id types.ID) {
 	c.removed[id] = true
 }
 
-func (c *RaftCluster) UpdateAttributes(id types.ID, attr Attributes) bool {
+func (c *RaftCluster) UpdateAttributes(id types.ID, attr Attributes) {
 	c.Lock()
 	defer c.Unlock()
 	if m, ok := c.members[id]; ok {
 		m.Attributes = attr
-		return true
+		if c.store != nil {
+			mustUpdateMemberAttrInStore(c.store, m)
+		}
+		return
 	}
 	_, ok := c.removed[id]
-	if ok {
-		plog.Warningf("skipped updating attributes of removed member %s", id)
-	} else {
+	if !ok {
 		plog.Panicf("error updating attributes of unknown member %s", id)
 	}
-	// TODO: update store in this function
-	return false
+	plog.Warningf("skipped updating attributes of removed member %s", id)
 }
 
 func (c *RaftCluster) UpdateRaftAttributes(id types.ID, raftAttr RaftAttributes) {

etcdserver/membership/store.go

@@ -25,9 +25,7 @@ import (
 )
 
 const (
-	// TODO: make this private after moving all membership storage logic
-	// from etcdserver pkg
-	AttributesSuffix     = "attributes"
+	attributesSuffix     = "attributes"
 	raftAttributesSuffix = "raftAttributes"
 
 	// the prefix for stroing membership related information in store provided by store pkg.
@@ -96,13 +94,24 @@ func mustUpdateMemberInStore(s store.Store, m *Member) {
 	}
 }
 
+func mustUpdateMemberAttrInStore(s store.Store, m *Member) {
+	b, err := json.Marshal(m.Attributes)
+	if err != nil {
+		plog.Panicf("marshal raftAttributes should never fail: %v", err)
+	}
+	p := path.Join(MemberStoreKey(m.ID), attributesSuffix)
+	if _, err := s.Set(p, false, string(b), store.TTLOptionSet{ExpireTime: store.Permanent}); err != nil {
+		plog.Panicf("update raftAttributes should never fail: %v", err)
+	}
+}
+
 // nodeToMember builds member from a key value node.
 // the child nodes of the given node MUST be sorted by key.
 func nodeToMember(n *store.NodeExtern) (*Member, error) {
 	m := &Member{ID: MustParseMemberIDFromKey(n.Key)}
 	attrs := make(map[string][]byte)
 	raftAttrKey := path.Join(n.Key, raftAttributesSuffix)
-	attrKey := path.Join(n.Key, AttributesSuffix)
+	attrKey := path.Join(n.Key, attributesSuffix)
 	for _, nn := range n.Nodes {
 		if nn.Key != raftAttrKey && nn.Key != attrKey {
 			return nil, fmt.Errorf("unknown key %q", nn.Key)
@@ -133,7 +142,7 @@ func MemberStoreKey(id types.ID) string {
 }
 
 func MemberAttributesStorePath(id types.ID) string {
-	return path.Join(MemberStoreKey(id), AttributesSuffix)
+	return path.Join(MemberStoreKey(id), attributesSuffix)
 }
 
 func MustParseMemberIDFromKey(key string) types.ID {

etcdserver/server.go

@@ -1091,18 +1091,15 @@ func (s *EtcdServer) applyRequest(r pb.Request) Response {
 		case r.PrevIndex > 0 || r.PrevValue != "":
 			return f(s.store.CompareAndSwap(r.Path, r.PrevValue, r.PrevIndex, r.Val, ttlOptions))
 		default:
-			// TODO (yicheng): cluster should be the owner of cluster prefix store
-			// we should not modify cluster store here.
 			if storeMemberAttributeRegexp.MatchString(r.Path) {
 				id := membership.MustParseMemberIDFromKey(path.Dir(r.Path))
 				var attr membership.Attributes
 				if err := json.Unmarshal([]byte(r.Val), &attr); err != nil {
 					plog.Panicf("unmarshal %s should never fail: %v", r.Val, err)
 				}
-				ok := s.cluster.UpdateAttributes(id, attr)
-				if !ok {
-					return Response{}
-				}
+				s.cluster.UpdateAttributes(id, attr)
+				// return an empty response since there is no consumer.
+				return Response{}
 			}
 			if r.Path == path.Join(StoreClusterPrefix, "version") {
 				s.cluster.SetVersion(semver.Must(semver.NewVersion(r.Val)))

etcdserver/server_test.go

@@ -21,7 +21,6 @@ import (
 	"os"
 	"path"
 	"reflect"
-	"strconv"
 	"testing"
 	"time"
 
@@ -466,7 +465,7 @@ func TestApplyRequestOnAdminMemberAttributes(t *testing.T) {
 	req := pb.Request{
 		Method: "PUT",
 		ID:     1,
-		Path:   path.Join(membership.StoreMembersPrefix, strconv.FormatUint(1, 16), membership.AttributesSuffix),
+		Path:   membership.MemberAttributesStorePath(1),
 		Val:    `{"Name":"abc","ClientURLs":["http://127.0.0.1:2379"]}`,
 	}
 	srv.applyRequest(req)