Merge pull request #795 from fatedier/lb

support lb
This commit is contained in:
fatedier 2018-05-26 22:42:03 +08:00 committed by GitHub
commit 85dd41c17b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 452 additions and 55 deletions

View File

@ -35,6 +35,7 @@ frp is a fast reverse proxy to help you expose a local server behind a NAT or fi
* [TCP Stream Multiplexing](#tcp-stream-multiplexing)
* [Support KCP Protocol](#support-kcp-protocol)
* [Connection Pool](#connection-pool)
* [Load balancing](#load-balancing)
* [Rewriting the Host Header](#rewriting-the-host-header)
* [Set Headers In HTTP Request](#set-headers-in-http-request)
* [Get Real IP](#get-real-ip)
@ -484,6 +485,32 @@ This feature is fit for a large number of short connections.
pool_count = 1
```
### Load balancing
Load balancing is supported by `group`.
This feature is available only for type `tcp` now.
```ini
# frpc.ini
[test1]
type = tcp
local_port = 8080
remote_port = 80
group = web
group_key = 123
[test2]
type = tcp
local_port = 8081
remote_port = 80
group = web
group_key = 123
```
`group_key` is used for authentication.
Proxies in same group will accept connections from port 80 randomly.
### Rewriting the Host Header
When forwarding to a local port, frp does not modify the tunneled HTTP requests at all, they are copied to your server byte-for-byte as they are received. Some application servers use the Host header for determining which development site to display. For this reason, frp can rewrite your requests with a modified host header. Use the `host_header_rewrite` switch to rewrite incoming HTTP requests.
@ -644,7 +671,6 @@ plugin_http_passwd = abc
* Log http request information in frps.
* Direct reverse proxy, like haproxy.
* Load balance to different service in frpc.
* kubernetes ingress support.
## Contributing

View File

@ -33,8 +33,9 @@ frp 是一个可用于内网穿透的高性能的反向代理应用,支持 tcp
* [TCP 多路复用](#tcp-多路复用)
* [底层通信可选 kcp 协议](#底层通信可选-kcp-协议)
* [连接池](#连接池)
* [负载均衡](#负载均衡)
* [修改 Host Header](#修改-host-header)
* [设置 http 请求的 header](#设置-http-请求的-header)
* [设置 HTTP 请求的 header](#设置-http-请求的-header)
* [获取用户真实 IP](#获取用户真实-ip)
* [通过密码保护你的 web 服务](#通过密码保护你的-web-服务)
* [自定义二级域名](#自定义二级域名)
@ -511,6 +512,32 @@ tcp_mux = false
pool_count = 1
```
### 负载均衡
可以将多个相同类型的 proxy 加入到同一个 group 中,从而实现负载均衡的功能。
目前只支持 tcp 类型的 proxy。
```ini
# fprc.ini
[test1]
type = tcp
local_port = 8080
remote_port = 80
group = web
group_key = 123
[test2]
type = tcp
local_port = 8081
remote_port = 80
group = web
group_key = 123
```
用户连接 frps 服务器的 80 端口frps 会将接收到的用户连接随机分发给其中一个存活的 proxy。这样可以在一台 frpc 机器挂掉后仍然有其他节点能够提供服务。
要求 `group_key` 相同,做权限验证,且 `remote_port` 相同。
### 修改 Host Header
通常情况下 frp 不会修改转发的任何数据。但有一些后端服务会根据 http 请求 header 中的 host 字段来展现不同的网站,例如 nginx 的虚拟主机服务,启用 host-header 的修改功能可以动态修改 http 请求中的 host 字段。该功能仅限于 http 类型的代理。
@ -526,7 +553,7 @@ host_header_rewrite = dev.yourdomain.com
原来 http 请求中的 host 字段 `test.yourdomain.com` 转发到后端服务时会被替换为 `dev.yourdomain.com`
### 设置 http 请求的 header
### 设置 HTTP 请求的 header
对于 `type = http` 的代理,可以设置在转发中动态添加的 header 参数。
@ -684,7 +711,6 @@ plugin_http_passwd = abc
* frps 记录 http 请求日志。
* frps 支持直接反向代理,类似 haproxy。
* frpc 支持负载均衡到后端不同服务。
* 集成对 k8s 等平台的支持。
## 为 frp 做贡献

View File

@ -45,7 +45,7 @@ login_fail_exit = true
protocol = tcp
# specify a dns server, so frpc will use this instead of default one
dns_server = 8.8.8.8
# dns_server = 8.8.8.8
# proxy names you want to start divided by ','
# default is empty, means all proxies
@ -69,6 +69,10 @@ use_encryption = false
use_compression = false
# remote port listen by frps
remote_port = 6001
# frps will load balancing connections for proxies in same group
group = test_group
# group should have same group key
group_key = 123456
[ssh_random]
type = tcp

View File

@ -100,8 +100,10 @@ type BaseProxyConf struct {
ProxyName string `json:"proxy_name"`
ProxyType string `json:"proxy_type"`
UseEncryption bool `json:"use_encryption"`
UseCompression bool `json:"use_compression"`
UseEncryption bool `json:"use_encryption"`
UseCompression bool `json:"use_compression"`
Group string `json:"group"`
GroupKey string `json:"group_key"`
}
func (cfg *BaseProxyConf) GetBaseInfo() *BaseProxyConf {
@ -112,7 +114,9 @@ func (cfg *BaseProxyConf) compare(cmp *BaseProxyConf) bool {
if cfg.ProxyName != cmp.ProxyName ||
cfg.ProxyType != cmp.ProxyType ||
cfg.UseEncryption != cmp.UseEncryption ||
cfg.UseCompression != cmp.UseCompression {
cfg.UseCompression != cmp.UseCompression ||
cfg.Group != cmp.Group ||
cfg.GroupKey != cmp.GroupKey {
return false
}
return true
@ -123,6 +127,8 @@ func (cfg *BaseProxyConf) UnmarshalFromMsg(pMsg *msg.NewProxy) {
cfg.ProxyType = pMsg.ProxyType
cfg.UseEncryption = pMsg.UseEncryption
cfg.UseCompression = pMsg.UseCompression
cfg.Group = pMsg.Group
cfg.GroupKey = pMsg.GroupKey
}
func (cfg *BaseProxyConf) UnmarshalFromIni(prefix string, name string, section ini.Section) error {
@ -142,6 +148,9 @@ func (cfg *BaseProxyConf) UnmarshalFromIni(prefix string, name string, section i
if ok && tmpStr == "true" {
cfg.UseCompression = true
}
cfg.Group = section["group"]
cfg.GroupKey = section["group_key"]
return nil
}
@ -150,6 +159,8 @@ func (cfg *BaseProxyConf) MarshalToMsg(pMsg *msg.NewProxy) {
pMsg.ProxyType = cfg.ProxyType
pMsg.UseEncryption = cfg.UseEncryption
pMsg.UseCompression = cfg.UseCompression
pMsg.Group = cfg.Group
pMsg.GroupKey = cfg.GroupKey
}
// Bind info

View File

@ -86,6 +86,8 @@ type NewProxy struct {
ProxyType string `json:"proxy_type"`
UseEncryption bool `json:"use_encryption"`
UseCompression bool `json:"use_compression"`
Group string `json:"group"`
GroupKey string `json:"group_key"`
// tcp and udp only
RemotePort int `json:"remote_port"`

25
server/group/group.go Normal file
View File

@ -0,0 +1,25 @@
// Copyright 2018 fatedier, fatedier@gmail.com
//
// 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 group
import (
"errors"
)
var (
ErrGroupAuthFailed = errors.New("group auth failed")
ErrGroupParamsInvalid = errors.New("group params invalid")
ErrListenerClosed = errors.New("group listener closed")
)

200
server/group/tcp.go Normal file
View File

@ -0,0 +1,200 @@
// Copyright 2018 fatedier, fatedier@gmail.com
//
// 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 group
import (
"fmt"
"net"
"sync"
"github.com/fatedier/frp/server/ports"
gerr "github.com/fatedier/golib/errors"
)
type TcpGroupListener struct {
groupName string
group *TcpGroup
addr net.Addr
closeCh chan struct{}
}
func newTcpGroupListener(name string, group *TcpGroup, addr net.Addr) *TcpGroupListener {
return &TcpGroupListener{
groupName: name,
group: group,
addr: addr,
closeCh: make(chan struct{}),
}
}
func (ln *TcpGroupListener) Accept() (c net.Conn, err error) {
var ok bool
select {
case <-ln.closeCh:
return nil, ErrListenerClosed
case c, ok = <-ln.group.Accept():
if !ok {
return nil, ErrListenerClosed
}
return c, nil
}
}
func (ln *TcpGroupListener) Addr() net.Addr {
return ln.addr
}
func (ln *TcpGroupListener) Close() (err error) {
close(ln.closeCh)
ln.group.CloseListener(ln)
return
}
type TcpGroup struct {
group string
groupKey string
addr string
port int
realPort int
acceptCh chan net.Conn
index uint64
tcpLn net.Listener
lns []*TcpGroupListener
ctl *TcpGroupCtl
mu sync.Mutex
}
func NewTcpGroup(ctl *TcpGroupCtl) *TcpGroup {
return &TcpGroup{
lns: make([]*TcpGroupListener, 0),
ctl: ctl,
acceptCh: make(chan net.Conn),
}
}
func (tg *TcpGroup) Listen(proxyName string, group string, groupKey string, addr string, port int) (ln *TcpGroupListener, realPort int, err error) {
tg.mu.Lock()
defer tg.mu.Unlock()
if len(tg.lns) == 0 {
realPort, err = tg.ctl.portManager.Acquire(proxyName, port)
if err != nil {
return
}
tcpLn, errRet := net.Listen("tcp", fmt.Sprintf("%s:%d", addr, port))
if errRet != nil {
err = errRet
return
}
ln = newTcpGroupListener(group, tg, tcpLn.Addr())
tg.group = group
tg.groupKey = groupKey
tg.addr = addr
tg.port = port
tg.realPort = realPort
tg.tcpLn = tcpLn
tg.lns = append(tg.lns, ln)
if tg.acceptCh == nil {
tg.acceptCh = make(chan net.Conn)
}
go tg.worker()
} else {
if tg.group != group || tg.addr != addr || tg.port != port {
err = ErrGroupParamsInvalid
return
}
if tg.groupKey != groupKey {
err = ErrGroupAuthFailed
return
}
ln = newTcpGroupListener(group, tg, tg.lns[0].Addr())
realPort = tg.realPort
tg.lns = append(tg.lns, ln)
}
return
}
func (tg *TcpGroup) worker() {
for {
c, err := tg.tcpLn.Accept()
if err != nil {
return
}
err = gerr.PanicToError(func() {
tg.acceptCh <- c
})
if err != nil {
return
}
}
}
func (tg *TcpGroup) Accept() <-chan net.Conn {
return tg.acceptCh
}
func (tg *TcpGroup) CloseListener(ln *TcpGroupListener) {
tg.mu.Lock()
defer tg.mu.Unlock()
for i, tmpLn := range tg.lns {
if tmpLn == ln {
tg.lns = append(tg.lns[:i], tg.lns[i+1:]...)
break
}
}
if len(tg.lns) == 0 {
close(tg.acceptCh)
tg.tcpLn.Close()
tg.ctl.portManager.Release(tg.realPort)
tg.ctl.RemoveGroup(tg.group)
}
}
type TcpGroupCtl struct {
groups map[string]*TcpGroup
portManager *ports.PortManager
mu sync.Mutex
}
func NewTcpGroupCtl(portManager *ports.PortManager) *TcpGroupCtl {
return &TcpGroupCtl{
groups: make(map[string]*TcpGroup),
portManager: portManager,
}
}
func (tgc *TcpGroupCtl) Listen(proxyNanme string, group string, groupKey string,
addr string, port int) (l net.Listener, realPort int, err error) {
tgc.mu.Lock()
defer tgc.mu.Unlock()
if tcpGroup, ok := tgc.groups[group]; ok {
return tcpGroup.Listen(proxyNanme, group, groupKey, addr, port)
} else {
tcpGroup = NewTcpGroup(tgc)
tgc.groups[group] = tcpGroup
return tcpGroup.Listen(proxyNanme, group, groupKey, addr, port)
}
}
func (tgc *TcpGroupCtl) RemoveGroup(group string) {
tgc.mu.Lock()
defer tgc.mu.Unlock()
delete(tgc.groups, group)
}

View File

@ -1,4 +1,4 @@
package server
package ports
import (
"errors"

View File

@ -181,27 +181,44 @@ type TcpProxy struct {
}
func (pxy *TcpProxy) Run() (remoteAddr string, err error) {
pxy.realPort, err = pxy.ctl.svr.tcpPortManager.Acquire(pxy.name, pxy.cfg.RemotePort)
if err != nil {
return
}
defer func() {
if err != nil {
pxy.ctl.svr.tcpPortManager.Release(pxy.realPort)
if pxy.cfg.Group != "" {
l, realPort, errRet := pxy.ctl.svr.tcpGroupCtl.Listen(pxy.name, pxy.cfg.Group, pxy.cfg.GroupKey, g.GlbServerCfg.ProxyBindAddr, pxy.cfg.RemotePort)
if errRet != nil {
err = errRet
return
}
}()
remoteAddr = fmt.Sprintf(":%d", pxy.realPort)
pxy.cfg.RemotePort = pxy.realPort
listener, errRet := frpNet.ListenTcp(g.GlbServerCfg.ProxyBindAddr, pxy.realPort)
if errRet != nil {
err = errRet
return
defer func() {
if err != nil {
l.Close()
}
}()
pxy.realPort = realPort
listener := frpNet.WrapLogListener(l)
listener.AddLogPrefix(pxy.name)
pxy.listeners = append(pxy.listeners, listener)
pxy.Info("tcp proxy listen port [%d] in group [%s]", pxy.cfg.RemotePort, pxy.cfg.Group)
} else {
pxy.realPort, err = pxy.ctl.svr.tcpPortManager.Acquire(pxy.name, pxy.cfg.RemotePort)
if err != nil {
return
}
defer func() {
if err != nil {
pxy.ctl.svr.tcpPortManager.Release(pxy.realPort)
}
}()
listener, errRet := frpNet.ListenTcp(g.GlbServerCfg.ProxyBindAddr, pxy.realPort)
if errRet != nil {
err = errRet
return
}
listener.AddLogPrefix(pxy.name)
pxy.listeners = append(pxy.listeners, listener)
pxy.Info("tcp proxy listen port [%d]", pxy.cfg.RemotePort)
}
listener.AddLogPrefix(pxy.name)
pxy.listeners = append(pxy.listeners, listener)
pxy.Info("tcp proxy listen port [%d]", pxy.cfg.RemotePort)
pxy.cfg.RemotePort = pxy.realPort
remoteAddr = fmt.Sprintf(":%d", pxy.realPort)
pxy.startListenHandler(pxy, HandleUserTcpConnection)
return
}
@ -212,7 +229,9 @@ func (pxy *TcpProxy) GetConf() config.ProxyConf {
func (pxy *TcpProxy) Close() {
pxy.BaseProxy.Close()
pxy.ctl.svr.tcpPortManager.Release(pxy.realPort)
if pxy.cfg.Group == "" {
pxy.ctl.svr.tcpPortManager.Release(pxy.realPort)
}
}
type HttpProxy struct {

View File

@ -24,6 +24,8 @@ import (
"github.com/fatedier/frp/assets"
"github.com/fatedier/frp/g"
"github.com/fatedier/frp/models/msg"
"github.com/fatedier/frp/server/group"
"github.com/fatedier/frp/server/ports"
"github.com/fatedier/frp/utils/log"
frpNet "github.com/fatedier/frp/utils/net"
"github.com/fatedier/frp/utils/util"
@ -40,38 +42,41 @@ const (
var ServerService *Service
// Server service.
// Server service
type Service struct {
// Dispatch connections to different handlers listen on same port.
// Dispatch connections to different handlers listen on same port
muxer *mux.Mux
// Accept connections from client.
// Accept connections from client
listener frpNet.Listener
// Accept connections using kcp.
// Accept connections using kcp
kcpListener frpNet.Listener
// For https proxies, route requests to different clients by hostname and other infomation.
// For https proxies, route requests to different clients by hostname and other infomation
VhostHttpsMuxer *vhost.HttpsMuxer
httpReverseProxy *vhost.HttpReverseProxy
// Manage all controllers.
// Manage all controllers
ctlManager *ControlManager
// Manage all proxies.
// Manage all proxies
pxyManager *ProxyManager
// Manage all visitor listeners.
// Manage all visitor listeners
visitorManager *VisitorManager
// Manage all tcp ports.
tcpPortManager *PortManager
// Manage all tcp ports
tcpPortManager *ports.PortManager
// Manage all udp ports.
udpPortManager *PortManager
// Manage all udp ports
udpPortManager *ports.PortManager
// Controller for nat hole connections.
// Tcp Group Controller
tcpGroupCtl *group.TcpGroupCtl
// Controller for nat hole connections
natHoleController *NatHoleController
}
@ -81,9 +86,10 @@ func NewService() (svr *Service, err error) {
ctlManager: NewControlManager(),
pxyManager: NewProxyManager(),
visitorManager: NewVisitorManager(),
tcpPortManager: NewPortManager("tcp", cfg.ProxyBindAddr, cfg.AllowPorts),
udpPortManager: NewPortManager("udp", cfg.ProxyBindAddr, cfg.AllowPorts),
tcpPortManager: ports.NewPortManager("tcp", cfg.ProxyBindAddr, cfg.AllowPorts),
udpPortManager: ports.NewPortManager("udp", cfg.ProxyBindAddr, cfg.AllowPorts),
}
svr.tcpGroupCtl = group.NewTcpGroupCtl(svr.tcpPortManager)
// Init assets.
err = assets.Load(cfg.AssetsDir)

View File

@ -23,6 +23,22 @@ remote_port = 10901
use_encryption = true
use_compression = true
[tcp_group1]
type = tcp
local_ip = 127.0.0.1
local_port = 10701
remote_port = 10802
group = test1
group_key = 123
[tcp_group2]
type = tcp
local_ip = 127.0.0.1
local_port = 10702
remote_port = 10802
group = test1
group_key = 123
[udp_normal]
type = udp
local_ip = 127.0.0.1

View File

@ -28,6 +28,24 @@ func StartTcpEchoServer() {
}
}
func StartTcpEchoServer2() {
l, err := frpNet.ListenTcp("127.0.0.1", TEST_TCP2_PORT)
if err != nil {
fmt.Printf("echo server2 listen error: %v\n", err)
return
}
for {
c, err := l.Accept()
if err != nil {
fmt.Printf("echo server2 accept error: %v\n", err)
return
}
go echoWorker2(c)
}
}
func StartUdpEchoServer() {
l, err := frpNet.ListenUDP("127.0.0.1", TEST_UDP_PORT)
if err != nil {
@ -85,3 +103,25 @@ func echoWorker(c net.Conn) {
c.Write(buf[:n])
}
}
func echoWorker2(c net.Conn) {
buf := make([]byte, 2048)
for {
n, err := c.Read(buf)
if err != nil {
if err == io.EOF {
c.Close()
break
} else {
fmt.Printf("echo server read error: %v\n", err)
return
}
}
var w []byte
w = append(w, buf[:n]...)
w = append(w, buf[:n]...)
c.Write(w)
}
}

View File

@ -12,7 +12,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/fatedier/frp/client"
"github.com/fatedier/frp/server"
"github.com/fatedier/frp/server/ports"
gnet "github.com/fatedier/golib/net"
)
@ -25,7 +25,9 @@ var (
TEST_STR = "frp is a fast reverse proxy to help you expose a local server behind a NAT or firewall to the internet."
TEST_TCP_PORT int = 10701
TEST_TCP2_PORT int = 10702
TEST_TCP_FRP_PORT int = 10801
TEST_TCP2_FRP_PORT int = 10802
TEST_TCP_EC_FRP_PORT int = 10901
TEST_TCP_ECHO_STR string = "tcp type:" + TEST_STR
@ -62,6 +64,7 @@ var (
func init() {
go StartTcpEchoServer()
go StartTcpEchoServer2()
go StartUdpEchoServer()
go StartUnixDomainServer()
go StartHttpServer()
@ -226,19 +229,19 @@ func TestAllowPorts(t *testing.T) {
status, err := getProxyStatus(ProxyTcpPortNotAllowed)
if assert.NoError(err) {
assert.Equal(client.ProxyStatusStartErr, status.Status)
assert.True(strings.Contains(status.Err, server.ErrPortNotAllowed.Error()))
assert.True(strings.Contains(status.Err, ports.ErrPortNotAllowed.Error()))
}
status, err = getProxyStatus(ProxyUdpPortNotAllowed)
if assert.NoError(err) {
assert.Equal(client.ProxyStatusStartErr, status.Status)
assert.True(strings.Contains(status.Err, server.ErrPortNotAllowed.Error()))
assert.True(strings.Contains(status.Err, ports.ErrPortNotAllowed.Error()))
}
status, err = getProxyStatus(ProxyTcpPortUnavailable)
if assert.NoError(err) {
assert.Equal(client.ProxyStatusStartErr, status.Status)
assert.True(strings.Contains(status.Err, server.ErrPortUnAvailable.Error()))
assert.True(strings.Contains(status.Err, ports.ErrPortUnAvailable.Error()))
}
// Port normal
@ -310,3 +313,25 @@ func TestRangePortsMapping(t *testing.T) {
}
}
}
func TestGroup(t *testing.T) {
assert := assert.New(t)
var (
p1 int
p2 int
)
addr := fmt.Sprintf("127.0.0.1:%d", TEST_TCP2_FRP_PORT)
for i := 0; i < 6; i++ {
res, err := sendTcpMsg(addr, TEST_TCP_ECHO_STR)
assert.NoError(err)
switch res {
case TEST_TCP_ECHO_STR:
p1++
case TEST_TCP_ECHO_STR + TEST_TCP_ECHO_STR:
p2++
}
}
assert.True(p1 > 0 && p2 > 0, "group proxies load balancing")
}

View File

@ -52,16 +52,13 @@ func (r *VhostRouters) Del(domain, location string) {
if !found {
return
}
for i, vr := range vrs {
if vr.location == location {
if len(vrs) > i+1 {
r.RouterByDomain[domain] = append(vrs[:i], vrs[i+1:]...)
} else {
r.RouterByDomain[domain] = vrs[:i]
}
newVrs := make([]*VhostRouter, 0)
for _, vr := range vrs {
if vr.location != location {
newVrs = append(newVrs, vr)
}
}
r.RouterByDomain[domain] = newVrs
}
func (r *VhostRouters) Get(host, path string) (vr *VhostRouter, exist bool) {