first commit
This commit is contained in:
commit
be3c522f14
11 changed files with 635 additions and 0 deletions
422
main.go
Normal file
422
main.go
Normal file
|
|
@ -0,0 +1,422 @@
|
|||
// 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"
|
||||
typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
|
||||
//"k8s.io/apimachinery/pkg/types"
|
||||
//"k8s.io/client-go/informers"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
"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 mapGitAPI(s string) string {
|
||||
switch s {
|
||||
case "ghe", "github":
|
||||
return "api/v3"
|
||||
case "gitea":
|
||||
return "api/v1"
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func mapState(s string) string {
|
||||
switch s {
|
||||
case "Running":
|
||||
return "pending"
|
||||
case "PipelineRunCancelled":
|
||||
return "error"
|
||||
case "Failed":
|
||||
return "failure"
|
||||
case "Succeeded":
|
||||
return "success"
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
type Status struct {
|
||||
Type string
|
||||
Date time.Time
|
||||
}
|
||||
|
||||
type StatusCache struct {
|
||||
sync.Mutex
|
||||
cache map[string]Status
|
||||
}
|
||||
|
||||
func NewStatusCache() *StatusCache {
|
||||
return &StatusCache{cache: make(map[string]Status)}
|
||||
}
|
||||
|
||||
func (sc *StatusCache) Set(sha, status string, st, fin time.Time) 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) {
|
||||
log.Printf("updating status in cache: %s %s", sha, status)
|
||||
sc.cache[sha] = Status{status, fin}
|
||||
return true
|
||||
}
|
||||
if st.After(s.Date) {
|
||||
log.Printf("updating status in cache: %s %s", sha, status)
|
||||
sc.cache[sha] = Status{status, st}
|
||||
return true
|
||||
}
|
||||
if fin.Equal(s.Date) && s.Type != status {
|
||||
log.Printf("updating status in cache: %s %s", sha, status)
|
||||
sc.cache[sha] = Status{status, fin}
|
||||
return true
|
||||
}
|
||||
if st.Equal(s.Date) && s.Type != status {
|
||||
log.Printf("updating status in cache: %s %s", sha, status)
|
||||
sc.cache[sha] = Status{status, st}
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
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 == "" {
|
||||
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
|
||||
}
|
||||
err = postComment(pr, url, issue, secret)
|
||||
if err != nil {
|
||||
log.Printf("error posting comment for %s:%s: %v", sha, status, err)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
const (
|
||||
contentType = "Content-Type"
|
||||
jsonType = "application/json"
|
||||
)
|
||||
|
||||
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 string) (username, password string, err error) {
|
||||
s, ok := sec.Data[host]
|
||||
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(s), ":", 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 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)
|
||||
}
|
||||
s := &GitStatus{
|
||||
Context: "tekton",
|
||||
Description: pr.Name,
|
||||
State: mapState(status),
|
||||
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)
|
||||
user, pass, err := getSecret(sec, u.Host)
|
||||
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 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)
|
||||
user, pass, err := getSecret(sec, u.Host)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req, err := http.NewRequest("GET", u.String(), nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set(contentType, jsonType)
|
||||
req.SetBasicAuth(user, pass)
|
||||
c := &http.Client{Transport: &http.Transport{
|
||||
DialContext: (&net.Dialer{
|
||||
Timeout: 30 * time.Second,
|
||||
}).DialContext,
|
||||
}}
|
||||
resp, err := c.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
body, err := ioutil.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if bytes.Contains(body, []byte(preview)) {
|
||||
return nil
|
||||
}
|
||||
|
||||
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 != "" {
|
||||
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())
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue