Skip to content

Commit

Permalink
client: Use first endpoint as http2 authority header
Browse files Browse the repository at this point in the history
This doesn't fully fix authority in multiple endpoint scenario, but
should at least fixes single endpoint scenario.
  • Loading branch information
serathius committed Sep 27, 2021
1 parent 2d91243 commit 8ebe2b9
Show file tree
Hide file tree
Showing 3 changed files with 45 additions and 18 deletions.
18 changes: 15 additions & 3 deletions client/v3/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -297,16 +297,28 @@ func (c *Client) dial(creds grpccredentials.TransportCredentials, dopts ...grpc.
dctx, cancel = context.WithTimeout(c.ctx, c.cfg.DialTimeout)
defer cancel() // TODO: Is this right for cases where grpc.WithBlock() is not set on the dial options?
}

initialEndpoints := strings.Join(c.Endpoints(), ";")
target := fmt.Sprintf("%s://%p/#initially=[%s]", resolver.Schema, c, initialEndpoints)
target := fmt.Sprintf("%s://%p/%s", resolver.Schema, c, authority(c.endpoints[0]))
conn, err := grpc.DialContext(dctx, target, opts...)
if err != nil {
return nil, err
}
return conn, nil
}

func authority(endpoint string) string {
spl := strings.SplitN(endpoint, "://", 2)
if len(spl) < 2 {
if strings.HasPrefix(endpoint, "unix:") {
return endpoint[len("unix:"):]
}
if strings.HasPrefix(endpoint, "unixs:") {
return endpoint[len("unixs:"):]
}
return endpoint
}
return spl[1]
}

func (c *Client) credentialsForEndpoint(ep string) grpccredentials.TransportCredentials {
r := endpoint.RequiresCredentials(ep)
switch r {
Expand Down
8 changes: 4 additions & 4 deletions tests/e2e/ctl_v3_grpc_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -32,19 +32,19 @@ func TestAuthority(t *testing.T) {
clientURLPattern string

// Pattern used to validate authority received by server. Fields filled:
// %s - list of endpoints concatenated with ";"
// %d - will be filled with first member grpc port
expectAuthorityPattern string
}{
// HTTP schema
{
name: "http://domain[:port]",
clientURLPattern: "http://localhost:%d",
expectAuthorityPattern: "#initially=[%s]",
expectAuthorityPattern: "localhost:%d",
},
{
name: "http://address[:port]",
clientURLPattern: "http://127.0.0.1:%d",
expectAuthorityPattern: "#initially=[%s]",
expectAuthorityPattern: "127.0.0.1:%d",
},
}
for _, tc := range tcs {
Expand All @@ -71,7 +71,7 @@ func TestAuthority(t *testing.T) {
}

executeWithTimeout(t, 5*time.Second, func() {
assertAuthority(t, fmt.Sprintf(tc.expectAuthorityPattern, strings.Join(endpoints, ";")), epc)
assertAuthority(t, fmt.Sprintf(tc.expectAuthorityPattern, 20000), epc)
})
})

Expand Down
37 changes: 26 additions & 11 deletions tests/integration/grpc_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -37,57 +37,58 @@ func TestAuthority(t *testing.T) {
clientURLPattern string

// Pattern used to validate authority received by server. Fields filled:
// %s - list of endpoints concatenated with ";"
// %d - will be filled with first member grpc port
// %s - will be filled with first member name
expectAuthorityPattern string
}{
{
name: "unix:path",
clientURLPattern: "unix:localhost:%s",
expectAuthorityPattern: "#initially=[%s]",
expectAuthorityPattern: "localhost:%s",
},
{
name: "unix://absolute_path",
clientURLPattern: "unix://localhost:%s",
expectAuthorityPattern: "#initially=[%s]",
expectAuthorityPattern: "localhost:%s",
},
// "unixs" is not standard schema supported by etcd
{
name: "unixs:absolute_path",
useTLS: true,
clientURLPattern: "unixs:localhost:%s",
expectAuthorityPattern: "#initially=[%s]",
expectAuthorityPattern: "localhost:%s",
},
{
name: "unixs://absolute_path",
useTLS: true,
clientURLPattern: "unixs://localhost:%s",
expectAuthorityPattern: "#initially=[%s]",
expectAuthorityPattern: "localhost:%s",
},
{
name: "http://domain[:port]",
useTCP: true,
clientURLPattern: "http://localhost:%d",
expectAuthorityPattern: "#initially=[%s]",
expectAuthorityPattern: "localhost:%d",
},
{
name: "https://domain[:port]",
useTLS: true,
useTCP: true,
clientURLPattern: "https://localhost:%d",
expectAuthorityPattern: "#initially=[%s]",
expectAuthorityPattern: "localhost:%d",
},
{
name: "http://address[:port]",
useTCP: true,
clientURLPattern: "http://127.0.0.1:%d",
expectAuthorityPattern: "#initially=[%s]",
expectAuthorityPattern: "127.0.0.1:%d",
},
{
name: "https://address[:port]",
useTCP: true,
useTLS: true,
clientURLPattern: "https://127.0.0.1:%d",
expectAuthorityPattern: "#initially=[%s]",
expectAuthorityPattern: "127.0.0.1:%d",
},
}
for _, tc := range tcs {
Expand All @@ -102,7 +103,6 @@ func TestAuthority(t *testing.T) {
cfg, tlsConfig := setupTLS(t, tc.useTLS, cfg)
clus := NewClusterV3(t, &cfg)
defer clus.Terminate(t)
endpoints := templateEndpoints(t, tc.clientURLPattern, clus)

kv := setupClient(t, tc.clientURLPattern, clus, tlsConfig)
defer kv.Close()
Expand All @@ -112,7 +112,7 @@ func TestAuthority(t *testing.T) {
t.Fatal(err)
}

assertAuthority(t, fmt.Sprintf(tc.expectAuthorityPattern, strings.Join(endpoints, ";")), clus)
assertAuthority(t, templateAuthority(t, tc.expectAuthorityPattern, clus.Members[0]), clus)
})
}
}
Expand Down Expand Up @@ -165,6 +165,21 @@ func templateEndpoints(t *testing.T, pattern string, clus *ClusterV3) []string {
return endpoints
}

func templateAuthority(t *testing.T, pattern string, m *member) string {
t.Helper()
authority := pattern
if strings.Contains(authority, "%d") {
authority = fmt.Sprintf(authority, GrpcPortNumber(m.UniqNumber, m.MemberNumber))
}
if strings.Contains(authority, "%s") {
authority = fmt.Sprintf(authority, m.Name)
}
if strings.Contains(authority, "%") {
t.Fatalf("Failed to template pattern, %% symbol left %q", authority)
}
return authority
}

func assertAuthority(t *testing.T, expectedAuthority string, clus *ClusterV3) {
t.Helper()
requestsFound := 0
Expand Down

0 comments on commit 8ebe2b9

Please sign in to comment.