Skip to content

Commit b043971

Browse files
mohammedfirdoussmohammedfirdouss
authored
[Multi_K8s-Plugin] Add per-stage multiTarget filtering (#6757)
Signed-off-by: Mohammed Firdous <124298708+mohammedfirdouss@users.noreply.github.com>
1 parent 1a4677a commit b043971

9 files changed

Lines changed: 151 additions & 37 deletions

File tree

‎pkg/app/pipedv1/plugin/kubernetes_multicluster/config/application.go‎

Lines changed: 22 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -173,6 +173,9 @@ type K8sSyncStageOptions struct {
173173
AddVariantLabelToSelector bool `json:"addVariantLabelToSelector"`
174174
// Whether the resources that are no longer defined in Git should be removed or not.
175175
Prune bool `json:"prune"`
176+
// Limit this stage to a subset of deploy targets by name.
177+
// Empty means the stage applies to all targets.
178+
MultiTargets []string `json:"multiTargets,omitempty"`
176179
}
177180

178181
// FindDeployTarget finds the deploy target configuration by the given name.
@@ -227,10 +230,17 @@ type K8sCanaryRolloutStageOptions struct {
227230
CreateService bool `json:"createService"`
228231
// List of patches used to customize manifests for CANARY variant.
229232
Patches []K8sResourcePatch `json:"patches"`
233+
// Limit this stage to a subset of deploy targets by name.
234+
// Empty means the stage applies to all targets.
235+
MultiTargets []string `json:"multiTargets,omitempty"`
230236
}
231237

232238
// K8sCanaryCleanStageOptions contains all configurable values for a K8S_CANARY_CLEAN stage.
233-
type K8sCanaryCleanStageOptions struct{}
239+
type K8sCanaryCleanStageOptions struct {
240+
// Limit this stage to a subset of deploy targets by name.
241+
// Empty means the stage applies to all targets.
242+
MultiTargets []string `json:"multiTargets,omitempty"`
243+
}
234244

235245
// K8sPrimaryRolloutStageOptions contains all configurable values for a K8S_PRIMARY_ROLLOUT stage.
236246
type K8sPrimaryRolloutStageOptions struct {
@@ -243,6 +253,9 @@ type K8sPrimaryRolloutStageOptions struct {
243253
AddVariantLabelToSelector bool `json:"addVariantLabelToSelector"`
244254
// Whether the resources that are no longer defined in Git should be removed or not.
245255
Prune bool `json:"prune"`
256+
// Limit this stage to a subset of deploy targets by name.
257+
// Empty means the stage applies to all targets.
258+
MultiTargets []string `json:"multiTargets,omitempty"`
246259
}
247260

248261
func (o *K8sPrimaryRolloutStageOptions) UnmarshalJSON(data []byte) error {
@@ -270,6 +283,9 @@ type K8sBaselineRolloutStageOptions struct {
270283
Suffix string `json:"suffix" default:"baseline"`
271284
// Whether the BASELINE service should be created.
272285
CreateService bool `json:"createService"`
286+
// Limit this stage to a subset of deploy targets by name.
287+
// Empty means the stage applies to all targets.
288+
MultiTargets []string `json:"multiTargets,omitempty"`
273289
}
274290

275291
func (o *K8sBaselineRolloutStageOptions) UnmarshalJSON(data []byte) error {
@@ -286,7 +302,11 @@ func (o *K8sBaselineRolloutStageOptions) UnmarshalJSON(data []byte) error {
286302
}
287303

288304
// K8sBaselineCleanStageOptions contains all configurable values for a K8S_BASELINE_CLEAN stage.
289-
type K8sBaselineCleanStageOptions struct{}
305+
type K8sBaselineCleanStageOptions struct {
306+
// Limit this stage to a subset of deploy targets by name.
307+
// Empty means the stage applies to all targets.
308+
MultiTargets []string `json:"multiTargets,omitempty"`
309+
}
290310

291311
// K8sResourcePatch represents a patch operation for a Kubernetes resource.
292312
type K8sResourcePatch struct {

‎pkg/app/pipedv1/plugin/kubernetes_multicluster/config/traffic.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,9 @@ type K8sTrafficRoutingStageOptions struct {
7070
Canary unit.Percentage `json:"canary"`
7171
// The percentage of traffic should be routed to BASELINE variant.
7272
Baseline unit.Percentage `json:"baseline"`
73+
// Limit this stage to a subset of deploy targets by name.
74+
// Empty means the stage applies to all targets.
75+
MultiTargets []string `json:"multiTargets,omitempty"`
7376
}
7477

7578
// Percentages returns the primary, canary, and baseline percentages from the K8sTrafficRoutingStageOptions.

‎pkg/app/pipedv1/plugin/kubernetes_multicluster/deployment/baseline.go‎

Lines changed: 14 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ func (p *Plugin) executeK8sMultiBaselineRolloutStage(ctx context.Context, input
4646
}
4747
}
4848

49-
targets := buildStageTargets(lp, dts, cfg.Spec.Input.MultiTargets)
49+
targets := filterStageTargets(buildStageTargets(lp, dts, cfg.Spec.Input.MultiTargets), stageCfg.MultiTargets)
5050
return runOnTargets(ctx, lp, targets, func(ctx context.Context, dt *sdk.DeployTarget[kubeconfig.KubernetesDeployTargetConfig], mt *kubeconfig.KubernetesMultiTarget) sdk.StageStatus {
5151
lp.Infof("Start baseline rollout for target %s", dt.Name)
5252
return p.baselineRollout(ctx, input, dt, mt, stageCfg)
@@ -197,20 +197,23 @@ func (p *Plugin) executeK8sMultiBaselineCleanStage(ctx context.Context, input *s
197197
return sdk.StageStatusFailure
198198
}
199199

200+
var stageCfg kubeconfig.K8sBaselineCleanStageOptions
201+
if len(input.Request.StageConfig) > 0 {
202+
if err := json.Unmarshal(input.Request.StageConfig, &stageCfg); err != nil {
203+
lp.Errorf("Failed while unmarshalling stage config (%v)", err)
204+
return sdk.StageStatusFailure
205+
}
206+
}
207+
200208
deployTargetMap := make(map[string]*sdk.DeployTarget[kubeconfig.KubernetesDeployTargetConfig], len(dts))
201209
for _, dt := range dts {
202210
deployTargetMap[dt.Name] = dt
203211
}
204212

205-
type targetConfig struct {
206-
deployTarget *sdk.DeployTarget[kubeconfig.KubernetesDeployTargetConfig]
207-
multiTarget *kubeconfig.KubernetesMultiTarget
208-
}
209-
210-
targetConfigs := make([]targetConfig, 0, len(dts))
213+
targetConfigs := make([]stageTarget, 0, len(dts))
211214
if len(cfg.Spec.Input.MultiTargets) == 0 {
212215
for _, dt := range dts {
213-
targetConfigs = append(targetConfigs, targetConfig{deployTarget: dt})
216+
targetConfigs = append(targetConfigs, stageTarget{deployTarget: dt})
214217
}
215218
} else {
216219
for _, mt := range cfg.Spec.Input.MultiTargets {
@@ -219,10 +222,12 @@ func (p *Plugin) executeK8sMultiBaselineCleanStage(ctx context.Context, input *s
219222
lp.Infof("Ignore multi target '%s': not matched any deployTarget", mt.Target.Name)
220223
continue
221224
}
222-
targetConfigs = append(targetConfigs, targetConfig{deployTarget: dt, multiTarget: &mt})
225+
targetConfigs = append(targetConfigs, stageTarget{deployTarget: dt, multiTarget: &mt})
223226
}
224227
}
225228

229+
targetConfigs = filterStageTargets(targetConfigs, stageCfg.MultiTargets)
230+
226231
eg, ctx := errgroup.WithContext(ctx)
227232
for _, tc := range targetConfigs {
228233
eg.Go(func() error {

‎pkg/app/pipedv1/plugin/kubernetes_multicluster/deployment/canary.go‎

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ func (p *Plugin) executeK8sMultiCanaryRolloutStage(ctx context.Context, input *s
4747
}
4848
}
4949

50-
targets := buildStageTargets(lp, dts, cfg.Spec.Input.MultiTargets)
50+
targets := filterStageTargets(buildStageTargets(lp, dts, cfg.Spec.Input.MultiTargets), stageCfg.MultiTargets)
5151
return runOnTargets(ctx, lp, targets, func(ctx context.Context, dt *sdk.DeployTarget[kubeconfig.KubernetesDeployTargetConfig], mt *kubeconfig.KubernetesMultiTarget) sdk.StageStatus {
5252
lp.Infof("Start canary rollout for target %s", dt.Name)
5353
return p.canaryRollout(ctx, input, dt, mt, stageCfg)
@@ -318,19 +318,23 @@ func (p *Plugin) executeK8sMultiCanaryCleanStage(ctx context.Context, input *sdk
318318
return sdk.StageStatusFailure
319319
}
320320

321+
var stageCfg kubeconfig.K8sCanaryCleanStageOptions
322+
if len(input.Request.StageConfig) > 0 {
323+
if err := json.Unmarshal(input.Request.StageConfig, &stageCfg); err != nil {
324+
lp.Errorf("Failed while unmarshalling stage config (%v)", err)
325+
return sdk.StageStatusFailure
326+
}
327+
}
328+
321329
deployTargetMap := make(map[string]*sdk.DeployTarget[kubeconfig.KubernetesDeployTargetConfig], len(dts))
322330
for _, dt := range dts {
323331
deployTargetMap[dt.Name] = dt
324332
}
325333

326-
type targetConfig struct {
327-
deployTarget *sdk.DeployTarget[kubeconfig.KubernetesDeployTargetConfig]
328-
}
329-
330-
targetConfigs := make([]targetConfig, 0, len(dts))
334+
targetConfigs := make([]stageTarget, 0, len(dts))
331335
if len(cfg.Spec.Input.MultiTargets) == 0 {
332336
for _, dt := range dts {
333-
targetConfigs = append(targetConfigs, targetConfig{deployTarget: dt})
337+
targetConfigs = append(targetConfigs, stageTarget{deployTarget: dt})
334338
}
335339
} else {
336340
for _, mt := range cfg.Spec.Input.MultiTargets {
@@ -339,10 +343,12 @@ func (p *Plugin) executeK8sMultiCanaryCleanStage(ctx context.Context, input *sdk
339343
lp.Infof("Ignore multi target '%s': not matched any deployTarget", mt.Target.Name)
340344
continue
341345
}
342-
targetConfigs = append(targetConfigs, targetConfig{deployTarget: dt})
346+
targetConfigs = append(targetConfigs, stageTarget{deployTarget: dt})
343347
}
344348
}
345349

350+
targetConfigs = filterStageTargets(targetConfigs, stageCfg.MultiTargets)
351+
346352
eg, ctx := errgroup.WithContext(ctx)
347353
for _, tc := range targetConfigs {
348354
eg.Go(func() error {

‎pkg/app/pipedv1/plugin/kubernetes_multicluster/deployment/misc.go‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,25 @@ import (
2727
"github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin/kubernetes_multicluster/provider"
2828
)
2929

30+
// filterStageTargets returns only the elements of tcs whose deploy target name
31+
// appears in stageTargets. If stageTargets is empty all elements are returned unchanged.
32+
func filterStageTargets(tcs []stageTarget, stageTargets []string) []stageTarget {
33+
if len(stageTargets) == 0 {
34+
return tcs
35+
}
36+
allowed := make(map[string]struct{}, len(stageTargets))
37+
for _, name := range stageTargets {
38+
allowed[name] = struct{}{}
39+
}
40+
out := make([]stageTarget, 0, len(stageTargets))
41+
for _, tc := range tcs {
42+
if _, ok := allowed[tc.deployTarget.Name]; ok {
43+
out = append(out, tc)
44+
}
45+
}
46+
return out
47+
}
48+
3049
func ensureVariantSelectorInWorkload(m provider.Manifest, variantLabel, variant string) error {
3150
variantMap := map[string]string{
3251
variantLabel: variant,

‎pkg/app/pipedv1/plugin/kubernetes_multicluster/deployment/misc_test.go‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,9 @@ import (
2020
"github.com/stretchr/testify/assert"
2121
"github.com/stretchr/testify/require"
2222

23+
sdk "github.com/pipe-cd/piped-plugin-sdk-go"
24+
25+
kubeconfig "github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin/kubernetes_multicluster/config"
2326
"github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin/kubernetes_multicluster/provider"
2427
)
2528

@@ -235,3 +238,59 @@ metadata:
235238
})
236239
}
237240
}
241+
242+
func TestFilterStageTargets(t *testing.T) {
243+
t.Parallel()
244+
245+
makeTC := func(name string) stageTarget {
246+
return stageTarget{
247+
deployTarget: &sdk.DeployTarget[kubeconfig.KubernetesDeployTargetConfig]{Name: name},
248+
}
249+
}
250+
251+
all := []stageTarget{makeTC("us"), makeTC("eu"), makeTC("ap")}
252+
253+
tests := []struct {
254+
name string
255+
stageTargets []string
256+
wantNames []string
257+
}{
258+
{
259+
name: "empty filter returns all targets",
260+
stageTargets: nil,
261+
wantNames: []string{"us", "eu", "ap"},
262+
},
263+
{
264+
name: "single target selected",
265+
stageTargets: []string{"us"},
266+
wantNames: []string{"us"},
267+
},
268+
{
269+
name: "multiple targets selected",
270+
stageTargets: []string{"us", "ap"},
271+
wantNames: []string{"us", "ap"},
272+
},
273+
{
274+
name: "unknown target name is silently ignored",
275+
stageTargets: []string{"unknown"},
276+
wantNames: []string{},
277+
},
278+
{
279+
name: "mix of known and unknown names",
280+
stageTargets: []string{"eu", "unknown"},
281+
wantNames: []string{"eu"},
282+
},
283+
}
284+
285+
for _, tt := range tests {
286+
t.Run(tt.name, func(t *testing.T) {
287+
t.Parallel()
288+
got := filterStageTargets(all, tt.stageTargets)
289+
names := make([]string, len(got))
290+
for i, tc := range got {
291+
names[i] = tc.deployTarget.Name
292+
}
293+
assert.Equal(t, tt.wantNames, names)
294+
})
295+
}
296+
}

‎pkg/app/pipedv1/plugin/kubernetes_multicluster/deployment/primary.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@ func (p *Plugin) executeK8sMultiPrimaryRolloutStage(ctx context.Context, input *
4545
}
4646
}
4747

48-
targets := buildStageTargets(lp, dts, cfg.Spec.Input.MultiTargets)
48+
targets := filterStageTargets(buildStageTargets(lp, dts, cfg.Spec.Input.MultiTargets), stageCfg.MultiTargets)
4949
return runOnTargets(ctx, lp, targets, func(ctx context.Context, dt *sdk.DeployTarget[kubeconfig.KubernetesDeployTargetConfig], mt *kubeconfig.KubernetesMultiTarget) sdk.StageStatus {
5050
lp.Infof("Start primary rollout for target %s", dt.Name)
5151
return p.primaryRollout(ctx, input, dt, mt, stageCfg)

‎pkg/app/pipedv1/plugin/kubernetes_multicluster/deployment/sync.go‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,15 @@ func (p *Plugin) executeK8sMultiSyncStage(ctx context.Context, input *sdk.Execut
3636
return sdk.StageStatusFailure
3737
}
3838

39-
targets := buildStageTargets(lp, dts, cfg.Spec.Input.MultiTargets)
39+
var stageCfg kubeconfig.K8sSyncStageOptions
40+
if len(input.Request.StageConfig) > 0 {
41+
if err := json.Unmarshal(input.Request.StageConfig, &stageCfg); err != nil {
42+
lp.Errorf("Failed while unmarshalling stage config (%v)", err)
43+
return sdk.StageStatusFailure
44+
}
45+
}
46+
47+
targets := filterStageTargets(buildStageTargets(lp, dts, cfg.Spec.Input.MultiTargets), stageCfg.MultiTargets)
4048
return runOnTargets(ctx, lp, targets, func(ctx context.Context, dt *sdk.DeployTarget[kubeconfig.KubernetesDeployTargetConfig], mt *kubeconfig.KubernetesMultiTarget) sdk.StageStatus {
4149
lp.Infof("Start syncing the deployment to the target %s", dt.Name)
4250
return p.sync(ctx, input, dt, mt)

‎pkg/app/pipedv1/plugin/kubernetes_multicluster/deployment/traffic.go‎

Lines changed: 10 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -97,15 +97,10 @@ func (p *Plugin) executeK8sMultiTrafficRoutingStagePodSelector(
9797
deployTargetMap[dt.Name] = dt
9898
}
9999

100-
type targetConfig struct {
101-
deployTarget *sdk.DeployTarget[kubeconfig.KubernetesDeployTargetConfig]
102-
multiTarget *kubeconfig.KubernetesMultiTarget
103-
}
104-
105-
targetConfigs := make([]targetConfig, 0, len(dts))
100+
targetConfigs := make([]stageTarget, 0, len(dts))
106101
if len(cfg.Spec.Input.MultiTargets) == 0 {
107102
for _, dt := range dts {
108-
targetConfigs = append(targetConfigs, targetConfig{deployTarget: dt})
103+
targetConfigs = append(targetConfigs, stageTarget{deployTarget: dt})
109104
}
110105
} else {
111106
for _, mt := range cfg.Spec.Input.MultiTargets {
@@ -114,10 +109,12 @@ func (p *Plugin) executeK8sMultiTrafficRoutingStagePodSelector(
114109
lp.Infof("Ignore multi target '%s': not matched any deployTarget", mt.Target.Name)
115110
continue
116111
}
117-
targetConfigs = append(targetConfigs, targetConfig{deployTarget: dt, multiTarget: &mt})
112+
targetConfigs = append(targetConfigs, stageTarget{deployTarget: dt, multiTarget: &mt})
118113
}
119114
}
120115

116+
targetConfigs = filterStageTargets(targetConfigs, stageCfg.MultiTargets)
117+
121118
eg, ctx := errgroup.WithContext(ctx)
122119
for _, tc := range targetConfigs {
123120
eg.Go(func() error {
@@ -228,15 +225,10 @@ func (p *Plugin) executeK8sMultiTrafficRoutingStageIstio(
228225
deployTargetMap[dt.Name] = dt
229226
}
230227

231-
type targetConfig struct {
232-
deployTarget *sdk.DeployTarget[kubeconfig.KubernetesDeployTargetConfig]
233-
multiTarget *kubeconfig.KubernetesMultiTarget
234-
}
235-
236-
targetConfigs := make([]targetConfig, 0, len(dts))
228+
targetConfigs := make([]stageTarget, 0, len(dts))
237229
if len(cfg.Spec.Input.MultiTargets) == 0 {
238230
for _, dt := range dts {
239-
targetConfigs = append(targetConfigs, targetConfig{deployTarget: dt})
231+
targetConfigs = append(targetConfigs, stageTarget{deployTarget: dt})
240232
}
241233
} else {
242234
for _, mt := range cfg.Spec.Input.MultiTargets {
@@ -245,10 +237,12 @@ func (p *Plugin) executeK8sMultiTrafficRoutingStageIstio(
245237
lp.Infof("Ignore multi target '%s': not matched any deployTarget", mt.Target.Name)
246238
continue
247239
}
248-
targetConfigs = append(targetConfigs, targetConfig{deployTarget: dt, multiTarget: &mt})
240+
targetConfigs = append(targetConfigs, stageTarget{deployTarget: dt, multiTarget: &mt})
249241
}
250242
}
251243

244+
targetConfigs = filterStageTargets(targetConfigs, stageCfg.MultiTargets)
245+
252246
eg, ctx := errgroup.WithContext(ctx)
253247
for _, tc := range targetConfigs {
254248
eg.Go(func() error {

0 commit comments

Comments
 (0)