Merge "discovery: add mdns plugin"
diff --git a/cmd/vdl/v23_internal_test.go b/cmd/vdl/v23_internal_test.go
index ae59080..a80e0ec 100644
--- a/cmd/vdl/v23_internal_test.go
+++ b/cmd/vdl/v23_internal_test.go
@@ -12,9 +12,11 @@
"testing"
"v.io/x/ref/test"
+ "v.io/x/ref/test/modules"
)
func TestMain(m *testing.M) {
test.Init()
+ modules.DispatchAndExitIfChild()
os.Exit(m.Run())
}
diff --git a/cmd/vdl/vdl_test.go b/cmd/vdl/vdl_test.go
index da17bb9..7a127cd 100644
--- a/cmd/vdl/vdl_test.go
+++ b/cmd/vdl/vdl_test.go
@@ -8,12 +8,11 @@
"bytes"
"fmt"
"io/ioutil"
- "os"
"path/filepath"
"strings"
"testing"
- "v.io/x/lib/cmdline"
+ "v.io/x/ref/test/v23tests"
)
const (
@@ -25,18 +24,15 @@
// Compares generated VDL files against the copy in the repo.
func TestVDLGenerator(t *testing.T) {
+ testEnv := v23tests.New(t)
+ defer testEnv.Cleanup()
+ vdlBin := testEnv.BuildGoPkg("v.io/x/ref/cmd/vdl")
+
// Use vdl to generate Go code from input, into a temporary directory.
- outDir, err := ioutil.TempDir("", "vdltest")
- if err != nil {
- t.Fatalf("TempDir() failed: %v", err)
- }
- defer os.RemoveAll(outDir)
+ outDir := testEnv.NewTempDir("")
// TODO(toddw): test the generated java and javascript files too.
outOpt := fmt.Sprintf("--go-out-dir=%s", outDir)
- env := cmdline.EnvFromOS()
- if err := cmdline.ParseAndRun(cmdVDL, env, []string{"generate", "--lang=go", outOpt, testDir}); err != nil {
- t.Fatalf("Execute() failed: %v", err)
- }
+ vdlBin.Run("generate", "--lang=go", outOpt, testDir)
// Check that each *.vdl.go file in the testDir matches the generated output.
entries, err := ioutil.ReadDir(testDir)
if err != nil {
diff --git a/runtime/internal/flow/conn/conn.go b/runtime/internal/flow/conn/conn.go
index 76a5206..a7991c7 100644
--- a/runtime/internal/flow/conn/conn.go
+++ b/runtime/internal/flow/conn/conn.go
@@ -159,6 +159,12 @@
// RemoteEndpoint returns the remote vanadium Endpoint
func (c *Conn) RemoteEndpoint() naming.Endpoint { return c.remote }
+// LocalBlessings returns the local blessings.
+func (c *Conn) LocalBlessings() security.Blessings { return c.lBlessings }
+
+// RemoteBlessings returns the remote blessings.
+func (c *Conn) RemoteBlessings() security.Blessings { return c.rBlessings }
+
// CommonVersion returns the RPCVersion negotiated between the local and remote endpoints.
func (c *Conn) CommonVersion() version.RPCVersion { return c.version }
diff --git a/runtime/internal/flow/manager/conncache.go b/runtime/internal/flow/manager/conncache.go
index cdb81a7..92f8348 100644
--- a/runtime/internal/flow/manager/conncache.go
+++ b/runtime/internal/flow/manager/conncache.go
@@ -183,6 +183,9 @@
// is identified by the provided rid. nil is returned if there is no such Conn.
// FindWithRoutingID will return an error iff the cache is closed.
func (c *ConnCache) FindWithRoutingID(rid naming.RoutingID) (*conn.Conn, error) {
+ if rid == naming.NullRoutingID {
+ return nil, nil
+ }
defer c.mu.Unlock()
c.mu.Lock()
if c.addrCache == nil {
diff --git a/runtime/internal/flow/manager/manager.go b/runtime/internal/flow/manager/manager.go
index 6ec43fb..d39b1ad 100644
--- a/runtime/internal/flow/manager/manager.go
+++ b/runtime/internal/flow/manager/manager.go
@@ -266,7 +266,11 @@
// The flow.Manager associated with ctx must be the receiver of the method,
// otherwise an error is returned.
func (m *manager) Dial(ctx *context.T, remote naming.Endpoint, fn flow.BlessingsForPeer) (flow.Flow, error) {
- return m.internalDial(ctx, remote, fn, &flowHandler{q: m.q, closed: m.closed})
+ var fh conn.FlowHandler
+ if m.rid != naming.NullRoutingID {
+ fh = &flowHandler{q: m.q, closed: m.closed}
+ }
+ return m.internalDial(ctx, remote, fn, fh)
}
func (m *manager) internalDial(ctx *context.T, remote naming.Endpoint, fn flow.BlessingsForPeer, fh conn.FlowHandler) (flow.Flow, error) {
diff --git a/runtime/internal/flow/manager/manager_test.go b/runtime/internal/flow/manager/manager_test.go
index 41c470f..62403da 100644
--- a/runtime/internal/flow/manager/manager_test.go
+++ b/runtime/internal/flow/manager/manager_test.go
@@ -15,6 +15,7 @@
"v.io/v23/naming"
_ "v.io/x/ref/runtime/factories/fake"
+ "v.io/x/ref/runtime/internal/flow/conn"
"v.io/x/ref/runtime/internal/flow/flowtest"
"v.io/x/ref/test"
)
@@ -82,6 +83,29 @@
testFlows(t, ctx, am, dm, flowtest.BlessingsForPeer)
}
+func TestNullClientBlessings(t *testing.T) {
+ ctx, shutdown := v23.Init()
+ defer shutdown()
+
+ am := New(ctx, naming.FixedRoutingID(0x5555))
+ if err := am.Listen(ctx, "tcp", "127.0.0.1:0"); err != nil {
+ t.Fatal(err)
+ }
+ dm := New(ctx, naming.NullRoutingID)
+ _, af := testFlows(t, ctx, dm, am, flowtest.BlessingsForPeer)
+ // Ensure that the remote blessings of the underlying conn of the accepted flow are zero.
+ if rBlessings := af.Conn().(*conn.Conn).RemoteBlessings(); !rBlessings.IsZero() {
+ t.Errorf("got %v, want zero-value blessings", rBlessings)
+ }
+ dm = New(ctx, naming.FixedRoutingID(0x1111))
+ _, af = testFlows(t, ctx, dm, am, flowtest.BlessingsForPeer)
+ // Ensure that the remote blessings of the underlying conn of the accepted flow are
+ // non-zero if we did specify a RoutingID.
+ if rBlessings := af.Conn().(*conn.Conn).RemoteBlessings(); rBlessings.IsZero() {
+ t.Errorf("got %v, want non-zero blessings", rBlessings)
+ }
+}
+
func testFlows(t *testing.T, ctx *context.T, dm, am flow.Manager, bFn flow.BlessingsForPeer) (df, af flow.Flow) {
eps := am.ListeningEndpoints()
if len(eps) == 0 {
diff --git a/runtime/internal/rt/runtime.go b/runtime/internal/rt/runtime.go
index f3452b0..c4f358e 100644
--- a/runtime/internal/rt/runtime.go
+++ b/runtime/internal/rt/runtime.go
@@ -190,7 +190,13 @@
}
// Add the flow.Manager to the context.
- ctx, _, err = r.ExperimentalWithNewFlowManager(ctx)
+ // This initial Flow Manager can only be used as a client.
+ ctx, _, err = r.setNewClientFlowManager(ctx)
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ // Add the Client to the context.
+ ctx, _, err = r.WithNewClient(ctx)
if err != nil {
return nil, nil, nil, err
}
@@ -321,21 +327,22 @@
return sm, nil
}
-func newFlowManager(ctx *context.T) (flow.Manager, error) {
- rid, err := naming.NewRoutingID()
- if err != nil {
- return nil, err
- }
- return manager.New(ctx, rid), nil
+func (r *Runtime) setNewClientFlowManager(ctx *context.T) (*context.T, flow.Manager, error) {
+ return r.setNewFlowManager(ctx, naming.NullRoutingID)
}
-func (r *Runtime) setNewFlowManager(ctx *context.T) (*context.T, flow.Manager, error) {
- fm, err := newFlowManager(ctx)
+func (r *Runtime) setNewBidiFlowManager(ctx *context.T) (*context.T, flow.Manager, error) {
+ rid, err := naming.NewRoutingID()
if err != nil {
return nil, nil, err
}
+ return r.setNewFlowManager(ctx, rid)
+}
+
+func (r *Runtime) setNewFlowManager(ctx *context.T, rid naming.RoutingID) (*context.T, flow.Manager, error) {
+ fm := manager.New(ctx, rid)
// TODO(mattr): How can we close a flow manager.
- if err = r.addChild(ctx, fm, func() {}); err != nil {
+ if err := r.addChild(ctx, fm, func() {}); err != nil {
return ctx, nil, err
}
newctx := context.WithValue(ctx, flowManagerKey, fm)
@@ -396,7 +403,12 @@
if newctx, err = r.setNewStreamManager(newctx); err != nil {
return ctx, err
}
- if newctx, _, err = r.setNewFlowManager(newctx); err != nil {
+ if rid := r.ExperimentalGetFlowManager(newctx).RoutingID(); rid == naming.NullRoutingID {
+ newctx, _, err = r.setNewClientFlowManager(newctx)
+ } else {
+ newctx, _, err = r.setNewBidiFlowManager(newctx)
+ }
+ if err != nil {
return ctx, err
}
if newctx, _, err = r.setNewNamespace(newctx, r.GetNamespace(ctx).Roots()...); err != nil {
@@ -559,7 +571,7 @@
func (r *Runtime) ExperimentalWithNewFlowManager(ctx *context.T) (*context.T, flow.Manager, error) {
defer apilog.LogCall(ctx)(ctx) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
- newctx, m, err := r.setNewFlowManager(ctx)
+ newctx, m, err := r.setNewBidiFlowManager(ctx)
if err != nil {
return ctx, nil, err
}
diff --git a/runtime/internal/rt/runtime_test.go b/runtime/internal/rt/runtime_test.go
index 9b7f6cd..fc70a76 100644
--- a/runtime/internal/rt/runtime_test.go
+++ b/runtime/internal/rt/runtime_test.go
@@ -161,6 +161,9 @@
if oldman == nil {
t.Error("ExperimentalGetFlowManager should have returned a non-nil value")
}
+ if rid := oldman.RoutingID(); rid != naming.NullRoutingID {
+ t.Errorf("Initial flow.Manager should have NullRoutingID, got %v", rid)
+ }
newctx, newman, err := r.ExperimentalWithNewFlowManager(ctx)
if err != nil || newman == nil || newman == oldman {
t.Fatalf("Could not create flow manager: %v", err)
@@ -172,4 +175,7 @@
if man != newman || man == oldman {
t.Error("ExperimentalWithNewFlowManager didn't update the context properly")
}
+ if man.RoutingID() == naming.NullRoutingID {
+ t.Error("Newly created flow.Manager should not have NullRoutingID")
+ }
}
diff --git a/services/device/device/publish.go b/services/device/device/publish.go
index 5862d86..f7d84a0 100644
--- a/services/device/device/publish.go
+++ b/services/device/device/publish.go
@@ -140,8 +140,7 @@
// TODO(caprita): use the profile detection machinery and/or let user
// specify the profile by hand.
profile := fmt.Sprintf("%s-%s", goosFlag, goarchFlag)
- // TODO(caprita): use a label e.g. "prod" instead of "0".
- appVON := naming.Join(applicationService, envelopeName, "0")
+ appVON := naming.Join(applicationService, envelopeName)
appClient := repository.ApplicationClient(appVON)
envelope, err := appClient.Match(ctx, []string{profile})
if verror.ErrorID(err) == verror.ErrNoExist.ID {
@@ -173,8 +172,16 @@
envelope.Binary.Signature = security.Signature{}
envelope.Publisher = security.Blessings{}
}
-
- if err := appClient.Put(ctx, profile, envelope, true); err != nil {
+ appVON = naming.Join(appVON, timestamp)
+ appClient = repository.ApplicationClient(appVON)
+ if err := appClient.Put(ctx, profile, envelope, false); err != nil {
+ // NOTE(caprita): We don't retry if an envelope already exists
+ // at the versioned name, as we do when uploading binaries. In
+ // the case of binaries, it's likely that the same binary is
+ // uploaded more than once in a given second, due to apps
+ // sharing the same binary. The scenarios where the same app is
+ // published repeatedly in a short time-frame are expected to be
+ // rare, and the operator can retry manually in such cases.
return err
}
fmt.Fprintf(env.Stdout, "Published %q\n", appVON)
diff --git a/services/device/dmrun/dmrun.go b/services/device/dmrun/dmrun.go
index c829566..d816542 100644
--- a/services/device/dmrun/dmrun.go
+++ b/services/device/dmrun/dmrun.go
@@ -234,6 +234,10 @@
cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", ref.EnvCredentials, creds))
} else if agentCreds := os.Getenv(ref.EnvAgentEndpoint); len(agentCreds) > 0 {
cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", ref.EnvAgentEndpoint, agentCreds))
+ } else if agentCreds := os.Getenv(ref.EnvAgentPath); len(agentCreds) > 0 {
+ cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", ref.EnvAgentPath, agentCreds))
+ } else {
+ fmt.Fprintf(os.Stderr, "WARNING: no credentials found. You'll probably have authorization issues later on.\n")
}
}
diff --git a/services/device/mgmt_v23_test.go b/services/device/mgmt_v23_test.go
index 5940bcc..78a7795 100644
--- a/services/device/mgmt_v23_test.go
+++ b/services/device/mgmt_v23_test.go
@@ -350,14 +350,14 @@
// Allow publishers to create and update envelopes
deviceBin.Run("acl", "set", appDName, "root/a", "Read,Write,Resolve")
- sampleAppName := appDName + "/testapp/0"
+ sampleAppName := appDName + "/testapp"
appPubName := "testbinaryd"
appEnvelopeFilename := filepath.Join(workDir, "app.envelope")
appEnvelope := fmt.Sprintf("{\"Title\":\"BINARYD\", \"Args\":[\"--name=%s\", \"--root-dir=./binstore\", \"--v23.tcp.address=127.0.0.1:0\", \"--http=127.0.0.1:0\"], \"Binary\":{\"File\":%q}, \"Env\":[]}", appPubName, sampleAppBinName)
ioutil.WriteFile(appEnvelopeFilename, []byte(appEnvelope), 0666)
defer os.Remove(appEnvelopeFilename)
- output := applicationBin.Run("put", sampleAppName, deviceProfile, appEnvelopeFilename)
+ output := applicationBin.Run("put", sampleAppName+"/0", deviceProfile, appEnvelopeFilename)
if got, want := output, fmt.Sprintf("Application envelope added for profile %s.", deviceProfile); got != want {
i.Fatalf("got %q, want %q", got, want)
}
diff --git a/services/syncbase/server/app.go b/services/syncbase/server/app.go
index e07094c..9bcec0d 100644
--- a/services/syncbase/server/app.go
+++ b/services/syncbase/server/app.go
@@ -9,7 +9,6 @@
"sync"
"v.io/v23/context"
- "v.io/v23/glob"
"v.io/v23/rpc"
"v.io/v23/security/access"
wire "v.io/v23/services/syncbase"
@@ -42,6 +41,8 @@
////////////////////////////////////////
// RPC methods
+// TODO(sadovsky): Implement Glob__ or GlobChildren__.
+
// TODO(sadovsky): Require the app name to match the client's blessing name.
// I.e. reserve names at the app level of the hierarchy.
func (a *app) Create(ctx *context.T, call rpc.ServerCall, perms access.Permissions) error {
@@ -83,17 +84,17 @@
return data.Perms, util.FormatVersion(data.Version), nil
}
-func (a *app) GlobChildren__(ctx *context.T, call rpc.GlobChildrenServerCall, matcher *glob.Element) error {
+func (a *app) ListDatabases(ctx *context.T, call rpc.ServerCall) ([]string, error) {
if !a.exists {
- return verror.New(verror.ErrNoExist, ctx, a.name)
+ return nil, verror.New(verror.ErrNoExist, ctx, a.name)
}
// Check perms.
sn := a.s.st.NewSnapshot()
if err := util.GetWithAuth(ctx, call, sn, a.stKey(), &appData{}); err != nil {
sn.Abort()
- return err
+ return nil, err
}
- return util.Glob(ctx, call, matcher, sn, sn.Abort, util.JoinKeyParts(util.DbInfoPrefix, a.name))
+ return util.ListChildren(ctx, call, sn, util.JoinKeyParts(util.DbInfoPrefix, a.name))
}
////////////////////////////////////////
diff --git a/services/syncbase/server/dispatcher.go b/services/syncbase/server/dispatcher.go
index dac3d34..06a6757 100644
--- a/services/syncbase/server/dispatcher.go
+++ b/services/syncbase/server/dispatcher.go
@@ -34,21 +34,26 @@
func (disp *dispatcher) Lookup(ctx *context.T, suffix string) (interface{}, security.Authorizer, error) {
suffix = strings.TrimPrefix(suffix, "/")
- parts := strings.SplitN(suffix, "/", 2)
if len(suffix) == 0 {
return wire.ServiceServer(disp.s), auth, nil
}
+ // If the first slash-separated component of suffix is SyncbaseSuffix,
+ // dispatch to the sync module.
+ parts := strings.SplitN(suffix, "/", 2)
if parts[0] == util.SyncbaseSuffix {
return interfaces.SyncServer(disp.s.sync), auth, nil
}
+ // Otherwise, split on NameSepWithSlashes to get hierarchy component names.
+ parts = strings.SplitN(suffix, pubutil.NameSepWithSlashes, 2)
+
// Validate all key atoms up front, so that we can avoid doing so in all our
// method implementations.
appName := parts[0]
if !pubutil.ValidName(appName) {
- return nil, nil, wire.NewErrInvalidName(nil, suffix)
+ return nil, nil, wire.NewErrInvalidName(ctx, suffix)
}
aExists := false
@@ -74,7 +79,7 @@
// All database, table, and row methods require the app to exist. If it
// doesn't, abort early.
if !aExists {
- return nil, nil, verror.New(verror.ErrNoExist, nil, a.name)
+ return nil, nil, verror.New(verror.ErrNoExist, ctx, a.name)
}
// Note, it's possible for the app to be deleted concurrently with downstream
diff --git a/services/syncbase/server/nosql/database.go b/services/syncbase/server/nosql/database.go
index 1755076..f6e14e1 100644
--- a/services/syncbase/server/nosql/database.go
+++ b/services/syncbase/server/nosql/database.go
@@ -7,13 +7,11 @@
import (
"math/rand"
"path"
- "strconv"
"strings"
"sync"
"time"
"v.io/v23/context"
- "v.io/v23/glob"
"v.io/v23/rpc"
"v.io/v23/security/access"
wire "v.io/v23/services/syncbase/nosql"
@@ -129,6 +127,8 @@
////////////////////////////////////////
// RPC methods
+// TODO(sadovsky): Implement Glob__ or GlobChildren__.
+
func (d *databaseReq) Create(ctx *context.T, call rpc.ServerCall, metadata *wire.SchemaMetadata, perms access.Permissions) error {
if d.exists {
return verror.New(verror.ErrExist, ctx, d.name)
@@ -178,24 +178,24 @@
d.mu.Lock()
defer d.mu.Unlock()
var id uint64
- var batchType string
+ var batchType util.BatchType
for {
id = uint64(rng.Int63())
if bo.ReadOnly {
if _, ok := d.sns[id]; !ok {
d.sns[id] = d.st.NewSnapshot()
- batchType = "sn"
+ batchType = util.BatchTypeSn
break
}
} else {
if _, ok := d.txs[id]; !ok {
d.txs[id] = d.st.NewTransaction()
- batchType = "tx"
+ batchType = util.BatchTypeTx
break
}
}
}
- return strings.Join([]string{d.name, batchType, strconv.FormatUint(id, 10)}, util.BatchSep), nil
+ return strings.Join([]string{d.name, util.JoinBatchInfo(batchType, id)}, util.BatchSep), nil
}
func (d *databaseReq) Commit(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error {
@@ -251,10 +251,20 @@
}
func (d *databaseReq) Exec(ctx *context.T, call wire.DatabaseExecServerCall, schemaVersion int32, q string) error {
+ if !d.exists {
+ return verror.New(verror.ErrNoExist, ctx, d.name)
+ }
if err := d.checkSchemaVersion(ctx, schemaVersion); err != nil {
return err
}
- impl := func(headers []string, rs query.ResultStream, err error) error {
+ impl := func(sntx store.SnapshotOrTransaction) error {
+ db := &queryDb{
+ ctx: ctx,
+ call: call,
+ req: d,
+ sntx: sntx,
+ }
+ headers, rs, err := exec.Exec(db, q)
if err != nil {
return err
}
@@ -275,23 +285,13 @@
}
return rs.Err()
}
- var sntx store.SnapshotOrTransaction
if d.batchId != nil {
- sntx = d.batchReader()
+ return impl(d.batchReader())
} else {
- sntx = d.st.NewSnapshot()
+ sntx := d.st.NewSnapshot()
defer sntx.Abort()
+ return impl(sntx)
}
- // queryDb implements the query.Database interface, which is needed by the
- // exec.Exec function.
- db := &queryDb{
- ctx: ctx,
- call: call,
- req: d,
- sntx: sntx,
- }
-
- return impl(exec.Exec(db, q))
}
func (d *databaseReq) SetPermissions(ctx *context.T, call rpc.ServerCall, perms access.Permissions, version string) error {
@@ -318,20 +318,25 @@
return data.Perms, util.FormatVersion(data.Version), nil
}
-func (d *databaseReq) GlobChildren__(ctx *context.T, call rpc.GlobChildrenServerCall, matcher *glob.Element) error {
+func (d *databaseReq) ListTables(ctx *context.T, call rpc.ServerCall) ([]string, error) {
if !d.exists {
- return verror.New(verror.ErrNoExist, ctx, d.name)
+ return nil, verror.New(verror.ErrNoExist, ctx, d.name)
+ }
+ impl := func(sntx store.SnapshotOrTransaction) ([]string, error) {
+ // Check perms.
+ if err := util.GetWithAuth(ctx, call, sntx, d.stKey(), &databaseData{}); err != nil {
+ sntx.Abort()
+ return nil, err
+ }
+ return util.ListChildren(ctx, call, sntx, util.TablePrefix)
}
if d.batchId != nil {
- return wire.NewErrBoundToBatch(ctx)
+ return impl(d.batchReader())
+ } else {
+ sntx := d.st.NewSnapshot()
+ defer sntx.Abort()
+ return impl(sntx)
}
- // Check perms.
- sn := d.st.NewSnapshot()
- if err := util.GetWithAuth(ctx, call, sn, d.stKey(), &databaseData{}); err != nil {
- sn.Abort()
- return err
- }
- return util.Glob(ctx, call, matcher, sn, sn.Abort, util.TablePrefix)
}
////////////////////////////////////////
diff --git a/services/syncbase/server/nosql/database_watch.go b/services/syncbase/server/nosql/database_watch.go
index e201987..374cab4 100644
--- a/services/syncbase/server/nosql/database_watch.go
+++ b/services/syncbase/server/nosql/database_watch.go
@@ -172,7 +172,7 @@
}
parts := util.SplitKeyParts(opKey)
// TODO(rogulenko): Currently we process only rows, i.e. keys of the form
- // $row:xxx:yyy. Consider processing other keys.
+ // <RowPrefix>:xxx:yyy. Consider processing other keys.
if len(parts) != 3 || parts[0] != util.RowPrefix {
continue
}
@@ -188,7 +188,7 @@
continue
}
change := watch.Change{
- Name: naming.Join(table, row),
+ Name: naming.Join(table, pubutil.NameSep, row),
Continued: true,
}
switch op := logEntry.Op.(type) {
diff --git a/services/syncbase/server/nosql/dispatcher.go b/services/syncbase/server/nosql/dispatcher.go
index 269a7cd..4ec16c0 100644
--- a/services/syncbase/server/nosql/dispatcher.go
+++ b/services/syncbase/server/nosql/dispatcher.go
@@ -5,7 +5,6 @@
package nosql
import (
- "strconv"
"strings"
"v.io/v23/context"
@@ -34,25 +33,25 @@
// RPC method implementations to perform proper authorization.
var auth security.Authorizer = security.AllowEveryone()
-func (disp *dispatcher) Lookup(_ *context.T, suffix string) (interface{}, security.Authorizer, error) {
+func (disp *dispatcher) Lookup(ctx *context.T, suffix string) (interface{}, security.Authorizer, error) {
suffix = strings.TrimPrefix(suffix, "/")
- parts := strings.Split(suffix, "/")
+ parts := strings.Split(suffix, pubutil.NameSepWithSlashes)
if len(parts) == 0 {
vlog.Fatal("invalid nosql.dispatcher Lookup")
}
- dParts := strings.Split(parts[0], util.BatchSep)
+ dParts := strings.SplitN(parts[0], util.BatchSep, 2)
dName := dParts[0]
// Validate all key atoms up front, so that we can avoid doing so in all our
// method implementations.
if !pubutil.ValidName(dName) {
- return nil, nil, wire.NewErrInvalidName(nil, suffix)
+ return nil, nil, wire.NewErrInvalidName(ctx, suffix)
}
for _, s := range parts[1:] {
if !pubutil.ValidName(s) {
- return nil, nil, wire.NewErrInvalidName(nil, suffix)
+ return nil, nil, wire.NewErrInvalidName(ctx, suffix)
}
}
@@ -75,8 +74,10 @@
}
dReq := &databaseReq{database: d}
- if !setBatchFields(dReq, dParts) {
- return nil, nil, wire.NewErrInvalidName(nil, suffix)
+ if len(dParts) == 2 {
+ if !setBatchFields(dReq, dParts[1]) {
+ return nil, nil, wire.NewErrInvalidName(ctx, suffix)
+ }
}
if len(parts) == 1 {
return nosqlWire.DatabaseServer(dReq), auth, nil
@@ -85,7 +86,7 @@
// All table and row methods require the database to exist. If it doesn't,
// abort early.
if !dExists {
- return nil, nil, verror.New(verror.ErrNoExist, nil, d.name)
+ return nil, nil, verror.New(verror.ErrNoExist, ctx, d.name)
}
// Note, it's possible for the database to be deleted concurrently with
@@ -108,20 +109,14 @@
return nosqlWire.RowServer(rReq), auth, nil
}
- return nil, nil, verror.NewErrNoExist(nil)
+ return nil, nil, verror.NewErrNoExist(ctx)
}
// setBatchFields sets the batch-related fields in databaseReq based on the
-// value of dParts, the parts of the database name component. It returns false
-// if dParts is malformed.
-func setBatchFields(d *databaseReq, dParts []string) bool {
- if len(dParts) == 1 {
- return true
- }
- if len(dParts) != 3 {
- return false
- }
- batchId, err := strconv.ParseUint(dParts[2], 0, 64)
+// value of batchInfo (suffix of the database name component). It returns false
+// if batchInfo is malformed.
+func setBatchFields(d *databaseReq, batchInfo string) bool {
+ batchType, batchId, err := util.SplitBatchInfo(batchInfo)
if err != nil {
return false
}
@@ -129,13 +124,11 @@
d.mu.Lock()
defer d.mu.Unlock()
var ok bool
- switch dParts[1] {
- case "sn":
+ switch batchType {
+ case util.BatchTypeSn:
d.sn, ok = d.sns[batchId]
- case "tx":
+ case util.BatchTypeTx:
d.tx, ok = d.txs[batchId]
- default:
- return false
}
return ok
}
diff --git a/services/syncbase/server/nosql/row.go b/services/syncbase/server/nosql/row.go
index 758c58a..1efe739 100644
--- a/services/syncbase/server/nosql/row.go
+++ b/services/syncbase/server/nosql/row.go
@@ -26,6 +26,8 @@
////////////////////////////////////////
// RPC methods
+// TODO(sadovsky): Implement Glob__ or GlobChildren__.
+
func (r *rowReq) Exists(ctx *context.T, call rpc.ServerCall, schemaVersion int32) (bool, error) {
_, err := r.Get(ctx, call, schemaVersion)
return util.ErrorToExists(err)
diff --git a/services/syncbase/server/nosql/table.go b/services/syncbase/server/nosql/table.go
index 27097b6..8d68d7a 100644
--- a/services/syncbase/server/nosql/table.go
+++ b/services/syncbase/server/nosql/table.go
@@ -8,7 +8,6 @@
"strings"
"v.io/v23/context"
- "v.io/v23/glob"
"v.io/v23/rpc"
"v.io/v23/security/access"
wire "v.io/v23/services/syncbase/nosql"
@@ -31,6 +30,8 @@
////////////////////////////////////////
// RPC methods
+// TODO(sadovsky): Implement Glob__ or GlobChildren__.
+
func (t *tableReq) Create(ctx *context.T, call rpc.ServerCall, schemaVersion int32, perms access.Permissions) error {
if t.d.batchId != nil {
return wire.NewErrBoundToBatch(ctx)
@@ -316,26 +317,6 @@
}
}
-func (t *tableReq) GlobChildren__(ctx *context.T, call rpc.GlobChildrenServerCall, matcher *glob.Element) error {
- impl := func(sntx store.SnapshotOrTransaction, closeSntx func() error) error {
- // Check perms.
- if err := t.checkAccess(ctx, call, sntx, ""); err != nil {
- closeSntx()
- return err
- }
- // TODO(rogulenko): Check prefix permissions for children.
- return util.Glob(ctx, call, matcher, sntx, closeSntx, util.JoinKeyParts(util.RowPrefix, t.name))
- }
- if t.d.batchId != nil {
- return impl(t.d.batchReader(), func() error {
- return nil
- })
- } else {
- sn := t.d.st.NewSnapshot()
- return impl(sn, sn.Abort)
- }
-}
-
////////////////////////////////////////
// Internal helpers
diff --git a/services/syncbase/server/service.go b/services/syncbase/server/service.go
index 6ca96f1..f2ec671 100644
--- a/services/syncbase/server/service.go
+++ b/services/syncbase/server/service.go
@@ -13,7 +13,6 @@
"sync"
"v.io/v23/context"
- "v.io/v23/glob"
"v.io/v23/rpc"
"v.io/v23/security/access"
wire "v.io/v23/services/syncbase"
@@ -136,6 +135,8 @@
////////////////////////////////////////
// RPC methods
+// TODO(sadovsky): Implement Glob__ or GlobChildren__.
+
func (s *service) SetPermissions(ctx *context.T, call rpc.ServerCall, perms access.Permissions, version string) error {
return store.RunInTransaction(s.st, func(tx store.Transaction) error {
data := &serviceData{}
@@ -158,14 +159,14 @@
return data.Perms, util.FormatVersion(data.Version), nil
}
-func (s *service) GlobChildren__(ctx *context.T, call rpc.GlobChildrenServerCall, matcher *glob.Element) error {
+func (s *service) ListApps(ctx *context.T, call rpc.ServerCall) ([]string, error) {
// Check perms.
sn := s.st.NewSnapshot()
if err := util.GetWithAuth(ctx, call, sn, s.stKey(), &serviceData{}); err != nil {
sn.Abort()
- return err
+ return nil, err
}
- return util.Glob(ctx, call, matcher, sn, sn.Abort, util.AppPrefix)
+ return util.ListChildren(ctx, call, sn, util.AppPrefix)
}
////////////////////////////////////////
diff --git a/services/syncbase/server/util/constants.go b/services/syncbase/server/util/constants.go
index ab2e401..a14d006 100644
--- a/services/syncbase/server/util/constants.go
+++ b/services/syncbase/server/util/constants.go
@@ -8,9 +8,9 @@
"time"
)
-// TODO(sadovsky): Consider using shorter strings.
-
// Constants related to storage engine keys.
+// Note, these are persisted and therefore must not be modified.
+// Below, they are ordered lexicographically by value.
const (
AppPrefix = "$app"
ClockPrefix = "$clock"
@@ -23,26 +23,48 @@
SyncPrefix = "$sync"
TablePrefix = "$table"
VersionPrefix = "$version"
+
+ // TODO(sadovsky): Changing these prefixes breaks various tests. Tests
+ // generally shouldn't depend on the values of these constants.
+ /*
+ AppPrefix = "a"
+ ClockPrefix = "c"
+ DatabasePrefix = "d"
+ DbInfoPrefix = "i"
+ LogPrefix = "l"
+ PermsPrefix = "p"
+ RowPrefix = "r"
+ ServicePrefix = "s"
+ TablePrefix = "t"
+ VersionPrefix = "v"
+ SyncPrefix = "y"
+ */
+
+ // Separator for parts of storage engine keys.
+ // TODO(sadovsky): Allow ":" in names and use a different separator here.
+ KeyPartSep = ":"
+
+ // PrefixRangeLimitSuffix is a key suffix that indicates the end of a prefix
+ // range. Must be greater than any character allowed in client-provided keys.
+ PrefixRangeLimitSuffix = "\xff"
)
// Constants related to object names.
const (
- // Service object name suffix for Syncbase-to-Syncbase RPCs.
- SyncbaseSuffix = "$sync"
+ // Object name component for Syncbase-to-Syncbase (sync) RPCs.
+ // Sync object names have the form:
+ // <syncbase>/@@sync/...
+ SyncbaseSuffix = "@@sync"
// Separator for batch info in database names.
- BatchSep = ":"
- // Separator for parts of storage engine keys.
- KeyPartSep = ":"
- // PrefixRangeLimitSuffix is the suffix of a key which indicates the end of
- // a prefix range. Should be more than any regular key in the store.
- // TODO(rogulenko): Change this constant to something out of the UTF8 space.
- PrefixRangeLimitSuffix = "~"
+ // Batch object names have the form:
+ // <syncbase>/<app>/$/<database>@@<batchInfo>/...
+ BatchSep = "@@"
)
// Constants related to syncbase clock.
const (
- // The pool.ntp.org project is a big virtual cluster of timeservers
- // providing reliable easy to use NTP service for millions of clients.
+ // The pool.ntp.org project is a big virtual cluster of timeservers,
+ // providing reliable, easy-to-use NTP service for millions of clients.
// See more at http://www.pool.ntp.org/en/
NtpServerPool = "pool.ntp.org"
NtpSampleCount = 15
diff --git a/services/syncbase/server/util/glob.go b/services/syncbase/server/util/glob.go
deleted file mode 100644
index c483347..0000000
--- a/services/syncbase/server/util/glob.go
+++ /dev/null
@@ -1,40 +0,0 @@
-// 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 util
-
-import (
- "v.io/v23/context"
- "v.io/v23/glob"
- "v.io/v23/naming"
- "v.io/v23/rpc"
- "v.io/x/lib/vlog"
- "v.io/x/ref/services/syncbase/store"
-)
-
-// NOTE(nlacasse): Syncbase handles Glob requests by implementing
-// GlobChildren__ at each level (service, app, database, table).
-
-// Glob performs a glob. It calls closeSntx to close sntx.
-func Glob(ctx *context.T, call rpc.GlobChildrenServerCall, matcher *glob.Element, sntx store.SnapshotOrTransaction, closeSntx func() error, stKeyPrefix string) error {
- prefix, _ := matcher.FixedPrefix()
- it := sntx.Scan(ScanPrefixArgs(stKeyPrefix, prefix))
- defer closeSntx()
- key := []byte{}
- for it.Advance() {
- key = it.Key(key)
- parts := SplitKeyParts(string(key))
- name := parts[len(parts)-1]
- if matcher.Match(name) {
- if err := call.SendStream().Send(naming.GlobChildrenReplyName{Value: name}); err != nil {
- return err
- }
- }
- }
- if err := it.Err(); err != nil {
- vlog.VI(1).Infof("Glob() failed: %v", err)
- call.SendStream().Send(naming.GlobChildrenReplyError{Value: naming.GlobError{Error: err}})
- }
- return nil
-}
diff --git a/services/syncbase/server/util/key_util.go b/services/syncbase/server/util/key_util.go
index 0ac6686..c0d79ad 100644
--- a/services/syncbase/server/util/key_util.go
+++ b/services/syncbase/server/util/key_util.go
@@ -5,18 +5,21 @@
package util
import (
+ "strconv"
"strings"
"v.io/v23/syncbase/util"
+ "v.io/v23/verror"
)
// JoinKeyParts builds keys for accessing data in the storage engine.
+// TODO(sadovsky): Allow ":" in names and use a different separator here.
func JoinKeyParts(parts ...string) string {
- // TODO(sadovsky): Figure out which delimiter makes the most sense.
return strings.Join(parts, KeyPartSep)
}
// SplitKeyParts is the inverse of JoinKeyParts.
+// TODO(sadovsky): Allow ":" in names and use a different separator here.
func SplitKeyParts(key string) []string {
return strings.Split(key, KeyPartSep)
}
@@ -35,3 +38,36 @@
}
return []byte(fullStart), []byte(fullLimit)
}
+
+type BatchType int
+
+const (
+ BatchTypeSn BatchType = iota // snapshot
+ BatchTypeTx // transaction
+)
+
+// JoinBatchInfo encodes batch type and id into a single "info" string.
+func JoinBatchInfo(batchType BatchType, batchId uint64) string {
+ return strings.Join([]string{strconv.Itoa(int(batchType)), strconv.FormatUint(batchId, 10)}, BatchSep)
+}
+
+// SplitBatchInfo is the inverse of JoinBatchInfo.
+func SplitBatchInfo(batchInfo string) (BatchType, uint64, error) {
+ parts := strings.Split(batchInfo, BatchSep)
+ if len(parts) != 2 {
+ return BatchTypeSn, 0, verror.New(verror.ErrBadArg, nil, batchInfo)
+ }
+ batchTypeInt, err := strconv.Atoi(parts[0])
+ if err != nil {
+ return BatchTypeSn, 0, err
+ }
+ batchType := BatchType(batchTypeInt)
+ if batchType != BatchTypeSn && batchType != BatchTypeTx {
+ return BatchTypeSn, 0, verror.New(verror.ErrBadArg, nil, batchInfo)
+ }
+ batchId, err := strconv.ParseUint(parts[1], 0, 64)
+ if err != nil {
+ return BatchTypeSn, 0, err
+ }
+ return batchType, batchId, nil
+}
diff --git a/services/syncbase/server/util/list_children.go b/services/syncbase/server/util/list_children.go
new file mode 100644
index 0000000..eaa68bd
--- /dev/null
+++ b/services/syncbase/server/util/list_children.go
@@ -0,0 +1,29 @@
+// 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 util
+
+import (
+ "v.io/v23/context"
+ "v.io/v23/rpc"
+ "v.io/x/ref/services/syncbase/store"
+)
+
+// ListChildren returns the names of all apps, databases, or tables with the
+// given key prefix. Designed for use by Service.ListApps, App.ListDatabases,
+// and Database.ListTables.
+func ListChildren(ctx *context.T, call rpc.ServerCall, sntx store.SnapshotOrTransaction, stKeyPrefix string) ([]string, error) {
+ it := sntx.Scan(ScanPrefixArgs(stKeyPrefix, ""))
+ key := []byte{}
+ res := []string{}
+ for it.Advance() {
+ key = it.Key(key)
+ parts := SplitKeyParts(string(key))
+ res = append(res, parts[len(parts)-1])
+ }
+ if err := it.Err(); err != nil {
+ return nil, err
+ }
+ return res, nil
+}
diff --git a/services/syncbase/testutil/layer.go b/services/syncbase/testutil/layer.go
index fc568b9..3f6cc54 100644
--- a/services/syncbase/testutil/layer.go
+++ b/services/syncbase/testutil/layer.go
@@ -12,6 +12,7 @@
"v.io/v23/context"
"v.io/v23/security"
"v.io/v23/security/access"
+ wire "v.io/v23/services/syncbase"
"v.io/v23/syncbase"
"v.io/v23/syncbase/nosql"
"v.io/v23/syncbase/util"
@@ -72,7 +73,8 @@
t.Errorf("Perms do not match: got %v, want %v", gotPerms, wantPerms)
}
- // Even though self2 exists, Exists returns false because Read access is needed.
+ // Even though self2 exists, Exists returns false because Read access is
+ // needed.
assertExists(t, ctx, self2, "self2", false)
// Test that create fails if the parent perms disallow access.
@@ -90,6 +92,33 @@
assertExists(t, ctx, self3, "self3", false)
}
+// Tests that non-ASCII UTF-8 chars are supported at all layers as long as they
+// satisfy util.ValidName.
+// This test only requires layer.Create to be implemented and thus works for
+// rows.
+func TestCreateNameValidation(t *testing.T, ctx *context.T, i interface{}) {
+ parent := makeLayer(i)
+
+ // Invalid names.
+ // TODO(sadovsky): Add names with slashes to this list once we implement
+ // client-side name validation. As it stands, some names with slashes result
+ // in RPCs against objects at the next layer of hierarchy, and naming.Join
+ // drops some leading and trailing slashes before they reach the server-side
+ // name-checking code.
+ for _, name := range []string{"a\x00", "\x00a", "@@", "a@@", "@@a", "@@a", "$", "a/$", "$/a"} {
+ if err := parent.Child(name).Create(ctx, nil); verror.ErrorID(err) != wire.ErrInvalidName.ID {
+ t.Fatalf("Create(%q) should have failed: %v", name, err)
+ }
+ }
+
+ // Valid names.
+ for _, name := range []string{"a", "aa", "*", "a*", "*a", "a*b", "a/b", "a/$$", "$$/a", "a/$$/b", "dev.v.io/a/admin@myapp.com", "alice/bob", "안녕하세요"} {
+ if err := parent.Child(name).Create(ctx, nil); err != nil {
+ t.Fatalf("Create(%q) failed: %v", name, err)
+ }
+ }
+}
+
// TestDestroy tests that object destruction works as expected.
func TestDestroy(t *testing.T, ctx *context.T, i interface{}) {
parent := makeLayer(i)
@@ -179,7 +208,7 @@
var err error
got, err = self.ListChildren(ctx)
- want = []string{}
+ want = []string(nil)
if err != nil {
t.Fatalf("self.ListChildren() failed: %v", err)
}
@@ -210,6 +239,19 @@
if !reflect.DeepEqual(got, want) {
t.Fatalf("Lists do not match: got %v, want %v", got, want)
}
+
+ hello := "안녕하세요"
+ if err := self.Child(hello).Create(ctx, nil); err != nil {
+ t.Fatalf("hello.Create() failed: %v", err)
+ }
+ got, err = self.ListChildren(ctx)
+ want = []string{"x", "y", hello}
+ if err != nil {
+ t.Fatalf("self.ListChildren() failed: %v", err)
+ }
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("Lists do not match: got %v, want %v", got, want)
+ }
}
// TestPerms tests that {Set,Get}Permissions work as expected.
diff --git a/services/syncbase/testutil/v23util.go b/services/syncbase/testutil/v23util.go
index ef2c3f6..8e370a8 100644
--- a/services/syncbase/testutil/v23util.go
+++ b/services/syncbase/testutil/v23util.go
@@ -55,7 +55,7 @@
}
}
-// RunClient runs modules.Program and waits until it terminates.
+// RunClient runs the given program and waits until it terminates.
func RunClient(t *v23tests.T, creds *modules.CustomCredentials, program modules.Program, args ...string) {
client, err := t.Shell().StartWithOpts(
t.Shell().DefaultStartOpts().WithCustomCredentials(creds),
diff --git a/services/xproxyd/proxy_test.go b/services/xproxyd/proxy_test.go
index 4257409..43f613a 100644
--- a/services/xproxyd/proxy_test.go
+++ b/services/xproxyd/proxy_test.go
@@ -43,14 +43,6 @@
func TestMultipleProxies(t *testing.T) {
pctx, shutdown := v23.Init()
defer shutdown()
- p2ctx, _, err := v23.ExperimentalWithNewFlowManager(pctx)
- if err != nil {
- t.Fatal(err)
- }
- p3ctx, _, err := v23.ExperimentalWithNewFlowManager(pctx)
- if err != nil {
- t.Fatal(err)
- }
actx, am, err := v23.ExperimentalWithNewFlowManager(pctx)
if err != nil {
t.Fatal(err)
@@ -62,9 +54,9 @@
pep := startProxy(t, pctx, address{"tcp", "127.0.0.1:0"})
- p2ep := startProxy(t, p2ctx, address{"v23", pep.String()}, address{"tcp", "127.0.0.1:0"})
+ p2ep := startProxy(t, pctx, address{"v23", pep.String()}, address{"tcp", "127.0.0.1:0"})
- p3ep := startProxy(t, p3ctx, address{"v23", p2ep.String()}, address{"tcp", "127.0.0.1:0"})
+ p3ep := startProxy(t, pctx, address{"v23", p2ep.String()}, address{"tcp", "127.0.0.1:0"})
if err := am.Listen(actx, "v23", p3ep.String()); err != nil {
t.Fatal(err)
@@ -147,7 +139,7 @@
ls.Addrs = append(ls.Addrs, addr)
}
ctx = v23.WithListenSpec(ctx, ls)
- proxy, err := xproxyd.New(ctx)
+ proxy, _, err := xproxyd.New(ctx)
if err != nil {
t.Fatal(err)
}
diff --git a/services/xproxyd/proxyd.go b/services/xproxyd/proxyd.go
index fc97ceb..d05a561 100644
--- a/services/xproxyd/proxyd.go
+++ b/services/xproxyd/proxyd.go
@@ -24,41 +24,45 @@
proxyEndpoints []naming.Endpoint
}
-func New(ctx *context.T) (*proxy, error) {
+func New(ctx *context.T) (*proxy, *context.T, error) {
+ ctx, mgr, err := v23.ExperimentalWithNewFlowManager(ctx)
+ if err != nil {
+ return nil, nil, err
+ }
p := &proxy{
- m: v23.ExperimentalGetFlowManager(ctx),
+ m: mgr,
}
for _, addr := range v23.GetListenSpec(ctx).Addrs {
if addr.Protocol == "v23" {
ep, err := v23.NewEndpoint(addr.Address)
if err != nil {
- return nil, err
+ return nil, nil, err
}
f, err := p.m.Dial(ctx, ep, proxyBlessingsForPeer{}.run)
if err != nil {
- return nil, err
+ return nil, nil, err
}
// Send a byte telling the acceptor that we are a proxy.
if err := writeMessage(ctx, &message.MultiProxyRequest{}, f); err != nil {
- return nil, err
+ return nil, nil, err
}
msg, err := readMessage(ctx, f)
if err != nil {
- return nil, err
+ return nil, nil, err
}
m, ok := msg.(*message.ProxyResponse)
if !ok {
- return nil, NewErrUnexpectedMessage(ctx, fmt.Sprintf("%t", m))
+ return nil, nil, NewErrUnexpectedMessage(ctx, fmt.Sprintf("%t", m))
}
p.mu.Lock()
p.proxyEndpoints = append(p.proxyEndpoints, m.Endpoints...)
p.mu.Unlock()
} else if err := p.m.Listen(ctx, addr.Protocol, addr.Address); err != nil {
- return nil, err
+ return nil, nil, err
}
}
go p.listenLoop(ctx)
- return p, nil
+ return p, ctx, nil
}
func (p *proxy) ListeningEndpoints() []naming.Endpoint {
