Skip to content

Commit f62cd92

Browse files
l-qingl-qing
committed
feat(pipeline): allow variable substitution in pipeline.tasks[].onError
fix #8564 In the reconcile logic of PipelineRun, add the ability to replace the `pipeline.tasks[].onError` field. At the same time, the validation of pipelineSpec legality has been delayed.
1 parent a44d224 commit f62cd92

File tree

6 files changed

+154
-54
lines changed

6 files changed

+154
-54
lines changed

‎docs/variables.md‎

Lines changed: 17 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -141,12 +141,25 @@ For instructions on using variable substitutions see the relevant section of [th
141141
| `TaskRun` | `spec.workspaces[].csi.driver` |
142142
| `TaskRun` | `spec.workspaces[].csi.nodePublishSecretRef.name` |
143143
| `Pipeline` | `spec.tasks[].params[].value` |
144-
| `Pipeline` | `spec.tasks[].conditions[].params[].value` |
145-
| `Pipeline` | `spec.results[].value` |
144+
| `Pipeline` | `spec.tasks[].matrix.params[].value` |
145+
| `Pipeline` | `spec.tasks[].matrix.include[].params[].value` |
146+
| `Pipeline` | `spec.tasks[].displayName` |
147+
| `Pipeline` | `spec.tasks[].workspaces[].subPath` |
146148
| `Pipeline` | `spec.tasks[].when[].input` |
147149
| `Pipeline` | `spec.tasks[].when[].values` |
148-
| `Pipeline` | `spec.tasks[].workspaces[].subPath` |
149-
| `Pipeline` | `spec.tasks[].displayName` |
150+
| `Pipeline` | `spec.tasks[].taskRef.params[].values` |
151+
| `Pipeline` | `spec.tasks[].taskRef.name` |
152+
| `Pipeline` | `spec.tasks[].onError` |
153+
| `Pipeline` | `spec.finally[].params[].value` |
154+
| `Pipeline` | `spec.finally[].matrix.params[].value` |
155+
| `Pipeline` | `spec.finally[].matrix.include[].params[].value` |
156+
| `Pipeline` | `spec.finally[].displayName` |
157+
| `Pipeline` | `spec.finally[].workspaces[].subPath` |
158+
| `Pipeline` | `spec.finally[].when[].input` |
159+
| `Pipeline` | `spec.finally[].when[].values` |
160+
| `Pipeline` | `spec.finally[].taskRef.params[].values` |
161+
| `Pipeline` | `spec.finally[].taskRef.name` |
162+
| `Pipeline` | `spec.finally[].onError` |
150163
| `PipelineRun` | `spec.workspaces[].subPath` |
151164
| `PipelineRun` | `spec.workspaces[].persistentVolumeClaim.claimName` |
152165
| `PipelineRun` | `spec.workspaces[].configMap.name` |

‎pkg/apis/pipeline/v1/pipeline_validation.go‎

Lines changed: 15 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -191,15 +191,7 @@ func (pt PipelineTask) Validate(ctx context.Context) (errs *apis.FieldError) {
191191
ClusterTaskRefKind: true,
192192
}
193193

194-
if pt.OnError != "" {
195-
errs = errs.Also(config.ValidateEnabledAPIFields(ctx, "OnError", config.BetaAPIFields))
196-
if pt.OnError != PipelineTaskContinue && pt.OnError != PipelineTaskStopAndFail {
197-
errs = errs.Also(apis.ErrInvalidValue(pt.OnError, "OnError", "PipelineTask OnError must be either \"continue\" or \"stopAndFail\""))
198-
}
199-
if pt.OnError == PipelineTaskContinue && pt.Retries > 0 {
200-
errs = errs.Also(apis.ErrGeneric("PipelineTask OnError cannot be set to \"continue\" when Retries is greater than 0"))
201-
}
202-
}
194+
errs = errs.Also(pt.ValidateOnError(ctx))
203195

204196
// Pipeline task having taskRef/taskSpec with APIVersion is classified as custom task
205197
switch {
@@ -217,6 +209,20 @@ func (pt PipelineTask) Validate(ctx context.Context) (errs *apis.FieldError) {
217209
return errs
218210
}
219211

212+
// ValidateOnError validates the OnError field of a PipelineTask
213+
func (pt PipelineTask) ValidateOnError(ctx context.Context) (errs *apis.FieldError) {
214+
if pt.OnError != "" && !isParamRefs(string(pt.OnError)) {
215+
errs = errs.Also(config.ValidateEnabledAPIFields(ctx, "OnError", config.BetaAPIFields))
216+
if pt.OnError != PipelineTaskContinue && pt.OnError != PipelineTaskStopAndFail {
217+
errs = errs.Also(apis.ErrInvalidValue(pt.OnError, "OnError", "PipelineTask OnError must be either \"continue\" or \"stopAndFail\""))
218+
}
219+
if pt.OnError == PipelineTaskContinue && pt.Retries > 0 {
220+
errs = errs.Also(apis.ErrGeneric("PipelineTask OnError cannot be set to \"continue\" when Retries is greater than 0"))
221+
}
222+
}
223+
return errs
224+
}
225+
220226
func (pt *PipelineTask) validateMatrix(ctx context.Context) (errs *apis.FieldError) {
221227
if pt.IsMatrixed() {
222228
// This is a beta feature and will fail validation if it's used in a pipeline spec

‎pkg/reconciler/pipelinerun/pipelinerun.go‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -563,6 +563,15 @@ func (c *Reconciler) reconcile(ctx context.Context, pr *v1.PipelineRun, getPipel
563563
// Update pipelinespec of pipelinerun's status field
564564
pr.Status.PipelineSpec = pipelineSpec
565565

566+
// validate pipelineSpec after apply parameters
567+
if err := validatePipelineSpecAfterApplyParameters(ctx, pipelineSpec); err != nil {
568+
// This Run has failed, so we need to mark it as failed and stop reconciling it
569+
pr.Status.MarkFailed(v1.PipelineRunReasonFailedValidation.String(),
570+
"Pipeline %s/%s can't be Run; it has an invalid spec: %s",
571+
pipelineMeta.Namespace, pipelineMeta.Name, pipelineErrors.WrapUserError(err))
572+
return controller.NewPermanentError(err)
573+
}
574+
566575
// pipelineState holds a list of pipeline tasks after fetching their resolved Task specs.
567576
// pipelineState also holds a taskRun for each pipeline task after the taskRun is created
568577
// pipelineState is instantiated and updated on every reconcile cycle
@@ -1679,3 +1688,19 @@ func conditionFromVerificationResult(verificationResult *trustedresources.Verifi
16791688
}
16801689
return condition, err
16811690
}
1691+
1692+
// validatePipelineSpecAfterApplyParameters validates the PipelineSpec after apply parameters
1693+
// Maybe some fields are modified during apply parameters, need to validate again. For example, tasks[].OnError.
1694+
func validatePipelineSpecAfterApplyParameters(ctx context.Context, pipelineSpec *v1.PipelineSpec) (errs *apis.FieldError) {
1695+
if pipelineSpec == nil {
1696+
errs = errs.Also(apis.ErrMissingField("PipelineSpec"))
1697+
return
1698+
}
1699+
tasks := make([]v1.PipelineTask, 0, len(pipelineSpec.Tasks)+len(pipelineSpec.Finally))
1700+
tasks = append(tasks, pipelineSpec.Tasks...)
1701+
tasks = append(tasks, pipelineSpec.Finally...)
1702+
for _, t := range tasks {
1703+
errs = errs.Also(t.ValidateOnError(ctx))
1704+
}
1705+
return errs
1706+
}

‎pkg/reconciler/pipelinerun/pipelinerun_test.go‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17994,6 +17994,55 @@ func Test_runNextSchedulableTask(t *testing.T) {
1799417994
}
1799517995
}
1799617996

17997+
func TestReconcile_InvalidOnErrorPipeline(t *testing.T) {
17998+
names.TestingSeed()
17999+
18000+
namespace := "foo"
18001+
prName := "test-pipeline-invalid-onerror"
18002+
18003+
prs := []*v1.PipelineRun{
18004+
parse.MustParseV1PipelineRun(t, `
18005+
metadata:
18006+
name: test-pipeline-invalid-onerror
18007+
namespace: foo
18008+
spec:
18009+
params:
18010+
- name: onerror
18011+
value: "invalid"
18012+
pipelineSpec:
18013+
tasks:
18014+
- name: echo
18015+
onError: $(params.onerror)
18016+
taskSpec:
18017+
steps:
18018+
- name: echo
18019+
image: ubuntu
18020+
script: |
18021+
echo "Hello, World!"
18022+
exit 1
18023+
`),
18024+
}
18025+
18026+
d := test.Data{
18027+
PipelineRuns: prs,
18028+
ConfigMaps: []*corev1.ConfigMap{newFeatureFlagsConfigMap()},
18029+
}
18030+
prt := newPipelineRunTest(t, d)
18031+
defer prt.Cancel()
18032+
18033+
wantEvents := []string{
18034+
"Normal Started",
18035+
"(?s)Warning Failed .*PipelineTask OnError must be either \"continue\" or \"stopAndFail\"",
18036+
"(?s)Warning InternalError .*OnError\nPipelineTask OnError must be either \"continue\" or \"stopAndFail\"",
18037+
}
18038+
reconciledRun, clients := prt.reconcileRun(namespace, prName, wantEvents, true)
18039+
18040+
// Check that the expected TaskRun was not created
18041+
taskRuns := getTaskRunsForPipelineRun(prt.TestAssets.Ctx, t, clients, namespace, prName)
18042+
validateTaskRunsCount(t, taskRuns, 0)
18043+
verifyTaskRunStatusesCount(t, reconciledRun.Status, 0)
18044+
}
18045+
1799718046
func getSignedV1Pipeline(unsigned *pipelinev1.Pipeline, signer signature.Signer, name string) (*pipelinev1.Pipeline, error) {
1799818047
signed := unsigned.DeepCopy()
1799918048
signed.Name = name

‎pkg/reconciler/pipelinerun/resources/apply.go‎

Lines changed: 28 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -298,55 +298,42 @@ func ApplyWorkspaces(p *v1.PipelineSpec, pr *v1.PipelineRun) *v1.PipelineSpec {
298298
return ApplyReplacements(p, replacements, map[string][]string{}, map[string]map[string]string{})
299299
}
300300

301-
// ApplyReplacements replaces placeholders for declared parameters with the specified replacements.
302-
func ApplyReplacements(p *v1.PipelineSpec, replacements map[string]string, arrayReplacements map[string][]string, objectReplacements map[string]map[string]string) *v1.PipelineSpec {
303-
p = p.DeepCopy()
304-
305-
for i := range p.Tasks {
306-
p.Tasks[i].Params = p.Tasks[i].Params.ReplaceVariables(replacements, arrayReplacements, objectReplacements)
307-
if p.Tasks[i].IsMatrixed() {
308-
p.Tasks[i].Matrix.Params = p.Tasks[i].Matrix.Params.ReplaceVariables(replacements, arrayReplacements, nil)
309-
for j := range p.Tasks[i].Matrix.Include {
310-
p.Tasks[i].Matrix.Include[j].Params = p.Tasks[i].Matrix.Include[j].Params.ReplaceVariables(replacements, nil, nil)
301+
// replaceVariablesInPipelineTasks handles variable replacement for a slice of PipelineTasks in-place
302+
func replaceVariablesInPipelineTasks(tasks []v1.PipelineTask, replacements map[string]string,
303+
arrayReplacements map[string][]string, objectReplacements map[string]map[string]string) {
304+
305+
for i := range tasks {
306+
tasks[i].Params = tasks[i].Params.ReplaceVariables(replacements, arrayReplacements, objectReplacements)
307+
if tasks[i].IsMatrixed() {
308+
tasks[i].Matrix.Params = tasks[i].Matrix.Params.ReplaceVariables(replacements, arrayReplacements, nil)
309+
for j := range tasks[i].Matrix.Include {
310+
tasks[i].Matrix.Include[j].Params = tasks[i].Matrix.Include[j].Params.ReplaceVariables(replacements, nil, nil)
311311
}
312312
} else {
313-
p.Tasks[i].DisplayName = substitution.ApplyReplacements(p.Tasks[i].DisplayName, replacements)
313+
tasks[i].DisplayName = substitution.ApplyReplacements(tasks[i].DisplayName, replacements)
314314
}
315-
for j := range p.Tasks[i].Workspaces {
316-
p.Tasks[i].Workspaces[j].SubPath = substitution.ApplyReplacements(p.Tasks[i].Workspaces[j].SubPath, replacements)
315+
for j := range tasks[i].Workspaces {
316+
tasks[i].Workspaces[j].SubPath = substitution.ApplyReplacements(tasks[i].Workspaces[j].SubPath, replacements)
317317
}
318-
p.Tasks[i].When = p.Tasks[i].When.ReplaceVariables(replacements, arrayReplacements)
319-
if p.Tasks[i].TaskRef != nil {
320-
if p.Tasks[i].TaskRef.Params != nil {
321-
p.Tasks[i].TaskRef.Params = p.Tasks[i].TaskRef.Params.ReplaceVariables(replacements, arrayReplacements, objectReplacements)
318+
tasks[i].When = tasks[i].When.ReplaceVariables(replacements, arrayReplacements)
319+
if tasks[i].TaskRef != nil {
320+
if tasks[i].TaskRef.Params != nil {
321+
tasks[i].TaskRef.Params = tasks[i].TaskRef.Params.ReplaceVariables(replacements, arrayReplacements, objectReplacements)
322322
}
323-
p.Tasks[i].TaskRef.Name = substitution.ApplyReplacements(p.Tasks[i].TaskRef.Name, replacements)
323+
tasks[i].TaskRef.Name = substitution.ApplyReplacements(tasks[i].TaskRef.Name, replacements)
324324
}
325-
p.Tasks[i] = propagateParams(p.Tasks[i], replacements, arrayReplacements, objectReplacements)
325+
tasks[i].OnError = v1.PipelineTaskOnErrorType(substitution.ApplyReplacements(string(tasks[i].OnError), replacements))
326+
tasks[i] = propagateParams(tasks[i], replacements, arrayReplacements, objectReplacements)
326327
}
328+
}
327329

328-
for i := range p.Finally {
329-
p.Finally[i].Params = p.Finally[i].Params.ReplaceVariables(replacements, arrayReplacements, objectReplacements)
330-
if p.Finally[i].IsMatrixed() {
331-
p.Finally[i].Matrix.Params = p.Finally[i].Matrix.Params.ReplaceVariables(replacements, arrayReplacements, nil)
332-
for j := range p.Finally[i].Matrix.Include {
333-
p.Finally[i].Matrix.Include[j].Params = p.Finally[i].Matrix.Include[j].Params.ReplaceVariables(replacements, nil, nil)
334-
}
335-
} else {
336-
p.Finally[i].DisplayName = substitution.ApplyReplacements(p.Finally[i].DisplayName, replacements)
337-
}
338-
for j := range p.Finally[i].Workspaces {
339-
p.Finally[i].Workspaces[j].SubPath = substitution.ApplyReplacements(p.Finally[i].Workspaces[j].SubPath, replacements)
340-
}
341-
p.Finally[i].When = p.Finally[i].When.ReplaceVariables(replacements, arrayReplacements)
342-
if p.Finally[i].TaskRef != nil {
343-
if p.Finally[i].TaskRef.Params != nil {
344-
p.Finally[i].TaskRef.Params = p.Finally[i].TaskRef.Params.ReplaceVariables(replacements, arrayReplacements, objectReplacements)
345-
}
346-
p.Finally[i].TaskRef.Name = substitution.ApplyReplacements(p.Finally[i].TaskRef.Name, replacements)
347-
}
348-
p.Finally[i] = propagateParams(p.Finally[i], replacements, arrayReplacements, objectReplacements)
349-
}
330+
// ApplyReplacements replaces placeholders for declared parameters with the specified replacements.
331+
func ApplyReplacements(p *v1.PipelineSpec, replacements map[string]string, arrayReplacements map[string][]string, objectReplacements map[string]map[string]string) *v1.PipelineSpec {
332+
p = p.DeepCopy()
333+
334+
// Replace variables in Tasks and Finally tasks
335+
replaceVariablesInPipelineTasks(p.Tasks, replacements, arrayReplacements, objectReplacements)
336+
replaceVariablesInPipelineTasks(p.Finally, replacements, arrayReplacements, objectReplacements)
350337

351338
return p
352339
}

‎pkg/reconciler/pipelinerun/resources/apply_test.go‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1771,6 +1771,26 @@ func TestApplyParameters(t *testing.T) {
17711771
}},
17721772
},
17731773
},
1774+
{
1775+
name: "parameter in onError",
1776+
original: v1.PipelineSpec{
1777+
Params: []v1.ParamSpec{
1778+
{Name: "onerror", Type: v1.ParamTypeString},
1779+
},
1780+
Tasks: []v1.PipelineTask{{
1781+
OnError: v1.PipelineTaskOnErrorType("$(params.onerror)"),
1782+
}},
1783+
},
1784+
params: v1.Params{{Name: "onerror", Value: *v1.NewStructuredValues("new-onerror-value")}},
1785+
expected: v1.PipelineSpec{
1786+
Params: []v1.ParamSpec{
1787+
{Name: "onerror", Type: v1.ParamTypeString},
1788+
},
1789+
Tasks: []v1.PipelineTask{{
1790+
OnError: v1.PipelineTaskOnErrorType("new-onerror-value"),
1791+
}},
1792+
},
1793+
},
17741794
} {
17751795
t.Run(tt.name, func(t *testing.T) {
17761796
t.Parallel()

0 commit comments

Comments
 (0)