mirrors/kube-arangodb
1
0
Fork 0
mirror of https://github.com/arangodb/kube-arangodb.git synced 2024-12-14 11:57:37 +00:00
Code Activity

[Feature] Deprecate ForeachServerGroup, ForeachServerInGroups and ForServerGroup functions and refactor code accordingly (#1069)

Browse source
This commit is contained in:
Nikita Vanyasin 2022-07-24 21:26:26 +03:00 committed by GitHub
parent f8fb748464
commit 738823e0db
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
24 changed files with 631 additions and 801 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
CHANGELOG.md
View file

@ -3,6 +3,7 @@
## [master](https://github.com/arangodb/kube-arangodb/tree/master) (N/A)
- (Feature) Add ArangoDeployment ServerGroupStatus
- (Feature) (EE) Ordered Member IDs
- (Refactor) Deprecate ForeachServerGroup, ForeachServerInGroups and ForServerGroup functions and refactor code accordingly
## [1.2.15](https://github.com/arangodb/kube-arangodb/tree/1.2.15) (2022-07-20)
- (Bugfix) Ensure pod names not too long

14
pkg/apis/deployment/v1/deployment_status_members.go
View file

@ -84,10 +84,13 @@ 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.
// Deprecated. Use AsList instead
func (ds DeploymentStatusMembers) ForeachServerGroup(cb MemberStatusFunc) error {
return ds.ForeachServerInGroups(cb, AllServerGroups...)
}
// ForeachServerInGroups calls the given callback for specified server groups.
// Deprecated. Use AsListInGroups instead
func (ds DeploymentStatusMembers) ForeachServerInGroups(cb MemberStatusFunc, groups ...ServerGroup) error {
for _, group := range groups {
if err := ds.ForServerGroup(cb, group); err != nil {
@ -98,6 +101,8 @@ func (ds DeploymentStatusMembers) ForeachServerInGroups(cb MemberStatusFunc, gro
return nil
}
// ForServerGroup calls the given callback for specified server group.
// Deprecated. Use AsListInGroup or MembersOfGroup
func (ds DeploymentStatusMembers) ForServerGroup(cb MemberStatusFunc, group ServerGroup) error {
switch group {
case ServerGroupSingle:
@ -292,14 +297,11 @@ func (ds DeploymentStatusMembers) MembersOfGroup(group ServerGroup) MemberStatus
func (ds DeploymentStatusMembers) PodNames() []string {
var n []string
ds.ForeachServerGroup(func(group ServerGroup, list MemberStatusList) error {
for _, m := range list {
if m.PodName != "" {
n = append(n, m.PodName)
for _, m := range ds.AsList() {
if m.Member.PodName != "" {
n = append(n, m.Member.PodName)
}
}
return nil
})
return n
}

32
pkg/apis/deployment/v1/deployment_status_members_test.go
View file

@ -39,46 +39,30 @@ func newMemberList() DeploymentStatusMembers {
func Test_StatusMemberList_EnsureDefaultExecutionOrder(t *testing.T) {
statusMembers := newMemberList()
order := AllServerGroups
orderIndex := 0
statusMembers.ForeachServerGroup(func(group ServerGroup, list MemberStatusList) error {
for _, e := range statusMembers.AsList() {
require.True(t, orderIndex < len(order))
require.Equal(t, order[orderIndex], group)
require.Len(t, list, 1)
require.Equal(t, order[orderIndex].AsRole(), list[0].ID)
require.Equal(t, order[orderIndex], e.Group)
require.Equal(t, order[orderIndex].AsRole(), e.Member.ID)
orderIndex += 1
return nil
})
}
}
func Test_StatusMemberList_CustomExecutionOrder(t *testing.T) {
statusMembers := newMemberList()
order := []ServerGroup{
ServerGroupDBServers,
}
orderIndex := 0
statusMembers.ForeachServerInGroups(func(group ServerGroup, list MemberStatusList) error {
for _, e := range statusMembers.AsListInGroups(order...) {
require.True(t, orderIndex < len(order))
require.Equal(t, order[orderIndex], group)
require.Len(t, list, 1)
require.Equal(t, order[orderIndex].AsRole(), list[0].ID)
require.Equal(t, order[orderIndex], e.Group)
require.Equal(t, order[orderIndex].AsRole(), e.Member.ID)
orderIndex += 1
return nil
}, order...)
}
}

14
pkg/apis/deployment/v2alpha1/deployment_status_members.go
View file

@ -84,10 +84,13 @@ 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.
// Deprecated. Use AsList instead
func (ds DeploymentStatusMembers) ForeachServerGroup(cb MemberStatusFunc) error {
return ds.ForeachServerInGroups(cb, AllServerGroups...)
}
// ForeachServerInGroups calls the given callback for specified server groups.
// Deprecated. Use AsListInGroups instead
func (ds DeploymentStatusMembers) ForeachServerInGroups(cb MemberStatusFunc, groups ...ServerGroup) error {
for _, group := range groups {
if err := ds.ForServerGroup(cb, group); err != nil {
@ -98,6 +101,8 @@ func (ds DeploymentStatusMembers) ForeachServerInGroups(cb MemberStatusFunc, gro
return nil
}
// ForServerGroup calls the given callback for specified server group.
// Deprecated. Use AsListInGroup or MembersOfGroup
func (ds DeploymentStatusMembers) ForServerGroup(cb MemberStatusFunc, group ServerGroup) error {
switch group {
case ServerGroupSingle:
@ -292,14 +297,11 @@ func (ds DeploymentStatusMembers) MembersOfGroup(group ServerGroup) MemberStatus
func (ds DeploymentStatusMembers) PodNames() []string {
var n []string
ds.ForeachServerGroup(func(group ServerGroup, list MemberStatusList) error {
for _, m := range list {
if m.PodName != "" {
n = append(n, m.PodName)
for _, m := range ds.AsList() {
if m.Member.PodName != "" {
n = append(n, m.Member.PodName)
}
}
return nil
})
return n
}

32
pkg/apis/deployment/v2alpha1/deployment_status_members_test.go
View file

@ -39,46 +39,30 @@ func newMemberList() DeploymentStatusMembers {
func Test_StatusMemberList_EnsureDefaultExecutionOrder(t *testing.T) {
statusMembers := newMemberList()
order := AllServerGroups
orderIndex := 0
statusMembers.ForeachServerGroup(func(group ServerGroup, list MemberStatusList) error {
for _, e := range statusMembers.AsList() {
require.True(t, orderIndex < len(order))
require.Equal(t, order[orderIndex], group)
require.Len(t, list, 1)
require.Equal(t, order[orderIndex].AsRole(), list[0].ID)
require.Equal(t, order[orderIndex], e.Group)
require.Equal(t, order[orderIndex].AsRole(), e.Member.ID)
orderIndex += 1
return nil
})
}
}
func Test_StatusMemberList_CustomExecutionOrder(t *testing.T) {
statusMembers := newMemberList()
order := []ServerGroup{
ServerGroupDBServers,
}
orderIndex := 0
statusMembers.ForeachServerInGroups(func(group ServerGroup, list MemberStatusList) error {
for _, e := range statusMembers.AsListInGroups(order...) {
require.True(t, orderIndex < len(order))
require.Equal(t, order[orderIndex], group)
require.Len(t, list, 1)
require.Equal(t, order[orderIndex].AsRole(), list[0].ID)
require.Equal(t, order[orderIndex], e.Group)
require.Equal(t, order[orderIndex].AsRole(), e.Member.ID)
orderIndex += 1
return nil
}, order...)
}
}

14
pkg/deployment/deployment_inspector.go
View file

@ -359,26 +359,22 @@ func (d *Deployment) isUpToDateStatus(status api.DeploymentStatus) (upToDate boo
if !status.Conditions.Check(api.ConditionTypeReachable).Exists().IsTrue().Evaluate() {
upToDate = false
return
}
status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
if !upToDate {
return nil
}
for _, member := range list {
for _, m := range status.Members.AsList() {
member := m.Member
if member.Conditions.IsTrue(api.ConditionTypeRestart) || member.Conditions.IsTrue(api.ConditionTypePendingRestart) {
upToDate = false
reason = "Pending restarts on members"
return nil
return
}
if member.Conditions.IsTrue(api.ConditionTypePVCResizePending) {
upToDate = false
reason = "PVC is resizing"
return nil
return
}
}
return nil
})
return
}

50
pkg/deployment/deployment_run_test.go
View file

@ -124,22 +124,19 @@ func runTestCase(t *testing.T, testCase testCaseStruct) {
}
// Set Pending phase
require.NoError(t, d.status.last.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
for _, e := range d.status.last.Members.AsList() {
m := e.Member
if m.Phase == api.MemberPhaseNone {
m.Phase = api.MemberPhasePending
if err := d.status.last.Members.Update(m, group); err != nil {
return err
require.NoError(t, d.status.last.Members.Update(m, e.Group))
}
}
}
return nil
}))
// Set members
if err := d.status.last.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
var loopErr error
for _, e := range d.status.last.Members.AsList() {
m := e.Member
group := e.Group
member := api.ArangoMember{
ObjectMeta: meta.ObjectMeta{
Namespace: d.GetNamespace(),
@ -153,8 +150,8 @@ func runTestCase(t *testing.T, testCase testCaseStruct) {
c := d.WithCurrentArangoMember(m.ArangoMemberName(d.GetName(), group))
if err := c.Create(context.Background(), &member); err != nil {
return err
if loopErr = c.Create(context.Background(), &member); loopErr != nil {
break
}
s := core.Service{
@ -164,8 +161,8 @@ func runTestCase(t *testing.T, testCase testCaseStruct) {
},
}
if _, err := d.ServicesModInterface().Create(context.Background(), &s, meta.CreateOptions{}); err != nil {
return err
if _, loopErr = d.ServicesModInterface().Create(context.Background(), &s, meta.CreateOptions{}); loopErr != nil {
break
}
require.NoError(t, d.acs.CurrentClusterCache().Refresh(context.Background()))
@ -175,9 +172,10 @@ func runTestCase(t *testing.T, testCase testCaseStruct) {
image, ok := d.resources.SelectImage(d.apiObject.Spec, d.status.last)
require.True(t, ok)
template, err := d.resources.RenderPodTemplateForMember(context.Background(), d.ACS(), d.apiObject.Spec, d.status.last, m.ID, image)
if err != nil {
return err
var template *core.PodTemplateSpec
template, loopErr = d.resources.RenderPodTemplateForMember(context.Background(), d.ACS(), d.apiObject.Spec, d.status.last, m.ID, image)
if loopErr != nil {
break
}
checksum, err := resources.ChecksumArangoPod(groupSpec, resources.CreatePodFromTemplate(template))
@ -189,28 +187,24 @@ func runTestCase(t *testing.T, testCase testCaseStruct) {
member.Status.Template = podTemplate
member.Spec.Template = podTemplate
if err := c.Update(context.Background(), func(obj *api.ArangoMember) bool {
if loopErr = c.Update(context.Background(), func(obj *api.ArangoMember) bool {
obj.Spec.Template = podTemplate
return true
}); err != nil {
return err
}); loopErr != nil {
break
}
if err := c.UpdateStatus(context.Background(), func(obj *api.ArangoMember, s *api.ArangoMemberStatus) bool {
if loopErr = c.UpdateStatus(context.Background(), func(obj *api.ArangoMember, s *api.ArangoMemberStatus) bool {
s.Template = podTemplate
return true
}); err != nil {
return err
}); loopErr != nil {
break
}
}
return nil
}); err != nil {
if testCase.ExpectedError != nil && assert.EqualError(t, err, testCase.ExpectedError.Error()) {
if loopErr != nil && testCase.ExpectedError != nil && assert.EqualError(t, loopErr, testCase.ExpectedError.Error()) {
return
}
require.NoError(t, err)
}
// Act
require.NoError(t, d.acs.CurrentClusterCache().Refresh(context.Background()))

62
pkg/deployment/reconcile/action_set_member_condition_v2.go
View file

@ -69,44 +69,52 @@ func (a actionSetMemberConditionV2) Start(ctx context.Context) (bool, error) {
as := a.action.Params[setConditionActionV2KeyStatus] == string(core.ConditionTrue)
if err := a.actionCtx.WithStatusUpdateErr(ctx, func(s *api.DeploymentStatus) (bool, error) {
var changed bool
s.Members.ForServerGroup(func(group api.ServerGroup, members api.MemberStatusList) error {
for i := range members {
if members[i].ID == a.action.MemberID {
changed = members[i].Conditions.UpdateWithHash(api.ConditionType(aa), as, ar, am, ah)
return nil
}
}
status, g, ok := s.Members.ElementByID(a.action.MemberID)
if !ok {
a.log.Info("can not set the condition because the member is gone already")
return nil
}, a.action.Group)
return false, nil
}
// If not found then false is returned.
return changed, nil
if g != a.action.Group {
a.log.Info("can not set the condition because of invalid groups")
return false, nil
}
if status.Conditions.UpdateWithHash(api.ConditionType(aa), as, ar, am, ah) {
if err := s.Members.Update(status, g); err != nil {
return false, err
}
return true, nil
}
return false, nil
}); err != nil {
a.log.Err(err).Warn("unable to update status")
return true, nil
}
case setConditionActionV2KeyTypeRemove:
if err := a.actionCtx.WithStatusUpdateErr(ctx, func(s *api.DeploymentStatus) (bool, error) {
var changed bool
s.Members.ForServerGroup(func(group api.ServerGroup, members api.MemberStatusList) error {
for i := range members {
if members[i].ID == a.action.MemberID {
changed = members[i].Conditions.Remove(api.ConditionType(aa))
return nil
}
status, g, ok := s.Members.ElementByID(a.action.MemberID)
if !ok {
a.log.Info("can not set the condition because the member is gone already")
return false, nil
}
a.log.Info("can not remove the condition because the member is gone already")
return nil
}, a.action.Group)
if g != a.action.Group {
a.log.Info("can not set the condition because of invalid groups")
return false, nil
}
// If not found then false is returned.
return changed, nil
if status.Conditions.Remove(api.ConditionType(aa)) {
if err := s.Members.Update(status, g); err != nil {
return false, err
}
return true, nil
}
return false, nil
}); err != nil {
a.log.Err(err).Warn("unable to update status")
return true, nil

8
pkg/deployment/reconcile/plan_builder_cluster.go
View file

@ -67,14 +67,10 @@ func (r *Reconciler) createClusterOperationPlan(ctx context.Context, apiObject k
membersHealth := health.Health
status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
delete(membersHealth, driver.ServerID(m.ID))
for _, e := range status.Members.AsList() {
delete(membersHealth, driver.ServerID(e.Member.ID))
}
return nil
})
if len(membersHealth) == 0 {
return nil
}

11
pkg/deployment/reconcile/plan_builder_encryption.go
View file

@ -226,12 +226,11 @@ func (r *Reconciler) createEncryptionKeyCleanPlan(ctx context.Context, apiObject
func (r *Reconciler) areEncryptionKeysUpToDate(ctx context.Context, spec api.DeploymentSpec,
status api.DeploymentStatus, context PlanBuilderContext, folder *core.Secret) (plan api.Plan, failed bool) {
status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, group := range api.AllServerGroups {
if !pod.GroupEncryptionSupported(spec.Mode.Get(), group) {
return nil
continue
}
for _, m := range list {
for _, m := range status.Members.MembersOfGroup(group) {
if updateRequired, failedMember := r.isEncryptionKeyUpToDate(ctx, status, context, group, m, folder); failedMember {
failed = true
continue
@ -240,9 +239,7 @@ func (r *Reconciler) areEncryptionKeysUpToDate(ctx context.Context, spec api.Dep
continue
}
}
return nil
})
}
return
}

63
pkg/deployment/reconcile/plan_builder_high.go
View file

@ -73,18 +73,14 @@ func (r *Reconciler) updateMemberPodTemplateSpec(ctx context.Context, apiObject
var plan api.Plan
// Update member specs
status.Members.ForeachServerGroup(func(group api.ServerGroup, members api.MemberStatusList) error {
for _, m := range members {
if m.Phase != api.MemberPhaseNone {
if reason, changed := r.arangoMemberPodTemplateNeedsUpdate(ctx, apiObject, spec, group, status, m, context); changed {
plan = append(plan, actions.NewAction(api.ActionTypeArangoMemberUpdatePodSpec, group, m, reason))
for _, e := range status.Members.AsList() {
if e.Member.Phase != api.MemberPhaseNone {
if reason, changed := r.arangoMemberPodTemplateNeedsUpdate(ctx, apiObject, spec, e.Group, status, e.Member, context); changed {
plan = append(plan, actions.NewAction(api.ActionTypeArangoMemberUpdatePodSpec, e.Group, e.Member, reason))
}
}
}
return nil
})
return plan
}
@ -94,25 +90,19 @@ func (r *Reconciler) updateMemberPhasePlan(ctx context.Context, apiObject k8suti
context PlanBuilderContext) api.Plan {
var plan api.Plan
status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
if m.Phase == api.MemberPhaseNone {
for _, e := range status.Members.AsList() {
if e.Member.Phase == api.MemberPhaseNone {
var p api.Plan
p = append(p,
actions.NewAction(api.ActionTypeArangoMemberUpdatePodSpec, group, m, "Propagating spec of pod"),
actions.NewAction(api.ActionTypeArangoMemberUpdatePodStatus, group, m, "Propagating status of pod"))
p = append(p, actions.NewAction(api.ActionTypeMemberPhaseUpdate, group, m,
"Move to Pending phase").AddParam(actionTypeMemberPhaseUpdatePhaseKey, api.MemberPhasePending.String()))
actions.NewAction(api.ActionTypeArangoMemberUpdatePodSpec, e.Group, e.Member, "Propagating spec of pod"),
actions.NewAction(api.ActionTypeArangoMemberUpdatePodStatus, e.Group, e.Member, "Propagating status of pod"),
actions.NewAction(api.ActionTypeMemberPhaseUpdate, e.Group, e.Member,
"Move to Pending phase").AddParam(actionTypeMemberPhaseUpdatePhaseKey, api.MemberPhasePending.String()),
)
plan = append(plan, p...)
}
}
return nil
})
return plan
}
@ -133,14 +123,13 @@ func (r *Reconciler) updateMemberUpdateConditionsPlan(ctx context.Context, apiOb
context PlanBuilderContext) api.Plan {
var plan api.Plan
if err := status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
if m.Conditions.IsTrue(api.ConditionTypeUpdating) {
for _, e := range status.Members.AsList() {
if e.Member.Conditions.IsTrue(api.ConditionTypeUpdating) {
// We are in updating phase
if status.Plan.IsEmpty() {
// If plan is empty then something went wrong
plan = append(plan,
actions.NewAction(api.ActionTypeSetMemberCondition, group, m, "Clean update actions after failure").
actions.NewAction(api.ActionTypeSetMemberCondition, e.Group, e.Member, "Clean update actions after failure").
AddParam(api.ConditionTypePendingUpdate.String(), "").
AddParam(api.ConditionTypeUpdating.String(), "").
AddParam(api.ConditionTypeUpdateFailed.String(), "T").
@ -150,12 +139,6 @@ func (r *Reconciler) updateMemberUpdateConditionsPlan(ctx context.Context, apiOb
}
}
return nil
}); err != nil {
r.log.Err(err).Error("Error while generating update plan")
return nil
}
return plan
}
@ -164,31 +147,25 @@ func (r *Reconciler) updateMemberRotationConditionsPlan(ctx context.Context, api
context PlanBuilderContext) api.Plan {
var plan api.Plan
if err := status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
cache, ok := context.ACS().ClusterCache(m.ClusterID)
for _, e := range status.Members.AsList() {
cache, ok := context.ACS().ClusterCache(e.Member.ClusterID)
if !ok {
continue
}
p, ok := cache.Pod().V1().GetSimple(m.PodName)
p, ok := cache.Pod().V1().GetSimple(e.Member.PodName)
if !ok {
p = nil
}
if p, err := r.updateMemberRotationConditions(apiObject, spec, m, group, p, context); err != nil {
return err
if p, err := r.updateMemberRotationConditions(apiObject, spec, e.Member, e.Group, p, context); err != nil {
r.log.Err(err).Error("Error while generating rotation plan")
return nil
} else if len(p) > 0 {
plan = append(plan, p...)
}
}
return nil
}); err != nil {
r.log.Err(err).Error("Error while generating rotation plan")
return nil
}
return plan
}

10
pkg/deployment/reconcile/plan_builder_jwt.go
View file

@ -202,22 +202,18 @@ func (r *Reconciler) areJWTTokensUpToDate(ctx context.Context, status api.Deploy
gCtx, c := context.WithTimeout(ctx, 2*time.Second)
defer c()
status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
for _, e := range status.Members.AsList() {
nCtx, c := context.WithTimeout(gCtx, 500*time.Millisecond)
defer c()
if updateRequired, failedMember := r.isJWTTokenUpToDate(nCtx, status, planCtx, group, m, folder); failedMember {
if updateRequired, failedMember := r.isJWTTokenUpToDate(nCtx, status, planCtx, e.Group, e.Member, folder); failedMember {
failed = true
continue
} else if updateRequired {
plan = append(plan, actions.NewAction(api.ActionTypeJWTRefresh, group, m))
plan = append(plan, actions.NewAction(api.ActionTypeJWTRefresh, e.Group, e.Member))
continue
}
}
return nil
})
return
}

7
pkg/deployment/reconcile/plan_builder_member_recovery.go
View file

@ -57,13 +57,15 @@ func (r *Reconciler) createMemberFailedRestoreInternal(_ context.Context, _ k8su
agencyState, agencyOK := context.GetAgencyCache()
// Check for members in failed state.
status.Members.ForeachServerGroup(func(group api.ServerGroup, members api.MemberStatusList) error {
for _, group := range api.AllServerGroups {
members := status.Members.MembersOfGroup(group)
failed := 0
for _, m := range members {
if m.Phase == api.MemberPhaseFailed {
failed++
}
}
for _, m := range members {
if m.Phase != api.MemberPhaseFailed || len(plan) > 0 {
continue
@ -116,8 +118,7 @@ func (r *Reconciler) createMemberFailedRestoreInternal(_ context.Context, _ k8su
}
}
}
return nil
})
}
if len(plan) == 0 && !agencyOK {
r.log.Warn("unable to build further plan without access to agency")

9
pkg/deployment/reconcile/plan_builder_rotate_upgrade.go
View file

@ -85,8 +85,9 @@ func (r *Reconciler) createMarkToRemovePlan(ctx context.Context, apiObject k8sut
context PlanBuilderContext) api.Plan {
var plan api.Plan
status.Members.ForeachServerInGroups(func(group api.ServerGroup, members api.MemberStatusList) error {
for _, m := range members {
for _, e := range status.Members.AsListInGroups(rotationByAnnotationOrder...) {
m := e.Member
group := e.Group
if m.Phase != api.MemberPhaseCreated || m.PodName == "" {
// Only rotate when phase is created
continue
@ -111,10 +112,6 @@ func (r *Reconciler) createMarkToRemovePlan(ctx context.Context, apiObject k8sut
}
}
}
return nil
}, rotationByAnnotationOrder...)
return plan
}

15
pkg/deployment/reconcile/plan_builder_scale.go
View file

@ -116,11 +116,14 @@ func (r *Reconciler) createReplaceMemberPlan(ctx context.Context, apiObject k8su
var plan api.Plan
// Replace is only allowed for Coordinators, DBServers & Agents
status.Members.ForeachServerInGroups(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, member := range list {
for _, e := range status.Members.AsListInGroups(api.ServerGroupAgents, api.ServerGroupDBServers, api.ServerGroupCoordinators) {
if !plan.IsEmpty() {
return nil
break
}
member := e.Member
group := e.Group
if member.Conditions.IsTrue(api.ConditionTypeMarkedToRemove) {
ready, message := groupReadyForRestart(context, status, member, group)
if !ready {
@ -134,27 +137,21 @@ func (r *Reconciler) createReplaceMemberPlan(ctx context.Context, apiObject k8su
r.planLogger.
Str("role", group.AsRole()).
Debug("Creating replacement plan")
return nil
case api.ServerGroupCoordinators:
plan = append(plan, actions.NewAction(api.ActionTypeRemoveMember, group, member))
r.planLogger.
Str("role", group.AsRole()).
Debug("Creating replacement plan")
return nil
case api.ServerGroupAgents:
plan = append(plan, actions.NewAction(api.ActionTypeRemoveMember, group, member),
actions.NewAction(api.ActionTypeAddMember, group, withPredefinedMember("")))
r.planLogger.
Str("role", group.AsRole()).
Debug("Creating replacement plan")
return nil
}
}
}
return nil
}, api.ServerGroupAgents, api.ServerGroupDBServers, api.ServerGroupCoordinators)
return plan
}

10
pkg/deployment/reconcile/plan_builder_test.go
View file

@ -977,14 +977,10 @@ func TestCreatePlan(t *testing.T) {
},
}
require.NoError(t, c.context.ArangoDeployment.Status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
m.Phase = api.MemberPhaseCreated
require.NoError(t, c.context.ArangoDeployment.Status.Members.Update(m, group))
for _, e := range c.context.ArangoDeployment.Status.Members.AsList() {
e.Member.Phase = api.MemberPhaseCreated
require.NoError(t, c.context.ArangoDeployment.Status.Members.Update(e.Member, e.Group))
}
return nil
}))
},
context: &testContext{
ArangoDeployment: deploymentTemplate.DeepCopy(),

57
pkg/deployment/reconcile/plan_builder_tls.go
View file

@ -288,9 +288,7 @@ func (r *Reconciler) createKeyfileRenewalPlanSynced(ctx context.Context, apiObje
var plan api.Plan
group := api.ServerGroupSyncMasters
for _, statusMember := range status.Members.AsListInGroup(group) {
member := statusMember.Member
for _, member := range status.Members.MembersOfGroup(group) {
if !plan.IsEmpty() {
continue
}
@ -321,17 +319,8 @@ func (r *Reconciler) createKeyfileRenewalPlanDefault(ctx context.Context, apiObj
var plan api.Plan
status.Members.ForeachServerGroup(func(group api.ServerGroup, members api.MemberStatusList) error {
if !group.IsArangod() {
return nil
}
for _, member := range members {
if !plan.IsEmpty() {
return nil
}
cache, ok := planCtx.ACS().ClusterCache(member.ClusterID)
for _, e := range status.Members.AsListInGroups(api.AllArangoDServerGroups...) {
cache, ok := planCtx.ACS().ClusterCache(e.Member.ClusterID)
if !ok {
continue
}
@ -339,15 +328,13 @@ func (r *Reconciler) createKeyfileRenewalPlanDefault(ctx context.Context, apiObj
lCtx, c := context.WithTimeout(ctx, 500*time.Millisecond)
defer c()
if renew, _ := r.keyfileRenewalRequired(lCtx, apiObject, spec.TLS, spec, cache, planCtx, group, member, api.TLSRotateModeRecreate); renew {
if renew, _ := r.keyfileRenewalRequired(lCtx, apiObject, spec.TLS, spec, cache, planCtx, e.Group, e.Member, api.TLSRotateModeRecreate); renew {
r.planLogger.Info("Renewal of keyfile required - Recreate (server)")
plan = append(plan, tlsRotateConditionAction(group, member.ID, "Restart server after keyfile removal"))
plan = append(plan, tlsRotateConditionAction(e.Group, e.Member.ID, "Restart server after keyfile removal"))
break
}
}
return nil
})
return plan
}
@ -360,13 +347,8 @@ func (r *Reconciler) createKeyfileRenewalPlanInPlace(ctx context.Context, apiObj
var plan api.Plan
status.Members.ForeachServerGroup(func(group api.ServerGroup, members api.MemberStatusList) error {
if !group.IsArangod() {
return nil
}
for _, member := range members {
cache, ok := planCtx.ACS().ClusterCache(member.ClusterID)
for _, e := range status.Members.AsListInGroups(api.AllArangoDServerGroups...) {
cache, ok := planCtx.ACS().ClusterCache(e.Member.ClusterID)
if !ok {
continue
}
@ -374,18 +356,15 @@ func (r *Reconciler) createKeyfileRenewalPlanInPlace(ctx context.Context, apiObj
lCtx, c := context.WithTimeout(ctx, 500*time.Millisecond)
defer c()
if renew, recreate := r.keyfileRenewalRequired(lCtx, apiObject, spec.TLS, spec, cache, planCtx, group, member, api.TLSRotateModeInPlace); renew {
if renew, recreate := r.keyfileRenewalRequired(lCtx, apiObject, spec.TLS, spec, cache, planCtx, e.Group, e.Member, api.TLSRotateModeInPlace); renew {
r.planLogger.Info("Renewal of keyfile required - InPlace (server)")
if recreate {
plan = append(plan, actions.NewAction(api.ActionTypeCleanTLSKeyfileCertificate, group, member, "Remove server keyfile and enforce renewal"))
plan = append(plan, actions.NewAction(api.ActionTypeCleanTLSKeyfileCertificate, e.Group, e.Member, "Remove server keyfile and enforce renewal"))
}
plan = append(plan, actions.NewAction(api.ActionTypeRefreshTLSKeyfileCertificate, group, member, "Renew Member Keyfile"))
plan = append(plan, actions.NewAction(api.ActionTypeRefreshTLSKeyfileCertificate, e.Group, e.Member, "Renew Member Keyfile"))
}
}
return nil
})
return plan
}
@ -419,17 +398,12 @@ func createKeyfileRenewalPlanMode(
mode := spec.TLS.Mode.Get()
status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, e := range status.Members.AsList() {
if mode != api.TLSRotateModeInPlace {
return nil
break
}
for _, member := range list {
if mode != api.TLSRotateModeInPlace {
return nil
}
if i, ok := status.Images.GetByImageID(member.ImageID); !ok {
if i, ok := status.Images.GetByImageID(e.Member.ImageID); !ok {
mode = api.TLSRotateModeRecreate
} else {
if !features.TLSRotation().Supported(i.ArangoDBVersion, i.Enterprise) {
@ -438,9 +412,6 @@ func createKeyfileRenewalPlanMode(
}
}
return nil
})
return mode
}

15
pkg/deployment/reconcile/plan_builder_tls_sni.go
View file

@ -56,15 +56,14 @@ func (r *Reconciler) createRotateTLSServerSNIPlan(ctx context.Context, apiObject
}
var plan api.Plan
status.Members.ForeachServerGroup(func(group api.ServerGroup, members api.MemberStatusList) error {
for _, group := range api.AllServerGroups {
if !pod.GroupSNISupported(spec.Mode.Get(), group) {
return nil
continue
}
for _, m := range members {
for _, m := range status.Members.MembersOfGroup(group) {
if !plan.IsEmpty() {
// Only 1 member at a time
return nil
break
}
if m.Phase != api.MemberPhaseCreated {
@ -96,8 +95,7 @@ func (r *Reconciler) createRotateTLSServerSNIPlan(ctx context.Context, apiObject
})
if err != nil {
r.planLogger.Err(err).Info("SNI compare failed")
return nil
break
} else if !ok {
switch spec.TLS.Mode.Get() {
case api.TLSRotateModeRecreate:
@ -111,7 +109,6 @@ func (r *Reconciler) createRotateTLSServerSNIPlan(ctx context.Context, apiObject
}
}
}
return nil
})
}
return plan
}

10
pkg/deployment/resilience/member_failure.go
View file

@ -41,8 +41,10 @@ const (
func (r *Resilience) CheckMemberFailure(ctx context.Context) error {
status, lastVersion := r.context.GetStatus()
updateStatusNeeded := false
if err := status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
for _, e := range status.Members.AsList() {
m := e.Member
group := e.Group
log := r.log("member-failure").
Str("id", m.ID).
Str("role", group.AsRole())
@ -101,10 +103,6 @@ func (r *Resilience) CheckMemberFailure(ctx context.Context) error {
}
}
return nil
}); err != nil {
return errors.WithStack(err)
}
if updateStatusNeeded {
if err := r.context.UpdateStatus(ctx, status, lastVersion); err != nil {
return errors.WithStack(err)

21
pkg/deployment/resources/member_cleanup.go
View file

@ -82,8 +82,9 @@ func (r *Resources) syncMembersInCluster(ctx context.Context, health memberState
status, lastVersion := r.context.GetStatus()
updateStatusNeeded := false
status.Members.ForeachServerInGroups(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
for _, e := range status.Members.AsListInGroups(api.ServerGroupCoordinators, api.ServerGroupDBServers) {
m := e.Member
group := e.Group
log := log.Str("member", m.ID).Str("role", group.AsRole())
if serverFound(m.ID) {
// Member is (still) found, skip it
@ -105,8 +106,6 @@ func (r *Resources) syncMembersInCluster(ctx context.Context, health memberState
log.Info("Member is no longer part of the ArangoDB cluster")
}
}
return nil
}, api.ServerGroupCoordinators, api.ServerGroupDBServers)
if updateStatusNeeded {
log.Debug("UpdateStatus needed")
@ -125,9 +124,8 @@ func (r *Resources) EnsureArangoMembers(ctx context.Context, cachedStatus inspec
s, _ := r.context.GetStatus()
obj := r.context.GetAPIObject()
if err := s.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, member := range list {
name := member.ArangoMemberName(r.context.GetAPIObject().GetName(), group)
for _, e := range s.Members.AsList() {
name := e.Member.ArangoMemberName(r.context.GetAPIObject().GetName(), e.Group)
c := r.context.WithCurrentArangoMember(name)
@ -141,8 +139,8 @@ func (r *Resources) EnsureArangoMembers(ctx context.Context, cachedStatus inspec
},
},
Spec: api.ArangoMemberSpec{
Group: group,
ID: member.ID,
Group: e.Group,
ID: e.Member.ID,
DeploymentUID: obj.GetUID(),
},
}
@ -174,11 +172,6 @@ func (r *Resources) EnsureArangoMembers(ctx context.Context, cachedStatus inspec
}
}
return nil
}); err != nil {
return err
}
if err := cachedStatus.ArangoMember().V1().Iterate(func(member *api.ArangoMember) error {
_, g, ok := s.Members.ElementByID(member.Spec.ID)

11
pkg/deployment/resources/pod_inspector.go
View file

@ -351,8 +351,9 @@ func (r *Resources) InspectPods(ctx context.Context, cachedStatus inspectorInter
}
// Go over all members, check for missing pods
status.Members.ForeachServerGroup(func(group api.ServerGroup, members api.MemberStatusList) error {
for _, m := range members {
for _, e := range status.Members.AsList() {
m := e.Member
group := e.Group
if podName := m.PodName; podName != "" {
if _, exists := cachedStatus.Pod().V1().GetSimple(podName); !exists {
log.Str("pod-name", podName).Debug("Does not exist")
@ -372,7 +373,7 @@ func (r *Resources) InspectPods(ctx context.Context, cachedStatus inspectorInter
}
// Save it
if err := status.Members.Update(m, group); err != nil {
return errors.WithStack(err)
return 0, errors.WithStack(err)
}
}
default:
@ -397,15 +398,13 @@ func (r *Resources) InspectPods(ctx context.Context, cachedStatus inspectorInter
if updateMemberNeeded {
// Save it
if err := status.Members.Update(m, group); err != nil {
return errors.WithStack(err)
return 0, errors.WithStack(err)
}
}
}
}
}
}
return nil
})
spec := r.context.GetSpec()
allMembersReady := status.Members.AllMembersReady(spec.GetMode(), spec.Sync.IsEnabled())

51
pkg/deployment/resources/pod_leader.go
View file

@ -61,15 +61,14 @@ func (r *Resources) EnsureLeader(ctx context.Context, cachedStatus inspectorInte
noLeader := len(leaderID) == 0
changed := false
group := api.ServerGroupAgents
agencyServers := func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
pod, exist := cachedStatus.Pod().V1().GetSimple(m.PodName)
for _, e := range status.Members.AsListInGroup(group) {
pod, exist := cachedStatus.Pod().V1().GetSimple(e.Member.PodName)
if !exist {
continue
}
labels := pod.GetLabels()
if noLeader || m.ID != leaderID {
if noLeader || e.Member.ID != leaderID {
// Unset a leader when:
// - leader is unknown.
// - leader does not belong to the current pod.
@ -83,7 +82,7 @@ func (r *Resources) EnsureLeader(ctx context.Context, cachedStatus inspectorInte
return err
}
log.Warn("leader label is removed from \"%s\" member", m.ID)
log.Warn("leader label is removed from \"%s\" member", e.Member.ID)
changed = true
}
@ -105,21 +104,13 @@ func (r *Resources) EnsureLeader(ctx context.Context, cachedStatus inspectorInte
log.Err(err).Error("Unable to update leader label")
return err
}
log.Warn("leader label is set on \"%s\" member", m.ID)
log.Warn("leader label is set on \"%s\" member", e.Member.ID)
changed = true
}
return nil
}
if err := status.Members.ForeachServerInGroups(agencyServers, group); err != nil {
return err
}
if changed {
return errors.Reconcile()
}
changed = false
if noLeader {
// There is no leader agency so service may not exist, or it can exist with empty list of endpoints.
@ -165,19 +156,12 @@ func (r *Resources) getSingleServerLeaderID(ctx context.Context) (string, error)
var leaderID string
var anyError error
dbServers := func(group api.ServerGroup, list api.MemberStatusList) error {
if len(list) == 0 {
return nil
}
ctxCancel, cancel := context.WithCancel(ctx)
defer func() {
cancel()
}()
defer cancel()
// Fetch availability of each single server.
var wg sync.WaitGroup
wg.Add(len(list))
for _, m := range list {
for _, m := range status.Members.Single {
wg.Add(1)
go func(id string) {
defer wg.Done()
err := globals.GetGlobalTimeouts().ArangoD().RunWithTimeout(ctxCancel, func(ctxChild context.Context) error {
@ -209,13 +193,6 @@ func (r *Resources) getSingleServerLeaderID(ctx context.Context) (string, error)
}
wg.Wait()
return nil
}
if err := status.Members.ForeachServerInGroups(dbServers, api.ServerGroupSingle); err != nil {
return "", err
}
if len(leaderID) > 0 {
return leaderID, nil
}
@ -240,8 +217,8 @@ func (r *Resources) ensureSingleServerLeader(ctx context.Context, cachedStatus i
}
}
singleServers := func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
status, _ := r.context.GetStatus()
for _, m := range status.Members.Single {
pod, exist := cachedStatus.Pod().V1().GetSimple(m.PodName)
if !exist {
continue
@ -271,14 +248,6 @@ func (r *Resources) ensureSingleServerLeader(ctx context.Context, cachedStatus i
changed = true
}
return nil
}
status, _ := r.context.GetStatus()
if err := status.Members.ForeachServerInGroups(singleServers, api.ServerGroupSingle); err != nil {
return err
}
if changed {
return errors.Reconcile()
}

14
pkg/deployment/resources/services.go
View file

@ -131,25 +131,24 @@ func (r *Resources) EnsureServices(ctx context.Context, cachedStatus inspectorIn
reconcileRequired := k8sutil.NewReconcile(cachedStatus)
// Ensure member services
if err := status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, e := range status.Members.AsList() {
var targetPort int32 = shared.ArangoPort
switch group {
switch e.Group {
case api.ServerGroupSyncMasters:
targetPort = shared.ArangoSyncMasterPort
case api.ServerGroupSyncWorkers:
targetPort = shared.ArangoSyncWorkerPort
}
for _, m := range list {
memberName := m.ArangoMemberName(r.context.GetAPIObject().GetName(), group)
memberName := e.Member.ArangoMemberName(r.context.GetAPIObject().GetName(), e.Group)
member, ok := cachedStatus.ArangoMember().V1().GetSimple(memberName)
if !ok {
return errors.Newf("Member %s not found", memberName)
}
selector := k8sutil.LabelsForMember(deploymentName, group.AsRole(), m.ID)
selector := k8sutil.LabelsForMember(deploymentName, e.Group.AsRole(), e.Member.ID)
if s, ok := cachedStatus.Service().V1().GetSimple(member.GetName()); !ok {
s := r.createService(member.GetName(), member.GetNamespace(), member.AsOwner(), targetPort, selector)
@ -177,11 +176,6 @@ func (r *Resources) EnsureServices(ctx context.Context, cachedStatus inspectorIn
}
}
return nil
}); err != nil {
return err
}
// Headless service
counterMetric.Inc()
if _, exists := cachedStatus.Service().V1().GetSimple(k8sutil.CreateHeadlessServiceName(deploymentName)); !exists {

51
pkg/deployment/server_api.go
View file

@ -66,9 +66,8 @@ func (d *Deployment) StateColor() server.StateColor {
allGood = false
}
status, _ := d.GetStatus()
status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
switch m.Phase {
for _, m := range status.Members.AsList() {
switch m.Member.Phase {
case api.MemberPhaseFailed:
failed = true
case api.MemberPhaseCreated:
@ -78,8 +77,6 @@ func (d *Deployment) StateColor() server.StateColor {
allGood = true
}
}
return nil
})
if failed {
return server.StateRed
}
@ -94,34 +91,22 @@ func (d *Deployment) StateColor() server.StateColor {
// PodCount returns the number of pods for the deployment
func (d *Deployment) PodCount() int {
count := 0
status, _ := d.GetStatus()
status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
if m.PodName != "" {
count++
}
}
return nil
})
return count
return len(status.Members.PodNames())
}
// ReadyPodCount returns the number of pods for the deployment that are in ready state
func (d *Deployment) ReadyPodCount() int {
count := 0
status, _ := d.GetStatus()
status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
if m.PodName == "" {
for _, e := range status.Members.AsList() {
if e.Member.PodName == "" {
continue
}
if m.Conditions.IsTrue(api.ConditionTypeReady) {
if e.Member.Conditions.IsTrue(api.ConditionTypeReady) {
count++
}
}
return nil
})
return count
}
@ -129,14 +114,11 @@ func (d *Deployment) ReadyPodCount() int {
func (d *Deployment) VolumeCount() int {
count := 0
status, _ := d.GetStatus()
status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
if m.PersistentVolumeClaimName != "" {
for _, e := range status.Members.AsList() {
if e.Member.PersistentVolumeClaimName != "" {
count++
}
}
return nil
})
return count
}
@ -145,22 +127,19 @@ func (d *Deployment) ReadyVolumeCount() int {
count := 0
status, _ := d.GetStatus()
pvcs, _ := d.GetOwnedPVCs() // Ignore errors on purpose
status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, m := range list {
if m.PersistentVolumeClaimName == "" {
for _, e := range status.Members.AsList() {
if e.Member.PersistentVolumeClaimName == "" {
continue
}
// Find status
for _, pvc := range pvcs {
if pvc.Name == m.PersistentVolumeClaimName {
if pvc.Name == e.Member.PersistentVolumeClaimName {
if pvc.Status.Phase == core.ClaimBound {
count++
}
}
}
}
return nil
})
return count
}
@ -229,7 +208,9 @@ func (d *Deployment) DatabaseVersion() (string, string) {
func (d *Deployment) Members() map[api.ServerGroup][]server.Member {
result := make(map[api.ServerGroup][]server.Member)
status, _ := d.GetStatus()
status.Members.ForeachServerGroup(func(group api.ServerGroup, list api.MemberStatusList) error {
for _, group := range api.AllServerGroups {
list := status.Members.MembersOfGroup(group)
members := make([]server.Member, len(list))
for i, m := range list {
members[i] = member{
@ -241,7 +222,7 @@ func (d *Deployment) Members() map[api.ServerGroup][]server.Member {
if len(members) > 0 {
result[group] = members
}
return nil
})
}
return result
}