task.go 9.46 KB
Newer Older
Lukas Burgey's avatar
Lukas Burgey committed
1
2
3
package main

import (
4
	"bytes"
Lukas Burgey's avatar
Lukas Burgey committed
5
6
	"encoding/json"
	"fmt"
Lukas Burgey's avatar
Lukas Burgey committed
7
	"io"
Lukas Burgey's avatar
Lukas Burgey committed
8
9
10
11
	"io/ioutil"
	"log"
	"net/http"
	"os/exec"
Lukas Burgey's avatar
Lukas Burgey committed
12
	"strings"
Lukas Burgey's avatar
Lukas Burgey committed
13
14
	"time"

Lukas Burgey's avatar
Lukas Burgey committed
15
	s "git.scc.kit.edu/feudal/feudalScripts"
Lukas Burgey's avatar
Lukas Burgey committed
16
17
18
19
)

type (
	service struct {
Lukas Burgey's avatar
Lukas Burgey committed
20
21
22
		Name        string `json:"name"`
		Command     string `json:"command"`
		Description string `json:"description"`
Lukas Burgey's avatar
Lukas Burgey committed
23
24
	}

Lukas Burgey's avatar
Lukas Burgey committed
25
26
27
28
	group struct {
		Name string `json:"name"`
	}

29
	// task describes a deployment action
Lukas Burgey's avatar
Lukas Burgey committed
30
	task struct {
Lukas Burgey's avatar
Lukas Burgey committed
31
32
33
34
35
36
37
		ID            int                       `json:"id"`
		StateTarget   s.State                   `json:"state_target"`
		User          s.User                    `json:"user"`
		Key           s.SSHKey                  `json:"key"`
		Credentials   map[string][]s.Credential `json:"credentials"`
		Questionnaire map[string]string         `json:"questionnaire"`
		Group         group                     `json:"group"`
Lukas Burgey's avatar
Lukas Burgey committed
38

39
		// Service may be overwritten by scheduleTask
Lukas Burgey's avatar
Lukas Burgey committed
40
41
42
		Service service `json:"service,omitempty"`

		// Exchange and RoutingKey are inserted when we receive the task via RabbitMQ
Lukas Burgey's avatar
Lukas Burgey committed
43
44
		Exchange   string
		RoutingKey string
Lukas Burgey's avatar
Lukas Burgey committed
45
46
	}

47
48
	// taskReply is sent back to the backend to inform it about executed deployment actions
	taskReply struct {
Lukas Burgey's avatar
Lukas Burgey committed
49
		// ID of the according task (at the backend: DeploymentState)
50
51
		ID     int      `json:"id"`
		Output s.Output `json:"output"`
Lukas Burgey's avatar
Lukas Burgey committed
52

53
		// Service is the service for which we performed the deploy action
Lukas Burgey's avatar
Lukas Burgey committed
54
		Service service `json:"service"`
Lukas Burgey's avatar
Lukas Burgey committed
55
56
	}

Lukas Burgey's avatar
Lukas Burgey committed
57
58
59
60
	// ackResponse response of the backend for our response
	ackResponse struct {
		Error string `json:"error"`
	}
Lukas Burgey's avatar
Lukas Burgey committed
61
62
63
64
65
66
67
)

func (s service) String() string {
	return s.Name
}

func (t task) String() string {
Lukas Burgey's avatar
Lukas Burgey committed
68
69
70
71
	return fmt.Sprintf("[Task:%v:%v]", t.ID, t.Service.Name)
}

func (t task) Log(formatString string, params ...interface{}) {
Lukas Burgey's avatar
Lukas Burgey committed
72
	log.Printf("%s "+formatString, append([]interface{}{t}, params...)...)
Lukas Burgey's avatar
Lukas Burgey committed
73
74
}

75
func (te taskReply) String() string {
Lukas Burgey's avatar
Lukas Burgey committed
76
77
78
	return fmt.Sprintf("[Task:%v:%v]", te.ID, te.Service.Name)
}

79
func (te taskReply) Log(formatString string, params ...interface{}) {
Lukas Burgey's avatar
Lukas Burgey committed
80
	log.Printf("%s "+formatString, append([]interface{}{te}, params...)...)
Lukas Burgey's avatar
Lukas Burgey committed
81
82
83
}

func (c *config) taskServices(t task) (services []service, err error) {
Lukas Burgey's avatar
Lukas Burgey committed
84
85
86
87
88
89
90
91
92
93
94
95
	// hacky: we need to determine exchange and routing key when we fetched manually from the rest
	// interface
	if t.Exchange == "" || t.RoutingKey == "" {
		if t.Service != (service{}) {
			t.Exchange = "services"
			t.RoutingKey = t.Service.Name
		} else if t.Group != (group{}) {
			t.Exchange = "groups"
			t.RoutingKey = t.Group.Name
		}
	}

Lukas Burgey's avatar
Lukas Burgey committed
96
97
98
99
100
101
102
103
	switch t.Exchange {
	case "services":
		for _, svc := range c.Services {
			if svc.Name == t.RoutingKey {
				services = []service{svc}
				return
			}
		}
Lukas Burgey's avatar
Lukas Burgey committed
104

Lukas Burgey's avatar
Lukas Burgey committed
105
106
107
108
109
		for _, groupServices := range c.GroupToServices {
			for _, groupService := range groupServices {
				if groupService.Name == t.RoutingKey {
					services = []service{groupService}
					return
Lukas Burgey's avatar
Lukas Burgey committed
110
111
112
113
				}
			}
		}

Lukas Burgey's avatar
Lukas Burgey committed
114
115
116
117
118
	case "groups":
		var ok bool
		services, ok = c.GroupToServices[t.RoutingKey]
		if !ok {
			err = fmt.Errorf(
Lukas Burgey's avatar
Lukas Burgey committed
119
				"%s: Group %s has no mapped services",
Lukas Burgey's avatar
Lukas Burgey committed
120
				t,
Lukas Burgey's avatar
Lukas Burgey committed
121
				t.RoutingKey,
Lukas Burgey's avatar
Lukas Burgey committed
122
123
124
125
126
127
128
			)
			return
		}

	case "sites":
		// TODO implement
	}
Lukas Burgey's avatar
Lukas Burgey committed
129
130

	if len(services) == 0 {
Lukas Burgey's avatar
Lukas Burgey committed
131
		t.Log("Unable to determine services for this task")
Lukas Burgey's avatar
Lukas Burgey committed
132
	}
Lukas Burgey's avatar
Lukas Burgey committed
133
	return
Lukas Burgey's avatar
Lukas Burgey committed
134
135
}

136
137
138
139
// fetches tasks (which rabbitmq missed) manually
func (c *config) taskFetcher() {
	// start ticker
	ticker := time.NewTicker(c.FetchInterval)
Lukas Burgey's avatar
Lukas Burgey committed
140

Lukas Burgey's avatar
Lukas Burgey committed
141
142
143
144
145
146
	for {
		go func() {
			if err := c.fetchTasks(); err != nil {
				log.Printf("[Fetch] error fetching: %s", err)
			}
		}()
Lukas Burgey's avatar
Lukas Burgey committed
147

Lukas Burgey's avatar
Lukas Burgey committed
148
149
150
		// wait for a tick
		<-ticker.C
	}
Lukas Burgey's avatar
Lukas Burgey committed
151
152
}

153
// handles tasks in c.NewTasks
Lukas Burgey's avatar
Lukas Burgey committed
154
155
func (c *config) taskHandler() {
	for newTask := range c.NewTasks {
Lukas Burgey's avatar
Lukas Burgey committed
156
		handleWrapper := func(t task) {
Lukas Burgey's avatar
Lukas Burgey committed
157
			if err := c.handleTask(t); err != nil {
Lukas Burgey's avatar
Lukas Burgey committed
158
159
				t.Log("Error handling task: %s", err)

160
				c.DoneTasks <- taskReply{
Lukas Burgey's avatar
Lukas Burgey committed
161
162
					ID:      t.ID,
					Service: t.Service,
163
164
					Output: s.Output{
						State: s.Failed,
Lukas Burgey's avatar
Lukas Burgey committed
165
166
167
168
						// TODO maybe another message for the user
						Msg: fmt.Sprint(err),
					},
				}
Lukas Burgey's avatar
Lukas Burgey committed
169
			}
Lukas Burgey's avatar
Lukas Burgey committed
170
171
172
173
174
175
176
177
		}

		if *sequentialExecution {
			handleWrapper(newTask)
		} else {
			// handle tasks asynchronously
			go handleWrapper(newTask)
		}
Lukas Burgey's avatar
Lukas Burgey committed
178
179
180
	}
}

Lukas Burgey's avatar
Lukas Burgey committed
181
// responds to the backend concering tasks in c.DoneTasks
182
func (c *config) taskResponder() {
Lukas Burgey's avatar
Lukas Burgey committed
183
184
	for doneTask := range c.DoneTasks {
		// ack tasks asynchronously
185
		go func(te taskReply) {
186
			err := c.respondToTask(te)
Lukas Burgey's avatar
Lukas Burgey committed
187
			if err != nil {
Lukas Burgey's avatar
Lukas Burgey committed
188
				te.Log("ACK-ERROR: %s", err)
189
190

				// reschedule failed responses
191
				go func(te taskReply) {
192
193
194
					time.Sleep(time.Minute)
					c.DoneTasks <- te
				}(te)
Lukas Burgey's avatar
Lukas Burgey committed
195
196
197
198
199
200
				return
			}
		}(doneTask)
	}
}

201
// IMPLEMENTATIONS
Lukas Burgey's avatar
Lukas Burgey committed
202
func (c *config) fetchTasks() (err error) {
Lukas Burgey's avatar
Lukas Burgey committed
203
	if len(c.Services) == 0 && len(c.GroupToServices) == 0 {
Lukas Burgey's avatar
Lukas Burgey committed
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
		log.Printf("[Fetch] Not fetching because the are no services to fetch")
		return
	}

	// construct a request
	uri := "https://" + c.Host + "/backend/clientapi/deployments"
	req, err := http.NewRequest("GET", uri, nil)
	if err != nil {
		return
	}

	// add query params for the services
	q := req.URL.Query()
	for _, service := range c.Services {
		q.Add("s", service.Name)
	}
Lukas Burgey's avatar
Lukas Burgey committed
220
221
222
	for groupName := range c.GroupToServices {
		q.Add("g", groupName)
	}
Lukas Burgey's avatar
Lukas Burgey committed
223
224
225
226
227
228
229
230
231
232
233
	req.URL.RawQuery = q.Encode()

	req.SetBasicAuth(c.Username, c.Password)

	// execute the request
	resp, err := client.Do(req)
	if err != nil {
		return
	}
	defer resp.Body.Close()

Lukas Burgey's avatar
Lukas Burgey committed
234
	if resp.StatusCode != 200 {
Lukas Burgey's avatar
Lukas Burgey committed
235
236
237
238
		err = fmt.Errorf("Unable to fetch tasks (response: %v)", resp.Status)
		return
	}

Lukas Burgey's avatar
Lukas Burgey committed
239
	var newTasks []task
Lukas Burgey's avatar
Lukas Burgey committed
240
241
242
243
	body, err := ioutil.ReadAll(resp.Body)

	err = json.Unmarshal(body, &newTasks)
	if err != nil {
Lukas Burgey's avatar
Lukas Burgey committed
244
		err = fmt.Errorf("Error unmarshaling: %s\n%s", err, body)
Lukas Burgey's avatar
Lukas Burgey committed
245
246
		return
	}
Lukas Burgey's avatar
Lukas Burgey committed
247

Lukas Burgey's avatar
Lukas Burgey committed
248
	log.Printf("[Fetch] %d new tasks: %v", len(newTasks), newTasks)
Lukas Burgey's avatar
Lukas Burgey committed
249
	for _, task := range newTasks {
Lukas Burgey's avatar
Lukas Burgey committed
250
251
		err = c.scheduleTask(task)
		if err != nil {
Lukas Burgey's avatar
Lukas Burgey committed
252
253
254
255
			log.Printf("[Fetch] %v", err)

			// we execute all tasks even when one fails
			err = nil
Lukas Burgey's avatar
Lukas Burgey committed
256
		}
Lukas Burgey's avatar
Lukas Burgey committed
257
258
259
260
261
	}
	return
}

// scheduleTask puts tasks in the newTasks channel
Lukas Burgey's avatar
Lukas Burgey committed
262
func (c *config) scheduleTask(t task) (err error) {
Lukas Burgey's avatar
Lukas Burgey committed
263
264
	// determine the services of the task
	var services []service
Lukas Burgey's avatar
Lukas Burgey committed
265
	services, err = c.taskServices(t)
Lukas Burgey's avatar
Lukas Burgey committed
266
267
268
269
270
271
	if err != nil {
		return
	}

	// schedule one task per service of the task
	for _, svc := range services {
Lukas Burgey's avatar
Lukas Burgey committed
272
273
274
		var serviceTask task
		serviceTask = t
		serviceTask.Service = svc
Lukas Burgey's avatar
Lukas Burgey committed
275

Lukas Burgey's avatar
Lukas Burgey committed
276
		serviceTask.Log("Scheduling execution for %s", svc.Name)
Lukas Burgey's avatar
Lukas Burgey committed
277

Lukas Burgey's avatar
Lukas Burgey committed
278
		// put the in the task queue
Lukas Burgey's avatar
Lukas Burgey committed
279
		c.NewTasks <- serviceTask
Lukas Burgey's avatar
Lukas Burgey committed
280
281
282
	}
	return
}
Lukas Burgey's avatar
Lukas Burgey committed
283

Lukas Burgey's avatar
Lukas Burgey committed
284
func (c *config) handleTask(t task) (err error) {
Lukas Burgey's avatar
Lukas Burgey committed
285

286
	var output s.Output
287
288

	// encode input as json
289
	input := s.Input{
Lukas Burgey's avatar
Lukas Burgey committed
290
291
292
		StateTarget:   t.StateTarget,
		User:          t.User,
		Questionnaire: t.Questionnaire,
Lukas Burgey's avatar
Lukas Burgey committed
293
		Credentials:   t.Credentials,
294
	}
295

296
297
298
299
300
	iBytes, err := input.Marshal()
	if err != nil {
		return
	}

Lukas Burgey's avatar
Lukas Burgey committed
301
302
303
304
305
	// TODO execute in parallel
	var (
		stdin          io.WriteCloser
		stdout, stderr io.ReadCloser
	)
Lukas Burgey's avatar
Lukas Burgey committed
306

Lukas Burgey's avatar
Lukas Burgey committed
307
308
	// execute the script
	if *scriptDebugging {
Lukas Burgey's avatar
Lukas Burgey committed
309
310
		t.Log("Executing: '%s'", t.Service.Command)
		t.Log("Input: %s", input)
Lukas Burgey's avatar
Lukas Burgey committed
311
	}
312

Lukas Burgey's avatar
Lukas Burgey committed
313
314
	commandParts := strings.Split(t.Service.Command, " ")
	cmd := exec.Command(commandParts[0], commandParts[1:]...)
Lukas Burgey's avatar
Lukas Burgey committed
315
316
317
318
319
320
321
322
323
324
325
326
	stdin, err = cmd.StdinPipe()
	if err != nil {
		return
	}
	stdout, err = cmd.StdoutPipe()
	if err != nil {
		return
	}
	stderr, err = cmd.StderrPipe()
	if err != nil {
		return
	}
327

Lukas Burgey's avatar
Lukas Burgey committed
328
329
330
331
332
333
	err = cmd.Start()
	if err != nil {
		return
	}
	stdin.Write(iBytes)
	stdin.Close()
334

Lukas Burgey's avatar
Lukas Burgey committed
335
336
337
338
339
340
	// decode json output
	var outputBytes, logOutputBytes []byte
	outputBytes, err = ioutil.ReadAll(stdout)
	if err != nil {
		return
	}
341

Lukas Burgey's avatar
Lukas Burgey committed
342
343
344
345
346
	logOutputBytes, err = ioutil.ReadAll(stderr)
	if err != nil {
		return
	}
	if *scriptDebugging {
Lukas Burgey's avatar
Lukas Burgey committed
347
348
349
350
		if len(logOutputBytes) > 0 {
			t.Log("Logs:\n%s", logOutputBytes)
			t.Log("End of Logs")
		}
Lukas Burgey's avatar
Lukas Burgey committed
351
	}
Lukas Burgey's avatar
Lukas Burgey committed
352

Lukas Burgey's avatar
Lukas Burgey committed
353
354
355
356
	err = cmd.Wait()
	if err != nil {
		return
	}
Lukas Burgey's avatar
Lukas Burgey committed
357

Lukas Burgey's avatar
Lukas Burgey committed
358
359
	err = json.Unmarshal(outputBytes, &output)
	if err != nil {
Lukas Burgey's avatar
Lukas Burgey committed
360
361
362
		if len(outputBytes) > 0 {
			t.Log("Raw output: %s", outputBytes)
		}
Lukas Burgey's avatar
Lukas Burgey committed
363
364
365
		return
	}
	if *scriptDebugging {
Lukas Burgey's avatar
Lukas Burgey committed
366
		t.Log("Output: %s", output)
Lukas Burgey's avatar
Lukas Burgey committed
367
	}
368

369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
	// if the script does not provide any credential specific states we expect all
	// credentials to have the same state as the whole execution (output.State)
	if output.UserCredentialStates == nil {
		if *scriptDebugging {
			t.Log("The script did not provided credential specific states")
		}

		var userCredentialStates = make(s.UserCredentialStates)
		for credType, credList := range t.Credentials {
			for _, credential := range credList {
				if stateMap, ok := userCredentialStates[credType]; !ok {
					userCredentialStates[credType] = map[string]s.State{
						credential.Name: output.State,
					}
				} else {
					stateMap[credential.Name] = output.State
385
386
387
				}
			}
		}
388
		output.UserCredentialStates = userCredentialStates
389
390
391
	}

	c.DoneTasks <- taskReply{
392
393
394
		ID:      t.ID,
		Service: t.Service,
		Output:  output,
Lukas Burgey's avatar
Lukas Burgey committed
395
	}
396
	return
Lukas Burgey's avatar
Lukas Burgey committed
397
}
Lukas Burgey's avatar
Lukas Burgey committed
398

399
func (c *config) respondToTask(te taskReply) (err error) {
Lukas Burgey's avatar
Lukas Burgey committed
400
	taskResponse, err := json.MarshalIndent(te, "", "    ")
Lukas Burgey's avatar
Lukas Burgey committed
401
402
403
404
	if err != nil {
		return
	}

405
406
407
408
	if *backendDebugging {
		log.Printf("Replying to the backend: %s", taskResponse)
	}

Lukas Burgey's avatar
Lukas Burgey committed
409
	url := fmt.Sprintf("https://%s/backend/clientapi/response", c.Host)
Lukas Burgey's avatar
Lukas Burgey committed
410
	req, err := http.NewRequest("POST", url, bytes.NewReader(taskResponse))
Lukas Burgey's avatar
Lukas Burgey committed
411
412
413
414
415
416
417
418
419
420
421
422
423
424
	if err != nil {
		return
	}
	req.SetBasicAuth(c.Username, c.Password)
	req.Header.Set("Content-Type", "application/json")

	// execute request
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return
	}
	defer resp.Body.Close()

	if resp.StatusCode == 200 {
Lukas Burgey's avatar
Lukas Burgey committed
425
		te.Log("Successful response")
Lukas Burgey's avatar
Lukas Burgey committed
426
	} else {
Lukas Burgey's avatar
Lukas Burgey committed
427
		te.Log("Response Status: %v", resp.StatusCode)
Lukas Burgey's avatar
Lukas Burgey committed
428
429
430
431
432
433
434
435
436
437

		var (
			ackResponseBytes []byte
			ackResponse      ackResponse
		)
		ackResponseBytes, err = ioutil.ReadAll(resp.Body)
		if err != nil {
			return
		}
		err = json.Unmarshal(ackResponseBytes, &ackResponse)
Lukas Burgey's avatar
Lukas Burgey committed
438
439
440
		if err != nil {
			return
		}
Lukas Burgey's avatar
Lukas Burgey committed
441
		te.Log("Response to our response: '%s'", ackResponse.Error)
Lukas Burgey's avatar
Lukas Burgey committed
442
	}
Lukas Burgey's avatar
Lukas Burgey committed
443
444
	return
}