release(v20.11): merge master to bring in latest changes

compose/Makefile

@@ -20,9 +20,12 @@ BUILD_FLAGS  ?= "-N -l"
 .PHONY: all
 all: install_dgraph $(BIN)
 
+.PHONY: install
+install: all
+
 .PHONY: install_dgraph
 install_dgraph:
-	$(MAKE) -C .. install
+	$(MAKE) -C ../dgraph install
 
 $(BIN): compose.go
 	go build -gcflags=$(BUILD_FLAGS) -o $(BIN)

dgraph/cmd/alpha/metrics_test.go

@@ -96,6 +96,7 @@ func TestMetrics(t *testing.T) {
 
 		// Dgraph Memory Metrics
 		"dgraph_memory_idle_bytes", "dgraph_memory_inuse_bytes", "dgraph_memory_proc_bytes",
+		"dgraph_memory_alloc_bytes",
 		// Dgraph Activity Metrics
 		"dgraph_active_mutations_total", "dgraph_pending_proposals_total",
 		"dgraph_pending_queries_total",

dgraph/cmd/alpha/run.go

@@ -425,6 +425,7 @@ func setupServer(closer *z.Closer) {
 	http.HandleFunc("/alter", alterHandler)
 	http.HandleFunc("/health", healthCheck)
 	http.HandleFunc("/state", stateHandler)
+	http.HandleFunc("/jemalloc", x.JemallocHandler)
 
 	// TODO: Figure out what this is for?
 	http.HandleFunc("/debug/store", storeStatsHandler)
@@ -434,6 +435,7 @@ func setupServer(closer *z.Closer) {
 	// Global Epoch is a lockless synchronization mechanism for graphql service.
 	// It's is just an atomic counter used by the graphql subscription to update its state.
 	// It's is used to detect the schema changes and server exit.
+	// It is also reported by /probe/graphql endpoint as the schemaUpdateCounter.
 
 	// Implementation for schema change:
 	// The global epoch is incremented when there is a schema change.
@@ -458,7 +460,8 @@ func setupServer(closer *z.Closer) {
 		}
 		w.WriteHeader(httpStatusCode)
 		w.Header().Set("Content-Type", "application/json")
-		x.Check2(w.Write([]byte(fmt.Sprintf(`{"status":"%s"}`, healthStatus.StatusMsg))))
+		x.Check2(w.Write([]byte(fmt.Sprintf(`{"status":"%s","schemaUpdateCounter":%d}`,
+			healthStatus.StatusMsg, atomic.LoadUint64(&globalEpoch)))))
 	})
 	http.Handle("/admin", allowedMethodsHandler(allowedMethods{
 		http.MethodGet:     true,
@@ -726,10 +729,16 @@ func run() {
 	go func() {
 		var numShutDownSig int
 		for range sdCh {
+			closer := admin.ServerCloser
+			if closer == nil {
+				glog.Infoln("Caught Ctrl-C. Terminating now.")
+				os.Exit(1)
+			}
+
 			select {
-			case <-admin.ServerCloser.HasBeenClosed():
+			case <-closer.HasBeenClosed():
 			default:
-				admin.ServerCloser.Signal()
+				closer.Signal()
 			}
 			numShutDownSig++
 			glog.Infoln("Caught Ctrl-C. Terminating now (this may take a few seconds)...")

dgraph/cmd/debug/run.go

@@ -17,17 +17,26 @@
 package debug
 
 import (
+	"bufio"
 	"bytes"
+	"context"
 	"encoding/hex"
 	"fmt"
 	"io"
 	"log"
 	"math"
+	"net/http"
+	_ "net/http/pprof"
+	"os"
 	"sort"
 	"strconv"
 	"strings"
+	"sync/atomic"
 
 	"github.com/dgraph-io/badger/v2"
+	bpb "github.com/dgraph-io/badger/v2/pb"
+	"github.com/dgraph-io/ristretto/z"
+
 	"github.com/dgraph-io/dgraph/codec"
 	"github.com/dgraph-io/dgraph/ee/enc"
 	"github.com/dgraph-io/dgraph/posting"
@@ -47,8 +56,10 @@ var (
 type flagOptions struct {
 	vals          bool
 	keyLookup     string
+	rollupKey     string
 	keyHistory    bool
 	predicate     string
+	prefix        string
 	readOnly      bool
 	pdir          string
 	itemMeta      bool
@@ -83,7 +94,9 @@ func init() {
 	flag.Uint64Var(&opt.readTs, "at", math.MaxUint64, "Set read timestamp for all txns.")
 	flag.BoolVarP(&opt.readOnly, "readonly", "o", true, "Open in read only mode.")
 	flag.StringVarP(&opt.predicate, "pred", "r", "", "Only output specified predicate.")
+	flag.StringVarP(&opt.prefix, "prefix", "", "", "Uses a hex prefix.")
 	flag.StringVarP(&opt.keyLookup, "lookup", "l", "", "Hex of key to lookup.")
+	flag.StringVar(&opt.rollupKey, "rollup", "", "Hex of key to rollup.")
 	flag.BoolVarP(&opt.keyHistory, "history", "y", false, "Show all versions of a key.")
 	flag.StringVarP(&opt.pdir, "postings", "p", "", "Directory where posting lists are stored.")
 	flag.BoolVar(&opt.sizeHistogram, "histogram", false,
@@ -432,6 +445,43 @@ func appendPosting(w io.Writer, o *pb.Posting) {
 	}
 	fmt.Fprintln(w, "")
 }
+func rollupKey(db *badger.DB) {
+	txn := db.NewTransactionAt(opt.readTs, false)
+	defer txn.Discard()
+
+	key, err := hex.DecodeString(opt.rollupKey)
+	x.Check(err)
+
+	iopts := badger.DefaultIteratorOptions
+	iopts.AllVersions = true
+	iopts.PrefetchValues = false
+	itr := txn.NewKeyIterator(key, iopts)
+	defer itr.Close()
+
+	itr.Rewind()
+	if !itr.Valid() {
+		log.Fatalf("Unable to seek to key: %s", hex.Dump(key))
+	}
+
+	item := itr.Item()
+	// Don't need to do anything if the bitdelta is not set.
+	if item.UserMeta()&posting.BitDeltaPosting == 0 {
+		fmt.Printf("First item has UserMeta:[b%04b]. Nothing to do\n", item.UserMeta())
+		return
+	}
+	pl, err := posting.ReadPostingList(item.KeyCopy(nil), itr)
+	x.Check(err)
+
+	alloc := z.NewAllocator(32 << 20)
+	defer alloc.Release()
+
+	kvs, err := pl.Rollup(alloc)
+	x.Check(err)
+
+	wb := db.NewManagedWriteBatch()
+	x.Check(wb.WriteList(&bpb.KVList{Kv: kvs}))
+	x.Check(wb.Flush())
+}
 
 func lookup(db *badger.DB) {
 	txn := db.NewTransactionAt(opt.readTs, false)
@@ -487,26 +537,20 @@ func lookup(db *badger.DB) {
 // {i} attr: name term: [8] woods  ts: 535 item: [28, b0100] sz: 81 dcnt: 3 key: 00000...6f6f6473
 // Fix the TestBulkLoadMultiShard accordingly, if the format changes.
 func printKeys(db *badger.DB) {
-	txn := db.NewTransactionAt(opt.readTs, false)
-	defer txn.Discard()
-
-	iopts := badger.DefaultIteratorOptions
-	iopts.AllVersions = true
-	iopts.PrefetchValues = false
-	itr := txn.NewIterator(iopts)
-	defer itr.Close()
-
 	var prefix []byte
 	if len(opt.predicate) > 0 {
 		prefix = x.PredicatePrefix(opt.predicate)
+	} else if len(opt.prefix) > 0 {
+		p, err := hex.DecodeString(opt.prefix)
+		x.Check(err)
+		prefix = p
 	}
-
 	fmt.Printf("prefix = %s\n", hex.Dump(prefix))
-	var loop int
-	for itr.Seek(prefix); itr.ValidForPrefix(prefix); {
+	stream := db.NewStreamAt(opt.readTs)
+	stream.Prefix = prefix
+	var total uint64
+	stream.KeyToList = func(key []byte, itr *badger.Iterator) (*bpb.KVList, error) {
 		item := itr.Item()
-		var key []byte
-		key = item.KeyCopy(key)
 		pk, err := x.Parse(key)
 		x.Check(err)
 		var buf bytes.Buffer
@@ -544,6 +588,7 @@ func printKeys(db *badger.DB) {
 		}
 
 		var sz, deltaCount int64
+	LOOP:
 		for ; itr.ValidForPrefix(prefix); itr.Next() {
 			item := itr.Item()
 			if !bytes.Equal(item.Key(), key) {
@@ -557,7 +602,7 @@ func printKeys(db *badger.DB) {
 			// This is rather a default case as one of the 4 bit must be set.
 			case posting.BitCompletePosting, posting.BitEmptyPosting, posting.BitSchemaPosting:
 				sz += item.EstimatedSize()
-				break
+				break LOOP
 			case posting.BitDeltaPosting:
 				sz += item.EstimatedSize()
 				deltaCount++
@@ -569,20 +614,54 @@ func printKeys(db *badger.DB) {
 				break
 			}
 		}
+		var invalidSz, invalidCount uint64
 		// skip all the versions of key
 		for ; itr.ValidForPrefix(prefix); itr.Next() {
 			item := itr.Item()
 			if !bytes.Equal(item.Key(), key) {
 				break
 			}
+			invalidSz += uint64(item.EstimatedSize())
+			invalidCount++
 		}
 
 		fmt.Fprintf(&buf, " sz: %d dcnt: %d", sz, deltaCount)
+		if invalidCount > 0 {
+			fmt.Fprintf(&buf, " isz: %d icount: %d", invalidSz, invalidCount)
+		}
 		fmt.Fprintf(&buf, " key: %s", hex.EncodeToString(key))
-		fmt.Println(buf.String())
-		loop++
+		// If total size is more than 1 GB or we have more than 1 million keys, flag this key.
+		if uint64(sz)+invalidSz > (1<<30) || uint64(deltaCount)+invalidCount > 10e6 {
+			fmt.Fprintf(&buf, " [HEAVY]")
+		}
+		buf.WriteRune('\n')
+		list := &bpb.KVList{}
+		list.Kv = append(list.Kv, &bpb.KV{
+			Value: buf.Bytes(),
+		})
+		// Don't call fmt.Println here. It is much slower.
+		return list, nil
 	}
-	fmt.Printf("Found %d keys\n", loop)
+
+	w := bufio.NewWriterSize(os.Stdout, 16<<20)
+	stream.Send = func(buf *z.Buffer) error {
+		var count int
+		err := buf.SliceIterate(func(s []byte) error {
+			var kv bpb.KV
+			if err := kv.Unmarshal(s); err != nil {
+				return err
+			}
+			x.Check2(w.Write(kv.Value))
+			count++
+			return nil
+		})
+		atomic.AddUint64(&total, uint64(count))
+		return err
+	}
+	x.Check(stream.Orchestrate(context.Background()))
+	w.Flush()
+	fmt.Println()
+	fmt.Printf("Found %d keys\n", atomic.LoadUint64(&total))
 }
 
 // Creates bounds for an histogram. The bounds are powers of two of the form
@@ -792,6 +871,16 @@ func printZeroProposal(buf *bytes.Buffer, zpr *pb.ZeroProposal) {
 }
 
 func run() {
+	go func() {
+		for i := 8080; i < 9080; i++ {
+			fmt.Printf("Listening for /debug HTTP requests at port: %d\n", i)
+			if err := http.ListenAndServe(fmt.Sprintf("localhost:%d", i), nil); err != nil {
+				fmt.Println("Port busy. Trying another one...")
+				continue
+			}
+		}
+	}()
+
 	var err error
 	dir := opt.pdir
 	isWal := false
@@ -806,7 +895,9 @@ func run() {
 
 	bopts := badger.DefaultOptions(dir).
 		WithReadOnly(opt.readOnly).
-		WithEncryptionKey(opt.key)
+		WithEncryptionKey(opt.key).
+		WithBlockCacheSize(1 << 30).
+		WithIndexCacheSize(1 << 30)
 
 	x.AssertTruef(len(bopts.Dir) > 0, "No posting or wal dir specified.")
 	fmt.Printf("Opening DB: %s\n", bopts.Dir)
@@ -853,6 +944,8 @@ func run() {
 	// fmt.Printf("Min commit: %d. Max commit: %d, w.r.t %d\n", min, max, opt.readTs)
 
 	switch {
+	case len(opt.rollupKey) > 0:
+		rollupKey(db)
 	case len(opt.keyLookup) > 0:
 		lookup(db)
 	case len(opt.jepsen) > 0:

dgraph/cmd/zero/run.go

@@ -247,6 +247,7 @@ func run() {
 	http.HandleFunc("/moveTablet", st.moveTablet)
 	http.HandleFunc("/assign", st.assign)
 	http.HandleFunc("/enterpriseLicense", st.applyEnterpriseLicense)
+	http.HandleFunc("/jemalloc", x.JemallocHandler)
 	zpages.Handle(http.DefaultServeMux, "/z")
 
 	// This must be here. It does not work if placed before Grpc init.

dgraph/main.go

@@ -101,9 +101,11 @@ func main() {
 	// Run the program.
 	cmd.Execute()
 	ticker.Stop()
+
+	glog.V(2).Infof("Num Allocated Bytes at program end: %d", z.NumAllocBytes())
 	if z.NumAllocBytes() > 0 {
 		glog.Warningf("MEMORY LEAK detected of size: %s\n",
 			humanize.Bytes(uint64(z.NumAllocBytes())))
-		z.PrintLeaks()
+		glog.Warningf("%s", z.Leaks())
 	}
 }

edgraph/server.go

@@ -937,18 +937,20 @@ func (s *Server) Health(ctx context.Context, all bool) (*api.Response, error) {
 			healthAll = append(healthAll, p.HealthInfo())
 		}
 	}
+
 	// Append self.
 	healthAll = append(healthAll, pb.HealthInfo{
-		Instance:   "alpha",
-		Address:    x.WorkerConfig.MyAddr,
-		Status:     "healthy",
-		Group:      strconv.Itoa(int(worker.GroupId())),
-		Version:    x.Version(),
-		Uptime:     int64(time.Since(x.WorkerConfig.StartTime) / time.Second),
-		LastEcho:   time.Now().Unix(),
-		Ongoing:    worker.GetOngoingTasks(),
-		Indexing:   schema.GetIndexingPredicates(),
-		EeFeatures: ee.GetEEFeaturesList(),
+		Instance:    "alpha",
+		Address:     x.WorkerConfig.MyAddr,
+		Status:      "healthy",
+		Group:       strconv.Itoa(int(worker.GroupId())),
+		Version:     x.Version(),
+		Uptime:      int64(time.Since(x.WorkerConfig.StartTime) / time.Second),
+		LastEcho:    time.Now().Unix(),
+		Ongoing:     worker.GetOngoingTasks(),
+		Indexing:    schema.GetIndexingPredicates(),
+		EeFeatures:  ee.GetEEFeaturesList(),
+		MaxAssigned: posting.Oracle().MaxAssigned(),
 	})
 
 	var err error