traefik/integration/grpc_test.go

431 lines
11 KiB
Go
Raw Normal View History

2017-08-06 09:55:42 +00:00
package integration
import (
2018-02-27 13:07:07 +00:00
"context"
2017-08-06 09:55:42 +00:00
"crypto/tls"
"crypto/x509"
2017-10-24 12:38:02 +00:00
"errors"
2018-05-23 15:48:04 +00:00
"math/rand"
2017-08-06 09:55:42 +00:00
"net"
"os"
"testing"
2017-08-06 09:55:42 +00:00
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/suite"
"github.com/traefik/traefik/v2/integration/helloworld"
"github.com/traefik/traefik/v2/integration/try"
"github.com/traefik/traefik/v2/pkg/log"
2017-08-06 09:55:42 +00:00
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
2023-03-20 14:38:08 +00:00
"google.golang.org/grpc/credentials/insecure"
2017-08-06 09:55:42 +00:00
)
2020-07-07 12:42:03 +00:00
var (
LocalhostCert []byte
LocalhostKey []byte
)
2017-08-06 09:55:42 +00:00
2018-05-23 15:48:04 +00:00
const randCharset = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890"
2020-05-11 10:06:07 +00:00
// GRPCSuite tests suite.
2017-08-06 09:55:42 +00:00
type GRPCSuite struct{ BaseSuite }
func TestGRPCSuite(t *testing.T) {
suite.Run(t, new(GRPCSuite))
}
2017-10-24 12:38:02 +00:00
type myserver struct {
stopStreamExample chan bool
}
2017-08-06 09:55:42 +00:00
func (s *GRPCSuite) SetupSuite() {
2017-08-06 09:55:42 +00:00
var err error
2021-03-04 19:08:03 +00:00
LocalhostCert, err = os.ReadFile("./resources/tls/local.cert")
assert.NoError(s.T(), err)
2021-03-04 19:08:03 +00:00
LocalhostKey, err = os.ReadFile("./resources/tls/local.key")
assert.NoError(s.T(), err)
2017-08-06 09:55:42 +00:00
}
func (s *myserver) SayHello(ctx context.Context, in *helloworld.HelloRequest) (*helloworld.HelloReply, error) {
2023-11-17 00:50:06 +00:00
return &helloworld.HelloReply{Message: "Hello " + in.GetName()}, nil
2017-08-06 09:55:42 +00:00
}
2017-10-24 12:38:02 +00:00
func (s *myserver) StreamExample(in *helloworld.StreamExampleRequest, server helloworld.Greeter_StreamExampleServer) error {
data := make([]byte, 512)
2018-05-23 15:48:04 +00:00
for i := range data {
data[i] = randCharset[rand.Intn(len(randCharset))]
}
2018-08-06 18:00:03 +00:00
if err := server.Send(&helloworld.StreamExampleReply{Data: string(data)}); err != nil {
2019-09-13 17:28:04 +00:00
log.WithoutContext().Error(err)
2018-08-06 18:00:03 +00:00
}
2017-10-24 12:38:02 +00:00
<-s.stopStreamExample
return nil
}
func startGRPCServer(lis net.Listener, server *myserver) error {
2017-08-06 09:55:42 +00:00
cert, err := tls.X509KeyPair(LocalhostCert, LocalhostKey)
if err != nil {
return err
}
creds := credentials.NewServerTLSFromCert(&cert)
serverOption := grpc.Creds(creds)
2017-10-24 12:38:02 +00:00
s := grpc.NewServer(serverOption)
2017-08-06 09:55:42 +00:00
defer s.Stop()
2017-10-24 12:38:02 +00:00
helloworld.RegisterGreeterServer(s, server)
2017-08-06 09:55:42 +00:00
return s.Serve(lis)
}
2018-05-24 08:52:04 +00:00
func starth2cGRPCServer(lis net.Listener, server *myserver) error {
s := grpc.NewServer()
defer s.Stop()
helloworld.RegisterGreeterServer(s, server)
return s.Serve(lis)
}
2017-10-24 12:38:02 +00:00
func getHelloClientGRPC() (helloworld.GreeterClient, func() error, error) {
2017-08-06 09:55:42 +00:00
roots := x509.NewCertPool()
roots.AppendCertsFromPEM(LocalhostCert)
credsClient := credentials.NewClientTLSFromCert(roots, "")
2017-08-06 09:55:42 +00:00
conn, err := grpc.Dial("127.0.0.1:4443", grpc.WithTransportCredentials(credsClient))
if err != nil {
2017-10-24 12:38:02 +00:00
return nil, func() error { return nil }, err
2017-08-06 09:55:42 +00:00
}
2017-10-24 12:38:02 +00:00
return helloworld.NewGreeterClient(conn), conn.Close, nil
}
2017-08-06 09:55:42 +00:00
2018-05-28 09:46:03 +00:00
func getHelloClientGRPCh2c() (helloworld.GreeterClient, func() error, error) {
2023-03-20 14:38:08 +00:00
conn, err := grpc.Dial("127.0.0.1:8081", grpc.WithTransportCredentials(insecure.NewCredentials()))
2018-05-28 09:46:03 +00:00
if err != nil {
return nil, func() error { return nil }, err
}
return helloworld.NewGreeterClient(conn), conn.Close, nil
}
func callHelloClientGRPC(name string, secure bool) (string, error) {
var client helloworld.GreeterClient
var closer func() error
var err error
if secure {
client, closer, err = getHelloClientGRPC()
} else {
client, closer, err = getHelloClientGRPCh2c()
}
2021-03-04 08:02:03 +00:00
defer func() { _ = closer() }()
2018-05-28 09:46:03 +00:00
2017-10-24 12:38:02 +00:00
if err != nil {
return "", err
}
2017-08-06 09:55:42 +00:00
r, err := client.SayHello(context.Background(), &helloworld.HelloRequest{Name: name})
if err != nil {
return "", err
}
2023-11-17 00:50:06 +00:00
return r.GetMessage(), nil
2017-08-06 09:55:42 +00:00
}
2017-10-24 12:38:02 +00:00
func callStreamExampleClientGRPC() (helloworld.Greeter_StreamExampleClient, func() error, error) {
client, closer, err := getHelloClientGRPC()
if err != nil {
return nil, closer, err
}
t, err := client.StreamExample(context.Background(), &helloworld.StreamExampleRequest{})
if err != nil {
return nil, closer, err
}
return t, closer, nil
}
func (s *GRPCSuite) TestGRPC() {
2017-08-06 09:55:42 +00:00
lis, err := net.Listen("tcp", ":0")
assert.NoError(s.T(), err)
2017-08-06 09:55:42 +00:00
_, port, err := net.SplitHostPort(lis.Addr().String())
assert.NoError(s.T(), err)
2017-08-06 09:55:42 +00:00
go func() {
2017-10-24 12:38:02 +00:00
err := startGRPCServer(lis, &myserver{})
assert.NoError(s.T(), err)
2017-08-06 09:55:42 +00:00
}()
file := s.adaptFile("fixtures/grpc/config.toml", struct {
2017-08-06 09:55:42 +00:00
CertContent string
KeyContent string
GRPCServerPort string
}{
CertContent: string(LocalhostCert),
KeyContent: string(LocalhostKey),
GRPCServerPort: port,
})
s.traefikCmd(withConfigFile(file))
2017-08-06 09:55:42 +00:00
// wait for Traefik
err = try.GetRequest("http://127.0.0.1:8080/api/rawdata", 1*time.Second, try.BodyContains("Host(`127.0.0.1`)"))
assert.NoError(s.T(), err)
2018-05-24 08:52:04 +00:00
2017-08-06 09:55:42 +00:00
var response string
err = try.Do(1*time.Second, func() error {
2018-05-28 09:46:03 +00:00
response, err = callHelloClientGRPC("World", true)
2017-08-06 09:55:42 +00:00
return err
})
assert.NoError(s.T(), err)
assert.Equal(s.T(), "Hello World", response)
2018-05-24 08:52:04 +00:00
}
func (s *GRPCSuite) TestGRPCh2c() {
2018-05-24 08:52:04 +00:00
lis, err := net.Listen("tcp", ":0")
assert.NoError(s.T(), err)
2018-05-24 08:52:04 +00:00
_, port, err := net.SplitHostPort(lis.Addr().String())
assert.NoError(s.T(), err)
2018-05-24 08:52:04 +00:00
go func() {
err := starth2cGRPCServer(lis, &myserver{})
assert.NoError(s.T(), err)
2018-05-24 08:52:04 +00:00
}()
file := s.adaptFile("fixtures/grpc/config_h2c.toml", struct {
2018-05-28 09:46:03 +00:00
GRPCServerPort string
}{
GRPCServerPort: port,
})
s.traefikCmd(withConfigFile(file))
2018-05-28 09:46:03 +00:00
// wait for Traefik
err = try.GetRequest("http://127.0.0.1:8080/api/rawdata", 1*time.Second, try.BodyContains("Host(`127.0.0.1`)"))
assert.NoError(s.T(), err)
2018-05-28 09:46:03 +00:00
var response string
err = try.Do(1*time.Second, func() error {
response, err = callHelloClientGRPC("World", false)
return err
})
assert.NoError(s.T(), err)
assert.Equal(s.T(), "Hello World", response)
2018-05-28 09:46:03 +00:00
}
func (s *GRPCSuite) TestGRPCh2cTermination() {
2018-05-28 09:46:03 +00:00
lis, err := net.Listen("tcp", ":0")
assert.NoError(s.T(), err)
2018-05-28 09:46:03 +00:00
_, port, err := net.SplitHostPort(lis.Addr().String())
assert.NoError(s.T(), err)
2018-05-28 09:46:03 +00:00
go func() {
err := starth2cGRPCServer(lis, &myserver{})
assert.NoError(s.T(), err)
2018-05-28 09:46:03 +00:00
}()
file := s.adaptFile("fixtures/grpc/config_h2c_termination.toml", struct {
2018-05-24 08:52:04 +00:00
CertContent string
KeyContent string
GRPCServerPort string
}{
CertContent: string(LocalhostCert),
KeyContent: string(LocalhostKey),
GRPCServerPort: port,
})
s.traefikCmd(withConfigFile(file))
2018-05-24 08:52:04 +00:00
// wait for Traefik
err = try.GetRequest("http://127.0.0.1:8080/api/rawdata", 1*time.Second, try.BodyContains("Host(`127.0.0.1`)"))
assert.NoError(s.T(), err)
2017-08-06 09:55:42 +00:00
2018-05-24 08:52:04 +00:00
var response string
err = try.Do(1*time.Second, func() error {
2018-05-28 09:46:03 +00:00
response, err = callHelloClientGRPC("World", true)
2018-05-24 08:52:04 +00:00
return err
})
assert.NoError(s.T(), err)
assert.Equal(s.T(), "Hello World", response)
2017-08-06 09:55:42 +00:00
}
2017-10-10 10:14:03 +00:00
func (s *GRPCSuite) TestGRPCInsecure() {
2017-10-10 10:14:03 +00:00
lis, err := net.Listen("tcp", ":0")
assert.NoError(s.T(), err)
2017-10-10 10:14:03 +00:00
_, port, err := net.SplitHostPort(lis.Addr().String())
assert.NoError(s.T(), err)
2017-10-10 10:14:03 +00:00
go func() {
2017-10-24 12:38:02 +00:00
err := startGRPCServer(lis, &myserver{})
assert.NoError(s.T(), err)
2017-10-10 10:14:03 +00:00
}()
file := s.adaptFile("fixtures/grpc/config_insecure.toml", struct {
2017-10-10 10:14:03 +00:00
CertContent string
KeyContent string
GRPCServerPort string
}{
CertContent: string(LocalhostCert),
KeyContent: string(LocalhostKey),
GRPCServerPort: port,
})
s.traefikCmd(withConfigFile(file))
2017-10-10 10:14:03 +00:00
// wait for Traefik
err = try.GetRequest("http://127.0.0.1:8080/api/rawdata", 1*time.Second, try.BodyContains("Host(`127.0.0.1`)"))
assert.NoError(s.T(), err)
2018-05-24 08:52:04 +00:00
2017-10-10 10:14:03 +00:00
var response string
err = try.Do(1*time.Second, func() error {
2018-05-28 09:46:03 +00:00
response, err = callHelloClientGRPC("World", true)
2017-10-10 10:14:03 +00:00
return err
})
assert.NoError(s.T(), err)
assert.Equal(s.T(), "Hello World", response)
2017-10-10 10:14:03 +00:00
}
2017-10-24 12:38:02 +00:00
func (s *GRPCSuite) TestGRPCBuffer() {
2017-10-24 12:38:02 +00:00
stopStreamExample := make(chan bool)
defer func() { stopStreamExample <- true }()
lis, err := net.Listen("tcp", ":0")
assert.NoError(s.T(), err)
2017-10-24 12:38:02 +00:00
_, port, err := net.SplitHostPort(lis.Addr().String())
assert.NoError(s.T(), err)
2017-10-24 12:38:02 +00:00
go func() {
err := startGRPCServer(lis, &myserver{
stopStreamExample: stopStreamExample,
})
assert.NoError(s.T(), err)
2017-10-24 12:38:02 +00:00
}()
file := s.adaptFile("fixtures/grpc/config.toml", struct {
2017-10-24 12:38:02 +00:00
CertContent string
KeyContent string
GRPCServerPort string
}{
CertContent: string(LocalhostCert),
KeyContent: string(LocalhostKey),
GRPCServerPort: port,
})
s.traefikCmd(withConfigFile(file))
2017-10-24 12:38:02 +00:00
// wait for Traefik
err = try.GetRequest("http://127.0.0.1:8080/api/rawdata", 1*time.Second, try.BodyContains("Host(`127.0.0.1`)"))
assert.NoError(s.T(), err)
2017-10-24 12:38:02 +00:00
var client helloworld.Greeter_StreamExampleClient
client, closer, err := callStreamExampleClientGRPC()
2021-03-04 08:02:03 +00:00
defer func() { _ = closer() }()
assert.NoError(s.T(), err)
2017-10-24 12:38:02 +00:00
received := make(chan bool)
go func() {
2017-11-22 17:20:03 +00:00
tr, err := client.Recv()
assert.NoError(s.T(), err)
assert.Len(s.T(), tr.GetData(), 512)
2017-10-24 12:38:02 +00:00
received <- true
}()
err = try.Do(10*time.Second, func() error {
2017-10-24 12:38:02 +00:00
select {
case <-received:
return nil
default:
return errors.New("failed to receive stream data")
}
})
assert.NoError(s.T(), err)
2017-10-24 12:38:02 +00:00
}
2018-10-29 17:42:03 +00:00
func (s *GRPCSuite) TestGRPCBufferWithFlushInterval() {
2018-10-29 17:42:03 +00:00
stopStreamExample := make(chan bool)
lis, err := net.Listen("tcp", ":0")
assert.NoError(s.T(), err)
2018-10-29 17:42:03 +00:00
_, port, err := net.SplitHostPort(lis.Addr().String())
assert.NoError(s.T(), err)
2018-10-29 17:42:03 +00:00
go func() {
err := startGRPCServer(lis, &myserver{
stopStreamExample: stopStreamExample,
})
assert.NoError(s.T(), err)
2018-10-29 17:42:03 +00:00
}()
file := s.adaptFile("fixtures/grpc/config.toml", struct {
2018-10-29 17:42:03 +00:00
CertContent string
KeyContent string
GRPCServerPort string
}{
CertContent: string(LocalhostCert),
KeyContent: string(LocalhostKey),
GRPCServerPort: port,
})
s.traefikCmd(withConfigFile(file))
2018-10-29 17:42:03 +00:00
// wait for Traefik
err = try.GetRequest("http://127.0.0.1:8080/api/rawdata", 1*time.Second, try.BodyContains("Host(`127.0.0.1`)"))
assert.NoError(s.T(), err)
2018-11-14 09:18:03 +00:00
2018-10-29 17:42:03 +00:00
var client helloworld.Greeter_StreamExampleClient
client, closer, err := callStreamExampleClientGRPC()
2021-03-04 08:02:03 +00:00
defer func() {
_ = closer()
stopStreamExample <- true
}()
assert.NoError(s.T(), err)
2018-10-29 17:42:03 +00:00
received := make(chan bool)
go func() {
tr, err := client.Recv()
assert.NoError(s.T(), err)
assert.Len(s.T(), tr.GetData(), 512)
2018-10-29 17:42:03 +00:00
received <- true
}()
2018-11-14 09:18:03 +00:00
err = try.Do(100*time.Millisecond, func() error {
2018-10-29 17:42:03 +00:00
select {
case <-received:
return nil
default:
return errors.New("failed to receive stream data")
}
})
assert.NoError(s.T(), err)
2018-10-29 17:42:03 +00:00
}
2019-02-01 08:50:04 +00:00
func (s *GRPCSuite) TestGRPCWithRetry() {
2019-02-01 08:50:04 +00:00
lis, err := net.Listen("tcp", ":0")
assert.NoError(s.T(), err)
2019-02-01 08:50:04 +00:00
_, port, err := net.SplitHostPort(lis.Addr().String())
assert.NoError(s.T(), err)
2019-02-01 08:50:04 +00:00
go func() {
err := startGRPCServer(lis, &myserver{})
assert.NoError(s.T(), err)
2019-02-01 08:50:04 +00:00
}()
file := s.adaptFile("fixtures/grpc/config_retry.toml", struct {
2019-02-01 08:50:04 +00:00
CertContent string
KeyContent string
GRPCServerPort string
}{
CertContent: string(LocalhostCert),
KeyContent: string(LocalhostKey),
GRPCServerPort: port,
})
s.traefikCmd(withConfigFile(file))
2019-02-01 08:50:04 +00:00
// wait for Traefik
err = try.GetRequest("http://127.0.0.1:8080/api/rawdata", 1*time.Second, try.BodyContains("Host(`127.0.0.1`)"))
assert.NoError(s.T(), err)
2019-02-01 08:50:04 +00:00
var response string
err = try.Do(1*time.Second, func() error {
response, err = callHelloClientGRPC("World", true)
return err
})
assert.NoError(s.T(), err)
assert.Equal(s.T(), "Hello World", response)
2019-02-01 08:50:04 +00:00
}