task.go

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"net/http"
	"os/exec"
	"strings"
	"time"

	s "git.scc.kit.edu/feudal/feudalScripts"
)

type (
	service struct {
		Name        string `json:"name"`
		Command     string `json:"command"`
		Description string `json:"description"`
	}

	vo struct {
		ID           int    `json:"id"`
		Name         string `json:"name"`
		ResourceType string `json:"resourcetype"`
	}

	// task describes a deployment action
	task struct {
		ID            int                       `json:"id"`
		StateTarget   s.State                   `json:"state_target"`
		User          s.User                    `json:"user"`
		Credentials   map[string][]s.Credential `json:"credentials"`
		Questionnaire map[string]string         `json:"questionnaire"`
		VO            vo                        `json:"vo"`

		// Service may be overwritten by scheduleTask
		Service service `json:"service,omitempty"`

		// Exchange and RoutingKey are inserted when we receive the task via RabbitMQ
		Exchange   string
		RoutingKey string
	}

	// taskReply is sent back to the backend to inform it about executed deployment actions
	taskReply struct {
		// ID of the according task (at the backend: DeploymentState)
		ID     int      `json:"id"`
		Output s.Output `json:"output"`

		// Service is the service for which we performed the deploy action
		Service service `json:"service"`
	}

	// ackResponse response of the backend for our response
	ackResponse struct {
		Error string `json:"error"`
	}
)

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

func (t task) String() string {
	return fmt.Sprintf("[Task:%v:%v]", t.ID, t.Service.Name)
}

func (t task) Log(formatString string, params ...interface{}) {
	log.Printf("%s "+formatString, append([]interface{}{t}, params...)...)
}

func (te taskReply) String() string {
	return fmt.Sprintf("[Task:%v:%v]", te.ID, te.Service.Name)
}

func (te taskReply) Log(formatString string, params ...interface{}) {
	log.Printf("%s "+formatString, append([]interface{}{te}, params...)...)
}

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.VO.ResourceType == "Group" {
			t.Exchange = "groups"
			t.RoutingKey = t.VO.Name
		} else if t.VO.ResourceType == "Entitlement" {
			t.Exchange = "entitlements"
			t.RoutingKey = t.VO.Name
		}
	}

	switch t.Exchange {
	case "groups":
		var ok bool
		services, ok = c.GroupToServices[t.RoutingKey]
		if !ok {
			err = fmt.Errorf(
				"%s: Group %s has no mapped services",
				t,
				t.RoutingKey,
			)
			return
		}

	case "entitlements":
		var ok bool
		services, ok = c.EntitlementToServices[t.RoutingKey]
		if !ok {
			err = fmt.Errorf(
				"%s: Entitlement %s has no mapped services",
				t,
				t.RoutingKey,
			)
			return
		}
	case "sites":
		// TODO implement
	}

	if len(services) == 0 {
		t.Log("Unable to determine services for this task")
	}
	return
}

// fetches tasks (which rabbitmq missed) manually
func (c *config) taskFetcher() {
	// start ticker
	ticker := time.NewTicker(c.FetchInterval)

	for {
		go func() {
			if err := c.fetchTasks(); err != nil {
				log.Printf("[Fetch] error fetching: %s", err)
			}
		}()

		// wait for a tick
		<-ticker.C
	}
}

// handles tasks in c.NewTasks
func (c *config) taskHandler() {
	for newTask := range c.NewTasks {
		handleWrapper := func(t task) {
			if err := c.handleTask(t); err != nil {
				t.Log("Error handling task: %s", err)

				c.DoneTasks <- taskReply{
					ID:      t.ID,
					Service: t.Service,
					Output: s.Output{
						State: s.Failed,
						// TODO maybe another message for the user
						Msg: fmt.Sprint(err),
					},
				}
			}
		}

		if *sequentialExecution {
			handleWrapper(newTask)
		} else {
			// handle tasks asynchronously
			go handleWrapper(newTask)
		}
	}
}

// responds to the backend concering tasks in c.DoneTasks
func (c *config) taskResponder() {
	for doneTask := range c.DoneTasks {
		// ack tasks asynchronously
		go func(te taskReply) {
			err := c.respondToTask(te)
			if err != nil {
				te.Log("ACK-ERROR: %s", err)

				// reschedule failed responses
				go func(te taskReply) {
					time.Sleep(time.Minute)
					c.DoneTasks <- te
				}(te)
				return
			}
		}(doneTask)
	}
}

// IMPLEMENTATIONS
func (c *config) fetchTasks() (err error) {
	if len(c.EntitlementToServices) == 0 && len(c.GroupToServices) == 0 {
		log.Printf("[HTTP] 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 groupName := range c.GroupToServices {
		q.Add("g", groupName)
	}
	for entitlementName := range c.EntitlementToServices {
		q.Add("e", entitlementName)
	}
	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()

	if resp.StatusCode != 200 {
		err = fmt.Errorf("[HTTP] Unable to fetch tasks (response: %v)", resp.Status)
		return
	}

	var newTasks []task
	body, err := ioutil.ReadAll(resp.Body)

	err = json.Unmarshal(body, &newTasks)
	if err != nil {
		err = fmt.Errorf("[HTTP] Error unmarshaling: %s\n%s", err, body)
		return
	}

	if *backendDebugging {
		indented := new(bytes.Buffer)
		if err := json.Indent(indented, body, "", "  "); err != nil {
			log.Printf("[HTTP] Fetched (unable to indent: %v):\n%s", err, body)
		} else {
			log.Printf("[HTTP] Fetched:\n%s", indented)
		}
	}

	for _, task := range newTasks {
		err = c.scheduleTask(task)
		if err != nil {
			log.Printf("[HTTP] Scheduling: %v", err)

			// we execute all tasks even when one fails
			err = nil
		}
	}
	return
}

// scheduleTask puts tasks in the newTasks channel
func (c *config) scheduleTask(t task) (err error) {
	// determine the services of the task
	var services []service
	services, err = c.taskServices(t)
	if err != nil {
		return
	}

	// schedule one task per service of the task
	for _, svc := range services {
		var serviceTask task
		serviceTask = t
		serviceTask.Service = svc

		serviceTask.Log("Scheduling execution for %s", svc.Name)

		// put the in the task queue
		c.NewTasks <- serviceTask
	}
	return
}

func (c *config) handleTask(t task) (err error) {

	var output s.Output

	// encode input as json
	input := s.Input{
		StateTarget:   t.StateTarget,
		User:          t.User,
		Questionnaire: t.Questionnaire,
		Credentials:   t.Credentials,
	}

	iBytes, err := input.Marshal()
	if err != nil {
		return
	}

	// TODO execute in parallel
	var (
		stdin          io.WriteCloser
		stdout, stderr io.ReadCloser
	)

	// execute the script
	if *scriptDebugging {
		t.Log("Executing: '%s'", t.Service.Command)
		t.Log("Input: %s", input)
	}

	commandParts := strings.Split(t.Service.Command, " ")
	cmd := exec.Command(commandParts[0], commandParts[1:]...)
	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()

	// 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
	}
	if *scriptDebugging {
		if len(logOutputBytes) > 0 {
			t.Log("Logs:\n%s", logOutputBytes)
			t.Log("End of Logs")
		}
	}

	err = cmd.Wait()
	if err != nil {
		return
	}
	if *scriptDebugging {
		var indented = new(bytes.Buffer)
		err = json.Indent(indented, outputBytes, "  ", "  ")
		if err != nil {
			t.Log("Output (unable to indent: %s):\n%s", err, outputBytes)
		} else {
			t.Log("Output:\n%s", indented.String())
		}
	}

	err = json.Unmarshal(outputBytes, &output)
	if err != nil {
		if len(outputBytes) > 0 {
			t.Log("Raw output: %s", outputBytes)
		}
		return
	}

	// 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 {
		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
				}
			}
		}
		output.UserCredentialStates = userCredentialStates
	}

	c.DoneTasks <- taskReply{
		ID:      t.ID,
		Service: t.Service,
		Output:  output,
	}
	return
}

func (c *config) respondToTask(te taskReply) (err error) {
	taskResponse, err := json.MarshalIndent(te, "", "    ")
	if err != nil {
		return
	}

	if *backendDebugging {
		log.Printf("Task Response:\n%s", taskResponse)
	}

	url := fmt.Sprintf("https://%s/backend/clientapi/response", c.Host)
	req, err := http.NewRequest("POST", url, bytes.NewReader(taskResponse))
	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 {
		te.Log("Successful response")
	} else {
		te.Log("Response Status: %v", 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 {
			return
		}
		te.Log("Response to our response: '%s'", ackResponse.Error)
	}
	return
}