-
Notifications
You must be signed in to change notification settings - Fork 1.7k
/
task.result.go
85 lines (72 loc) · 2.3 KB
/
task.result.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
package pipeline
import (
"context"
"database/sql/driver"
"encoding/json"
"fmt"
"github.com/pkg/errors"
"gopkg.in/guregu/null.v4"
)
// TODO: This task type is no longer necessary, we should deprecate/remove it.
// See: https://www.pivotaltracker.com/story/show/176557536
// ResultTask exists solely as a Postgres performance optimization. It's added
// automatically to the end of every pipeline, and it receives the outputs of all
// tasks that have no successor tasks. This allows the pipeline runner to detect
// when it has reached the end a given pipeline simply by checking the `successor_id`
// field, rather than having to try to SELECT all of the pipeline run's task runs,
// (which must be done from inside of a transaction, and causes lock contention
// and serialization anomaly issues).
type ResultTask struct {
BaseTask `mapstructure:",squash"`
}
var _ Task = (*ResultTask)(nil)
func (t *ResultTask) Type() TaskType {
return TaskTypeResult
}
func (t *ResultTask) SetDefaults(inputValues map[string]string, g TaskDAG, self taskDAGNode) error {
return nil
}
func (t *ResultTask) Run(_ context.Context, taskRun TaskRun, inputs []Result) Result {
values := make([]interface{}, len(inputs))
errors := make(FinalErrors, len(inputs))
for i, input := range inputs {
values[i] = input.Value
if input.Error != nil {
errors[i] = null.StringFrom(input.Error.Error())
}
}
return Result{Value: values, Error: errors}
}
// FIXME: This error/FinalErrors conflation exists solely because of the __result__ task.
// It is confusing and needs to go, making this note to remove it along with the
// special __result__ task.
// https://www.pivotaltracker.com/story/show/176557536
type FinalErrors []null.String
func (fe FinalErrors) HasErrors() bool {
for _, err := range fe {
if !err.IsZero() {
return true
}
}
return false
}
func (fe FinalErrors) Error() string {
bs, err := json.Marshal(fe)
if err != nil {
return `["could not unmarshal final pipeline errors"]`
}
return string(bs)
}
func (fe FinalErrors) Value() (driver.Value, error) {
return fe.Error(), nil
}
func (fe *FinalErrors) Scan(value interface{}) error {
switch v := value.(type) {
case []byte:
return json.Unmarshal(v, fe)
case string:
return json.Unmarshal([]byte(v), fe)
default:
return errors.New(fmt.Sprintf("%s", value))
}
}