TyleoDvDelaware3217/GoDoxy
1
0
Fork 0
mirror of https://github.com/yusing/godoxy.git synced 2025-05-31 00:52:35 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

refactor, merge Entry, RawEntry and Route into one. Implement fileserver. Still buggy

Browse source
This commit is contained in:
yusing 2025-02-03 08:55:07 +08:00
parent 58d5e81449
commit c61d893403
54 changed files with 996 additions and 1096 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
.golangci.yml
View file

@ -9,9 +9,6 @@ linters-settings:
- fieldalignment
gocyclo:
min-complexity: 14
goconst:
min-len: 3
min-occurrences: 4
misspell:
locale: US
funlen:
@ -102,13 +99,14 @@ linters:
- depguard # Not relevant
- nakedret # Too strict
- lll # Not relevant
- gocyclo # FIXME must be fixed
- gocyclo # must be fixed
- gocognit # Too strict
- nestif # Too many false-positive.
- prealloc # Too many false-positive.
- makezero # Not relevant
- dupl # Too strict
- gci # I don't care
- goconst # Too annoying
- gosec # Too strict
- gochecknoinits
- gochecknoglobals

10
.trunk/trunk.yaml
View file

@ -2,12 +2,12 @@
# To learn more about the format of this file, see https://docs.trunk.io/reference/trunk-yaml
version: 0.1
cli:
version: 1.22.8
version: 1.22.9
# Trunk provides extensibility via plugins. (https://docs.trunk.io/plugins)
plugins:
sources:
- id: trunk
ref: v1.6.6
ref: v1.6.7
uri: https://github.com/trunk-io/plugins
# Many linters and tools depend on runtimes - configure them here. (https://docs.trunk.io/runtimes)
runtimes:
@ -22,8 +22,8 @@ lint:
- yamllint
enabled:
- hadolint@2.12.1-beta
- actionlint@1.7.6
- checkov@3.2.352
- actionlint@1.7.7
- checkov@3.2.360
- git-diff-check
- gofmt@1.20.4
- golangci-lint@1.63.4
@ -32,7 +32,7 @@ lint:
- prettier@3.4.2
- shellcheck@0.10.0
- shfmt@3.6.0
- trufflehog@3.88.2
- trufflehog@3.88.4
actions:
disabled:
- trunk-announce

2
cmd/main.go
View file

@ -120,7 +120,7 @@ func main() {
printJSON(cfg.Value())
return
case common.CommandDebugListEntries:
printJSON(cfg.DumpEntries())
printJSON(cfg.DumpRoutes())
return
case common.CommandDebugListProviders:
printJSON(cfg.DumpRouteProviders())

2
internal/api/v1/favicon/cache.go
View file

@ -78,7 +78,7 @@ func pruneExpiredIconCache() {
}
func routeKey(r route.HTTPRoute) string {
return r.RawEntry().Provider + ":" + r.TargetName()
return r.ProviderName() + ":" + r.TargetName()
}
func PruneRouteIconCache(route route.HTTPRoute) {

4
internal/api/v1/favicon/favicon.go
View file

@ -87,7 +87,7 @@ func GetFavIcon(w http.ResponseWriter, req *http.Request) {
}
var result *fetchResult
hp := r.RawEntry().Homepage.GetOverride()
hp := r.HomepageConfig().GetOverride()
if !hp.IsEmpty() && hp.Icon != nil {
if hp.Icon.IconSource == homepage.IconSourceRelative {
result = findIcon(r, req, hp.Icon.Value)
@ -189,7 +189,7 @@ func findIcon(r route.HTTPRoute, req *http.Request, uri string) *fetchResult {
}
result := fetchIcon("png", sanitizeName(r.TargetName()))
cont := r.RawEntry().Container
cont := r.ContainerInfo()
if !result.OK() && cont != nil {
result = fetchIcon("png", sanitizeName(cont.ImageName))
}

9
internal/config/query.go
View file

@ -1,16 +1,15 @@
package config
import (
route "github.com/yusing/go-proxy/internal/route"
"github.com/yusing/go-proxy/internal/route"
"github.com/yusing/go-proxy/internal/route/provider"
"github.com/yusing/go-proxy/internal/route/types"
)
func (cfg *Config) DumpEntries() map[string]*types.RawEntry {
entries := make(map[string]*types.RawEntry)
func (cfg *Config) DumpRoutes() map[string]*route.Route {
entries := make(map[string]*route.Route)
cfg.providers.RangeAll(func(_ string, p *provider.Provider) {
p.RangeRoutes(func(alias string, r *route.Route) {
entries[alias] = r.Entry
entries[alias] = r
})
})
return entries

2
internal/docker/container.go
View file

@ -12,7 +12,7 @@ import (
)
type (
PortMapping = map[string]types.Port
PortMapping = map[int]types.Port
Container struct {
_ U.NoCopy

16
internal/docker/container_helper.go
View file

@ -44,7 +44,7 @@ func (c containerHelper) getPublicPortMapping() PortMapping {
if v.PublicPort == 0 {
continue
}
res[strutils.PortString(v.PublicPort)] = v
res[int(v.PublicPort)] = v
}
return res
}
@ -52,7 +52,7 @@ func (c containerHelper) getPublicPortMapping() PortMapping {
func (c containerHelper) getPrivatePortMapping() PortMapping {
res := make(PortMapping)
for _, v := range c.Ports {
res[strutils.PortString(v.PrivatePort)] = v
res[int(v.PrivatePort)] = v
}
return res
}
@ -66,14 +66,6 @@ var databaseMPs = map[string]struct{}{
"/var/lib/rabbitmq": {},
}
var databasePrivPorts = map[uint16]struct{}{
5432: {}, // postgres
3306: {}, // mysql, mariadb
6379: {}, // redis
11211: {}, // memcached
27017: {}, // mongodb
}
func (c containerHelper) isDatabase() bool {
for _, m := range c.Mounts {
if _, ok := databaseMPs[m.Destination]; ok {
@ -82,7 +74,9 @@ func (c containerHelper) isDatabase() bool {
}
for _, v := range c.Ports {
if _, ok := databasePrivPorts[v.PrivatePort]; ok {
switch v.PrivatePort {
// postgres, mysql or mariadb, redis, memcached, mongodb
case 5432, 3306, 6379, 11211, 27017:
return true
}
}

24
internal/docker/idlewatcher/waker.go
View file

@ -38,32 +38,32 @@ const (
// TODO: support stream
func newWaker(parent task.Parent, entry route.Entry, rp *reverseproxy.ReverseProxy, stream net.Stream) (Waker, E.Error) {
hcCfg := entry.RawEntry().HealthCheck
func newWaker(parent task.Parent, route route.Route, rp *reverseproxy.ReverseProxy, stream net.Stream) (Waker, E.Error) {
hcCfg := route.HealthCheckConfig()
hcCfg.Timeout = idleWakerCheckTimeout
waker := &waker{
rp: rp,
stream: stream,
}
task := parent.Subtask("idlewatcher." + entry.TargetName())
watcher, err := registerWatcher(task, entry, waker)
task := parent.Subtask("idlewatcher." + route.TargetName())
watcher, err := registerWatcher(task, route, waker)
if err != nil {
return nil, E.Errorf("register watcher: %w", err)
}
switch {
case rp != nil:
waker.hc = monitor.NewHTTPHealthChecker(entry.TargetURL(), hcCfg)
waker.hc = monitor.NewHTTPHealthChecker(route.TargetURL(), hcCfg)
case stream != nil:
waker.hc = monitor.NewRawHealthChecker(entry.TargetURL(), hcCfg)
waker.hc = monitor.NewRawHealthChecker(route.TargetURL(), hcCfg)
default:
panic("both nil")
}
if common.PrometheusEnabled {
m := metrics.GetServiceMetrics()
fqn := parent.Name() + "/" + entry.TargetName()
fqn := parent.Name() + "/" + route.TargetName()
waker.metric = m.HealthStatus.With(metrics.HealthMetricLabels(fqn))
waker.metric.Set(float64(watcher.Status()))
}
@ -71,12 +71,12 @@ func newWaker(parent task.Parent, entry route.Entry, rp *reverseproxy.ReversePro
}
// lifetime should follow route provider.
func NewHTTPWaker(parent task.Parent, entry route.Entry, rp *reverseproxy.ReverseProxy) (Waker, E.Error) {
return newWaker(parent, entry, rp, nil)
func NewHTTPWaker(parent task.Parent, route route.Route, rp *reverseproxy.ReverseProxy) (Waker, E.Error) {
return newWaker(parent, route, rp, nil)
}
func NewStreamWaker(parent task.Parent, entry route.Entry, stream net.Stream) (Waker, E.Error) {
return newWaker(parent, entry, nil, stream)
func NewStreamWaker(parent task.Parent, route route.Route, stream net.Stream) (Waker, E.Error) {
return newWaker(parent, route, nil, stream)
}
// Start implements health.HealthMonitor.
@ -155,7 +155,7 @@ func (w *Watcher) getStatusUpdateReady() health.Status {
// MarshalJSON implements health.HealthMonitor.
func (w *Watcher) MarshalJSON() ([]byte, error) {
var url net.URL
var url *net.URL
if w.hc.URL().Port() != "0" {
url = w.hc.URL()
}

4
internal/docker/idlewatcher/watcher.go
View file

@ -50,8 +50,8 @@ var (
const dockerReqTimeout = 3 * time.Second
func registerWatcher(watcherTask *task.Task, entry route.Entry, waker *waker) (*Watcher, error) {
cfg := entry.IdlewatcherConfig()
func registerWatcher(watcherTask *task.Task, route route.Route, waker *waker) (*Watcher, error) {
cfg := route.IdlewatcherConfig()
if cfg.IdleTimeout == 0 {
panic(errShouldNotReachHere)

10
internal/net/http/loadbalancer/types/server.go
View file

@ -4,7 +4,7 @@ import (
"net/http"
idlewatcher "github.com/yusing/go-proxy/internal/docker/idlewatcher/types"
"github.com/yusing/go-proxy/internal/net/types"
net "github.com/yusing/go-proxy/internal/net/types"
U "github.com/yusing/go-proxy/internal/utils"
F "github.com/yusing/go-proxy/internal/utils/functional"
"github.com/yusing/go-proxy/internal/watcher/health"
@ -15,7 +15,7 @@ type (
_ U.NoCopy
name string
url types.URL
url *net.URL
weight Weight
http.Handler `json:"-"`
@ -26,7 +26,7 @@ type (
http.Handler
health.HealthMonitor
Name() string
URL() types.URL
URL() *net.URL
Weight() Weight
SetWeight(weight Weight)
TryWake() error
@ -37,7 +37,7 @@ type (
var NewServerPool = F.NewMap[Pool]
func NewServer(name string, url types.URL, weight Weight, handler http.Handler, healthMon health.HealthMonitor) Server {
func NewServer(name string, url *net.URL, weight Weight, handler http.Handler, healthMon health.HealthMonitor) Server {
srv := &server{
name: name,
url: url,
@ -59,7 +59,7 @@ func (srv *server) Name() string {
return srv.name
}
func (srv *server) URL() types.URL {
func (srv *server) URL() *net.URL {
return srv.url
}

28
internal/net/http/middleware/middleware.go
View file

@ -196,34 +196,6 @@ func (m *Middleware) ServeHTTP(next http.HandlerFunc, w http.ResponseWriter, r *
next(w, r)
}
// TODO: check conflict or duplicates.
func compileMiddlewares(middlewaresMap map[string]OptionsRaw) ([]*Middleware, E.Error) {
middlewares := make([]*Middleware, 0, len(middlewaresMap))
errs := E.NewBuilder("middlewares compile error")
invalidOpts := E.NewBuilder("options compile error")
for name, opts := range middlewaresMap {
m, err := Get(name)
if err != nil {
errs.Add(err)
continue
}
m, err = m.New(opts)
if err != nil {
invalidOpts.Add(err.Subject(name))
continue
}
middlewares = append(middlewares, m)
}
if invalidOpts.HasError() {
errs.Add(invalidOpts.Error())
}
return middlewares, errs.Error()
}
func PatchReverseProxy(rp *ReverseProxy, middlewaresMap map[string]OptionsRaw) (err E.Error) {
var middlewares []*Middleware
middlewares, err = compileMiddlewares(middlewaresMap)

38
internal/net/http/middleware/middleware_builder.go
View file

@ -4,6 +4,7 @@ import (
"fmt"
"os"
"path"
"sort"
E "github.com/yusing/go-proxy/internal/error"
"gopkg.in/yaml.v3"
@ -39,6 +40,43 @@ func BuildMiddlewaresFromYAML(source string, data []byte, eb *E.Builder) map[str
return middlewares
}
func compileMiddlewares(middlewaresMap map[string]OptionsRaw) ([]*Middleware, E.Error) {
middlewares := make([]*Middleware, 0, len(middlewaresMap))
errs := E.NewBuilder("middlewares compile error")
invalidOpts := E.NewBuilder("options compile error")
for name, opts := range middlewaresMap {
m, err := Get(name)
if err != nil {
errs.Add(err)
continue
}
m, err = m.New(opts)
if err != nil {
invalidOpts.Add(err.Subject(name))
continue
}
middlewares = append(middlewares, m)
}
if invalidOpts.HasError() {
errs.Add(invalidOpts.Error())
}
sort.Sort(ByPriority(middlewares))
return middlewares, errs.Error()
}
func BuildMiddlewareFromMap(name string, middlewaresMap map[string]OptionsRaw) (*Middleware, E.Error) {
compiled, err := compileMiddlewares(middlewaresMap)
if err != nil {
return nil, err
}
return NewMiddlewareChain(name, compiled), nil
}
// TODO: check conflict or duplicates.
func BuildMiddlewareFromChainRaw(name string, defs []map[string]any) (*Middleware, E.Error) {
chainErr := E.NewBuilder("")
chain := make([]*Middleware, 0, len(defs))

8
internal/net/http/middleware/test_utils.go
View file

@ -79,11 +79,11 @@ type TestResult struct {
type testArgs struct {
middlewareOpt OptionsRaw
upstreamURL types.URL
upstreamURL *types.URL
realRoundTrip bool
reqURL types.URL
reqURL *types.URL
reqMethod string
headers http.Header
body []byte
@ -94,13 +94,13 @@ type testArgs struct {
}
func (args *testArgs) setDefaults() {
if args.reqURL.Nil() {
if args.reqURL == nil {
args.reqURL = E.Must(types.ParseURL("https://example.com"))
}
if args.reqMethod == "" {
args.reqMethod = http.MethodGet
}
if args.upstreamURL.Nil() {
if args.upstreamURL == nil {
args.upstreamURL = E.Must(types.ParseURL("https://10.0.0.1:8443")) // dummy url, no actual effect
}
if args.respHeaders == nil {

6
internal/net/http/reverseproxy/reverse_proxy_mod.go
View file

@ -96,7 +96,7 @@ type ReverseProxy struct {
HandlerFunc http.HandlerFunc
TargetName string
TargetURL types.URL
TargetURL *types.URL
}
type httpMetricLogger struct {
@ -167,7 +167,7 @@ func joinURLPath(a, b *url.URL) (path, rawpath string) {
// URLs to the scheme, host, and base path provided in target. If the
// target's path is "/base" and the incoming request was for "/dir",
// the target request will be for /base/dir.
func NewReverseProxy(name string, target types.URL, transport http.RoundTripper) *ReverseProxy {
func NewReverseProxy(name string, target *types.URL, transport http.RoundTripper) *ReverseProxy {
if transport == nil {
panic("nil transport")
}
@ -189,7 +189,7 @@ func (p *ReverseProxy) rewriteRequestURL(req *http.Request) {
targetQuery := p.TargetURL.RawQuery
req.URL.Scheme = p.TargetURL.Scheme
req.URL.Host = p.TargetURL.Host
req.URL.Path, req.URL.RawPath = joinURLPath(p.TargetURL.URL, req.URL)
req.URL.Path, req.URL.RawPath = joinURLPath(&p.TargetURL.URL, req.URL)
if targetQuery == "" || req.URL.RawQuery == "" {
req.URL.RawQuery = targetQuery + req.URL.RawQuery
} else {

3
internal/net/types/url.go
View file

@ -19,7 +19,8 @@ func MustParseURL(url string) *URL {
return u
}
func ParseURL(url string) (u *URL, err error) {
func ParseURL(url string) (*URL, error) {
u := &URL{}
return u, u.Parse(url)
}

3
internal/notif/dispatcher.go
View file

@ -83,6 +83,9 @@ func (disp *Dispatcher) start() {
}
func (disp *Dispatcher) dispatch(msg *LogMessage) {
if true {
return
}
task := disp.task.Subtask("dispatcher")
defer task.Finish("notif dispatched")

62
internal/route/entry/entry.go
View file

@ -1,62 +0,0 @@
package entry
import (
E "github.com/yusing/go-proxy/internal/error"
route "github.com/yusing/go-proxy/internal/route/types"
)
type Entry = route.Entry
func ValidateEntry(m *route.RawEntry) (Entry, E.Error) {
scheme, err := route.NewScheme(m.Scheme)
if err != nil {
return nil, E.From(err)
}
var entry Entry
errs := E.NewBuilder("entry validation failed")
if scheme.IsStream() {
entry = validateStreamEntry(m, errs)
} else {
entry = validateRPEntry(m, scheme, errs)
}
if errs.HasError() {
return nil, errs.Error()
}
if !UseHealthCheck(entry) && (UseLoadBalance(entry) || UseIdleWatcher(entry)) {
return nil, E.New("healthCheck.disable cannot be true when loadbalancer or idlewatcher is enabled")
}
return entry, nil
}
func IsDocker(entry Entry) bool {
iw := entry.IdlewatcherConfig()
return iw != nil && iw.ContainerID != ""
}
func IsZeroPort(entry Entry) bool {
return entry.TargetURL().Port() == "0"
}
func ShouldNotServe(entry Entry) bool {
return IsZeroPort(entry) && !UseIdleWatcher(entry)
}
func UseLoadBalance(entry Entry) bool {
lb := entry.RawEntry().LoadBalance
return lb != nil && lb.Link != ""
}
func UseIdleWatcher(entry Entry) bool {
iw := entry.IdlewatcherConfig()
return iw != nil && iw.IdleTimeout > 0
}
func UseHealthCheck(entry Entry) bool {
hc := entry.RawEntry().HealthCheck
return hc != nil && !hc.Disable
}
func UseAccessLog(entry Entry) bool {
return entry.RawEntry().AccessLog != nil
}

61
internal/route/entry/reverse_proxy.go
View file

@ -1,61 +0,0 @@
package entry
import (
"fmt"
"net/url"
"github.com/yusing/go-proxy/internal/docker"
idlewatcher "github.com/yusing/go-proxy/internal/docker/idlewatcher/types"
E "github.com/yusing/go-proxy/internal/error"
net "github.com/yusing/go-proxy/internal/net/types"
route "github.com/yusing/go-proxy/internal/route/types"
)
type ReverseProxyEntry struct { // real model after validation
Raw *route.RawEntry `json:"raw"`
URL net.URL `json:"url"`
/* Docker only */
Idlewatcher *idlewatcher.Config `json:"idlewatcher,omitempty"`
}
func (rp *ReverseProxyEntry) TargetName() string {
return rp.Raw.Alias
}
func (rp *ReverseProxyEntry) TargetURL() net.URL {
return rp.URL
}
func (rp *ReverseProxyEntry) RawEntry() *route.RawEntry {
return rp.Raw
}
func (rp *ReverseProxyEntry) IdlewatcherConfig() *idlewatcher.Config {
return rp.Idlewatcher
}
func validateRPEntry(m *route.RawEntry, s route.Scheme, errs *E.Builder) *ReverseProxyEntry {
cont := m.Container
if cont == nil {
cont = docker.DummyContainer
}
if m.LoadBalance != nil && m.LoadBalance.Link == "" {
m.LoadBalance = nil
}
port := E.Collect(errs, route.ValidatePort, m.Port)
url := E.Collect(errs, url.Parse, fmt.Sprintf("%s://%s:%d", s, m.Host, port))
iwCfg := E.Collect(errs, idlewatcher.ValidateConfig, cont)
if errs.HasError() {
return nil
}
return &ReverseProxyEntry{
Raw: m,
URL: net.NewURL(url),
Idlewatcher: iwCfg,
}
}

65
internal/route/entry/stream.go
View file

@ -1,65 +0,0 @@
package entry
import (
"fmt"
"github.com/yusing/go-proxy/internal/docker"
idlewatcher "github.com/yusing/go-proxy/internal/docker/idlewatcher/types"
E "github.com/yusing/go-proxy/internal/error"
net "github.com/yusing/go-proxy/internal/net/types"
route "github.com/yusing/go-proxy/internal/route/types"
)
type StreamEntry struct {
Raw *route.RawEntry `json:"raw"`
Scheme route.StreamScheme `json:"scheme"`
URL net.URL `json:"url"`
ListenURL net.URL `json:"listening_url"`
Port route.StreamPort `json:"port,omitempty"`
/* Docker only */
Idlewatcher *idlewatcher.Config `json:"idlewatcher,omitempty"`
}
func (s *StreamEntry) TargetName() string {
return s.Raw.Alias
}
func (s *StreamEntry) TargetURL() net.URL {
return s.URL
}
func (s *StreamEntry) RawEntry() *route.RawEntry {
return s.Raw
}
func (s *StreamEntry) IdlewatcherConfig() *idlewatcher.Config {
return s.Idlewatcher
}
func validateStreamEntry(m *route.RawEntry, errs *E.Builder) *StreamEntry {
cont := m.Container
if cont == nil {
cont = docker.DummyContainer
}
port := E.Collect(errs, route.ValidateStreamPort, m.Port)
scheme := E.Collect(errs, route.ValidateStreamScheme, m.Scheme)
url := E.Collect(errs, net.ParseURL, fmt.Sprintf("%s://%s:%d", scheme.ProxyScheme, m.Host, port.ProxyPort))
listenURL := E.Collect(errs, net.ParseURL, fmt.Sprintf("%s://:%d", scheme.ListeningScheme, port.ListeningPort))
idleWatcherCfg := E.Collect(errs, idlewatcher.ValidateConfig, cont)
if errs.HasError() {
return nil
}
return &StreamEntry{
Raw: m,
Scheme: *scheme,
URL: url,
ListenURL: listenURL,
Port: port,
Idlewatcher: idleWatcherCfg,
}
}

109
internal/route/fileserver.go Normal file
View file

@ -0,0 +1,109 @@
package route
import (
"net/http"
"time"
"github.com/yusing/go-proxy/internal/net/http/middleware"
"github.com/yusing/go-proxy/internal/task"
"github.com/yusing/go-proxy/internal/watcher/health"
"github.com/yusing/go-proxy/internal/watcher/health/monitor"
E "github.com/yusing/go-proxy/internal/error"
)
type (
FileServer struct {
*Route
task *task.Task
middleware *middleware.Middleware
handler http.Handler
startTime time.Time
}
)
func handler(root string) http.Handler {
return http.FileServer(http.Dir(root))
}
func NewFileServer(base *Route) (*FileServer, E.Error) {
s := &FileServer{Route: base}
s.handler = handler(s.Root)
if len(s.Rules) > 0 {
s.handler = s.Rules.BuildHandler(s.Alias, s.handler)
}
if len(s.Middlewares) > 0 {
mid, err := middleware.BuildMiddlewareFromMap(s.Alias, s.Middlewares)
if err != nil {
return nil, err
}
s.middleware = mid
}
return s, nil
}
// Start implements task.TaskStarter.
func (s *FileServer) Start(parent task.Parent) E.Error {
s.startTime = time.Now()
s.task = parent.Subtask("fileserver."+s.Name(), false)
return nil
}
func (s *FileServer) Task() *task.Task {
return s.task
}
// Finish implements task.TaskFinisher.
func (s *FileServer) Finish(reason any) {
s.task.Finish(reason)
}
// ServeHTTP implements http.Handler.
func (s *FileServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
if s.middleware != nil {
s.middleware.ServeHTTP(s.handler.ServeHTTP, w, req)
}
s.handler.ServeHTTP(w, req)
}
// Status implements health.HealthMonitor.
func (s *FileServer) Status() health.Status {
return health.StatusHealthy
}
// Uptime implements health.HealthMonitor.
func (s *FileServer) Uptime() time.Duration {
return time.Since(s.startTime)
}
// Latency implements health.HealthMonitor.
func (s *FileServer) Latency() time.Duration {
return 0
}
// MarshalJSON implements json.Marshaler.
func (s *FileServer) MarshalJSON() ([]byte, error) {
return (&monitor.JSONRepresentation{
Name: s.Alias,
Config: nil,
Status: s.Status(),
Started: s.startTime,
Uptime: s.Uptime(),
Latency: s.Latency(),
LastSeen: time.Now(),
Detail: "",
URL: nil,
}).MarshalJSON()
}
func (s *FileServer) String() string {
return "FileServer " + s.Alias
}
func (s *FileServer) Name() string {
return s.Alias
}

70
internal/route/http.go
View file

@ -16,9 +16,7 @@ import (
loadbalance "github.com/yusing/go-proxy/internal/net/http/loadbalancer/types"
"github.com/yusing/go-proxy/internal/net/http/middleware"
"github.com/yusing/go-proxy/internal/net/http/reverseproxy"
"github.com/yusing/go-proxy/internal/route/entry"
"github.com/yusing/go-proxy/internal/route/routes"
route "github.com/yusing/go-proxy/internal/route/types"
"github.com/yusing/go-proxy/internal/task"
"github.com/yusing/go-proxy/internal/watcher/health"
"github.com/yusing/go-proxy/internal/watcher/health/monitor"
@ -26,7 +24,7 @@ import (
type (
HTTPRoute struct {
*entry.ReverseProxyEntry
*Route
HealthMon health.HealthMonitor `json:"health,omitempty"`
@ -43,9 +41,9 @@ type (
// var globalMux = http.NewServeMux() // TODO: support regex subdomain matching.
func NewHTTPRoute(entry *entry.ReverseProxyEntry) (impl, E.Error) {
func NewHTTPRoute(base *Route) (*HTTPRoute, E.Error) {
trans := gphttp.DefaultTransport
httpConfig := entry.Raw.HTTPConfig
httpConfig := base.HTTPConfig
if httpConfig.NoTLSVerify {
trans = gphttp.DefaultTransportNoTLS
@ -55,21 +53,21 @@ func NewHTTPRoute(entry *entry.ReverseProxyEntry) (impl, E.Error) {
trans.ResponseHeaderTimeout = httpConfig.ResponseHeaderTimeout
}
service := entry.TargetName()
rp := reverseproxy.NewReverseProxy(service, entry.URL, trans)
service := base.TargetName()
rp := reverseproxy.NewReverseProxy(service, base.pURL, trans)
if len(entry.Raw.Middlewares) > 0 {
err := middleware.PatchReverseProxy(rp, entry.Raw.Middlewares)
if len(base.Middlewares) > 0 {
err := middleware.PatchReverseProxy(rp, base.Middlewares)
if err != nil {
return nil, err
}
}
r := &HTTPRoute{
ReverseProxyEntry: entry,
rp: rp,
Route: base,
rp: rp,
l: logging.With().
Str("type", entry.URL.Scheme).
Str("type", string(base.Scheme)).
Str("name", service).
Logger(),
}
@ -82,38 +80,34 @@ func (r *HTTPRoute) String() string {
// Start implements task.TaskStarter.
func (r *HTTPRoute) Start(parent task.Parent) E.Error {
if entry.ShouldNotServe(r) {
return nil
}
r.task = parent.Subtask("http."+r.TargetName(), false)
switch {
case entry.UseIdleWatcher(r):
waker, err := idlewatcher.NewHTTPWaker(parent, r.ReverseProxyEntry, r.rp)
case r.UseIdleWatcher():
waker, err := idlewatcher.NewHTTPWaker(parent, r, r.rp)
if err != nil {
r.task.Finish(err)
return err
}
r.handler = waker
r.HealthMon = waker
case entry.UseHealthCheck(r):
if entry.IsDocker(r) {
client, err := docker.ConnectClient(r.Idlewatcher.DockerHost)
case r.UseHealthCheck():
if r.IsDocker() {
client, err := docker.ConnectClient(r.idlewatcher.DockerHost)
if err == nil {
fallback := monitor.NewHTTPHealthChecker(r.rp.TargetURL, r.Raw.HealthCheck)
r.HealthMon = monitor.NewDockerHealthMonitor(client, r.Idlewatcher.ContainerID, r.TargetName(), r.Raw.HealthCheck, fallback)
fallback := monitor.NewHTTPHealthChecker(r.rp.TargetURL, r.HealthCheck)
r.HealthMon = monitor.NewDockerHealthMonitor(client, r.idlewatcher.ContainerID, r.TargetName(), r.HealthCheck, fallback)
r.task.OnCancel("close_docker_client", client.Close)
}
}
if r.HealthMon == nil {
r.HealthMon = monitor.NewHTTPHealthMonitor(r.rp.TargetURL, r.Raw.HealthCheck)
r.HealthMon = monitor.NewHTTPHealthMonitor(r.rp.TargetURL, r.HealthCheck)
}
}
if entry.UseAccessLog(r) {
if r.UseAccessLog() {
var err error
r.rp.AccessLogger, err = accesslog.NewFileAccessLogger(r.task, r.Raw.AccessLog)
r.rp.AccessLogger, err = accesslog.NewFileAccessLogger(r.task, r.AccessLog)
if err != nil {
r.task.Finish(err)
return E.From(err)
@ -121,7 +115,7 @@ func (r *HTTPRoute) Start(parent task.Parent) E.Error {
}
if r.handler == nil {
pathPatterns := r.Raw.PathPatterns
pathPatterns := r.PathPatterns
switch {
case len(pathPatterns) == 0:
r.handler = r.rp
@ -144,8 +138,8 @@ func (r *HTTPRoute) Start(parent task.Parent) E.Error {
}
}
if len(r.Raw.Rules) > 0 {
r.handler = r.Raw.Rules.BuildHandler(r.TargetName(), r.handler)
if len(r.Rules) > 0 {
r.handler = r.Rules.BuildHandler(r.TargetName(), r.handler)
}
if r.HealthMon != nil {
@ -154,7 +148,7 @@ func (r *HTTPRoute) Start(parent task.Parent) E.Error {
}
}
if entry.UseLoadBalance(r) {
if r.UseLoadBalance() {
r.addToLoadBalancer(parent)
} else {
routes.SetHTTPRoute(r.TargetName(), r)
@ -191,15 +185,15 @@ func (r *HTTPRoute) HealthMonitor() health.HealthMonitor {
func (r *HTTPRoute) addToLoadBalancer(parent task.Parent) {
var lb *loadbalancer.LoadBalancer
cfg := r.Raw.LoadBalance
cfg := r.LoadBalance
l, ok := routes.GetHTTPRoute(cfg.Link)
var linked *HTTPRoute
if ok {
linked = l.(*HTTPRoute)
lb = linked.loadBalancer
lb.UpdateConfigIfNeeded(cfg)
if linked.Raw.Homepage.IsEmpty() && !r.Raw.Homepage.IsEmpty() {
linked.Raw.Homepage = r.Raw.Homepage
if linked.Homepage.IsEmpty() && !r.Homepage.IsEmpty() {
linked.Homepage = r.Homepage
}
} else {
lb = loadbalancer.New(cfg)
@ -207,11 +201,9 @@ func (r *HTTPRoute) addToLoadBalancer(parent task.Parent) {
panic(err) // should always return nil
}
linked = &HTTPRoute{
ReverseProxyEntry: &entry.ReverseProxyEntry{
Raw: &route.RawEntry{
Alias: cfg.Link,
Homepage: r.Raw.Homepage,
},
Route: &Route{
Alias: cfg.Link,
Homepage: r.Homepage,
},
HealthMon: lb,
loadBalancer: lb,
@ -220,7 +212,7 @@ func (r *HTTPRoute) addToLoadBalancer(parent task.Parent) {
routes.SetHTTPRoute(cfg.Link, linked)
}
r.loadBalancer = lb
r.server = loadbalance.NewServer(r.task.Name(), r.rp.TargetURL, r.Raw.LoadBalance.Weight, r.handler, r.HealthMon)
r.server = loadbalance.NewServer(r.task.Name(), r.rp.TargetURL, r.LoadBalance.Weight, r.handler, r.HealthMon)
lb.AddServer(r.server)
r.task.OnCancel("lb_remove_server", func() {
lb.RemoveServer(r.server)

70
internal/route/provider/docker.go
View file

@ -3,7 +3,6 @@ package provider
import (
"fmt"
"strconv"
"strings"
"github.com/docker/docker/client"
"github.com/rs/zerolog"
@ -62,15 +61,13 @@ func (p *DockerProvider) NewWatcher() watcher.Watcher {
}
func (p *DockerProvider) loadRoutesImpl() (route.Routes, E.Error) {
routes := route.NewRoutes()
entries := route.NewProxyEntries()
containers, err := docker.ListContainers(p.dockerHost)
if err != nil {
return routes, E.From(err)
return nil, E.From(err)
}
errs := E.NewBuilder("")
routes := make(route.Routes)
for _, c := range containers {
container := docker.FromDocker(&c, p.dockerHost)
@ -78,47 +75,34 @@ func (p *DockerProvider) loadRoutesImpl() (route.Routes, E.Error) {
continue
}
newEntries, err := p.entriesFromContainerLabels(container)
newEntries, err := p.routesFromContainerLabels(container)
if err != nil {
errs.Add(err.Subject(container.ContainerName))
}
// although err is not nil
// there may be some valid entries in `en`
dups := entries.MergeFrom(newEntries)
// add the duplicate proxy entries to the error
dups.RangeAll(func(k string, v *route.RawEntry) {
errs.Addf("duplicated alias %s", k)
})
for k, v := range newEntries {
if routes.Contains(k) {
errs.Addf("duplicated alias %s", k)
} else {
routes[k] = v
}
}
}
routes, err = route.FromEntries(p.ShortName(), entries)
errs.Add(err)
return routes, errs.Error()
}
func (p *DockerProvider) shouldIgnore(container *docker.Container) bool {
return container.IsExcluded ||
!container.IsExplicit && p.IsExplicitOnly() ||
!container.IsExplicit && container.IsDatabase ||
strings.HasSuffix(container.ContainerName, "-old")
}
// Returns a list of proxy entries for a container.
// Always non-nil.
func (p *DockerProvider) entriesFromContainerLabels(container *docker.Container) (entries route.RawEntries, _ E.Error) {
entries = route.NewProxyEntries()
if p.shouldIgnore(container) {
return
func (p *DockerProvider) routesFromContainerLabels(container *docker.Container) (route.Routes, E.Error) {
if !container.IsExplicit && p.IsExplicitOnly() {
return nil, nil
}
routes := make(route.Routes, len(container.Aliases))
// init entries map for all aliases
for _, a := range container.Aliases {
entries.Store(a, &route.RawEntry{
Alias: a,
Container: container,
})
routes[a] = &route.Route{Container: container}
}
errs := E.NewBuilder("label errors")
@ -170,32 +154,28 @@ func (p *DockerProvider) entriesFromContainerLabels(container *docker.Container)
}
// init entry if not exist
en, ok := entries.Load(alias)
r, ok := routes[alias]
if !ok {
en = &route.RawEntry{
Alias: alias,
Container: container,
}
entries.Store(alias, en)
r = &route.Route{Container: container}
routes[alias] = r
}
// deserialize map into entry object
err := U.Deserialize(entryMap, en)
err := U.Deserialize(entryMap, r)
if err != nil {
errs.Add(err.Subject(alias))
} else {
entries.Store(alias, en)
routes[alias] = r
}
}
if wildcardProps != nil {
entries.Range(func(alias string, re *route.RawEntry) bool {
for _, re := range routes {
if err := U.Deserialize(wildcardProps, re); err != nil {
errs.Add(err.Subject(docker.WildcardAlias))
return false
break
}
return true
})
}
}
return entries, errs.Error()
return routes, errs.Error()
}

6
internal/route/provider/docker_labels_test.go
View file

@ -20,7 +20,7 @@ func TestParseDockerLabels(t *testing.T) {
labels := make(map[string]string)
ExpectNoError(t, yaml.Unmarshal(testDockerLabelsYAML, &labels))
routes, err := provider.entriesFromContainerLabels(
routes, err := provider.routesFromContainerLabels(
docker.FromDocker(&types.Container{
Names: []string{"container"},
Labels: labels,
@ -31,6 +31,6 @@ func TestParseDockerLabels(t *testing.T) {
}, "/var/run/docker.sock"),
)
ExpectNoError(t, err)
ExpectTrue(t, routes.Has("app"))
ExpectTrue(t, routes.Has("app1"))
ExpectTrue(t, routes.Contains("app"))
ExpectTrue(t, routes.Contains("app1"))
}

139
internal/route/provider/docker_test.go
View file

@ -11,7 +11,6 @@ import (
D "github.com/yusing/go-proxy/internal/docker"
E "github.com/yusing/go-proxy/internal/error"
"github.com/yusing/go-proxy/internal/route"
"github.com/yusing/go-proxy/internal/route/entry"
T "github.com/yusing/go-proxy/internal/route/types"
. "github.com/yusing/go-proxy/internal/utils/testing"
)
@ -23,7 +22,7 @@ const (
testDockerIP = "172.17.0.123"
)
func makeEntries(cont *types.Container, dockerHostIP ...string) route.RawEntries {
func makeRoutes(cont *types.Container, dockerHostIP ...string) route.Routes {
var p DockerProvider
var host string
if len(dockerHostIP) > 0 {
@ -32,11 +31,11 @@ func makeEntries(cont *types.Container, dockerHostIP ...string) route.RawEntries
host = client.DefaultDockerHost
}
p.name = "test"
entries := E.Must(p.entriesFromContainerLabels(D.FromDocker(cont, host)))
entries.RangeAll(func(k string, v *route.RawEntry) {
v.Finalize()
})
return entries
routes := E.Must(p.routesFromContainerLabels(D.FromDocker(cont, host)))
for _, r := range routes {
r.Finalize()
}
return routes
}
func TestExplicitOnly(t *testing.T) {
@ -66,7 +65,7 @@ func TestApplyLabel(t *testing.T) {
"prop4": "value4",
},
}
entries := makeEntries(&types.Container{
entries := makeRoutes(&types.Container{
Names: dummyNames,
Labels: map[string]string{
D.LabelAliases: "a,b",
@ -91,9 +90,9 @@ func TestApplyLabel(t *testing.T) {
},
})
a, ok := entries.Load("a")
a, ok := entries["a"]
ExpectTrue(t, ok)
b, ok := entries.Load("b")
b, ok := entries["b"]
ExpectTrue(t, ok)
ExpectEqual(t, a.Scheme, "https")
@ -102,8 +101,8 @@ func TestApplyLabel(t *testing.T) {
ExpectEqual(t, a.Host, "app")
ExpectEqual(t, b.Host, "app")
ExpectEqual(t, a.Port, "4567")
ExpectEqual(t, b.Port, "4567")
ExpectEqual(t, a.Port.Proxy, 4567)
ExpectEqual(t, b.Port.Proxy, 4567)
ExpectTrue(t, a.NoTLSVerify)
ExpectTrue(t, b.NoTLSVerify)
@ -139,7 +138,7 @@ func TestApplyLabel(t *testing.T) {
}
func TestApplyLabelWithAlias(t *testing.T) {
entries := makeEntries(&types.Container{
entries := makeRoutes(&types.Container{
Names: dummyNames,
State: "running",
Labels: map[string]string{
@ -150,23 +149,23 @@ func TestApplyLabelWithAlias(t *testing.T) {
"proxy.c.scheme": "https",
},
})
a, ok := entries.Load("a")
a, ok := entries["a"]
ExpectTrue(t, ok)
b, ok := entries.Load("b")
b, ok := entries["b"]
ExpectTrue(t, ok)
c, ok := entries.Load("c")
c, ok := entries["c"]
ExpectTrue(t, ok)
ExpectEqual(t, a.Scheme, "http")
ExpectEqual(t, a.Port, "3333")
ExpectEqual(t, a.Port.Proxy, 3333)
ExpectEqual(t, a.NoTLSVerify, true)
ExpectEqual(t, b.Scheme, "http")
ExpectEqual(t, b.Port, "1234")
ExpectEqual(t, b.Port.Proxy, 1234)
ExpectEqual(t, c.Scheme, "https")
}
func TestApplyLabelWithRef(t *testing.T) {
entries := makeEntries(&types.Container{
entries := makeRoutes(&types.Container{
Names: dummyNames,
State: "running",
Labels: map[string]string{
@ -178,19 +177,19 @@ func TestApplyLabelWithRef(t *testing.T) {
"proxy.#3.scheme": "https",
},
})
a, ok := entries.Load("a")
a, ok := entries["a"]
ExpectTrue(t, ok)
b, ok := entries.Load("b")
b, ok := entries["b"]
ExpectTrue(t, ok)
c, ok := entries.Load("c")
c, ok := entries["c"]
ExpectTrue(t, ok)
ExpectEqual(t, a.Scheme, "http")
ExpectEqual(t, a.Host, "localhost")
ExpectEqual(t, a.Port, "4444")
ExpectEqual(t, b.Port, "9999")
ExpectEqual(t, a.Port.Proxy, 4444)
ExpectEqual(t, b.Port.Proxy, 9999)
ExpectEqual(t, c.Scheme, "https")
ExpectEqual(t, c.Port, "1111")
ExpectEqual(t, c.Port.Proxy, 1111)
}
func TestApplyLabelWithRefIndexError(t *testing.T) {
@ -204,7 +203,7 @@ func TestApplyLabelWithRefIndexError(t *testing.T) {
},
}, "")
var p DockerProvider
_, err := p.entriesFromContainerLabels(c)
_, err := p.routesFromContainerLabels(c)
ExpectError(t, ErrAliasRefIndexOutOfRange, err)
c = D.FromDocker(&types.Container{
@ -215,7 +214,7 @@ func TestApplyLabelWithRefIndexError(t *testing.T) {
"proxy.#0.host": "localhost",
},
}, "")
_, err = p.entriesFromContainerLabels(c)
_, err = p.routesFromContainerLabels(c)
ExpectError(t, ErrAliasRefIndexOutOfRange, err)
}
@ -229,17 +228,17 @@ func TestDynamicAliases(t *testing.T) {
},
}
entries := makeEntries(c)
entries := makeRoutes(c)
raw, ok := entries.Load("app1")
r, ok := entries["app1"]
ExpectTrue(t, ok)
ExpectEqual(t, raw.Scheme, "http")
ExpectEqual(t, raw.Port, "1234")
ExpectEqual(t, r.Scheme, "http")
ExpectEqual(t, r.Port.Proxy, 1234)
raw, ok = entries.Load("app1_backend")
r, ok = entries["app1_backend"]
ExpectTrue(t, ok)
ExpectEqual(t, raw.Scheme, "http")
ExpectEqual(t, raw.Port, "5678")
ExpectEqual(t, r.Scheme, "http")
ExpectEqual(t, r.Port.Proxy, 5678)
}
func TestDisableHealthCheck(t *testing.T) {
@ -251,22 +250,22 @@ func TestDisableHealthCheck(t *testing.T) {
"proxy.a.port": "1234",
},
}
raw, ok := makeEntries(c).Load("a")
r, ok := makeRoutes(c)["a"]
ExpectTrue(t, ok)
ExpectEqual(t, raw.HealthCheck, nil)
ExpectEqual(t, r.HealthCheck, nil)
}
func TestPublicIPLocalhost(t *testing.T) {
c := &types.Container{Names: dummyNames, State: "running"}
raw, ok := makeEntries(c).Load("a")
r, ok := makeRoutes(c)["a"]
ExpectTrue(t, ok)
ExpectEqual(t, raw.Container.PublicIP, "127.0.0.1")
ExpectEqual(t, raw.Host, raw.Container.PublicIP)
ExpectEqual(t, r.Container.PublicIP, "127.0.0.1")
ExpectEqual(t, r.Host, r.Container.PublicIP)
}
func TestPublicIPRemote(t *testing.T) {
c := &types.Container{Names: dummyNames, State: "running"}
raw, ok := makeEntries(c, testIP).Load("a")
raw, ok := makeRoutes(c, testIP)["a"]
ExpectTrue(t, ok)
ExpectEqual(t, raw.Container.PublicIP, testIP)
ExpectEqual(t, raw.Host, raw.Container.PublicIP)
@ -283,10 +282,10 @@ func TestPrivateIPLocalhost(t *testing.T) {
},
},
}
raw, ok := makeEntries(c).Load("a")
r, ok := makeRoutes(c)["a"]
ExpectTrue(t, ok)
ExpectEqual(t, raw.Container.PrivateIP, testDockerIP)
ExpectEqual(t, raw.Host, raw.Container.PrivateIP)
ExpectEqual(t, r.Container.PrivateIP, testDockerIP)
ExpectEqual(t, r.Host, r.Container.PrivateIP)
}
func TestPrivateIPRemote(t *testing.T) {
@ -301,11 +300,11 @@ func TestPrivateIPRemote(t *testing.T) {
},
},
}
raw, ok := makeEntries(c, testIP).Load("a")
r, ok := makeRoutes(c, testIP)["a"]
ExpectTrue(t, ok)
ExpectEqual(t, raw.Container.PrivateIP, "")
ExpectEqual(t, raw.Container.PublicIP, testIP)
ExpectEqual(t, raw.Host, raw.Container.PublicIP)
ExpectEqual(t, r.Container.PrivateIP, "")
ExpectEqual(t, r.Container.PublicIP, testIP)
ExpectEqual(t, r.Host, r.Container.PublicIP)
}
func TestStreamDefaultValues(t *testing.T) {
@ -328,59 +327,57 @@ func TestStreamDefaultValues(t *testing.T) {
}
t.Run("local", func(t *testing.T) {
raw, ok := makeEntries(cont).Load("a")
r, ok := makeRoutes(cont)["a"]
ExpectTrue(t, ok)
en := E.Must(entry.ValidateEntry(raw))
a := ExpectType[*entry.StreamEntry](t, en)
ExpectEqual(t, a.Scheme.ListeningScheme, T.Scheme("udp"))
ExpectEqual(t, a.Scheme.ProxyScheme, T.Scheme("udp"))
ExpectEqual(t, a.URL.Hostname(), privIP)
ExpectEqual(t, a.Port.ListeningPort, 0)
ExpectEqual(t, a.Port.ProxyPort, T.Port(privPort))
ExpectNoError(t, r.Validate())
ExpectEqual(t, r.Scheme, T.Scheme("udp"))
ExpectEqual(t, r.TargetURL().Hostname(), privIP)
ExpectEqual(t, r.Port.Listening, 0)
ExpectEqual(t, r.Port.Proxy, int(privPort))
})
t.Run("remote", func(t *testing.T) {
raw, ok := makeEntries(cont, testIP).Load("a")
r, ok := makeRoutes(cont, testIP)["a"]
ExpectTrue(t, ok)
en := E.Must(entry.ValidateEntry(raw))
a := ExpectType[*entry.StreamEntry](t, en)
ExpectEqual(t, a.Scheme.ListeningScheme, T.Scheme("udp"))
ExpectEqual(t, a.Scheme.ProxyScheme, T.Scheme("udp"))
ExpectEqual(t, a.URL.Hostname(), testIP)
ExpectEqual(t, a.Port.ListeningPort, 0)
ExpectEqual(t, a.Port.ProxyPort, T.Port(pubPort))
ExpectNoError(t, r.Validate())
ExpectEqual(t, r.Scheme, T.Scheme("udp"))
ExpectEqual(t, r.TargetURL().Hostname(), testIP)
ExpectEqual(t, r.Port.Listening, 0)
ExpectEqual(t, r.Port.Proxy, int(pubPort))
})
}
func TestExplicitExclude(t *testing.T) {
_, ok := makeEntries(&types.Container{
_, ok := makeRoutes(&types.Container{
Names: dummyNames,
Labels: map[string]string{
D.LabelAliases: "a",
D.LabelExclude: "true",
"proxy.a.no_tls_verify": "true",
},
}, "").Load("a")
}, "")["a"]
ExpectFalse(t, ok)
}
func TestImplicitExcludeDatabase(t *testing.T) {
t.Run("mount path detection", func(t *testing.T) {
_, ok := makeEntries(&types.Container{
r, ok := makeRoutes(&types.Container{
Names: dummyNames,
Mounts: []types.MountPoint{
{Source: "/data", Destination: "/var/lib/postgresql/data"},
},
}).Load("a")
ExpectFalse(t, ok)
})["a"]
ExpectTrue(t, ok)
ExpectTrue(t, r.ShouldNotServe())
})
t.Run("exposed port detection", func(t *testing.T) {
_, ok := makeEntries(&types.Container{
r, ok := makeRoutes(&types.Container{
Names: dummyNames,
Ports: []types.Port{
{Type: "tcp", PrivatePort: 5432, PublicPort: 5432},
},
}).Load("a")
ExpectFalse(t, ok)
})["a"]
ExpectTrue(t, ok)
ExpectTrue(t, r.ShouldNotServe())
})
}

39
internal/route/provider/event_handler.go
View file

@ -4,7 +4,6 @@ import (
"github.com/yusing/go-proxy/internal/common"
E "github.com/yusing/go-proxy/internal/error"
"github.com/yusing/go-proxy/internal/route"
"github.com/yusing/go-proxy/internal/route/entry"
"github.com/yusing/go-proxy/internal/route/provider/types"
"github.com/yusing/go-proxy/internal/task"
"github.com/yusing/go-proxy/internal/watcher"
@ -31,10 +30,10 @@ func (p *Provider) newEventHandler() *EventHandler {
func (handler *EventHandler) Handle(parent task.Parent, events []watcher.Event) {
oldRoutes := handler.provider.routes
newRoutes, err := handler.provider.loadRoutesImpl()
newRoutes, err := handler.provider.loadRoutes()
if err != nil {
handler.errs.Add(err)
if newRoutes.Size() == 0 {
if len(newRoutes) == 0 {
return
}
}
@ -47,34 +46,34 @@ func (handler *EventHandler) Handle(parent task.Parent, events []watcher.Event)
E.LogDebug(eventsLog.About(), eventsLog.Error(), handler.provider.Logger())
oldRoutesLog := E.NewBuilder("old routes")
oldRoutes.RangeAllParallel(func(k string, r *route.Route) {
for k := range oldRoutes {
oldRoutesLog.Adds(k)
})
}
E.LogDebug(oldRoutesLog.About(), oldRoutesLog.Error(), handler.provider.Logger())
newRoutesLog := E.NewBuilder("new routes")
newRoutes.RangeAllParallel(func(k string, r *route.Route) {
for k := range newRoutes {
newRoutesLog.Adds(k)
})
}
E.LogDebug(newRoutesLog.About(), newRoutesLog.Error(), handler.provider.Logger())
}
oldRoutes.RangeAll(func(k string, oldr *route.Route) {
newr, ok := newRoutes.Load(k)
for k, oldr := range oldRoutes {
newr, ok := newRoutes[k]
switch {
case !ok:
handler.Remove(oldr)
case handler.matchAny(events, newr):
handler.Update(parent, oldr, newr)
case entry.ShouldNotServe(newr):
case newr.ShouldNotServe():
handler.Remove(oldr)
}
})
newRoutes.RangeAll(func(k string, newr *route.Route) {
if !(oldRoutes.Has(k) || entry.ShouldNotServe(newr)) {
}
for k, newr := range newRoutes {
if _, ok := oldRoutes[k]; !(ok || newr.ShouldNotServe()) {
handler.Add(parent, newr)
}
})
}
}
func (handler *EventHandler) matchAny(events []watcher.Event, route *route.Route) bool {
@ -89,8 +88,8 @@ func (handler *EventHandler) matchAny(events []watcher.Event, route *route.Route
func (handler *EventHandler) match(event watcher.Event, route *route.Route) bool {
switch handler.provider.GetType() {
case types.ProviderTypeDocker:
return route.Entry.Container.ContainerID == event.ActorID ||
route.Entry.Container.ContainerName == event.ActorName
return route.Container.ContainerID == event.ActorID ||
route.Container.ContainerName == event.ActorName
case types.ProviderTypeFile:
return true
}
@ -103,14 +102,14 @@ func (handler *EventHandler) Add(parent task.Parent, route *route.Route) {
if err != nil {
handler.errs.Add(err.Subject("add"))
} else {
handler.added.Adds(route.Entry.Alias)
handler.added.Adds(route.Alias)
}
}
func (handler *EventHandler) Remove(route *route.Route) {
route.Finish("route removed")
handler.provider.routes.Delete(route.Entry.Alias)
handler.removed.Adds(route.Entry.Alias)
delete(handler.provider.routes, route.Alias)
handler.removed.Adds(route.Alias)
}
func (handler *EventHandler) Update(parent task.Parent, oldRoute *route.Route, newRoute *route.Route) {
@ -119,7 +118,7 @@ func (handler *EventHandler) Update(parent task.Parent, oldRoute *route.Route, n
if err != nil {
handler.errs.Add(err.Subject("update"))
} else {
handler.updated.Adds(newRoute.Entry.Alias)
handler.updated.Adds(newRoute.Alias)
}
}

22
internal/route/provider/file.go
View file

@ -33,16 +33,13 @@ func FileProviderImpl(filename string) (ProviderImpl, error) {
return impl, nil
}
func validate(provider string, data []byte) (route.Routes, E.Error) {
entries, err := utils.DeserializeYAMLMap[*route.RawEntry](data)
if err != nil {
return route.NewRoutes(), err
}
return route.FromEntries(provider, entries)
func validate(data []byte) (routes route.Routes, err E.Error) {
err = utils.DeserializeYAML(data, &routes)
return
}
func Validate(data []byte) (err E.Error) {
_, err = validate("", data)
_, err = validate(data)
return
}
@ -63,14 +60,15 @@ func (p *FileProvider) Logger() *zerolog.Logger {
}
func (p *FileProvider) loadRoutesImpl() (route.Routes, E.Error) {
routes := route.NewRoutes()
data, err := os.ReadFile(p.path)
if err != nil {
return routes, E.From(err)
return nil, E.Wrap(err)
}
return validate(p.ShortName(), data)
routes, err := validate(data)
if err != nil {
return nil, E.Wrap(err)
}
return routes, nil
}
func (p *FileProvider) NewWatcher() W.Watcher {

2
internal/route/provider/file_test.go
View file

@ -12,6 +12,6 @@ import (
var testAllFieldsYAML []byte
func TestFile(t *testing.T) {
_, err := validate("", testAllFieldsYAML)
_, err := validate(testAllFieldsYAML)
ExpectNoError(t, err)
}

81
internal/route/provider/provider.go
View file

@ -8,7 +8,8 @@ import (
"github.com/rs/zerolog"
E "github.com/yusing/go-proxy/internal/error"
R "github.com/yusing/go-proxy/internal/route"
"github.com/yusing/go-proxy/internal/logging"
"github.com/yusing/go-proxy/internal/route"
"github.com/yusing/go-proxy/internal/route/provider/types"
"github.com/yusing/go-proxy/internal/task"
W "github.com/yusing/go-proxy/internal/watcher"
@ -20,7 +21,7 @@ type (
ProviderImpl `json:"-"`
t types.ProviderType
routes R.Routes
routes route.Routes
watcher W.Watcher
}
@ -28,7 +29,7 @@ type (
fmt.Stringer
ShortName() string
IsExplicitOnly() bool
loadRoutesImpl() (R.Routes, E.Error)
loadRoutesImpl() (route.Routes, E.Error)
NewWatcher() W.Watcher
Logger() *zerolog.Logger
}
@ -41,10 +42,7 @@ const (
var ErrEmptyProviderName = errors.New("empty provider name")
func newProvider(t types.ProviderType) *Provider {
return &Provider{
t: t,
routes: R.NewRoutes(),
}
return &Provider{t: t}
}
func NewFileProvider(filename string) (p *Provider, err error) {
@ -84,13 +82,15 @@ func (p *Provider) MarshalText() ([]byte, error) {
return []byte(p.String()), nil
}
func (p *Provider) startRoute(parent task.Parent, r *R.Route) E.Error {
func (p *Provider) startRoute(parent task.Parent, r *route.Route) E.Error {
if r.ShouldNotServe() {
logging.Debug().Str("alias", r.Alias).Str("provider", p.ShortName()).Msg("route excluded")
return nil
}
err := r.Start(parent)
if err != nil {
return err.Subject(r.Entry.Alias)
return err.Subject(r.Alias)
}
p.routes.Store(r.Entry.Alias, r)
return nil
}
@ -98,11 +98,10 @@ func (p *Provider) startRoute(parent task.Parent, r *R.Route) E.Error {
func (p *Provider) Start(parent task.Parent) E.Error {
t := parent.Subtask("provider."+p.String(), false)
// routes and event queue will stop on config reload
errs := p.routes.CollectErrorsParallel(
func(alias string, r *R.Route) error {
return p.startRoute(t, r)
})
errs := E.NewBuilder("routes error")
for _, r := range p.routes {
errs.Add(p.startRoute(t, r))
}
eventQueue := events.NewEventQueue(
t.Subtask("event_queue", false),
@ -119,32 +118,54 @@ func (p *Provider) Start(parent task.Parent) E.Error {
)
eventQueue.Start(p.watcher.Events(t.Context()))
if err := E.Join(errs...); err != nil {
if err := errs.Error(); err != nil {
return err.Subject(p.String())
}
return nil
}
func (p *Provider) RangeRoutes(do func(string, *R.Route)) {
p.routes.RangeAll(do)
func (p *Provider) RangeRoutes(do func(string, *route.Route)) {
for alias, r := range p.routes {
do(alias, r)
}
}
func (p *Provider) GetRoute(alias string) (*R.Route, bool) {
return p.routes.Load(alias)
func (p *Provider) GetRoute(alias string) (r *route.Route, ok bool) {
r, ok = p.routes[alias]
return
}
func (p *Provider) LoadRoutes() E.Error {
var err E.Error
p.routes, err = p.loadRoutesImpl()
if p.routes.Size() > 0 {
return err
func (p *Provider) loadRoutes() (routes route.Routes, err E.Error) {
routes, err = p.loadRoutesImpl()
if err != nil && len(routes) == 0 {
return nil, err
}
if err == nil {
return nil
errs := E.NewBuilder("routes error")
errs.Add(err)
// check for exclusion
// set alias and provider, then validate
for alias, r := range routes {
r.Alias = alias
r.Provider = p.ShortName()
r.Finalize()
if err := r.Validate(); err != nil {
errs.Add(err.Subject(alias))
delete(routes, alias)
continue
}
if r.ShouldExclude() {
delete(routes, alias)
continue
}
}
return err
return routes, errs.Error()
}
func (p *Provider) LoadRoutes() (err E.Error) {
p.routes, err = p.loadRoutes()
return
}
func (p *Provider) NumRoutes() int {
return p.routes.Size()
return len(p.routes)
}

28
internal/route/provider/stats.go
View file

@ -17,10 +17,11 @@ type (
NumUnknown uint16 `json:"unknown"`
}
ProviderStats struct {
Total uint16 `json:"total"`
RPs RouteStats `json:"reverse_proxies"`
Streams RouteStats `json:"streams"`
Type types.ProviderType `json:"type"`
Total uint16 `json:"total"`
RPs RouteStats `json:"reverse_proxies"`
FileServers RouteStats `json:"file_servers"`
Streams RouteStats `json:"streams"`
Type types.ProviderType `json:"type"`
}
)
@ -55,19 +56,22 @@ func (stats *RouteStats) AddOther(other RouteStats) {
}
func (p *Provider) Statistics() ProviderStats {
var rps, streams RouteStats
p.routes.RangeAll(func(_ string, r *R.Route) {
switch r.Type {
var rps, fileServers, streams RouteStats
for _, r := range p.routes {
switch r.Type() {
case route.RouteTypeReverseProxy:
rps.Add(r)
case route.RouteTypeStream:
streams.Add(r)
default:
fileServers.Add(r)
}
})
}
return ProviderStats{
Total: rps.Total + streams.Total,
RPs: rps,
Streams: streams,
Type: p.t,
Total: rps.Total + streams.Total,
RPs: rps,
FileServers: fileServers,
Streams: streams,
Type: p.t,
}
}

399
internal/route/route.go Executable file → Normal file
View file

@ -1,104 +1,355 @@
package route
import (
"fmt"
"strconv"
"strings"
"github.com/yusing/go-proxy/internal/docker"
E "github.com/yusing/go-proxy/internal/error"
url "github.com/yusing/go-proxy/internal/net/types"
"github.com/yusing/go-proxy/internal/route/entry"
"github.com/yusing/go-proxy/internal/route/types"
idlewatcher "github.com/yusing/go-proxy/internal/docker/idlewatcher/types"
"github.com/yusing/go-proxy/internal/homepage"
"github.com/yusing/go-proxy/internal/logging"
net "github.com/yusing/go-proxy/internal/net/types"
"github.com/yusing/go-proxy/internal/task"
U "github.com/yusing/go-proxy/internal/utils"
F "github.com/yusing/go-proxy/internal/utils/functional"
"github.com/yusing/go-proxy/internal/watcher/health"
dockertypes "github.com/docker/docker/api/types"
"github.com/yusing/go-proxy/internal/common"
E "github.com/yusing/go-proxy/internal/error"
"github.com/yusing/go-proxy/internal/net/http/accesslog"
loadbalance "github.com/yusing/go-proxy/internal/net/http/loadbalancer/types"
"github.com/yusing/go-proxy/internal/route/rules"
"github.com/yusing/go-proxy/internal/route/types"
"github.com/yusing/go-proxy/internal/utils"
)
type (
Route struct {
_ U.NoCopy
impl
Type types.RouteType
Entry *RawEntry
}
Routes = F.Map[string, *Route]
_ utils.NoCopy
impl interface {
types.Route
task.TaskStarter
task.TaskFinisher
String() string
TargetURL() url.URL
Alias string `json:"alias"`
Scheme types.Scheme `json:"scheme,omitempty"`
Host string `json:"host,omitempty"`
Port types.Port `json:"port,omitempty"`
Root string `json:"root,omitempty"`
types.HTTPConfig
PathPatterns []string `json:"path_patterns,omitempty"`
Rules rules.Rules `json:"rules,omitempty" validate:"omitempty,unique=Name"`
HealthCheck *health.HealthCheckConfig `json:"healthcheck,omitempty"`
LoadBalance *loadbalance.Config `json:"load_balance,omitempty"`
Middlewares map[string]docker.LabelMap `json:"middlewares,omitempty"`
Homepage *homepage.Item `json:"homepage,omitempty"`
AccessLog *accesslog.Config `json:"access_log,omitempty"`
/* Docker only */
Container *docker.Container `json:"container,omitempty"`
Provider string `json:"provider,omitempty"`
// private fields
lURL, pURL *net.URL // listening url and proxy url
idlewatcher *idlewatcher.Config
impl types.Route
isValidated bool
}
RawEntry = types.RawEntry
RawEntries = types.RawEntries
Routes map[string]*Route
)
// function alias.
var (
NewRoutes = F.NewMap[Routes]
NewProxyEntries = types.NewProxyEntries
)
func (rt *Route) Container() *docker.Container {
if rt.Entry.Container == nil {
return docker.DummyContainer
}
return rt.Entry.Container
func (r Routes) Contains(alias string) bool {
_, ok := r[alias]
return ok
}
func NewRoute(raw *RawEntry) (*Route, E.Error) {
raw.Finalize()
en, err := entry.ValidateEntry(raw)
if err != nil {
return nil, err
func (r *Route) Validate() E.Error {
if r.isValidated {
return nil
}
r.isValidated = true
if r.ShouldNotServe() {
return nil
}
var t types.RouteType
var rt impl
errs := E.NewBuilder("entry validation failed")
switch e := en.(type) {
case *entry.StreamEntry:
t = types.RouteTypeStream
rt, err = NewStreamRoute(e)
case *entry.ReverseProxyEntry:
t = types.RouteTypeReverseProxy
rt, err = NewHTTPRoute(e)
switch r.Scheme {
case types.SchemeFileServer:
return nil
case types.SchemeHTTP, types.SchemeHTTPS:
if r.Port.Listening != 0 {
errs.Addf("unexpected listening port for %s scheme", r.Scheme)
}
fallthrough
case types.SchemeTCP, types.SchemeUDP:
r.lURL = E.Collect(errs, net.ParseURL, fmt.Sprintf("%s://%s:%d", r.Scheme, r.Host, r.Port.Listening))
fallthrough
default:
panic("bug: should not reach here")
if r.Port.Proxy == 0 && !r.UseIdleWatcher() {
errs.Adds("missing proxy port")
}
if r.LoadBalance != nil && r.LoadBalance.Link == "" {
r.LoadBalance = nil
}
r.pURL = E.Collect(errs, net.ParseURL, fmt.Sprintf("%s://%s:%d", r.Scheme, r.Host, r.Port.Proxy))
r.idlewatcher = E.Collect(errs, idlewatcher.ValidateConfig, r.Container)
}
if !r.UseHealthCheck() && (r.UseLoadBalance() || r.UseIdleWatcher()) {
errs.Adds("healthCheck.disable cannot be true when loadbalancer or idlewatcher is enabled")
}
return errs.Error()
}
func (r *Route) Start(parent task.Parent) (err E.Error) {
switch r.Scheme {
case types.SchemeFileServer:
r.impl, err = NewFileServer(r)
case types.SchemeHTTP, types.SchemeHTTPS:
r.impl, err = NewHTTPRoute(r)
case types.SchemeTCP, types.SchemeUDP:
r.impl, err = NewStreamRoute(r)
default:
panic(fmt.Errorf("unexpected scheme %s for alias %s", r.Scheme, r.Alias))
}
if err != nil {
return nil, err
return err
}
return &Route{
impl: rt,
Type: t,
Entry: raw,
}, nil
return r.impl.Start(parent)
}
func FromEntries(provider string, entries RawEntries) (Routes, E.Error) {
b := E.NewBuilder("errors in routes")
func (r *Route) Finish(reason any) {
if r.impl == nil {
return
}
r.impl.Finish(reason)
}
routes := NewRoutes()
entries.RangeAllParallel(func(alias string, en *RawEntry) {
if en == nil {
en = new(RawEntry)
}
en.Alias = alias
en.Provider = provider
if strings.HasPrefix(alias, "x-") { // x properties
return
}
r, err := NewRoute(en)
func (r *Route) ProviderName() string {
return r.Provider
}
func (r *Route) TargetName() string {
return r.Alias
}
func (r *Route) TargetURL() *net.URL {
return r.pURL
}
func (r *Route) Type() types.RouteType {
switch r.Scheme {
case types.SchemeHTTP, types.SchemeHTTPS:
return types.RouteTypeReverseProxy
case types.SchemeTCP, types.SchemeUDP:
return types.RouteTypeStream
default:
return types.RouteTypeFileServer
}
}
func (r *Route) HealthMonitor() health.HealthMonitor {
return r.impl.HealthMonitor()
}
func (r *Route) IdlewatcherConfig() *idlewatcher.Config {
return r.idlewatcher
}
func (r *Route) HealthCheckConfig() *health.HealthCheckConfig {
return r.HealthCheck
}
func (r *Route) LoadBalanceConfig() *loadbalance.Config {
return r.LoadBalance
}
func (r *Route) HomepageConfig() *homepage.Item {
return r.Homepage
}
func (r *Route) ContainerInfo() *docker.Container {
return r.Container
}
func (r *Route) IsDocker() bool {
if r.Container == nil {
return false
}
return r.Container.ContainerID != ""
}
func (r *Route) IsZeroPort() bool {
return r.Port.Proxy == 0
}
func (r *Route) ShouldExclude() bool {
if r.Container != nil {
switch {
case err != nil:
b.Add(err.Subject(alias))
case entry.ShouldNotServe(r):
return
default:
routes.Store(alias, r)
case r.Container.IsExcluded:
return true
case r.IsZeroPort() && !r.UseIdleWatcher():
logging.Debug().Str("container", r.Container.ContainerName).Msg("route excluded")
return true
case strings.HasPrefix(r.Container.ContainerName, "buildx_"):
return true
}
})
return routes, b.Error()
} else if r.IsZeroPort() {
return true
}
if strings.HasPrefix(r.Alias, "x-") ||
strings.HasSuffix(r.Alias, "-old") {
return true
}
return false
}
func (r *Route) ShouldNotServe() bool {
if r.Container != nil && r.Container.IsDatabase && !r.Container.IsExplicit {
return true
}
return false
}
func (r *Route) UseLoadBalance() bool {
return r.LoadBalance != nil && r.LoadBalance.Link != ""
}
func (r *Route) UseIdleWatcher() bool {
return r.idlewatcher != nil && r.idlewatcher.IdleTimeout > 0
}
func (r *Route) UseHealthCheck() bool {
return !r.HealthCheck.Disable
}
func (r *Route) UseAccessLog() bool {
return r.AccessLog != nil
}
func (r *Route) Finalize() {
isDocker := r.Container != nil
cont := r.Container
if !isDocker {
cont = docker.DummyContainer
}
if r.Host == "" {
switch {
case cont.PrivateIP != "":
r.Host = cont.PrivateIP
case cont.PublicIP != "":
r.Host = cont.PublicIP
case !isDocker:
r.Host = "localhost"
}
}
lp, pp := r.Port.Listening, r.Port.Proxy
if port, ok := common.ServiceNamePortMapTCP[cont.ImageName]; ok {
if pp == 0 {
pp = port
}
if r.Scheme == "" {
r.Scheme = "tcp"
}
} else if port, ok := common.ImageNamePortMap[cont.ImageName]; ok {
if pp == 0 {
pp = port
}
if r.Scheme == "" {
r.Scheme = "http"
}
}
if pp == 0 {
switch {
case r.Scheme == "https":
pp = 443
case !isDocker:
pp = 80
default:
pp = lowestPort(cont.PrivatePortMapping)
if pp == 0 {
pp = lowestPort(cont.PublicPortMapping)
}
}
}
// replace private port with public port if using public IP.
if r.Host == cont.PublicIP {
if p, ok := cont.PrivatePortMapping[pp]; ok {
pp = int(p.PublicPort)
}
}
// replace public port with private port if using private IP.
if r.Host == cont.PrivateIP {
if p, ok := cont.PublicPortMapping[pp]; ok {
pp = int(p.PrivatePort)
}
}
if r.Scheme == "" && isDocker {
switch {
case r.Host == cont.PublicIP && cont.PublicPortMapping[pp].Type == "udp":
r.Scheme = "udp"
case r.Host == cont.PrivateIP && cont.PrivatePortMapping[pp].Type == "udp":
r.Scheme = "udp"
}
}
if r.Scheme == "" {
switch {
case lp != 0:
r.Scheme = "tcp"
case strings.HasSuffix(strconv.Itoa(pp), "443"):
r.Scheme = "https"
default: // assume its http
r.Scheme = "http"
}
}
r.Port.Listening, r.Port.Proxy = lp, pp
if r.HealthCheck == nil {
r.HealthCheck = health.DefaultHealthConfig
}
if !r.HealthCheck.Disable {
if r.HealthCheck.Interval == 0 {
r.HealthCheck.Interval = common.HealthCheckIntervalDefault
}
if r.HealthCheck.Timeout == 0 {
r.HealthCheck.Timeout = common.HealthCheckTimeoutDefault
}
}
if cont.IdleTimeout != "" {
if cont.WakeTimeout == "" {
cont.WakeTimeout = common.WakeTimeoutDefault
}
if cont.StopTimeout == "" {
cont.StopTimeout = common.StopTimeoutDefault
}
if cont.StopMethod == "" {
cont.StopMethod = common.StopMethodDefault
}
}
if r.Homepage.IsEmpty() {
r.Homepage = homepage.NewItem(r.Alias)
}
}
func lowestPort(ports map[int]dockertypes.Port) (res int) {
cmp := (uint16)(65535)
for port, v := range ports {
if v.PrivatePort < cmp {
cmp = v.PrivatePort
res = port
}
}
return
}

27
internal/route/routes/routequery/query.go
View file

@ -6,7 +6,6 @@ import (
"github.com/yusing/go-proxy/internal"
"github.com/yusing/go-proxy/internal/homepage"
"github.com/yusing/go-proxy/internal/route/entry"
provider "github.com/yusing/go-proxy/internal/route/provider/types"
"github.com/yusing/go-proxy/internal/route/routes"
route "github.com/yusing/go-proxy/internal/route/types"
@ -44,15 +43,15 @@ func HomepageCategories() []string {
check := make(map[string]struct{})
categories := make([]string, 0)
routes.GetHTTPRoutes().RangeAll(func(alias string, r route.HTTPRoute) {
en := r.RawEntry()
if en.Homepage.IsEmpty() || en.Homepage.Category == "" {
homepage := r.HomepageConfig()
if homepage.IsEmpty() || homepage.Category == "" {
return
}
if _, ok := check[en.Homepage.Category]; ok {
if _, ok := check[homepage.Category]; ok {
return
}
check[en.Homepage.Category] = struct{}{}
categories = append(categories, en.Homepage.Category)
check[homepage.Category] = struct{}{}
categories = append(categories, homepage.Category)
})
return categories
}
@ -61,8 +60,7 @@ func HomepageConfig(useDefaultCategories bool, categoryFilter, providerFilter st
hpCfg := homepage.NewHomePageConfig()
routes.GetHTTPRoutes().RangeAll(func(alias string, r route.HTTPRoute) {
en := r.RawEntry()
item := en.Homepage
item := r.HomepageConfig()
if item.IsEmpty() {
item = homepage.NewItem(alias)
@ -78,7 +76,7 @@ func HomepageConfig(useDefaultCategories bool, categoryFilter, providerFilter st
}
item.Alias = alias
item.Provider = r.RawEntry().Provider
item.Provider = r.ProviderName()
if providerFilter != "" && item.Provider != providerFilter {
return
@ -86,7 +84,7 @@ func HomepageConfig(useDefaultCategories bool, categoryFilter, providerFilter st
if item.Name == "" {
reference := r.TargetName()
cont := r.RawEntry().Container
cont := r.ContainerInfo()
if cont != nil {
reference = cont.ImageName
}
@ -104,8 +102,9 @@ func HomepageConfig(useDefaultCategories bool, categoryFilter, providerFilter st
}
if useDefaultCategories {
if en.Container != nil && item.Category == "" {
if category, ok := homepage.PredefinedCategories[en.Container.ImageName]; ok {
container := r.ContainerInfo()
if container != nil && item.Category == "" {
if category, ok := homepage.PredefinedCategories[container.ImageName]; ok {
item.Category = category
}
}
@ -122,12 +121,12 @@ func HomepageConfig(useDefaultCategories bool, categoryFilter, providerFilter st
}
switch {
case entry.IsDocker(r):
case r.IsDocker():
if item.Category == "" {
item.Category = "Docker"
}
item.SourceType = string(provider.ProviderTypeDocker)
case entry.UseLoadBalance(r):
case r.UseLoadBalance():
if item.Category == "" {
item.Category = "Load-balanced"
}

4
internal/route/rules/do.go
View file

@ -94,7 +94,7 @@ var commands = map[string]struct {
},
validate: validateURL,
build: func(args any) CommandHandler {
target := args.(types.URL).String()
target := args.(*types.URL).String()
return ReturningCommand(func(w http.ResponseWriter, r *http.Request) {
http.Redirect(w, r, target, http.StatusTemporaryRedirect)
})
@ -159,7 +159,7 @@ var commands = map[string]struct {
},
validate: validateAbsoluteURL,
build: func(args any) CommandHandler {
target := args.(types.URL)
target := args.(*types.URL)
if target.Scheme == "" {
target.Scheme = "http"
}

39
internal/route/stream.go
View file

@ -10,8 +10,8 @@ import (
E "github.com/yusing/go-proxy/internal/error"
"github.com/yusing/go-proxy/internal/logging"
net "github.com/yusing/go-proxy/internal/net/types"
"github.com/yusing/go-proxy/internal/route/entry"
"github.com/yusing/go-proxy/internal/route/routes"
route "github.com/yusing/go-proxy/internal/route/types"
"github.com/yusing/go-proxy/internal/task"
"github.com/yusing/go-proxy/internal/watcher/health"
"github.com/yusing/go-proxy/internal/watcher/health/monitor"
@ -19,7 +19,7 @@ import (
// TODO: support stream load balance.
type StreamRoute struct {
*entry.StreamEntry
*Route
net.Stream `json:"-"`
@ -30,16 +30,13 @@ type StreamRoute struct {
l zerolog.Logger
}
func NewStreamRoute(entry *entry.StreamEntry) (impl, E.Error) {
func NewStreamRoute(base *Route) (route.Route, E.Error) {
// TODO: support non-coherent scheme
if !entry.Scheme.IsCoherent() {
return nil, E.Errorf("unsupported scheme: %v -> %v", entry.Scheme.ListeningScheme, entry.Scheme.ProxyScheme)
}
return &StreamRoute{
StreamEntry: entry,
Route: base,
l: logging.With().
Str("type", string(entry.Scheme.ListeningScheme)).
Str("name", entry.TargetName()).
Str("type", string(base.Scheme)).
Str("name", base.TargetName()).
Logger(),
}, nil
}
@ -50,10 +47,6 @@ func (r *StreamRoute) String() string {
// Start implements task.TaskStarter.
func (r *StreamRoute) Start(parent task.Parent) E.Error {
if entry.ShouldNotServe(r) {
return nil
}
r.task = parent.Subtask("stream." + r.TargetName())
r.Stream = NewStream(r)
parent.OnCancel("finish", func() {
@ -61,25 +54,25 @@ func (r *StreamRoute) Start(parent task.Parent) E.Error {
})
switch {
case entry.UseIdleWatcher(r):
waker, err := idlewatcher.NewStreamWaker(parent, r.StreamEntry, r.Stream)
case r.UseIdleWatcher():
waker, err := idlewatcher.NewStreamWaker(parent, r, r.Stream)
if err != nil {
r.task.Finish(err)
return err
}
r.Stream = waker
r.HealthMon = waker
case entry.UseHealthCheck(r):
if entry.IsDocker(r) {
client, err := docker.ConnectClient(r.Idlewatcher.DockerHost)
case r.UseHealthCheck():
if r.IsDocker() {
client, err := docker.ConnectClient(r.IdlewatcherConfig().DockerHost)
if err == nil {
fallback := monitor.NewRawHealthChecker(r.TargetURL(), r.Raw.HealthCheck)
r.HealthMon = monitor.NewDockerHealthMonitor(client, r.Idlewatcher.ContainerID, r.TargetName(), r.Raw.HealthCheck, fallback)
fallback := monitor.NewRawHealthChecker(r.TargetURL(), r.HealthCheck)
r.HealthMon = monitor.NewDockerHealthMonitor(client, r.IdlewatcherConfig().ContainerID, r.TargetName(), r.HealthCheck, fallback)
r.task.OnCancel("close_docker_client", client.Close)
}
}
if r.HealthMon == nil {
r.HealthMon = monitor.NewRawHealthMonitor(r.TargetURL(), r.Raw.HealthCheck)
r.HealthMon = monitor.NewRawHealthMonitor(r.TargetURL(), r.HealthCheck)
}
}
@ -88,9 +81,7 @@ func (r *StreamRoute) Start(parent task.Parent) E.Error {
return E.From(err)
}
r.l.Info().
Int("port", int(r.Port.ListeningPort)).
Msg("listening")
r.l.Info().Int("port", r.Port.Listening).Msg("listening")
if r.HealthMon != nil {
if err := r.HealthMon.Start(r.task); err != nil {

15
internal/route/stream_impl.go
View file

@ -8,7 +8,6 @@ import (
"time"
"github.com/yusing/go-proxy/internal/net/types"
T "github.com/yusing/go-proxy/internal/route/types"
U "github.com/yusing/go-proxy/internal/utils"
)
@ -45,25 +44,25 @@ func (stream *Stream) Setup() error {
ctx := stream.task.Context()
switch stream.Scheme.ListeningScheme {
switch stream.Scheme {
case "tcp":
stream.targetAddr, err = net.ResolveTCPAddr("tcp", stream.URL.Host)
stream.targetAddr, err = net.ResolveTCPAddr("tcp", stream.pURL.Host)
if err != nil {
return err
}
tcpListener, err := lcfg.Listen(ctx, "tcp", stream.ListenURL.Host)
tcpListener, err := lcfg.Listen(ctx, "tcp", stream.lURL.Host)
if err != nil {
return err
}
// in case ListeningPort was zero, get the actual port
stream.Port.ListeningPort = T.Port(tcpListener.Addr().(*net.TCPAddr).Port)
stream.Port.Listening = tcpListener.Addr().(*net.TCPAddr).Port
stream.listener = types.NetListener(tcpListener)
case "udp":
stream.targetAddr, err = net.ResolveUDPAddr("udp", stream.URL.Host)
stream.targetAddr, err = net.ResolveUDPAddr("udp", stream.pURL.Host)
if err != nil {
return err
}
udpListener, err := lcfg.ListenPacket(ctx, "udp", stream.ListenURL.Host)
udpListener, err := lcfg.ListenPacket(ctx, "udp", stream.lURL.Host)
if err != nil {
return err
}
@ -72,7 +71,7 @@ func (stream *Stream) Setup() error {
udpListener.Close()
return errors.New("udp listener is not *net.UDPConn")
}
stream.Port.ListeningPort = T.Port(udpConn.LocalAddr().(*net.UDPAddr).Port)
stream.Port.Listening = udpConn.LocalAddr().(*net.UDPAddr).Port
stream.listener = NewUDPForwarder(ctx, udpConn, stream.targetAddr)
default:
panic("should not reach here")

13
internal/route/types/entry.go
View file

@ -1,13 +0,0 @@
package types
import (
idlewatcher "github.com/yusing/go-proxy/internal/docker/idlewatcher/types"
net "github.com/yusing/go-proxy/internal/net/types"
)
type Entry interface {
TargetName() string
TargetURL() net.URL
RawEntry() *RawEntry
IdlewatcherConfig() *idlewatcher.Config
}

19
internal/route/types/headers.go
View file

@ -1,19 +0,0 @@
package types
import (
"net/http"
E "github.com/yusing/go-proxy/internal/error"
"github.com/yusing/go-proxy/internal/utils/strutils"
)
func ValidateHTTPHeaders(headers map[string]string) (http.Header, E.Error) {
h := make(http.Header)
for k, v := range headers {
vSplit := strutils.CommaSeperatedList(v)
for _, header := range vSplit {
h.Add(k, header)
}
}
return h, nil
}

12
internal/route/types/http_config_test.go
View file

@ -1,9 +1,11 @@
package types
package types_test
import (
"testing"
"time"
. "github.com/yusing/go-proxy/internal/route"
"github.com/yusing/go-proxy/internal/route/types"
"github.com/yusing/go-proxy/internal/utils"
. "github.com/yusing/go-proxy/internal/utils/testing"
)
@ -12,14 +14,14 @@ func TestHTTPConfigDeserialize(t *testing.T) {
tests := []struct {
name string
input map[string]any
expected HTTPConfig
expected types.HTTPConfig
}{
{
name: "no_tls_verify",
input: map[string]any{
"no_tls_verify": "true",
},
expected: HTTPConfig{
expected: types.HTTPConfig{
NoTLSVerify: true,
},
},
@ -28,7 +30,7 @@ func TestHTTPConfigDeserialize(t *testing.T) {
input: map[string]any{
"response_header_timeout": "1s",
},
expected: HTTPConfig{
expected: types.HTTPConfig{
ResponseHeaderTimeout: 1 * time.Second,
},
},
@ -36,7 +38,7 @@ func TestHTTPConfigDeserialize(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
cfg := RawEntry{}
cfg := Route{}
err := utils.Deserialize(tt.input, &cfg)
if err != nil {
ExpectNoError(t, err)

58
internal/route/types/port.go
View file

@ -7,37 +7,55 @@ import (
"github.com/yusing/go-proxy/internal/utils/strutils"
)
type Port int
type Port struct {
Listening int `json:"listening"`
Proxy int `json:"proxy"`
}
var ErrPortOutOfRange = E.New("port out of range")
var (
ErrInvalidPortSyntax = E.New("invalid port syntax, expect [listening_port:]target_port")
ErrPortOutOfRange = E.New("port out of range")
)
// Parse implements strutils.Parser.
func (p *Port) Parse(v string) (err error) {
parts := strutils.SplitRune(v, ':')
switch len(parts) {
case 1:
p.Listening = 0
p.Proxy, err = strconv.Atoi(v)
case 2:
var err2 error
p.Listening, err = strconv.Atoi(parts[0])
p.Proxy, err2 = strconv.Atoi(parts[1])
err = E.Join(err, err2)
default:
return ErrInvalidPortSyntax.Subject(v)
}
func ValidatePort[String ~string](v String) (Port, error) {
p, err := strutils.Atoi(string(v))
if err != nil {
return ErrPort, err
return err
}
return ValidatePortInt(p)
}
func ValidatePortInt[Int int | uint16](v Int) (Port, error) {
p := Port(v)
if !p.inBound() {
return ErrPort, ErrPortOutOfRange.Subject(strconv.Itoa(int(p)))
if p.Listening < MinPort || p.Listening > MaxPort {
return ErrPortOutOfRange.Subjectf("%d", p.Listening)
}
return p, nil
if p.Proxy < MinPort || p.Proxy > MaxPort {
return ErrPortOutOfRange.Subjectf("%d", p.Proxy)
}
return nil
}
func (p Port) inBound() bool {
return p >= MinPort && p <= MaxPort
}
func (p Port) String() string {
return strconv.Itoa(int(p))
func (p *Port) String() string {
if p.Listening == 0 {
return strconv.Itoa(p.Proxy)
}
return strconv.Itoa(p.Listening) + ":" + strconv.Itoa(p.Proxy)
}
const (
MinPort = 0
MaxPort = 65535
ErrPort = Port(-1)
NoPort = Port(0)
)

106
internal/route/types/port_test.go Normal file
View file

@ -0,0 +1,106 @@
package types
import (
"errors"
"strconv"
"testing"
)
var invalidPorts = []string{
"",
"123:",
"0:",
":1234",
"qwerty",
"asdfgh:asdfgh",
"1234:asdfgh",
}
var tooManyColonsPorts = []string{
"1234:1234:1234",
}
var outOfRangePorts = []string{
"-1:1234",
"1234:-1",
"65536",
"0:65536",
}
func TestPortInvalid(t *testing.T) {
tests := []struct {
name string
inputs []string
wantErr error
}{
{
name: "invalid",
inputs: invalidPorts,
wantErr: strconv.ErrSyntax,
},
{
name: "too many colons",
inputs: tooManyColonsPorts,
wantErr: ErrInvalidPortSyntax,
},
{
name: "out of range",
inputs: outOfRangePorts,
wantErr: ErrPortOutOfRange,
},
}
for _, tc := range tests {
for _, input := range tc.inputs {
t.Run(tc.name, func(t *testing.T) {
p := &Port{}
err := p.Parse(input)
if !errors.Is(err, tc.wantErr) {
t.Errorf("expected error %v, got %v", tc.wantErr, err)
}
})
}
}
}
func TestPortValid(t *testing.T) {
tests := []struct {
name string
inputs string
expect Port
}{
{
name: "valid_lp",
inputs: "1234:5678",
expect: Port{
Listening: 1234,
Proxy: 5678,
},
},
{
name: "valid_p",
inputs: "5678",
expect: Port{
Listening: 0,
Proxy: 5678,
},
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
p := &Port{}
err := p.Parse(tc.inputs)
if err != nil {
t.Errorf("expected no error, got %v", err)
}
if p.Listening != tc.expect.Listening {
t.Errorf("expected listening port %d, got %d", tc.expect.Listening, p.Listening)
}
if p.Proxy != tc.expect.Proxy {
t.Errorf("expected proxy port %d, got %d", tc.expect.Proxy, p.Proxy)
}
})
}
}

221
internal/route/types/raw_entry.go
View file

@ -1,221 +0,0 @@
//nolint:goconst
package types
import (
"strconv"
"strings"
"github.com/docker/docker/api/types"
"github.com/yusing/go-proxy/internal/common"
"github.com/yusing/go-proxy/internal/docker"
"github.com/yusing/go-proxy/internal/homepage"
"github.com/yusing/go-proxy/internal/logging"
"github.com/yusing/go-proxy/internal/net/http/accesslog"
loadbalance "github.com/yusing/go-proxy/internal/net/http/loadbalancer/types"
"github.com/yusing/go-proxy/internal/route/rules"
U "github.com/yusing/go-proxy/internal/utils"
F "github.com/yusing/go-proxy/internal/utils/functional"
"github.com/yusing/go-proxy/internal/utils/strutils"
"github.com/yusing/go-proxy/internal/watcher/health"
)
type (
RawEntry struct {
_ U.NoCopy
// raw entry object before validation
// loaded from docker labels or yaml file
Alias string `json:"alias"`
Scheme string `json:"scheme,omitempty"`
Host string `json:"host,omitempty"`
Port string `json:"port,omitempty"`
HTTPConfig
PathPatterns []string `json:"path_patterns,omitempty"`
Rules rules.Rules `json:"rules,omitempty" validate:"omitempty,unique=Name"`
HealthCheck *health.HealthCheckConfig `json:"healthcheck,omitempty"`
LoadBalance *loadbalance.Config `json:"load_balance,omitempty"`
Middlewares map[string]docker.LabelMap `json:"middlewares,omitempty"`
Homepage *homepage.Item `json:"homepage,omitempty"`
AccessLog *accesslog.Config `json:"access_log,omitempty"`
/* Docker only */
Container *docker.Container `json:"container,omitempty"`
Provider string `json:"provider,omitempty"`
finalized bool
}
RawEntries = F.Map[string, *RawEntry]
)
var NewProxyEntries = F.NewMapOf[string, *RawEntry]
func (e *RawEntry) Finalize() {
if e.finalized {
return
}
isDocker := e.Container != nil
cont := e.Container
if !isDocker {
cont = docker.DummyContainer
}
if e.Host == "" {
switch {
case cont.PrivateIP != "":
e.Host = cont.PrivateIP
case cont.PublicIP != "":
e.Host = cont.PublicIP
case !isDocker:
e.Host = "localhost"
}
}
lp, pp, extra := e.splitPorts()
if port, ok := common.ServiceNamePortMapTCP[cont.ImageName]; ok {
if pp == "" {
pp = strconv.Itoa(port)
}
if e.Scheme == "" {
e.Scheme = "tcp"
}
} else if port, ok := common.ImageNamePortMap[cont.ImageName]; ok {
if pp == "" {
pp = strconv.Itoa(port)
}
if e.Scheme == "" {
e.Scheme = "http"
}
} else if pp == "" && e.Scheme == "https" {
pp = "443"
} else if pp == "" {
if p := lowestPort(cont.PrivatePortMapping); p != "" {
pp = p
} else if p := lowestPort(cont.PublicPortMapping); p != "" {
pp = p
} else if !isDocker {
pp = "80"
} else {
logging.Debug().Msg("no port found for " + e.Alias)
}
}
// replace private port with public port if using public IP.
if e.Host == cont.PublicIP {
if p, ok := cont.PrivatePortMapping[pp]; ok {
pp = strutils.PortString(p.PublicPort)
}
}
// replace public port with private port if using private IP.
if e.Host == cont.PrivateIP {
if p, ok := cont.PublicPortMapping[pp]; ok {
pp = strutils.PortString(p.PrivatePort)
}
}
if e.Scheme == "" && isDocker {
switch {
case e.Host == cont.PublicIP && cont.PublicPortMapping[pp].Type == "udp":
e.Scheme = "udp"
case e.Host == cont.PrivateIP && cont.PrivatePortMapping[pp].Type == "udp":
e.Scheme = "udp"
}
}
if e.Scheme == "" {
switch {
case lp != "":
e.Scheme = "tcp"
case strings.HasSuffix(pp, "443"):
e.Scheme = "https"
default: // assume its http
e.Scheme = "http"
}
}
if e.HealthCheck == nil {
e.HealthCheck = new(health.HealthCheckConfig)
}
if e.HealthCheck.Disable {
e.HealthCheck = nil
} else {
if e.HealthCheck.Interval == 0 {
e.HealthCheck.Interval = common.HealthCheckIntervalDefault
}
if e.HealthCheck.Timeout == 0 {
e.HealthCheck.Timeout = common.HealthCheckTimeoutDefault
}
}
if cont.IdleTimeout != "" {
if cont.WakeTimeout == "" {
cont.WakeTimeout = common.WakeTimeoutDefault
}
if cont.StopTimeout == "" {
cont.StopTimeout = common.StopTimeoutDefault
}
if cont.StopMethod == "" {
cont.StopMethod = common.StopMethodDefault
}
}
e.Port = joinPorts(lp, pp, extra)
if e.Port == "" || e.Host == "" {
if lp != "" {
e.Port = lp + ":0"
} else {
e.Port = "0"
}
}
if e.Homepage.IsEmpty() {
e.Homepage = homepage.NewItem(e.Alias)
}
e.finalized = true
}
func (e *RawEntry) splitPorts() (lp string, pp string, extra string) {
portSplit := strutils.SplitRune(e.Port, ':')
if len(portSplit) == 1 {
pp = portSplit[0]
} else {
lp = portSplit[0]
pp = portSplit[1]
if len(portSplit) > 2 {
extra = strutils.JoinRune(portSplit[2:], ':')
}
}
return
}
func joinPorts(lp string, pp string, extra string) string {
s := make([]string, 0, 3)
if lp != "" {
s = append(s, lp)
}
if pp != "" {
s = append(s, pp)
}
if extra != "" {
s = append(s, extra)
}
return strutils.JoinRune(s, ':')
}
func lowestPort(ports map[string]types.Port) string {
var cmp uint16
var res string
for port, v := range ports {
if v.PrivatePort < cmp || cmp == 0 {
cmp = v.PrivatePort
res = port
}
}
return res
}

25
internal/route/types/route.go
View file

@ -3,14 +3,37 @@ package types
import (
"net/http"
"github.com/yusing/go-proxy/internal/docker"
idlewatcher "github.com/yusing/go-proxy/internal/docker/idlewatcher/types"
"github.com/yusing/go-proxy/internal/homepage"
net "github.com/yusing/go-proxy/internal/net/types"
"github.com/yusing/go-proxy/internal/task"
"github.com/yusing/go-proxy/internal/watcher/health"
loadbalance "github.com/yusing/go-proxy/internal/net/http/loadbalancer/types"
)
type (
//nolint:interfacebloat // this is for avoiding circular imports
Route interface {
Entry
task.TaskStarter
task.TaskFinisher
ProviderName() string
TargetName() string
TargetURL() *net.URL
HealthMonitor() health.HealthMonitor
IdlewatcherConfig() *idlewatcher.Config
HealthCheckConfig() *health.HealthCheckConfig
LoadBalanceConfig() *loadbalance.Config
HomepageConfig() *homepage.Item
ContainerInfo() *docker.Container
IsDocker() bool
UseLoadBalance() bool
UseIdleWatcher() bool
UseHealthCheck() bool
UseAccessLog() bool
}
HTTPRoute interface {
Route

1
internal/route/types/route_type.go
View file

@ -5,4 +5,5 @@ type RouteType string
const (
RouteTypeStream RouteType = "stream"
RouteTypeReverseProxy RouteType = "reverse_proxy"
RouteTypeFileServer RouteType = "file_server"
)

24
internal/route/types/scheme.go
View file

@ -8,16 +8,22 @@ type Scheme string
var ErrInvalidScheme = E.New("invalid scheme")
func NewScheme(s string) (Scheme, error) {
const (
SchemeHTTP Scheme = "http"
SchemeHTTPS Scheme = "https"
SchemeTCP Scheme = "tcp"
SchemeUDP Scheme = "udp"
SchemeFileServer Scheme = "fileserver"
)
func (s Scheme) Validate() E.Error {
switch s {
case "http", "https", "tcp", "udp":
return Scheme(s), nil
case SchemeHTTP, SchemeHTTPS,
SchemeTCP, SchemeUDP, SchemeFileServer:
return nil
}
return "", ErrInvalidScheme.Subject(s)
return ErrInvalidScheme.Subject(string(s))
}
func (s Scheme) IsHTTP() bool { return s == "http" }
func (s Scheme) IsHTTPS() bool { return s == "https" }
func (s Scheme) IsTCP() bool { return s == "tcp" }
func (s Scheme) IsUDP() bool { return s == "udp" }
func (s Scheme) IsStream() bool { return s.IsTCP() || s.IsUDP() }
func (s Scheme) IsReverseProxy() bool { return s == SchemeHTTP || s == SchemeHTTPS }
func (s Scheme) IsStream() bool { return s == SchemeTCP || s == SchemeUDP }

34
internal/route/types/stream_port.go
View file

@ -1,34 +0,0 @@
package types
import (
E "github.com/yusing/go-proxy/internal/error"
"github.com/yusing/go-proxy/internal/utils/strutils"
)
type StreamPort struct {
ListeningPort Port `json:"listening"`
ProxyPort Port `json:"proxy"`
}
var ErrStreamPortTooManyColons = E.New("too many colons")
func ValidateStreamPort(p string) (StreamPort, error) {
split := strutils.SplitRune(p, ':')
switch len(split) {
case 1:
split = []string{"0", split[0]}
case 2:
break
default:
return StreamPort{}, ErrStreamPortTooManyColons.Subject(p)
}
listeningPort, lErr := ValidatePort(split[0])
proxyPort, pErr := ValidatePort(split[1])
if err := E.Join(lErr, pErr); err != nil {
return StreamPort{}, err
}
return StreamPort{listeningPort, proxyPort}, nil
}

54
internal/route/types/stream_port_test.go
View file

@ -1,54 +0,0 @@
package types
import (
"strconv"
"testing"
. "github.com/yusing/go-proxy/internal/utils/testing"
)
var validPorts = []string{
"1234:5678",
"0:2345",
"2345",
}
var invalidPorts = []string{
"",
"123:",
"0:",
":1234",
"qwerty",
"asdfgh:asdfgh",
"1234:asdfgh",
}
var outOfRangePorts = []string{
"-1:1234",
"1234:-1",
"65536",
"0:65536",
}
var tooManyColonsPorts = []string{
"1234:1234:1234",
}
func TestStreamPort(t *testing.T) {
for _, port := range validPorts {
_, err := ValidateStreamPort(port)
ExpectNoError(t, err)
}
for _, port := range invalidPorts {
_, err := ValidateStreamPort(port)
ExpectError2(t, port, strconv.ErrSyntax, err)
}
for _, port := range outOfRangePorts {
_, err := ValidateStreamPort(port)
ExpectError2(t, port, ErrPortOutOfRange, err)
}
for _, port := range tooManyColonsPorts {
_, err := ValidateStreamPort(port)
ExpectError2(t, port, ErrStreamPortTooManyColons, err)
}
}

42
internal/route/types/stream_scheme.go
View file

@ -1,42 +0,0 @@
package types
import (
E "github.com/yusing/go-proxy/internal/error"
"github.com/yusing/go-proxy/internal/utils/strutils"
)
type StreamScheme struct {
ListeningScheme Scheme `json:"listening"`
ProxyScheme Scheme `json:"proxy"`
}
func ValidateStreamScheme(s string) (*StreamScheme, error) {
ss := &StreamScheme{}
parts := strutils.SplitRune(s, ':')
if len(parts) == 1 {
parts = []string{s, s}
} else if len(parts) != 2 {
return nil, ErrInvalidScheme.Subject(s)
}
var lErr, pErr error
ss.ListeningScheme, lErr = NewScheme(parts[0])
ss.ProxyScheme, pErr = NewScheme(parts[1])
if err := E.Join(lErr, pErr); err != nil {
return nil, err
}
return ss, nil
}
func (s StreamScheme) String() string {
return string(s.ListeningScheme) + " -> " + string(s.ProxyScheme)
}
// IsCoherent checks if the ListeningScheme and ProxyScheme of the StreamScheme are equal.
//
// It returns a boolean value indicating whether the ListeningScheme and ProxyScheme are equal.
func (s StreamScheme) IsCoherent() bool {
return s.ListeningScheme == s.ProxyScheme
}

37
internal/route/types/stream_scheme_test.go
View file

@ -1,37 +0,0 @@
package types
import (
"testing"
. "github.com/yusing/go-proxy/internal/utils/testing"
)
var (
validStreamSchemes = []string{
"tcp:tcp",
"tcp:udp",
"udp:tcp",
"udp:udp",
"tcp",
"udp",
}
invalidStreamSchemes = []string{
"tcp:tcp:",
"tcp:",
":udp:",
":udp",
"top",
}
)
func TestNewStreamScheme(t *testing.T) {
for _, s := range validStreamSchemes {
_, err := ValidateStreamScheme(s)
ExpectNoError(t, err)
}
for _, s := range invalidStreamSchemes {
_, err := ValidateStreamScheme(s)
ExpectError(t, ErrInvalidScheme, err)
}
}

7
internal/watcher/health/config.go
View file

@ -2,6 +2,8 @@ package health
import (
"time"
"github.com/yusing/go-proxy/internal/common"
)
type HealthCheckConfig struct {
@ -11,3 +13,8 @@ type HealthCheckConfig struct {
Interval time.Duration `json:"interval" validate:"omitempty,min=1s"`
Timeout time.Duration `json:"timeout" validate:"omitempty,min=1s"`
}
var DefaultHealthConfig = &HealthCheckConfig{
Interval: common.HealthCheckIntervalDefault,
Timeout: common.HealthCheckTimeoutDefault,
}

4
internal/watcher/health/monitor/http.go
View file

@ -26,7 +26,7 @@ var pinger = &http.Client{
},
}
func NewHTTPHealthMonitor(url types.URL, config *health.HealthCheckConfig) *HTTPHealthMonitor {
func NewHTTPHealthMonitor(url *types.URL, config *health.HealthCheckConfig) *HTTPHealthMonitor {
mon := new(HTTPHealthMonitor)
mon.monitor = newMonitor(url, config, mon.CheckHealth)
if config.UseGet {
@ -37,7 +37,7 @@ func NewHTTPHealthMonitor(url types.URL, config *health.HealthCheckConfig) *HTTP
return mon
}
func NewHTTPHealthChecker(url types.URL, config *health.HealthCheckConfig) health.HealthChecker {
func NewHTTPHealthChecker(url *types.URL, config *health.HealthCheckConfig) health.HealthChecker {
return NewHTTPHealthMonitor(url, config)
}

4
internal/watcher/health/monitor/json.go
View file

@ -5,7 +5,7 @@ import (
"strconv"
"time"
"github.com/yusing/go-proxy/internal/net/types"
net "github.com/yusing/go-proxy/internal/net/types"
"github.com/yusing/go-proxy/internal/utils/strutils"
"github.com/yusing/go-proxy/internal/watcher/health"
)
@ -19,7 +19,7 @@ type JSONRepresentation struct {
Latency time.Duration
LastSeen time.Time
Detail string
URL types.URL
URL *net.URL
Extra map[string]any
}

10
internal/watcher/health/monitor/monitor.go
View file

@ -23,7 +23,7 @@ type (
monitor struct {
service string
config *health.HealthCheckConfig
url atomic.Value[types.URL]
url atomic.Value[*types.URL]
status atomic.Value[health.Status]
lastResult *health.HealthCheckResult
@ -39,7 +39,7 @@ type (
var ErrNegativeInterval = errors.New("negative interval")
func newMonitor(url types.URL, config *health.HealthCheckConfig, healthCheckFunc HealthCheckFunc) *monitor {
func newMonitor(url *types.URL, config *health.HealthCheckConfig, healthCheckFunc HealthCheckFunc) *monitor {
mon := &monitor{
config: config,
checkHealth: healthCheckFunc,
@ -118,12 +118,12 @@ func (mon *monitor) Finish(reason any) {
}
// UpdateURL implements HealthChecker.
func (mon *monitor) UpdateURL(url types.URL) {
func (mon *monitor) UpdateURL(url *types.URL) {
mon.url.Store(url)
}
// URL implements HealthChecker.
func (mon *monitor) URL() types.URL {
func (mon *monitor) URL() *types.URL {
return mon.url.Load()
}
@ -205,7 +205,7 @@ func (mon *monitor) checkUpdateHealth() error {
if !result.Healthy {
extras.Add("Last Seen", strutils.FormatLastSeen(GetLastSeen(mon.service)))
}
if !mon.url.Load().Nil() {
if mon.url.Load() != nil {
extras.Add("Service URL", mon.url.Load().String())
}
if result.Detail != "" {

4
internal/watcher/health/monitor/raw.go
View file

@ -15,7 +15,7 @@ type (
}
)
func NewRawHealthMonitor(url types.URL, config *health.HealthCheckConfig) *RawHealthMonitor {
func NewRawHealthMonitor(url *types.URL, config *health.HealthCheckConfig) *RawHealthMonitor {
mon := new(RawHealthMonitor)
mon.monitor = newMonitor(url, config, mon.CheckHealth)
mon.dialer = &net.Dialer{
@ -25,7 +25,7 @@ func NewRawHealthMonitor(url types.URL, config *health.HealthCheckConfig) *RawHe
return mon
}
func NewRawHealthChecker(url types.URL, config *health.HealthCheckConfig) health.HealthChecker {
func NewRawHealthChecker(url *types.URL, config *health.HealthCheckConfig) health.HealthChecker {
return NewRawHealthMonitor(url, config)
}

4
internal/watcher/health/health_checker.go → internal/watcher/health/types.go
View file

@ -30,8 +30,8 @@ type (
}
HealthChecker interface {
CheckHealth() (result *HealthCheckResult, err error)
URL() types.URL
URL() *types.URL
Config() *HealthCheckConfig
UpdateURL(url types.URL)
UpdateURL(url *types.URL)
}
)