cloudstack-k8s-upgrader/pkg/upgrader/upgrader.go

package upgrader

import (
	"bytes"
	_ "embed"
	"errors"
	"fmt"
	"net/http"
	"net/smtp"
	"strings"
	"text/template"
	"time"

	"github.com/rs/zerolog/log"
	"golang.org/x/net/html"

	"git.csclub.uwaterloo.ca/cloud/cloudstack-k8s-upgrader/pkg/cloudstack"
	"git.csclub.uwaterloo.ca/cloud/cloudstack-k8s-upgrader/pkg/config"
)

//go:embed email-message.txt
var emailMessageTemplate string

type Upgrader struct {
	cfg    *config.Config
	client *cloudstack.CloudstackClient
}

func New(cfg *config.Config) *Upgrader {
	return &Upgrader{
		cfg:    cfg,
		client: cloudstack.New(cfg),
	}
}

type semanticVersion struct {
	major, minor, patch int
}

func getSemanticVersion(s string) *semanticVersion {
	semver := semanticVersion{}
	n, err := fmt.Sscanf(s, "%d.%d.%d", &semver.major, &semver.minor, &semver.patch)
	if n != 3 || err != nil {
		panic(fmt.Sprintf("Invalid semantic version string: %s", s))
	}
	return &semver
}

func (ver *semanticVersion) toString() string {
	return fmt.Sprintf("%d.%d.%d", ver.major, ver.minor, ver.patch)
}

const ISO_DOWNLOAD_BASE_URL = "https://download.cloudstack.org/cks/"

func isHyperlink(node *html.Node) bool {
	return node.Type == html.ElementNode && node.Data == "a" &&
		node.FirstChild != nil && node.FirstChild.Type == html.TextNode
}

func getNextUpgradableVersionFromHtmlNode(currentVer *semanticVersion, n *html.Node) *semanticVersion {
	if isHyperlink(n) {
		nextVer := semanticVersion{}
		n, err := fmt.Sscanf(
			n.FirstChild.Data, "setup-%d.%d.%d.iso",
			&nextVer.major, &nextVer.minor, &nextVer.patch,
		)
		if n == 3 && err == nil && nextVer.major >= currentVer.major {
			if nextVer.major > currentVer.major {
				panic("A new major version was detected. This kind of upgrade " +
					"is too dangerous to perform automatically.")
			}
			if nextVer.minor > currentVer.minor && nextVer.minor-currentVer.minor == 1 {
				return &nextVer
			}
		}
		return nil
	}
	for c := n.FirstChild; c != nil; c = c.NextSibling {
		nextVer := getNextUpgradableVersionFromHtmlNode(currentVer, c)
		if nextVer != nil {
			return nextVer
		}
	}
	return nil
}

func getNextUpgradableVersion(currentVer *semanticVersion) *semanticVersion {
	resp, err := http.Get(ISO_DOWNLOAD_BASE_URL)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()
	doc, err := html.Parse(resp.Body)
	if err != nil {
		panic(err)
	}
	return getNextUpgradableVersionFromHtmlNode(currentVer, doc)
}

func semverCompare(ver1, ver2 *semanticVersion) int {
	if ver1.major > ver2.major {
		return 1
	}
	if ver1.major < ver2.major {
		return -1
	}
	if ver1.minor > ver2.minor {
		return 1
	}
	if ver1.minor < ver2.minor {
		return -1
	}
	if ver1.patch > ver2.patch {
		return 1
	}
	if ver1.patch < ver2.patch {
		return -1
	}
	return 0
}

func getLatestExistingISO(versions []cloudstack.KubernetesSupportedVersionResponse) *cloudstack.KubernetesSupportedVersionResponse {
	latestSoFar := &versions[0]
	latestVer := getSemanticVersion(latestSoFar.SemanticVersion)
	for i, _ := range versions {
		current := &versions[i]
		currentVer := getSemanticVersion(current.SemanticVersion)
		if semverCompare(currentVer, latestVer) > 0 {
			latestSoFar = current
			latestVer = currentVer
		}
	}
	return latestSoFar
}

func (u *Upgrader) DoUpgrade() {
	cluster := u.client.GetKubernetesCluster(u.cfg.ClusterName)
	currentVersions := u.client.ListKubernetesSupportedVersions()
	if currentVersions.Count == 0 {
		panic("No k8s ISOs present")
	}
	currentISO := getLatestExistingISO(currentVersions.KubernetesSupportedVersion)
	currentVerStr := currentISO.SemanticVersion
	currentVer := getSemanticVersion(currentVerStr)
	nextVer := getNextUpgradableVersion(currentVer)
	if nextVer == nil {
		return
	}
	nextVerStr := nextVer.toString()
	log.Info().
		Str("currentVer", currentVerStr).
		Str("nextVer", nextVerStr).
		Msg("New version available")
	downloadURL := ISO_DOWNLOAD_BASE_URL + "setup-" + nextVerStr + ".iso"

	log.Info().Msg("Adding new k8s ISO")
	newISO := u.client.AddKubernetesSupportedVersion(nextVerStr, downloadURL)

	log.Info().Msg("Upgrading k8s cluster")
	err := u.client.UpgradeKubernetesCluster(cluster.Id, newISO.Id)
	var jobFailedError *cloudstack.JobFailedError
	if errors.As(err, &jobFailedError) &&
		strings.Contains(jobFailedError.ErrorText, "unable to upgrade Kubernetes node on VM") {
		// The upgrade might fail the first time.
		// Reboot the cluster and try again.
		log.Warn().
			Str("jobID", jobFailedError.JobID).
			Int("resultCode", jobFailedError.ResultCode).
			Msg("Job failed")
		log.Warn().Msg("Stopping cluster")
		u.client.StopKubernetesCluster(cluster.Id)
		log.Warn().Msg("Starting cluster")
		u.client.StartKubernetesCluster(cluster.Id)
		log.Debug().Msg("Waiting for cluster to stabilize after reboot")
		time.Sleep(5 * time.Minute)
		log.Info().Msg("Upgrading k8s cluster (2nd attempt)")
		err = u.client.UpgradeKubernetesCluster(cluster.Id, newISO.Id)
	}
	if err != nil {
		panic(err)
	}

	log.Info().Msg("Deleting old k8s ISO")
	err = u.client.DeleteKubernetesSupportedVersion(currentISO.Id)
	if err != nil {
		log.Warn().Msg("Could not delete old k8s ISO: " + err.Error())
	}
	u.editKubeControllerManagerFlags(cluster)
	u.sendEmailNotification(nextVerStr)
}

func (u *Upgrader) sendEmailNotification(nextVerStr string) {
	tmpl := template.Must(template.New("email-message").Parse(emailMessageTemplate))
	data := map[string]interface{}{
		"cfg":        u.cfg,
		"date":       time.Now().Format(time.RFC1123Z),
		"nextVerStr": nextVerStr,
	}
	var buf bytes.Buffer
	if err := tmpl.Execute(&buf, data); err != nil {
		panic(err)
	}
	// The lines of the body need to be CRLF terminated
	// See https://pkg.go.dev/net/smtp#SendMail
	// Make sure that email-message.txt uses Unix-style LF endings
	msg := bytes.ReplaceAll(buf.Bytes(), []byte("\n"), []byte("\r\n"))
	log.Debug().
		Str("to", u.cfg.EmailRecipient).
		Msg("sending email notification")
	if err := smtp.SendMail(
		u.cfg.EmailServer,
		nil, /* auth */
		u.cfg.EmailSender,
		[]string{u.cfg.EmailRecipient},
		msg,
	); err != nil {
		panic(err)
	}
}