status-reporter/main.go

// Copyright (C) 2019 Marius Schellenberger

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/signal"
	"strings"
	"sync"
	"syscall"
	"time"

	pipelinev1alpha1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1alpha1"
	"github.com/tektoncd/pipeline/pkg/client/clientset/versioned"
	extinf "github.com/tektoncd/pipeline/pkg/client/informers/externalversions"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/cache"
)

type PipelineRun = pipelinev1alpha1.PipelineRun

func init() {
	log.SetFlags(log.Flags() | log.Lshortfile)
}

func getNS() (string, error) {
	if ns := os.Getenv("POD_NAMESPACE"); ns != "" {
		return ns, nil
	}
	if data, err := ioutil.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace"); err == nil {
		if ns := strings.TrimSpace(string(data)); len(ns) > 0 {
			return ns, nil
		}
	}

	return "default", nil
}

func makeKeys(host, owner, repo string) []string {
	return []string{
		host + secretDelim + owner + secretDelim + repo,
		host + secretDelim + owner,
		host,
	}
}

func mapGitAPI(s string) string {
	switch s {
	case "ghe", "github":
		return "api/v3"
	case "gitea":
		return "api/v1"
	}
	return ""
}

func mapState(s string) (state, description string) {
	switch s {
	case "Running":
		return "pending", "Job triggered."
	case "PipelineRunCancelled":
		return "error", "Job cancelled."
	case "Failed":
		return "failure", "Job failed."
	case "Succeeded":
		return "success", "Job succeeded."
	}
	return "", ""
}

type Status struct {
	Type string
	Date time.Time
}

type StatusCache struct {
	sync.Mutex
	cache map[string]Status

	comment sync.Mutex
}

func NewStatusCache() *StatusCache {
	return &StatusCache{cache: make(map[string]Status)}
}

func (sc *StatusCache) Set(sha, status string, st, fin time.Time) (ret bool) {
	sc.Lock()
	defer sc.Unlock()
	s, ok := sc.cache[sha]
	if !ok {
		log.Printf("adding status to cache: %s %s", sha, status)
		if fin.After(st) {
			sc.cache[sha] = Status{status, fin}
			return true
		}
		sc.cache[sha] = Status{status, st}
		return true
	}
	if fin.After(s.Date) {
		sc.cache[sha] = Status{status, fin}
		ret = true
	}
	if !ret && st.After(s.Date) {
		sc.cache[sha] = Status{status, st}
		ret = true
	}
	if !ret && fin.Equal(s.Date) && s.Type != status {
		sc.cache[sha] = Status{status, fin}
		ret = true
	}
	if !ret && st.Equal(s.Date) && s.Type != status {
		sc.cache[sha] = Status{status, st}
		ret = true
	}
	if ret {
		log.Printf("updating status in cache: %s %s", sha, status)
	}
	return ret
}

func (sc *StatusCache) Remove(sha string) {
	log.Printf("removing status from cache: %s", sha)
	sc.Lock()
	delete(sc.cache, sha)
	sc.Unlock()
}

func getGit(pr *PipelineRun) (rev, url string) {
	for _, v := range pr.Spec.Resources {
		rs := v.ResourceSpec
		if rs == nil {
			continue
		}
		if rs.Type != "git" {
			continue
		}
		for _, rp := range rs.Params {
			if rp.Name == "revision" {
				rev = rp.Value
			}
			if rp.Name == "url" {
				url = rp.Value
			}
			if rev != "" && url != "" {
				return
			}
		}
	}
	return
}

func getParam(pr *PipelineRun, name string) string {
	for _, p := range pr.Spec.Params {
		if p.Name == name {
			return p.Value.StringVal
		}
	}
	return ""
}

func statusUpdate(obj interface{}, s *StatusCache, sec typedcorev1.SecretInterface) {
	pr, ok := obj.(*PipelineRun)
	if !ok {
		return
	}
	if len(pr.Status.Conditions) < 1 {
		return
	}
	status := pr.Status.Conditions[0].Reason
	var (
		start time.Time
		fin   time.Time
	)
	st := pr.Status.StartTime
	f := pr.Status.CompletionTime
	if st != nil {
		start = (*st).Time
	}
	if f != nil {
		fin = (*f).Time
	}
	sha, url := getGit(pr)
	if sha == "" || sha == "master" {
		return
	}
	if !s.Set(sha, status, start, fin) {
		return
	}
	secret, err := sec.Get("git-auth", metav1.GetOptions{})
	if err != nil {
		log.Printf("error getting secret for %s:%s: %v", sha, status, err)
		s.Remove(sha)
		return
	}
	err = postStatus(pr, url, sha, status, secret)
	if err != nil {
		log.Printf("error posting status for %s:%s: %v", sha, status, err)
		s.Remove(sha)
		return
	}

	if status != "Succeeded" {
		return
	}
	issue := getParam(pr, "issue")
	if issue == "" || issue == "master" {
		log.Printf("skipped comment on issue %s for %s/%s", issue, pr.Namespace, pr.Name)
		return
	}
	s.comment.Lock()
	defer s.comment.Unlock()
	err = postComment(pr, url, issue, secret)
	if err != nil {
		log.Printf("error posting comment for %s issue %s: %v", url, issue, err)
		return
	}
}

const (
	contentType = "Content-Type"
	jsonType    = "application/json"

	secretDelim = "_"
)

type GitStatus struct {
	Context     string `json:"context"`
	Description string `json:"description"`
	State       string `json:"state"`
	TargetURL   string `json:"target_url"`
}

func getSecret(sec *corev1.Secret, host, owner, repo string) (username, password string, err error) {
	var (
		secret []byte
		ok     bool
	)
	for _, v := range makeKeys(host, owner, repo) {
		secret, ok = sec.Data[v]
		if ok {
			break
		}
	}
	if !ok {
		err = fmt.Errorf("no key found for host %s in secret %s/%s", host, sec.Namespace, sec.Name)
		return
	}
	a := strings.SplitN(string(secret), ":", 2)
	if len(a) < 2 {
		err = fmt.Errorf("missing credentials for host %s in secret %s/%s", host, sec.Namespace, sec.Name)
		return
	}
	username = a[0]
	password = a[1]
	return
}

func getOwnerRepo(uri string) (owner, repo string) {
	a := strings.SplitN(uri, "/", 2)
	if len(a) < 2 {
		return
	}
	owner = a[0]
	repo = a[1]
	return
}

func postStatus(pr *PipelineRun, gitUrl, sha, status string, sec *corev1.Secret) error {
	u, err := url.Parse(gitUrl)
	if err != nil {
		return err
	}
	uri := strings.TrimPrefix(strings.TrimSuffix(u.RequestURI(), ".git"), "/")
	t := getParam(pr, "gitapi")
	api := mapGitAPI(t)
	if api == "" {
		return fmt.Errorf("error parsing gittype: %s", t)
	}
	u.Path = strings.Join([]string{api, "repos",
		uri, "statuses", sha}, "/")
	tektonurl := getParam(pr, "tektonurl")
	if tektonurl == "" {
		log.Printf("missing tektonurl parameter for %s/%s", pr.Namespace, pr.Name)
	}
	pipeline := "tekton"
	if pr.Spec.PipelineRef != nil {
		pipeline += "-" + pr.Spec.PipelineRef.Name
	}
	state, desc := mapState(status)
	s := &GitStatus{
		Context:     pipeline,
		Description: desc,
		State:       state,
		TargetURL:   tektonurl,
	}
	buf := new(bytes.Buffer)
	err = json.NewEncoder(buf).Encode(s)
	if err != nil {
		return err
	}
	req, err := http.NewRequest("POST", u.String(), buf)
	if err != nil {
		return err
	}
	req.Header.Set(contentType, jsonType)
	owner, repo := getOwnerRepo(uri)
	user, pass, err := getSecret(sec, u.Host, owner, repo)
	if err != nil {
		return err
	}
	req.SetBasicAuth(user, pass)
	c := &http.Client{Transport: &http.Transport{
		DialContext: (&net.Dialer{
			Timeout: 30 * time.Second,
		}).DialContext,
	}}
	_, err = c.Do(req)
	return err
}

type Comment struct {
	Body string `json:"body"`
}

func parseLink(link string) (next string, last string) {
	if len(link) < 2 {
		return
	}
	a := strings.Split(link, ", ")
	for _, v := range a {
		b := strings.Split(v, "; ")
		if len(b) == 2 {
			url := b[0][1 : len(b[0])-1]
			switch b[1] {
			case `rel="next"`:
				next = url
			case `rel="last"`:
				last = url
			}
		}
	}
	return
}

func readComments(c *http.Client, url, user, pass, preview string) (found bool, next, last string, err error) {
	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		return
	}
	req.Header.Set(contentType, jsonType)
	req.SetBasicAuth(user, pass)
	resp, err := c.Do(req)
	if err != nil {
		return
	}
	defer resp.Body.Close()
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return
	}
	if resp.StatusCode != http.StatusOK {
		err = fmt.Errorf("error reading comments: %s", url)
		return
	}
	next, last = parseLink(resp.Header.Get("Link"))
	found = bytes.Contains(body, []byte(preview))
	return
}

func postComment(pr *PipelineRun, gitUrl, issue string, sec *corev1.Secret) error {
	u, err := url.Parse(gitUrl)
	if err != nil {
		return err
	}
	uri := strings.TrimPrefix(strings.TrimSuffix(u.RequestURI(), ".git"), "/")
	t := getParam(pr, "gitapi")
	api := mapGitAPI(t)
	if api == "" {
		return fmt.Errorf("error parsing gittype: %s", t)
	}
	u.Path = strings.Join([]string{api, "repos",
		uri, "issues", issue, "comments"}, "/")
	previewurl := getParam(pr, "previewurl")
	if previewurl == "" {
		return fmt.Errorf("missing previewurl parameter for %s/%s", pr.Namespace, pr.Name)
	}
	scheme := getParam(pr, "scheme")
	preview := fmt.Sprintf("Preview Environment: [Link](%s%s)", scheme, previewurl)
	owner, repo := getOwnerRepo(uri)
	user, pass, err := getSecret(sec, u.Host, owner, repo)
	if err != nil {
		return err
	}
	c := &http.Client{Transport: &http.Transport{
		DialContext: (&net.Dialer{
			Timeout: 30 * time.Second,
		}).DialContext,
	}}
	commentsUrl := u.String()
	for {
		found, next, last, err := readComments(c, commentsUrl, user, pass, preview)
		if err != nil {
			return err
		}
		if found {
			return nil
		}
		if next == "" {
			break
		}
		if commentsUrl == last {
			break
		}
		commentsUrl = next
	}

	comment := &Comment{preview}
	buf := new(bytes.Buffer)
	err = json.NewEncoder(buf).Encode(comment)
	if err != nil {
		return err
	}
	req, err := http.NewRequest("POST", u.String(), buf)
	if err != nil {
		return err
	}
	req.Header.Set(contentType, jsonType)
	req.SetBasicAuth(user, pass)
	_, err = c.Do(req)
	if err == nil {
		log.Printf("set git comment: %s %s", u.String(), previewurl)
	}
	return err
}

func main() {
	cfg, err := rest.InClusterConfig()
	if err != nil {
		log.Fatalf("error configuring kube client: %s\n", err)
	}

	cl, err := versioned.NewForConfig(cfg)
	if err != nil {
		log.Fatalf("error creating new versioned kube client: %s\n", err)
	}
	kcl, err := kubernetes.NewForConfig(cfg)
	if err != nil {
		log.Fatalf("error creating new kube client: %s\n", err)
	}
	ns, err := getNS()
	if err != nil {
		log.Fatalf("error getting current namespace %s\n", err)
	}
	sec := kcl.CoreV1().Secrets(ns)

	extFactory := extinf.NewSharedInformerFactory(cl, time.Second*30)
	prInformer := extFactory.Tekton().V1alpha1().PipelineRuns().Informer()

	s := NewStatusCache()

	prInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			statusUpdate(obj, s, sec)
		},
		UpdateFunc: func(_, obj interface{}) {
			statusUpdate(obj, s, sec)
		},
		DeleteFunc: func(obj interface{}) {
			pr, ok := obj.(*PipelineRun)
			if !ok {
				return
			}
			sha, _ := getGit(pr)
			if sha != "" && sha != "master" {
				s.Remove(sha)
			}
		},
	})

	stop := make(chan struct{})
	extFactory.Start(stop)
	log.Println("started pipelinerun watcher")

	sigs := make(chan os.Signal)
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
	log.Println("started signal handler")
	sig := <-sigs
	close(stop)
	log.Printf("signal: %s\n", sig.String())
}