arcad/emissary
16
0
Fork 0
Code Issues 14 Pull Requests Packages Projects Releases 3 Wiki Activity

feat: host-based routing in proxy

Browse Source
This commit is contained in:
wpetit
2023-03-22 18:15:22 +01:00
parent 8ada7e3b84
commit cd796fff89
10 changed files with 165 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
internal/agent/controller/proxy/controller.go
View File

@ -2,16 +2,23 @@ package proxy
import (
"context"
"net/url"
"forge.cadoles.com/Cadoles/emissary/internal/agent"
"forge.cadoles.com/Cadoles/emissary/internal/spec/proxy"
edgeProxy "forge.cadoles.com/arcad/edge/pkg/proxy"
"github.com/mitchellh/hashstructure/v2"
"github.com/pkg/errors"
"gitlab.com/wpetit/goweb/logger"
)
type proxyEntry struct {
SpecHash uint64
Proxy *ReverseProxy
}
type Controller struct {
proxies map[proxy.ID]*ReverseProxy
currentSpecRevision int
proxies map[proxy.ID]*proxyEntry
}
// Name implements node.Controller.
@ -25,7 +32,7 @@ func (c *Controller) Reconcile(ctx context.Context, state *agent.State) error {
if err := state.GetSpec(proxy.NameProxy, proxySpec); err != nil {
if errors.Is(err, agent.ErrSpecNotFound) {
logger.Info(ctx, "could not find proxy spec, stopping all remaining proxies")
logger.Info(ctx, "could not find proxy spec")
c.stopAllProxies(ctx)
@ -37,25 +44,20 @@ func (c *Controller) Reconcile(ctx context.Context, state *agent.State) error {
logger.Info(ctx, "retrieved spec", logger.F("spec", proxySpec.SpecName()), logger.F("revision", proxySpec.SpecRevision()))
if c.currentSpecRevision == proxySpec.SpecRevision() {
logger.Info(ctx, "spec revision did not change, doing nothing")
return nil
}
c.updateProxies(ctx, proxySpec)
c.currentSpecRevision = proxySpec.SpecRevision()
logger.Info(ctx, "updating current spec revision", logger.F("revision", c.currentSpecRevision))
return nil
}
func (c *Controller) stopAllProxies(ctx context.Context) {
for proxyID, proxy := range c.proxies {
if len(c.proxies) > 0 {
logger.Info(ctx, "stopping all proxies")
}
for proxyID, entry := range c.proxies {
logger.Info(ctx, "stopping proxy", logger.F("proxyID", proxyID))
if err := proxy.Stop(); err != nil {
if err := entry.Proxy.Stop(); err != nil {
logger.Error(
ctx, "error while stopping proxy",
logger.F("proxyID", proxyID),
@ -69,14 +71,14 @@ func (c *Controller) stopAllProxies(ctx context.Context) {
func (c *Controller) updateProxies(ctx context.Context, spec *proxy.Spec) {
// Stop and remove obsolete proxys
for proxyID, proxy := range c.proxies {
for proxyID, entry := range c.proxies {
if _, exists := spec.Proxies[proxyID]; exists {
continue
}
logger.Info(ctx, "stopping proxy", logger.F("proxyID", proxyID))
if err := proxy.Stop(); err != nil {
if err := entry.Proxy.Stop(); err != nil {
logger.Error(
ctx, "error while stopping proxy",
logger.F("proxyID", proxyID),
@ -89,35 +91,89 @@ func (c *Controller) updateProxies(ctx context.Context, spec *proxy.Spec) {
// (Re)start proxys
for proxyID, proxySpec := range spec.Proxies {
proxy, exists := c.proxies[proxyID]
if !exists {
proxy = NewReverseProxy()
c.proxies[proxyID] = proxy
}
proxyCtx := logger.With(ctx, logger.F("proxyID", proxyID))
logger.Info(
ctx, "starting proxy",
logger.F("proxyID", proxyID),
logger.F("addr", proxySpec.Address),
logger.F("target", proxySpec.Target),
)
if err := proxy.Start(ctx, proxySpec.Address, proxySpec.Target); err != nil {
logger.Error(
ctx, "error while starting proxy",
logger.F("proxyID", proxyID),
logger.E(errors.WithStack(err)),
)
delete(c.proxies, proxyID)
if err := c.updateProxy(ctx, proxyID, proxySpec); err != nil {
logger.Error(proxyCtx, "could not update proxy", logger.E(errors.WithStack(err)))
continue
}
}
}
func (c *Controller) updateProxy(ctx context.Context, proxyID proxy.ID, proxySpec proxy.ProxyEntry) (err error) {
newProxySpecHash, err := hashstructure.Hash(proxySpec, hashstructure.FormatV2, nil)
if err != nil {
return errors.WithStack(err)
}
var entry *proxyEntry
entry, exists := c.proxies[proxyID]
if !exists {
logger.Info(ctx, "proxy currently not running")
}
if entry == nil {
entry = &proxyEntry{
Proxy: NewReverseProxy(),
SpecHash: 0,
}
c.proxies[proxyID] = entry
}
specChanged := newProxySpecHash != entry.SpecHash
if entry.Proxy.Running() && !specChanged {
return nil
}
if specChanged && entry.SpecHash != 0 {
logger.Info(
ctx, "restarting proxy",
logger.F("address", proxySpec.Address),
)
} else {
logger.Info(
ctx, "starting proxy",
logger.F("address", proxySpec.Address),
)
}
options := make([]edgeProxy.OptionFunc, 0)
allowedHosts := make([]string, len(proxySpec.Mappings))
mappings := make(map[string]*url.URL, len(proxySpec.Mappings))
for _, m := range proxySpec.Mappings {
target, err := url.Parse(m.Target)
if err != nil {
return errors.WithStack(err)
}
mappings[m.HostPattern] = target
allowedHosts = append(allowedHosts, m.HostPattern)
}
options = append(
options,
edgeProxy.WithAllowedHosts(allowedHosts...),
edgeProxy.WithRewriteHosts(mappings),
)
if err := entry.Proxy.Start(ctx, proxySpec.Address, options...); err != nil {
delete(c.proxies, proxyID)
return errors.Wrap(err, "could not start app")
}
entry.SpecHash = newProxySpecHash
return nil
}
func NewController() *Controller {
return &Controller{
proxies: make(map[proxy.ID]*ReverseProxy),
currentSpecRevision: -1,
proxies: make(map[proxy.ID]*proxyEntry),
}
}

46
internal/agent/controller/proxy/reverse_proxy.go
View File

@ -3,26 +3,21 @@ package proxy
import (
"context"
"net/http"
"net/http/httputil"
"net/url"
"sync"
"github.com/pkg/errors"
"gitlab.com/wpetit/goweb/logger"
"forge.cadoles.com/arcad/edge/pkg/proxy"
)
type ReverseProxy struct {
addr string
target string
server *http.Server
mutex sync.RWMutex
}
func (p *ReverseProxy) Start(ctx context.Context, addr, target string) error {
alreadyRunning := p.server != nil && target == p.target && addr == p.target
if alreadyRunning {
return nil
}
func (p *ReverseProxy) Start(ctx context.Context, addr string, funcs ...proxy.OptionFunc) error {
if p.server != nil {
if err := p.Stop(); err != nil {
return errors.WithStack(err)
@ -33,33 +28,40 @@ func (p *ReverseProxy) Start(ctx context.Context, addr, target string) error {
Addr: addr,
}
url, err := url.Parse(target)
if err != nil {
return errors.WithStack(err)
}
proxy := httputil.NewSingleHostReverseProxy(url)
proxy := proxy.New(funcs...)
server.Handler = proxy
p.mutex.Lock()
p.server = server
p.addr = addr
p.target = target
p.mutex.Unlock()
go func() {
defer func() {
if err := p.Stop(); err != nil {
logger.Error(ctx, "error while stopping gateway", logger.E(errors.WithStack(err)))
}
}()
if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
logger.Error(ctx, "error while listening", logger.E(errors.WithStack(err)))
}
if err := p.Stop(); err != nil {
logger.Error(ctx, "error while stopping gateway", logger.E(errors.WithStack(err)))
}
}()
return nil
}
func (p *ReverseProxy) Running() bool {
p.mutex.RLock()
defer p.mutex.RUnlock()
return p.server != nil
}
func (p *ReverseProxy) Stop() error {
p.mutex.Lock()
defer p.mutex.Unlock()
if p.server == nil {
return nil
}