Fix the task fetch

consumer.go

@@ -43,12 +43,14 @@ func (c *config) consumer() (cons *consumer) {
 			exchange{
 				"services", "topic", serviceRoutingKeys,
 			},
-			exchange{
-				"sites", "topic", []string{c.Site},
-			},
 			exchange{
 				"groups", "topic", groupRoutingKeys,
 			},
+			/*
+				exchange{
+					"sites", "topic", []string{c.Site},
+				},
+			*/
 		},
 		consumer:         c.consume,
 		reconnectTimeout: c.ReconnectTimeout,
@@ -199,8 +201,8 @@ func (c *config) consume(deliveries <-chan amqp.Delivery) {
 			newTask.RoutingKey = d.RoutingKey
 
 			// enqueue the task
-			log.Printf("[Task] %v: RECEIVED", newTask)
-			c.NewTasks <- newTask
+			log.Printf("[Task] %v: RECEIVED via %v-%v", newTask, d.Exchange, d.RoutingKey)
+			c.scheduleTask(newTask)
 			d.Ack(
 				false, // multiple
 			)

main.go

@@ -2,13 +2,10 @@ package main
 
 import (
 	"encoding/json"
-	"fmt"
 	"io/ioutil"
 	"log"
 	"net/http"
 	"os"
-	"os/signal"
-	"syscall"
 	"time"
 
 	"gopkg.in/alecthomas/kingpin.v2"
@@ -37,7 +34,7 @@ type (
 
 		// GroupToServices maps a group name to services provided for this group
 		// this is for  deployment per _group_
-		GroupToServices map[string]([]service) `json:"group_to_services"` //
+		GroupToServices map[string]([]service) `json:"group_to_services"`
 
 		FetchIntervalString    string `json:"fetch_interval"`    // string parsed by time.ParseDuration
 		ReconnectTimeoutString string `json:"reconnect_timeout"` // string parsed by time.ParseDuration
@@ -60,7 +57,9 @@ var (
 	).Version(
 		"0.3.0",
 	)
-	configFile = app.Arg("config", "Config file to file to use.").Required().String()
+	configFile       = app.Arg("config", "Config file to file to use.").Required().String()
+	scriptDebugging  = app.Flag("debug-scripts", "Display debugging info concerning executed scripts").Bool()
+	backendDebugging = app.Flag("debug-backend", "Display debugging info concerning the backend").Bool()
 )
 
 func logError(err error, msg string) {
@@ -183,55 +182,9 @@ func getConfig(configFile string) (c config, err error) {
 	return
 }
 
-func signalHandler() {
-	exits := make(chan int)
-	signals := make(chan os.Signal, 1)
-
-	go func(exits chan int) {
-		for e := range exits {
-			os.Exit(e)
-		}
-	}(exits)
-
-	signal.Notify(signals,
-		syscall.SIGHUP,
-		syscall.SIGINT,
-		syscall.SIGTERM,
-		syscall.SIGQUIT,
-	)
-	for signal := range signals {
-		switch signal {
-		case syscall.SIGHUP:
-			fmt.Println("hungup")
-			exits <- 0
-
-		// kill -SIGINT XXXX or Ctrl+c
-		case syscall.SIGINT:
-			fmt.Println("Warikomi")
-			exits <- 0
-
-		// kill -SIGTERM XXXX
-		case syscall.SIGTERM:
-			fmt.Println("force stop")
-			exits <- 0
-
-		// kill -SIGQUIT XXXX
-		case syscall.SIGQUIT:
-			fmt.Println("stop and core dump")
-			exits <- 1
-
-		default:
-			fmt.Println("Unknown signal.")
-
-		}
-	}
-}
-
 func main() {
 	var err error
 
-	go signalHandler()
-
 	// get arguments
 	kingpin.MustParse(app.Parse(os.Args[1:]))
 

task.go

@@ -21,29 +21,39 @@ type (
 		Description string `json:"description"`
 	}
 
+	group struct {
+		Name string `json:"name"`
+	}
+
 	task struct {
 		ID            int               `json:"id"`
 		StateTarget   string            `json:"state_target"`
 		User          scripts.User      `json:"user"`
 		Key           scripts.SSHKey    `json:"key"`
 		Questionnaire map[string]string `json:"questionnaire"`
+		Group         group             `json:"group"`
 
+		// maybe overwritten by scheduleTask
+		Service service `json:"service,omitempty"`
+
+		// Exchange and RoutingKey are inserted when we receive the task via RabbitMQ
 		Exchange   string
 		RoutingKey string
 	}
 
 	taskExecution struct {
-		// ID of the according task
+		// ID of the according task (at the backend: DeploymentState)
 		ID     int            `json:"id"`
 		Output scripts.Output `json:"output"`
 
 		// we need to inform the backend for which service we deployed
 		Service service `json:"service"`
 	}
-)
 
-var (
-	scriptDebugging = true
+	// ackResponse response of the backend for our response
+	ackResponse struct {
+		Error string `json:"error"`
+	}
 )
 
 func (s service) String() string {
@@ -55,6 +65,18 @@ func (t task) String() string {
 }
 
 func (c *config) taskServices(t task) (services []service, err error) {
+	// 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
+		}
+	}
+
 	switch t.Exchange {
 	case "services":
 		for _, svc := range c.Services {
@@ -63,7 +85,20 @@ func (c *config) taskServices(t task) (services []service, err error) {
 				return
 			}
 		}
-		return
+
+		// try the group based services
+		// TODO is this desired?
+		if len(services) == 0 {
+			for _, groupServices := range c.GroupToServices {
+				for _, groupService := range groupServices {
+					if groupService.Name == t.RoutingKey {
+						services = []service{groupService}
+						return
+					}
+				}
+			}
+		}
+
 	case "groups":
 		var ok bool
 		services, ok = c.GroupToServices[t.RoutingKey]
@@ -76,10 +111,13 @@ func (c *config) taskServices(t task) (services []service, err error) {
 			return
 		}
 
-		return
 	case "sites":
 		// TODO implement
 	}
+
+	if len(services) == 0 {
+		log.Printf("Unable to determine services for task %s", t)
+	}
 	return
 }
 
@@ -109,19 +147,21 @@ func (c *config) taskHandler() {
 		go func(t task) {
 			if err := c.handleTask(t); err != nil {
 				log.Printf("[Task] Error handling task: %s", err)
-				/*
-					TODO
-					go func(t task) {
-						time.Sleep(1 * time.Minute)
-						c.NewTasks <- t
-					}(t)
-				*/
+				c.DoneTasks <- taskExecution{
+					ID: t.ID,
+					Output: scripts.Output{
+						Status: scripts.Failed,
+						// TODO maybe another message for the user
+						Msg: fmt.Sprint(err),
+					},
+					Service: t.Service,
+				}
 			}
 		}(newTask)
 	}
 }
 
-// acks tasks in c.DoneTasks
+// responds to the backend concering tasks in c.DoneTasks
 func (c *config) taskResponder() {
 	for doneTask := range c.DoneTasks {
 		// ack tasks asynchronously
@@ -131,7 +171,6 @@ func (c *config) taskResponder() {
 				log.Printf("[Task] %v: ACK-ERROR: %s", te.ID, err)
 
 				// reschedule failed responses
-				// TODO does this work?
 				go func(te taskExecution) {
 					time.Sleep(time.Minute)
 					c.DoneTasks <- te
@@ -184,14 +223,38 @@ func (c *config) fetchTasks() (err error) {
 		return
 	}
 
-	log.Printf("[Fetch] %d new tasks", len(newTasks))
+	log.Printf("[Fetch] %d new tasks:%v", len(newTasks), newTasks)
 	for _, task := range newTasks {
+		c.scheduleTask(task)
+	}
+	return
+}
+
+// scheduleTask puts tasks in the newTasks channel
+func (c *config) scheduleTask(ti task) (err error) {
+	// determine the services of the task
+	var services []service
+	services, err = c.taskServices(ti)
+	if err != nil {
+		return
+	}
+
+	// schedule one task per service of the task
+	for _, svc := range services {
+		var t task
+		t = ti
+		t.Service = svc
+
+		log.Printf("[Task:%v] scheduling for %s", t.ID, svc.Name)
+
 		// put the in the task queue
-		c.NewTasks <- task
+		c.NewTasks <- t
 	}
 	return
 }
+
 func (c *config) handleTask(ti task) (err error) {
+
 	var output scripts.Output
 
 	// encode input as json
@@ -206,81 +269,77 @@ func (c *config) handleTask(ti task) (err error) {
 		return
 	}
 
-	// determine the services of the task
-	var services []service
-	services, err = c.taskServices(ti)
-	if err != nil {
-		return
-	}
-
-	for _, svc := range services {
-		// TODO execute in parallel
-		var (
-			stdin          io.WriteCloser
-			stdout, stderr io.ReadCloser
-		)
+	// TODO execute in parallel
+	var (
+		stdin          io.WriteCloser
+		stdout, stderr io.ReadCloser
+	)
 
-		commandName := svc.Command
-		// execute the script
+	commandName := ti.Service.Command
+	// execute the script
+	if *scriptDebugging {
 		log.Printf("[Task:%v] Executing: '%s'", ti.ID, commandName)
-		if scriptDebugging {
-			log.Printf("[Task:%v] Input: %s", ti.ID, input)
-		}
+		log.Printf("[Task:%v] Input: %s", ti.ID, input)
+	}
 
-		cmd := exec.Command(commandName)
-		stdin, err = cmd.StdinPipe()
-		if err != nil {
-			return
-		}
-		stdout, err = cmd.StdoutPipe()
-		if err != nil {
-			return
-		}
-		stderr, err = cmd.StderrPipe()
-		if err != nil {
-			return
-		}
+	cmd := exec.Command(commandName)
+	stdin, err = cmd.StdinPipe()
+	if err != nil {
+		return
+	}
+	stdout, err = cmd.StdoutPipe()
+	if err != nil {
+		return
+	}
+	stderr, err = cmd.StderrPipe()
+	if err != nil {
+		return
+	}
 
-		err = cmd.Start()
-		if err != nil {
-			return
-		}
-		stdin.Write(iBytes)
-		stdin.Close()
+	err = cmd.Start()
+	if err != nil {
+		return
+	}
+	stdin.Write(iBytes)
+	stdin.Close()
 
-		// decode json output
-		var outputBytes, logOutputBytes []byte
-		outputBytes, err = ioutil.ReadAll(stdout)
-		if err != nil {
-			return
-		}
+	// decode json output
+	var outputBytes, logOutputBytes []byte
+	outputBytes, err = ioutil.ReadAll(stdout)
+	if err != nil {
+		return
+	}
 
-		logOutputBytes, err = ioutil.ReadAll(stderr)
-		if err != nil {
-			return
-		}
+	logOutputBytes, err = ioutil.ReadAll(stderr)
+	if err != nil {
+		return
+	}
+	if *scriptDebugging {
 		log.Printf("[Task:%v] Logs:\n%s", ti.ID, logOutputBytes)
 		log.Printf("[Task:%v] End of Logs", ti.ID)
+	}
 
-		err = cmd.Wait()
-		if err != nil {
-			return
-		}
+	err = cmd.Wait()
+	if err != nil {
+		return
+	}
 
-		err = json.Unmarshal(outputBytes, &output)
-		if err != nil {
-			return
-		}
+	err = json.Unmarshal(outputBytes, &output)
+	if err != nil {
+		return
+	}
+	if *scriptDebugging {
 		log.Printf("[Task:%v] Output: %s", ti.ID, output)
+	}
 
-		c.DoneTasks <- taskExecution{
-			ID:      ti.ID,
-			Output:  output,
-			Service: svc,
-		}
+	c.DoneTasks <- taskExecution{
+		ID:      ti.ID,
+		Output:  output,
+		Service: ti.Service,
 	}
 	return
 }
+
 func (c *config) respondToTask(te taskExecution) (err error) {
 	taskResponse, err := json.MarshalIndent(te, "", "    ")
 	if err != nil {
@@ -303,16 +362,23 @@ func (c *config) respondToTask(te taskExecution) (err error) {
 	defer resp.Body.Close()
 
 	if resp.StatusCode == 200 {
-		log.Printf("[Task:%v] Successful ACK", te.ID)
+		log.Printf("[Task:%v:%v] Successful response", te.ID, te.Service.Name)
 	} else {
-		var ackResponse []byte
-		ackResponse, err = ioutil.ReadAll(resp.Body)
+		log.Printf("[Task:%v:%v] Response Status: %v", te.ID, te.Service.Name, resp.StatusCode)
+
+		var (
+			ackResponseBytes []byte
+			ackResponse      ackResponse
+		)
+		ackResponseBytes, err = ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return
+		}
+		err = json.Unmarshal(ackResponseBytes, &ackResponse)
 		if err != nil {
-			log.Printf("[Task:%v] ACK Status: %v", te.ID, resp.StatusCode)
 			return
 		}
-		log.Printf("[Tsak:%v] ACK Response: %s", te.ID, ackResponse)
-		log.Printf("[Task:%v] ACK Status: %v", te.ID, resp.StatusCode)
+		log.Printf("[Task:%v:%v] Response to our response: '%s'", te.ID, te.Service.Name, ackResponse.Error)
 	}
 	return
 }