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

add HTTP plugin

Browse Source
This commit is contained in:
ginuerzh
2023-09-20 22:56:43 +08:00
parent afddd2d29e
commit 1760151500
29 changed files with 1032 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
ingress/ingress.go
View File

@ -9,7 +9,7 @@ import (
"sync"
"time"
ingress_pkg "github.com/go-gost/core/ingress"
"github.com/go-gost/core/ingress"
"github.com/go-gost/core/logger"
"github.com/go-gost/x/internal/loader"
"google.golang.org/grpc"
@ -74,7 +74,7 @@ func LoggerOption(logger logger.Logger) Option {
}
}
type ingress struct {
type localIngress struct {
rules map[string]Rule
cancelFunc context.CancelFunc
options options
@ -82,7 +82,7 @@ type ingress struct {
}
// NewIngress creates and initializes a new Ingress.
func NewIngress(opts ...Option) ingress_pkg.Ingress {
func NewIngress(opts ...Option) ingress.Ingress {
var options options
for _, opt := range opts {
opt(&options)
@ -90,7 +90,7 @@ func NewIngress(opts ...Option) ingress_pkg.Ingress {
ctx, cancel := context.WithCancel(context.TODO())
ing := &ingress{
ing := &localIngress{
cancelFunc: cancel,
options: options,
}
@ -105,7 +105,7 @@ func NewIngress(opts ...Option) ingress_pkg.Ingress {
return ing
}
func (ing *ingress) periodReload(ctx context.Context) error {
func (ing *localIngress) periodReload(ctx context.Context) error {
period := ing.options.period
if period < time.Second {
period = time.Second
@ -126,7 +126,7 @@ func (ing *ingress) periodReload(ctx context.Context) error {
}
}
func (ing *ingress) reload(ctx context.Context) error {
func (ing *localIngress) reload(ctx context.Context) error {
rules := make(map[string]Rule)
fn := func(rule Rule) {
@ -160,7 +160,7 @@ func (ing *ingress) reload(ctx context.Context) error {
return nil
}
func (ing *ingress) load(ctx context.Context) (rules []Rule, err error) {
func (ing *localIngress) load(ctx context.Context) (rules []Rule, err error) {
if ing.options.fileLoader != nil {
if lister, ok := ing.options.fileLoader.(loader.Lister); ok {
list, er := lister.List(ctx)
@ -211,7 +211,7 @@ func (ing *ingress) load(ctx context.Context) (rules []Rule, err error) {
return
}
func (ing *ingress) parseRules(r io.Reader) (rules []Rule, err error) {
func (ing *localIngress) parseRules(r io.Reader) (rules []Rule, err error) {
if r == nil {
return
}
@ -227,7 +227,7 @@ func (ing *ingress) parseRules(r io.Reader) (rules []Rule, err error) {
return
}
func (ing *ingress) Get(ctx context.Context, host string) string {
func (ing *localIngress) Get(ctx context.Context, host string) string {
if host == "" || ing == nil {
return ""
}
@ -267,7 +267,7 @@ func (ing *ingress) Get(ctx context.Context, host string) string {
return ep
}
func (ing *ingress) lookup(host string) string {
func (ing *localIngress) lookup(host string) string {
if ing == nil || len(ing.rules) == 0 {
return ""
}
@ -278,7 +278,7 @@ func (ing *ingress) lookup(host string) string {
return ing.rules[host].Endpoint
}
func (ing *ingress) parseLine(s string) (rule Rule) {
func (ing *localIngress) parseLine(s string) (rule Rule) {
line := strings.Replace(s, "\t", " ", -1)
line = strings.TrimSpace(line)
if n := strings.IndexByte(line, '#'); n >= 0 {
@ -300,7 +300,7 @@ func (ing *ingress) parseLine(s string) (rule Rule) {
}
}
func (ing *ingress) Close() error {
func (ing *localIngress) Close() error {
ing.cancelFunc()
if ing.options.fileLoader != nil {
ing.options.fileLoader.Close()

101
ingress/plugin.go
View File

@ -1,12 +1,16 @@
package ingress
import (
"bytes"
"context"
"encoding/json"
"io"
"net/http"
ingress_pkg "github.com/go-gost/core/ingress"
"github.com/go-gost/core/ingress"
"github.com/go-gost/core/logger"
"github.com/go-gost/plugin/ingress/proto"
"github.com/go-gost/x/internal/util/plugin"
"google.golang.org/grpc"
)
@ -16,14 +20,25 @@ type grpcPluginIngress struct {
log logger.Logger
}
// NewGRPCPluginIngress creates a ingress plugin based on gRPC.
func NewGRPCPluginIngress(name string, conn grpc.ClientConnInterface) ingress_pkg.Ingress {
// NewGRPCPluginIngress creates an Ingress plugin based on gRPC.
func NewGRPCPluginIngress(name string, addr string, opts ...plugin.Option) ingress.Ingress {
var options plugin.Options
for _, opt := range opts {
opt(&options)
}
log := logger.Default().WithFields(map[string]any{
"kind": "ingress",
"ingress": name,
})
conn, err := plugin.NewGRPCConn(addr, &options)
if err != nil {
log.Error(err)
}
p := &grpcPluginIngress{
conn: conn,
log: logger.Default().WithFields(map[string]any{
"kind": "ingress",
"ingress": name,
}),
log: log,
}
if conn != nil {
p.client = proto.NewIngressClient(conn)
@ -53,3 +68,75 @@ func (p *grpcPluginIngress) Close() error {
}
return nil
}
type httpIngressRequest struct {
Host string `json:"host"`
}
type httpIngressResponse struct {
Endpoint string `json:"endpoint"`
}
type httpPluginIngress struct {
url string
client *http.Client
header http.Header
log logger.Logger
}
// NewHTTPPluginIngress creates an Ingress plugin based on HTTP.
func NewHTTPPluginIngress(name string, url string, opts ...plugin.Option) ingress.Ingress {
var options plugin.Options
for _, opt := range opts {
opt(&options)
}
return &httpPluginIngress{
url: url,
client: plugin.NewHTTPClient(&options),
header: options.Header,
log: logger.Default().WithFields(map[string]any{
"kind": "ingress",
"ingress": name,
}),
}
}
func (p *httpPluginIngress) Get(ctx context.Context, host string) (endpoint string) {
if p.client == nil {
return
}
rb := httpIngressRequest{
Host: host,
}
v, err := json.Marshal(&rb)
if err != nil {
return
}
req, err := http.NewRequest(http.MethodPost, p.url, bytes.NewReader(v))
if err != nil {
return
}
if p.header != nil {
req.Header = p.header.Clone()
}
req.Header.Set("Content-Type", "application/json")
resp, err := p.client.Do(req)
if err != nil {
return
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return
}
res := httpIngressResponse{}
if err := json.NewDecoder(resp.Body).Decode(&res); err != nil {
return
}
return res.Endpoint
}