Skip to content

Commit c586d09

Browse files
ajanikowajanikow
authored
[Feature] Allow to disable ClusterScalingIntegration and add proper Scheduled label to pods (#857)
1 parent eaf627c commit c586d09

File tree

15 files changed

+154
-20
lines changed

15 files changed

+154
-20
lines changed

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
- Add Agency Cache internally
77
- Add Recovery during PlanBuild operation
88
- Fix Exporter in Deployments without authentication
9+
- Allow to disable ClusterScalingIntegration and add proper Scheduled label to pods
910

1011
## [1.2.5](https://github.com/arangodb/kube-arangodb/tree/1.2.5) (2021-10-25)
1112
- Split & Unify Lifecycle management functionality

‎main.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,8 @@ var (
116116
enableBackup bool // Run backup operator
117117
versionOnly bool // Run only version endpoint, explicitly disabled with other
118118

119+
scalingIntegrationEnabled bool
120+
119121
alpineImage, metricsExporterImage, arangoImage string
120122

121123
singleMode bool
@@ -158,6 +160,7 @@ func init() {
158160
f.StringVar(&operatorOptions.scope, "scope", scope.DefaultScope.String(), "Define scope on which Operator works. Legacy - pre 1.1.0 scope with limited cluster access")
159161
f.DurationVar(&timeouts.k8s, "timeout.k8s", time.Second*3, "The request timeout to the kubernetes")
160162
f.DurationVar(&timeouts.arangoD, "timeout.arangod", time.Second*10, "The request timeout to the ArangoDB")
163+
f.BoolVar(&operatorOptions.scalingIntegrationEnabled, "scaling-integration", false, "Enable Scaling Integration")
161164
features.Init(&cmdMain)
162165
}
163166

@@ -368,6 +371,7 @@ func newOperatorConfigAndDeps(id, namespace, name string) (operator.Config, oper
368371
EnableStorage: operatorOptions.enableStorage,
369372
EnableBackup: operatorOptions.enableBackup,
370373
AllowChaos: chaosOptions.allowed,
374+
ScalingIntegrationEnabled: operatorOptions.scalingIntegrationEnabled,
371375
ArangoImage: operatorOptions.arangoImage,
372376
SingleMode: operatorOptions.singleMode,
373377
Scope: scope,

‎pkg/apis/deployment/v1/topology_member_status.go‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,17 @@ package v1
2222

2323
import "k8s.io/apimachinery/pkg/types"
2424

25+
type TopologyMemberStatusInitPhase string
26+
27+
const (
28+
TopologyMemberStatusInitPhaseNone TopologyMemberStatusInitPhase = ""
29+
TopologyMemberStatusInitPhasePending TopologyMemberStatusInitPhase = "pending"
30+
TopologyMemberStatusInitPhaseOK TopologyMemberStatusInitPhase = "ok"
31+
)
32+
2533
type TopologyMemberStatus struct {
26-
ID types.UID `json:"id"`
27-
Zone int `json:"rack"`
28-
Label string `json:"label,omitempty"`
34+
ID types.UID `json:"id"`
35+
Zone int `json:"rack"`
36+
Label string `json:"label,omitempty"`
37+
InitPhase TopologyMemberStatusInitPhase `json:"init_phase,omitempty"`
2938
}

‎pkg/apis/deployment/v1/topology_status.go‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,28 @@ func (t *TopologyStatus) IsTopologyOwned(m *TopologyMemberStatus) bool {
103103
return t.ID == m.ID
104104
}
105105

106+
func (t *TopologyStatus) IsTopologyEvenlyDistributed(group ServerGroup) bool {
107+
if t == nil {
108+
return true
109+
}
110+
111+
max, min := 0, math.MaxInt64
112+
113+
for _, z := range t.Zones {
114+
l := len(z.Members[group.AsRoleAbbreviated()])
115+
116+
if min > l {
117+
min = l
118+
}
119+
120+
if max < l {
121+
max = l
122+
}
123+
}
124+
125+
return min+1 >= max
126+
}
127+
106128
func (t *TopologyStatus) Enabled() bool {
107129
return t != nil
108130
}

‎pkg/apis/deployment/v2alpha1/topology_member_status.go‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,17 @@ package v2alpha1
2222

2323
import "k8s.io/apimachinery/pkg/types"
2424

25+
type TopologyMemberStatusInitPhase string
26+
27+
const (
28+
TopologyMemberStatusInitPhaseNone TopologyMemberStatusInitPhase = ""
29+
TopologyMemberStatusInitPhasePending TopologyMemberStatusInitPhase = "pending"
30+
TopologyMemberStatusInitPhaseOK TopologyMemberStatusInitPhase = "ok"
31+
)
32+
2533
type TopologyMemberStatus struct {
26-
ID types.UID `json:"id"`
27-
Zone int `json:"rack"`
28-
Label string `json:"label,omitempty"`
34+
ID types.UID `json:"id"`
35+
Zone int `json:"rack"`
36+
Label string `json:"label,omitempty"`
37+
InitPhase TopologyMemberStatusInitPhase `json:"init_phase,omitempty"`
2938
}

‎pkg/apis/deployment/v2alpha1/topology_status.go‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,28 @@ func (t *TopologyStatus) IsTopologyOwned(m *TopologyMemberStatus) bool {
103103
return t.ID == m.ID
104104
}
105105

106+
func (t *TopologyStatus) IsTopologyEvenlyDistributed(group ServerGroup) bool {
107+
if t == nil {
108+
return true
109+
}
110+
111+
max, min := 0, math.MaxInt64
112+
113+
for _, z := range t.Zones {
114+
l := len(z.Members[group.AsRoleAbbreviated()])
115+
116+
if min > l {
117+
min = l
118+
}
119+
120+
if max < l {
121+
max = l
122+
}
123+
}
124+
125+
return min+1 >= max
126+
}
127+
106128
func (t *TopologyStatus) Enabled() bool {
107129
return t != nil
108130
}

‎pkg/deployment/cluster_scaling_integration.go‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,9 +85,14 @@ func (ci *clusterScalingIntegration) checkScalingCluster(ctx context.Context, ex
8585
ci.scaleEnabled.mutex.Lock()
8686
defer ci.scaleEnabled.mutex.Unlock()
8787

88+
if !ci.depl.config.ScalingIntegrationEnabled {
89+
return false
90+
}
91+
92+
status, _ := ci.depl.GetStatus()
93+
8894
if !ci.scaleEnabled.enabled {
8995
// Check if it is possible to turn on scaling without any issue
90-
status, _ := ci.depl.GetStatus()
9196
if status.Plan.IsEmpty() && ci.setNumberOfServers(ctx) == nil {
9297
// Scaling should be enabled because there is no Plan.
9398
// It can happen when the enabling action fails

‎pkg/deployment/context_impl.go‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,9 @@ import (
3131
"strconv"
3232
"time"
3333

34+
"github.com/arangodb/kube-arangodb/pkg/deployment/patch"
35+
"k8s.io/apimachinery/pkg/types"
36+
3437
"github.com/arangodb/kube-arangodb/pkg/deployment/resources/inspector"
3538

3639
"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/arangomember"
@@ -710,3 +713,23 @@ func (d *Deployment) WithArangoMemberStatusUpdate(ctx context.Context, namespace
710713

711714
return nil
712715
}
716+
717+
func (d *Deployment) ApplyPatchOnPod(ctx context.Context, pod *core.Pod, p ...patch.Item) error {
718+
parser := patch.Patch(p)
719+
720+
data, err := parser.Marshal()
721+
if err != nil {
722+
return err
723+
}
724+
725+
c := d.deps.KubeCli.CoreV1().Pods(pod.GetNamespace())
726+
727+
ctxChild, cancel := context.WithTimeout(ctx, k8sutil.GetRequestTimeout())
728+
defer cancel()
729+
_, err = c.Patch(ctxChild, pod.GetName(), types.JSONPatchType, data, meta.PatchOptions{})
730+
if err != nil {
731+
return err
732+
}
733+
734+
return nil
735+
}

‎pkg/deployment/deployment.go‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -70,11 +70,12 @@ import (
7070

7171
// Config holds configuration settings for a Deployment
7272
type Config struct {
73-
ServiceAccount string
74-
AllowChaos bool
75-
OperatorImage string
76-
ArangoImage string
77-
Scope scope.Scope
73+
ServiceAccount string
74+
AllowChaos bool
75+
ScalingIntegrationEnabled bool
76+
OperatorImage string
77+
ArangoImage string
78+
Scope scope.Scope
7879
}
7980

8081
// Dependencies holds dependent services for a Deployment

‎pkg/deployment/resources/context.go‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,8 @@ package resources
2626
import (
2727
"context"
2828

29+
"github.com/arangodb/kube-arangodb/pkg/deployment/patch"
30+
2931
agencyCache "github.com/arangodb/kube-arangodb/pkg/deployment/agency"
3032

3133
"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/arangomember"
@@ -138,6 +140,10 @@ type ArangoAgency interface {
138140
RefreshAgencyCache(ctx context.Context) (uint64, error)
139141
}
140142

143+
type ArangoApplier interface {
144+
ApplyPatchOnPod(ctx context.Context, pod *core.Pod, p ...patch.Item) error
145+
}
146+
141147
// Context provides all functions needed by the Resources service
142148
// to perform its service.
143149
type Context interface {
@@ -148,6 +154,7 @@ type Context interface {
148154
DeploymentModInterfaces
149155
DeploymentCachedStatus
150156
ArangoAgency
157+
ArangoApplier
151158

152159
// GetAPIObject returns the deployment as k8s object.
153160
GetAPIObject() k8sutil.APIObject

0 commit comments

Comments
 (0)