converge/cmd/converge/converge.go

package main

import (
	"context"
	"fmt"
	"git.wamblee.org/converge/pkg/models"
	"git.wamblee.org/converge/pkg/server/matchmaker"
	"git.wamblee.org/converge/pkg/support/websocketutil"
	"log"
	"net"
	"net/http"
	_ "net/http/pprof"
	"os"
	"regexp"
	"strings"
	_ "time/tzdata"
)

func parsePublicId(path string) (publicId models.RendezVousId, _ error) {
	pattern := regexp.MustCompile("/([^/]+)$")
	matches := pattern.FindStringSubmatch(path)
	if len(matches) != 2 {
		return "", fmt.Errorf("Invalid URL path '%s'", path)
	}
	return models.RendezVousId(matches[1]), nil
}

func catchAllHandler(contextPath string) func(w http.ResponseWriter, r *http.Request) {
	return func(w http.ResponseWriter, r *http.Request) {
		http.Redirect(w, r, contextPath+"docs", http.StatusFound)
	}
}

func printHelp(msg string) {
	if msg != "" {
		fmt.Fprintf(os.Stderr, "ERROR: %s\n\n", msg)
	}
	helpText := "Usage: converge [options]\n" +
		"\n" +
		"Converge server is a rendez-vous server for debugging continuous integration\n" +
		"jobs be providing the capability to log into the agents where jobs are running.\n" +
		"This is achieve by starting an agent in the continuous integration job\n" +
		"which connects to Converge using a websocket connection. The end user also connects\n" +
		"to Converge using ssh over websockets. The server then matches the end-user with\n" +
		"the agent running in the continous integration job (the rendez-vous) and sets up\n" +
		"an end-to-end SSH connection between end-user and agent, with the agent providing\n" +
		"an embedded SSH server to provide interactive access to the end-user. This works\n" +
		"both on linux and on windows.\n" +
		"\n" +
		"Options\n" +
		"--static <contentdir>:     directory where static content of converge is placed\n" +
		"--downloads <downloaddir>: directory where downloads of converge are placed\n" +
		"--ccntext <contextpath>:   by default all content is served at /. Use this option to specify\n" +
		"                           a different context path. For instance to host converge at a base\n" +
		"                           URL of https://example.com/converge/, specify /converge/ (with\n" +
		"                           trailing slash. \n" +
		"--pprof:                   Enable the pprof endpoint at /debug/pprof"
	fmt.Fprintln(os.Stderr, helpText)
	os.Exit(1)
}

func main() {
	// for debugging hangs relating to using unbuffered channels
	//go func() {
	//	buf := make([]byte, 1000000)
	//	n := runtime.Stack(buf, true)
	//	time.Sleep(10 * time.Second)
	//	log.Println("LOG: \n" + string(buf[:n]))
	//}()

	downloaddir := "."
	staticdir := "../static"
	contextpath := "/"
	pprof := false

	args := os.Args[1:]
	for len(args) > 0 && strings.HasPrefix(args[0], "-") {
		switch args[0] {
		case "--downloads":
			if len(args) <= 1 {
				printHelp("The --downloads option expects an argument")
			}
			downloaddir = args[1]
			args = args[1:]
		case "--static":
			if len(args) <= 1 {
				printHelp("The --static option expects an argument")
			}
			staticdir = args[1]
			args = args[1:]
		case "--context":
			if len(args) <= 1 {
				printHelp("The --context option expects an argument")
			}
			contextpath = args[1]
			args = args[1:]
		case "--pprof":
			pprof = true
		default:
			printHelp("Unknown option " + args[0])
		}
		args = args[1:]
	}
	log.Println("Content directory", staticdir)

	if len(args) != 0 {
		printHelp("")
	}

	// Initializing the ocre system.
	// Setup of websession handling for pushing notifications to the client
	// Prometheus
	// And the MatchMaker. The MatchMakers sends state notifications to websessions
	// and prometheus.
	notifications := NewStateNotifier()
	websessions := matchmaker.NewWebSessions(notifications.webNotificationChannel)
	// monitoring
	prometheusMux := http.NewServeMux()
	setupPrometheus(prometheusMux, notifications.prometheusNotificationChannel)
	go func() {
		log.Fatal(http.ListenAndServe(":8001", prometheusMux))
	}()
	admin := matchmaker.NewMatchMaker(notifications)

	registrationService, clientService, sessionService := setupWebSockets(admin, websessions)

	context := HttpContext{mux: http.NewServeMux(), path: contextpath}

	if pprof {
		registerPprof(context.mux, "/debug/pprof")
	}
	setupWebUI(context, registrationService, clientService, sessionService, staticdir, downloaddir)
	// Catch all for the web UI
	context.mux.HandleFunc("/", catchAllHandler(contextpath))

	// Start HTTP server
	fmt.Println("Rendez-vous server listening on :8000")
	log.Fatal(http.ListenAndServe(":8000", context.mux))
}

func setupWebUI(context HttpContext, registrationService websocketutil.WebSocketService, clientService websocketutil.WebSocketService, sessionService websocketutil.WebSocketService, staticdir string, downloaddir string) HttpContext {
	// The web UI

	context.HandleFunc("agent/", registrationService.Handle)
	context.HandleFunc("client/", clientService.Handle)
	context.HandleFunc("ws/sessions", sessionService.Handle)

	// create filehandler with templating for html files.
	context.Handle("docs/", http.StripPrefix("docs/", http.HandlerFunc(pageHandler)))
	context.Handle("static/", http.StripPrefix("static/",
		http.FileServer(http.Dir(staticdir))))
	context.Handle("downloads/", http.StripPrefix("downloads/",
		http.FileServer(http.Dir(downloaddir))))
	// create usage generator
	context.HandleFunc("usage", generateCLIExammple)

	return context
}

func setupWebSockets(admin *matchmaker.MatchMaker, websessions *matchmaker.WebSessions) (websocketutil.WebSocketService, websocketutil.WebSocketService, websocketutil.WebSocketService) {
	// websocket endpoints

	// For agents connecting
	registrationService := websocketutil.WebSocketService{
		Handler: func(w http.ResponseWriter, r *http.Request, conn net.Conn) {
			publicId, err := parsePublicId(r.URL.Path)
			if err != nil {
				log.Printf("Cannot parse public id from url: '%v'\n", err)
				return
			}
			log.Printf("Got registration connection: '%s'\n", publicId)
			err = admin.Register(publicId, conn)
			if err != nil {
				log.Printf("Error %v\n", err)
			}
		},
	}

	// For users connecting with ssh
	clientService := websocketutil.WebSocketService{
		Handler: func(w http.ResponseWriter, r *http.Request, conn net.Conn) {
			publicId, err := parsePublicId(r.URL.Path)
			if err != nil {
				log.Printf("Cannot parse public id from url: '%v'\n", err)
				return
			}
			_, wsProxyMode := r.URL.Query()["wsproxy"]
			log.Printf("Got client connection: '%s'\n", publicId)
			err = admin.Connect(wsProxyMode, publicId, conn)
			if err != nil {
				log.Printf("Error %v\n", err)
			}
		},
	}

	// for the web browser getting live status updates.
	sessionService := websocketutil.WebSocketService{
		Handler: func(w http.ResponseWriter, r *http.Request, conn net.Conn) {
			ctx, cancel := context.WithCancel(context.Background())
			websession := websessions.NewSession(conn, ctx, cancel)
			defer websessions.SessionClosed(websession)
			location, err := matchmaker.GetUserLocation(r)
			if err != nil {
				panic(err)
			}
			websession.WriteNotifications(location, ctx, cancel)
		},
		Text: true,
	}
	return registrationService, clientService, sessionService
}