Skip to content

Commit b296fad

Browse files
burmanmburmanm
committed
Fix status updates and add a new status field, contextName for the Datacenter
1 parent 15ae6ac commit b296fad

File tree

12 files changed

+93
-113
lines changed

12 files changed

+93
-113
lines changed

‎apis/k8ssandra/v1alpha1/k8ssandracluster_types.go

+1
Original file line numberDiff line numberDiff line change
@@ -143,6 +143,7 @@ type K8ssandraClusterCondition struct {
143143

144144
// K8ssandraStatus defines the observed of a k8ssandra instance
145145
type K8ssandraStatus struct {
146+
ContextName string `json:"contextName,omitempty"`
146147
DecommissionProgress DecommissionProgress `json:"decommissionProgress,omitempty"`
147148
Cassandra *cassdcapi.CassandraDatacenterStatus `json:"cassandra,omitempty"`
148149
Stargate *stargateapi.StargateStatus `json:"stargate,omitempty"`

‎charts/k8ssandra-operator/crds/k8ssandra-operator-crds.yaml

+2
Original file line numberDiff line numberDiff line change
@@ -31176,6 +31176,8 @@ spec:
3117631176
format: date-time
3117731177
type: string
3117831178
type: object
31179+
contextName:
31180+
type: string
3117931181
decommissionProgress:
3118031182
type: string
3118131183
reaper:

‎config/crd/bases/k8ssandra.io_k8ssandraclusters.yaml

+2
Original file line numberDiff line numberDiff line change
@@ -31114,6 +31114,8 @@ spec:
3111431114
format: date-time
3111531115
type: string
3111631116
type: object
31117+
contextName:
31118+
type: string
3111731119
decommissionProgress:
3111831120
type: string
3111931121
reaper:

‎controllers/k8ssandra/cleanup.go

+41-39
Original file line numberDiff line numberDiff line change
@@ -144,7 +144,7 @@ func (r *K8ssandraClusterReconciler) checkFinalizer(ctx context.Context, kc *api
144144
}
145145

146146
func (r *K8ssandraClusterReconciler) checkDcDeletion(ctx context.Context, kc *api.K8ssandraCluster, logger logr.Logger) result.ReconcileResult {
147-
dcName, dcNameOverride := k8ssandra.GetDatacenterForDecommission(kc)
147+
dcName := k8ssandra.GetDatacenterForDecommission(kc)
148148
if dcName == "" {
149149
return result.Continue()
150150
}
@@ -163,76 +163,78 @@ func (r *K8ssandraClusterReconciler) checkDcDeletion(ctx context.Context, kc *ap
163163
default:
164164
logger.Info("Proceeding with DC deletion", "DC", dcName)
165165

166-
cassDcName := dcName
167-
if dcNameOverride != "" {
168-
cassDcName = dcNameOverride
169-
}
170-
return r.deleteDc(ctx, kc, dcName, cassDcName, logger)
166+
return r.deleteDc(ctx, kc, dcName, logger)
171167
}
172168
}
173169

174-
func (r *K8ssandraClusterReconciler) deleteDc(ctx context.Context, kc *api.K8ssandraCluster, dcName string, cassDcName string, logger logr.Logger) result.ReconcileResult {
170+
func (r *K8ssandraClusterReconciler) deleteDc(ctx context.Context, kc *api.K8ssandraCluster, dcName string, logger logr.Logger) result.ReconcileResult {
175171
kcKey := utils.GetKey(kc)
176172

177-
stargate, remoteClient, err := r.findStargateForDeletion(ctx, kcKey, cassDcName, nil)
173+
dcRemoteClient, err := r.ClientCache.GetRemoteClient(kc.Status.Datacenters[dcName].ContextName)
174+
if err != nil {
175+
return result.Error(err)
176+
}
177+
178+
dc, _, err := r.findDcForDeletion(ctx, kcKey, dcName, dcRemoteClient)
179+
if err != nil {
180+
return result.Error(err)
181+
}
182+
183+
if dc == nil {
184+
// Deletion was already done
185+
delete(kc.Status.Datacenters, dcName)
186+
logger.Info("DC deletion finished", "DC", dcName)
187+
return result.Continue()
188+
}
189+
190+
stargate, remoteClient, err := r.findStargateForDeletion(ctx, kcKey, dc.DatacenterName(), nil)
178191
if err != nil {
179192
return result.Error(err)
180193
}
181194

182195
if stargate != nil {
183196
if err = remoteClient.Delete(ctx, stargate); err != nil && !errors.IsNotFound(err) {
184-
return result.Error(fmt.Errorf("failed to delete Stargate for dc (%s): %v", cassDcName, err))
197+
return result.Error(fmt.Errorf("failed to delete Stargate for dc (%s): %v", dc.DatacenterName(), err))
185198
}
186199
logger.Info("Deleted Stargate", "Stargate", utils.GetKey(stargate))
187200
}
188201

189-
reaper, remoteClient, err := r.findReaperForDeletion(ctx, kcKey, cassDcName, remoteClient)
202+
reaper, remoteClient, err := r.findReaperForDeletion(ctx, kcKey, dc.DatacenterName(), remoteClient)
190203
if err != nil {
191204
return result.Error(err)
192205
}
193206

194207
if reaper != nil {
195208
if err = remoteClient.Delete(ctx, reaper); err != nil && !errors.IsNotFound(err) {
196-
return result.Error(fmt.Errorf("failed to delete Reaper for dc (%s): %v", cassDcName, err))
209+
return result.Error(fmt.Errorf("failed to delete Reaper for dc (%s): %v", dc.DatacenterName(), err))
197210
}
198211
logger.Info("Deleted Reaper", "Reaper", utils.GetKey(reaper))
199212
}
200213

201-
dc, remoteClient, err := r.findDcForDeletion(ctx, kcKey, dcName, remoteClient)
202-
if err != nil {
214+
if err := r.deleteContactPointsService(ctx, kc, dc, logger); err != nil {
203215
return result.Error(err)
204216
}
205217

206-
if dc != nil {
207-
if err := r.deleteContactPointsService(ctx, kc, dc, logger); err != nil {
208-
return result.Error(err)
209-
}
210-
211-
if dc.GetConditionStatus(cassdcapi.DatacenterDecommission) == corev1.ConditionTrue {
212-
logger.Info("CassandraDatacenter decommissioning in progress", "CassandraDatacenter", utils.GetKey(dc))
213-
// There is no need to requeue here. Reconciliation will be trigger by updates made by cass-operator.
214-
return result.Done()
215-
}
216-
217-
if !annotations.HasAnnotationWithValue(dc, cassdcapi.DecommissionOnDeleteAnnotation, "true") {
218-
patch := client.MergeFrom(dc.DeepCopy())
219-
annotations.AddAnnotation(dc, cassdcapi.DecommissionOnDeleteAnnotation, "true")
220-
if err = remoteClient.Patch(ctx, dc, patch); err != nil {
221-
return result.Error(fmt.Errorf("failed to add %s annotation to dc: %v", cassdcapi.DecommissionOnDeleteAnnotation, err))
222-
}
223-
}
224-
225-
if err = remoteClient.Delete(ctx, dc); err != nil && !errors.IsNotFound(err) {
226-
return result.Error(fmt.Errorf("failed to delete CassandraDatacenter (%s): %v", dcName, err))
227-
}
228-
logger.Info("Deleted CassandraDatacenter", "CassandraDatacenter", utils.GetKey(dc))
218+
if dc.GetConditionStatus(cassdcapi.DatacenterDecommission) == corev1.ConditionTrue {
219+
logger.Info("CassandraDatacenter decommissioning in progress", "CassandraDatacenter", utils.GetKey(dc))
229220
// There is no need to requeue here. Reconciliation will be trigger by updates made by cass-operator.
230221
return result.Done()
231222
}
232223

233-
delete(kc.Status.Datacenters, dcName)
234-
logger.Info("DC deletion finished", "DC", dcName)
235-
return result.Continue()
224+
if !annotations.HasAnnotationWithValue(dc, cassdcapi.DecommissionOnDeleteAnnotation, "true") {
225+
patch := client.MergeFrom(dc.DeepCopy())
226+
annotations.AddAnnotation(dc, cassdcapi.DecommissionOnDeleteAnnotation, "true")
227+
if err = dcRemoteClient.Patch(ctx, dc, patch); err != nil {
228+
return result.Error(fmt.Errorf("failed to add %s annotation to dc: %v", cassdcapi.DecommissionOnDeleteAnnotation, err))
229+
}
230+
}
231+
232+
if err = dcRemoteClient.Delete(ctx, dc); err != nil && !errors.IsNotFound(err) {
233+
return result.Error(fmt.Errorf("failed to delete CassandraDatacenter (%s): %v", dcName, err))
234+
}
235+
logger.Info("Deleted CassandraDatacenter", "CassandraDatacenter", utils.GetKey(dc))
236+
// There is no need to requeue here. Reconciliation will be trigger by updates made by cass-operator.
237+
return result.Done()
236238
}
237239

238240
func (r *K8ssandraClusterReconciler) findStargateForDeletion(

‎controllers/k8ssandra/datacenters.go

+9-3
Original file line numberDiff line numberDiff line change
@@ -148,7 +148,7 @@ func (r *K8ssandraClusterReconciler) reconcileDatacenters(ctx context.Context, k
148148
return result.Error(fmt.Errorf("CassandraDatacenter %s has cluster name %s, but expected %s. Cluster name cannot be changed in an existing cluster", dcKey, actualDc.Spec.ClusterName, cassClusterName)), actualDcs
149149
}
150150

151-
r.setStatusForDatacenter(kc, actualDc)
151+
r.setStatusForDatacenter(kc, actualDc, dcConfig.K8sContext)
152152

153153
r.reconcileContactPointsService(ctx, kc, actualDc, remoteClient, dcLogger)
154154

@@ -309,7 +309,7 @@ func (r *K8ssandraClusterReconciler) reconcileDatacenters(ctx context.Context, k
309309
return result.Continue(), actualDcs
310310
}
311311

312-
func (r *K8ssandraClusterReconciler) setStatusForDatacenter(kc *api.K8ssandraCluster, dc *cassdcapi.CassandraDatacenter) {
312+
func (r *K8ssandraClusterReconciler) setStatusForDatacenter(kc *api.K8ssandraCluster, dc *cassdcapi.CassandraDatacenter, targetContext string) {
313313
if len(kc.Status.Datacenters) == 0 {
314314
kc.Status.Datacenters = make(map[string]api.K8ssandraStatus, 0)
315315
}
@@ -318,9 +318,15 @@ func (r *K8ssandraClusterReconciler) setStatusForDatacenter(kc *api.K8ssandraClu
318318

319319
if found {
320320
dc.Status.DeepCopyInto(kdcStatus.Cassandra)
321+
if kdcStatus.ContextName != targetContext {
322+
// This is pretty fatal situation if it happens to actually change the context, but for updates from previous versions we need it
323+
kdcStatus.ContextName = targetContext
324+
}
325+
kc.Status.Datacenters[dc.Name] = kdcStatus
321326
} else {
322327
kc.Status.Datacenters[dc.Name] = api.K8ssandraStatus{
323-
Cassandra: dc.Status.DeepCopy(),
328+
ContextName: targetContext,
329+
Cassandra: dc.Status.DeepCopy(),
324330
}
325331
}
326332
}

‎controllers/k8ssandra/k8ssandracluster_controller_test.go

+3-3
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ import (
4747

4848
const (
4949
timeout = time.Second * 5
50-
interval = time.Millisecond * 500
50+
interval = time.Millisecond * 100
5151
)
5252

5353
var (
@@ -69,8 +69,8 @@ func TestK8ssandraCluster(t *testing.T) {
6969

7070
reconcilerConfig := config.InitConfig()
7171

72-
reconcilerConfig.DefaultDelay = 100 * time.Millisecond
73-
reconcilerConfig.LongDelay = 300 * time.Millisecond
72+
reconcilerConfig.DefaultDelay = 50 * time.Millisecond
73+
reconcilerConfig.LongDelay = 200 * time.Millisecond
7474

7575
err := testEnv.Start(ctx, t, func(mgr manager.Manager, clientCache *clientcache.ClientCache, clusters []cluster.Cluster) error {
7676
err := (&K8ssandraClusterReconciler{

‎controllers/k8ssandra/reaper.go

+3-5
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ package k8ssandra
1919
import (
2020
"context"
2121
"fmt"
22+
2223
"github.com/go-logr/logr"
2324
cassdcapi "github.com/k8ssandra/cass-operator/apis/cassandra/v1beta1"
2425
api "github.com/k8ssandra/k8ssandra-operator/apis/k8ssandra/v1alpha1"
@@ -242,11 +243,8 @@ func (r *K8ssandraClusterReconciler) setStatusForReaper(kc *api.K8ssandraCluster
242243

243244
func (r *K8ssandraClusterReconciler) removeReaperStatus(kc *api.K8ssandraCluster, dcName string) {
244245
if kdcStatus, found := kc.Status.Datacenters[dcName]; found {
245-
kc.Status.Datacenters[dcName] = api.K8ssandraStatus{
246-
Reaper: nil,
247-
Cassandra: kdcStatus.Cassandra.DeepCopy(),
248-
Stargate: kdcStatus.Stargate.DeepCopy(),
249-
}
246+
kdcStatus.Reaper = nil
247+
kc.Status.Datacenters[dcName] = kdcStatus
250248
}
251249
}
252250

‎controllers/k8ssandra/schemas.go

+18-5
Original file line numberDiff line numberDiff line change
@@ -56,25 +56,38 @@ func (r *K8ssandraClusterReconciler) checkSchemas(
5656
}
5757
}
5858

59-
decommCassDcName, _ := k8ssandra.GetDatacenterForDecommission(kc)
59+
decommCassDcName := k8ssandra.GetDatacenterForDecommission(kc)
6060

6161
logger.Info("Checking if user keyspace replication needs to be updated", "decommissioning_dc", decommCassDcName)
62+
logger.Info("Status of datacenters", "status", kc.Status.Datacenters)
6263
decommission := false
64+
status := kc.Status.Datacenters[decommCassDcName]
6365
if decommCassDcName != "" {
64-
decommission = kc.Status.Datacenters[decommCassDcName].DecommissionProgress == api.DecommUpdatingReplication
66+
decommission = status.DecommissionProgress == api.DecommUpdatingReplication
6567
}
66-
status := kc.Status.Datacenters[decommCassDcName]
6768

6869
if decommission {
6970
kcKey := utils.GetKey(kc)
70-
dc, _, err = r.findDcForDeletion(ctx, kcKey, decommCassDcName, remoteClient)
71+
logger.Info("Decommissioning DC", "dc", decommCassDcName, "context", status.ContextName)
72+
73+
var dcRemoteClient client.Client
74+
if status.ContextName == "" {
75+
dcRemoteClient = remoteClient
76+
} else {
77+
dcRemoteClient, err = r.ClientCache.GetRemoteClient(status.ContextName)
78+
if err != nil {
79+
return result.Error(err)
80+
}
81+
}
82+
83+
dc, _, err = r.findDcForDeletion(ctx, kcKey, decommCassDcName, dcRemoteClient)
7184
if err != nil {
7285
return result.Error(err)
7386
}
7487

7588
decommDcName := decommCassDcName
7689
if dc.Spec.DatacenterName != "" {
77-
decommCassDcName = dc.Spec.DatacenterName
90+
decommDcName = dc.Spec.DatacenterName
7891
}
7992

8093
if recResult := r.checkUserKeyspacesReplicationForDecommission(kc, decommDcName, mgmtApi, logger); recResult.Completed() {

‎controllers/k8ssandra/stargate.go

+3-6
Original file line numberDiff line numberDiff line change
@@ -121,10 +121,10 @@ func (r *K8ssandraClusterReconciler) setStatusForStargate(kc *api.K8ssandraClust
121121
if found {
122122
if kdcStatus.Stargate == nil {
123123
kdcStatus.Stargate = stargate.Status.DeepCopy()
124-
kc.Status.Datacenters[dcName] = kdcStatus
125124
} else {
126125
stargate.Status.DeepCopyInto(kdcStatus.Stargate)
127126
}
127+
kc.Status.Datacenters[dcName] = kdcStatus
128128
} else {
129129
kc.Status.Datacenters[dcName] = api.K8ssandraStatus{
130130
Stargate: stargate.Status.DeepCopy(),
@@ -166,10 +166,7 @@ func (r *K8ssandraClusterReconciler) reconcileStargateAuthSchema(
166166

167167
func (r *K8ssandraClusterReconciler) removeStargateStatus(kc *api.K8ssandraCluster, dcName string) {
168168
if kdcStatus, found := kc.Status.Datacenters[dcName]; found {
169-
kc.Status.Datacenters[dcName] = api.K8ssandraStatus{
170-
Stargate: nil,
171-
Cassandra: kdcStatus.Cassandra.DeepCopy(),
172-
Reaper: kdcStatus.Reaper.DeepCopy(),
173-
}
169+
kdcStatus.Stargate = nil
170+
kc.Status.Datacenters[dcName] = kdcStatus
174171
}
175172
}

‎pkg/k8ssandra/util.go

+3-10
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ import (
55
"k8s.io/utils/strings/slices"
66
)
77

8-
func GetDatacenterForDecommission(kc *api.K8ssandraCluster) (string, string) {
8+
func GetDatacenterForDecommission(kc *api.K8ssandraCluster) string {
99
dcNames := make([]string, 0)
1010
for _, dc := range kc.Spec.Cassandra.Datacenters {
1111
dcNames = append(dcNames, dc.Meta.Name)
@@ -15,24 +15,17 @@ func GetDatacenterForDecommission(kc *api.K8ssandraCluster) (string, string) {
1515
for dcName, status := range kc.Status.Datacenters {
1616
if !slices.Contains(dcNames, dcName) {
1717
if status.DecommissionProgress != api.DecommNone {
18-
return dcName, dcNameOverride(kc.Status.Datacenters[dcName].Cassandra.DatacenterName)
18+
return dcName
1919
}
2020
}
2121
}
2222

2323
// No decommissions are in progress. Pick the first one we find.
2424
for dcName := range kc.Status.Datacenters {
2525
if !slices.Contains(dcNames, dcName) {
26-
return dcName, dcNameOverride(kc.Status.Datacenters[dcName].Cassandra.DatacenterName)
26+
return dcName
2727
}
2828
}
2929

30-
return "", ""
31-
}
32-
33-
func dcNameOverride(datacenterName *string) string {
34-
if datacenterName != nil {
35-
return *datacenterName
36-
}
3730
return ""
3831
}

‎pkg/k8ssandra/util_test.go

-34
This file was deleted.

0 commit comments

Comments
 (0)