feat(sql): add database/sql plugin (#22)

README.md

@@ -14,6 +14,7 @@ The plugins of [go2sky](https://github.com/SkyAPM/go2sky)
 1. [go-micro](micro/README.md)
 1. [go-restful](go-restful/README.md)
 1. [go-kratos](kratos/README.md)
+1. [sql](sql/README.md)
 
 ### Log Plugins
 1. [logrus](logrus/README.md)

sql/README.md

@@ -0,0 +1,42 @@
+# Go2Sky with database/sql
+
+## Installation
+
+```bash
+go get -u github.com/SkyAPM/go2sky-plugins/sql
+```
+
+## Usage
+
+```go
+import (
+	sqlPlugin "github.com/SkyAPM/go2sky-plugins/sql"
+
+	"github.com/SkyAPM/go2sky"
+	"github.com/SkyAPM/go2sky/reporter"
+	_ "github.com/go-sql-driver/mysql"
+)
+
+// init reporter
+re, err := reporter.NewLogReporter()
+defer re.Close()
+
+// init tracer
+tracer, err := go2sky.NewTracer("service-name", go2sky.WithReporter(re))
+if err != nil {
+    log.Fatalf("init tracer error: %v", err)
+}
+
+// use sql plugin to open db with tracer
+db, err := sqlPlugin.Open("mysql", dsn, tracer,
+    sqlPlugin.WithSqlDBType(sqlPlugin.MYSQL),
+    sqlPlugin.WithQueryReport(),
+    sqlPlugin.WithParamReport(),
+    sqlPlugin.WithPeerAddr("127.0.0.1:3306"),
+)
+if err != nil {
+	log.Fatalf("open db error: %v \n", err)
+}
+
+// use db handler as usual.
+```

sql/common.go

@@ -0,0 +1,121 @@
+//
+// Copyright 2021 SkyAPM org
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+package sql
+
+import (
+	"context"
+	"database/sql/driver"
+	"errors"
+	"fmt"
+	"regexp"
+	"strings"
+
+	"github.com/SkyAPM/go2sky"
+	agentv3 "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
+)
+
+const (
+	componentIDUnknown = 0
+	componentIDMysql   = 5012
+)
+
+const (
+	tagDbType          = "db.type"
+	tagDbInstance      = "db.instance"
+	tagDbStatement     = "db.statement"
+	tagDbSqlParameters = "db.sql.parameters"
+)
+
+var ErrUnsupportedOp = errors.New("operation unsupported by the underlying driver")
+
+// namedValueToValueString converts driver arguments of NamedValue format to Value string format.
+func namedValueToValueString(named []driver.NamedValue) string {
+	b := make([]string, 0, len(named))
+	for _, param := range named {
+		b = append(b, fmt.Sprintf("%v", param.Value))
+	}
+	return strings.Join(b, ",")
+}
+
+// namedValueToValue converts driver arguments of NamedValue format to Value format.
+// Implemented in the same way as in database/sql/ctxutil.go.
+func namedValueToValue(named []driver.NamedValue) ([]driver.Value, error) {
+	dargs := make([]driver.Value, len(named))
+	for n, param := range named {
+		if len(param.Name) > 0 {
+			return nil, errors.New("sql: driver does not support the use of Named Parameters")
+		}
+		dargs[n] = param.Value
+	}
+	return dargs, nil
+}
+
+func argsToString(args []interface{}) string {
+	sb := strings.Builder{}
+	for _, arg := range args {
+		sb.WriteString(fmt.Sprintf("%v, ", arg))
+	}
+	return sb.String()
+}
+
+func createSpan(ctx context.Context, tracer *go2sky.Tracer, opts *options, operation string) (go2sky.Span, error) {
+	s, _, err := tracer.CreateLocalSpan(ctx,
+		go2sky.WithSpanType(go2sky.SpanTypeExit),
+		go2sky.WithOperationName(opts.getOpName(operation)),
+	)
+	if err != nil {
+		return nil, err
+	}
+	s.SetPeer(opts.peer)
+	s.SetComponent(opts.componentID)
+	s.SetSpanLayer(agentv3.SpanLayer_Database)
+	s.Tag(tagDbType, string(opts.dbType))
+	s.Tag(tagDbInstance, opts.peer)
+	return s, nil
+}
+
+func createLocalSpan(ctx context.Context, tracer *go2sky.Tracer, opts *options, operation string) (go2sky.Span, context.Context, error) {
+	s, nCtx, err := tracer.CreateLocalSpan(ctx,
+		go2sky.WithSpanType(go2sky.SpanTypeLocal),
+		go2sky.WithOperationName(opts.getOpName(operation)),
+	)
+	if err != nil {
+		return nil, nil, err
+	}
+	s.SetComponent(opts.componentID)
+	s.SetSpanLayer(agentv3.SpanLayer_Database)
+	s.Tag(tagDbType, string(opts.dbType))
+	s.Tag(tagDbInstance, opts.peer)
+	return s, nCtx, nil
+}
+
+// parseDsn parse dsn to a endpoint addr string (host:port)
+func parseDsn(dbType DBType, dsn string) string {
+	var addr string
+	switch dbType {
+	case MYSQL:
+		// [user[:password]@][net[(addr)]]/dbname[?param1=value1&paramN=valueN]
+		re := regexp.MustCompile(`\(.+\)`)
+		addr = re.FindString(dsn)
+		addr = addr[1 : len(addr)-1]
+	case IPV4:
+		// ipv4 addr
+		re := regexp.MustCompile(`((2(5[0-5]|[0-4]\d))|[0-1]?\d{1,2})(\.((2(5[0-5]|[0-4]\d))|[0-1]?\d{1,2})){3}:\d{1,5}`)
+		addr = re.FindString(dsn)
+	}
+	return addr
+}

sql/conn.go

@@ -0,0 +1,132 @@
+//
+// Copyright 2021 SkyAPM org
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+package sql
+
+import (
+	"context"
+	"database/sql"
+	"time"
+)
+
+type Conn struct {
+	*sql.Conn
+
+	db *DB
+}
+
+func (c *Conn) PingContext(ctx context.Context) error {
+	span, err := createSpan(ctx, c.db.tracer, c.db.opts, "ping")
+	if err != nil {
+		return err
+	}
+	defer span.End()
+	err = c.Conn.PingContext(ctx)
+	if err != nil {
+		span.Error(time.Now(), err.Error())
+	}
+	return err
+}
+
+func (c *Conn) ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error) {
+	span, err := createSpan(ctx, c.db.tracer, c.db.opts, "execute")
+	if err != nil {
+		return nil, err
+	}
+	defer span.End()
+
+	if c.db.opts.reportQuery {
+		span.Tag(tagDbStatement, query)
+	}
+	if c.db.opts.reportParam {
+		span.Tag(tagDbSqlParameters, argsToString(args))
+	}
+
+	res, err := c.Conn.ExecContext(ctx, query, args...)
+	if err != nil {
+		span.Error(time.Now(), err.Error())
+	}
+	return res, err
+}
+
+func (c *Conn) QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error) {
+	span, err := createSpan(ctx, c.db.tracer, c.db.opts, "query")
+	if err != nil {
+		return nil, err
+	}
+	defer span.End()
+
+	if c.db.opts.reportQuery {
+		span.Tag(tagDbStatement, query)
+	}
+	if c.db.opts.reportParam {
+		span.Tag(tagDbSqlParameters, argsToString(args))
+	}
+
+	rows, err := c.Conn.QueryContext(ctx, query, args...)
+	if err != nil {
+		span.Error(time.Now(), err.Error())
+	}
+	return rows, err
+}
+
+func (c *Conn) QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row {
+	span, err := createSpan(ctx, c.db.tracer, c.db.opts, "query")
+	if err != nil {
+		return nil
+	}
+	defer span.End()
+
+	if c.db.opts.reportQuery {
+		span.Tag(tagDbStatement, query)
+	}
+	if c.db.opts.reportParam {
+		span.Tag(tagDbSqlParameters, argsToString(args))
+	}
+
+	return c.Conn.QueryRowContext(ctx, query, args...)
+}
+
+func (c *Conn) PrepareContext(ctx context.Context, query string) (*Stmt, error) {
+	stmt, err := c.Conn.PrepareContext(ctx, query)
+	return &Stmt{
+		Stmt:  stmt,
+		db:    c.db,
+		query: query,
+	}, err
+}
+
+func (c *Conn) BeginTx(ctx context.Context, opts *sql.TxOptions) (*Tx, error) {
+	span, nCtx, err := createLocalSpan(ctx, c.db.tracer, c.db.opts, "transaction")
+	if err != nil {
+		return nil, err
+	}
+
+	tx, err := c.Conn.BeginTx(ctx, opts)
+	if err != nil {
+		span.Error(time.Now(), err.Error())
+		span.End()
+		return nil, err
+	}
+
+	return &Tx{
+		Tx:   tx,
+		db:   c.db,
+		span: span,
+		ctx:  nCtx,
+	}, nil
+
+}