mirror of
https://github.com/eolinker/apinto
synced 2025-10-26 02:00:40 +08:00
grpc重写插件,首次提交
This commit is contained in:
@@ -19,6 +19,7 @@ import (
|
|||||||
circuit_breaker "github.com/eolinker/apinto/drivers/plugins/circuit-breaker"
|
circuit_breaker "github.com/eolinker/apinto/drivers/plugins/circuit-breaker"
|
||||||
"github.com/eolinker/apinto/drivers/plugins/cors"
|
"github.com/eolinker/apinto/drivers/plugins/cors"
|
||||||
extra_params "github.com/eolinker/apinto/drivers/plugins/extra-params"
|
extra_params "github.com/eolinker/apinto/drivers/plugins/extra-params"
|
||||||
|
grpc_proxy_rewrite "github.com/eolinker/apinto/drivers/plugins/grpc-proxy-rewrite"
|
||||||
"github.com/eolinker/apinto/drivers/plugins/gzip"
|
"github.com/eolinker/apinto/drivers/plugins/gzip"
|
||||||
ip_restriction "github.com/eolinker/apinto/drivers/plugins/ip-restriction"
|
ip_restriction "github.com/eolinker/apinto/drivers/plugins/ip-restriction"
|
||||||
"github.com/eolinker/apinto/drivers/plugins/monitor"
|
"github.com/eolinker/apinto/drivers/plugins/monitor"
|
||||||
@@ -126,4 +127,6 @@ func Register(extenderRegister eosc.IExtenderDriverRegister) {
|
|||||||
fuse.Register(extenderRegister)
|
fuse.Register(extenderRegister)
|
||||||
fuse_strategy.Register(extenderRegister)
|
fuse_strategy.Register(extenderRegister)
|
||||||
|
|
||||||
|
grpc_proxy_rewrite.Register(extenderRegister)
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
10
drivers/plugins/grpc-proxy-rewrite/config.go
Normal file
10
drivers/plugins/grpc-proxy-rewrite/config.go
Normal file
@@ -0,0 +1,10 @@
|
|||||||
|
package grpc_proxy_rewrite
|
||||||
|
|
||||||
|
type Config struct {
|
||||||
|
Service string `json:"service" label:"服务名称"`
|
||||||
|
Method string `json:"method" label:"方法名称"`
|
||||||
|
Authority string `json:"authority" label:"虚拟主机域名(Authority)"`
|
||||||
|
Headers map[string]string `json:"headers" label:"请求头部"`
|
||||||
|
Tls bool `json:"tls" label:"TLS传输"`
|
||||||
|
SkipCertificate bool `json:"skip_certificate" label:"跳过证书检查"`
|
||||||
|
}
|
||||||
31
drivers/plugins/grpc-proxy-rewrite/driver.go
Normal file
31
drivers/plugins/grpc-proxy-rewrite/driver.go
Normal file
@@ -0,0 +1,31 @@
|
|||||||
|
package grpc_proxy_rewrite
|
||||||
|
|
||||||
|
import (
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
"github.com/eolinker/apinto/drivers"
|
||||||
|
"github.com/eolinker/eosc"
|
||||||
|
)
|
||||||
|
|
||||||
|
func check(v interface{}) (*Config, error) {
|
||||||
|
conf, err := drivers.Assert[Config](v)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
conf.Authority = strings.TrimSpace(conf.Authority)
|
||||||
|
|
||||||
|
return conf, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func Create(id, name string, conf *Config, workers map[eosc.RequireId]eosc.IWorker) (eosc.IWorker, error) {
|
||||||
|
|
||||||
|
pw := &ProxyRewrite{
|
||||||
|
WorkerBase: drivers.Worker(id, name),
|
||||||
|
host: strings.TrimSpace(conf.Authority),
|
||||||
|
headers: conf.Headers,
|
||||||
|
tls: conf.Tls,
|
||||||
|
skipCertificate: conf.SkipCertificate,
|
||||||
|
}
|
||||||
|
|
||||||
|
return pw, nil
|
||||||
|
}
|
||||||
18
drivers/plugins/grpc-proxy-rewrite/factory.go
Normal file
18
drivers/plugins/grpc-proxy-rewrite/factory.go
Normal file
@@ -0,0 +1,18 @@
|
|||||||
|
package grpc_proxy_rewrite
|
||||||
|
|
||||||
|
import (
|
||||||
|
"github.com/eolinker/apinto/drivers"
|
||||||
|
"github.com/eolinker/eosc"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
Name = "grpc-proxy_write"
|
||||||
|
)
|
||||||
|
|
||||||
|
func Register(register eosc.IExtenderDriverRegister) {
|
||||||
|
register.RegisterExtenderDriver(Name, NewFactory())
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewFactory() eosc.IExtenderDriverFactory {
|
||||||
|
return drivers.NewFactory[Config](Create)
|
||||||
|
}
|
||||||
77
drivers/plugins/grpc-proxy-rewrite/proxy-rewrite.go
Normal file
77
drivers/plugins/grpc-proxy-rewrite/proxy-rewrite.go
Normal file
@@ -0,0 +1,77 @@
|
|||||||
|
package grpc_proxy_rewrite
|
||||||
|
|
||||||
|
import (
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
grpc_context "github.com/eolinker/eosc/eocontext/grpc-context"
|
||||||
|
|
||||||
|
"github.com/eolinker/apinto/drivers"
|
||||||
|
"github.com/eolinker/eosc"
|
||||||
|
"github.com/eolinker/eosc/eocontext"
|
||||||
|
http_service "github.com/eolinker/eosc/eocontext/http-context"
|
||||||
|
)
|
||||||
|
|
||||||
|
var _ eocontext.IFilter = (*ProxyRewrite)(nil)
|
||||||
|
var _ grpc_context.GrpcFilter = (*ProxyRewrite)(nil)
|
||||||
|
|
||||||
|
var (
|
||||||
|
regexpErrInfo = `[plugin proxy-rewrite2 config err] Compile regexp fail. err regexp: %s `
|
||||||
|
notMatchErrInfo = `[plugin proxy-rewrite2 err] Proxy path rewrite fail. Request path can't match any rewrite-path. request path: %s `
|
||||||
|
)
|
||||||
|
|
||||||
|
type ProxyRewrite struct {
|
||||||
|
drivers.WorkerBase
|
||||||
|
|
||||||
|
host string
|
||||||
|
headers map[string]string
|
||||||
|
tls bool
|
||||||
|
skipCertificate bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *ProxyRewrite) DoFilter(ctx eocontext.EoContext, next eocontext.IChain) (err error) {
|
||||||
|
return grpc_context.DoGrpcFilter(p, ctx, next)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *ProxyRewrite) DoGrpcFilter(ctx grpc_context.IGrpcContext, next eocontext.IChain) (err error) {
|
||||||
|
if p.host != "" {
|
||||||
|
ctx.Proxy().SetHost(p.host)
|
||||||
|
}
|
||||||
|
ctx.EnableTls(p.tls)
|
||||||
|
ctx.InsecureCertificateVerify(p.skipCertificate)
|
||||||
|
for key, value := range p.headers {
|
||||||
|
ctx.Proxy().Headers().Set(key, value)
|
||||||
|
}
|
||||||
|
if next != nil {
|
||||||
|
return next.DoChain(ctx)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *ProxyRewrite) Start() error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *ProxyRewrite) Reset(v interface{}, workers map[eosc.RequireId]eosc.IWorker) error {
|
||||||
|
conf, err := check(v)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
p.skipCertificate = conf.SkipCertificate
|
||||||
|
p.headers = conf.Headers
|
||||||
|
p.tls = conf.Tls
|
||||||
|
p.host = strings.TrimSpace(conf.Authority)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *ProxyRewrite) Stop() error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *ProxyRewrite) Destroy() {
|
||||||
|
|
||||||
|
p.headers = nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *ProxyRewrite) CheckSkill(skill string) bool {
|
||||||
|
return http_service.FilterSkillName == skill
|
||||||
|
}
|
||||||
@@ -42,7 +42,6 @@ func (h *Complete) Complete(org eocontext.EoContext) error {
|
|||||||
ctx.Response().SetErr(lastErr)
|
ctx.Response().SetErr(lastErr)
|
||||||
ctx.SetLabel("handler", "proxy")
|
ctx.SetLabel("handler", "proxy")
|
||||||
}()
|
}()
|
||||||
|
|
||||||
timeOut := app.TimeOut()
|
timeOut := app.TimeOut()
|
||||||
for index := 0; index <= h.retry; index++ {
|
for index := 0; index <= h.retry; index++ {
|
||||||
|
|
||||||
|
|||||||
@@ -203,6 +203,15 @@ func (c *Context) Invoke(address string, timeout time.Duration) error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
passHost, targetHost := c.GetUpstreamHostHandler().PassHost()
|
||||||
|
switch passHost {
|
||||||
|
case eocontext.PassHost:
|
||||||
|
|
||||||
|
case eocontext.NodeHost:
|
||||||
|
c.proxy.Headers().Set(":authority", address)
|
||||||
|
case eocontext.ReWriteHost:
|
||||||
|
c.proxy.Headers().Set(":authority", targetHost)
|
||||||
|
}
|
||||||
c.proxy.Headers().Set("grpc-timeout", fmt.Sprintf("%dn", timeout))
|
c.proxy.Headers().Set("grpc-timeout", fmt.Sprintf("%dn", timeout))
|
||||||
clientCtx, _ := context.WithCancel(metadata.NewOutgoingContext(c.Context(), c.proxy.Headers().Copy()))
|
clientCtx, _ := context.WithCancel(metadata.NewOutgoingContext(c.Context(), c.proxy.Headers().Copy()))
|
||||||
clientStream, err := grpc.NewClientStream(clientCtx, clientStreamDescForProxying, clientConn, c.proxy.FullMethodName())
|
clientStream, err := grpc.NewClientStream(clientCtx, clientStreamDescForProxying, clientConn, c.proxy.FullMethodName())
|
||||||
|
|||||||
@@ -126,7 +126,6 @@ func (cc *ClientPool) connect() (*grpc.ClientConn, error) {
|
|||||||
Timeout: cc.option.KeepAliveTimeout,
|
Timeout: cc.option.KeepAliveTimeout,
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
|
|
||||||
conn, err := grpc.DialContext(ctx, cc.target, opts...)
|
conn, err := grpc.DialContext(ctx, cc.target, opts...)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
|||||||
@@ -30,6 +30,10 @@ type Request struct {
|
|||||||
realIP string
|
realIP string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (r *Request) SetHost(s string) {
|
||||||
|
r.host = s
|
||||||
|
}
|
||||||
|
|
||||||
func (r *Request) SetService(service string) {
|
func (r *Request) SetService(service string) {
|
||||||
r.service = service
|
r.service = service
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"sort"
|
"sort"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
|
||||||
grpc_context "github.com/eolinker/eosc/eocontext/grpc-context"
|
grpc_context "github.com/eolinker/eosc/eocontext/grpc-context"
|
||||||
|
|
||||||
@@ -30,7 +31,11 @@ func newHostMatcher(children map[string]router.IMatcher) router.IMatcher {
|
|||||||
children: children,
|
children: children,
|
||||||
name: "host",
|
name: "host",
|
||||||
read: func(port int, request grpc_context.IRequest) (string, bool) {
|
read: func(port int, request grpc_context.IRequest) (string, bool) {
|
||||||
return request.Host(), true
|
orgHost := request.Host()
|
||||||
|
if i := strings.Index(orgHost, ":"); i > 0 {
|
||||||
|
return orgHost[:i], true
|
||||||
|
}
|
||||||
|
return orgHost, true
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user