Skip to content

Commit

Permalink
grpcproxy: add test to reproduce issue with auth token
Browse files Browse the repository at this point in the history
Signed-off-by: Kristoffer Johansson <[email protected]>
  • Loading branch information
krijohs committed Dec 19, 2024
1 parent b65c4a6 commit 8d9e89f
Show file tree
Hide file tree
Showing 2 changed files with 100 additions and 0 deletions.
96 changes: 96 additions & 0 deletions tests/e2e/etcd_grpcproxy_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@ package e2e
import (
"context"
"strings"
"sync"
"sync/atomic"
"testing"
"time"

Expand Down Expand Up @@ -142,3 +144,97 @@ func waitForEndpointInLog(ctx context.Context, proxyProc *expect.ExpectProcess,

return err
}

func TestGRPCProxyWatchersAfterTokenExpiry(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cluster, err := e2e.NewEtcdProcessCluster(ctx, t,
e2e.WithClusterSize(1),
e2e.WithAuthTokenOpts("simple"),
e2e.WithAuthTokenTTL(1),
)
require.NoError(t, err)
t.Cleanup(func() { require.NoError(t, cluster.Stop()) })

cli := cluster.Etcdctl()

createUsers(ctx, t, cli)

require.NoError(t, cli.AuthEnable(ctx))

var (
node1ClientURL = cluster.Procs[0].Config().ClientURL
proxyClientURL = "127.0.0.1:42379"
)

proxyProc, err := e2e.SpawnCmd([]string{
e2e.BinPath.Etcd, "grpc-proxy", "start",
"--advertise-client-url", proxyClientURL,
"--listen-addr", proxyClientURL,
"--endpoints", node1ClientURL,
}, nil)
require.NoError(t, err)
t.Cleanup(func() { require.NoError(t, proxyProc.Stop()) })

var totalEventsCount int64

handler := func(events clientv3.WatchChan) {
for {
select {
case ev, open := <-events:
if !open {
return
}
if ev.Err() != nil {
t.Logf("watch response error: %s", ev.Err())
continue
}
atomic.AddInt64(&totalEventsCount, 1)
case <-ctx.Done():
return
}
}
}

withAuth := e2e.WithAuth("root", "rootPassword")
withEndpoint := e2e.WithEndpoints([]string{proxyClientURL})

events := cluster.Etcdctl(withAuth, withEndpoint).Watch(ctx, "/test", config.WatchOptions{Prefix: true, Revision: 1})

wg := sync.WaitGroup{}

wg.Add(1)
go func() {
defer wg.Done()
handler(events)
}()

clusterCli := cluster.Etcdctl(withAuth)
require.NoError(t, clusterCli.Put(ctx, "/test/1", "test", config.PutOptions{}))
require.NoError(t, err)

time.Sleep(time.Second * 2)

events2 := cluster.Etcdctl(withAuth, withEndpoint).Watch(ctx, "/test", config.WatchOptions{Prefix: true, Revision: 1})

wg.Add(1)
go func() {
defer wg.Done()
handler(events2)
}()

events3 := cluster.Etcdctl(withAuth, withEndpoint).Watch(ctx, "/test", config.WatchOptions{Prefix: true, Revision: 1})

wg.Add(1)
go func() {
defer wg.Done()
handler(events3)
}()

time.Sleep(time.Second)

cancel()
wg.Wait()

assert.Equal(t, int64(3), atomic.LoadInt64(&totalEventsCount))
}
4 changes: 4 additions & 0 deletions tests/framework/e2e/cluster.go
Original file line number Diff line number Diff line change
Expand Up @@ -291,6 +291,10 @@ func WithAuthTokenOpts(token string) EPClusterOption {
return func(c *EtcdProcessClusterConfig) { c.ServerConfig.AuthToken = token }
}

func WithAuthTokenTTL(ttl uint) EPClusterOption {
return func(c *EtcdProcessClusterConfig) { c.ServerConfig.AuthTokenTTL = ttl }
}

func WithRollingStart(rolling bool) EPClusterOption {
return func(c *EtcdProcessClusterConfig) { c.RollingStart = rolling }
}
Expand Down

0 comments on commit 8d9e89f

Please sign in to comment.