implement collect

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-15 15:49:54 +03:00
parent 315cce9348
commit b023ed8cec
4 changed files with 234 additions and 1 deletions
--- a/README.md
+++ b/README.md
@@ -1 +1,52 @@
-# micro-wrapper-sql
+# micro-wrapper-sqlpackage postgres
+
+Example for For postgres 
+
+```go
+import (
+	"fmt"
+	"net/url"
+	"time"
+
+	"github.com/jackc/pgx/v4"
+	"github.com/jackc/pgx/v4/stdlib"
+	"github.com/jmoiron/sqlx"
+)
+
+func Connect(cfg *PostgresConfig) (*sqlx.DB, error) {
+	// format connection string
+	dbstr := fmt.Sprintf(
+		"postgres://%s:%s@%s/%s?sslmode=disable&statement_cache_mode=describe",
+		cfg.Login,
+		url.QueryEscape(cfg.Passw),
+		cfg.Addr,
+		cfg.DBName,
+	)
+
+	// parse connection string
+	dbConf, err := pgx.ParseConfig(dbstr)
+	if err != nil {
+		return nil, err
+	}
+
+	// needed for pgbouncer
+	dbConf.RuntimeParams = map[string]string{
+		"standard_conforming_strings": "on",
+		"application_name":            cfg.AppName,
+	}
+	// may be needed for pbbouncer, needs to check
+	//dbConf.PreferSimpleProtocol = true
+	// register pgx conn
+	connStr := stdlib.RegisterConnConfig(dbConf)
+
+	db, err := sqlx.Connect("pgx", connStr)
+	if err != nil {
+		return nil, err
+	}
+	db.SetMaxOpenConns(int(cfg.ConnMax))
+	db.SetMaxIdleConns(int(cfg.ConnMax / 2))
+	db.SetConnMaxLifetime(time.Duration(cfg.ConnLifetime) * time.Second)
+	db.SetConnMaxIdleTime(time.Duration(cfg.ConnMaxIdleTime) * time.Second)
+	return db, nil
+}
+```
--- a/go.mod
+++ b/go.mod
@@ -0,0 +1,5 @@
+module github.com/unistack-org/micro-wrapper-sql/v3
+
+go 1.16
+
+require github.com/unistack-org/micro/v3 v3.2.22
--- a/go.sum
+++ b/go.sum
@@ -0,0 +1,18 @@
+github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
+github.com/ef-ds/deque v1.0.4/go.mod h1:gXDnTC3yqvBcHbq2lcExjtAcVrOnJCbMcZXmuj8Z4tg=
+github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
+github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/imdario/mergo v0.3.12 h1:b6R2BslTbIEToALKP7LxUvijTsNI9TAe80pLWN2g/HU=
+github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
+github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
+github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
+github.com/silas/dag v0.0.0-20210121180416-41cf55125c34/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
+github.com/unistack-org/micro/v3 v3.2.22 h1:AXyLtRpfcPGczhaA1f9KR0ctK+1Zpqvb+rBJrZtp3Oo=
+github.com/unistack-org/micro/v3 v3.2.22/go.mod h1:oI8H/uGq1h4i5cvUycEoFKJQC7G8yChZQNIDNWGSLRU=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
--- a/wrapper.go
+++ b/wrapper.go
@@ -0,0 +1,159 @@
+package wrapper
+
+import (
+	"context"
+	"database/sql"
+	"time"
+
+	"github.com/unistack-org/micro/v3/logger"
+	"github.com/unistack-org/micro/v3/meter"
+	"github.com/unistack-org/micro/v3/tracer"
+)
+
+var (
+	DefaultStatsInterval = 5 * time.Second
+	// default metric prefix
+	DefaultMetricPrefix = "micro_sql_"
+	// default label prefix
+	DefaultLabelPrefix = "micro_"
+)
+
+var (
+	DatabaseHostLabel   = "database_host"
+	DatabaseNameLabel   = "database_name"
+	ServiceNameLabel    = "service_name"
+	ServiceVersionLabel = "service_version"
+	ServiceIDLabel      = "service_id"
+
+	MaxOpenConnectionsLabel = "max_open_connections"
+	OpenConnectionsLabel    = "open_connections"
+	InuseConnectionsLabel   = "inuse_connections"
+	IdleConnectionsLabel    = "idle_connections"
+	WaitConnectionsLabel    = "wait_connections"
+	BlockedSecondsLabel     = "blocked_seconds"
+	MaxIdleClosedLabel      = "max_idle_closed"
+	MaxLifetimeClosedLabel  = "max_lifetime_closed"
+
+	//srequest_total // counter
+	//slatency_microseconds // summary
+	//srequest_duration_seconds // histogramm
+
+)
+
+type Options struct {
+	Logger         logger.Logger
+	Meter          meter.Meter
+	Tracer         tracer.Tracer
+	DatabaseHost   string
+	DatabaseName   string
+	ServiceName    string
+	ServiceVersion string
+	ServiceID      string
+}
+
+type Option func(*Options)
+
+func DatabaseHost(host string) Option {
+	return func(opts *Options) {
+		opts.DatabaseHost = host
+	}
+}
+
+func DatabaseName(name string) Option {
+	return func(opts *Options) {
+		opts.DatabaseName = name
+	}
+}
+
+func ServiceName(name string) Option {
+	return func(opts *Options) {
+		opts.ServiceName = name
+	}
+}
+
+func ServiceVersion(version string) Option {
+	return func(opts *Options) {
+		opts.ServiceVersion = version
+	}
+}
+
+func ServiceID(id string) Option {
+	return func(opts *Options) {
+		opts.ServiceID = id
+	}
+}
+
+func Meter(m meter.Meter) Option {
+	return func(opts *Options) {
+		opts.Meter = m
+	}
+}
+func Logger(l logger.Logger) Option {
+	return func(opts *Options) {
+		opts.Logger = l
+	}
+}
+
+func Tracer(t tracer.Tracer) Option {
+	return func(opts *Options) {
+		opts.Tracer = t
+	}
+}
+
+type queryKey struct{}
+
+func QueryName(ctx context.Context, name string) context.Context {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	return context.WithValue(ctx, queryKey{}, name)
+}
+
+func getName(ctx context.Context) string {
+	name := "Unknown"
+
+	val, ok := ctx.Value(queryKey{}).(string)
+	if ok && len(val) > 0 {
+		name = val
+	}
+
+	return name
+}
+
+type Wrapper struct {
+	db   *sql.DB
+	opts Options
+}
+
+func (w *Wrapper) collect() {
+	labels := []string{
+		DatabaseHostLabel, w.opts.DatabaseHost,
+		DatabaseNameLabel, w.opts.DatabaseName,
+		ServiceNameLabel, w.opts.ServiceName,
+		ServiceVersionLabel, w.opts.ServiceVersion,
+		ServiceIDLabel, w.opts.ServiceID,
+	}
+
+	ticker := time.NewTicker(DefaultStatsInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			if w.db == nil {
+				continue
+			}
+
+			stats := w.db.Stats()
+			w.opts.Meter.FloatCounter(MaxOpenConnectionsLabel, meter.Labels(labels...)).Set(float64(stats.MaxOpenConnections))
+			w.opts.Meter.FloatCounter(OpenConnectionsLabel, meter.Labels(labels...)).Set(float64(stats.OpenConnections))
+			w.opts.Meter.FloatCounter(InuseConnectionsLabel, meter.Labels(labels...)).Set(float64(stats.InUse))
+			w.opts.Meter.FloatCounter(IdleConnectionsLabel, meter.Labels(labels...)).Set(float64(stats.Idle))
+			w.opts.Meter.FloatCounter(WaitConnectionsLabel, meter.Labels(labels...)).Set(float64(stats.WaitCount))
+			w.opts.Meter.FloatCounter(BlockedSecondsLabel, meter.Labels(labels...)).Set(stats.WaitDuration.Seconds())
+			w.opts.Meter.FloatCounter(MaxIdleClosedLabel, meter.Labels(labels...)).Set(float64(stats.MaxIdleClosed))
+			w.opts.Meter.FloatCounter(MaxLifetimeClosedLabel, meter.Labels(labels...)).Set(float64(stats.MaxLifetimeClosed))
+		}
+	}
+}