[Feature] Topology Support (#804)

CHANGELOG.md

@@ -5,6 +5,7 @@
 - Add "Short Names" feature
 - Switch ArangoDB Image Discovery process from Headless Service to Pod IP
 - Fix PVC Resize for Single servers
+- Add Topology support
 
 ## [1.2.3](https://github.com/arangodb/kube-arangodb/tree/1.2.3) (2021-09-24)
 - Update UBI Image to 8.4

go.mod

@@ -29,7 +29,7 @@ require (
 	github.com/arangodb/go-driver v0.0.0-20210621075908-e7a6fa0cbd18
 	github.com/arangodb/go-upgrade-rules v0.0.0-20180809110947-031b4774ff21
 	github.com/cenkalti/backoff v2.2.1+incompatible
-	github.com/coreos/go-semver v0.3.0
+	github.com/coreos/go-semver v0.3.0 // indirect
 	github.com/dchest/uniuri v0.0.0-20160212164326-8902c56451e9
 	github.com/evanphx/json-patch v4.9.0+incompatible
 	github.com/ghodss/yaml v1.0.0
@@ -43,7 +43,7 @@ require (
 	github.com/jessevdk/go-assets v0.0.0-20160921144138-4f4301a06e15
 	github.com/jessevdk/go-assets-builder v0.0.0-20130903091706-b8483521738f // indirect
 	github.com/jessevdk/go-flags v1.4.0 // indirect
-	github.com/json-iterator/go v1.1.11
+	github.com/json-iterator/go v1.1.11 // indirect
 	github.com/julienschmidt/httprouter v1.3.0
 	github.com/kevinburke/rest v0.0.0-20210222204520-f7a2e216372f // indirect
 	github.com/magiconair/properties v1.8.0
@@ -62,7 +62,7 @@ require (
 	github.com/ugorji/go/codec v1.2.6 // indirect
 	github.com/voxelbrain/goptions v0.0.0-20180630082107-58cddc247ea2 // indirect
 	github.com/zenazn/goji v0.9.0 // indirect
-	golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4
+	golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4 // indirect
 	golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c
 	golang.org/x/tools v0.1.1-0.20210504181558-0bb7e5c47b1a // indirect
 	google.golang.org/protobuf v1.27.1 // indirect

pkg/apis/deployment/v1/conditions.go

@@ -83,6 +83,8 @@ const (
 	ConditionTypeUpdating ConditionType = "Updating"
 	// ConditionTypeUpdateFailed indicates that runtime update failed
 	ConditionTypeUpdateFailed ConditionType = "UpdateFailed"
+	// ConditionTypeTopologyAware indicates that the member is deployed with TopologyAwareness.
+	ConditionTypeTopologyAware ConditionType = "TopologyAware"
 )
 
 // Condition represents one current condition of a deployment or deployment member.

pkg/apis/deployment/v1/deployment_spec.go

@@ -163,6 +163,9 @@ type DeploymentSpec struct {
 
 	// CommunicationMethod define communication method used in deployment
 	CommunicationMethod *DeploymentCommunicationMethod `json:"communicationMethod,omitempty"`
+
+	// Topology define topology adjustment details, Enterprise only
+	Topology *TopologySpec `json:"topology,omitempty"`
 }
 
 // GetAllowMemberRecreation returns member recreation policy based on group and settings

pkg/apis/deployment/v1/deployment_status.go

@@ -81,6 +81,8 @@ type DeploymentStatus struct {
 
 	// Agency keeps information about agency
 	Agency *DeploymentStatusAgencyInfo `json:"agency,omitempty"`
+
+	Topology *TopologyStatus `json:"topology,omitempty"`
 }
 
 // Equal checks for equality

pkg/apis/deployment/v1/list.go

@@ -0,0 +1,55 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+package v1
+
+import "sort"
+
+type List []string
+
+func (l List) Contains(v string) bool {
+	for _, z := range l {
+		if z == v {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (l List) Sort() {
+	sort.Strings(l)
+}
+
+func (l List) Remove(values ...string) List {
+	var m List
+
+	toRemove := List(values)
+
+	for _, v := range l {
+		if toRemove.Contains(v) {
+			continue
+		}
+
+		m = append(m, v)
+	}
+
+	return m
+}

pkg/apis/deployment/v1/member_status.go

@@ -41,6 +41,9 @@ type MemberStatus struct {
 	// ID holds the unique ID of the member.
 	// This id is also used within the ArangoDB cluster to identify this server.
 	ID string `json:"id"`
+	// UID holds the unique UID of the member.
+	// UID is created once member run in AddMember action.
+	UID types.UID `json:"uid,omitempty"`
 	// RID holds the ID of the member run.
 	// Value is updated in Pending Phase.
 	RID types.UID `json:"rid,omitempty"`
@@ -80,11 +83,14 @@ type MemberStatus struct {
 	Upgrade bool `json:"upgrade,omitempty"`
 	// Endpoint definition how member should be reachable
 	Endpoint *string `json:"endpoint,omitempty"`
+	// Topology define topology member status assignment
+	Topology *TopologyMemberStatus `json:"topology,omitempty"`
 }
 
 // Equal checks for equality
 func (s MemberStatus) Equal(other MemberStatus) bool {
 	return s.ID == other.ID &&
+		s.UID == other.UID &&
 		s.RID == other.RID &&
 		s.Phase == other.Phase &&
 		util.TimeCompareEqual(s.CreatedAt, other.CreatedAt) &&

pkg/apis/deployment/v1/member_status_list.go

@@ -138,10 +138,12 @@ func (l *MemberStatusList) removeByID(id string) error {
 	return errors.WithStack(errors.Wrapf(NotFoundError, "Member '%s' is not a member", id))
 }
 
+type MemberToRemoveSelector func(m MemberStatusList) (string, error)
+
 // SelectMemberToRemove selects a member from the given list that should
 // be removed in a scale down action.
 // Returns an error if the list is empty.
-func (l MemberStatusList) SelectMemberToRemove() (MemberStatus, error) {
+func (l MemberStatusList) SelectMemberToRemove(selectors ...MemberToRemoveSelector) (MemberStatus, error) {
 	if len(l) > 0 {
 		// Try to find member with phase to be removed
 		for _, m := range l {
@@ -165,6 +167,20 @@ func (l MemberStatusList) SelectMemberToRemove() (MemberStatus, error) {
 				return m, nil
 			}
 		}
+
+		// Run conditional picker
+		for _, selector := range selectors {
+			if m, err := selector(l); err != nil {
+				return MemberStatus{}, err
+			} else if m != "" {
+				if member, ok := l.ElementByID(m); ok {
+					return member, nil
+				} else {
+					return MemberStatus{}, errors.Newf("Unable to find member with id %s", m)
+				}
+			}
+		}
+
 		// Pick a random member that is in created state
 		perm := rand.Perm(len(l))
 		for _, idx := range perm {

pkg/apis/deployment/v1/plan.go

@@ -171,6 +171,11 @@ const (
 	ActionTypeRuntimeContainerImageUpdate ActionType = "RuntimeContainerImageUpdate"
 	// ActionTypeRuntimeContainerArgsLogLevelUpdate updates the container's executor arguments.
 	ActionTypeRuntimeContainerArgsLogLevelUpdate ActionType = "RuntimeContainerArgsLogLevelUpdate"
+
+	// Topology
+	ActionTypeTopologyEnable           ActionType = "TopologyEnable"
+	ActionTypeTopologyDisable          ActionType = "TopologyDisable"
+	ActionTypeTopologyMemberAssignment ActionType = "TopologyMemberAssignment"
 )
 
 const (
@@ -346,3 +351,45 @@ func (p Plan) Wrap(before, after Action) Plan {
 
 	return n
 }
+
+// AfterFirst adds actions when condition will return false
+func (p Plan) AfterFirst(condition func(a Action) bool, actions ...Action) Plan {
+	var r Plan
+	c := p
+	for {
+		if len(c) == 0 {
+			break
+		}
+
+		if !condition(c[0]) {
+			r = append(r, actions...)
+
+			r = append(r, c...)
+
+			break
+		}
+
+		r = append(r, c[0])
+
+		if len(c) == 1 {
+			break
+		}
+
+		c = c[1:]
+	}
+
+	return r
+}
+
+// Filter filter list of the actions
+func (p Plan) Filter(condition func(a Action) bool) Plan {
+	var r Plan
+
+	for _, a := range p {
+		if condition(a) {
+			r = append(r, a)
+		}
+	}
+
+	return r
+}

pkg/apis/deployment/v1/topology_member_status.go

@@ -0,0 +1,28 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+package v1
+
+import "k8s.io/apimachinery/pkg/types"
+
+type TopologyMemberStatus struct {
+	ID   types.UID `json:"id"`
+	Zone int       `json:"rack"`
+}

pkg/apis/deployment/v1/topology_spec.go

@@ -0,0 +1,53 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+package v1
+
+const DefaultTopologySpecLabel = "topology.kubernetes.io/zone"
+
+type TopologySpec struct {
+	Enabled bool    `json:"enabled,omitempty"`
+	Zones   int     `json:"zones,omitempty"`
+	Label   *string `json:"label,omitempty"`
+}
+
+func (t *TopologySpec) IsEnabled() bool {
+	if t == nil {
+		return false
+	}
+
+	return t.Enabled && t.Zones > 0
+}
+
+func (t *TopologySpec) GetZones() int {
+	if t == nil {
+		return 0
+	}
+
+	return t.Zones
+}
+
+func (t *TopologySpec) GetLabel() string {
+	if t == nil || t.Label == nil {
+		return DefaultTopologySpecLabel
+	}
+
+	return *t.Label
+}

pkg/apis/deployment/v1/topology_status.go

@@ -0,0 +1,175 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+package v1
+
+import (
+	"math"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/uuid"
+)
+
+type TopologyStatus struct {
+	ID types.UID `json:"id"`
+
+	Size int `json:"size,omitempty"`
+
+	Zones TopologyStatusZones `json:"zones,omitempty"`
+
+	Label string `json:"label,omitempty"`
+}
+
+func (t *TopologyStatus) GetLeastUsedZone(group ServerGroup) int {
+	if t == nil {
+		return -1
+	}
+
+	r, m := -1, math.MaxInt64
+
+	for i, z := range t.Zones {
+		if n, ok := z.Members[group.AsRoleAbbreviated()]; ok {
+			if v := len(n); v < m {
+				r, m = i, v
+			}
+		} else {
+			if v := 0; v < m {
+				r, m = i, v
+			}
+		}
+	}
+
+	return r
+}
+
+func (t *TopologyStatus) RegisterTopologyLabel(zone int, label string) bool {
+	if t == nil {
+		return false
+	}
+
+	if zone < 0 || zone >= t.Size {
+		return false
+	}
+
+	if t.Zones[zone].Labels.Contains(label) {
+		return false
+	}
+
+	t.Zones[zone].Labels = append(t.Zones[zone].Labels, label)
+	t.Zones[zone].Labels.Sort()
+
+	return true
+}
+
+func (t *TopologyStatus) RemoveMember(group ServerGroup, id string) bool {
+	if t == nil {
+		return false
+	}
+
+	for _, zone := range t.Zones {
+		if zone.RemoveMember(group, id) {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (t *TopologyStatus) IsTopologyOwned(m *TopologyMemberStatus) bool {
+	if t == nil {
+		return false
+	}
+
+	if m == nil {
+		return false
+	}
+
+	return t.ID == m.ID
+}
+
+func (t *TopologyStatus) Enabled() bool {
+	return t != nil
+}
+
+type TopologyStatusZones []TopologyStatusZone
+
+type TopologyStatusZoneMembers map[string]List
+
+type TopologyStatusZone struct {
+	ID int `json:"id"`
+
+	Labels List `json:"labels,omitempty"`
+
+	Members TopologyStatusZoneMembers `json:"members,omitempty"`
+}
+
+func (t *TopologyStatusZone) AddMember(group ServerGroup, id string) {
+	if t.Members == nil {
+		t.Members = TopologyStatusZoneMembers{}
+	}
+
+	t.Members[group.AsRoleAbbreviated()] = append(t.Members[group.AsRoleAbbreviated()], id)
+
+	t.Members[group.AsRoleAbbreviated()].Sort()
+}
+
+func (t *TopologyStatusZone) RemoveMember(group ServerGroup, id string) bool {
+	if t == nil {
+		return false
+	}
+	if t.Members == nil {
+		return false
+	}
+	if !t.Members[group.AsRoleAbbreviated()].Contains(id) {
+		return false
+	}
+	t.Members[group.AsRoleAbbreviated()] = t.Members[group.AsRoleAbbreviated()].Remove(id)
+	return true
+}
+
+func (t *TopologyStatusZone) Get(group ServerGroup) List {
+	if t == nil {
+		return nil
+	}
+
+	if v, ok := t.Members[group.AsRoleAbbreviated()]; ok {
+		return v
+	} else {
+		return nil
+	}
+}
+
+func NewTopologyStatus(spec *TopologySpec) *TopologyStatus {
+	if spec == nil {
+		return nil
+	}
+	zones := make(TopologyStatusZones, spec.Zones)
+
+	for i := 0; i < spec.Zones; i++ {
+		zones[i] = TopologyStatusZone{ID: i}
+	}
+
+	return &TopologyStatus{
+		ID:    uuid.NewUUID(),
+		Size:  spec.Zones,
+		Zones: zones,
+		Label: spec.GetLabel(),
+	}
+}

pkg/apis/deployment/v1/topology_status_test.go

@@ -0,0 +1,41 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+package v1
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func Test_GetLeastUsedZone(t *testing.T) {
+	v := NewTopologyStatus(&TopologySpec{Enabled: true, Zones: 3})
+
+	require.Equal(t, 0, v.GetLeastUsedZone(ServerGroupDBServers))
+
+	v.Zones[0].AddMember(ServerGroupDBServers, "M-0")
+
+	require.Equal(t, 1, v.GetLeastUsedZone(ServerGroupDBServers))
+
+	v.Zones[0].RemoveMember(ServerGroupDBServers, "M-0")
+
+	require.Equal(t, 0, v.GetLeastUsedZone(ServerGroupDBServers))
+}

pkg/apis/deployment/v1/zz_generated.deepcopy.go

@@ -610,6 +610,11 @@ func (in *DeploymentSpec) DeepCopyInto(out *DeploymentSpec) {
 		*out = new(DeploymentCommunicationMethod)
 		**out = **in
 	}
+	if in.Topology != nil {
+		in, out := &in.Topology, &out.Topology
+		*out = new(TopologySpec)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -684,6 +689,11 @@ func (in *DeploymentStatus) DeepCopyInto(out *DeploymentStatus) {
 		*out = new(DeploymentStatusAgencyInfo)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.Topology != nil {
+		in, out := &in.Topology, &out.Topology
+		*out = new(TopologyStatus)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -1066,6 +1076,26 @@ func (in *LifecycleSpec) DeepCopy() *LifecycleSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in List) DeepCopyInto(out *List) {
+	{
+		in := &in
+		*out = make(List, len(*in))
+		copy(*out, *in)
+		return
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new List.
+func (in List) DeepCopy() List {
+	if in == nil {
+		return nil
+	}
+	out := new(List)
+	in.DeepCopyInto(out)
+	return *out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *MemberStatus) DeepCopyInto(out *MemberStatus) {
 	*out = *in
@@ -1106,6 +1136,11 @@ func (in *MemberStatus) DeepCopyInto(out *MemberStatus) {
 		*out = new(string)
 		**out = **in
 	}
+	if in.Topology != nil {
+		in, out := &in.Topology, &out.Topology
+		*out = new(TopologyMemberStatus)
+		**out = **in
+	}
 	return
 }
 
@@ -2209,3 +2244,151 @@ func (in *Timeouts) DeepCopy() *Timeouts {
 	in.DeepCopyInto(out)
 	return out
 }
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *TopologyMemberStatus) DeepCopyInto(out *TopologyMemberStatus) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TopologyMemberStatus.
+func (in *TopologyMemberStatus) DeepCopy() *TopologyMemberStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(TopologyMemberStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *TopologySpec) DeepCopyInto(out *TopologySpec) {
+	*out = *in
+	if in.Label != nil {
+		in, out := &in.Label, &out.Label
+		*out = new(string)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TopologySpec.
+func (in *TopologySpec) DeepCopy() *TopologySpec {
+	if in == nil {
+		return nil
+	}
+	out := new(TopologySpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *TopologyStatus) DeepCopyInto(out *TopologyStatus) {
+	*out = *in
+	if in.Zones != nil {
+		in, out := &in.Zones, &out.Zones
+		*out = make(TopologyStatusZones, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TopologyStatus.
+func (in *TopologyStatus) DeepCopy() *TopologyStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(TopologyStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *TopologyStatusZone) DeepCopyInto(out *TopologyStatusZone) {
+	*out = *in
+	if in.Labels != nil {
+		in, out := &in.Labels, &out.Labels
+		*out = make(List, len(*in))
+		copy(*out, *in)
+	}
+	if in.Members != nil {
+		in, out := &in.Members, &out.Members
+		*out = make(TopologyStatusZoneMembers, len(*in))
+		for key, val := range *in {
+			var outVal []string
+			if val == nil {
+				(*out)[key] = nil
+			} else {
+				in, out := &val, &outVal
+				*out = make(List, len(*in))
+				copy(*out, *in)
+			}
+			(*out)[key] = outVal
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TopologyStatusZone.
+func (in *TopologyStatusZone) DeepCopy() *TopologyStatusZone {
+	if in == nil {
+		return nil
+	}
+	out := new(TopologyStatusZone)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in TopologyStatusZoneMembers) DeepCopyInto(out *TopologyStatusZoneMembers) {
+	{
+		in := &in
+		*out = make(TopologyStatusZoneMembers, len(*in))
+		for key, val := range *in {
+			var outVal []string
+			if val == nil {
+				(*out)[key] = nil
+			} else {
+				in, out := &val, &outVal
+				*out = make(List, len(*in))
+				copy(*out, *in)
+			}
+			(*out)[key] = outVal
+		}
+		return
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TopologyStatusZoneMembers.
+func (in TopologyStatusZoneMembers) DeepCopy() TopologyStatusZoneMembers {
+	if in == nil {
+		return nil
+	}
+	out := new(TopologyStatusZoneMembers)
+	in.DeepCopyInto(out)
+	return *out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in TopologyStatusZones) DeepCopyInto(out *TopologyStatusZones) {
+	{
+		in := &in
+		*out = make(TopologyStatusZones, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+		return
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TopologyStatusZones.
+func (in TopologyStatusZones) DeepCopy() TopologyStatusZones {
+	if in == nil {
+		return nil
+	}
+	out := new(TopologyStatusZones)
+	in.DeepCopyInto(out)
+	return *out
+}

pkg/apis/deployment/v2alpha1/conditions.go

@@ -83,6 +83,8 @@ const (
 	ConditionTypeUpdating ConditionType = "Updating"
 	// ConditionTypeUpdateFailed indicates that runtime update failed
 	ConditionTypeUpdateFailed ConditionType = "UpdateFailed"
+	// ConditionTypeTopologyAware indicates that the member is deployed with TopologyAwareness.
+	ConditionTypeTopologyAware ConditionType = "TopologyAware"
 )
 
 // Condition represents one current condition of a deployment or deployment member.

pkg/apis/deployment/v2alpha1/deployment_spec.go

@@ -163,6 +163,9 @@ type DeploymentSpec struct {
 
 	// CommunicationMethod define communication method used in deployment
 	CommunicationMethod *DeploymentCommunicationMethod `json:"communicationMethod,omitempty"`
+
+	// Topology define topology adjustment details, Enterprise only
+	Topology *TopologySpec `json:"topology,omitempty"`
 }
 
 // GetAllowMemberRecreation returns member recreation policy based on group and settings

pkg/apis/deployment/v2alpha1/deployment_status.go

@@ -81,6 +81,8 @@ type DeploymentStatus struct {
 
 	// Agency keeps information about agency
 	Agency *DeploymentStatusAgencyInfo `json:"agency,omitempty"`
+
+	Topology *TopologyStatus `json:"topology,omitempty"`
 }
 
 // Equal checks for equality

pkg/apis/deployment/v2alpha1/list.go

@@ -0,0 +1,55 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+package v2alpha1
+
+import "sort"
+
+type List []string
+
+func (l List) Contains(v string) bool {
+	for _, z := range l {
+		if z == v {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (l List) Sort() {
+	sort.Strings(l)
+}
+
+func (l List) Remove(values ...string) List {
+	var m List
+
+	toRemove := List(values)
+
+	for _, v := range l {
+		if toRemove.Contains(v) {
+			continue
+		}
+
+		m = append(m, v)
+	}
+
+	return m
+}

pkg/apis/deployment/v2alpha1/member_status.go

@@ -41,6 +41,9 @@ type MemberStatus struct {
 	// ID holds the unique ID of the member.
 	// This id is also used within the ArangoDB cluster to identify this server.
 	ID string `json:"id"`
+	// UID holds the unique UID of the member.
+	// UID is created once member run in AddMember action.
+	UID types.UID `json:"uid,omitempty"`
 	// RID holds the ID of the member run.
 	// Value is updated in Pending Phase.
 	RID types.UID `json:"rid,omitempty"`
@@ -80,11 +83,14 @@ type MemberStatus struct {
 	Upgrade bool `json:"upgrade,omitempty"`
 	// Endpoint definition how member should be reachable
 	Endpoint *string `json:"endpoint,omitempty"`
+	// Topology define topology member status assignment
+	Topology *TopologyMemberStatus `json:"topology,omitempty"`
 }
 
 // Equal checks for equality
 func (s MemberStatus) Equal(other MemberStatus) bool {
 	return s.ID == other.ID &&
+		s.UID == other.UID &&
 		s.RID == other.RID &&
 		s.Phase == other.Phase &&
 		util.TimeCompareEqual(s.CreatedAt, other.CreatedAt) &&

pkg/apis/deployment/v2alpha1/member_status_list.go

@@ -138,10 +138,12 @@ func (l *MemberStatusList) removeByID(id string) error {
 	return errors.WithStack(errors.Wrapf(NotFoundError, "Member '%s' is not a member", id))
 }
 
+type MemberToRemoveSelector func(m MemberStatusList) (string, error)
+
 // SelectMemberToRemove selects a member from the given list that should
 // be removed in a scale down action.
 // Returns an error if the list is empty.
-func (l MemberStatusList) SelectMemberToRemove() (MemberStatus, error) {
+func (l MemberStatusList) SelectMemberToRemove(selectors ...MemberToRemoveSelector) (MemberStatus, error) {
 	if len(l) > 0 {
 		// Try to find member with phase to be removed
 		for _, m := range l {
@@ -165,6 +167,20 @@ func (l MemberStatusList) SelectMemberToRemove() (MemberStatus, error) {
 				return m, nil
 			}
 		}
+
+		// Run conditional picker
+		for _, selector := range selectors {
+			if m, err := selector(l); err != nil {
+				return MemberStatus{}, err
+			} else if m != "" {
+				if member, ok := l.ElementByID(m); ok {
+					return member, nil
+				} else {
+					return MemberStatus{}, errors.Newf("Unable to find member with id %s", m)
+				}
+			}
+		}
+
 		// Pick a random member that is in created state
 		perm := rand.Perm(len(l))
 		for _, idx := range perm {

pkg/apis/deployment/v2alpha1/plan.go

@@ -171,6 +171,11 @@ const (
 	ActionTypeRuntimeContainerImageUpdate ActionType = "RuntimeContainerImageUpdate"
 	// ActionTypeRuntimeContainerArgsLogLevelUpdate updates the container's executor arguments.
 	ActionTypeRuntimeContainerArgsLogLevelUpdate ActionType = "RuntimeContainerArgsLogLevelUpdate"
+
+	// Topology
+	ActionTypeTopologyEnable           ActionType = "TopologyEnable"
+	ActionTypeTopologyDisable          ActionType = "TopologyDisable"
+	ActionTypeTopologyMemberAssignment ActionType = "TopologyMemberAssignment"
 )
 
 const (
@@ -346,3 +351,45 @@ func (p Plan) Wrap(before, after Action) Plan {
 
 	return n
 }
+
+// AfterFirst adds actions when condition will return false
+func (p Plan) AfterFirst(condition func(a Action) bool, actions ...Action) Plan {
+	var r Plan
+	c := p
+	for {
+		if len(c) == 0 {
+			break
+		}
+
+		if !condition(c[0]) {
+			r = append(r, actions...)
+
+			r = append(r, c...)
+
+			break
+		}
+
+		r = append(r, c[0])
+
+		if len(c) == 1 {
+			break
+		}
+
+		c = c[1:]
+	}
+
+	return r
+}
+
+// Filter filter list of the actions
+func (p Plan) Filter(condition func(a Action) bool) Plan {
+	var r Plan
+
+	for _, a := range p {
+		if condition(a) {
+			r = append(r, a)
+		}
+	}
+
+	return r
+}

pkg/apis/deployment/v2alpha1/topology_member_status.go

@@ -0,0 +1,28 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+package v2alpha1
+
+import "k8s.io/apimachinery/pkg/types"
+
+type TopologyMemberStatus struct {
+	ID   types.UID `json:"id"`
+	Zone int       `json:"rack"`
+}

pkg/apis/deployment/v2alpha1/topology_spec.go

@@ -0,0 +1,53 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+package v2alpha1
+
+const DefaultTopologySpecLabel = "topology.kubernetes.io/zone"
+
+type TopologySpec struct {
+	Enabled bool    `json:"enabled,omitempty"`
+	Zones   int     `json:"zones,omitempty"`
+	Label   *string `json:"label,omitempty"`
+}
+
+func (t *TopologySpec) IsEnabled() bool {
+	if t == nil {
+		return false
+	}
+
+	return t.Enabled && t.Zones > 0
+}
+
+func (t *TopologySpec) GetZones() int {
+	if t == nil {
+		return 0
+	}
+
+	return t.Zones
+}
+
+func (t *TopologySpec) GetLabel() string {
+	if t == nil || t.Label == nil {
+		return DefaultTopologySpecLabel
+	}
+
+	return *t.Label
+}

pkg/apis/deployment/v2alpha1/topology_status.go

@@ -0,0 +1,175 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+package v2alpha1
+
+import (
+	"math"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/uuid"
+)
+
+type TopologyStatus struct {
+	ID types.UID `json:"id"`
+
+	Size int `json:"size,omitempty"`
+
+	Zones TopologyStatusZones `json:"zones,omitempty"`
+
+	Label string `json:"label,omitempty"`
+}
+
+func (t *TopologyStatus) GetLeastUsedZone(group ServerGroup) int {
+	if t == nil {
+		return -1
+	}
+
+	r, m := -1, math.MaxInt64
+
+	for i, z := range t.Zones {
+		if n, ok := z.Members[group.AsRoleAbbreviated()]; ok {
+			if v := len(n); v < m {
+				r, m = i, v
+			}
+		} else {
+			if v := 0; v < m {
+				r, m = i, v
+			}
+		}
+	}
+
+	return r
+}
+
+func (t *TopologyStatus) RegisterTopologyLabel(zone int, label string) bool {
+	if t == nil {
+		return false
+	}
+
+	if zone < 0 || zone >= t.Size {
+		return false
+	}
+
+	if t.Zones[zone].Labels.Contains(label) {
+		return false
+	}
+
+	t.Zones[zone].Labels = append(t.Zones[zone].Labels, label)
+	t.Zones[zone].Labels.Sort()
+
+	return true
+}
+
+func (t *TopologyStatus) RemoveMember(group ServerGroup, id string) bool {
+	if t == nil {
+		return false
+	}
+
+	for _, zone := range t.Zones {
+		if zone.RemoveMember(group, id) {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (t *TopologyStatus) IsTopologyOwned(m *TopologyMemberStatus) bool {
+	if t == nil {
+		return false
+	}
+
+	if m == nil {
+		return false
+	}
+
+	return t.ID == m.ID
+}
+
+func (t *TopologyStatus) Enabled() bool {
+	return t != nil
+}
+
+type TopologyStatusZones []TopologyStatusZone
+
+type TopologyStatusZoneMembers map[string]List
+
+type TopologyStatusZone struct {
+	ID int `json:"id"`
+
+	Labels List `json:"labels,omitempty"`
+
+	Members TopologyStatusZoneMembers `json:"members,omitempty"`
+}
+
+func (t *TopologyStatusZone) AddMember(group ServerGroup, id string) {
+	if t.Members == nil {
+		t.Members = TopologyStatusZoneMembers{}
+	}
+
+	t.Members[group.AsRoleAbbreviated()] = append(t.Members[group.AsRoleAbbreviated()], id)
+
+	t.Members[group.AsRoleAbbreviated()].Sort()
+}
+
+func (t *TopologyStatusZone) RemoveMember(group ServerGroup, id string) bool {
+	if t == nil {
+		return false
+	}
+	if t.Members == nil {
+		return false
+	}
+	if !t.Members[group.AsRoleAbbreviated()].Contains(id) {
+		return false
+	}
+	t.Members[group.AsRoleAbbreviated()] = t.Members[group.AsRoleAbbreviated()].Remove(id)
+	return true
+}
+
+func (t *TopologyStatusZone) Get(group ServerGroup) List {
+	if t == nil {
+		return nil
+	}
+
+	if v, ok := t.Members[group.AsRoleAbbreviated()]; ok {
+		return v
+	} else {
+		return nil
+	}
+}
+
+func NewTopologyStatus(spec *TopologySpec) *TopologyStatus {
+	if spec == nil {
+		return nil
+	}
+	zones := make(TopologyStatusZones, spec.Zones)
+
+	for i := 0; i < spec.Zones; i++ {
+		zones[i] = TopologyStatusZone{ID: i}
+	}
+
+	return &TopologyStatus{
+		ID:    uuid.NewUUID(),
+		Size:  spec.Zones,
+		Zones: zones,
+		Label: spec.GetLabel(),
+	}
+}

pkg/apis/deployment/v2alpha1/topology_status_test.go

@@ -0,0 +1,41 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+package v2alpha1
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func Test_GetLeastUsedZone(t *testing.T) {
+	v := NewTopologyStatus(&TopologySpec{Enabled: true, Zones: 3})
+
+	require.Equal(t, 0, v.GetLeastUsedZone(ServerGroupDBServers))
+
+	v.Zones[0].AddMember(ServerGroupDBServers, "M-0")
+
+	require.Equal(t, 1, v.GetLeastUsedZone(ServerGroupDBServers))
+
+	v.Zones[0].RemoveMember(ServerGroupDBServers, "M-0")
+
+	require.Equal(t, 0, v.GetLeastUsedZone(ServerGroupDBServers))
+}

pkg/apis/deployment/v2alpha1/zz_generated.deepcopy.go

@@ -610,6 +610,11 @@ func (in *DeploymentSpec) DeepCopyInto(out *DeploymentSpec) {
 		*out = new(DeploymentCommunicationMethod)
 		**out = **in
 	}
+	if in.Topology != nil {
+		in, out := &in.Topology, &out.Topology
+		*out = new(TopologySpec)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -684,6 +689,11 @@ func (in *DeploymentStatus) DeepCopyInto(out *DeploymentStatus) {
 		*out = new(DeploymentStatusAgencyInfo)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.Topology != nil {
+		in, out := &in.Topology, &out.Topology
+		*out = new(TopologyStatus)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -1066,6 +1076,26 @@ func (in *LifecycleSpec) DeepCopy() *LifecycleSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in List) DeepCopyInto(out *List) {
+	{
+		in := &in
+		*out = make(List, len(*in))
+		copy(*out, *in)
+		return
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new List.
+func (in List) DeepCopy() List {
+	if in == nil {
+		return nil
+	}
+	out := new(List)
+	in.DeepCopyInto(out)
+	return *out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *MemberStatus) DeepCopyInto(out *MemberStatus) {
 	*out = *in
@@ -1106,6 +1136,11 @@ func (in *MemberStatus) DeepCopyInto(out *MemberStatus) {
 		*out = new(string)
 		**out = **in
 	}
+	if in.Topology != nil {
+		in, out := &in.Topology, &out.Topology
+		*out = new(TopologyMemberStatus)
+		**out = **in
+	}
 	return
 }
 
@@ -2209,3 +2244,151 @@ func (in *Timeouts) DeepCopy() *Timeouts {
 	in.DeepCopyInto(out)
 	return out
 }
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *TopologyMemberStatus) DeepCopyInto(out *TopologyMemberStatus) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TopologyMemberStatus.
+func (in *TopologyMemberStatus) DeepCopy() *TopologyMemberStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(TopologyMemberStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *TopologySpec) DeepCopyInto(out *TopologySpec) {
+	*out = *in
+	if in.Label != nil {
+		in, out := &in.Label, &out.Label
+		*out = new(string)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TopologySpec.
+func (in *TopologySpec) DeepCopy() *TopologySpec {
+	if in == nil {
+		return nil
+	}
+	out := new(TopologySpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *TopologyStatus) DeepCopyInto(out *TopologyStatus) {
+	*out = *in
+	if in.Zones != nil {
+		in, out := &in.Zones, &out.Zones
+		*out = make(TopologyStatusZones, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TopologyStatus.
+func (in *TopologyStatus) DeepCopy() *TopologyStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(TopologyStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *TopologyStatusZone) DeepCopyInto(out *TopologyStatusZone) {
+	*out = *in
+	if in.Labels != nil {
+		in, out := &in.Labels, &out.Labels
+		*out = make(List, len(*in))
+		copy(*out, *in)
+	}
+	if in.Members != nil {
+		in, out := &in.Members, &out.Members
+		*out = make(TopologyStatusZoneMembers, len(*in))
+		for key, val := range *in {
+			var outVal []string
+			if val == nil {
+				(*out)[key] = nil
+			} else {
+				in, out := &val, &outVal
+				*out = make(List, len(*in))
+				copy(*out, *in)
+			}
+			(*out)[key] = outVal
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TopologyStatusZone.
+func (in *TopologyStatusZone) DeepCopy() *TopologyStatusZone {
+	if in == nil {
+		return nil
+	}
+	out := new(TopologyStatusZone)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in TopologyStatusZoneMembers) DeepCopyInto(out *TopologyStatusZoneMembers) {
+	{
+		in := &in
+		*out = make(TopologyStatusZoneMembers, len(*in))
+		for key, val := range *in {
+			var outVal []string
+			if val == nil {
+				(*out)[key] = nil
+			} else {
+				in, out := &val, &outVal
+				*out = make(List, len(*in))
+				copy(*out, *in)
+			}
+			(*out)[key] = outVal
+		}
+		return
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TopologyStatusZoneMembers.
+func (in TopologyStatusZoneMembers) DeepCopy() TopologyStatusZoneMembers {
+	if in == nil {
+		return nil
+	}
+	out := new(TopologyStatusZoneMembers)
+	in.DeepCopyInto(out)
+	return *out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in TopologyStatusZones) DeepCopyInto(out *TopologyStatusZones) {
+	{
+		in := &in
+		*out = make(TopologyStatusZones, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+		return
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TopologyStatusZones.
+func (in TopologyStatusZones) DeepCopy() TopologyStatusZones {
+	if in == nil {
+		return nil
+	}
+	out := new(TopologyStatusZones)
+	in.DeepCopyInto(out)
+	return *out
+}

pkg/deployment/context_impl.go

@@ -31,6 +31,8 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/arangodb/kube-arangodb/pkg/deployment/reconcile"
+
 	"github.com/arangodb/kube-arangodb/pkg/generated/clientset/versioned"
 	inspectorInterface "github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/secret"
@@ -361,19 +363,22 @@ func (d *Deployment) GetSyncServerClient(ctx context.Context, group api.ServerGr
 
 // CreateMember adds a new member to the given group.
 // If ID is non-empty, it will be used, otherwise a new ID is created.
-func (d *Deployment) CreateMember(ctx context.Context, group api.ServerGroup, id string) (string, error) {
+func (d *Deployment) CreateMember(ctx context.Context, group api.ServerGroup, id string, mods ...reconcile.CreateMemberMod) (string, error) {
 	log := d.deps.Log
-	status, lastVersion := d.GetStatus()
+	if err := d.WithStatusUpdateErr(ctx, func(s *api.DeploymentStatus) (bool, error) {
-	id, err := createMember(log, &status, group, id, d.apiObject)
+		nid, err := createMember(log, s, group, id, d.apiObject, mods...)
 		if err != nil {
 			log.Debug().Err(err).Str("group", group.AsRole()).Msg("Failed to create member")
-		return "", errors.WithStack(err)
+			return false, errors.WithStack(err)
 		}
-	// Save added member
+
-	if err := d.UpdateStatus(ctx, status, lastVersion); err != nil {
+		id = nid
-		log.Debug().Err(err).Msg("Updating CR status failed")
+
-		return "", errors.WithStack(err)
+		return true, nil
+	}); err != nil {
+		return "", err
 	}
+
 	// Create event about it
 	d.CreateEvent(k8sutil.NewMemberAddEvent(id, group.AsRole(), d.apiObject))
 
@@ -609,13 +614,17 @@ func (d *Deployment) GetArangoImage() string {
 	return d.config.ArangoImage
 }
 
-func (d *Deployment) WithStatusUpdate(ctx context.Context, action resources.DeploymentStatusUpdateFunc, force ...bool) error {
+func (d *Deployment) WithStatusUpdateErr(ctx context.Context, action resources.DeploymentStatusUpdateErrFunc, force ...bool) error {
 	d.status.mutex.Lock()
 	defer d.status.mutex.Unlock()
 
 	status, version := d.getStatus()
 
-	changed := action(&status)
+	changed, err := action(&status)
+
+	if err != nil {
+		return err
+	}
 
 	if !changed {
 		return nil
@@ -624,6 +633,12 @@ func (d *Deployment) WithStatusUpdate(ctx context.Context, action resources.Depl
 	return d.updateStatus(ctx, status, version, force...)
 }
 
+func (d *Deployment) WithStatusUpdate(ctx context.Context, action resources.DeploymentStatusUpdateFunc, force ...bool) error {
+	return d.WithStatusUpdateErr(ctx, func(s *api.DeploymentStatus) (bool, error) {
+		return action(s), nil
+	}, force...)
+}
+
 func (d *Deployment) SecretsInterface() k8sutil.SecretInterface {
 	return d.GetKubeCli().CoreV1().Secrets(d.GetNamespace())
 }

pkg/deployment/deployment.go

@@ -275,14 +275,9 @@ func (d *Deployment) run() {
 			}
 		}
 
-		// Create members
+		// Create initial topology
-		if err := d.createInitialMembers(context.TODO(), d.apiObject); err != nil {
+		if err := d.createInitialTopology(context.TODO()); err != nil {
-			d.CreateEvent(k8sutil.NewErrorEvent("Failed to create initial members", err, d.GetAPIObject()))
+			d.CreateEvent(k8sutil.NewErrorEvent("Failed to create initial topology", err, d.GetAPIObject()))
-		}
-
-		// Create Pod Disruption Budgets
-		if err := d.resources.EnsurePDBs(context.TODO()); err != nil {
-			d.CreateEvent(k8sutil.NewErrorEvent("Failed to create pdbs", err, d.GetAPIObject()))
 		}
 
 		status, lastVersion := d.GetStatus()

pkg/deployment/members.community.go

@@ -0,0 +1,31 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+// +build !enterprise
+
+package deployment
+
+import (
+	"context"
+)
+
+func (d *Deployment) createInitialTopology(ctx context.Context) error {
+	return nil
+}

pkg/deployment/members.go

@@ -25,6 +25,10 @@ package deployment
 import (
 	"context"
 
+	"k8s.io/apimachinery/pkg/util/uuid"
+
+	"github.com/arangodb/kube-arangodb/pkg/deployment/reconcile"
+
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/names"
 
 	"github.com/arangodb/kube-arangodb/pkg/util/errors"
@@ -36,41 +40,6 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
 )
 
-// createInitialMembers creates all members needed for the initial state of the deployment.
-// Note: This does not create any pods of PVCs
-func (d *Deployment) createInitialMembers(ctx context.Context, apiObject *api.ArangoDeployment) error {
-	log := d.deps.Log
-	log.Debug().Msg("creating initial members...")
-
-	// Go over all groups and create members
-	var events []*k8sutil.Event
-	status, lastVersion := d.GetStatus()
-	if err := apiObject.ForeachServerGroup(func(group api.ServerGroup, spec api.ServerGroupSpec, members *api.MemberStatusList) error {
-		for len(*members) < spec.GetCount() {
-			id, err := createMember(log, &status, group, "", apiObject)
-			if err != nil {
-				return errors.WithStack(err)
-			}
-			events = append(events, k8sutil.NewMemberAddEvent(id, group.AsRole(), apiObject))
-		}
-		return nil
-	}, &status); err != nil {
-		return errors.WithStack(err)
-	}
-
-	// Save status
-	log.Debug().Msg("saving initial members...")
-	if err := d.UpdateStatus(ctx, status, lastVersion); err != nil {
-		return errors.WithStack(err)
-	}
-	// Save events
-	for _, evt := range events {
-		d.CreateEvent(evt)
-	}
-
-	return nil
-}
-
 func (d *Deployment) createAgencyMapping(ctx context.Context) error {
 	spec := d.GetSpec()
 	status, _ := d.GetStatus()
@@ -116,10 +85,29 @@ func (d *Deployment) createAgencyMapping(ctx context.Context) error {
 // createMember creates member and adds it to the applicable member list.
 // Note: This does not create any pods of PVCs
 // Note: The updated status is not yet written to the apiserver.
-func createMember(log zerolog.Logger, status *api.DeploymentStatus, group api.ServerGroup, id string, apiObject *api.ArangoDeployment) (string, error) {
+func createMember(log zerolog.Logger, status *api.DeploymentStatus, group api.ServerGroup, id string, apiObject *api.ArangoDeployment, mods ...reconcile.CreateMemberMod) (string, error) {
+	m, err := renderMember(log, status, group, id, apiObject)
+	if err != nil {
+		return "", err
+	}
+
+	for _, mod := range mods {
+		if err := mod(status, group, m); err != nil {
+			return "", err
+		}
+	}
+
+	if err := status.Members.Add(*m, group); err != nil {
+		return "", err
+	}
+
+	return m.ID, nil
+}
+
+func renderMember(log zerolog.Logger, status *api.DeploymentStatus, group api.ServerGroup, id string, apiObject *api.ArangoDeployment) (*api.MemberStatus, error) {
 	if group == api.ServerGroupAgents {
 		if status.Agency == nil {
-			return "", errors.New("Agency is not yet defined")
+			return nil, errors.New("Agency is not yet defined")
 		}
 		// In case of agents we need to use hardcoded ids
 		if id == "" {
@@ -142,7 +130,7 @@ func createMember(log zerolog.Logger, status *api.DeploymentStatus, group api.Se
 		}
 	}
 	if id == "" {
-		return "nil", errors.New("Unable to get ID")
+		return nil, errors.New("Unable to get ID")
 	}
 	deploymentName := apiObject.GetName()
 	role := group.AsRole()
@@ -150,79 +138,71 @@ func createMember(log zerolog.Logger, status *api.DeploymentStatus, group api.Se
 	switch group {
 	case api.ServerGroupSingle:
 		log.Debug().Str("id", id).Msg("Adding single server")
-		if err := status.Members.Add(api.MemberStatus{
+		return &api.MemberStatus{
 			ID:                        id,
+			UID:                       uuid.NewUUID(),
 			CreatedAt:                 metav1.Now(),
 			Phase:                     api.MemberPhaseNone,
 			PersistentVolumeClaimName: k8sutil.CreatePersistentVolumeClaimName(deploymentName, role, id),
 			PodName:                   "",
 			Image:                     apiObject.Status.CurrentImage,
-		}, group); err != nil {
+		}, nil
-			return "", errors.WithStack(err)
-		}
 	case api.ServerGroupAgents:
 		log.Debug().Str("id", id).Msg("Adding agent")
-		if err := status.Members.Add(api.MemberStatus{
+		return &api.MemberStatus{
 			ID:                        id,
+			UID:                       uuid.NewUUID(),
 			CreatedAt:                 metav1.Now(),
 			Phase:                     api.MemberPhaseNone,
 			PersistentVolumeClaimName: k8sutil.CreatePersistentVolumeClaimName(deploymentName, role, id),
 			PodName:                   "",
 			Image:                     apiObject.Status.CurrentImage,
-		}, group); err != nil {
+		}, nil
-			return "", errors.WithStack(err)
-		}
 	case api.ServerGroupDBServers:
 		log.Debug().Str("id", id).Msg("Adding dbserver")
-		if err := status.Members.Add(api.MemberStatus{
+		return &api.MemberStatus{
 			ID:                        id,
+			UID:                       uuid.NewUUID(),
 			CreatedAt:                 metav1.Now(),
 			Phase:                     api.MemberPhaseNone,
 			PersistentVolumeClaimName: k8sutil.CreatePersistentVolumeClaimName(deploymentName, role, id),
 			PodName:                   "",
 			Image:                     apiObject.Status.CurrentImage,
-		}, group); err != nil {
+		}, nil
-			return "", errors.WithStack(err)
-		}
 	case api.ServerGroupCoordinators:
 		log.Debug().Str("id", id).Msg("Adding coordinator")
-		if err := status.Members.Add(api.MemberStatus{
+		return &api.MemberStatus{
 			ID:                        id,
+			UID:                       uuid.NewUUID(),
 			CreatedAt:                 metav1.Now(),
 			Phase:                     api.MemberPhaseNone,
 			PersistentVolumeClaimName: "",
 			PodName:                   "",
 			Image:                     apiObject.Status.CurrentImage,
-		}, group); err != nil {
+		}, nil
-			return "", errors.WithStack(err)
-		}
 	case api.ServerGroupSyncMasters:
 		log.Debug().Str("id", id).Msg("Adding syncmaster")
-		if err := status.Members.Add(api.MemberStatus{
+		return &api.MemberStatus{
 			ID:                        id,
+			UID:                       uuid.NewUUID(),
 			CreatedAt:                 metav1.Now(),
 			Phase:                     api.MemberPhaseNone,
 			PersistentVolumeClaimName: "",
 			PodName:                   "",
 			Image:                     apiObject.Status.CurrentImage,
-		}, group); err != nil {
+		}, nil
-			return "", errors.WithStack(err)
-		}
 	case api.ServerGroupSyncWorkers:
 		log.Debug().Str("id", id).Msg("Adding syncworker")
-		if err := status.Members.Add(api.MemberStatus{
+		return &api.MemberStatus{
 			ID:                        id,
+			UID:                       uuid.NewUUID(),
 			CreatedAt:                 metav1.Now(),
 			Phase:                     api.MemberPhaseNone,
 			PersistentVolumeClaimName: "",
 			PodName:                   "",
 			Image:                     apiObject.Status.CurrentImage,
-		}, group); err != nil {
+		}, nil
-			return "", errors.WithStack(err)
-		}
 	default:
-		return "", errors.WithStack(errors.Newf("Unknown server group %d", group))
+		return nil, errors.WithStack(errors.Newf("Unknown server group %d", group))
 	}
-
-	return id, nil
 }

pkg/deployment/reconcile/action_add_member.go

@@ -27,6 +27,8 @@ import (
 	"context"
 	"time"
 
+	"github.com/arangodb/kube-arangodb/pkg/deployment/topology"
+
 	"github.com/arangodb/kube-arangodb/pkg/util/errors"
 
 	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
@@ -67,7 +69,7 @@ type actionAddMember struct {
 // Returns true if the action is completely finished, false in case
 // the start time needs to be recorded and a ready condition needs to be checked.
 func (a *actionAddMember) Start(ctx context.Context) (bool, error) {
-	newID, err := a.actionCtx.CreateMember(ctx, a.action.Group, a.action.MemberID)
+	newID, err := a.actionCtx.CreateMember(ctx, a.action.Group, a.action.MemberID, topology.WithTopologyMod)
 	if err != nil {
 		log.Debug().Err(err).Msg("Failed to create member")
 		return false, errors.WithStack(err)
@@ -90,7 +92,9 @@ func (a *actionAddMember) ActionPlanAppender(current api.Plan) (api.Plan, bool)
 	}
 
 	if len(app) > 0 {
-		return append(app, current...), true
+		return app.AfterFirst(func(a api.Action) bool {
+			return a.Type == api.ActionTypeAddMember
+		}, app...), true
 	}
 
 	return current, false

pkg/deployment/reconcile/action_context.go

@@ -89,7 +89,7 @@ type ActionContext interface {
 	GetMemberStatusAndGroupByID(id string) (api.MemberStatus, api.ServerGroup, bool)
 	// CreateMember adds a new member to the given group.
 	// If ID is non-empty, it will be used, otherwise a new ID is created.
-	CreateMember(ctx context.Context, group api.ServerGroup, id string) (string, error)
+	CreateMember(ctx context.Context, group api.ServerGroup, id string, mods ...CreateMemberMod) (string, error)
 	// UpdateMember updates the deployment status wrt the given member.
 	UpdateMember(ctx context.Context, member api.MemberStatus) error
 	// RemoveMemberByID removes a member with given id.
@@ -236,6 +236,10 @@ func (ac *actionContext) GetBackup(ctx context.Context, backup string) (*backupA
 	return ac.context.GetBackup(ctx, backup)
 }
 
+func (ac *actionContext) WithStatusUpdateErr(ctx context.Context, action resources.DeploymentStatusUpdateErrFunc, force ...bool) error {
+	return ac.context.WithStatusUpdateErr(ctx, action, force...)
+}
+
 func (ac *actionContext) WithStatusUpdate(ctx context.Context, action resources.DeploymentStatusUpdateFunc, force ...bool) error {
 	return ac.context.WithStatusUpdate(ctx, action, force...)
 }
@@ -351,8 +355,8 @@ func (ac *actionContext) GetMemberStatusAndGroupByID(id string) (api.MemberStatu
 
 // CreateMember adds a new member to the given group.
 // If ID is non-empty, it will be used, otherwise a new ID is created.
-func (ac *actionContext) CreateMember(ctx context.Context, group api.ServerGroup, id string) (string, error) {
+func (ac *actionContext) CreateMember(ctx context.Context, group api.ServerGroup, id string, mods ...CreateMemberMod) (string, error) {
-	result, err := ac.context.CreateMember(ctx, group, id)
+	result, err := ac.context.CreateMember(ctx, group, id, mods...)
 	if err != nil {
 		return "", errors.WithStack(err)
 	}

pkg/deployment/reconcile/action_mark_to_remove_member.go

@@ -50,7 +50,7 @@ type actionMarkToRemove struct {
 }
 
 func (a *actionMarkToRemove) Start(ctx context.Context) (bool, error) {
-	if a.action.Group != api.ServerGroupDBServers && a.action.Group != api.ServerGroupAgents {
+	if a.action.Group != api.ServerGroupDBServers && a.action.Group != api.ServerGroupAgents && a.action.Group != api.ServerGroupCoordinators {
 		return true, nil
 	}
 

pkg/deployment/reconcile/action_member_phase_update.go

@@ -35,7 +35,7 @@ func init() {
 }
 
 const (
-	ActionTypeMemberPhaseUpdatePhaseKey string = "phase"
+	actionTypeMemberPhaseUpdatePhaseKey string = "phase"
 )
 
 func newMemberPhaseUpdate(log zerolog.Logger, action api.Action, actionCtx ActionContext) Action {
@@ -60,7 +60,7 @@ func (a *memberPhaseUpdateAction) Start(ctx context.Context) (bool, error) {
 		return true, nil
 	}
 
-	phaseString, ok := a.action.Params[ActionTypeMemberPhaseUpdatePhaseKey]
+	phaseString, ok := a.action.Params[actionTypeMemberPhaseUpdatePhaseKey]
 	if !ok {
 		log.Error().Msg("Phase not defined")
 		return true, nil

pkg/deployment/reconcile/action_remove_member.go

@@ -127,6 +127,11 @@ func (a *actionRemoveMember) Start(ctx context.Context) (bool, error) {
 	if err := a.actionCtx.RemoveMemberByID(ctx, a.action.MemberID); err != nil {
 		return false, errors.WithStack(err)
 	}
+	if err := a.actionCtx.WithStatusUpdate(ctx, func(s *api.DeploymentStatus) bool {
+		return s.Topology.RemoveMember(a.action.Group, a.action.MemberID)
+	}); err != nil {
+		return false, errors.WithStack(err)
+	}
 	// Check that member has been removed
 	if _, found := a.actionCtx.GetMemberStatusByID(a.action.MemberID); found {
 		return false, errors.WithStack(errors.Newf("Member %s still exists", a.action.MemberID))

pkg/deployment/reconcile/action_topology_disable.community.go

@@ -0,0 +1,27 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+// +build !enterprise
+
+package reconcile
+
+type topologyDisable struct {
+	actionEmpty
+}

pkg/deployment/reconcile/action_topology_disable.go

@@ -0,0 +1,38 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+package reconcile
+
+import (
+	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
+	"github.com/rs/zerolog"
+)
+
+func init() {
+	registerAction(api.ActionTypeTopologyDisable, newTopologyDisable)
+}
+
+func newTopologyDisable(log zerolog.Logger, action api.Action, actionCtx ActionContext) Action {
+	a := &topologyDisable{}
+
+	a.actionImpl = newActionImplDefRef(log, action, actionCtx, defaultTimeout)
+
+	return a
+}

pkg/deployment/reconcile/action_topology_enable.community.go

@@ -1,7 +1,7 @@
 //
 // DISCLAIMER
 //
-// Copyright 2021 ArangoDB GmbH, Cologne, Germany
+// Copyright 2020 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.
@@ -17,7 +17,11 @@
 //
 // Copyright holder is ArangoDB GmbH, Cologne, Germany
 //
-// Author Adam Janikowski
+
-//
+// +build !enterprise
 
 package reconcile
+
+type topologyEnable struct {
+	actionEmpty
+}

pkg/deployment/reconcile/action_topology_enable.go

@@ -0,0 +1,38 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+package reconcile
+
+import (
+	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
+	"github.com/rs/zerolog"
+)
+
+func init() {
+	registerAction(api.ActionTypeTopologyEnable, newTopologyEnable)
+}
+
+func newTopologyEnable(log zerolog.Logger, action api.Action, actionCtx ActionContext) Action {
+	a := &topologyEnable{}
+
+	a.actionImpl = newActionImplDefRef(log, action, actionCtx, defaultTimeout)
+
+	return a
+}

pkg/deployment/reconcile/action_topology_member_assignment.community.go

@@ -0,0 +1,27 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+// +build !enterprise
+
+package reconcile
+
+type topologyMemberAssignment struct {
+	actionEmpty
+}

pkg/deployment/reconcile/action_topology_member_assignment.go

@@ -0,0 +1,38 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+package reconcile
+
+import (
+	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
+	"github.com/rs/zerolog"
+)
+
+func init() {
+	registerAction(api.ActionTypeTopologyMemberAssignment, newTopologyMemberAssignment)
+}
+
+func newTopologyMemberAssignment(log zerolog.Logger, action api.Action, actionCtx ActionContext) Action {
+	a := &topologyMemberAssignment{}
+
+	a.actionImpl = newActionImplDefRef(log, action, actionCtx, defaultTimeout)
+
+	return a
+}

pkg/deployment/reconcile/context.go

@@ -39,6 +39,8 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
 )
 
+type CreateMemberMod func(s *api.DeploymentStatus, g api.ServerGroup, m *api.MemberStatus) error
+
 // Context provides methods to the reconcile package.
 type Context interface {
 	resources.DeploymentStatusUpdate
@@ -77,7 +79,7 @@ type Context interface {
 	// CreateMember adds a new member to the given group.
 	// If ID is non-empty, it will be used, otherwise a new ID is created.
 	// Returns ID, error
-	CreateMember(ctx context.Context, group api.ServerGroup, id string) (string, error)
+	CreateMember(ctx context.Context, group api.ServerGroup, id string, mods ...CreateMemberMod) (string, error)
 	// GetPod returns pod.
 	GetPod(ctx context.Context, podName string) (*v1.Pod, error)
 	// DeletePod deletes a pod with given name in the namespace

pkg/deployment/reconcile/plan_builder_high.go

@@ -89,6 +89,7 @@ func createHighPlan(ctx context.Context, log zerolog.Logger, apiObject k8sutil.A
 		ApplyIfEmpty(createCleanOutPlan).
 		ApplyIfEmpty(updateMemberUpdateConditionsPlan).
 		ApplyIfEmpty(updateMemberRotationConditionsPlan).
+		ApplyIfEmpty(createTopologyMemberConditionPlan).
 		Plan(), true
 }
 
@@ -132,7 +133,7 @@ func updateMemberPhasePlan(ctx context.Context,
 					api.NewAction(api.ActionTypeArangoMemberUpdatePodStatus, group, m.ID, "Propagating status of pod"))
 
 				p = append(p, api.NewAction(api.ActionTypeMemberPhaseUpdate, group, m.ID,
-					"Move to Pending phase").AddParam(ActionTypeMemberPhaseUpdatePhaseKey, api.MemberPhasePending.String()))
+					"Move to Pending phase").AddParam(actionTypeMemberPhaseUpdatePhaseKey, api.MemberPhasePending.String()))
 
 				plan = append(plan, p...)
 			}

pkg/deployment/reconcile/plan_builder_normal.go

@@ -81,14 +81,20 @@ func createNormalPlan(ctx context.Context, log zerolog.Logger, apiObject k8sutil
 	}
 
 	return newPlanAppender(NewWithPlanBuilder(ctx, log, apiObject, spec, status, cachedStatus, builderCtx), nil).
+		// Adjust topology settings
+		ApplyIfEmpty(createTopologyMemberAdjustmentPlan).
+		// Define topology
+		ApplyIfEmpty(createTopologyEnablementPlan).
+		// Check for scale up
+		ApplyIfEmpty(createScaleUPMemberPlan).
 		// Check for failed members
 		ApplyIfEmpty(createMemberFailedRestorePlan).
+		// Check for scale up/down
+		ApplyIfEmpty(createScaleMemberPlan).
 		// Update status
 		ApplySubPlanIfEmpty(createEncryptionKeyStatusPropagatedFieldUpdate, createEncryptionKeyStatusUpdate).
 		ApplyIfEmpty(createTLSStatusUpdate).
 		ApplyIfEmpty(createJWTStatusUpdate).
-		// Check for scale up/down
-		ApplyIfEmpty(createScaleMemberPlan).
 		// Check for cleaned out dbserver in created state
 		ApplyIfEmpty(createRemoveCleanedDBServersPlan).
 		// Check for members to be removed

pkg/deployment/reconcile/plan_builder_rotate_upgrade.go

@@ -188,7 +188,7 @@ func createRotateOrUpgradePlanInternal(log zerolog.Logger, apiObject k8sutil.API
 			}
 
 			if pod.Annotations != nil {
-				if _, ok := pod.Annotations[deployment.ArangoDeploymentPodReplaceAnnotation]; ok && (group == api.ServerGroupDBServers || group == api.ServerGroupAgents) {
+				if _, ok := pod.Annotations[deployment.ArangoDeploymentPodReplaceAnnotation]; ok && (group == api.ServerGroupDBServers || group == api.ServerGroupAgents || group == api.ServerGroupCoordinators) {
 					if !m.Conditions.IsTrue(api.ConditionTypeMarkedToRemove) {
 						newPlan = api.Plan{api.NewAction(api.ActionTypeMarkToRemoveMember, group, m.ID, "Replace flag present")}
 						continue

pkg/deployment/reconcile/plan_builder_scale.go

@@ -31,6 +31,15 @@ import (
 	"github.com/rs/zerolog"
 )
 
+func createScaleUPMemberPlan(ctx context.Context,
+	log zerolog.Logger, apiObject k8sutil.APIObject,
+	spec api.DeploymentSpec, status api.DeploymentStatus,
+	cachedStatus inspectorInterface.Inspector, context PlanBuilderContext) api.Plan {
+	return createScaleMemberPlan(ctx, log, apiObject, spec, status, cachedStatus, context).Filter(func(a api.Action) bool {
+		return a.Type == api.ActionTypeAddMember
+	})
+}
+
 func createScaleMemberPlan(ctx context.Context,
 	log zerolog.Logger, apiObject k8sutil.APIObject,
 	spec api.DeploymentSpec, status api.DeploymentStatus,
@@ -42,24 +51,30 @@ func createScaleMemberPlan(ctx context.Context,
 	case api.DeploymentModeSingle:
 		// Never scale down
 	case api.DeploymentModeActiveFailover:
-		// Only scale singles
+		// Only scale agents & singles
-		plan = append(plan, createScalePlan(log, status.Members.Single, api.ServerGroupSingle, spec.Single.GetCount())...)
+		if a := status.Agency; a != nil && a.Size != nil {
+			plan = append(plan, createScalePlan(log, status, status.Members.Agents, api.ServerGroupAgents, int(*a.Size))...)
+		}
+		plan = append(plan, createScalePlan(log, status, status.Members.Single, api.ServerGroupSingle, spec.Single.GetCount())...)
 	case api.DeploymentModeCluster:
-		// Scale dbservers, coordinators
+		// Scale agents, dbservers, coordinators
-		plan = append(plan, createScalePlan(log, status.Members.DBServers, api.ServerGroupDBServers, spec.DBServers.GetCount())...)
+		if a := status.Agency; a != nil && a.Size != nil {
-		plan = append(plan, createScalePlan(log, status.Members.Coordinators, api.ServerGroupCoordinators, spec.Coordinators.GetCount())...)
+			plan = append(plan, createScalePlan(log, status, status.Members.Agents, api.ServerGroupAgents, int(*a.Size))...)
+		}
+		plan = append(plan, createScalePlan(log, status, status.Members.DBServers, api.ServerGroupDBServers, spec.DBServers.GetCount())...)
+		plan = append(plan, createScalePlan(log, status, status.Members.Coordinators, api.ServerGroupCoordinators, spec.Coordinators.GetCount())...)
 	}
 	if spec.GetMode().SupportsSync() {
 		// Scale syncmasters & syncworkers
-		plan = append(plan, createScalePlan(log, status.Members.SyncMasters, api.ServerGroupSyncMasters, spec.SyncMasters.GetCount())...)
+		plan = append(plan, createScalePlan(log, status, status.Members.SyncMasters, api.ServerGroupSyncMasters, spec.SyncMasters.GetCount())...)
-		plan = append(plan, createScalePlan(log, status.Members.SyncWorkers, api.ServerGroupSyncWorkers, spec.SyncWorkers.GetCount())...)
+		plan = append(plan, createScalePlan(log, status, status.Members.SyncWorkers, api.ServerGroupSyncWorkers, spec.SyncWorkers.GetCount())...)
 	}
 
 	return plan
 }
 
 // createScalePlan creates a scaling plan for a single server group
-func createScalePlan(log zerolog.Logger, members api.MemberStatusList, group api.ServerGroup, count int) api.Plan {
+func createScalePlan(log zerolog.Logger, status api.DeploymentStatus, members api.MemberStatusList, group api.ServerGroup, count int) api.Plan {
 	var plan api.Plan
 	if len(members) < count {
 		// Scale up
@@ -75,7 +90,7 @@ func createScalePlan(log zerolog.Logger, members api.MemberStatusList, group api
 			Msg("Creating scale-up plan")
 	} else if len(members) > count {
 		// Note, we scale down 1 member at a time
-		if m, err := members.SelectMemberToRemove(); err != nil {
+		if m, err := members.SelectMemberToRemove(topologyMissingMemberToRemoveSelector(status.Topology), topologyAwarenessMemberToRemoveSelector(group, status.Topology)); err != nil {
 			log.Warn().Err(err).Str("role", group.AsRole()).Msg("Failed to select member to remove")
 		} else {
 
@@ -102,7 +117,7 @@ func createReplaceMemberPlan(ctx context.Context,
 
 	var plan api.Plan
 
-	// Replace is only allowed for DBServers & Agents
+	// Replace is only allowed for Coordinators, DBServers & Agents
 	status.Members.ForeachServerInGroups(func(group api.ServerGroup, list api.MemberStatusList) error {
 		for _, member := range list {
 			if !plan.IsEmpty() {
@@ -118,6 +133,12 @@ func createReplaceMemberPlan(ctx context.Context,
 						Str("role", group.AsRole()).
 						Msg("Creating replacement plan")
 					return nil
+				case api.ServerGroupCoordinators:
+					plan = append(plan, api.NewAction(api.ActionTypeRemoveMember, group, member.ID))
+					log.Debug().
+						Str("role", group.AsRole()).
+						Msg("Creating replacement plan")
+					return nil
 				case api.ServerGroupAgents:
 					plan = append(plan, api.NewAction(api.ActionTypeRemoveMember, group, member.ID),
 						api.NewAction(api.ActionTypeAddMember, group, "").
@@ -132,7 +153,7 @@ func createReplaceMemberPlan(ctx context.Context,
 		}
 
 		return nil
-	}, api.ServerGroupAgents, api.ServerGroupDBServers)
+	}, api.ServerGroupAgents, api.ServerGroupDBServers, api.ServerGroupCoordinators)
 
 	return plan
 }

pkg/deployment/reconcile/plan_builder_test.go

@@ -79,6 +79,10 @@ type testContext struct {
 	RecordedEvent    *k8sutil.Event
 }
 
+func (c *testContext) WithStatusUpdateErr(ctx context.Context, action resources.DeploymentStatusUpdateErrFunc, force ...bool) error {
+	panic("implement me")
+}
+
 func (c *testContext) GetKubeCli() kubernetes.Interface {
 	panic("implement me")
 }
@@ -224,7 +228,7 @@ func (c *testContext) GetSyncServerClient(ctx context.Context, group api.ServerG
 	panic("implement me")
 }
 
-func (c *testContext) CreateMember(_ context.Context, group api.ServerGroup, id string) (string, error) {
+func (c *testContext) CreateMember(_ context.Context, group api.ServerGroup, id string, mods ...CreateMemberMod) (string, error) {
 	panic("implement me")
 }
 

pkg/deployment/reconcile/plan_builder_topology.community.go

@@ -0,0 +1,61 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+// +build !enterprise
+
+package reconcile
+
+import (
+	"context"
+
+	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
+	inspectorInterface "github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector"
+	"github.com/rs/zerolog"
+)
+
+func createTopologyEnablementPlan(ctx context.Context,
+	log zerolog.Logger, apiObject k8sutil.APIObject,
+	spec api.DeploymentSpec, status api.DeploymentStatus,
+	cachedStatus inspectorInterface.Inspector, context PlanBuilderContext) api.Plan {
+	return nil
+}
+
+func createTopologyMemberConditionPlan(ctx context.Context,
+	log zerolog.Logger, apiObject k8sutil.APIObject,
+	spec api.DeploymentSpec, status api.DeploymentStatus,
+	cachedStatus inspectorInterface.Inspector, context PlanBuilderContext) api.Plan {
+	return nil
+}
+
+func createTopologyMemberAdjustmentPlan(ctx context.Context,
+	log zerolog.Logger, apiObject k8sutil.APIObject,
+	spec api.DeploymentSpec, status api.DeploymentStatus,
+	cachedStatus inspectorInterface.Inspector, context PlanBuilderContext) api.Plan {
+	return nil
+}
+
+func topologyMissingMemberToRemoveSelector(s *api.TopologyStatus) api.MemberToRemoveSelector {
+	return nil
+}
+
+func topologyAwarenessMemberToRemoveSelector(g api.ServerGroup, s *api.TopologyStatus) api.MemberToRemoveSelector {
+	return nil
+}

pkg/deployment/reconcile/reconciler.go

@@ -53,14 +53,7 @@ func (r *Reconciler) CheckDeployment(ctx context.Context) error {
 
 	if spec.GetMode().HasCoordinators() {
 		// Check if there are coordinators
-		if len(status.Members.Coordinators) == 0 {
+		if status.Members.Coordinators.AllFailed() {
-			// No more coordinators! Take immediate action
-			r.log.Error().Msg("No Coordinator members! Create one member immediately")
-			_, err := r.context.CreateMember(ctx, api.ServerGroupCoordinators, "")
-			if err != nil {
-				return err
-			}
-		} else if status.Members.Coordinators.AllFailed() {
 			r.log.Error().Msg("All coordinators failed - reset")
 			for _, m := range status.Members.Coordinators {
 				if err := r.context.DeletePod(ctx, m.PodName); err != nil {

pkg/deployment/resources/context.go

@@ -52,9 +52,12 @@ type ServerGroupIterator interface {
 	ForeachServerGroup(cb api.ServerGroupFunc, status *api.DeploymentStatus) error
 }
 
+type DeploymentStatusUpdateErrFunc func(s *api.DeploymentStatus) (bool, error)
 type DeploymentStatusUpdateFunc func(s *api.DeploymentStatus) bool
 
 type DeploymentStatusUpdate interface {
+	// WithStatusUpdateErr update status of ArangoDeployment with defined modifier. If action returns True action is taken
+	WithStatusUpdateErr(ctx context.Context, action DeploymentStatusUpdateErrFunc, force ...bool) error
 	// WithStatusUpdate update status of ArangoDeployment with defined modifier. If action returns True action is taken
 	WithStatusUpdate(ctx context.Context, action DeploymentStatusUpdateFunc, force ...bool) error
 }

pkg/deployment/resources/pod_creator.go

@@ -623,6 +623,14 @@ func (r *Resources) createPodForMember(ctx context.Context, spec api.DeploymentS
 	// Record new member phase
 	m.Phase = newPhase
 
+	if status.Topology.Enabled() {
+		if m.Topology != nil && m.Topology.ID == status.Topology.ID {
+			m.Conditions.Update(api.ConditionTypeTopologyAware, true, "Topology Aware", "Topology Aware")
+		} else {
+			m.Conditions.Update(api.ConditionTypeTopologyAware, false, "Topology spec missing", "Topology spec missing")
+		}
+	}
+
 	m.Upgrade = false
 	r.log.Info().Str("pod", m.PodName).Msgf("Updating member")
 	if err := status.Members.Update(m, group); err != nil {

pkg/deployment/resources/pod_creator_arangod.go

@@ -27,6 +27,8 @@ import (
 	"math"
 	"os"
 
+	"github.com/arangodb/kube-arangodb/pkg/deployment/topology"
+
 	"github.com/arangodb/kube-arangodb/pkg/deployment/features"
 
 	"github.com/arangodb/kube-arangodb/pkg/util/collection"
@@ -267,6 +269,8 @@ func (m *MemberArangoDPod) GetPodAntiAffinity() *core.PodAntiAffinity {
 
 	pod.AppendPodAntiAffinityDefault(m, &a)
 
+	pod.MergePodAntiAffinity(&a, topology.GetTopologyAffinityRules(m.context.GetName(), m.group, m.deploymentStatus.Topology, m.status.Topology).PodAntiAffinity)
+
 	pod.MergePodAntiAffinity(&a, m.groupSpec.AntiAffinity)
 
 	return pod.ReturnPodAntiAffinityOrNil(a)
@@ -277,6 +281,8 @@ func (m *MemberArangoDPod) GetPodAffinity() *core.PodAffinity {
 
 	pod.MergePodAffinity(&a, m.groupSpec.Affinity)
 
+	pod.MergePodAffinity(&a, topology.GetTopologyAffinityRules(m.context.GetName(), m.group, m.deploymentStatus.Topology, m.status.Topology).PodAffinity)
+
 	return pod.ReturnPodAffinityOrNil(a)
 }
 
@@ -287,6 +293,8 @@ func (m *MemberArangoDPod) GetNodeAffinity() *core.NodeAffinity {
 
 	pod.MergeNodeAffinity(&a, m.groupSpec.NodeAffinity)
 
+	pod.MergeNodeAffinity(&a, topology.GetTopologyAffinityRules(m.context.GetName(), m.group, m.deploymentStatus.Topology, m.status.Topology).NodeAffinity)
+
 	return pod.ReturnNodeAffinityOrNil(a)
 }
 
@@ -583,5 +591,16 @@ func (m *MemberArangoDPod) Annotations() map[string]string {
 }
 
 func (m *MemberArangoDPod) Labels() map[string]string {
-	return collection.ReservedLabels().Filter(collection.MergeAnnotations(m.spec.Labels, m.groupSpec.Labels))
+	l := collection.ReservedLabels().Filter(collection.MergeAnnotations(m.spec.Labels, m.groupSpec.Labels))
+
+	if m.group.IsArangod() && m.status.Topology != nil && m.deploymentStatus.Topology.Enabled() && m.deploymentStatus.Topology.ID == m.status.Topology.ID {
+		if l == nil {
+			l = map[string]string{}
+		}
+
+		l[k8sutil.LabelKeyArangoZone] = fmt.Sprintf("%d", m.status.Topology.Zone)
+		l[k8sutil.LabelKeyArangoTopology] = string(m.status.Topology.ID)
+	}
+
+	return l
 }

pkg/deployment/topology/mods.community.go

@@ -0,0 +1,29 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+// +build !enterprise
+
+package topology
+
+import api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
+
+func WithTopologyMod(s *api.DeploymentStatus, g api.ServerGroup, m *api.MemberStatus) error {
+	return nil
+}

pkg/deployment/topology/topology.community.go

@@ -0,0 +1,32 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+
+// +build !enterprise
+
+package topology
+
+import (
+	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
+	core "k8s.io/api/core/v1"
+)
+
+func GetTopologyAffinityRules(name string, group api.ServerGroup, status *api.TopologyStatus, member *api.TopologyMemberStatus) core.Affinity {
+	return core.Affinity{}
+}

pkg/util/k8sutil/util.go

@@ -46,6 +46,10 @@ const (
 	LabelKeyArangoExporter = "arango_exporter"
 	// LabelKeyArangoMember is the key of the label used to store the ArangoDeployment member ID in
 	LabelKeyArangoMember = "deployment.arangodb.com/member"
+	// LabelKeyArangoZone is the key of the label used to store the ArangoDeployment zone ID in
+	LabelKeyArangoZone = "deployment.arangodb.com/zone"
+	// LabelKeyArangoTopology is the key of the label used to store the ArangoDeployment topology ID in
+	LabelKeyArangoTopology = "deployment.arangodb.com/topology"
 
 	// AppName is the fixed value for the "app" label
 	AppName = "arangodb"

pkg/version/version.community.go

@@ -18,7 +18,7 @@
 // Copyright holder is ArangoDB GmbH, Cologne, Germany
 //
 
-// +build community
+// +build !enterprise
 
 package version