veyron2: make it possible to specify a 'reserved name' dispatcher.
This allows us to move the dependencies on the debug server out
of the core runtime. It is now configured via a profile and as
such is more flexible.
Change-Id: I3ee3e2663c2a02e9d4c4fd4162c1363aac7a9651
diff --git a/profiles/roaming/roaming.go b/profiles/roaming/roaming.go
index 816b4a1..a28feea 100644
--- a/profiles/roaming/roaming.go
+++ b/profiles/roaming/roaming.go
@@ -23,6 +23,9 @@
"veyron.io/veyron/veyron/lib/netstate"
"veyron.io/veyron/veyron/profiles"
"veyron.io/veyron/veyron/profiles/internal"
+ "veyron.io/veyron/veyron/services/mgmt/debug"
+ // TODO(cnicolaou,ashankar): move this into flags.
+ sflag "veyron.io/veyron/veyron/security/flag"
)
const (
@@ -69,6 +72,8 @@
func (p *profile) Init(rt veyron2.Runtime, publisher *config.Publisher) error {
log := rt.Logger()
+ rt.ConfigureReservedName("__debug", debug.NewDispatcher(log.LogDir(), sflag.NewAuthorizerOrDie()))
+
lf := commonFlags.ListenFlags()
ListenSpec = ipc.ListenSpec{
Protocol: lf.ListenProtocol.Protocol,
diff --git a/profiles/static/static.go b/profiles/static/static.go
index ba90318..18b7625 100644
--- a/profiles/static/static.go
+++ b/profiles/static/static.go
@@ -15,6 +15,9 @@
"veyron.io/veyron/veyron/lib/netstate"
"veyron.io/veyron/veyron/profiles"
"veyron.io/veyron/veyron/profiles/internal"
+ "veyron.io/veyron/veyron/services/mgmt/debug"
+ // TODO(cnicolaou,ashankar): move this into flags.
+ sflag "veyron.io/veyron/veyron/security/flag"
)
var (
@@ -55,6 +58,8 @@
func (p *static) Init(rt veyron2.Runtime, _ *config.Publisher) error {
log := rt.Logger()
+ rt.ConfigureReservedName("debug", debug.NewDispatcher(log.LogDir(), sflag.NewAuthorizerOrDie()))
+
lf := commonFlags.ListenFlags()
ListenSpec = ipc.ListenSpec{
Protocol: lf.ListenProtocol.Protocol,
diff --git a/runtimes/google/ipc/debug_test.go b/runtimes/google/ipc/debug_test.go
index 6292622..e05c185 100644
--- a/runtimes/google/ipc/debug_test.go
+++ b/runtimes/google/ipc/debug_test.go
@@ -9,12 +9,14 @@
"veyron.io/veyron/veyron2/naming"
"veyron.io/veyron/veyron2/options"
"veyron.io/veyron/veyron2/services/mounttable/types"
+ "veyron.io/veyron/veyron2/vlog"
"veyron.io/veyron/veyron/lib/stats"
"veyron.io/veyron/veyron/runtimes/google/ipc/stream/manager"
"veyron.io/veyron/veyron/runtimes/google/ipc/stream/sectest"
"veyron.io/veyron/veyron/runtimes/google/ipc/stream/vc"
tnaming "veyron.io/veyron/veyron/runtimes/google/testing/mocks/naming"
+ "veyron.io/veyron/veyron/services/mgmt/debug"
)
func TestDebugServer(t *testing.T) {
@@ -28,10 +30,13 @@
pclient.AddToRoots(bclient) // Client recognizes "server" as a root of blessings.
pclient.BlessingStore().Set(bclient, "server") // Client presents bclient to server
+ debugDisp := debug.NewDispatcher(vlog.Log.LogDir(), nil)
+
sm := manager.InternalNew(naming.FixedRoutingID(0x555555555))
defer sm.Shutdown()
ns := tnaming.NewSimpleNamespace()
- server, err := InternalNewServer(testContext(), sm, ns, vc.LocalPrincipal{pserver})
+
+ server, err := InternalNewServer(testContext(), sm, ns, options.ReservedNameDispatcher{"__debug", debugDisp}, vc.LocalPrincipal{pserver})
if err != nil {
t.Fatalf("InternalNewServer failed: %v", err)
}
diff --git a/runtimes/google/ipc/server.go b/runtimes/google/ipc/server.go
index 08fb74a..10d4263 100644
--- a/runtimes/google/ipc/server.go
+++ b/runtimes/google/ipc/server.go
@@ -28,7 +28,6 @@
"veyron.io/veyron/veyron/runtimes/google/lib/publisher"
inaming "veyron.io/veyron/veyron/runtimes/google/naming"
ivtrace "veyron.io/veyron/veyron/runtimes/google/vtrace"
- "veyron.io/veyron/veyron/services/mgmt/debug"
)
var (
@@ -48,8 +47,7 @@
stoppedChan chan struct{} // closed when the server has been stopped.
ns naming.Namespace
servesMountTable bool
- debugAuthorizer security.Authorizer
- debugDisp ipc.Dispatcher
+ reservedOpt options.ReservedNameDispatcher
// TODO(cnicolaou): add roaming stats to ipcStats
stats *ipcStats // stats for this server.
}
@@ -82,11 +80,10 @@
s.listenerOpts = append(s.listenerOpts, opt)
case options.ServesMountTable:
s.servesMountTable = bool(opt)
- case options.DebugAuthorizer:
- s.debugAuthorizer = opt.Authorizer
+ case options.ReservedNameDispatcher:
+ s.reservedOpt = opt
}
}
- s.debugDisp = debug.NewDispatcher(vlog.Log.LogDir(), s.debugAuthorizer)
return s, nil
}
@@ -622,12 +619,12 @@
// flow that's already connected to the client.
type flowServer struct {
context.T
- server *server // ipc.Server that this flow server belongs to
- disp ipc.Dispatcher // ipc.Dispatcher that will serve RPCs on this flow
- dec *vom.Decoder // to decode requests and args from the client
- enc *vom.Encoder // to encode responses and results to the client
- flow stream.Flow // underlying flow
- debugDisp ipc.Dispatcher // internal debug dispatcher
+ server *server // ipc.Server that this flow server belongs to
+ disp ipc.Dispatcher // ipc.Dispatcher that will serve RPCs on this flow
+ dec *vom.Decoder // to decode requests and args from the client
+ enc *vom.Encoder // to encode responses and results to the client
+ flow stream.Flow // underlying flow
+ reservedOpt options.ReservedNameDispatcher
// Fields filled in during the server invocation.
blessings security.Blessings
@@ -652,11 +649,11 @@
server: server,
disp: disp,
// TODO(toddw): Support different codecs
- dec: vom.NewDecoder(flow),
- enc: vom.NewEncoder(flow),
- flow: flow,
- debugDisp: server.debugDisp,
- discharges: make(map[string]security.Discharge),
+ dec: vom.NewDecoder(flow),
+ enc: vom.NewEncoder(flow),
+ flow: flow,
+ reservedOpt: server.reservedOpt,
+ discharges: make(map[string]security.Discharge),
}
}
@@ -857,14 +854,16 @@
func (fs *flowServer) lookup(name, method string) (ipc.Invoker, security.Authorizer, string, verror.E) {
name = strings.TrimLeft(name, "/")
if method == "Glob" && len(name) == 0 {
- return ipc.ReflectInvoker(&globInvoker{fs}), &acceptAllAuthorizer{}, name, nil
+ return ipc.ReflectInvoker(&globInvoker{fs.reservedOpt.Prefix, fs}), &acceptAllAuthorizer{}, name, nil
}
disp := fs.disp
- if name == ipc.DebugKeyword || strings.HasPrefix(name, ipc.DebugKeyword+"/") {
- name = strings.TrimPrefix(name, ipc.DebugKeyword)
+ prefix := fs.reservedOpt.Prefix
+ if len(prefix) > 0 && (name == prefix || strings.HasPrefix(name, prefix+"/")) {
+ name = strings.TrimPrefix(name, prefix)
name = strings.TrimLeft(name, "/")
- disp = fs.debugDisp
+ disp = fs.reservedOpt.Dispatcher
}
+
if disp != nil {
invoker, auth, err := disp.Lookup(name, method)
switch {
@@ -884,7 +883,8 @@
}
type globInvoker struct {
- fs *flowServer
+ prefix string
+ fs *flowServer
}
// Glob matches the pattern against internal object names if the double-
@@ -895,15 +895,12 @@
if err != nil {
return err
}
- if strings.HasPrefix(pattern, "__") {
+ if strings.HasPrefix(pattern, naming.ReservedNamePrefix) {
var err error
// Match against internal object names.
- internalLeaves := []string{ipc.DebugKeyword}
- for _, leaf := range internalLeaves {
- if ok, _, left := g.MatchInitialSegment(leaf); ok {
- if ierr := i.invokeGlob(call, i.fs.debugDisp, leaf, left.String()); ierr != nil {
- err = ierr
- }
+ if ok, _, left := g.MatchInitialSegment(i.prefix); ok {
+ if ierr := i.invokeGlob(call, i.fs.reservedOpt.Dispatcher, i.prefix, left.String()); ierr != nil {
+ err = ierr
}
}
return err
@@ -938,6 +935,7 @@
if err != nil {
return err
}
+
if len(results) != 1 {
return verror.BadArgf("unexpected number of results. Got %d, want 1", len(results))
}
diff --git a/runtimes/google/rt/ipc.go b/runtimes/google/rt/ipc.go
index 293657a..4923176 100644
--- a/runtimes/google/rt/ipc.go
+++ b/runtimes/google/rt/ipc.go
@@ -93,6 +93,11 @@
}
// Add the option that provides the principal to the server.
otherOpts = append(otherOpts, vc.LocalPrincipal{rt.principal})
+ if rt.reservedDisp != nil {
+ ropts := options.ReservedNameDispatcher{rt.reservedPrefix, rt.reservedDisp}
+ otherOpts = append(otherOpts, ropts)
+ otherOpts = append(otherOpts, rt.reservedOpts...)
+ }
ctx := rt.NewContext()
return iipc.InternalNewServer(ctx, sm, ns, otherOpts...)
}
diff --git a/runtimes/google/rt/rt.go b/runtimes/google/rt/rt.go
index 8383b6d..7bb5e3b 100644
--- a/runtimes/google/rt/rt.go
+++ b/runtimes/google/rt/rt.go
@@ -5,6 +5,7 @@
"fmt"
"os"
"path/filepath"
+ "strings"
"sync"
"veyron.io/veyron/veyron2"
@@ -30,17 +31,20 @@
type vrt struct {
mu sync.Mutex
- profile veyron2.Profile
- publisher *config.Publisher
- sm []stream.Manager // GUARDED_BY(mu)
- ns naming.Namespace
- signals chan os.Signal
- principal security.Principal
- client ipc.Client
- mgmt *mgmtImpl
- flags flags.RuntimeFlags
- nServers int // GUARDED_BY(mu)
- cleaningUp bool // GUARDED_BY(mu)
+ profile veyron2.Profile
+ publisher *config.Publisher
+ sm []stream.Manager // GUARDED_BY(mu)
+ ns naming.Namespace
+ signals chan os.Signal
+ principal security.Principal
+ client ipc.Client
+ mgmt *mgmtImpl
+ flags flags.RuntimeFlags
+ reservedDisp ipc.Dispatcher
+ reservedPrefix string
+ reservedOpts []ipc.ServerOpt
+ nServers int // GUARDED_BY(mu)
+ cleaningUp bool // GUARDED_BY(mu)
lang i18n.LangID // Language, from environment variables.
program string // Program name, from os.Args[0].
@@ -118,6 +122,11 @@
return nil, fmt.Errorf("failed to get child handle: %s", err)
}
+ rt.publisher = config.NewPublisher()
+ if err := rt.profile.Init(rt, rt.publisher); err != nil {
+ return nil, err
+ }
+
// TODO(caprita, cnicolaou): how is this to be configured?
// Can it ever be anything other than a localhost/loopback address?
listenSpec := ipc.ListenSpec{Protocol: "tcp", Address: "127.0.0.1:0"}
@@ -125,11 +134,6 @@
return nil, err
}
- rt.publisher = config.NewPublisher()
- if err := rt.profile.Init(rt, rt.publisher); err != nil {
- return nil, err
- }
-
vlog.VI(2).Infof("rt.Init done")
return rt, nil
}
@@ -138,8 +142,18 @@
return rt.publisher
}
-func (r *vrt) Profile() veyron2.Profile {
- return r.profile
+func (rt *vrt) Profile() veyron2.Profile {
+ return rt.profile
+}
+
+func (rt *vrt) ConfigureReservedName(prefix string, server ipc.Dispatcher, opts ...ipc.ServerOpt) {
+ rt.mu.Lock()
+ defer rt.mu.Unlock()
+ rt.reservedDisp = server
+ prefix = strings.TrimLeft(prefix, "_")
+ rt.reservedPrefix = naming.ReservedNamePrefix + prefix
+ rt.reservedOpts = make([]ipc.ServerOpt, 0, len(opts))
+ copy(rt.reservedOpts, opts)
}
func (rt *vrt) Cleanup() {
diff --git a/runtimes/google/testing/mocks/runtime/panic_runtime.go b/runtimes/google/testing/mocks/runtime/panic_runtime.go
index 215fb5c..cccc4d7 100644
--- a/runtimes/google/testing/mocks/runtime/panic_runtime.go
+++ b/runtimes/google/testing/mocks/runtime/panic_runtime.go
@@ -45,4 +45,7 @@
func (*PanicRuntime) AdvanceGoal(delta int) { panic(badRuntime) }
func (*PanicRuntime) AdvanceProgress(delta int) { panic(badRuntime) }
func (*PanicRuntime) TrackTask(chan<- veyron2.Task) { panic(badRuntime) }
-func (*PanicRuntime) Cleanup() { panic(badRuntime) }
+func (*PanicRuntime) ConfigureReservedName(string, ipc.Dispatcher, ...ipc.ServerOpt) {
+ panic(badRuntime)
+}
+func (*PanicRuntime) Cleanup() { panic(badRuntime) }
diff --git a/services/mgmt/lib/toplevelglob/invoker.go b/services/mgmt/lib/toplevelglob/invoker.go
index fd605a4..16d1676 100644
--- a/services/mgmt/lib/toplevelglob/invoker.go
+++ b/services/mgmt/lib/toplevelglob/invoker.go
@@ -46,6 +46,9 @@
if err != nil {
return err
}
+ if invoker == nil {
+ return verror.BadArgf("failed to find invoker for %q", leaf)
+ }
argptrs := []interface{}{&pattern}
leafCall := &localServerCall{call, leaf}
results, err := invoker.Invoke("Glob", leafCall, argptrs)
diff --git a/services/mgmt/node/impl/proxy_invoker_test.go b/services/mgmt/node/impl/proxy_invoker_test.go
index 7f84099..dd87323 100644
--- a/services/mgmt/node/impl/proxy_invoker_test.go
+++ b/services/mgmt/node/impl/proxy_invoker_test.go
@@ -11,8 +11,6 @@
"veyron.io/veyron/veyron2/security"
"veyron.io/veyron/veyron2/services/mgmt/stats"
"veyron.io/veyron/veyron2/services/mounttable"
-
- "veyron.io/veyron/veyron/profiles"
)
func TestProxyInvoker(t *testing.T) {
@@ -24,7 +22,8 @@
t.Fatalf("NewServer: %v", err)
}
defer server1.Stop()
- ep1, err := server1.Listen(profiles.LocalListenSpec)
+ localSpec := ipc.ListenSpec{Protocol: "tcp", Address: "127.0.0.1:0"}
+ ep1, err := server1.Listen(localSpec)
if err != nil {
t.Fatalf("Listen: %v", err)
}
@@ -38,7 +37,7 @@
t.Fatalf("NewServer: %v", err)
}
defer server2.Stop()
- ep2, err := server2.Listen(profiles.LocalListenSpec)
+ ep2, err := server2.Listen(localSpec)
if err != nil {
t.Fatalf("Listen: %v", err)
}
@@ -79,7 +78,7 @@
}
stream, err := c.Glob(rt.R().NewContext(), pattern)
if err != nil {
- t.Errorf("Glob failed: %v", err)
+ t.Fatalf("Glob failed: %v", err)
}
results := []string{}
iterator := stream.RecvStream()
