zivildienst/deploymentagent/src/deploymentagent/server.go

package main

import (
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"net/http"
	"sort"
	"time"

	"gitlab.com/infektcommon/settings"
)

const (
	jsonContentType = "application/json"
)

var (
	basicAuthUsername     string
	basicAuthPassword     string
	deploymentStateFile   string
	nixosRebuildSocketUrl string
	inputQueue            *DeploymentQueue // queued
	processQueue          *DeploymentQueue // deploying, failed, deployed
)

func init() {
	basicAuthUsername = settings.GetSetting("BASIC_AUTH_USERNAME", "default")
	basicAuthPassword = settings.GetSetting("BASIC_AUTH_PASSWORD", "default")
	deploymentStateFile = settings.GetSetting("DEPLOYMENT_STATE_FILE", "./guidelines.json")
	nixosRebuildSocketUrl = settings.GetSetting("NIXOS_REBUILD_SOCKET_URL", "localhost:4444")

	inputQueue = new(DeploymentQueue)
	processQueue = new(DeploymentQueue)
}

type DeploymentAgentServer struct {
	http.Handler
}

type DeploymentSpec []struct {
	App     string `json:"app"`
	Version string `json:"version"`
}

type QueuedDeploymentSpec struct {
	Id              int64
	Status          string // one of: queued (input queue), deploying, failed, deployed (process queue)
	StatusMessage   string // status or error message
	DeploymentSpec  DeploymentSpec
	DeploymentState map[string]string
}

type DeploymentQueue struct {
	items []QueuedDeploymentSpec
}

func (q *DeploymentQueue) Enqueue(id int64, status, statusMessage string, deploymentSpec DeploymentSpec) {
	q.items = append(q.items, QueuedDeploymentSpec{
		Id:             id,
		Status:         status,
		StatusMessage:  statusMessage,
		DeploymentSpec: deploymentSpec,
	})

}

func (q *DeploymentQueue) Dequeue() QueuedDeploymentSpec {
	item := q.items[0]
	q.items = q.items[1:len(q.items)]
	return item
}

func (q *DeploymentQueue) GetAll() []QueuedDeploymentSpec {
	return q.items
}

func (q *DeploymentQueue) updateStatus(id int64, status, statusMessage string, deploymentState map[string]string) {
	for i := range q.items {
		if q.items[i].Id == id {
			p := &q.items[i]
			p.Status = status
			p.StatusMessage = statusMessage
			p.DeploymentState = deploymentState
			break
		}
	}
}

func (q *DeploymentQueue) String() string {
	return fmt.Sprint(q.items)
}

func newDeploymentAgentServer() *DeploymentAgentServer {
	server := new(DeploymentAgentServer)
	router := http.NewServeMux()

	router.Handle("/deploy", http.HandlerFunc(server.deploymentHandler))
	router.Handle("/status", http.HandlerFunc(server.statusHandler))

	server.Handler = router

	return server
}

func isAuthenticated(w http.ResponseWriter, r http.Request) (http.ResponseWriter, bool) {
	user, pass, ok := r.BasicAuth()
	if !ok {
		w.WriteHeader(http.StatusUnauthorized)
		w.Header().Add("WWW-Authenticate", `Basic realm=""`)
		w.Write([]byte(`{"Error": "No credentials present"}`))
		return w, false
	}

	if (user != basicAuthUsername && user != "default") || (pass != basicAuthPassword && pass != "default") {
		w.WriteHeader(http.StatusUnauthorized)
		w.Header().Add("WWW-Authenticate", `Basic realm=""`)
		w.Write([]byte(`{"Error": "Invalid credentials"}`))
		return w, false
	}

	return w, true
}

func (s *DeploymentAgentServer) deploymentHandler(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", jsonContentType)
	if authenticatedResponse, ok := isAuthenticated(w, *r); ok {
		w = authenticatedResponse
	} else {
		return
	}

	switch r.Method {
	case http.MethodPost:
		queueDeploymentSpec(w, r)
	default:
		w.WriteHeader(http.StatusMethodNotAllowed)
		w.Write([]byte(`{"Error": "Method not allowed"}`))
	}
}

func (s *DeploymentAgentServer) statusHandler(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", jsonContentType)
	if authenticatedResponse, ok := isAuthenticated(w, *r); ok {
		w = authenticatedResponse
	} else {
		return
	}

	switch r.Method {
	case http.MethodGet:
		var status []QueuedDeploymentSpec
		status = append(status, inputQueue.GetAll()...)
		status = append(status, processQueue.GetAll()...)
		sort.Slice(status, func(i, j int) bool {
			return status[i].Id < status[j].Id
		})

		w.WriteHeader(http.StatusAccepted)
		json.NewEncoder(w).Encode(status)
	default:
		w.WriteHeader(http.StatusMethodNotAllowed)
		w.Write([]byte(`{"Error": "Method not allowed"}`))
	}
}

func queueDeploymentSpec(w http.ResponseWriter, r *http.Request) {
	var error struct{ Error string }
	var deploymentSpec DeploymentSpec

	body, _ := ioutil.ReadAll(r.Body)
	err := json.Unmarshal(body, &deploymentSpec)
	if err != nil {
		error.Error = err.Error()
		w.WriteHeader(http.StatusBadRequest)
		json.NewEncoder(w).Encode(&error)
	} else {
		message := "Queued for deployment"
		deploymentId := time.Now().UnixNano() / 1000
		inputQueue.Enqueue(deploymentId, "queued", message, deploymentSpec)
		w.WriteHeader(http.StatusAccepted)
		w.Write([]byte(fmt.Sprintf(`{"Message": %q, "DeploymentId": "%d"}`, message, deploymentId)))
	}
}

func processDeploymentSpecs(pc *podmanClient, hc *http.Client, url string) {
	fmt.Println("Server ready to schedule deployments")
	for {
		time.Sleep(5 * time.Second)
		if len(inputQueue.GetAll()) > 0 {
			var err error

			dequequedDeploymentSpec := inputQueue.Dequeue()
			processQueue.Enqueue(dequequedDeploymentSpec.Id, "deploying", "Deployment in progress", dequequedDeploymentSpec.DeploymentSpec)

			err = pc.pullImages(dequequedDeploymentSpec.DeploymentSpec, hc, url)
			if err != nil {
				processQueue.updateStatus(dequequedDeploymentSpec.Id, "failed", ("When pulling images with Podman: " + err.Error()), getDeploymentState())
			} else {
				err := mergeDeploymentState(dequequedDeploymentSpec.DeploymentSpec)
				if err != nil {
					processQueue.updateStatus(dequequedDeploymentSpec.Id, "failed", ("When updating the deploymentstate file" + err.Error()), getDeploymentState())
				} else {
					err = triggerNixosRebuild()
					if err != nil {
						processQueue.updateStatus(dequequedDeploymentSpec.Id, "failed", ("When triggering nixos-rebuild with Systemd: " + err.Error()), getDeploymentState())
					} else {
						processQueue.updateStatus(dequequedDeploymentSpec.Id, "deployed", "Deployment successful", getDeploymentState())
					}
				}
			}

			fmt.Println(processQueue.GetAll())
		}
	}
}

func getDeploymentState() map[string]string {
	state := make(map[string]string)

	file, _ := ioutil.ReadFile(deploymentStateFile)
	err := json.Unmarshal([]byte(file), &state)
	if err != nil {
		fmt.Println(err.Error())
	}
	return state
}

func mergeDeploymentState(deploymentSpec DeploymentSpec) error {
	var deployedApps = []string{}
	deploymentState := getDeploymentState()

	for k := range deploymentState {
		deployedApps = append(deployedApps, k)
	}

	for _, app := range deploymentSpec {
		deploymentState[app.App] = app.Version
	}

	file, _ := json.MarshalIndent(deploymentState, "", " ")
	err := ioutil.WriteFile(deploymentStateFile, file, 0644)
	if err != nil {
		return err
	}

	return nil
}

// triggerNixosRebuild connects to the activation socket of a
// systemd unit which will trigger the nixos-rebuild unit.
// The connection will be kept open and closed as soon as the
// build has been completed, however, there is no way of
// knowing whether there has been an error in the build phase
// and thus the switch has not been applied.
func triggerNixosRebuild() error {
	data := make([]byte, 1)

	c, err := net.Dial("tcp", nixosRebuildSocketUrl)
	if err != nil {
		return err
	}

	if _, err := c.Read(data); err == io.EOF {
		fmt.Println("nixos-rebuild switch has finished")
		c.Close()
	} else {
		c.Close()
		return err
	}
	return nil
}