Merge branch 'master' into bugfix/abort-on-cleanout-failure

Author: ewoutp

pkg/apis/deployment/v1alpha/deployment_status_members.go

@@ -75,23 +75,23 @@ func (ds DeploymentStatusMembers) ElementByID(id string) (MemberStatus, ServerGr
 // ForeachServerGroup calls the given callback for all server groups.
 // If the callback returns an error, this error is returned and the callback is
 // not called for the remaining groups.
-func (ds DeploymentStatusMembers) ForeachServerGroup(cb func(group ServerGroup, list *MemberStatusList) error) error {
-	if err := cb(ServerGroupSingle, &ds.Single); err != nil {
+func (ds DeploymentStatusMembers) ForeachServerGroup(cb func(group ServerGroup, list MemberStatusList) error) error {
+	if err := cb(ServerGroupSingle, ds.Single); err != nil {
 		return maskAny(err)
 	}
-	if err := cb(ServerGroupAgents, &ds.Agents); err != nil {
+	if err := cb(ServerGroupAgents, ds.Agents); err != nil {
 		return maskAny(err)
 	}
-	if err := cb(ServerGroupDBServers, &ds.DBServers); err != nil {
+	if err := cb(ServerGroupDBServers, ds.DBServers); err != nil {
 		return maskAny(err)
 	}
-	if err := cb(ServerGroupCoordinators, &ds.Coordinators); err != nil {
+	if err := cb(ServerGroupCoordinators, ds.Coordinators); err != nil {
 		return maskAny(err)
 	}
-	if err := cb(ServerGroupSyncMasters, &ds.SyncMasters); err != nil {
+	if err := cb(ServerGroupSyncMasters, ds.SyncMasters); err != nil {
 		return maskAny(err)
 	}
-	if err := cb(ServerGroupSyncWorkers, &ds.SyncWorkers); err != nil {
+	if err := cb(ServerGroupSyncWorkers, ds.SyncWorkers); err != nil {
 		return maskAny(err)
 	}
 	return nil
@@ -137,22 +137,47 @@ func (ds DeploymentStatusMembers) MemberStatusByPVCName(pvcName string) (MemberS
 	return MemberStatus{}, 0, false
 }
 
-// UpdateMemberStatus updates the given status in the given group.
-func (ds *DeploymentStatusMembers) UpdateMemberStatus(status MemberStatus, group ServerGroup) error {
+// Add adds the given status in the given group.
+func (ds *DeploymentStatusMembers) Add(status MemberStatus, group ServerGroup) error {
 	var err error
 	switch group {
 	case ServerGroupSingle:
-		err = ds.Single.Update(status)
+		err = ds.Single.add(status)
 	case ServerGroupAgents:
-		err = ds.Agents.Update(status)
+		err = ds.Agents.add(status)
 	case ServerGroupDBServers:
-		err = ds.DBServers.Update(status)
+		err = ds.DBServers.add(status)
 	case ServerGroupCoordinators:
-		err = ds.Coordinators.Update(status)
+		err = ds.Coordinators.add(status)
 	case ServerGroupSyncMasters:
-		err = ds.SyncMasters.Update(status)
+		err = ds.SyncMasters.add(status)
 	case ServerGroupSyncWorkers:
-		err = ds.SyncWorkers.Update(status)
+		err = ds.SyncWorkers.add(status)
+	default:
+		return maskAny(errors.Wrapf(NotFoundError, "ServerGroup %d is not known", group))
+	}
+	if err != nil {
+		return maskAny(err)
+	}
+	return nil
+}
+
+// Update updates the given status in the given group.
+func (ds *DeploymentStatusMembers) Update(status MemberStatus, group ServerGroup) error {
+	var err error
+	switch group {
+	case ServerGroupSingle:
+		err = ds.Single.update(status)
+	case ServerGroupAgents:
+		err = ds.Agents.update(status)
+	case ServerGroupDBServers:
+		err = ds.DBServers.update(status)
+	case ServerGroupCoordinators:
+		err = ds.Coordinators.update(status)
+	case ServerGroupSyncMasters:
+		err = ds.SyncMasters.update(status)
+	case ServerGroupSyncWorkers:
+		err = ds.SyncWorkers.update(status)
 	default:
 		return maskAny(errors.Wrapf(NotFoundError, "ServerGroup %d is not known", group))
 	}
@@ -168,17 +193,17 @@ func (ds *DeploymentStatusMembers) RemoveByID(id string, group ServerGroup) erro
 	var err error
 	switch group {
 	case ServerGroupSingle:
-		err = ds.Single.RemoveByID(id)
+		err = ds.Single.removeByID(id)
 	case ServerGroupAgents:
-		err = ds.Agents.RemoveByID(id)
+		err = ds.Agents.removeByID(id)
 	case ServerGroupDBServers:
-		err = ds.DBServers.RemoveByID(id)
+		err = ds.DBServers.removeByID(id)
 	case ServerGroupCoordinators:
-		err = ds.Coordinators.RemoveByID(id)
+		err = ds.Coordinators.removeByID(id)
 	case ServerGroupSyncMasters:
-		err = ds.SyncMasters.RemoveByID(id)
+		err = ds.SyncMasters.removeByID(id)
 	case ServerGroupSyncWorkers:
-		err = ds.SyncWorkers.RemoveByID(id)
+		err = ds.SyncWorkers.removeByID(id)
 	default:
 		return maskAny(errors.Wrapf(NotFoundError, "ServerGroup %d is not known", group))
 	}
@@ -190,8 +215,8 @@ func (ds *DeploymentStatusMembers) RemoveByID(id string, group ServerGroup) erro
 
 // AllMembersReady returns true when all members are in the Ready state.
 func (ds DeploymentStatusMembers) AllMembersReady() bool {
-	if err := ds.ForeachServerGroup(func(group ServerGroup, list *MemberStatusList) error {
-		for _, x := range *list {
+	if err := ds.ForeachServerGroup(func(group ServerGroup, list MemberStatusList) error {
+		for _, x := range list {
 			if !x.Conditions.IsTrue(ConditionTypeReady) {
 				return fmt.Errorf("not ready")
 			}

pkg/apis/deployment/v1alpha/member_status_list.go

@@ -76,7 +76,7 @@ func (l MemberStatusList) ElementByPVCName(pvcName string) (MemberStatus, bool)
 
 // Add a member to the list.
 // Returns an AlreadyExistsError if the ID of the given member already exists.
-func (l *MemberStatusList) Add(m MemberStatus) error {
+func (l *MemberStatusList) add(m MemberStatus) error {
 	src := *l
 	for _, x := range src {
 		if x.ID == m.ID {
@@ -89,7 +89,7 @@ func (l *MemberStatusList) Add(m MemberStatus) error {
 
 // Update a member in the list.
 // Returns a NotFoundError if the ID of the given member cannot be found.
-func (l MemberStatusList) Update(m MemberStatus) error {
+func (l MemberStatusList) update(m MemberStatus) error {
 	for i, x := range l {
 		if x.ID == m.ID {
 			l[i] = m
@@ -101,7 +101,7 @@ func (l MemberStatusList) Update(m MemberStatus) error {
 
 // RemoveByID a member with given ID from the list.
 // Returns a NotFoundError if the ID of the given member cannot be found.
-func (l *MemberStatusList) RemoveByID(id string) error {
+func (l *MemberStatusList) removeByID(id string) error {
 	src := *l
 	for i, x := range src {
 		if x.ID == id {

pkg/apis/deployment/v1alpha/member_status_list_test.go

@@ -0,0 +1,69 @@
+//
+// DISCLAIMER
+//
+// Copyright 2018 ArangoDB GmbH, Cologne, Germany
+//
+// 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.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+// Author Ewout Prangsma
+//
+
+package v1alpha
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// TestMemberStatusList tests modifying a MemberStatusList.
+func TestMemberStatusList(t *testing.T) {
+	list := &MemberStatusList{}
+	m1 := MemberStatus{ID: "m1"}
+	m2 := MemberStatus{ID: "m2"}
+	m3 := MemberStatus{ID: "m3"}
+	assert.Equal(t, 0, len(*list))
+
+	assert.NoError(t, list.add(m1))
+	assert.Equal(t, 1, len(*list))
+
+	assert.NoError(t, list.add(m2))
+	assert.NoError(t, list.add(m3))
+	assert.Equal(t, 3, len(*list))
+
+	assert.Error(t, list.add(m2))
+	assert.Equal(t, 3, len(*list))
+
+	assert.NoError(t, list.removeByID(m3.ID))
+	assert.Equal(t, 2, len(*list))
+	assert.False(t, list.ContainsID(m3.ID))
+	assert.Equal(t, m1.ID, (*list)[0].ID)
+	assert.Equal(t, m2.ID, (*list)[1].ID)
+
+	m2.PodName = "foo"
+	assert.NoError(t, list.update(m2))
+	assert.Equal(t, 2, len(*list))
+	assert.True(t, list.ContainsID(m2.ID))
+	x, found := list.ElementByPodName("foo")
+	assert.True(t, found)
+	assert.Equal(t, "foo", x.PodName)
+	assert.Equal(t, m2.ID, x.ID)
+
+	assert.NoError(t, list.add(m3))
+	assert.Equal(t, 3, len(*list))
+	assert.Equal(t, m1.ID, (*list)[0].ID)
+	assert.Equal(t, m2.ID, (*list)[1].ID)
+	assert.Equal(t, m3.ID, (*list)[2].ID)
+}

pkg/deployment/cluster_scaling_integration.go

@@ -71,7 +71,7 @@ func (ci *clusterScalingIntegration) ListenForClusterEvents(stopCh <-chan struct
 		delay := time.Second * 2
 
 		// Is deployment in running state
-		if ci.depl.status.Phase == api.DeploymentPhaseRunning {
+		if ci.depl.GetPhase() == api.DeploymentPhaseRunning {
 			// Update cluster with our state
 			ctx := context.Background()
 			safeToAskCluster, err := ci.updateClusterServerCount(ctx)

pkg/deployment/context_impl.go

@@ -24,6 +24,7 @@ package deployment
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/arangodb/arangosync/client"
 	"github.com/arangodb/arangosync/tasks"
@@ -65,20 +66,44 @@ func (d *Deployment) GetNamespace() string {
 	return d.apiObject.GetNamespace()
 }
 
+// GetPhase returns the current phase of the deployment
+func (d *Deployment) GetPhase() api.DeploymentPhase {
+	return d.status.last.Phase
+}
+
 // GetSpec returns the current specification
 func (d *Deployment) GetSpec() api.DeploymentSpec {
 	return d.apiObject.Spec
 }
 
 // GetStatus returns the current status of the deployment
-func (d *Deployment) GetStatus() api.DeploymentStatus {
-	return d.status
+// together with the current version of that status.
+func (d *Deployment) GetStatus() (api.DeploymentStatus, int32) {
+	d.status.mutex.Lock()
+	defer d.status.mutex.Unlock()
+
+	version := d.status.version
+	return *d.status.last.DeepCopy(), version
 }
 
 // UpdateStatus replaces the status of the deployment with the given status and
 // updates the resources in k8s.
-func (d *Deployment) UpdateStatus(status api.DeploymentStatus, force ...bool) error {
-	d.status = status
+// If the given last version does not match the actual last version of the status object,
+// an error is returned.
+func (d *Deployment) UpdateStatus(status api.DeploymentStatus, lastVersion int32, force ...bool) error {
+	d.status.mutex.Lock()
+	defer d.status.mutex.Unlock()
+
+	if d.status.version != lastVersion {
+		// Status is obsolete
+		d.deps.Log.Error().
+			Int32("expected-version", lastVersion).
+			Int32("actual-version", d.status.version).
+			Msg("UpdateStatus version conflict error.")
+		return maskAny(fmt.Errorf("Status conflict error. Expected version %d, got %d", lastVersion, d.status.version))
+	}
+	d.status.version++
+	d.status.last = *status.DeepCopy()
 	if err := d.updateCRStatus(force...); err != nil {
 		return maskAny(err)
 	}
@@ -107,7 +132,7 @@ func (d *Deployment) GetServerClient(ctx context.Context, group api.ServerGroup,
 // GetAgencyClients returns a client connection for every agency member.
 // If the given predicate is not nil, only agents are included where the given predicate returns true.
 func (d *Deployment) GetAgencyClients(ctx context.Context, predicate func(id string) bool) ([]driver.Connection, error) {
-	agencyMembers := d.status.Members.Agents
+	agencyMembers := d.status.last.Members.Agents
 	result := make([]driver.Connection, 0, len(agencyMembers))
 	for _, m := range agencyMembers {
 		if predicate != nil && !predicate(m.ID) {
@@ -168,13 +193,14 @@ func (d *Deployment) GetSyncServerClient(ctx context.Context, group api.ServerGr
 // If ID is non-empty, it will be used, otherwise a new ID is created.
 func (d *Deployment) CreateMember(group api.ServerGroup, id string) error {
 	log := d.deps.Log
-	id, err := d.createMember(group, id, d.apiObject)
+	status, lastVersion := d.GetStatus()
+	id, err := createMember(log, &status, group, id, d.apiObject)
 	if err != nil {
 		log.Debug().Err(err).Str("group", group.AsRole()).Msg("Failed to create member")
 		return maskAny(err)
 	}
 	// Save added member
-	if err := d.updateCRStatus(); err != nil {
+	if err := d.UpdateStatus(status, lastVersion); err != nil {
 		log.Debug().Err(err).Msg("Updating CR status failed")
 		return maskAny(err)
 	}