forked from argoproj/argo-workflows
-
Notifications
You must be signed in to change notification settings - Fork 0
/
resource.go
299 lines (268 loc) · 9.39 KB
/
resource.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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
package executor
import (
"bufio"
"bytes"
"encoding/json"
"fmt"
"io/ioutil"
"os/exec"
"strings"
"time"
log "github.com/sirupsen/logrus"
"github.com/tidwall/gjson"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/util/wait"
"github.com/argoproj/argo/errors"
)
// ExecResource will run kubectl action against a manifest
func (we *WorkflowExecutor) ExecResource(action string, manifestPath string) (string, string, error) {
isDelete := action == "delete"
args := []string{
action,
}
output := "json"
if isDelete {
args = append(args, "--ignore-not-found")
output = "name"
}
if action == "patch" {
mergeStrategy := "strategic"
if we.Template.Resource.MergeStrategy != "" {
mergeStrategy = we.Template.Resource.MergeStrategy
}
args = append(args, "--type")
args = append(args, mergeStrategy)
args = append(args, "-p")
buff, err := ioutil.ReadFile(manifestPath)
if err != nil {
return "", "", errors.New(errors.CodeBadRequest, err.Error())
}
args = append(args, string(buff))
}
args = append(args, "-f")
args = append(args, manifestPath)
args = append(args, "-o")
args = append(args, output)
cmd := exec.Command("kubectl", args...)
log.Info(strings.Join(cmd.Args, " "))
out, err := cmd.Output()
if err != nil {
exErr := err.(*exec.ExitError)
errMsg := strings.TrimSpace(string(exErr.Stderr))
return "", "", errors.New(errors.CodeBadRequest, errMsg)
}
if action == "delete" {
return "", "", nil
}
obj := unstructured.Unstructured{}
err = json.Unmarshal(out, &obj)
if err != nil {
return "", "", err
}
resourceName := fmt.Sprintf("%s.%s/%s", obj.GroupVersionKind().Kind, obj.GroupVersionKind().Group, obj.GetName())
log.Infof("%s/%s", obj.GetNamespace(), resourceName)
return obj.GetNamespace(), resourceName, nil
}
// gjsonLabels is an implementation of labels.Labels interface
// which allows us to take advantage of k8s labels library
// for the purposes of evaluating fail and success conditions
type gjsonLabels struct {
json []byte
}
// Has returns whether the provided label exists.
func (g gjsonLabels) Has(label string) bool {
return gjson.GetBytes(g.json, label).Exists()
}
// Get returns the value for the provided label.
func (g gjsonLabels) Get(label string) string {
return gjson.GetBytes(g.json, label).String()
}
// WaitResource waits for a specific resource to satisfy either the success or failure condition
func (we *WorkflowExecutor) WaitResource(resourceNamespace string, resourceName string) error {
if we.Template.Resource.SuccessCondition == "" && we.Template.Resource.FailureCondition == "" {
return nil
}
var successReqs labels.Requirements
if we.Template.Resource.SuccessCondition != "" {
successSelector, err := labels.Parse(we.Template.Resource.SuccessCondition)
if err != nil {
return errors.Errorf(errors.CodeBadRequest, "success condition '%s' failed to parse: %v", we.Template.Resource.SuccessCondition, err)
}
log.Infof("Waiting for conditions: %s", successSelector)
successReqs, _ = successSelector.Requirements()
}
var failReqs labels.Requirements
if we.Template.Resource.FailureCondition != "" {
failSelector, err := labels.Parse(we.Template.Resource.FailureCondition)
if err != nil {
return errors.Errorf(errors.CodeBadRequest, "fail condition '%s' failed to parse: %v", we.Template.Resource.FailureCondition, err)
}
log.Infof("Failing for conditions: %s", failSelector)
failReqs, _ = failSelector.Requirements()
}
// Start the condition result reader using PollImmediateInfinite
// Poll intervall of 5 seconds serves as a backoff intervall in case of immediate result reader failure
err := wait.PollImmediateInfinite(time.Second*5,
func() (bool, error) {
isErrRetry, err := checkResourceState(resourceNamespace, resourceName, successReqs, failReqs)
if err == nil {
log.Infof("Returning from successful wait for resource %s", resourceName)
return true, nil
}
if isErrRetry {
log.Infof("Waiting for resource %s resulted in retryable error %v", resourceName, err)
return false, nil
}
log.Warnf("Waiting for resource %s resulted in non-retryable error %v", resourceName, err)
return false, err
})
if err != nil {
if err == wait.ErrWaitTimeout {
log.Warnf("Waiting for resource %s resulted in timeout due to repeated errors", resourceName)
} else {
log.Warnf("Waiting for resource %s resulted in error %v", resourceName, err)
}
}
return err
}
// Function to do the kubectl get -w command and then waiting on json reading.
func checkResourceState(resourceNamespace string, resourceName string, successReqs labels.Requirements, failReqs labels.Requirements) (bool, error) {
cmd, reader, err := startKubectlWaitCmd(resourceNamespace, resourceName)
if err != nil {
return false, err
}
defer func() {
_ = cmd.Process.Kill()
}()
for {
jsonBytes, err := readJSON(reader)
if err != nil {
resultErr := err
log.Warnf("Json reader returned error %v. Calling kill (usually superfluous)", err)
// We don't want to write OS specific code so we don't want to call syscall package code. But that means
// there is no way to figure out if a process is running or not in an asynchronous manner. exec.Wait will
// always block and we need to call that to get the exit code of the process. So we will unconditionally
// call exec.Process.Kill and then assume that wait will not block after that. Two things may happen:
// 1. Process already exited and kill does nothing (returns error which we ignore) and then we call
// Wait and get the proper return value
// 2. Process is running gets, killed with exec.Process.Kill call and Wait returns an error code and we give up
// and don't retry
_ = cmd.Process.Kill()
log.Warnf("Command for kubectl get -w for %s exited. Getting return value using Wait", resourceName)
err = cmd.Wait()
if err != nil {
log.Warnf("cmd.Wait for kubectl get -w command for resource %s returned error %v",
resourceName, err)
resultErr = err
} else {
log.Infof("readJSon failed for resource %s but cmd.Wait for kubectl get -w command did not error", resourceName)
}
return true, resultErr
}
log.Info(string(jsonBytes))
ls := gjsonLabels{json: jsonBytes}
for _, req := range failReqs {
failed := req.Matches(ls)
msg := fmt.Sprintf("failure condition '%s' evaluated %v", req, failed)
log.Infof(msg)
if failed {
// TODO: need a better error code instead of BadRequest
return false, errors.Errorf(errors.CodeBadRequest, msg)
}
}
numMatched := 0
for _, req := range successReqs {
matched := req.Matches(ls)
log.Infof("success condition '%s' evaluated %v", req, matched)
if matched {
numMatched++
}
}
log.Infof("%d/%d success conditions matched", numMatched, len(successReqs))
if numMatched >= len(successReqs) {
return false, nil
}
}
}
// Start Kubectl command Get with -w return error if unable to start command
func startKubectlWaitCmd(resourceNamespace string, resourceName string) (*exec.Cmd, *bufio.Reader, error) {
args := []string{"get", resourceName, "-w", "-o", "json"}
if resourceNamespace != "" {
args = append(args, "-n", resourceNamespace)
}
cmd := exec.Command("kubectl", args...)
stdout, err := cmd.StdoutPipe()
if err != nil {
return nil, nil, errors.InternalWrapError(err)
}
reader := bufio.NewReader(stdout)
log.Info(strings.Join(cmd.Args, " "))
if err := cmd.Start(); err != nil {
return nil, nil, errors.InternalWrapError(err)
}
return cmd, reader, nil
}
// readJSON reads from a reader line-by-line until it reaches "}\n" indicating end of json
func readJSON(reader *bufio.Reader) ([]byte, error) {
var buffer bytes.Buffer
for {
line, err := reader.ReadBytes('\n')
if err != nil {
return nil, err
}
isDelimiter := len(line) == 2 && line[0] == byte('}')
line = bytes.TrimSpace(line)
_, err = buffer.Write(line)
if err != nil {
return nil, err
}
if isDelimiter {
break
}
}
return buffer.Bytes(), nil
}
// SaveResourceParameters will save any resource output parameters
func (we *WorkflowExecutor) SaveResourceParameters(resourceNamespace string, resourceName string) error {
if len(we.Template.Outputs.Parameters) == 0 {
log.Infof("No output parameters")
return nil
}
log.Infof("Saving resource output parameters")
for i, param := range we.Template.Outputs.Parameters {
if param.ValueFrom == nil {
continue
}
var cmd *exec.Cmd
if param.ValueFrom.JSONPath != "" {
args := []string{"get", resourceName, "-o", fmt.Sprintf("jsonpath=%s", param.ValueFrom.JSONPath)}
if resourceNamespace != "" {
args = append(args, "-n", resourceNamespace)
}
cmd = exec.Command("kubectl", args...)
} else if param.ValueFrom.JQFilter != "" {
resArgs := []string{resourceName}
if resourceNamespace != "" {
resArgs = append(resArgs, "-n", resourceNamespace)
}
cmdStr := fmt.Sprintf("kubectl get %s -o json | jq -c '%s'", strings.Join(resArgs, " "), param.ValueFrom.JQFilter)
cmd = exec.Command("sh", "-c", cmdStr)
} else {
continue
}
log.Info(cmd.Args)
out, err := cmd.Output()
if err != nil {
if exErr, ok := err.(*exec.ExitError); ok {
log.Errorf("`%s` stderr:\n%s", cmd.Args, string(exErr.Stderr))
}
return errors.InternalWrapError(err)
}
output := string(out)
we.Template.Outputs.Parameters[i].Value = &output
log.Infof("Saved output parameter: %s, value: %s", param.Name, output)
}
err := we.AnnotateOutputs(nil)
return err
}