Exposes through REST leaving a tribe agreement

mgmt/rest/rbody/body.go

@@ -111,6 +111,8 @@ func UnmarshalBody(t string, b []byte) (Body, error) {
 		return unmarshalAndHandleError(b, &TribeMemberShow{})
 	case TribeJoinAgreementType:
 		return unmarshalAndHandleError(b, &TribeJoinAgreement{})
+	case TribeLeaveAgreementType:
+		return unmarshalAndHandleError(b, &TribeLeaveAgreement{})
 	case TribeGetAgreementType:
 		return unmarshalAndHandleError(b, &TribeGetAgreement{})
 	case PluginConfigItemType:

mgmt/rest/rbody/tribe.go

@@ -22,13 +22,14 @@ package rbody
 import "github.com/intelsdi-x/pulse/mgmt/tribe/agreement"
 
 const (
-	TribeAgreementListType = "tribe_agreement_list_returned"
-	TribeGetAgreementType  = "tribe_agreement_returned"
-	TribeAddAgreementType  = "tribe_agreement_created"
-	TribeAddMemberType     = "tribe_member_added"
-	TribeJoinAgreementType = "tribe_agreement_joined"
-	TribeMemberListType    = "tribe_member_list_returned"
-	TribeMemberShowType    = "tribe_member_details_returned"
+	TribeAgreementListType  = "tribe_agreement_list_returned"
+	TribeGetAgreementType   = "tribe_agreement_returned"
+	TribeAddAgreementType   = "tribe_agreement_created"
+	TribeAddMemberType      = "tribe_member_added"
+	TribeJoinAgreementType  = "tribe_agreement_joined"
+	TribeLeaveAgreementType = "tribe_agreement_left"
+	TribeMemberListType     = "tribe_member_list_returned"
+	TribeMemberShowType     = "tribe_member_details_returned"
 )
 
 type TribeAddAgreement struct {
@@ -79,6 +80,18 @@ func (t *TribeJoinAgreement) ResponseBodyType() string {
 	return TribeJoinAgreementType
 }
 
+type TribeLeaveAgreement struct {
+	Agreement *agreement.Agreement `json:"agreement"`
+}
+
+func (t *TribeLeaveAgreement) ResponseBodyMessage() string {
+	return "Tribe agreement left"
+}
+
+func (t *TribeLeaveAgreement) ResponseBodyType() string {
+	return TribeLeaveAgreementType
+}
+
 type TribeMemberList struct {
 	Members []string `json:"members"`
 }

mgmt/rest/server.go

@@ -98,6 +98,7 @@ type managesTribe interface {
 	GetAgreements() map[string]*agreement.Agreement
 	AddAgreement(name string) perror.PulseError
 	JoinAgreement(agreementName, memberName string) perror.PulseError
+	LeaveAgreement(agreementName, memberName string) perror.PulseError
 	GetMembers() []string
 	GetMember(name string) *agreement.Member
 }
@@ -220,7 +221,8 @@ func (s *Server) addRoutes() {
 		s.r.GET("/v1/tribe/agreements", s.getAgreements)
 		s.r.POST("/v1/tribe/agreements", s.addAgreement)
 		s.r.GET("/v1/tribe/agreements/:name", s.getAgreement)
-		s.r.POST("/v1/tribe/agreements/:name/join", s.joinAgreement)
+		s.r.POST("/v1/tribe/agreements/:name/join", s.joinAgreement) // TODO change to put
+		s.r.DELETE("/v1/tribe/agreements/:name/leave", s.leaveAgreement)
 		s.r.GET("/v1/tribe/members", s.getMembers)
 		s.r.GET("/v1/tribe/member/:name", s.getMember)
 	}

mgmt/rest/tribe.go

@@ -110,10 +110,55 @@ func (s *Server) joinAgreement(w http.ResponseWriter, r *http.Request, p httprou
 		respond(400, rbody.FromPulseError(perr), w)
 		return
 	}
-	respond(200, &rbody.TribeJoinAgreement{}, w)
+	agreement, _ := s.tr.GetAgreement(name)
+	respond(200, &rbody.TribeJoinAgreement{Agreement: agreement}, w)
 
 }
 
+func (s *Server) leaveAgreement(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
+	tribeLogger = tribeLogger.WithField("_block", "leaveAgreement")
+	name := p.ByName("name")
+	if _, ok := s.tr.GetAgreements()[name]; !ok {
+		fields := map[string]interface{}{
+			"agreement_name": name,
+		}
+		tribeLogger.WithFields(fields).Error(ErrAgreementDoesNotExist)
+		respond(400, rbody.FromPulseError(perror.New(ErrAgreementDoesNotExist, fields)), w)
+		return
+	}
+
+	b, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		tribeLogger.Error(err)
+		respond(500, rbody.FromError(err), w)
+		return
+	}
+
+	m := struct {
+		MemberName string `json:"member_name"`
+	}{}
+	err = json.Unmarshal(b, &m)
+	if err != nil {
+		fields := map[string]interface{}{
+			"error": err,
+			"hint":  `The body of the request should be of the form '{"member_name": "some_value"}'`,
+		}
+		pe := perror.New(ErrInvalidJSON, fields)
+		tribeLogger.WithFields(fields).Error(ErrInvalidJSON)
+		respond(400, rbody.FromPulseError(pe), w)
+		return
+	}
+
+	perr := s.tr.LeaveAgreement(name, m.MemberName)
+	if perr != nil {
+		tribeLogger.Error(perr)
+		respond(400, rbody.FromPulseError(perr), w)
+		return
+	}
+	agreement, _ := s.tr.GetAgreement(name)
+	respond(200, &rbody.TribeLeaveAgreement{Agreement: agreement}, w)
+}
+
 func (s *Server) getMembers(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
 	members := s.tr.GetMembers()
 	respond(200, &rbody.TribeMemberList{Members: members}, w)

mgmt/rest/tribe_test.go

@@ -95,6 +95,27 @@ func joinAgreement(port int, memberName, agreementName string) *rbody.APIRespons
 	return getAPIResponse(resp)
 }
 
+func leaveAgreement(port int, memberName, agreementName string) *rbody.APIResponse {
+	ja, err := json.Marshal(struct {
+		MemberName string `json:"member_name"`
+	}{
+		MemberName: memberName,
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+	b := bytes.NewReader(ja)
+	client := &http.Client{}
+	uri := fmt.Sprintf("http://127.0.0.1:%d/v1/tribe/agreements/%s/leave", port, agreementName)
+	req, err := http.NewRequest("DELETE", uri, b)
+	req.Header.Add("Content-Type", "application/json")
+	resp, err := client.Do(req)
+	if err != nil {
+		log.Fatal(err)
+	}
+	return getAPIResponse(resp)
+}
+
 func addAgreement(port int, name string) *rbody.APIResponse {
 	a, err := json.Marshal(struct {
 		Name string
@@ -419,6 +440,37 @@ func TestTribePluginAgreements(t *testing.T) {
 										}
 										wg.Wait()
 										So(timedOut, ShouldEqual, false)
+
+										Convey("A node leaves the agreement", func() {
+											leaveAgreement(mgtPorts[0], fmt.Sprintf("member-%d", mgtPorts[0]), aName)
+											var wg sync.WaitGroup
+											timedOut := false
+											for _, i := range mgtPorts {
+												timer := time.After(15 * time.Second)
+												wg.Add(1)
+												go func(port int, name string) {
+													defer wg.Done()
+													for {
+														select {
+														case <-timer:
+															timedOut = true
+															return
+														default:
+															resp := getAgreement(port, aName)
+															if resp.Meta.Code == 200 {
+																c.So(resp.Body.(*rbody.TribeGetAgreement), ShouldHaveSameTypeAs, new(rbody.TribeGetAgreement))
+																if len(resp.Body.(*rbody.TribeGetAgreement).Agreement.Members) == numOfNodes-1 {
+																	return
+																}
+															}
+															time.Sleep(200 * time.Millisecond)
+														}
+													}
+												}(i, fmt.Sprintf("member-%d", i))
+											}
+											wg.Wait()
+											So(timedOut, ShouldEqual, false)
+										})
 									})
 								})
 							})