promgate/web.go

package main

import (
	"bufio"
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"log"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"
)

type Web struct {
	config *Config
	server *http.Server
}

// NewWeb creates a new Web
func NewWeb(config *Config) *Web {
	mux := http.NewServeMux()
	w := &Web{
		config: config,
		server: &http.Server{
			Addr:    config.Listen,
			Handler: mux,
		},
	}
	if config.DisableTLS {
		log.Printf("running without tls")
		log.Printf("everyone can access this server via: http://%s", config.Listen)
		mux.HandleFunc("/metrics", w.metrics)
	} else {
		w.server.TLSConfig = &tls.Config{
			ClientAuth: tls.RequireAndVerifyClientCert,
			ClientCAs:  config.ClientCA.CertPool,
		}
		mux.HandleFunc("/metrics", w.withTlsClientCheck(w.metrics))
	}
	return w
}

// ListenAndServer starts the webserver according to config
func (w *Web) ListenAndServe() error {
	if w.config.DisableTLS {
		return w.server.ListenAndServe()
	}
	return w.server.ListenAndServeTLS(w.config.Cert, w.config.Key)
}

// withTlsClientCheck is a middleware that validates the client cert against the ca and crl
func (web *Web) withTlsClientCheck(next http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		for _, peer := range r.TLS.PeerCertificates {
			for _, revoked := range web.config.CRL.TBSCertList.RevokedCertificates {
				if peer.SerialNumber.Cmp(revoked.SerialNumber) == 0 {
					log.Printf("Revoked certificate: %s from %s", peer.Subject, r.RemoteAddr)
					w.WriteHeader(403)
					return
				}
			}

		}
		next.ServeHTTP(w, r)
	}
}

// metrics is a http HandleFunc that gathers metrics from URLs in parallel
func (web *Web) metrics(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	c := make(chan string)
	wg := sync.WaitGroup{}
	for _, urlParsed := range web.config.URLs.targets {
		wg.Add(1)
		go fetch(urlParsed, c, ctx, &wg)
	}

	// wait until all requests are finished or aborted
	go func() {
		wg.Wait()
		close(c)
	}()

	w.Header().Add("content-type", "text/plain")
	w.WriteHeader(200)
	for res := range c {
		fmt.Fprintf(w, "%s\n", res)
	}
}

// extend finds parses prometheus metric lines and injects instance and path labels
func extend(line, instance, path string) (string, error) {
	if strings.HasPrefix(line, "#") {
		return "", nil
	}
	match := re.FindStringSubmatch(line)
	if len(match) != 4 {
		return line, errors.New("Invalid Line.")
	}

	lineName := match[1]
	lineLabels := match[2]
	lineValue := match[3]
	if lineLabels == "" {
		lineLabels = fmt.Sprintf("sub_instance=%q,sub_path=%q", instance, path)
	} else {
		lineLabels = fmt.Sprintf("%s,sub_instance=%q,sub_path=%q", lineLabels, instance, path)
	}
	line = fmt.Sprintf("%s{%s} %s", lineName, lineLabels, lineValue)
	return line, nil
}

// fetch scans the result of a url linewise into c
func fetch(u *url.URL, c chan string, ctx context.Context, wg *sync.WaitGroup) {
	up := 0
	err := error(nil)

	defer wg.Done()
	defer func() {
		c <- fmt.Sprintf("up {sub_instance=%q, sub_path=%q} %d", u.Host, u.Path, up)
		if err != nil {
			log.Printf("Error: %s", err)
		}
	}()

	req, err := http.NewRequest("GET", u.String(), nil)
	if err != nil {
		return
	}
	req = req.WithContext(ctx)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return
	}
	scanner := bufio.NewScanner(resp.Body)
	for scanner.Scan() {
		line := scanner.Text()
		parsed, err := extend(line, u.Host, u.Path)
		if err != nil {
			log.Printf("unable to parse line %q: %s", line, err)
		}
		if parsed != "" {
			c <- parsed
		}
	}
	if err := scanner.Err(); err != nil {
		return
	}
	up = 1
}