Enrich Zipkin db tracing

Add info regarding

- database type and address
- approximation for queries results size

cmd/cataloguesvc/main.go

@@ -106,7 +106,10 @@ func main() {
  }
 
  // Data domain.
- db, err := sqlx.Open("mysql", *dsn)
+ var db catalogue.Database
+ sqlxdb, err := sqlx.Open("mysql", *dsn)
+ db = &catalogue.SqlxDb{Db: sqlxdb}
+ db = catalogue.DbTracingMiddleware()(db)
  if err != nil {
  logger.Log("err", err)
  os.Exit(1)

db.go

@@ -0,0 +1,48 @@
+package catalogue
+
+import(
+ "context"
+ "database/sql"
+
+ "github.com/jmoiron/sqlx"
+)
+
+type Database interface {
+ Close() error
+ Ping() error
+ Select(ctx context.Context, dest interface{}, query string, args ...interface{}) error
+ Prepare(query string) (StmtMiddleware, error)
+ Get(ctx context.Context, dest interface{}, query string, args ...interface{}) error
+ Query(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
+}
+
+// SqlxDb meets the Database interface requirements
+type SqlxDb struct {
+ // db is a reference for the underlying database implementation
+ Db *sqlx.DB
+}
+
+func (sqlxdb *SqlxDb) Close() error {
+ return sqlxdb.Db.Close()
+}
+
+func (sqlxdb *SqlxDb) Ping() error {
+ return sqlxdb.Db.Ping()
+}
+
+func (sqlxdb *SqlxDb) Select(ctx context.Context, dest interface{}, query string, args ...interface{}) error {
+ return sqlxdb.Db.Select(dest, query, args...)
+}
+
+func (sqlxdb *SqlxDb) Prepare(query string) (StmtMiddleware, error) {
+ sel, err := sqlxdb.Db.Prepare(query)
+ return StmtMiddleware{next: sel}, err
+}
+
+func (sqlxdb *SqlxDb) Get(ctx context.Context, dest interface{}, query string, args ...interface{}) error {
+ return sqlxdb.Db.Get(dest, query, args...)
+}
+
+func (sqlxdb *SqlxDb) Query(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error) {
+ return sqlxdb.Db.Query(query, args...)
+}

db_tracing_middleware.go

@@ -0,0 +1,87 @@
+package catalogue
+
+import (
+ "context"
+ "database/sql"
+ "unsafe"
+
+ stdopentracing "github.com/opentracing/opentracing-go"
+)
+
+// Middleware decorates a database.
+type DbMiddleware func(Database) Database
+
+// DbTracingMiddleware traces database calls.
+func DbTracingMiddleware() DbMiddleware {
+ return func(next Database) Database {
+ return dbTracingMiddleware{
+ next: next,
+ }
+ }
+}
+
+type dbTracingMiddleware struct {
+ next Database
+}
+
+type StmtMiddleware struct {
+ next *sql.Stmt
+}
+
+func (stmt StmtMiddleware) Close() error {
+ return stmt.next.Close()
+}
+
+func (stmt StmtMiddleware) QueryRow(ctx context.Context, args ...interface{}) *sql.Row {
+ span := startSpan(ctx, "rows from database")
+ rows := stmt.next.QueryRow(args...)
+ finishSpan(span, unsafe.Sizeof(rows))
+ return rows
+}
+
+func (mw dbTracingMiddleware) Close() error {
+ return mw.next.Close()
+}
+
+func (mw dbTracingMiddleware) Ping() error {
+ return mw.next.Ping()
+}
+
+func (mw dbTracingMiddleware) Select(ctx context.Context, dest interface{}, query string, args ...interface{}) error {
+ span := startSpan(ctx, "socks from database")
+ err := mw.next.Select(ctx, dest, query, args...)
+ finishSpan(span, unsafe.Sizeof(dest))
+ return err
+}
+
+func (mw dbTracingMiddleware) Prepare(query string) (StmtMiddleware, error) {
+ return mw.next.Prepare(query)
+}
+
+func (mw dbTracingMiddleware) Get(ctx context.Context, dest interface{}, query string, args ...interface{}) error {
+ span := startSpan(ctx, "get from database")
+ err := mw.next.Get(ctx, dest, query, args...)
+ finishSpan(span, unsafe.Sizeof(dest))
+ return err
+}
+
+func (mw dbTracingMiddleware) Query(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error) {
+ span := startSpan(ctx, "query from database")
+ rows, err := mw.next.Query(ctx, query, args...)
+ finishSpan(span, unsafe.Sizeof(rows))
+ return rows, err
+}
+
+func startSpan(ctx context.Context, n string) stdopentracing.Span {
+ var span stdopentracing.Span
+ span, ctx = stdopentracing.StartSpanFromContext(ctx, n)
+ span.SetTag("span.kind", "client")
+ span.SetTag("db.type", "mysql")
+ span.SetTag("peer.address", "catalogue-db:3306")
+ return span
+}
+
+func finishSpan(span stdopentracing.Span, size uintptr) {
+ span.SetTag("db.query.result.size", size)
+ span.Finish()
+}

endpoints.go

@@ -36,7 +36,7 @@ func MakeEndpoints(s Service, tracer stdopentracing.Tracer) Endpoints {
 func MakeListEndpoint(s Service) endpoint.Endpoint {
  return func(ctx context.Context, request interface{}) (response interface{}, err error) {
  req := request.(listRequest)
- socks, err := s.List(req.Tags, req.Order, req.PageNum, req.PageSize)
+ socks, err := s.List(ctx, req.Tags, req.Order, req.PageNum, req.PageSize)
  return listResponse{Socks: socks, Err: err}, err
  }
 }
@@ -45,7 +45,7 @@ func MakeListEndpoint(s Service) endpoint.Endpoint {
 func MakeCountEndpoint(s Service) endpoint.Endpoint {
  return func(ctx context.Context, request interface{}) (response interface{}, err error) {
  req := request.(countRequest)
- n, err := s.Count(req.Tags)
+ n, err := s.Count(ctx, req.Tags)
  return countResponse{N: n, Err: err}, err
  }
 }
@@ -54,15 +54,15 @@ func MakeCountEndpoint(s Service) endpoint.Endpoint {
 func MakeGetEndpoint(s Service) endpoint.Endpoint {
  return func(ctx context.Context, request interface{}) (response interface{}, err error) {
  req := request.(getRequest)
- sock, err := s.Get(req.ID)
+ sock, err := s.Get(ctx, req.ID)
  return getResponse{Sock: sock, Err: err}, err
  }
 }
 
 // MakeTagsEndpoint returns an endpoint via the given service.
 func MakeTagsEndpoint(s Service) endpoint.Endpoint {
  return func(ctx context.Context, request interface{}) (response interface{}, err error) {
- tags, err := s.Tags()
+ tags, err := s.Tags(ctx)
  return tagsResponse{Tags: tags, Err: err}, err
  }
 }

logging.go

@@ -1,6 +1,7 @@
 package catalogue
 
 import (
+ "context"
  "strings"
  "time"
 
@@ -22,7 +23,7 @@ type loggingMiddleware struct {
  logger log.Logger
 }
 
-func (mw loggingMiddleware) List(tags []string, order string, pageNum, pageSize int) (socks []Sock, err error) {
+func (mw loggingMiddleware) List(ctx context.Context, tags []string, order string, pageNum, pageSize int) (socks []Sock, err error) {
  defer func(begin time.Time) {
  mw.logger.Log(
  "method", "List",
@@ -35,10 +36,10 @@ func (mw loggingMiddleware) List(tags []string, order string, pageNum, pageSize
  "took", time.Since(begin),
  )
  }(time.Now())
- return mw.next.List(tags, order, pageNum, pageSize)
+ return mw.next.List(ctx, tags, order, pageNum, pageSize)
 }
 
-func (mw loggingMiddleware) Count(tags []string) (n int, err error) {
+func (mw loggingMiddleware) Count(ctx context.Context, tags []string) (n int, err error) {
  defer func(begin time.Time) {
  mw.logger.Log(
  "method", "Count",
@@ -48,10 +49,10 @@ func (mw loggingMiddleware) Count(tags []string) (n int, err error) {
  "took", time.Since(begin),
  )
  }(time.Now())
- return mw.next.Count(tags)
+ return mw.next.Count(ctx, tags)
 }
 
-func (mw loggingMiddleware) Get(id string) (s Sock, err error) {
+func (mw loggingMiddleware) Get(ctx context.Context, id string) (s Sock, err error) {
  defer func(begin time.Time) {
  mw.logger.Log(
  "method", "Get",
@@ -61,10 +62,10 @@ func (mw loggingMiddleware) Get(id string) (s Sock, err error) {
  "took", time.Since(begin),
  )
  }(time.Now())
- return mw.next.Get(id)
+ return mw.next.Get(ctx, id)
 }
 
-func (mw loggingMiddleware) Tags() (tags []string, err error) {
+func (mw loggingMiddleware) Tags(ctx context.Context) (tags []string, err error) {
  defer func(begin time.Time) {
  mw.logger.Log(
  "method", "Tags",
@@ -73,7 +74,7 @@ func (mw loggingMiddleware) Tags() (tags []string, err error) {
  "took", time.Since(begin),
  )
  }(time.Now())
- return mw.next.Tags()
+ return mw.next.Tags(ctx)
 }
 
 func (mw loggingMiddleware) Health() (health []Health) {

service.go

@@ -4,21 +4,21 @@ package catalogue
 // catalogue service. Everything here is agnostic to the transport (HTTP).
 
 import (
+ "context"
  "errors"
  "strings"
  "time"
 
  "github.com/go-kit/kit/log"
- "github.com/jmoiron/sqlx"
 )
 
 // Service is the catalogue service, providing read operations on a saleable
 // catalogue of sock products.
 type Service interface {
- List(tags []string, order string, pageNum, pageSize int) ([]Sock, error) // GET /catalogue
- Count(tags []string) (int, error) // GET /catalogue/size
- Get(id string) (Sock, error) // GET /catalogue/{id}
- Tags() ([]string, error) // GET /tags
+ List(ctx context.Context, tags []string, order string, pageNum, pageSize int) ([]Sock, error) // GET /catalogue
+ Count(ctx context.Context, tags []string) (int, error) // GET /catalogue/size
+ Get(ctx context.Context, id string) (Sock, error) // GET /catalogue/{id}
+ Tags(ctx context.Context) ([]string, error) // GET /tags
  Health() []Health // GET /health
 }
 
@@ -56,19 +56,19 @@ var baseQuery = "SELECT sock.sock_id AS id, sock.name, sock.description, sock.pr
 
 // NewCatalogueService returns an implementation of the Service interface,
 // with connection to an SQL database.
-func NewCatalogueService(db *sqlx.DB, logger log.Logger) Service {
+func NewCatalogueService(db Database, logger log.Logger) Service {
  return &catalogueService{
  db: db,
  logger: logger,
  }
 }
 
 type catalogueService struct {
- db *sqlx.DB
+ db Database
  logger log.Logger
 }
 
-func (s *catalogueService) List(tags []string, order string, pageNum, pageSize int) ([]Sock, error) {
+func (s *catalogueService) List(ctx context.Context, tags []string, order string, pageNum, pageSize int) ([]Sock, error) {
  var socks []Sock
  query := baseQuery
 
@@ -93,7 +93,7 @@ func (s *catalogueService) List(tags []string, order string, pageNum, pageSize i
 
  query += ";"
 
- err := s.db.Select(&socks, query, args...)
+ err := s.db.Select(ctx, &socks, query, args...)
  if err != nil {
  s.logger.Log("database error", err)
  return []Sock{}, ErrDBConnection
@@ -111,7 +111,7 @@ func (s *catalogueService) List(tags []string, order string, pageNum, pageSize i
  return socks, nil
 }
 
-func (s *catalogueService) Count(tags []string) (int, error) {
+func (s *catalogueService) Count(ctx context.Context, tags []string) (int, error) {
  query := "SELECT COUNT(DISTINCT sock.sock_id) FROM sock JOIN sock_tag ON sock.sock_id=sock_tag.sock_id JOIN tag ON sock_tag.tag_id=tag.tag_id"
 
  var args []interface{}
@@ -137,7 +137,7 @@ func (s *catalogueService) Count(tags []string) (int, error) {
  defer sel.Close()
 
  var count int
- err = sel.QueryRow(args...).Scan(&count)
+ err = sel.QueryRow(ctx, args...).Scan(&count)
 
  if err != nil {
  s.logger.Log("database error", err)
@@ -147,11 +147,11 @@ func (s *catalogueService) Count(tags []string) (int, error) {
  return count, nil
 }
 
-func (s *catalogueService) Get(id string) (Sock, error) {
+func (s *catalogueService) Get(ctx context.Context, id string) (Sock, error) {
  query := baseQuery + " WHERE sock.sock_id =? GROUP BY sock.sock_id;"
 
  var sock Sock
- err := s.db.Get(&sock, query, id)
+ err := s.db.Get(ctx, &sock, query, id)
  if err != nil {
  s.logger.Log("database error", err)
  return Sock{}, ErrNotFound
@@ -181,10 +181,10 @@ func (s *catalogueService) Health() []Health {
  return health
 }
 
-func (s *catalogueService) Tags() ([]string, error) {
+func (s *catalogueService) Tags(ctx context.Context) ([]string, error) {
  var tags []string
  query := "SELECT name FROM tag;"
- rows, err := s.db.Query(query)
+ rows, err := s.db.Query(ctx, query)
  if err != nil {
  s.logger.Log("database error", err)
  return []string{}, ErrDBConnection