Merge pull request src-d#154 from kuba--/opentracing

Opentracing upgraded to 2.13.0 
It uses configuration based on env. vars.

Author: kuba--

.gitignore

@@ -28,3 +28,5 @@ benchmark/*tbl
 vendor
 Makefile.main
 .ci/
+_example/main
+_example/*.exe

Gopkg.lock

@@ -28,7 +28,7 @@
 
 [[constraint]]
   name = "github.com/uber/jaeger-client-go"
-  version = "2.12.0"
+  version = "2.13.0"
 
 [[constraint]]
   name = "gopkg.in/src-d/go-errors.v1"

Gopkg.toml

@@ -32,7 +32,12 @@ func main() {
 		Password: "password1",
 	}}
 
-	s, err := server.NewDefaultServer("tcp", "localhost:5123", auth, driver)
+	s, err := server.NewDefaultServer(server.Config{
+		Protocol: "tcp",
+		Address:  "localhost:5123",
+		Auth:     auth,
+	}, driver)
+
 	if err != nil {
 		panic(err)
 	}
@@ -41,12 +46,12 @@ func main() {
 }
 
 func createTestDatabase() *mem.Database {
-	db := mem.NewDatabase("test")
+	db := mem.NewDatabase("test").(*mem.Database)
 	table := mem.NewTable("mytable", sql.Schema{
-		{Name: "name", Type: sql.Text},
-		{Name: "email", Type: sql.Text},
-		{Name: "phone_numbers", Type: sql.JSON},
-		{Name: "created_at", Type: sql.Timestamp},
+		{Name: "name", Type: sql.Text, Source: "mytable"},
+		{Name: "email", Type: sql.Text, Source: "mytable"},
+		{Name: "phone_numbers", Type: sql.JSON, Source: "mytable"},
+		{Name: "created_at", Type: sql.Timestamp, Source: "mytable"},
 	})
 	db.AddTable("mytable", table)
 	table.Insert(sql.NewRow("John Doe", "john@doe.com", []string{"555-555-555"}, time.Now()))

_example/main.go

@@ -1,6 +1,7 @@
 package sqle
 
 import (
+	opentracing "github.com/opentracing/opentracing-go"
 	"gopkg.in/src-d/go-mysql-server.v0/sql"
 	"gopkg.in/src-d/go-mysql-server.v0/sql/analyzer"
 	"gopkg.in/src-d/go-mysql-server.v0/sql/expression/function"
@@ -27,6 +28,9 @@ func (e *Engine) Query(
 	ctx *sql.Context,
 	query string,
 ) (sql.Schema, sql.RowIter, error) {
+	span, ctx := ctx.Span("query", opentracing.Tag{Key: "query", Value: query})
+	defer span.Finish()
+
 	parsed, err := parse.Parse(ctx, query)
 	if err != nil {
 		return nil, nil, err

engine.go

@@ -416,12 +416,12 @@ const expectedTree = `Offset(2)
 func TestPrintTree(t *testing.T) {
 	require := require.New(t)
 	node, err := parse.Parse(sql.NewEmptyContext(), `
-		SELECT t.foo, bar.baz 
-		FROM tbl t 
-		INNER JOIN bar 
-			ON foo = baz 
-		WHERE foo > qux 
-		LIMIT 5 
+		SELECT t.foo, bar.baz
+		FROM tbl t
+		INNER JOIN bar
+			ON foo = baz
+		WHERE foo > qux
+		LIMIT 5
 		OFFSET 2`)
 	require.NoError(err)
 	require.Equal(expectedTree, node.String())
@@ -439,10 +439,10 @@ func TestTracing(t *testing.T) {
 
 	ctx := sql.NewContext(context.TODO(), sql.WithTracer(tracer))
 
-	_, iter, err := e.Query(ctx, `SELECT DISTINCT i 
-		FROM mytable 
-		WHERE s = 'first row' 
-		ORDER BY i DESC 
+	_, iter, err := e.Query(ctx, `SELECT DISTINCT i
+		FROM mytable
+		WHERE s = 'first row'
+		ORDER BY i DESC
 		LIMIT 1`)
 	require.NoError(err)
 
@@ -463,7 +463,7 @@ func TestTracing(t *testing.T) {
 		"plan.Sort",
 	}
 
-	require.Len(spans, 76)
+	require.Len(spans, 77)
 
 	var spanOperations []string
 	for _, s := range spans {

engine_test.go

@@ -0,0 +1,43 @@
+package server
+
+import (
+	"io"
+
+	opentracing "github.com/opentracing/opentracing-go"
+	jaeger "github.com/uber/jaeger-client-go"
+	jaegercfg "github.com/uber/jaeger-client-go/config"
+	"github.com/uber/jaeger-lib/metrics"
+	"gopkg.in/src-d/go-vitess.v0/mysql"
+)
+
+const (
+	jaegerDefaultServiceName = "go-mysql-server"
+)
+
+// Config for the mysql server.
+type Config struct {
+	// Protocol for the connection.
+	Protocol string
+	// Address of the server.
+	Address string
+	// Auth of the server.
+	Auth mysql.AuthServer
+}
+
+// Tracer creates a new tracer for the current configuration. It also returns
+// an io.Closer to close the tracer and an error, if any.
+func (c Config) Tracer() (opentracing.Tracer, io.Closer, error) {
+	cfg, err := jaegercfg.FromEnv()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	if cfg.ServiceName == "" {
+		cfg.ServiceName = jaegerDefaultServiceName
+	}
+
+	return cfg.NewTracer(
+		jaegercfg.Logger(jaeger.StdLogger),
+		jaegercfg.Metrics(metrics.NullFactory),
+	)
+}

server/config.go

@@ -4,69 +4,11 @@ import (
 	"io"
 
 	opentracing "github.com/opentracing/opentracing-go"
-	jaeger "github.com/uber/jaeger-client-go"
 	"gopkg.in/src-d/go-mysql-server.v0"
 
 	"gopkg.in/src-d/go-vitess.v0/mysql"
 )
 
-// Config for the mysql server.
-type Config struct {
-	// Protocol for the connection.
-	Protocol string
-	// Address of the server.
-	Address string
-	// Auth of the server.
-	Auth mysql.AuthServer
-	// EnableTracing will enable the tracing, if it's true.
-	EnableTracing bool
-	// TracingAddr is the address where tracing will be sent.
-	// If this is empty, and tracing is enabled, it will be reported
-	// to the logs via stdout.
-	TracingAddr string
-	// TracingMaxPacketSize is the max packet size for sending traces
-	// to the remote endpoint.
-	TracingMaxPacketSize uint64
-	// TracingSamplingRate is the rate of traces we want to sample.
-	// Only takes effect is TracingAddr is not empty.
-	TracingSamplingRate float64
-}
-
-type nopCloser struct{}
-
-func (nopCloser) Close() error { return nil }
-
-// Tracer creates a new tracer for the current configuration. It also returns
-// an io.Closer to close the tracer and an error, if any.
-func (c Config) Tracer() (opentracing.Tracer, io.Closer, error) {
-	if !c.EnableTracing {
-		return opentracing.NoopTracer{}, nopCloser{}, nil
-	}
-
-	var reporter jaeger.Reporter
-	var sampler jaeger.Sampler
-	if c.TracingAddr == "" {
-		reporter = jaeger.NewLoggingReporter(jaeger.StdLogger)
-		sampler = jaeger.NewConstSampler(true)
-	} else {
-		transport, err := jaeger.NewUDPTransport(
-			c.TracingAddr,
-			int(c.TracingMaxPacketSize),
-		)
-		if err != nil {
-			return nil, nil, err
-		}
-		reporter = jaeger.NewRemoteReporter(transport)
-		sampler, err = jaeger.NewProbabilisticSampler(c.TracingSamplingRate)
-		if err != nil {
-			return nil, nil, err
-		}
-	}
-
-	tracer, closer := jaeger.NewTracer("go-mysql-server", sampler, reporter)
-	return tracer, closer, nil
-}
-
 // Server is a MySQL server for SQLe engines.
 type Server struct {
 	Listener *mysql.Listener
@@ -85,6 +27,7 @@ func NewServer(cfg Config, e *sqle.Engine, sb SessionBuilder) (*Server, error) {
 	if err != nil {
 		return nil, err
 	}
+	opentracing.SetGlobalTracer(tracer)
 
 	handler := NewHandler(e, NewSessionManager(sb, tracer))
 	l, err := mysql.NewListener(cfg.Protocol, cfg.Address, cfg.Auth, handler)

server/server.go

@@ -85,18 +85,23 @@ func (a *Analyzer) Log(msg string, args ...interface{}) {
 // Analyze the node and all its children.
 func (a *Analyzer) Analyze(ctx *sql.Context, n sql.Node) (sql.Node, error) {
 	span, ctx := ctx.Span("analyze")
-	defer span.Finish()
+	span.LogKV("plan", n.String())
 
 	prev := n
-
 	a.Log("starting analysis of node of type: %T", n)
 	cur, err := a.analyzeOnce(ctx, n)
+	defer func() {
+		if cur != nil {
+			span.SetTag("IsResolved", cur.Resolved())
+		}
+		span.Finish()
+	}()
+
 	if err != nil {
 		return nil, err
 	}
 
-	i := 0
-	for !reflect.DeepEqual(prev, cur) {
+	for i := 0; !reflect.DeepEqual(prev, cur); {
 		a.Log("previous node does not match new node, analyzing again, iteration: %d", i)
 		prev = cur
 		cur, err = a.analyzeOnce(ctx, cur)
@@ -111,18 +116,17 @@ func (a *Analyzer) Analyze(ctx *sql.Context, n sql.Node) (sql.Node, error) {
 	}
 
 	if errs := a.validate(ctx, cur); len(errs) != 0 {
-		var err error
 		for _, e := range errs {
 			err = multierror.Append(err, e)
 		}
-		return cur, err
 	}
 
-	return cur, nil
+	return cur, err
 }
 
 func (a *Analyzer) analyzeOnce(ctx *sql.Context, n sql.Node) (sql.Node, error) {
 	span, ctx := ctx.Span("analyze_once")
+	span.LogKV("plan", n.String())
 	defer span.Finish()
 
 	result := n

sql/analyzer/analyzer.go

@@ -39,6 +39,9 @@ func (a *Avg) IsNullable() bool {
 
 // Eval implements AggregationExpression interface. (AggregationExpression[Expression]])
 func (a *Avg) Eval(ctx *sql.Context, buffer sql.Row) (interface{}, error) {
+	span, ctx := ctx.Span("aggregation.Avg_Eval")
+	defer span.Finish()
+
 	isNoNum := buffer[2].(bool)
 	if isNoNum {
 		return float64(0), nil
@@ -50,6 +53,7 @@ func (a *Avg) Eval(ctx *sql.Context, buffer sql.Row) (interface{}, error) {
 	}
 
 	avg := buffer[0]
+	span.LogKV("avg", avg)
 	return avg, nil
 }
 
@@ -75,9 +79,6 @@ func (a *Avg) NewBuffer() sql.Row {
 
 // Update implements AggregationExpression interface. (AggregationExpression)
 func (a *Avg) Update(ctx *sql.Context, buffer, row sql.Row) error {
-	span, ctx := ctx.Span("aggregation.Avg_Update")
-	defer span.Finish()
-
 	v, err := a.Child.Eval(ctx, row)
 	if err != nil {
 		return err
@@ -112,9 +113,6 @@ func (a *Avg) Update(ctx *sql.Context, buffer, row sql.Row) error {
 
 // Merge implements AggregationExpression interface. (AggregationExpression)
 func (a *Avg) Merge(ctx *sql.Context, buffer, partial sql.Row) error {
-	span, _ := ctx.Span("aggregation.Avg_Merge")
-	defer span.Finish()
-
 	bufferAvg := buffer[0].(float64)
 	bufferRows := buffer[1].(float64)
 

sql/expression/function/aggregation/avg.go

@@ -56,9 +56,6 @@ func (c *Count) TransformUp(f sql.TransformExprFunc) (sql.Expression, error) {
 
 // Update implements the Aggregation interface.
 func (c *Count) Update(ctx *sql.Context, buffer, row sql.Row) error {
-	span, ctx := ctx.Span("aggregation.Count_Update")
-	defer span.Finish()
-
 	var inc bool
 	if _, ok := c.Child.(*expression.Star); ok {
 		inc = true
@@ -82,14 +79,16 @@ func (c *Count) Update(ctx *sql.Context, buffer, row sql.Row) error {
 
 // Merge implements the Aggregation interface.
 func (c *Count) Merge(ctx *sql.Context, buffer, partial sql.Row) error {
-	span, _ := ctx.Span("aggregation.Count_Merge")
-	defer span.Finish()
-
 	buffer[0] = buffer[0].(int32) + partial[0].(int32)
 	return nil
 }
 
 // Eval implements the Aggregation interface.
 func (c *Count) Eval(ctx *sql.Context, buffer sql.Row) (interface{}, error) {
-	return buffer[0], nil
+	span, ctx := ctx.Span("aggregation.Count_Eval")
+	count := buffer[0]
+	span.LogKV("count", count)
+	span.Finish()
+
+	return count, nil
 }