vegvisir/pkg/server/server.go

//           ___   ____               ___       ___
// \ \  / / | _   |  __| \ \  / / || | __  || || _ |
//  \ \/ /  |___  | |__   \ \/ /  || |___  || ||___|
//   \  /   | _   | _  |   \  /   ||  __ | || ||\\
//    \/    |___  |___ |    \/    || ____| || || \\
//
// Copyright (c) 2021 Piotr Biernat. https://pbiernat.dev. MIT License
// Repo: https://git.pbiernat.dev/golang/vegvisir

package server

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"regexp"
	"strings"
	"time"
	"vegvisir/pkg/cache"
	"vegvisir/pkg/config"

	"github.com/valyala/fasthttp"
)

const (
	Version = "0.1"
	Name    = "Vegvisir/" + Version
)

type Server struct {
	Config config.Config

	cFilePath string                       // Path to config file
	rCache    map[string]*cache.RouteCache // Internal route cache
	routeCM   cache.Manager                // Redis route cache
	respCM    cache.Manager                // Redis response cache
}

func NewServer(cPath string) *Server {
	datastore := cache.NewRedisDatastore("127.0.0.7", 6379) // FIXME use config or env...

	return &Server{
		cFilePath: cPath,
		rCache:    make(map[string]*cache.RouteCache),
		routeCM:   cache.NewManager(datastore, "prefix_", 5),   //FIXME use ttl(seconds) from config or env...
		respCM:    cache.NewManager(datastore, "response_", 5), //FIXME use ttl(seconds) from config or env...
	}
}

func (s *Server) Run() {
	if err := s.Config.Load(s.cFilePath); err != nil {
		log.Fatalln("Unable to find config file: ", s.cFilePath, err)
	}

	go func() {
		serverAddress := s.Config.Server.Address + ":" + fmt.Sprint(s.Config.Server.Port)
		if err := fasthttp.ListenAndServe(serverAddress, s.mainHandler); err != nil {
			log.Fatalf("Server panic! Error message: %s", err)
		}
	}()

	log.Println("Server started")

	// Wait for an interrupt
	interrupt := make(chan os.Signal, 1)
	signal.Notify(interrupt, os.Interrupt, os.Kill)
	<-interrupt

	log.Println("SIGKILL or SIGINT caught, shutting down...")

	// Attempt a graceful shutdown
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	s.Shutdown(ctx)
	log.Println("Server shutdown successfully.")
}

func (s *Server) Shutdown(ctx context.Context) { // TODO: wait for all connections to finish
	log.Println("Shuting down finished")
}

func (s *Server) mainHandler(ctx *fasthttp.RequestCtx) {
	ctx.Response.Header.Add(fasthttp.HeaderServer, Name)

	// move all below logic to concrete handler or sth....
	reqUrl, sReqUrl, sReqMethod := ctx.RequestURI(), string(ctx.RequestURI()), string(ctx.Method())
	// log.Println("Incoming request:", sReqMethod, sReqUrl)

	found, route := s.findRouteByRequestUrl(reqUrl)
	if !found {
		// FIXME: return 404 or 5xx error in response? Maybe define it in concrete Backend config?
		ctx.SetStatusCode(fasthttp.StatusNotFound)
		return
	}

	memo := cache.New(s.processUrl)
	defer memo.Close()
	response, err := memo.Read(sReqUrl, sReqMethod, route)
	//err, response := s.processUrl(sReqUrl, sReqMethod, route)
	if err != nil {
		// FIXME: Response read error(sending 500 error response)
		ctx.SetStatusCode(fasthttp.StatusInternalServerError)
		log.Println("Response read error(sending 500 error response)", err)
		return
	}

	ctx.SetStatusCode(response.Code)
	ctx.SetBody(response.Body)
	ctx.Response.Header.SetBytesV(fasthttp.HeaderContentType, response.Headers.ContentType())
}

func (s *Server) findRouteByRequestUrl(url []byte) (bool, *cache.RouteCache) {
	var sUrl = string(url)

	for bId := range s.Config.Backends {
		bck := s.Config.Backends[bId]
		if !strings.Contains(sUrl, bck.PrefixUrl) {
			continue
		}

		for rId := range bck.Routes {
			route := &bck.Routes[rId]

			if cRoute, ok := s.rCache[sUrl]; ok {
				return true, cRoute
			}

			rgxp := regexp.MustCompile(fmt.Sprintf("%s%s", bck.PrefixUrl, route.Pattern))
			if rgxp.Match(url) {
				targetUrl := bck.BackendAddress + rgxp.ReplaceAllString(sUrl, route.Target)

				cRoute := cache.NewRouteCache(sUrl, targetUrl)

				// s.rCacheManager.Save(sUrl, cRoute)
				s.rCache[sUrl] = cRoute

				return true, cRoute
			}
		}
	}

	return false, &cache.RouteCache{}
}

func (s *Server) processUrl(url, method string, route *cache.RouteCache) (error, *cache.ResponseCache) {
	// handle response caching
	cacheKey := method + "_" + url
	if ok, data := s.respCM.Load(cacheKey, &cache.ResponseCache{}); ok {
		//log.Println("Read resp from cache: ", route.TargetUrl, url)

		return nil, data.(*cache.ResponseCache)
	} else {
		//log.Println("Send req to backend url: ", route.TargetUrl, url)
		start := time.Now()
		bckReq := fasthttp.AcquireRequest()
		bckResp := fasthttp.AcquireResponse()
		defer fasthttp.ReleaseRequest(bckReq)
		defer fasthttp.ReleaseResponse(bckResp)

		// copy headers from backend response and prepare request for backend - separate
		bckReq.SetRequestURI(route.TargetUrl)
		bckReq.Header.SetMethod(method)

		err := fasthttp.Do(bckReq, bckResp)
		if err != nil {
			return err, nil
		}

		body, code := bckResp.Body(), bckResp.StatusCode()
		log.Printf("%s, %s, %d bytes\n", url, time.Since(start), len(body))
		// save response to cache
		respCache := cache.NewResponseCache(url, method, body, code, &bckResp.Header)
		s.respCM.Save(cacheKey, respCache)

		return nil, respCache
	}
}