XiaoMo/frp
1
0
Fork 0
mirror of https://github.com/fatedier/frp.git synced 2026-01-11 22:23:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

virtual-net: initial (#4751)

Browse Source
This commit is contained in:
fatedier
2025-04-16 16:05:54 +08:00
committed by GitHub
parent 773169e0c4
commit a78814a2e9
48 changed files with 1822 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
pkg/plugin/client/http2http.go
View File

@@ -14,13 +14,11 @@
//go:build !frps
package plugin
package client
import (
"context"
"io"
stdlog "log"
"net"
"net/http"
"net/http/httputil"
@@ -42,7 +40,7 @@ type HTTP2HTTPPlugin struct {
s *http.Server
}
func NewHTTP2HTTPPlugin(options v1.ClientPluginOptions) (Plugin, error) {
func NewHTTP2HTTPPlugin(_ PluginContext, options v1.ClientPluginOptions) (Plugin, error) {
opts := options.(*v1.HTTP2HTTPPluginOptions)
listener := NewProxyListener()
@@ -80,8 +78,8 @@ func NewHTTP2HTTPPlugin(options v1.ClientPluginOptions) (Plugin, error) {
return p, nil
}
func (p *HTTP2HTTPPlugin) Handle(_ context.Context, conn io.ReadWriteCloser, realConn net.Conn, _ *ExtraInfo) {
wrapConn := netpkg.WrapReadWriteCloserToConn(conn, realConn)
func (p *HTTP2HTTPPlugin) Handle(_ context.Context, connInfo *ConnectionInfo) {
wrapConn := netpkg.WrapReadWriteCloserToConn(connInfo.Conn, connInfo.UnderlyingConn)
_ = p.l.PutConn(wrapConn)
}

10
pkg/plugin/client/http2https.go
View File

@@ -14,14 +14,12 @@
//go:build !frps
package plugin
package client
import (
"context"
"crypto/tls"
"io"
stdlog "log"
"net"
"net/http"
"net/http/httputil"
@@ -43,7 +41,7 @@ type HTTP2HTTPSPlugin struct {
s *http.Server
}
func NewHTTP2HTTPSPlugin(options v1.ClientPluginOptions) (Plugin, error) {
func NewHTTP2HTTPSPlugin(_ PluginContext, options v1.ClientPluginOptions) (Plugin, error) {
opts := options.(*v1.HTTP2HTTPSPluginOptions)
listener := NewProxyListener()
@@ -89,8 +87,8 @@ func NewHTTP2HTTPSPlugin(options v1.ClientPluginOptions) (Plugin, error) {
return p, nil
}
func (p *HTTP2HTTPSPlugin) Handle(_ context.Context, conn io.ReadWriteCloser, realConn net.Conn, _ *ExtraInfo) {
wrapConn := netpkg.WrapReadWriteCloserToConn(conn, realConn)
func (p *HTTP2HTTPSPlugin) Handle(_ context.Context, connInfo *ConnectionInfo) {
wrapConn := netpkg.WrapReadWriteCloserToConn(connInfo.Conn, connInfo.UnderlyingConn)
_ = p.l.PutConn(wrapConn)
}

8
pkg/plugin/client/http_proxy.go
View File

@@ -14,7 +14,7 @@
//go:build !frps
package plugin
package client
import (
"bufio"
@@ -45,7 +45,7 @@ type HTTPProxy struct {
s *http.Server
}
func NewHTTPProxyPlugin(options v1.ClientPluginOptions) (Plugin, error) {
func NewHTTPProxyPlugin(_ PluginContext, options v1.ClientPluginOptions) (Plugin, error) {
opts := options.(*v1.HTTPProxyPluginOptions)
listener := NewProxyListener()
@@ -69,8 +69,8 @@ func (hp *HTTPProxy) Name() string {
return v1.PluginHTTPProxy
}
func (hp *HTTPProxy) Handle(_ context.Context, conn io.ReadWriteCloser, realConn net.Conn, _ *ExtraInfo) {
wrapConn := netpkg.WrapReadWriteCloserToConn(conn, realConn)
func (hp *HTTPProxy) Handle(_ context.Context, connInfo *ConnectionInfo) {
wrapConn := netpkg.WrapReadWriteCloserToConn(connInfo.Conn, connInfo.UnderlyingConn)
sc, rd := libnet.NewSharedConn(wrapConn)
firstBytes := make([]byte, 7)

14
pkg/plugin/client/https2http.go
View File

@@ -14,15 +14,13 @@
//go:build !frps
package plugin
package client
import (
"context"
"crypto/tls"
"fmt"
"io"
stdlog "log"
"net"
"net/http"
"net/http/httputil"
"time"
@@ -48,7 +46,7 @@ type HTTPS2HTTPPlugin struct {
s *http.Server
}
func NewHTTPS2HTTPPlugin(options v1.ClientPluginOptions) (Plugin, error) {
func NewHTTPS2HTTPPlugin(_ PluginContext, options v1.ClientPluginOptions) (Plugin, error) {
opts := options.(*v1.HTTPS2HTTPPluginOptions)
listener := NewProxyListener()
@@ -106,10 +104,10 @@ func NewHTTPS2HTTPPlugin(options v1.ClientPluginOptions) (Plugin, error) {
return p, nil
}
func (p *HTTPS2HTTPPlugin) Handle(_ context.Context, conn io.ReadWriteCloser, realConn net.Conn, extra *ExtraInfo) {
wrapConn := netpkg.WrapReadWriteCloserToConn(conn, realConn)
if extra.SrcAddr != nil {
wrapConn.SetRemoteAddr(extra.SrcAddr)
func (p *HTTPS2HTTPPlugin) Handle(_ context.Context, connInfo *ConnectionInfo) {
wrapConn := netpkg.WrapReadWriteCloserToConn(connInfo.Conn, connInfo.UnderlyingConn)
if connInfo.SrcAddr != nil {
wrapConn.SetRemoteAddr(connInfo.SrcAddr)
}
_ = p.l.PutConn(wrapConn)
}

14
pkg/plugin/client/https2https.go
View File

@@ -14,15 +14,13 @@
//go:build !frps
package plugin
package client
import (
"context"
"crypto/tls"
"fmt"
"io"
stdlog "log"
"net"
"net/http"
"net/http/httputil"
"time"
@@ -48,7 +46,7 @@ type HTTPS2HTTPSPlugin struct {
s *http.Server
}
func NewHTTPS2HTTPSPlugin(options v1.ClientPluginOptions) (Plugin, error) {
func NewHTTPS2HTTPSPlugin(_ PluginContext, options v1.ClientPluginOptions) (Plugin, error) {
opts := options.(*v1.HTTPS2HTTPSPluginOptions)
listener := NewProxyListener()
@@ -112,10 +110,10 @@ func NewHTTPS2HTTPSPlugin(options v1.ClientPluginOptions) (Plugin, error) {
return p, nil
}
func (p *HTTPS2HTTPSPlugin) Handle(_ context.Context, conn io.ReadWriteCloser, realConn net.Conn, extra *ExtraInfo) {
wrapConn := netpkg.WrapReadWriteCloserToConn(conn, realConn)
if extra.SrcAddr != nil {
wrapConn.SetRemoteAddr(extra.SrcAddr)
func (p *HTTPS2HTTPSPlugin) Handle(_ context.Context, connInfo *ConnectionInfo) {
wrapConn := netpkg.WrapReadWriteCloserToConn(connInfo.Conn, connInfo.UnderlyingConn)
if connInfo.SrcAddr != nil {
wrapConn.SetRemoteAddr(connInfo.SrcAddr)
}
_ = p.l.PutConn(wrapConn)
}

26
pkg/plugin/client/plugin.go
View File

@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package plugin
package client
import (
"context"
@@ -25,13 +25,18 @@ import (
pp "github.com/pires/go-proxyproto"
v1 "github.com/fatedier/frp/pkg/config/v1"
"github.com/fatedier/frp/pkg/vnet"
)
type PluginContext struct {
Name string
VnetController *vnet.Controller
}
// Creators is used for create plugins to handle connections.
var creators = make(map[string]CreatorFn)
// params has prefix "plugin_"
type CreatorFn func(options v1.ClientPluginOptions) (Plugin, error)
type CreatorFn func(pluginCtx PluginContext, options v1.ClientPluginOptions) (Plugin, error)
func Register(name string, fn CreatorFn) {
if _, exist := creators[name]; exist {
@@ -40,16 +45,19 @@ func Register(name string, fn CreatorFn) {
creators[name] = fn
}
func Create(name string, options v1.ClientPluginOptions) (p Plugin, err error) {
if fn, ok := creators[name]; ok {
p, err = fn(options)
func Create(pluginName string, pluginCtx PluginContext, options v1.ClientPluginOptions) (p Plugin, err error) {
if fn, ok := creators[pluginName]; ok {
p, err = fn(pluginCtx, options)
} else {
err = fmt.Errorf("plugin [%s] is not registered", name)
err = fmt.Errorf("plugin [%s] is not registered", pluginName)
}
return
}
type ExtraInfo struct {
type ConnectionInfo struct {
Conn io.ReadWriteCloser
UnderlyingConn net.Conn
ProxyProtocolHeader *pp.Header
SrcAddr net.Addr
DstAddr net.Addr
@@ -58,7 +66,7 @@ type ExtraInfo struct {
type Plugin interface {
Name() string
Handle(ctx context.Context, conn io.ReadWriteCloser, realConn net.Conn, extra *ExtraInfo)
Handle(ctx context.Context, connInfo *ConnectionInfo)
Close() error
}

11
pkg/plugin/client/socks5.go
View File

@@ -14,13 +14,12 @@
//go:build !frps
package plugin
package client
import (
"context"
"io"
"log"
"net"
gosocks5 "github.com/armon/go-socks5"
@@ -36,7 +35,7 @@ type Socks5Plugin struct {
Server *gosocks5.Server
}
func NewSocks5Plugin(options v1.ClientPluginOptions) (p Plugin, err error) {
func NewSocks5Plugin(_ PluginContext, options v1.ClientPluginOptions) (p Plugin, err error) {
opts := options.(*v1.Socks5PluginOptions)
cfg := &gosocks5.Config{
@@ -51,9 +50,9 @@ func NewSocks5Plugin(options v1.ClientPluginOptions) (p Plugin, err error) {
return
}
func (sp *Socks5Plugin) Handle(_ context.Context, conn io.ReadWriteCloser, realConn net.Conn, _ *ExtraInfo) {
defer conn.Close()
wrapConn := netpkg.WrapReadWriteCloserToConn(conn, realConn)
func (sp *Socks5Plugin) Handle(_ context.Context, connInfo *ConnectionInfo) {
defer connInfo.Conn.Close()
wrapConn := netpkg.WrapReadWriteCloserToConn(connInfo.Conn, connInfo.UnderlyingConn)
_ = sp.Server.ServeConn(wrapConn)
}

10
pkg/plugin/client/static_file.go
View File

@@ -14,12 +14,10 @@
//go:build !frps
package plugin
package client
import (
"context"
"io"
"net"
"net/http"
"time"
@@ -40,7 +38,7 @@ type StaticFilePlugin struct {
s *http.Server
}
func NewStaticFilePlugin(options v1.ClientPluginOptions) (Plugin, error) {
func NewStaticFilePlugin(_ PluginContext, options v1.ClientPluginOptions) (Plugin, error) {
opts := options.(*v1.StaticFilePluginOptions)
listener := NewProxyListener()
@@ -70,8 +68,8 @@ func NewStaticFilePlugin(options v1.ClientPluginOptions) (Plugin, error) {
return sp, nil
}
func (sp *StaticFilePlugin) Handle(_ context.Context, conn io.ReadWriteCloser, realConn net.Conn, _ *ExtraInfo) {
wrapConn := netpkg.WrapReadWriteCloserToConn(conn, realConn)
func (sp *StaticFilePlugin) Handle(_ context.Context, connInfo *ConnectionInfo) {
wrapConn := netpkg.WrapReadWriteCloserToConn(connInfo.Conn, connInfo.UnderlyingConn)
_ = sp.l.PutConn(wrapConn)
}

9
pkg/plugin/client/tls2raw.go
View File

@@ -14,12 +14,11 @@
//go:build !frps
package plugin
package client
import (
"context"
"crypto/tls"
"io"
"net"
libio "github.com/fatedier/golib/io"
@@ -40,7 +39,7 @@ type TLS2RawPlugin struct {
tlsConfig *tls.Config
}
func NewTLS2RawPlugin(options v1.ClientPluginOptions) (Plugin, error) {
func NewTLS2RawPlugin(_ PluginContext, options v1.ClientPluginOptions) (Plugin, error) {
opts := options.(*v1.TLS2RawPluginOptions)
p := &TLS2RawPlugin{
@@ -55,10 +54,10 @@ func NewTLS2RawPlugin(options v1.ClientPluginOptions) (Plugin, error) {
return p, nil
}
func (p *TLS2RawPlugin) Handle(ctx context.Context, conn io.ReadWriteCloser, realConn net.Conn, _ *ExtraInfo) {
func (p *TLS2RawPlugin) Handle(ctx context.Context, connInfo *ConnectionInfo) {
xl := xlog.FromContextSafe(ctx)
wrapConn := netpkg.WrapReadWriteCloserToConn(conn, realConn)
wrapConn := netpkg.WrapReadWriteCloserToConn(connInfo.Conn, connInfo.UnderlyingConn)
tlsConn := tls.Server(wrapConn, p.tlsConfig)
if err := tlsConn.Handshake(); err != nil {

13
pkg/plugin/client/unix_domain_socket.go
View File

@@ -14,11 +14,10 @@
//go:build !frps
package plugin
package client
import (
"context"
"io"
"net"
libio "github.com/fatedier/golib/io"
@@ -35,7 +34,7 @@ type UnixDomainSocketPlugin struct {
UnixAddr *net.UnixAddr
}
func NewUnixDomainSocketPlugin(options v1.ClientPluginOptions) (p Plugin, err error) {
func NewUnixDomainSocketPlugin(_ PluginContext, options v1.ClientPluginOptions) (p Plugin, err error) {
opts := options.(*v1.UnixDomainSocketPluginOptions)
unixAddr, errRet := net.ResolveUnixAddr("unix", opts.UnixPath)
@@ -50,20 +49,20 @@ func NewUnixDomainSocketPlugin(options v1.ClientPluginOptions) (p Plugin, err er
return
}
func (uds *UnixDomainSocketPlugin) Handle(ctx context.Context, conn io.ReadWriteCloser, _ net.Conn, extra *ExtraInfo) {
func (uds *UnixDomainSocketPlugin) Handle(ctx context.Context, connInfo *ConnectionInfo) {
xl := xlog.FromContextSafe(ctx)
localConn, err := net.DialUnix("unix", nil, uds.UnixAddr)
if err != nil {
xl.Warnf("dial to uds %s error: %v", uds.UnixAddr, err)
return
}
if extra.ProxyProtocolHeader != nil {
if _, err := extra.ProxyProtocolHeader.WriteTo(localConn); err != nil {
if connInfo.ProxyProtocolHeader != nil {
if _, err := connInfo.ProxyProtocolHeader.WriteTo(localConn); err != nil {
return
}
}
libio.Join(localConn, conn)
libio.Join(localConn, connInfo.Conn)
}
func (uds *UnixDomainSocketPlugin) Name() string {

71
pkg/plugin/client/virtual_net.go Normal file
View File

@@ -0,0 +1,71 @@
// Copyright 2025 The frp Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//go:build !frps
package client
import (
"context"
v1 "github.com/fatedier/frp/pkg/config/v1"
"github.com/fatedier/frp/pkg/util/xlog"
)
func init() {
Register(v1.PluginVirtualNet, NewVirtualNetPlugin)
}
type VirtualNetPlugin struct {
pluginCtx PluginContext
opts *v1.VirtualNetPluginOptions
}
func NewVirtualNetPlugin(pluginCtx PluginContext, options v1.ClientPluginOptions) (Plugin, error) {
opts := options.(*v1.VirtualNetPluginOptions)
p := &VirtualNetPlugin{
pluginCtx: pluginCtx,
opts: opts,
}
return p, nil
}
func (p *VirtualNetPlugin) Handle(ctx context.Context, connInfo *ConnectionInfo) {
xl := xlog.FromContextSafe(ctx)
// Verify if virtual network controller is available
if p.pluginCtx.VnetController == nil {
return
}
// Register the connection with the controller
routeName := p.pluginCtx.Name
err := p.pluginCtx.VnetController.RegisterServerConn(ctx, routeName, connInfo.Conn)
if err != nil {
xl.Errorf("virtual net failed to register server connection: %v", err)
return
}
}
func (p *VirtualNetPlugin) Name() string {
return v1.PluginVirtualNet
}
func (p *VirtualNetPlugin) Close() error {
if p.pluginCtx.VnetController != nil {
p.pluginCtx.VnetController.UnregisterServerConn(p.pluginCtx.Name)
}
return nil
}

2
pkg/plugin/server/http.go
View File

@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package plugin
package server
import (
"bytes"

2
pkg/plugin/server/manager.go
View File

@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package plugin
package server
import (
"context"

2
pkg/plugin/server/plugin.go
View File

@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package plugin
package server
import (
"context"

2
pkg/plugin/server/tracer.go
View File

@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package plugin
package server
import (
"context"

2
pkg/plugin/server/types.go
View File

@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package plugin
package server
import (
"github.com/fatedier/frp/pkg/msg"

58
pkg/plugin/visitor/plugin.go Normal file
View File

@@ -0,0 +1,58 @@
// Copyright 2025 The frp Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package visitor
import (
"context"
"fmt"
"net"
v1 "github.com/fatedier/frp/pkg/config/v1"
"github.com/fatedier/frp/pkg/vnet"
)
type PluginContext struct {
Name string
Ctx context.Context
VnetController *vnet.Controller
HandleConn func(net.Conn)
}
// Creators is used for create plugins to handle connections.
var creators = make(map[string]CreatorFn)
type CreatorFn func(pluginCtx PluginContext, options v1.VisitorPluginOptions) (Plugin, error)
func Register(name string, fn CreatorFn) {
if _, exist := creators[name]; exist {
panic(fmt.Sprintf("plugin [%s] is already registered", name))
}
creators[name] = fn
}
func Create(pluginName string, pluginCtx PluginContext, options v1.VisitorPluginOptions) (p Plugin, err error) {
if fn, ok := creators[pluginName]; ok {
p, err = fn(pluginCtx, options)
} else {
err = fmt.Errorf("plugin [%s] is not registered", pluginName)
}
return
}
type Plugin interface {
Name() string
Start()
Close() error
}

232
pkg/plugin/visitor/virtual_net.go Normal file
View File

@@ -0,0 +1,232 @@
// Copyright 2025 The frp Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//go:build !frps
package visitor
import (
"context"
"errors"
"fmt"
"net"
"sync"
"time"
v1 "github.com/fatedier/frp/pkg/config/v1"
netutil "github.com/fatedier/frp/pkg/util/net"
"github.com/fatedier/frp/pkg/util/xlog"
)
func init() {
Register(v1.VisitorPluginVirtualNet, NewVirtualNetPlugin)
}
type VirtualNetPlugin struct {
pluginCtx PluginContext
routes []net.IPNet
mu sync.Mutex
controllerConn net.Conn
closeSignal chan struct{}
ctx context.Context
cancel context.CancelFunc
}
func NewVirtualNetPlugin(pluginCtx PluginContext, options v1.VisitorPluginOptions) (Plugin, error) {
opts := options.(*v1.VirtualNetVisitorPluginOptions)
p := &VirtualNetPlugin{
pluginCtx: pluginCtx,
routes: make([]net.IPNet, 0),
}
p.ctx, p.cancel = context.WithCancel(pluginCtx.Ctx)
if opts.DestinationIP == "" {
return nil, errors.New("destinationIP is required")
}
// Parse DestinationIP as a single IP and create a host route
ip := net.ParseIP(opts.DestinationIP)
if ip == nil {
return nil, fmt.Errorf("invalid destination IP address [%s]", opts.DestinationIP)
}
var mask net.IPMask
if ip.To4() != nil {
mask = net.CIDRMask(32, 32) // /32 for IPv4
} else {
mask = net.CIDRMask(128, 128) // /128 for IPv6
}
p.routes = append(p.routes, net.IPNet{IP: ip, Mask: mask})
return p, nil
}
func (p *VirtualNetPlugin) Name() string {
return v1.VisitorPluginVirtualNet
}
func (p *VirtualNetPlugin) Start() {
xl := xlog.FromContextSafe(p.pluginCtx.Ctx)
if p.pluginCtx.VnetController == nil {
return
}
routeStr := "unknown"
if len(p.routes) > 0 {
routeStr = p.routes[0].String()
}
xl.Infof("Starting VirtualNetPlugin for visitor [%s], attempting to register routes for %s", p.pluginCtx.Name, routeStr)
go p.run()
}
func (p *VirtualNetPlugin) run() {
xl := xlog.FromContextSafe(p.ctx)
reconnectDelay := 10 * time.Second
for {
// Create a signal channel for this connection attempt
currentCloseSignal := make(chan struct{})
// Store the signal channel under lock
p.mu.Lock()
p.closeSignal = currentCloseSignal
p.mu.Unlock()
select {
case <-p.ctx.Done():
xl.Infof("VirtualNetPlugin run loop for visitor [%s] stopping (context cancelled before pipe creation).", p.pluginCtx.Name)
// Ensure controllerConn from previous loop is cleaned up if necessary
p.cleanupControllerConn(xl)
return
default:
}
controllerConn, pluginConn := net.Pipe()
// Store controllerConn under lock for cleanup purposes
p.mu.Lock()
p.controllerConn = controllerConn
p.mu.Unlock()
// Wrap pluginConn using CloseNotifyConn
pluginNotifyConn := netutil.WrapCloseNotifyConn(pluginConn, func() {
close(currentCloseSignal) // Signal the run loop
})
xl.Infof("Attempting to register client route for visitor [%s]", p.pluginCtx.Name)
err := p.pluginCtx.VnetController.RegisterClientRoute(p.ctx, p.pluginCtx.Name, p.routes, controllerConn)
if err != nil {
xl.Errorf("Failed to register client route for visitor [%s]: %v. Retrying after %v", p.pluginCtx.Name, err, reconnectDelay)
p.cleanupPipePair(xl, controllerConn, pluginConn) // Close both ends on registration failure
// Wait before retrying registration, unless context is cancelled
select {
case <-time.After(reconnectDelay):
continue // Retry the loop
case <-p.ctx.Done():
xl.Infof("VirtualNetPlugin registration retry wait interrupted for visitor [%s]", p.pluginCtx.Name)
return // Exit loop if context is cancelled during wait
}
}
xl.Infof("Successfully registered client route for visitor [%s]. Starting connection handler with CloseNotifyConn.", p.pluginCtx.Name)
// Pass the CloseNotifyConn to HandleConn.
// HandleConn is responsible for calling Close() on pluginNotifyConn.
p.pluginCtx.HandleConn(pluginNotifyConn)
// Wait for either the plugin context to be cancelled or the wrapper's Close() to be called via the signal channel.
select {
case <-p.ctx.Done():
xl.Infof("VirtualNetPlugin run loop stopping for visitor [%s] (context cancelled while waiting).", p.pluginCtx.Name)
// Context cancelled, ensure controller side is closed if HandleConn didn't close its side yet.
p.cleanupControllerConn(xl)
return
case <-currentCloseSignal:
xl.Infof("Detected connection closed via CloseNotifyConn for visitor [%s].", p.pluginCtx.Name)
// HandleConn closed the plugin side (pluginNotifyConn). The closeFn was called, closing currentCloseSignal.
// We still need to close the controller side.
p.cleanupControllerConn(xl)
// Add a delay before attempting to reconnect, respecting context cancellation.
xl.Infof("Waiting %v before attempting reconnection for visitor [%s]...", reconnectDelay, p.pluginCtx.Name)
select {
case <-time.After(reconnectDelay):
// Delay completed, loop will continue.
case <-p.ctx.Done():
xl.Infof("VirtualNetPlugin reconnection delay interrupted for visitor [%s]", p.pluginCtx.Name)
return // Exit loop if context is cancelled during wait
}
// Loop will continue to reconnect.
}
// Loop will restart, context check at the beginning of the loop is sufficient.
xl.Infof("Re-establishing virtual connection for visitor [%s]...", p.pluginCtx.Name)
}
}
// cleanupControllerConn closes the current controllerConn (if it exists) under lock.
func (p *VirtualNetPlugin) cleanupControllerConn(xl *xlog.Logger) {
p.mu.Lock()
defer p.mu.Unlock()
if p.controllerConn != nil {
xl.Debugf("Cleaning up controllerConn for visitor [%s]", p.pluginCtx.Name)
p.controllerConn.Close()
p.controllerConn = nil
}
// Also clear the closeSignal reference for the completed/cancelled connection attempt
p.closeSignal = nil
}
// cleanupPipePair closes both ends of a pipe, used typically when registration fails.
func (p *VirtualNetPlugin) cleanupPipePair(xl *xlog.Logger, controllerConn, pluginConn net.Conn) {
xl.Debugf("Cleaning up pipe pair for visitor [%s] after registration failure", p.pluginCtx.Name)
controllerConn.Close()
pluginConn.Close()
p.mu.Lock()
p.controllerConn = nil // Ensure field is nil if it was briefly set
p.closeSignal = nil // Ensure field is nil if it was briefly set
p.mu.Unlock()
}
// Close initiates the plugin shutdown.
func (p *VirtualNetPlugin) Close() error {
xl := xlog.FromContextSafe(p.pluginCtx.Ctx) // Use base context for close logging
xl.Infof("Closing VirtualNetPlugin for visitor [%s]", p.pluginCtx.Name)
// 1. Signal the run loop goroutine to stop via context cancellation.
p.cancel()
// 2. Unregister the route from the controller.
// This might implicitly cause the VnetController to close its end of the pipe (controllerConn).
if p.pluginCtx.VnetController != nil {
p.pluginCtx.VnetController.UnregisterClientRoute(p.pluginCtx.Name)
xl.Infof("Unregistered client route for visitor [%s]", p.pluginCtx.Name)
} else {
xl.Warnf("VnetController is nil during close for visitor [%s], cannot unregister route", p.pluginCtx.Name)
}
// 3. Explicitly close the controller side of the pipe managed by this plugin.
// This ensures the pipe is broken even if the run loop is stuck or HandleConn hasn't closed its end.
p.cleanupControllerConn(xl)
xl.Infof("Finished cleaning up connections during close for visitor [%s]", p.pluginCtx.Name)
return nil
}