go-gost/x
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

router: router.id is optional

Browse Source
This commit is contained in:
ginuerzh
2025-02-11 20:54:52 +08:00
parent fc12e33786
commit ef440e8b4a
5 changed files with 86 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
handler/router/associate.go
+52 -13
View File
@@ -10,6 +10,7 @@ import (
"math"
"net"
"sync"
"time"
"github.com/go-gost/core/common/bufpool"
"github.com/go-gost/core/limiter"
@@ -48,7 +49,7 @@ func (h *routerHandler) handleAssociate(ctx context.Context, conn net.Conn, netw
rid = parseRouterID(rule.Endpoint)
}
if rid.IsZero() || !rid.Equal(routerID) {
if !rid.Equal(routerID) {
resp.Status = relay.StatusHostUnreachable
resp.WriteTo(conn)
err := fmt.Errorf("no route to host %s", host)
@@ -111,13 +112,19 @@ func (h *routerHandler) handleAssociate(ctx context.Context, conn net.Conn, netw
Name: clientID,
Node: h.id,
})
ctx, cancel := context.WithCancel(ctx)
defer cancel()
go h.sdRenew(ctx, clientID, connectorID.String())
}
log.Debugf("%s/%s: router=%s, connector=%s, weight=%d established", host, network, routerID, connectorID, connectorID.Weight())
for {
b := bufpool.Get(h.md.bufferSize)
b := bufpool.Get(h.md.bufferSize)
defer bufpool.Put(b)
for {
n, err := conn.Read(b)
if err != nil {
if err == io.EOF {
@@ -130,9 +137,25 @@ func (h *routerHandler) handleAssociate(ctx context.Context, conn net.Conn, netw
}
}
func (h *routerHandler) handlePacket(ctx context.Context, data []byte, routerID relay.TunnelID, log logger.Logger) error {
defer bufpool.Put(data)
func (h *routerHandler) sdRenew(ctx context.Context, clientID string, connectorID string) {
tc := time.NewTicker(h.md.sdRenewInterval)
defer tc.Stop()
for {
select {
case <-tc.C:
h.md.sd.Renew(ctx, &sd.Service{
ID: connectorID,
Name: clientID,
Node: h.id,
})
case <-ctx.Done():
return
}
}
}
func (h *routerHandler) handlePacket(ctx context.Context, data []byte, routerID relay.TunnelID, log logger.Logger) error {
var dstIP net.IP
if waterutil.IsIPv4(data) {
header, err := ipv4.ParseHeader(data)
@@ -167,16 +190,10 @@ func (h *routerHandler) handlePacket(ctx context.Context, data []byte, routerID
rid := routerID.String()
var route *router.Route
if r := registry.RouterRegistry().Get(rid); r != nil {
route = r.GetRoute(ctx, dstIP.String(), router.IDOption(rid))
}
if route == nil && h.md.router != nil {
route = h.md.router.GetRoute(ctx, dstIP.String(), router.IDOption(rid))
}
route := h.getRoute(ctx, rid, dstIP.String())
if route == nil || route.Gateway == "" {
// no route to host, discard
return fmt.Errorf("route to host %s", dstIP)
return fmt.Errorf("no route to host %s", dstIP)
}
if log.IsLevelEnabled(logger.TraceLevel) {
@@ -218,6 +235,28 @@ func (h *routerHandler) handlePacket(ctx context.Context, data []byte, routerID
return nil
}
func (h *routerHandler) getRoute(ctx context.Context, rid string, dst string) *router.Route {
if h.md.routerCacheEnabled {
if item := h.routeCache.Get(dst); item != nil && !item.Expired() {
v, _ := item.Value().(*router.Route)
return v
}
}
var route *router.Route
if r := registry.RouterRegistry().Get(rid); r != nil {
route = r.GetRoute(ctx, dst, router.IDOption(rid))
}
if route == nil && h.md.router != nil {
route = h.md.router.GetRoute(ctx, dst, router.IDOption(rid))
}
if h.md.routerCacheEnabled {
h.routeCache.Set(dst, cache.NewItem(route, h.md.routerCacheExpiration))
}
return route
}
func (h *routerHandler) getAddrforRoute(ctx context.Context, routerID, gateway string) net.Addr {
if h.md.sd == nil {
return nil
handler/router/handler.go
+14 -18
View File
@@ -37,16 +37,17 @@ func init() {
}
type routerHandler struct {
id string
options handler.Options
pool *ConnectorPool
epConn net.PacketConn
md metadata
log logger.Logger
stats *stats_util.HandlerStats
limiter traffic.TrafficLimiter
cancel context.CancelFunc
sdCache *cache.Cache
id string
options handler.Options
pool *ConnectorPool
epConn net.PacketConn
md metadata
log logger.Logger
stats *stats_util.HandlerStats
limiter traffic.TrafficLimiter
cancel context.CancelFunc
sdCache *cache.Cache
routeCache *cache.Cache
}
func NewHandler(opts ...handler.Option) handler.Handler {
@@ -56,8 +57,9 @@ func NewHandler(opts ...handler.Option) handler.Handler {
}
return &routerHandler{
options: options,
sdCache: cache.NewCache(time.Minute),
options: options,
sdCache: cache.NewCache(time.Minute),
routeCache: cache.NewCache(time.Minute),
}
}
@@ -203,12 +205,6 @@ func (h *routerHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl
}
}
if routerID.IsZero() {
resp.Status = relay.StatusBadRequest
resp.WriteTo(conn)
return ErrRouterID
}
if user != "" {
log = log.WithFields(map[string]any{"user": user})
}
handler/router/metadata.go
+15 -2
View File
@@ -25,7 +25,11 @@ type metadata struct {
ingress ingress.Ingress
sd sd.SD
sdCacheExpiration time.Duration
router router.Router
sdRenewInterval time.Duration
router router.Router
routerCacheEnabled bool
routerCacheExpiration time.Duration
observerPeriod time.Duration
observerResetTraffic bool
@@ -45,12 +49,21 @@ func (h *routerHandler) parseMetadata(md mdata.Metadata) (err error) {
h.md.ingress = registry.IngressRegistry().Get(mdutil.GetString(md, "ingress"))
h.md.sd = registry.SDRegistry().Get(mdutil.GetString(md, "sd"))
h.md.sdCacheExpiration = mdutil.GetDuration(md, "sd.cacheExpiration")
h.md.sdCacheExpiration = mdutil.GetDuration(md, "sd.cache.expiration")
if h.md.sdCacheExpiration <= 0 {
h.md.sdCacheExpiration = defaultCacheExpiration
}
h.md.sdRenewInterval = mdutil.GetDuration(md, "sd.renewInterval")
if h.md.sdRenewInterval < time.Second {
h.md.sdRenewInterval = defaultTTL
}
h.md.router = registry.RouterRegistry().Get(mdutil.GetString(md, "router"))
h.md.routerCacheEnabled = mdutil.GetBool(md, "router.cache")
h.md.routerCacheExpiration = mdutil.GetDuration(md, "router.cache.expiration")
if h.md.routerCacheExpiration <= 0 {
h.md.routerCacheExpiration = defaultCacheExpiration
}
h.md.observerPeriod = mdutil.GetDuration(md, "observePeriod", "observer.period", "observer.observePeriod")
if h.md.observerPeriod == 0 {