| // Copyright 2015 The Vanadium Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style |
| // license that can be found in the LICENSE file. |
| |
| package test |
| |
| import ( |
| "fmt" |
| "io" |
| "os" |
| "reflect" |
| "sort" |
| "strings" |
| "testing" |
| "time" |
| |
| "v.io/x/lib/vlog" |
| |
| "v.io/v23" |
| "v.io/v23/context" |
| "v.io/v23/namespace" |
| "v.io/v23/naming" |
| "v.io/v23/options" |
| "v.io/v23/rpc" |
| "v.io/v23/security" |
| "v.io/v23/verror" |
| "v.io/v23/vtrace" |
| "v.io/x/ref/lib/flags" |
| _ "v.io/x/ref/profiles" |
| "v.io/x/ref/profiles/internal/lib/publisher" |
| inaming "v.io/x/ref/profiles/internal/naming" |
| irpc "v.io/x/ref/profiles/internal/rpc" |
| imanager "v.io/x/ref/profiles/internal/rpc/stream/manager" |
| "v.io/x/ref/profiles/internal/rpc/stream/proxy" |
| tnaming "v.io/x/ref/profiles/internal/testing/mocks/naming" |
| ivtrace "v.io/x/ref/profiles/internal/vtrace" |
| "v.io/x/ref/test/modules" |
| "v.io/x/ref/test/testutil" |
| ) |
| |
| func testContext() (*context.T, func()) { |
| ctx, shutdown := v23.Init() |
| ctx, _ = context.WithTimeout(ctx, 20*time.Second) |
| var err error |
| if ctx, err = ivtrace.Init(ctx, flags.VtraceFlags{}); err != nil { |
| panic(err) |
| } |
| ctx, _ = vtrace.WithNewTrace(ctx) |
| return ctx, shutdown |
| } |
| |
| func proxyServer(stdin io.Reader, stdout, stderr io.Writer, env map[string]string, args ...string) error { |
| ctx, shutdown := v23.Init() |
| defer shutdown() |
| |
| expected := len(args) |
| |
| listenSpec := rpc.ListenSpec{Addrs: rpc.ListenAddrs{{"tcp", "127.0.0.1:0"}}} |
| proxyShutdown, proxyEp, err := proxy.New(ctx, listenSpec) |
| if err != nil { |
| fmt.Fprintf(stderr, "%s\n", verror.DebugString(err)) |
| return err |
| } |
| defer proxyShutdown() |
| fmt.Fprintf(stdout, "PID=%d\n", os.Getpid()) |
| if expected > 0 { |
| pub := publisher.New(ctx, v23.GetNamespace(ctx), time.Minute) |
| defer pub.WaitForStop() |
| defer pub.Stop() |
| pub.AddServer(proxyEp.String()) |
| for _, name := range args { |
| if len(name) == 0 { |
| return fmt.Errorf("empty name specified on the command line") |
| } |
| pub.AddName(name, false, false) |
| } |
| // Wait for all the entries to be published. |
| for { |
| pubState := pub.Status() |
| if expected == len(pubState) { |
| break |
| } |
| delay := time.Second |
| time.Sleep(delay) |
| } |
| } |
| fmt.Fprintf(stdout, "PROXY_NAME=%s\n", proxyEp.Name()) |
| modules.WaitForEOF(stdin) |
| fmt.Fprintf(stdout, "DONE\n") |
| return nil |
| } |
| |
| type testServer struct{} |
| |
| func (*testServer) Echo(_ *context.T, call rpc.ServerCall, arg string) (string, error) { |
| return fmt.Sprintf("method:%q,suffix:%q,arg:%q", "Echo", call.Suffix(), arg), nil |
| } |
| |
| type testServerAuthorizer struct{} |
| |
| func (testServerAuthorizer) Authorize(*context.T, security.Call) error { |
| return nil |
| } |
| |
| type testServerDisp struct{ server interface{} } |
| |
| func (t testServerDisp) Lookup(suffix string) (interface{}, security.Authorizer, error) { |
| return t.server, testServerAuthorizer{}, nil |
| } |
| |
| type proxyHandle struct { |
| ns namespace.T |
| sh *modules.Shell |
| proxy modules.Handle |
| name string |
| } |
| |
| func (h *proxyHandle) Start(t *testing.T, ctx *context.T, args ...string) error { |
| sh, err := modules.NewShell(nil, nil, testing.Verbose(), t) |
| if err != nil { |
| t.Fatalf("unexpected error: %s", err) |
| } |
| h.sh = sh |
| p, err := sh.Start("proxyServer", nil, args...) |
| if err != nil { |
| t.Fatalf("unexpected error: %s", err) |
| } |
| h.proxy = p |
| p.ReadLine() |
| h.name = p.ExpectVar("PROXY_NAME") |
| if len(h.name) == 0 { |
| h.proxy.Shutdown(os.Stderr, os.Stderr) |
| t.Fatalf("failed to get PROXY_NAME from proxyd") |
| } |
| return h.ns.Mount(ctx, "proxy", h.name, time.Hour) |
| } |
| |
| func (h *proxyHandle) Stop(ctx *context.T) error { |
| defer h.sh.Cleanup(os.Stderr, os.Stderr) |
| if err := h.proxy.Shutdown(os.Stderr, os.Stderr); err != nil { |
| return err |
| } |
| if len(h.name) == 0 { |
| return nil |
| } |
| return h.ns.Unmount(ctx, "proxy", h.name) |
| } |
| |
| func TestProxyOnly(t *testing.T) { |
| listenSpec := rpc.ListenSpec{Proxy: "proxy"} |
| testProxy(t, listenSpec) |
| } |
| |
| func TestProxy(t *testing.T) { |
| proxyListenSpec := rpc.ListenSpec{ |
| Addrs: rpc.ListenAddrs{{"tcp", "127.0.0.1:0"}}, |
| Proxy: "proxy", |
| } |
| testProxy(t, proxyListenSpec) |
| } |
| |
| func TestWSProxy(t *testing.T) { |
| proxyListenSpec := rpc.ListenSpec{ |
| Addrs: rpc.ListenAddrs{{"tcp", "127.0.0.1:0"}}, |
| Proxy: "proxy", |
| } |
| // The proxy uses websockets only, but the server is using tcp. |
| testProxy(t, proxyListenSpec, "--v23.tcp.protocol=ws") |
| } |
| |
| func testProxy(t *testing.T, spec rpc.ListenSpec, args ...string) { |
| ctx, shutdown := testContext() |
| defer shutdown() |
| |
| var ( |
| pserver = testutil.NewPrincipal("server") |
| pclient = testutil.NewPrincipal("client") |
| serverKey = pserver.PublicKey() |
| // We use different stream managers for the client and server |
| // to prevent VIF re-use (in other words, we want to test VIF |
| // creation from both the client and server end). |
| smserver = imanager.InternalNew(naming.FixedRoutingID(0x555555555)) |
| smclient = imanager.InternalNew(naming.FixedRoutingID(0x444444444)) |
| ns = tnaming.NewSimpleNamespace() |
| ) |
| defer smserver.Shutdown() |
| defer smclient.Shutdown() |
| client, err := irpc.InternalNewClient(smserver, ns) |
| if err != nil { |
| t.Fatal(err) |
| } |
| defer client.Close() |
| serverCtx, _ := v23.WithPrincipal(ctx, pserver) |
| server, err := irpc.InternalNewServer(serverCtx, smserver, ns, nil, pserver) |
| if err != nil { |
| t.Fatal(err) |
| } |
| defer server.Stop() |
| |
| // The client must recognize the server's blessings, otherwise it won't |
| // communicate with it. |
| pclient.AddToRoots(pserver.BlessingStore().Default()) |
| |
| // If no address is specified then we'll only 'listen' via |
| // the proxy. |
| hasLocalListener := len(spec.Addrs) > 0 && len(spec.Addrs[0].Address) != 0 |
| |
| name := "mountpoint/server/suffix" |
| makeCall := func(opts ...rpc.CallOpt) (string, error) { |
| clientCtx, _ := v23.WithPrincipal(ctx, pclient) |
| clientCtx, _ = context.WithDeadline(clientCtx, time.Now().Add(5*time.Second)) |
| call, err := client.StartCall(clientCtx, name, "Echo", []interface{}{"batman"}, opts...) |
| if err != nil { |
| // proxy is down, we should return here/.... prepend |
| // the error with a well known string so that we can test for that. |
| return "", fmt.Errorf("RESOLVE: %s", err) |
| } |
| var result string |
| if err = call.Finish(&result); err != nil { |
| return "", err |
| } |
| return result, nil |
| } |
| proxy := &proxyHandle{ns: ns} |
| if err := proxy.Start(t, ctx, args...); err != nil { |
| t.Fatal(err) |
| } |
| defer proxy.Stop(ctx) |
| addrs := verifyMount(t, ctx, ns, spec.Proxy) |
| if len(addrs) != 1 { |
| t.Fatalf("failed to lookup proxy") |
| } |
| |
| eps, err := server.Listen(spec) |
| if err != nil { |
| t.Fatal(err) |
| } |
| if err := server.ServeDispatcher("mountpoint/server", testServerDisp{&testServer{}}); err != nil { |
| t.Fatal(err) |
| } |
| |
| // Proxy connections are started asynchronously, so we need to wait.. |
| waitForMountTable := func(ch chan int, expect int) { |
| then := time.Now().Add(time.Minute) |
| for { |
| me, err := ns.Resolve(ctx, name) |
| if err != nil { |
| continue |
| } |
| for i, s := range me.Servers { |
| vlog.Infof("%d: %s", i, s) |
| } |
| if err == nil && len(me.Servers) == expect { |
| ch <- 1 |
| return |
| } |
| if time.Now().After(then) { |
| t.Fatalf("timed out waiting for %d servers, found %d", expect, len(me.Servers)) |
| } |
| time.Sleep(100 * time.Millisecond) |
| } |
| } |
| waitForServerStatus := func(ch chan int, proxy string) { |
| then := time.Now().Add(time.Minute) |
| for { |
| status := server.Status() |
| if len(status.Proxies) == 1 && status.Proxies[0].Proxy == proxy { |
| ch <- 2 |
| return |
| } |
| if time.Now().After(then) { |
| t.Fatalf("timed out") |
| } |
| time.Sleep(100 * time.Millisecond) |
| } |
| } |
| proxyEP, _ := naming.SplitAddressName(addrs[0]) |
| proxiedEP, err := inaming.NewEndpoint(proxyEP) |
| if err != nil { |
| t.Fatalf("unexpected error for %q: %s", proxyEP, err) |
| } |
| proxiedEP.RID = naming.FixedRoutingID(0x555555555) |
| proxiedEP.Blessings = []string{"server"} |
| expectedNames := []string{naming.JoinAddressName(proxiedEP.String(), "suffix")} |
| if hasLocalListener { |
| expectedNames = append(expectedNames, naming.JoinAddressName(eps[0].String(), "suffix")) |
| } |
| |
| // Proxy connetions are created asynchronously, so we wait for the |
| // expected number of endpoints to appear for the specified service name. |
| ch := make(chan int, 2) |
| go waitForMountTable(ch, len(expectedNames)) |
| go waitForServerStatus(ch, spec.Proxy) |
| select { |
| case <-time.After(time.Minute): |
| t.Fatalf("timedout waiting for two entries in the mount table and server status") |
| case i := <-ch: |
| select { |
| case <-time.After(time.Minute): |
| t.Fatalf("timedout waiting for two entries in the mount table or server status") |
| case j := <-ch: |
| if !((i == 1 && j == 2) || (i == 2 && j == 1)) { |
| t.Fatalf("unexpected return values from waiters") |
| } |
| } |
| } |
| |
| status := server.Status() |
| if got, want := status.Proxies[0].Endpoint, proxiedEP; !reflect.DeepEqual(got, want) { |
| t.Fatalf("got %q, want %q", got, want) |
| } |
| |
| got := []string{} |
| for _, s := range verifyMount(t, ctx, ns, name) { |
| got = append(got, s) |
| } |
| sort.Strings(got) |
| sort.Strings(expectedNames) |
| if !reflect.DeepEqual(got, expectedNames) { |
| t.Errorf("got %v, want %v", got, expectedNames) |
| } |
| |
| if hasLocalListener { |
| // Listen will publish both the local and proxied endpoint with the |
| // mount table, given that we're trying to test the proxy, we remove |
| // the local endpoint from the mount table entry! We have to remove both |
| // the tcp and the websocket address. |
| sep := eps[0].String() |
| ns.Unmount(ctx, "mountpoint/server", sep) |
| } |
| |
| addrs = verifyMount(t, ctx, ns, name) |
| if len(addrs) != 1 { |
| t.Fatalf("failed to lookup proxy: addrs %v", addrs) |
| } |
| |
| // Proxied endpoint should be published and RPC should succeed (through proxy). |
| // Additionally, any server authorizaton options must only apply to the end server |
| // and not the proxy. |
| const expected = `method:"Echo",suffix:"suffix",arg:"batman"` |
| if result, err := makeCall(options.ServerPublicKey{serverKey}); result != expected || err != nil { |
| t.Fatalf("Got (%v, %v) want (%v, nil)", result, err, expected) |
| } |
| |
| // Proxy dies, calls should fail and the name should be unmounted. |
| if err := proxy.Stop(ctx); err != nil { |
| t.Fatal(err) |
| } |
| |
| if result, err := makeCall(options.NoRetry{}); err == nil || (!strings.HasPrefix(err.Error(), "RESOLVE") && !strings.Contains(err.Error(), "EOF")) { |
| t.Fatalf(`Got (%v, %v) want ("", "RESOLVE: <err>" or "EOF") as proxy is down`, result, err) |
| } |
| |
| for { |
| if _, err := ns.Resolve(ctx, name); err != nil { |
| break |
| } |
| time.Sleep(10 * time.Millisecond) |
| } |
| verifyMountMissing(t, ctx, ns, name) |
| |
| status = server.Status() |
| if got, want := len(status.Proxies), 1; got != want { |
| t.Logf("Proxies: %v", status.Proxies) |
| t.Fatalf("got %v, want %v", got, want) |
| } |
| if got, want := status.Proxies[0].Proxy, spec.Proxy; got != want { |
| t.Fatalf("got %v, want %v", got, want) |
| } |
| if got, want := verror.ErrorID(status.Proxies[0].Error), verror.ErrNoServers.ID; got != want { |
| t.Fatalf("got %v, want %v", got, want) |
| } |
| |
| // Proxy restarts, calls should eventually start succeeding. |
| if err := proxy.Start(t, ctx, args...); err != nil { |
| t.Fatal(err) |
| } |
| |
| retries := 0 |
| for { |
| if result, err := makeCall(); err == nil { |
| if result != expected { |
| t.Errorf("Got (%v, %v) want (%v, nil)", result, err, expected) |
| } |
| break |
| } else { |
| retries++ |
| if retries > 10 { |
| t.Fatalf("Failed after 10 attempts: err: %s", err) |
| } |
| } |
| } |
| } |
| |
| func verifyMount(t *testing.T, ctx *context.T, ns namespace.T, name string) []string { |
| me, err := ns.Resolve(ctx, name) |
| if err != nil { |
| t.Errorf("%s not found in mounttable", name) |
| return nil |
| } |
| return me.Names() |
| } |
| |
| func verifyMountMissing(t *testing.T, ctx *context.T, ns namespace.T, name string) { |
| if me, err := ns.Resolve(ctx, name); err == nil { |
| names := me.Names() |
| t.Errorf("%s not supposed to be found in mounttable; got %d servers instead: %v", name, len(names), names) |
| } |
| } |