mirror of
https://github.com/go-kratos/kratos.git
synced 2025-03-17 21:07:54 +02:00
grpc‘s secure discovery (#1270)
* add gprc secure discovery * add http insecure filter
This commit is contained in:
parent
a636fd52a4
commit
8f4e78b47d
64
examples/etcd_tls/cert/server.crt
Normal file
64
examples/etcd_tls/cert/server.crt
Normal file
@ -0,0 +1,64 @@
|
||||
|
||||
-----BEGIN CERTIFICATE-----
|
||||
MIIFZTCCA02gAwIBAgIBATANBgkqhkiG9w0BAQUFADB8MQswCQYDVQQGEwJDTjEL
|
||||
MAkGA1UECAwCU0gxCzAJBgNVBAcMAlNIMQ8wDQYDVQQKDAZrcmF0b3MxCzAJBgNV
|
||||
BAsMAklUMRMwEQYDVQQDDAprcmF0b3MuY29tMSAwHgYJKoZIhvcNAQkBFhFrcmF0
|
||||
b3NAa3JhdG9zLmNvbTAeFw0yMTA3MjcxNTM1MjJaFw0yMjA3MjcxNTM1MjJaME0x
|
||||
CzAJBgNVBAYTAkNOMQswCQYDVQQIDAJTSDELMAkGA1UEBwwCU0gxDzANBgNVBAoM
|
||||
BmtyYXRvczETMBEGA1UEAwwKa3JhdG9zLmNvbTCCAiIwDQYJKoZIhvcNAQEBBQAD
|
||||
ggIPADCCAgoCggIBALfEWAuPqUHTwTiOT8dtiCM3aEm1D7I6K/PY2mSDMMOI4f5z
|
||||
TYi7LsKLLItMQR44cEDein/kl0U0QNYJRHqVCr/3IXA4ds0maiq+npY/S2KABDiI
|
||||
Z38TZ1PZ2bcD3Jb0o1gw0GcSokOGzVtuNKSiASc3D711AepGerChD5UrbixrZdg2
|
||||
wAeWZj39Tl93+zCcldVrCHMSM7LmDluHJ5KZ8T6auuK6ypQPVhqLz8VseHhB/IRw
|
||||
s9/o9OlJ3kv4wB/CWlFIvU6ZGVnshZAGOk3Brq25kw600RRDr2MpxNhFY1Xvrt7D
|
||||
tAJ7fK3/3VaV+I2C4OzzmztK7T1WlL8XqZj6I3t1ZCZryMvDIUPzC4mgiRqEJiF7
|
||||
VzuVm1sjyTKkD4oX6ZjJYXJ/6pbvd8+AexuwVAsQJKqaF1iQC9jThg55RUTkGo8F
|
||||
DFErW7XHKHe8vKXuRLG5k3xZBiHK7gsVyHzx0ouuSuZMFHg7L9ACeeqtqxqKNd+0
|
||||
fIo4N0vNb3GEj+YaTLoadhDSBEsynyQNTfrDf+oFRmQUg3q0W4VJHJFhkmTHRRcq
|
||||
Qj6xEAJDHC1xr8yr3jif9BKWG2+zEbvpiTcRXKhycv2OUI11dK72aMnOycusJjRe
|
||||
8pOqcYhSVQZnz31WWlkmX9TRiQtEeUUkFCYAnIArSKm5rNwOddLCzC8Z4js1AgMB
|
||||
AAGjITAfMB0GA1UdEQQWMBSCDCoua3JhdG9zLmNvbYcEfwAAATANBgkqhkiG9w0B
|
||||
AQUFAAOCAgEAjr3SXzNOcN8+JQuroS6hKHadrcp6djepd3r5YKSEjKBNxVAU0gj6
|
||||
QGl0zjSqhxSFwN4wCqXU/4JJVOyAJCV+t992j5wNdaGI+Tcu4whK2LtPi0O68ttq
|
||||
Nn5H/8bmotW0IZ/YDcq1V8EVWiTZPECk4QLx26S2sjG4HOKNUAs8o+PoUmQE5bKJ
|
||||
XBFWmjsOfPnI0WBGnuCvGUw5wP1ipLiuK+OhoTNKA6SXPopm5KMDv7gjYPlcmPyI
|
||||
sJcpve75m9EXQxrDvJtvws8MIZnkWvWi7bW6uQ7274S7YjMZL09/sQTolQOtwl59
|
||||
pvEbkQNPzgdvQYAfrlJjSBtbq3OtHF+j+p2K+7R0TY2F1OW3LeV8vkcq7IOt38Er
|
||||
FK5feNEL3t9GlrF8ASmJp/JvhWQiJo5tZJxWZ68CjKfLmVd406ehsK5XNlHJemnl
|
||||
hNpuAegeV6WDglvMNavvQCJfK5mKokn73HujtQveZ8vwPKV4f6Wg3sHJ2yyP7ou2
|
||||
2UZ10Qbp6UvFYfYuMvLuq8kznapWqQ0dIJzPDs+CzVtjk1eINUF9UjsSvqX4oCvy
|
||||
+ZvaM9k4kU/fJMRkLstc/Dx2G13T/moI/l5sw0qFtck12zs/SQ7xgcW3b0ruGX9S
|
||||
11opfl5R86GpXXbz+vNL9fWP+8cIvoGZK8RAC872bcMPEoJPjPIwAbI=
|
||||
-----END CERTIFICATE-----
|
||||
-----BEGIN CERTIFICATE-----
|
||||
MIIFdDCCA1wCCQDHq+cGa349DzANBgkqhkiG9w0BAQsFADB8MQswCQYDVQQGEwJD
|
||||
TjELMAkGA1UECAwCU0gxCzAJBgNVBAcMAlNIMQ8wDQYDVQQKDAZrcmF0b3MxCzAJ
|
||||
BgNVBAsMAklUMRMwEQYDVQQDDAprcmF0b3MuY29tMSAwHgYJKoZIhvcNAQkBFhFr
|
||||
cmF0b3NAa3JhdG9zLmNvbTAeFw0yMTA3MjcxNTMwMzlaFw0zMTA3MjUxNTMwMzla
|
||||
MHwxCzAJBgNVBAYTAkNOMQswCQYDVQQIDAJTSDELMAkGA1UEBwwCU0gxDzANBgNV
|
||||
BAoMBmtyYXRvczELMAkGA1UECwwCSVQxEzARBgNVBAMMCmtyYXRvcy5jb20xIDAe
|
||||
BgkqhkiG9w0BCQEWEWtyYXRvc0BrcmF0b3MuY29tMIICIjANBgkqhkiG9w0BAQEF
|
||||
AAOCAg8AMIICCgKCAgEApy2GV9MiECuelNk3fz1Qwh6+wj8Ip11NG+LxEGK4/MLD
|
||||
JRJtbgAg/7s3vzrm4WDKATDO27W6wewNFOvEnGWyh9wyjAtSgnAcJreq7F2DMbpO
|
||||
+E2guIQHSCCzfa10s4BgwXKdBRPPvwTADIHXPtlbq4BItJqzt/AhLQbdDAp93mHX
|
||||
NCzFdlIr4wflT2OW7EO24K2LgMZLWCzaESei9fL6AYm7jEvfaFYZksI3rjJNAj1q
|
||||
wccMu1o6TvdWRA5fvBi6h15Z0ekR8C2LbM1A54zziZwd+YjcwdQHJJgWJFH7yNSt
|
||||
Oe/AJZzP1nRk/5H0EvxBnF7du6vfeSjZJytp8cXMlbYg4NGGkSy782tBaUaDIb43
|
||||
iLqSjfHVZDLzbDGNy/u/mzfo4xS8lxZ92zE7z21d0WyUAJ75Z72v4kaNTr2tnMuE
|
||||
NTTG1787e3NB0CaV6gjeP8XbMV8gwNTrJmTW3dS+DT6sKtTIISOvsUCZ6h5qFvKU
|
||||
RWqJ7MiaSxg31DPg51caYDjiVLkkES8GRpPM/Njsg9WpFTQeqcecKbbOdw0ihvoV
|
||||
fq0FgHpp+jbjm0KkmcNdSX7Ld5XeHp2rBPkA283IdXIAvjjthlyWJTmSP6kDDdEI
|
||||
km4Did1Bg0wcXNnFlbHHavCfeRTQbVoIYVkWH7I323Vp8itvKobz1GirSOK0Hr0C
|
||||
AwEAATANBgkqhkiG9w0BAQsFAAOCAgEAFBYIOOyABIbUUOjjjvx2FSDRBNLpee5O
|
||||
45KmznuCGerhR7ad3rUNhaakA9HJzmLMUXtmyzy6+ej0HzdqZE7RRzdDVftBGaOf
|
||||
thwEzUAiHfqeX0o039sTQvJSqUY2sBko+tyDRmeNtmd3TPE5VZZSWG+TUtrOofr7
|
||||
K28UquMthJrmtSC8IJQOvA78Nc/FCPNaGZrcS8ZvrgbrkCLPN4dIJvY9I38xaWQ+
|
||||
G0gNQazxPzdp89/UMzkczyJAKjYj4JyLCbrjzTZXjtu+rYJezxyS+3QSf0F0xVBr
|
||||
/8HCeXX6xG16WZY54Z6AijqI3sjiRBSQ25rLJJ00sWa9k1oH0Poyiv4pQG3RHgIs
|
||||
jJEQh7RQE5zAsBx2NHZ8FcsUGX2oOjSS+vtX//Bg37kN7oYx5HEtK9c+a0wWUxo8
|
||||
8cqIzqrQSWOd1CipxbE5CUUNKdGQzNCLTfLO68KitLdaSOCGoU/PaoAncbr6EjdB
|
||||
kzqJcHooqq8asl6fu1DVnYqCpEEp30ldU3p4MdclcMV0XZUq6bXFq1ylmfZXfC0t
|
||||
zSGnBUjxB5lohn5fs1S/DxyoUR8LbIFVWCljEf4jJtMRnCQV3bR3xeVTbGh3ti81
|
||||
21uhQKjIP/X2BRNAvRo1qUbrzEeHAJG3EbIG78rRFvSz6MkWVTZzeH1SgCr9Onw7
|
||||
djovWE7E6ic=
|
||||
-----END CERTIFICATE-----
|
81
examples/etcd_tls/client/main.go
Normal file
81
examples/etcd_tls/client/main.go
Normal file
@ -0,0 +1,81 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"github.com/go-kratos/kratos/v2/transport/http"
|
||||
"io/ioutil"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/go-kratos/etcd/registry"
|
||||
"github.com/go-kratos/kratos/examples/helloworld/helloworld"
|
||||
"github.com/go-kratos/kratos/v2/transport/grpc"
|
||||
clientv3 "go.etcd.io/etcd/client/v3"
|
||||
)
|
||||
|
||||
func main() {
|
||||
|
||||
b, err := ioutil.ReadFile("../cert/server.crt")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
cp := x509.NewCertPool()
|
||||
if !cp.AppendCertsFromPEM(b) {
|
||||
panic(err)
|
||||
}
|
||||
tlsConf := &tls.Config{ServerName: "www.kratos.com", RootCAs: cp}
|
||||
|
||||
cli, err := clientv3.New(clientv3.Config{
|
||||
Endpoints: []string{"127.0.0.1:2379"},
|
||||
})
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
r := registry.New(cli)
|
||||
for {
|
||||
callGRPC(r, tlsConf)
|
||||
callHTTP(r, tlsConf)
|
||||
time.Sleep(time.Second)
|
||||
}
|
||||
}
|
||||
|
||||
func callGRPC(r *registry.Registry, tlsConf *tls.Config) {
|
||||
conn, err := grpc.Dial(
|
||||
context.Background(),
|
||||
grpc.WithEndpoint("discovery:///helloworld"),
|
||||
grpc.WithDiscovery(r),
|
||||
grpc.WithTLSConfig(tlsConf),
|
||||
)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
client := helloworld.NewGreeterClient(conn)
|
||||
reply, err := client.SayHello(context.Background(), &helloworld.HelloRequest{Name: "kratos"})
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
log.Printf("[grpc] SayHello %+v\n", reply)
|
||||
}
|
||||
|
||||
func callHTTP(r *registry.Registry, tlsConf *tls.Config) {
|
||||
conn, err := http.NewClient(
|
||||
context.Background(),
|
||||
http.WithEndpoint("discovery:///helloworld"),
|
||||
http.WithDiscovery(r),
|
||||
http.WithBlock(),
|
||||
http.WithTLSConfig(tlsConf),
|
||||
)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
client := helloworld.NewGreeterHTTPClient(conn)
|
||||
reply, err := client.SayHello(context.Background(), &helloworld.HelloRequest{Name: "kratos"})
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
log.Printf("[http] SayHello %+v\n", reply)
|
||||
}
|
75
examples/etcd_tls/server/main.go
Normal file
75
examples/etcd_tls/server/main.go
Normal file
@ -0,0 +1,75 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"fmt"
|
||||
"github.com/go-kratos/kratos/v2/transport/http"
|
||||
"log"
|
||||
|
||||
"github.com/go-kratos/etcd/registry"
|
||||
pb "github.com/go-kratos/kratos/examples/helloworld/helloworld"
|
||||
"github.com/go-kratos/kratos/v2"
|
||||
"github.com/go-kratos/kratos/v2/middleware/recovery"
|
||||
"github.com/go-kratos/kratos/v2/transport/grpc"
|
||||
etcd "go.etcd.io/etcd/client/v3"
|
||||
)
|
||||
|
||||
// server is used to implement helloworld.GreeterServer.
|
||||
type server struct {
|
||||
pb.UnimplementedGreeterServer
|
||||
}
|
||||
|
||||
// SayHello implements helloworld.GreeterServer
|
||||
func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
|
||||
return &pb.HelloReply{Message: fmt.Sprintf("use tls:Welcome %+v!", in.Name)}, nil
|
||||
}
|
||||
|
||||
func main() {
|
||||
|
||||
cert, err := tls.LoadX509KeyPair("../cert/server.crt", "../cert/server.key")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
tlsConf := &tls.Config{Certificates: []tls.Certificate{cert}}
|
||||
|
||||
client, err := etcd.New(etcd.Config{
|
||||
Endpoints: []string{"127.0.0.1:2379"},
|
||||
})
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
|
||||
grpcSrv := grpc.NewServer(
|
||||
grpc.Address(":9000"),
|
||||
grpc.Middleware(
|
||||
recovery.Recovery(),
|
||||
),
|
||||
grpc.TLSConfig(tlsConf),
|
||||
)
|
||||
|
||||
httpSrv := http.NewServer(
|
||||
http.Address(":8000"),
|
||||
http.Middleware(
|
||||
recovery.Recovery(),
|
||||
),
|
||||
http.TLSConfig(tlsConf),
|
||||
)
|
||||
|
||||
s := &server{}
|
||||
pb.RegisterGreeterServer(grpcSrv, s)
|
||||
pb.RegisterGreeterHTTPServer(httpSrv, s)
|
||||
|
||||
r := registry.New(client)
|
||||
app := kratos.New(
|
||||
kratos.Name("helloworld"),
|
||||
kratos.Server(
|
||||
grpcSrv,
|
||||
httpSrv,
|
||||
),
|
||||
kratos.Registrar(r),
|
||||
)
|
||||
if err := app.Run(); err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
}
|
65
examples/etcd_tls/server2/main.go
Normal file
65
examples/etcd_tls/server2/main.go
Normal file
@ -0,0 +1,65 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/go-kratos/kratos/v2/transport/http"
|
||||
"log"
|
||||
|
||||
"github.com/go-kratos/etcd/registry"
|
||||
pb "github.com/go-kratos/kratos/examples/helloworld/helloworld"
|
||||
"github.com/go-kratos/kratos/v2"
|
||||
"github.com/go-kratos/kratos/v2/middleware/recovery"
|
||||
"github.com/go-kratos/kratos/v2/transport/grpc"
|
||||
etcd "go.etcd.io/etcd/client/v3"
|
||||
)
|
||||
|
||||
// server is used to implement helloworld.GreeterServer.
|
||||
type server struct {
|
||||
pb.UnimplementedGreeterServer
|
||||
}
|
||||
|
||||
// SayHello implements helloworld.GreeterServer
|
||||
func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
|
||||
return &pb.HelloReply{Message: fmt.Sprintf("not use tls:Welcome %+v!", in.Name)}, nil
|
||||
}
|
||||
|
||||
func main() {
|
||||
client, err := etcd.New(etcd.Config{
|
||||
Endpoints: []string{"127.0.0.1:2379"},
|
||||
})
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
|
||||
grpcSrv := grpc.NewServer(
|
||||
grpc.Address(":9001"),
|
||||
grpc.Middleware(
|
||||
recovery.Recovery(),
|
||||
),
|
||||
)
|
||||
|
||||
httpSrv := http.NewServer(
|
||||
http.Address(":8001"),
|
||||
http.Middleware(
|
||||
recovery.Recovery(),
|
||||
),
|
||||
)
|
||||
|
||||
s := &server{}
|
||||
pb.RegisterGreeterServer(grpcSrv, s)
|
||||
pb.RegisterGreeterHTTPServer(httpSrv, s)
|
||||
|
||||
r := registry.New(client)
|
||||
app := kratos.New(
|
||||
kratos.Name("helloworld"),
|
||||
kratos.Server(
|
||||
grpcSrv,
|
||||
httpSrv,
|
||||
),
|
||||
kratos.Registrar(r),
|
||||
)
|
||||
if err := app.Run(); err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
}
|
22
internal/endpoint/endpoint.go
Normal file
22
internal/endpoint/endpoint.go
Normal file
@ -0,0 +1,22 @@
|
||||
package endpoint
|
||||
|
||||
import (
|
||||
"net/url"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
func NewEndpoint(scheme, host string, isSecure bool) *url.URL {
|
||||
var query string
|
||||
if isSecure {
|
||||
query = "isSecure=true"
|
||||
}
|
||||
return &url.URL{Scheme: scheme, Host: host, RawQuery: query}
|
||||
}
|
||||
|
||||
func IsSecure(url *url.URL) bool {
|
||||
ok, err := strconv.ParseBool(url.Query().Get("isSecure"))
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
return ok
|
||||
}
|
42
internal/endpoint/endpoint_test.go
Normal file
42
internal/endpoint/endpoint_test.go
Normal file
@ -0,0 +1,42 @@
|
||||
package endpoint
|
||||
|
||||
import (
|
||||
"net/url"
|
||||
"reflect"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestEndPoint(t *testing.T) {
|
||||
type args struct {
|
||||
url *url.URL
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
want bool
|
||||
}{
|
||||
// TODO: Add test cases.
|
||||
{
|
||||
name: "grpc://127.0.0.1?isSecure=false",
|
||||
args: args{NewEndpoint("grpc", "127.0.0.1", false)},
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
name: "grpc://127.0.0.1?isSecure=true",
|
||||
args: args{NewEndpoint("http", "127.0.0.1", true)},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "grpc://127.0.0.1",
|
||||
args: args{NewEndpoint("grpc", "localhost", false)},
|
||||
want: false,
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := IsSecure(tt.args.url); !reflect.DeepEqual(got, tt.want) {
|
||||
t.Errorf("GetQuery() = %v, want %v", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
@ -111,7 +111,7 @@ func dial(ctx context.Context, insecure bool, opts ...ClientOption) (*grpc.Clien
|
||||
grpc.WithChainUnaryInterceptor(ints...),
|
||||
}
|
||||
if options.discovery != nil {
|
||||
grpcOpts = append(grpcOpts, grpc.WithResolvers(discovery.NewBuilder(options.discovery)))
|
||||
grpcOpts = append(grpcOpts, grpc.WithResolvers(discovery.NewBuilder(options.discovery, discovery.WithInsecure(insecure))))
|
||||
}
|
||||
if insecure {
|
||||
grpcOpts = append(grpcOpts, grpc.WithInsecure())
|
||||
|
@ -29,10 +29,18 @@ func WithTimeout(timeout time.Duration) Option {
|
||||
}
|
||||
}
|
||||
|
||||
// WithInsecure with isSecure option.
|
||||
func WithInsecure(insecure bool) Option {
|
||||
return func(b *builder) {
|
||||
b.insecure = insecure
|
||||
}
|
||||
}
|
||||
|
||||
type builder struct {
|
||||
discoverer registry.Discovery
|
||||
logger log.Logger
|
||||
timeout time.Duration
|
||||
insecure bool
|
||||
}
|
||||
|
||||
// NewBuilder creates a builder which is used to factory registry resolvers.
|
||||
@ -41,6 +49,7 @@ func NewBuilder(d registry.Discovery, opts ...Option) resolver.Builder {
|
||||
discoverer: d,
|
||||
logger: log.DefaultLogger,
|
||||
timeout: time.Second * 10,
|
||||
insecure: false,
|
||||
}
|
||||
for _, o := range opts {
|
||||
o(b)
|
||||
@ -69,11 +78,12 @@ func (b *builder) Build(target resolver.Target, cc resolver.ClientConn, opts res
|
||||
return nil, err
|
||||
}
|
||||
r := &discoveryResolver{
|
||||
w: w,
|
||||
cc: cc,
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
log: log.NewHelper(b.logger),
|
||||
w: w,
|
||||
cc: cc,
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
log: log.NewHelper(b.logger),
|
||||
insecure: b.insecure,
|
||||
}
|
||||
go r.watch()
|
||||
return r, nil
|
||||
|
@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"github.com/go-kratos/kratos/v2/internal/endpoint"
|
||||
"net/url"
|
||||
"time"
|
||||
|
||||
@ -20,6 +21,8 @@ type discoveryResolver struct {
|
||||
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
|
||||
insecure bool
|
||||
}
|
||||
|
||||
func (r *discoveryResolver) watch() {
|
||||
@ -45,7 +48,7 @@ func (r *discoveryResolver) watch() {
|
||||
func (r *discoveryResolver) update(ins []*registry.ServiceInstance) {
|
||||
var addrs []resolver.Address
|
||||
for _, in := range ins {
|
||||
endpoint, err := parseEndpoint(in.Endpoints)
|
||||
endpoint, err := parseEndpoint(in.Endpoints, r.insecure)
|
||||
if err != nil {
|
||||
r.log.Errorf("[resovler] Failed to parse discovery endpoint: %v", err)
|
||||
continue
|
||||
@ -76,14 +79,18 @@ func (r *discoveryResolver) Close() {
|
||||
|
||||
func (r *discoveryResolver) ResolveNow(options resolver.ResolveNowOptions) {}
|
||||
|
||||
func parseEndpoint(endpoints []string) (string, error) {
|
||||
func parseEndpoint(endpoints []string, insecure bool) (string, error) {
|
||||
|
||||
for _, e := range endpoints {
|
||||
u, err := url.Parse(e)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
if u.Scheme == "grpc" {
|
||||
return u.Host, nil
|
||||
if endpoint.IsSecure(u) != insecure {
|
||||
return u.Host, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
return "", nil
|
||||
|
@ -3,6 +3,7 @@ package grpc
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"github.com/go-kratos/kratos/v2/internal/endpoint"
|
||||
"net"
|
||||
"net/url"
|
||||
"sync"
|
||||
@ -158,7 +159,8 @@ func (s *Server) Endpoint() (*url.URL, error) {
|
||||
return
|
||||
}
|
||||
s.lis = lis
|
||||
s.endpoint = &url.URL{Scheme: "grpc", Host: addr}
|
||||
|
||||
s.endpoint = endpoint.NewEndpoint("grpc", addr, s.tlsConf != nil)
|
||||
})
|
||||
if s.err != nil {
|
||||
return nil, s.err
|
||||
|
@ -174,7 +174,7 @@ func NewClient(ctx context.Context, opts ...ClientOption) (*Client, error) {
|
||||
var r *resolver
|
||||
if options.discovery != nil {
|
||||
if target.Scheme == "discovery" {
|
||||
if r, err = newResolver(ctx, options.discovery, target, options.balancer, options.block); err != nil {
|
||||
if r, err = newResolver(ctx, options.discovery, target, options.balancer, options.block, !isSecure); err != nil {
|
||||
return nil, fmt.Errorf("[http client] new resolver failed!err: %v", options.endpoint)
|
||||
}
|
||||
} else if _, _, err := host.ExtractHostPort(options.endpoint); err != nil {
|
||||
@ -244,7 +244,7 @@ func (client *Client) invoke(ctx context.Context, req *http.Request, args interf
|
||||
if node, done, err = client.opts.balancer.Pick(ctx); err != nil {
|
||||
return nil, errors.ServiceUnavailable("NODE_NOT_FOUND", err.Error())
|
||||
}
|
||||
scheme, addr, err := parseEndpoint(node.Endpoints)
|
||||
scheme, addr, err := parseEndpoint(node.Endpoints, client.opts.tlsConf == nil)
|
||||
if err != nil {
|
||||
return nil, errors.ServiceUnavailable("NODE_NOT_FOUND", err.Error())
|
||||
}
|
||||
|
@ -3,8 +3,8 @@ package http
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"github.com/go-kratos/kratos/v2/internal/endpoint"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@ -52,18 +52,21 @@ type resolver struct {
|
||||
target *Target
|
||||
watcher registry.Watcher
|
||||
logger *log.Helper
|
||||
|
||||
insecure bool
|
||||
}
|
||||
|
||||
func newResolver(ctx context.Context, discovery registry.Discovery, target *Target, updater Updater, block bool) (*resolver, error) {
|
||||
func newResolver(ctx context.Context, discovery registry.Discovery, target *Target, updater Updater, block, insecure bool) (*resolver, error) {
|
||||
watcher, err := discovery.Watch(ctx, target.Endpoint)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
r := &resolver{
|
||||
target: target,
|
||||
watcher: watcher,
|
||||
logger: log.NewHelper(log.DefaultLogger),
|
||||
updater: updater,
|
||||
target: target,
|
||||
watcher: watcher,
|
||||
logger: log.NewHelper(log.DefaultLogger),
|
||||
updater: updater,
|
||||
insecure: insecure,
|
||||
}
|
||||
if block {
|
||||
done := make(chan error, 1)
|
||||
@ -115,7 +118,7 @@ func newResolver(ctx context.Context, discovery registry.Discovery, target *Targ
|
||||
func (r *resolver) update(services []*registry.ServiceInstance) {
|
||||
var nodes []*registry.ServiceInstance
|
||||
for _, in := range services {
|
||||
_, endpoint, err := parseEndpoint(in.Endpoints)
|
||||
_, endpoint, err := parseEndpoint(in.Endpoints, r.insecure)
|
||||
if err != nil {
|
||||
r.logger.Errorf("Failed to parse (%v) discovery endpoint: %v error %v", r.target, in.Endpoints, err)
|
||||
continue
|
||||
@ -139,19 +142,21 @@ func (r *resolver) Close() error {
|
||||
return r.watcher.Stop()
|
||||
}
|
||||
|
||||
func parseEndpoint(endpoints []string) (string, string, error) {
|
||||
func parseEndpoint(endpoints []string, insecure bool) (string, string, error) {
|
||||
for _, e := range endpoints {
|
||||
u, err := url.Parse(e)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
if u.Scheme == "http" {
|
||||
isSecure, _ := strconv.ParseBool(u.Query().Get("isSecure"))
|
||||
isSecure := endpoint.IsSecure(u)
|
||||
scheme := "http"
|
||||
if isSecure {
|
||||
scheme = "https"
|
||||
}
|
||||
return scheme, u.Host, nil
|
||||
if isSecure != insecure {
|
||||
return scheme, u.Host, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
return "", "", nil
|
||||
|
@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"github.com/go-kratos/kratos/v2/internal/endpoint"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
@ -217,11 +218,8 @@ func (s *Server) Endpoint() (*url.URL, error) {
|
||||
return
|
||||
}
|
||||
s.lis = lis
|
||||
var query string
|
||||
if s.tlsConf != nil {
|
||||
query = "isSecure=true"
|
||||
}
|
||||
s.endpoint = &url.URL{Scheme: "http", Host: addr, RawQuery: query}
|
||||
|
||||
s.endpoint = endpoint.NewEndpoint("http", addr, s.tlsConf != nil)
|
||||
})
|
||||
if s.err != nil {
|
||||
return nil, s.err
|
||||
|
Loading…
x
Reference in New Issue
Block a user