Skip to content

Commit 858389c

Browse files
leon-apeleon-ape
committed
fix test
1 parent 8b82680 commit 858389c

File tree

7 files changed

+545
-128
lines changed

7 files changed

+545
-128
lines changed

‎controllers/apps/cluster/transformer_cluster_component.go‎

Lines changed: 74 additions & 68 deletions
Original file line numberDiff line numberDiff line change
@@ -909,7 +909,7 @@ func (h *clusterShardingHandler) update(transCtx *clusterTransformContext, dag *
909909
return err
910910
}
911911

912-
err3, errorSkip := h.handleShardAddNRemove(transCtx, name, runningCompsMap, protoCompsMap, toCreate, toDelete, toUpdate)
912+
errorSkip, err3 := h.handleShardAddNRemove(transCtx, name, runningCompsMap, protoCompsMap, toCreate, toDelete, toUpdate)
913913

914914
// TODO: update strategy
915915
h.deleteComps(transCtx, dag, runningCompsMap, toDelete.Difference(errorSkip))
@@ -1131,21 +1131,7 @@ func (h *clusterShardingHandler) handlePostProvision(transCtx *clusterTransformC
11311131
shardingDef = h.shardingDef(transCtx, shardingName)
11321132

11331133
upsert = func(status appsv1.LifecycleActionStatus) {
1134-
shardingStatus := transCtx.Cluster.Status.Shardings[shardingName]
1135-
shardingStatus.ShardingDef = shardingDef.Name
1136-
if shardingStatus.PostProvision == nil {
1137-
shardingStatus.PostProvision = &status
1138-
} else {
1139-
shardingStatus.PostProvision.Phase = status.Phase
1140-
shardingStatus.PostProvision.Message = status.Message
1141-
if shardingStatus.PostProvision.StartTime == nil {
1142-
shardingStatus.PostProvision.StartTime = status.StartTime
1143-
}
1144-
if shardingStatus.PostProvision.CompletionTime == nil {
1145-
shardingStatus.PostProvision.CompletionTime = status.CompletionTime
1146-
}
1147-
}
1148-
transCtx.Cluster.Status.Shardings[shardingName] = shardingStatus
1134+
h.updateActionStatus(transCtx, shardingName, shardingDef, status, true)
11491135
}
11501136

11511137
init = func() error {
@@ -1187,7 +1173,8 @@ func (h *clusterShardingHandler) handlePostProvision(transCtx *clusterTransformC
11871173
initialized = transCtx.Cluster.Status.Shardings[shardingName].PostProvision != nil
11881174

11891175
defined = func() bool {
1190-
return shardingDef.Spec.LifecycleActions != nil && shardingDef.Spec.LifecycleActions.PostProvision != nil
1176+
return shardingDef != nil && shardingDef.Spec.LifecycleActions != nil &&
1177+
shardingDef.Spec.LifecycleActions.PostProvision != nil
11911178
}
11921179

11931180
done = func() bool {
@@ -1220,42 +1207,20 @@ func (h *clusterShardingHandler) handlePostProvision(transCtx *clusterTransformC
12201207

12211208
func (h *clusterShardingHandler) handlePreTerminate(transCtx *clusterTransformContext, shardingName string, runningComps []*appsv1.Component) error {
12221209
var (
1223-
shardingStatus = transCtx.Cluster.Status.Shardings[shardingName]
1224-
shardingDef = func() *appsv1.ShardingDefinition {
1225-
obj := &appsv1.ShardingDefinition{}
1210+
shardingStatus = transCtx.Cluster.Status.Shardings[shardingName]
1211+
shardingDef, initErr = func() (*appsv1.ShardingDefinition, error) {
12261212
if shardingStatus.ShardingDef == "" {
1227-
return obj
1213+
return nil, nil
12281214
}
1215+
obj := &appsv1.ShardingDefinition{}
12291216
if err := transCtx.Client.Get(transCtx.Context, types.NamespacedName{Name: shardingStatus.ShardingDef}, obj); err != nil {
1230-
return obj
1217+
return nil, client.IgnoreNotFound(err)
12311218
}
1232-
return obj
1219+
return obj, nil
12331220
}()
12341221

12351222
upsert = func(status appsv1.LifecycleActionStatus) {
1236-
shardingStatus2 := transCtx.Cluster.Status.Shardings[shardingName]
1237-
shardingStatus2.ShardingDef = shardingDef.Name
1238-
if shardingStatus2.PreTerminate == nil {
1239-
shardingStatus2.PreTerminate = &status
1240-
} else {
1241-
shardingStatus2.PreTerminate.Phase = status.Phase
1242-
shardingStatus2.PreTerminate.Message = status.Message
1243-
if shardingStatus2.PreTerminate.StartTime == nil {
1244-
shardingStatus2.PreTerminate.StartTime = status.StartTime
1245-
}
1246-
if shardingStatus2.PreTerminate.CompletionTime == nil {
1247-
shardingStatus2.PreTerminate.CompletionTime = status.CompletionTime
1248-
}
1249-
}
1250-
transCtx.Cluster.Status.Shardings[shardingName] = shardingStatus2
1251-
}
1252-
1253-
init = func() error {
1254-
upsert(appsv1.LifecycleActionStatus{
1255-
Phase: appsv1.LifecycleActionPending,
1256-
StartTime: &metav1.Time{Time: time.Now()},
1257-
})
1258-
return ictrlutil.NewDelayedRequeueError(time.Second, "requeue to schedule the sharding pre-terminate action")
1223+
h.updateActionStatus(transCtx, shardingName, shardingDef, status, false)
12591224
}
12601225

12611226
skipAsNotDefined = func() error {
@@ -1281,26 +1246,30 @@ func (h *clusterShardingHandler) handlePreTerminate(transCtx *clusterTransformCo
12811246
}
12821247

12831248
succeed = func() error {
1249+
now := &metav1.Time{Time: time.Now()}
12841250
upsert(appsv1.LifecycleActionStatus{
12851251
Phase: appsv1.LifecycleActionSucceeded,
12861252
Message: "",
1287-
CompletionTime: &metav1.Time{Time: time.Now()},
1253+
StartTime: now,
1254+
CompletionTime: now,
12881255
})
12891256
return nil
12901257
}
12911258

12921259
fail = func(err error) error {
12931260
upsert(appsv1.LifecycleActionStatus{
1294-
Phase: appsv1.LifecycleActionFailed,
1295-
Message: err.Error(),
1261+
Phase: appsv1.LifecycleActionFailed,
1262+
Message: err.Error(),
1263+
StartTime: &metav1.Time{Time: time.Now()},
12961264
})
12971265
return err // TODO: delay the requeue?
12981266
}
12991267

13001268
initialized = transCtx.Cluster.Status.Shardings[shardingName].PreTerminate != nil
13011269

13021270
defined = func() bool {
1303-
return shardingDef.Spec.LifecycleActions != nil && shardingDef.Spec.LifecycleActions.PreTerminate != nil
1271+
return shardingDef != nil && shardingDef.Spec.LifecycleActions != nil &&
1272+
shardingDef.Spec.LifecycleActions.PreTerminate != nil
13041273
}
13051274

13061275
done = func() bool {
@@ -1309,6 +1278,10 @@ func (h *clusterShardingHandler) handlePreTerminate(transCtx *clusterTransformCo
13091278
}
13101279
)
13111280

1281+
if initErr != nil {
1282+
return initErr
1283+
}
1284+
13121285
if !defined() {
13131286
if !initialized || !done() {
13141287
return skipAsNotDefined()
@@ -1327,26 +1300,62 @@ func (h *clusterShardingHandler) handlePreTerminate(transCtx *clusterTransformCo
13271300
}
13281301
}
13291302

1330-
if !initialized {
1331-
return init()
1332-
}
1333-
13341303
if err := h.shardingAction(transCtx, shardingName, shardingPreTerminateAction,
13351304
shardingDef.Spec.LifecycleActions.PreTerminate, nil, runningComps, nil); err != nil {
13361305
return fail(err)
13371306
}
13381307
return succeed()
13391308
}
13401309

1310+
func (h *clusterShardingHandler) updateActionStatus(transCtx *clusterTransformContext,
1311+
shardingName string, shardingDef *appsv1.ShardingDefinition, status appsv1.LifecycleActionStatus, postProvision bool) {
1312+
shardingStatus := transCtx.Cluster.Status.Shardings[shardingName]
1313+
if shardingDef == nil {
1314+
shardingStatus.ShardingDef = ""
1315+
} else {
1316+
shardingStatus.ShardingDef = shardingDef.Name
1317+
}
1318+
1319+
merge := func(action *appsv1.LifecycleActionStatus) {
1320+
action.Phase = status.Phase
1321+
action.Message = status.Message
1322+
if action.StartTime == nil {
1323+
action.StartTime = status.StartTime
1324+
}
1325+
if action.CompletionTime == nil {
1326+
action.CompletionTime = status.CompletionTime
1327+
}
1328+
}
1329+
1330+
if postProvision {
1331+
if shardingStatus.PostProvision == nil {
1332+
shardingStatus.PostProvision = &status
1333+
} else {
1334+
merge(shardingStatus.PostProvision)
1335+
}
1336+
} else {
1337+
if shardingStatus.PreTerminate == nil {
1338+
shardingStatus.PreTerminate = &status
1339+
} else {
1340+
merge(shardingStatus.PreTerminate)
1341+
}
1342+
}
1343+
1344+
if transCtx.Cluster.Status.Shardings == nil {
1345+
transCtx.Cluster.Status.Shardings = map[string]appsv1.ClusterShardingStatus{}
1346+
}
1347+
transCtx.Cluster.Status.Shardings[shardingName] = shardingStatus
1348+
}
1349+
13411350
func (h *clusterShardingHandler) handleShardAddNRemove(transCtx *clusterTransformContext, shardingName string,
13421351
runningCompsMap map[string]*appsv1.Component, protoCompsMap map[string]*appsv1.Component,
1343-
toCreate, toDelete, toUpdate sets.Set[string]) (error, sets.Set[string]) {
1352+
toCreate, toDelete, toUpdate sets.Set[string]) (sets.Set[string], error) {
13441353
var (
13451354
errorSkip = sets.Set[string]{}
13461355

13471356
create = func() {
13481357
shardingDef := h.shardingDef(transCtx, shardingName)
1349-
if shardingDef.Spec.LifecycleActions != nil && shardingDef.Spec.LifecycleActions.ShardAdd != nil {
1358+
if shardingDef != nil && shardingDef.Spec.LifecycleActions != nil && shardingDef.Spec.LifecycleActions.ShardAdd != nil {
13501359
now := time.Now().Format(time.RFC3339Nano)
13511360
for name := range toCreate {
13521361
protoComp := protoCompsMap[name]
@@ -1394,9 +1403,9 @@ func (h *clusterShardingHandler) handleShardAddNRemove(transCtx *clusterTransfor
13941403
err2 := _delete()
13951404

13961405
if err1 != nil {
1397-
return err1, errorSkip
1406+
return errorSkip, err1
13981407
}
1399-
return err2, errorSkip
1408+
return errorSkip, err2
14001409
}
14011410

14021411
func (h *clusterShardingHandler) handleShardAdd(transCtx *clusterTransformContext,
@@ -1408,27 +1417,24 @@ func (h *clusterShardingHandler) handleShardAdd(transCtx *clusterTransformContex
14081417
return runningComp.Annotations[shardingAddShardKey] != ""
14091418
}
14101419

1411-
fail = func(err error) error {
1412-
if protoComp.Annotations == nil {
1413-
protoComp.Annotations = make(map[string]string)
1414-
}
1415-
protoComp.Annotations[shardingAddShardKey] = runningComp.Annotations[shardingAddShardKey]
1416-
return err
1420+
succeed = func() error {
1421+
delete(runningComp.Annotations, shardingAddShardKey)
1422+
return nil
14171423
}
14181424
)
14191425

1420-
if shardingDef.Spec.LifecycleActions == nil || shardingDef.Spec.LifecycleActions.ShardAdd == nil {
1426+
if shardingDef == nil || shardingDef.Spec.LifecycleActions == nil || shardingDef.Spec.LifecycleActions.ShardAdd == nil {
14211427
return nil
14221428
}
14231429

14241430
if pending() {
14251431
args := map[string]string{shardingAddShardNameVar: runningComp.Name}
14261432
if err := h.shardingAction(transCtx, shardingName, shardingAddShardAction,
14271433
shardingDef.Spec.LifecycleActions.ShardAdd, args, runningComps, runningComp); err != nil {
1428-
return fail(err)
1434+
return err
14291435
}
14301436
}
1431-
return nil
1437+
return succeed()
14321438
}
14331439

14341440
func (h *clusterShardingHandler) handleShardRemove(transCtx *clusterTransformContext,
@@ -1445,7 +1451,7 @@ func (h *clusterShardingHandler) handleShardRemove(transCtx *clusterTransformCon
14451451
}
14461452
)
14471453

1448-
if shardingDef.Spec.LifecycleActions == nil || shardingDef.Spec.LifecycleActions.ShardRemove == nil {
1454+
if shardingDef == nil || shardingDef.Spec.LifecycleActions == nil || shardingDef.Spec.LifecycleActions.ShardRemove == nil {
14491455
return nil
14501456
}
14511457

‎controllers/apps/cluster/transformer_cluster_component_status_test.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -409,7 +409,7 @@ var _ = Describe("cluster component status transformer", func() {
409409

410410
It("sharding spec deleted", func() {
411411
// have seen the sharding1 and sharding2 objects in the cluster
412-
transCtx.Cluster.Status.Shardings = map[string]appsv1.ClusterComponentStatus{
412+
transCtx.Cluster.Status.Shardings = map[string]appsv1.ClusterShardingStatus{
413413
"sharding1": {
414414
Phase: appsv1.RunningComponentPhase,
415415
},
@@ -467,7 +467,7 @@ var _ = Describe("cluster component status transformer", func() {
467467

468468
It("sharding object deleted", func() {
469469
// have seen the sharding1 and sharding2 objects in the cluster
470-
transCtx.Cluster.Status.Shardings = map[string]appsv1.ClusterComponentStatus{
470+
transCtx.Cluster.Status.Shardings = map[string]appsv1.ClusterShardingStatus{
471471
"sharding1": {
472472
Phase: appsv1.RunningComponentPhase,
473473
},
@@ -509,7 +509,7 @@ var _ = Describe("cluster component status transformer", func() {
509509

510510
It("sharding deleted", func() {
511511
// have seen the sharding1 and sharding2 objects in the cluster
512-
transCtx.Cluster.Status.Shardings = map[string]appsv1.ClusterComponentStatus{
512+
transCtx.Cluster.Status.Shardings = map[string]appsv1.ClusterShardingStatus{
513513
"sharding1": {
514514
Phase: appsv1.RunningComponentPhase,
515515
},
@@ -642,7 +642,7 @@ var _ = Describe("cluster component status transformer", func() {
642642
})
643643

644644
It("phase changed", func() {
645-
transCtx.Cluster.Status.Shardings = map[string]appsv1.ClusterComponentStatus{
645+
transCtx.Cluster.Status.Shardings = map[string]appsv1.ClusterShardingStatus{
646646
"sharding1": {
647647
Phase: appsv1.CreatingComponentPhase,
648648
},

0 commit comments

Comments
 (0)