fix(lambda): monitor lambda server, fix performance issue, remove lambda logs from extensions (#8006)

Running lambda script in a new context adds a performance penalty.
This PR adds a health monitor in alpha that periodically pings lambda servers to check their responsiveness. The unresponsive lambda server ( configurable via --lambda=restart-after flag, default is10s) is restarted.
Also, this PR removes the lambda logs from the GraphQL response's extension. Instead, they are spit out in alpha logs itself with the prefix [LAMBDA-${ns}]

Author: NamanJain8

compose/compose.go

@@ -705,7 +705,8 @@ func main() {
 		"./docker-compose.yml", "name of output file")
 	cmd.PersistentFlags().BoolVarP(&opts.LocalBin, "local", "l", true,
 		"use locally-compiled binary if true, otherwise use binary from docker container")
-	cmd.PersistentFlags().StringVar(&opts.Image, "image", "dgraph/dgraph",
+	// TODO(Naman): Change this to dgraph/dgraph once the lambda changes are released.
+	cmd.PersistentFlags().StringVar(&opts.Image, "image", "public.ecr.aws/n1e3y0t3/dgraph-lambda",
 		"Docker image for alphas and zeros.")
 	cmd.PersistentFlags().StringVarP(&opts.Tag, "tag", "t", "latest",
 		"Docker tag for the --image image. Requires -l=false to use binary from docker container.")

contrib/bench-lambda/load-data.sh

@@ -35,8 +35,8 @@ func main() {
 			callDgraph(client, url)
 		}
 		if num := atomic.AddInt32(&count, 1); num%1000 == 0 {
-			elasped := time.Since(start).Round(time.Second).Seconds()
-			if elasped == 0 {
+			elasped := time.Since(start).Seconds()
+			if elasped < 1 {
 				return
 			}
 			fmt.Printf("[Chan: %d] Done %d requests in time: %f QPS: %d\n",

contrib/bench-lambda/main.go

@@ -25,13 +25,6 @@
  * MIT Licensed
  */
 
-/*!
- * connect-timeout
- * Copyright(c) 2014 Jonathan Ong
- * Copyright(c) 2014-2015 Douglas Christopher Wilson
- * MIT Licensed
- */
-
 /*!
  * content-disposition
  * Copyright(c) 2014-2017 Douglas Christopher Wilson
@@ -196,12 +189,6 @@
  * MIT Licensed
  */
 
-/*!
- * on-headers
- * Copyright(c) 2014 Douglas Christopher Wilson
- * MIT Licensed
- */
-
 /*!
  * parseurl
  * Copyright(c) 2014 Jonathan Ong

dgraph/cmd/alpha/dist/index.js

@@ -26,7 +26,7 @@ import (
 	"math"
 	"net"
 	"net/http"
-	_ "net/http/pprof" // http profiler
+	_ "net/http/pprof" // http profile
 	"net/url"
 	"os"
 	"os/exec"
@@ -243,6 +243,8 @@ they form a Raft group and provide synchronous replication.
 			"Number of JS lambda servers to be launched by alpha.").
 		Flag("port",
 			"The starting port at which the lambda server listens.").
+		Flag("restart-after",
+			"Restarts the lambda server after given duration of unresponsiveness").
 		String())
 
 	flag.String("cdc", worker.CDCDefaults, z.NewSuperFlagHelp(worker.CDCDefaults).
@@ -468,11 +470,7 @@ func setupLambdaServer(closer *z.Closer) {
 		return
 	}
 
-	glog.Infoln("Setting up lambda servers")
-	dgraphUrl := fmt.Sprintf("http://localhost:%d", httpPort())
-	// Entry point of the script is index.js.
-	filename := filepath.Join(x.WorkerConfig.TmpDir, "index.js")
-
+	// Copy over all the embedded files to actual files.
 	dir := "dist"
 	files, err := jsLambda.ReadDir(dir)
 	x.Check(err)
@@ -488,28 +486,84 @@ func setupLambdaServer(closer *z.Closer) {
 		x.Check(file.Close())
 	}
 
+	type lambda struct {
+		cmd        *exec.Cmd
+		active     bool
+		lastActive int64
+		health     string
+		port       int
+	}
+
+	lambdas := make([]*lambda, 0, num)
 	for i := 0; i < num; i++ {
+		lambdas = append(lambdas, &lambda{
+			port:   port + i,
+			health: fmt.Sprintf("http://127.0.0.1:%d/health", port+i),
+		})
+	}
+
+	// Entry point of the script is index.js.
+	filename := filepath.Join(x.WorkerConfig.TmpDir, "index.js")
+	dgraphUrl := fmt.Sprintf("http://127.0.0.1:%d", httpPort())
+
+	glog.Infoln("Setting up lambda servers")
+	for i := range lambdas {
 		go func(i int) {
 			for {
 				select {
 				case <-closer.HasBeenClosed():
-					break
+					return
 				default:
 					cmd := exec.CommandContext(closer.Ctx(), "node", filename)
-					cmd.Env = append(cmd.Env, fmt.Sprintf("PORT=%d", port+i))
+					cmd.Env = append(cmd.Env, fmt.Sprintf("PORT=%d", lambdas[i].port))
 					cmd.Env = append(cmd.Env, fmt.Sprintf("DGRAPH_URL="+dgraphUrl))
 					cmd.Stdout = os.Stdout
 					cmd.Stderr = os.Stderr
+					lambdas[i].cmd = cmd
+					lambdas[i].lastActive = time.Now().UnixNano()
+					lambdas[i].active = true
 					glog.Infof("Running node command: %+v\n", cmd)
-					err := cmd.Run()
-					if err != nil {
-						glog.Errorf("Lambda server idx: %d stopped with error %v", i, err)
+					if err := cmd.Run(); err != nil {
+						glog.Errorf("Lambda server at port: %d stopped with error: %v",
+							lambdas[i].port, err)
 					}
 					time.Sleep(2 * time.Second)
 				}
 			}
 		}(i)
 	}
+
+	// Monitor the lambda servers. If the server is unresponsive for more than restart-after time,
+	// restart it.
+	client := http.Client{Timeout: 1 * time.Second}
+	go func() {
+		ticker := time.NewTicker(2 * time.Second)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-closer.HasBeenClosed():
+				return
+			case <-ticker.C:
+				timestamp := time.Now().UnixNano()
+				for _, l := range lambdas {
+					if !l.active {
+						continue
+					}
+					resp, err := client.Get(l.health)
+					if err != nil || resp.StatusCode != 200 {
+						if time.Duration(timestamp-l.lastActive) > x.Config.Lambda.RestartAfter {
+							glog.Warningf("Lambda Server at port: %d not responding."+
+								" Killed it with err: %v", l.port, l.cmd.Process.Kill())
+							l.active = false
+						}
+						continue
+					}
+					resp.Body.Close()
+					l.lastActive = timestamp
+				}
+			}
+		}
+	}()
 }
 
 func serveGRPC(l net.Listener, tlsCfg *tls.Config, closer *z.Closer) {
@@ -809,9 +863,10 @@ func run() {
 	lambda := z.NewSuperFlag(Alpha.Conf.GetString("lambda")).MergeAndCheckDefault(
 		worker.LambdaDefaults)
 	x.Config.Lambda = x.LambdaOptions{
-		Url:  lambda.GetString("url"),
-		Num:  lambda.GetUint32("num"),
-		Port: lambda.GetUint32("port"),
+		Url:          lambda.GetString("url"),
+		Num:          lambda.GetUint32("num"),
+		Port:         lambda.GetUint32("port"),
+		RestartAfter: lambda.GetDuration("restart-after"),
 	}
 	if x.Config.Lambda.Url != "" {
 		graphqlLambdaUrl, err := url.Parse(x.Config.Lambda.Url)

dgraph/cmd/alpha/dist/index.js.LICENSE.txt

@@ -80,7 +80,7 @@ func processToFastJSON(q string) string {
 		log.Fatal(err)
 	}
 
-	buf, _, err := query.ToJson(context.Background(), &l, qr.Subgraphs, nil)
+	buf, err := query.ToJson(context.Background(), &l, qr.Subgraphs, nil)
 	if err != nil {
 		log.Fatal(err)
 	}

dgraph/cmd/alpha/run.go

@@ -1564,7 +1564,6 @@ func processQuery(ctx context.Context, qc *queryContext) (*api.Response, error)
 		return resp, errors.Wrap(err, "")
 	}
 
-	var logs []string
 	if len(er.SchemaNode) > 0 || len(er.Types) > 0 {
 		if err = authorizeSchemaQuery(ctx, &er); err != nil {
 			return resp, err
@@ -1587,7 +1586,7 @@ func processQuery(ctx context.Context, qc *queryContext) (*api.Response, error)
 	} else if qc.req.RespFormat == api.Request_RDF {
 		resp.Rdf, err = query.ToRDF(qc.latency, er.Subgraphs)
 	} else {
-		resp.Json, logs, err = query.ToJson(ctx, qc.latency, er.Subgraphs, qc.gqlField)
+		resp.Json, err = query.ToJson(ctx, qc.latency, er.Subgraphs, qc.gqlField)
 	}
 	// if err is just some error from GraphQL encoding, then we need to continue the normal
 	// execution ignoring the error as we still need to assign metrics and latency info to resp.
@@ -1642,7 +1641,6 @@ func processQuery(ctx context.Context, qc *queryContext) (*api.Response, error)
 
 	resp.Metrics = &api.Metrics{
 		NumUids: er.Metrics,
-		Logs:    logs,
 	}
 
 	var total uint64

dgraph/cmd/alpha/run_test.go

@@ -20,7 +20,7 @@ require (
 	github.com/blevesearch/bleve v1.0.13
 	github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd
 	github.com/dgraph-io/badger/v3 v3.0.0-20210825061050-c2b23c471f5e
-	github.com/dgraph-io/dgo/v210 v210.0.0-20210825123656-d3f867fe9cc3
+	github.com/dgraph-io/dgo/v210 v210.0.0-20210421093152-78a2fece3ebd
 	github.com/dgraph-io/gqlgen v0.13.2
 	github.com/dgraph-io/gqlparser/v2 v2.2.1
 	github.com/dgraph-io/graphql-transport-ws v0.0.0-20210511143556-2cef522f1f15

edgraph/server.go

@@ -168,8 +168,8 @@ github.com/dgraph-io/badger v1.6.0 h1:DshxFxZWXUcO0xX476VJC07Xsr6ZCBVRHKZ93Oh7Ev
 github.com/dgraph-io/badger v1.6.0/go.mod h1:zwt7syl517jmP8s94KqSxTlM6IMsdhYy6psNgSztDR4=
 github.com/dgraph-io/badger/v3 v3.0.0-20210825061050-c2b23c471f5e h1:lugmhvI1tMal0wKW0g5uxIRHUqXpE5y1lgq/vm/UP/8=
 github.com/dgraph-io/badger/v3 v3.0.0-20210825061050-c2b23c471f5e/go.mod h1:dULbq6ehJ5K0cGW/1TQ9iSfUk0gbSiToDWmWmTsJ53E=
-github.com/dgraph-io/dgo/v210 v210.0.0-20210825123656-d3f867fe9cc3 h1:/S7Dett03h3+KWRenJeuKE/1jZv76MaB9C1mbR/1Tns=
-github.com/dgraph-io/dgo/v210 v210.0.0-20210825123656-d3f867fe9cc3/go.mod h1:dCzdThGGTPYOAuNtrM6BiXj/86voHn7ZzkPL6noXR3s=
+github.com/dgraph-io/dgo/v210 v210.0.0-20210421093152-78a2fece3ebd h1:bKck5FnruuJxL1oCmrDSYWRl634IxBwL/IwwWx4UgEM=
+github.com/dgraph-io/dgo/v210 v210.0.0-20210421093152-78a2fece3ebd/go.mod h1:dCzdThGGTPYOAuNtrM6BiXj/86voHn7ZzkPL6noXR3s=
 github.com/dgraph-io/gqlgen v0.13.2 h1:TNhndk+eHKj5qE7BenKKSYdSIdOGhLqxR1rCiMso9KM=
 github.com/dgraph-io/gqlgen v0.13.2/go.mod h1:iCOrOv9lngN7KAo+jMgvUPVDlYHdf7qDwsTkQby2Sis=
 github.com/dgraph-io/gqlparser/v2 v2.1.1/go.mod h1:MYS4jppjyx8b9tuUtjV7jU1UFZK6P9fvO8TsIsQtRKU=