fix(bulk): upsert guardian/groot for all existing namespaces (#7759)

Issue:
If the data was loaded into some namespace using bulk loader with --force-namespace, that data wasn't actually accessible.
This is because the users/groups were not created for it, if they were not present originally in the RDF.

Solution:
This PR fixes that by upserting the default user and password for that namespace, when the alpha starts (to be precise, whenever ACL is reset).

(cherry picked from commit c4f4964)

dgraph/cmd/bulk/run.go

@@ -323,6 +323,7 @@ func run() {
 
 		loader.prog.mapEdgeCount = bulkMeta.EdgeCount
 		loader.schema.schemaMap = bulkMeta.SchemaMap
+		loader.schema.types = bulkMeta.Types
 	} else {
 		loader.mapStage()
 		mergeMapShardsIntoReduceShards(&opt)
@@ -331,6 +332,7 @@ func run() {
 		bulkMeta := pb.BulkMeta{
 			EdgeCount: loader.prog.mapEdgeCount,
 			SchemaMap: loader.schema.schemaMap,
+			Types:     loader.schema.types,
 		}
 		bulkMetaData, err := bulkMeta.Marshal()
 		if err != nil {

dgraph/cmd/bulk/schema.go

@@ -51,6 +51,7 @@ func newSchemaStore(initial *schema.ParsedSchema, opt *options, state *state) *s
 	// whenever we see data for a new namespace.
 	s.checkAndSetInitialSchema(x.GalaxyNamespace)
 
+	s.types = initial.Types
 	// This is from the schema read from the schema file.
 	for _, sch := range initial.Preds {
 		p := sch.Predicate
@@ -63,8 +64,6 @@ func newSchemaStore(initial *schema.ParsedSchema, opt *options, state *state) *s
 		s.schemaMap[p] = sch
 	}
 
-	s.types = initial.Types
-
 	return s
 }
 
@@ -102,6 +101,7 @@ func (s *schemaStore) checkAndSetInitialSchema(namespace uint64) {
 	for _, update := range schema.CompleteInitialSchema(namespace) {
 		s.schemaMap[update.Predicate] = update
 	}
+	s.types = append(s.types, schema.CompleteInitialTypes(namespace)...)
 
 	if s.opt.StoreXids {
 		s.schemaMap[x.NamespaceAttr(namespace, "xid")] = &pb.SchemaUpdate{

edgraph/access_ee.go

@@ -413,28 +413,35 @@ func ResetAcl(closer *z.Closer) {
 		// The acl feature is not turned on.
 		return
 	}
-	for closer.Ctx().Err() == nil {
-		ctx, cancel := context.WithTimeout(closer.Ctx(), time.Minute)
-		defer cancel()
-		ctx = x.AttachNamespace(ctx, x.GalaxyNamespace)
-		if err := upsertGuardian(ctx); err != nil {
-			glog.Infof("Unable to upsert the guardian group. Error: %v", err)
-			time.Sleep(100 * time.Millisecond)
-			continue
+
+	upsertGuardianAndGroot := func(ns uint64) {
+		for closer.Ctx().Err() == nil {
+			ctx, cancel := context.WithTimeout(closer.Ctx(), time.Minute)
+			defer cancel()
+			ctx = x.AttachNamespace(ctx, ns)
+			if err := upsertGuardian(ctx); err != nil {
+				glog.Infof("Unable to upsert the guardian group. Error: %v", err)
+				time.Sleep(100 * time.Millisecond)
+				continue
+			}
+			break
 		}
-		break
-	}
 
-	for closer.Ctx().Err() == nil {
-		ctx, cancel := context.WithTimeout(closer.Ctx(), time.Minute)
-		defer cancel()
-		ctx = x.AttachNamespace(ctx, x.GalaxyNamespace)
-		if err := upsertGroot(ctx, "password"); err != nil {
-			glog.Infof("Unable to upsert the groot account. Error: %v", err)
-			time.Sleep(100 * time.Millisecond)
-			continue
+		for closer.Ctx().Err() == nil {
+			ctx, cancel := context.WithTimeout(closer.Ctx(), time.Minute)
+			defer cancel()
+			ctx = x.AttachNamespace(ctx, ns)
+			if err := upsertGroot(ctx, "password"); err != nil {
+				glog.Infof("Unable to upsert the groot account. Error: %v", err)
+				time.Sleep(100 * time.Millisecond)
+				continue
+			}
+			break
 		}
-		break
+	}
+
+	for ns := range schema.State().Namespaces() {
+		upsertGuardianAndGroot(ns)
 	}
 }
 

protos/pb.proto

@@ -736,6 +736,7 @@ message UpdateGraphQLSchemaResponse {
 message BulkMeta {
 	int64 edge_count = 1;
 	map<string, SchemaUpdate> schema_map = 2;
+	repeated TypeUpdate types = 3;
 }
 
 message DeleteNsRequest {

schema/schema.go

@@ -138,6 +138,22 @@ func (s *state) DeleteType(typeName string) error {
 	return nil
 }
 
+// Namespaces returns the active namespaces based on the current types.
+func (s *state) Namespaces() map[uint64]struct{} {
+	if s == nil {
+		return nil
+	}
+
+	s.RLock()
+	defer s.RUnlock()
+
+	ns := make(map[uint64]struct{})
+	for typ := range s.types {
+		ns[x.ParseNamespace(typ)] = struct{}{}
+	}
+	return ns
+}
+
 // DeletePredsForNs deletes the predicate information for the namespace from the schema.
 func (s *state) DeletePredsForNs(delNs uint64) {
 	if s == nil {

systest/bulk_live/bulk/alpha_acl.yml

@@ -0,0 +1,25 @@
+version: "3.5"
+services:
+  alpha1:
+    image: dgraph/dgraph:latest
+    working_dir: /data/alpha1
+    labels:
+      cluster: test
+    ports:
+      - "8080"
+      - "9080"
+    volumes:
+      - type: bind
+        source: $GOPATH/bin
+        target: /gobin
+        read_only: true
+      - type: bind
+        source: ./data/out/0/p
+        target: /posting
+        read_only: false
+      - type: bind
+        source: ../../../ee/acl/hmac-secret
+        target: /dgraph-acl/hmac-secret
+        read_only: true
+    command: /gobin/dgraph alpha --my=alpha1:7080 --zero=zero1:5080 --logtostderr -v=2 -p=/posting
+      --security "whitelist=0.0.0.0/0;" --acl "secret-file=/dgraph-acl/hmac-secret; "

systest/bulk_live/bulk/bulk_test.go

@@ -25,3 +25,7 @@ import (
 func TestBulkCases(t *testing.T) {
 	t.Run("bulk test cases", common.RunBulkCases)
 }
+
+func TestBulkCasesAcl(t *testing.T) {
+	t.Run("bulk test cases with acl", common.RunBulkCasesAcl)
+}

systest/bulk_live/common/bulk_live_cases.go

@@ -22,6 +22,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
+	"math"
 	"net/http"
 	"os"
 	"os/exec"
@@ -77,6 +78,29 @@ func RunBulkCases(t *testing.T) {
 	suite.cleanup(t)
 }
 
+func RunBulkCasesAcl(t *testing.T) {
+	opts := suiteOpts{
+		schema:    helloWorldSchema,
+		gqlSchema: "",
+		rdfs:      helloWorldData,
+		bulkSuite: true,
+		bulkOpts:  bulkOpts{alpha: "../bulk/alpha_acl.yml", forceNs: 0x10},
+	}
+	suite := newSuiteInternal(t, opts)
+
+	t.Run("Pan and Jackson", testCaseWithAcl(`
+		{q(func: anyofterms(name, "Peter")) {
+			name
+		}}
+	`, `
+		{"q": [
+			{ "name": "Peter Pan" },
+			{ "name": "Peter Jackson" }
+		]}
+	`, "groot", "password", 0x10))
+	suite.cleanup(t)
+}
+
 // run this in sequential order. cleanup is necessary for live loader to work
 func RunLiveCases(t *testing.T) {
 	suite := helloWorldSetup(t, false)
@@ -130,6 +154,7 @@ func remoteHelloWorldSetup(t *testing.T, isBulkLoader bool) *suite {
 		rdfs:      helloWorldData,
 		bulkSuite: isBulkLoader,
 		remote:    true,
+		bulkOpts:  bulkOpts{alpha: "../bulk/alpha.yml", forceNs: math.MaxUint64},
 	})
 }
 

systest/bulk_live/common/bulk_live_fixture.go

@@ -19,6 +19,7 @@ package common
 import (
 	"context"
 	"io/ioutil"
+	"math"
 	"os"
 	"path/filepath"
 	"testing"
@@ -42,9 +43,15 @@ type suiteOpts struct {
 	gqlSchema string
 	rdfs      string
 	bulkSuite bool
+	bulkOpts  bulkOpts
 	remote    bool
 }
 
+type bulkOpts struct {
+	alpha   string
+	forceNs uint64
+}
+
 func newSuiteInternal(t *testing.T, opts suiteOpts) *suite {
 	if testing.Short() {
 		t.Skip("Skipping system test with long runtime.")
@@ -54,7 +61,6 @@ func newSuiteInternal(t *testing.T, opts suiteOpts) *suite {
 		t:    t,
 		opts: opts,
 	}
-
 	require.NoError(s.t, makeDirEmpty(rootDir))
 	rdfFile := filepath.Join(rootDir, "rdfs.rdf")
 	require.NoError(s.t, ioutil.WriteFile(rdfFile, []byte(opts.rdfs), 0644))
@@ -95,6 +101,7 @@ func newBulkOnlySuite(t *testing.T, schema, rdfs, gqlSchema string) *suite {
 		gqlSchema: gqlSchema,
 		rdfs:      rdfs,
 		bulkSuite: true,
+		bulkOpts:  bulkOpts{alpha: "../bulk/alpha.yml", forceNs: math.MaxUint64}, // preserve ns
 	}
 	return newSuiteInternal(t, opts)
 }
@@ -125,10 +132,11 @@ func (s *suite) setup(t *testing.T, schemaFile, rdfFile, gqlSchemaFile string) {
 			GQLSchemaFile: gqlSchemaFile,
 			Dir:           rootDir,
 			Env:           env,
+			Namespace:     s.opts.bulkOpts.forceNs,
 		})
 
 		require.NoError(t, err)
-		err = testutil.StartAlphas("../bulk/alpha.yml")
+		err = testutil.StartAlphas(s.opts.bulkOpts.alpha)
 		require.NoError(t, err)
 		return
 	}
@@ -157,7 +165,7 @@ func (s *suite) cleanup(t *testing.T) {
 	// NOTE: Shouldn't raise any errors here or fail a test, since this is
 	// called when we detect an error (don't want to mask the original problem).
 	if s.opts.bulkSuite {
-		isRace := testutil.StopAlphasAndDetectRace("../bulk/alpha.yml")
+		isRace := testutil.StopAlphasAndDetectRace(s.opts.bulkOpts.alpha)
 		_ = os.RemoveAll(rootDir)
 		if isRace {
 			t.Fatalf("Failing because race condition is detected. " +
@@ -188,3 +196,19 @@ func testCase(query, wantResult string) func(*testing.T) {
 		testutil.CompareJSON(t, wantResult, string(resp.GetJson()))
 	}
 }
+
+func testCaseWithAcl(query, wantResult, user, password string, ns uint64) func(*testing.T) {
+	return func(t *testing.T) {
+		// Check results of the bulk loader.
+		dg, err := testutil.DgraphClient(testutil.ContainerAddr("alpha1", 9080))
+		require.NoError(t, err)
+		ctx2, cancel2 := context.WithTimeout(context.Background(), time.Minute)
+		defer cancel2()
+		require.NoError(t, dg.LoginIntoNamespace(ctx2, user, password, ns))
+
+		txn := dg.NewTxn()
+		resp, err := txn.Query(ctx2, query)
+		require.NoError(t, err)
+		testutil.CompareJSON(t, wantResult, string(resp.GetJson()))
+	}
+}

testutil/bulk.go

@@ -25,6 +25,7 @@ import (
 	"os/exec"
 	"strconv"
 
+	"github.com/dgraph-io/dgraph/x"
 	"github.com/pkg/errors"
 )
 
@@ -48,13 +49,13 @@ func LiveLoad(opts LiveOpts) error {
 		"--alpha", opts.Alpha,
 		"--zero", opts.Zero,
 	}
-	if opts.ForceNs != 0 {
-		args = append(args, "--force-namespace", strconv.FormatInt(opts.ForceNs, 10))
-	}
 	if opts.Ludicrous {
 		args = append(args, "--ludicrous")
 	}
 	if opts.Creds != nil {
+		if opts.Creds.Namespace == x.GalaxyNamespace || opts.ForceNs != 0 {
+			args = append(args, "--force-namespace", strconv.FormatInt(opts.ForceNs, 10))
+		}
 		args = append(args, "--creds")
 		args = append(args, fmt.Sprintf("user=%s;password=%s;namespace=%d",
 			opts.Creds.UserID, opts.Creds.Passwd, opts.Creds.Namespace))
@@ -88,6 +89,7 @@ type BulkOpts struct {
 	GQLSchemaFile string
 	Dir           string
 	Env           []string
+	Namespace     uint64
 }
 
 func BulkLoad(opts BulkOpts) error {
@@ -100,6 +102,7 @@ func BulkLoad(opts BulkOpts) error {
 		"--map_shards="+strconv.Itoa(opts.Shards),
 		"--store_xids=true",
 		"--zero", opts.Zero,
+		"--force-namespace", strconv.FormatUint(opts.Namespace, 10),
 	)
 
 	if opts.Dir != "" {

x/keys.go

@@ -84,24 +84,20 @@ func GalaxyAttr(attr string) string {
 
 // ParseNamespaceAttr returns the namespace and attr from the given value.
 func ParseNamespaceAttr(attr string) (uint64, string) {
-	AssertTrue(len(attr) >= 8)
 	return binary.BigEndian.Uint64([]byte(attr[:8])), attr[8:]
 }
 
 func ParseNamespaceBytes(attr string) ([]byte, string) {
-	AssertTrue(len(attr) >= 8)
 	return []byte(attr[:8]), attr[8:]
 }
 
 // ParseAttr returns the attr from the given value.
 func ParseAttr(attr string) string {
-	AssertTrue(len(attr) >= 8)
 	return attr[8:]
 }
 
 // ParseNamespace returns the namespace from the given value.
 func ParseNamespace(attr string) uint64 {
-	AssertTrue(len(attr) >= 8)
 	return binary.BigEndian.Uint64([]byte(attr[:8]))
 }
 
@@ -114,7 +110,6 @@ func ParseAttrList(attrs []string) []string {
 }
 
 func IsReverseAttr(attr string) bool {
-	AssertTrue(len(attr) >= 8)
 	return attr[8] == '~'
 }