feat: miner deps: harmonydb

lib/harmony/harmonydb/doc.go

@@ -0,0 +1,35 @@
+/*
+# HarmonyDB provides database abstractions over SP-wide Postgres-compatible instance(s).
+
+# Features
+
+	Rolling to secondary database servers on connection failure
+	Convenience features for Go + SQL
+	Prevention of SQL injection vulnerabilities
+	Monitors behavior via Prometheus stats and logging of errors.
+
+# Usage
+
+Processes should use New() to instantiate a *DB and keep it.
+Consumers can use this *DB concurrently.
+Creating and changing tables & views should happen in ./sql/ folder.
+Name the file "today's date" in the format: YYYYMMDD.sql (ex: 20231231.sql for the year's last day)
+
+	a. CREATE TABLE should NOT have a schema:
+		GOOD: CREATE TABLE foo ();
+		BAD:  CREATE TABLE me.foo ();
+	b. Schema is managed for you. It provides isolation for integraton tests & multi-use.
+	c. Git Merges: All run once, so old-after-new is OK when there are no deps.
+	d. NEVER change shipped sql files. Have later files make corrections.
+	e. Anything not ran will be ran, so an older date making it to master is OK.
+
+Write SQL with context, raw strings, and args:
+
+	name := "Alice"
+	var ID int
+	err := QueryRow(ctx, "SELECT id from people where first_name=?", name).Scan(&ID)
+	fmt.Println(ID)
+
+Note: Scan() is column-oriented, while Select() & StructScan() is field name/tag oriented.
+*/
+package harmonydb

lib/harmony/harmonydb/harmonydb.go

@@ -0,0 +1,266 @@
+package harmonydb
+
+import (
+	"context"
+	"embed"
+	"errors"
+	"fmt"
+	"math/rand"
+	"regexp"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	logging "github.com/ipfs/go-log/v2"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/filecoin-project/lotus/node/config"
+)
+
+type ITestID string
+
+// ItestNewID see ITestWithID doc
+func ITestNewID() ITestID {
+	return ITestID(strconv.Itoa(rand.Intn(99999)))
+}
+
+type DB struct {
+	pgx       *pgxpool.Pool
+	cfg       *pgxpool.Config
+	schema    string
+	hostnames []string
+	log       func(string)
+}
+
+var logger = logging.Logger("harmonydb")
+
+// NewFromConfig is a convenience function.
+// In usage:
+//
+//	db, err := NewFromConfig(config.HarmonyDB)  // in binary init
+func NewFromConfig(cfg config.HarmonyDB) (*DB, error) {
+	return New(
+		cfg.Hosts,
+		cfg.Username,
+		cfg.Password,
+		cfg.Database,
+		cfg.Port,
+		"",
+		func(s string) { logger.Error(s) },
+	)
+}
+
+func NewFromConfigWithITestID(cfg config.HarmonyDB) func(id ITestID) (*DB, error) {
+	return func(id ITestID) (*DB, error) {
+		return New(
+			cfg.Hosts,
+			cfg.Username,
+			cfg.Password,
+			cfg.Database,
+			cfg.Port,
+			id,
+			func(s string) { logger.Error(s) },
+		)
+	}
+}
+
+// New is to be called once per binary to establish the pool.
+// log() is for errors. It returns an upgraded database's connection.
+// This entry point serves both production and integration tests, so it's more DI.
+func New(hosts []string, username, password, database, port string, itestID ITestID, log func(string)) (*DB, error) {
+	itest := string(itestID)
+	connString := ""
+	if len(hosts) > 0 {
+		connString = "host=" + hosts[0] + " "
+	}
+	for k, v := range map[string]string{"user": username, "password": password, "dbname": database, "port": port} {
+		if strings.TrimSpace(v) != "" {
+			connString += k + "=" + v + " "
+		}
+	}
+
+	schema := "lotus"
+	if itest != "" {
+		schema = "itest_" + itest
+	}
+
+	if err := ensureSchemaExists(connString, schema); err != nil {
+		return nil, err
+	}
+	cfg, err := pgxpool.ParseConfig(connString + "search_path=" + schema)
+	if err != nil {
+		return nil, err
+	}
+
+	// enable multiple fallback hosts.
+	for _, h := range hosts[1:] {
+		cfg.ConnConfig.Fallbacks = append(cfg.ConnConfig.Fallbacks, &pgconn.FallbackConfig{Host: h})
+	}
+
+	cfg.ConnConfig.OnNotice = func(conn *pgconn.PgConn, n *pgconn.Notice) {
+		log("database notice: " + n.Message + ": " + n.Detail)
+		DBMeasures.Errors.M(1)
+	}
+
+	db := DB{cfg: cfg, schema: schema, hostnames: hosts, log: log} // pgx populated in AddStatsAndConnect
+	if err := db.addStatsAndConnect(); err != nil {
+		return nil, err
+	}
+
+	return &db, db.upgrade()
+}
+
+type tracer struct {
+}
+
+type ctxkey string
+
+var sqlStart = ctxkey("sqlStart")
+
+func (t tracer) TraceQueryStart(ctx context.Context, conn *pgx.Conn, data pgx.TraceQueryStartData) context.Context {
+	return context.WithValue(ctx, sqlStart, time.Now())
+}
+func (t tracer) TraceQueryEnd(ctx context.Context, conn *pgx.Conn, data pgx.TraceQueryEndData) {
+	DBMeasures.Hits.M(1)
+	ms := time.Since(ctx.Value(sqlStart).(time.Time)).Milliseconds()
+	DBMeasures.TotalWait.M(ms)
+	DBMeasures.Waits.Observe(float64(ms))
+	if data.Err != nil {
+		DBMeasures.Errors.M(1)
+	}
+	// Can log what type of query it is, but not what tables
+	// Can log rows affected.
+}
+
+// addStatsAndConnect connects a prometheus logger. Be sure to run this before using the DB.
+func (db *DB) addStatsAndConnect() error {
+
+	db.cfg.ConnConfig.Tracer = tracer{}
+
+	hostnameToIndex := map[string]float64{}
+	for i, h := range db.hostnames {
+		hostnameToIndex[h] = float64(i)
+	}
+	db.cfg.AfterConnect = func(ctx context.Context, c *pgx.Conn) error {
+		s := db.pgx.Stat()
+		DBMeasures.OpenConnections.M(int64(s.TotalConns()))
+		DBMeasures.WhichHost.Observe(hostnameToIndex[c.Config().Host])
+
+		//FUTURE place for any connection seasoning
+		return nil
+	}
+
+	var err error
+	db.pgx, err = pgxpool.NewWithConfig(context.Background(), db.cfg)
+	if err != nil {
+		db.log(fmt.Sprintf("Unable to connect to database: %v\n", err))
+		return err
+	}
+	return nil
+}
+
+// ITestDeleteAll will delete everything created for "this" integration test.
+// This must be called at the end of each integration test.
+func (db *DB) ITestDeleteAll() {
+	if !strings.HasPrefix(db.schema, "itest_") {
+		fmt.Println("Warning: this should never be called on anything but an itest schema.")
+		return
+	}
+	defer db.pgx.Close()
+	_, err := db.pgx.Exec(context.Background(), "DROP SCHEMA "+db.schema+" CASCADE")
+	if err != nil {
+		fmt.Println("warning: unclean itest shutdown: cannot delete schema: " + err.Error())
+		return
+	}
+}
+
+var schemaREString = "^[A-Za-z0-9_]+$"
+var schemaRE = regexp.MustCompile(schemaREString)
+
+func ensureSchemaExists(connString, schema string) error {
+	// FUTURE allow using fallback DBs for start-up.
+	p, err := pgx.Connect(context.Background(), connString)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = p.Close(context.Background()) }()
+
+	if len(schema) < 5 || !schemaRE.MatchString(schema) {
+		return errors.New("schema must be of the form " + schemaREString + "\n Got: " + schema)
+	}
+	_, err = p.Exec(context.Background(), "CREATE SCHEMA IF NOT EXISTS "+schema)
+	if err != nil {
+		return fmt.Errorf("cannot create schema: %w", err)
+	}
+	return nil
+}
+
+//go:embed sql
+var fs embed.FS
+
+func (db *DB) upgrade() error {
+	// Does the version table exist? if not, make it.
+	// NOTE: This cannot change except via the next sql file.
+	err := db.Exec(context.Background(), `CREATE TABLE IF NOT EXISTS base (
+		id SERIAL PRIMARY KEY,
+		entry CHAR(12),
+		applied TIMESTAMP DEFAULT current_timestamp
+	)`)
+	if err != nil {
+		db.log("Upgrade failed.")
+		return err
+	}
+
+	// __Run scripts in order.__
+
+	landed := map[string]bool{}
+	{
+		var landedEntries []struct{ Entry string }
+		err = db.Select(context.Background(), &landedEntries, "SELECT entry FROM base")
+		if err != nil {
+			db.log("Cannot read entries: " + err.Error())
+			return err
+		}
+		for _, l := range landedEntries {
+			landed[l.Entry] = true
+		}
+	}
+	dir, err := fs.ReadDir("sql")
+	if err != nil {
+		db.log("Cannot read fs entries: " + err.Error())
+		return err
+	}
+	sort.Slice(dir, func(i, j int) bool { return dir[i].Name() < dir[j].Name() })
+	for _, e := range dir {
+		name := e.Name()
+		if landed[name] || !strings.HasSuffix(name, ".sql") {
+			continue
+		}
+		file, err := fs.ReadFile("sql/" + name)
+		if err != nil {
+			db.log("weird embed file read err")
+			return err
+		}
+		for _, s := range strings.Split(string(file), ";") { // Implement the changes.
+			if len(strings.TrimSpace(s)) == 0 {
+				continue
+			}
+			_, err = db.pgx.Exec(context.Background(), s)
+			if err != nil {
+				db.log(fmt.Sprintf("Could not upgrade! File %s, Query: %s, Returned: %s", name, s, err.Error()))
+				return err
+			}
+		}
+
+		// Mark Completed.
+		err = db.Exec(context.Background(), "INSERT INTO base (entry) VALUES ($1)", name)
+		if err != nil {
+			db.log("Cannot update base: " + err.Error())
+			return fmt.Errorf("cannot insert into base: %w", err)
+		}
+	}
+	return nil
+}

lib/harmony/harmonydb/metrics.go

@@ -0,0 +1,77 @@
+package harmonydb
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
+
+	"github.com/filecoin-project/lotus/metrics"
+)
+
+var (
+	dbTag, _         = tag.NewKey("db_name")
+	pre              = "harmonydb_base_"
+	waitsBuckets     = []float64{0, 10, 20, 30, 50, 80, 130, 210, 340, 550, 890}
+	whichHostBuckets = []float64{0, 1, 2, 3, 4, 5}
+)
+
+// DBMeasures groups all db metrics.
+var DBMeasures = struct {
+	Hits            *stats.Int64Measure
+	TotalWait       *stats.Int64Measure
+	Waits           prometheus.Histogram
+	OpenConnections *stats.Int64Measure
+	Errors          *stats.Int64Measure
+	WhichHost       prometheus.Histogram
+}{
+	Hits:      stats.Int64(pre+"hits", "Total number of uses.", stats.UnitDimensionless),
+	TotalWait: stats.Int64(pre+"total_wait", "Total delay. A numerator over hits to get average wait.", stats.UnitMilliseconds),
+	Waits: prometheus.NewHistogram(prometheus.HistogramOpts{
+		Name:    pre + "waits",
+		Buckets: waitsBuckets,
+		Help:    "The histogram of waits for query completions.",
+	}),
+	OpenConnections: stats.Int64(pre+"open_connections", "Total connection count.", stats.UnitDimensionless),
+	Errors:          stats.Int64(pre+"errors", "Total error count.", stats.UnitDimensionless),
+	WhichHost: prometheus.NewHistogram(prometheus.HistogramOpts{
+		Name:    pre + "which_host",
+		Buckets: whichHostBuckets,
+		Help:    "The index of the hostname being used",
+	}),
+}
+
+// CacheViews groups all cache-related default views.
+func init() {
+	metrics.RegisterViews(
+		&view.View{
+			Measure:     DBMeasures.Hits,
+			Aggregation: view.Sum(),
+			TagKeys:     []tag.Key{dbTag},
+		},
+		&view.View{
+			Measure:     DBMeasures.TotalWait,
+			Aggregation: view.Sum(),
+			TagKeys:     []tag.Key{dbTag},
+		},
+		&view.View{
+			Measure:     DBMeasures.OpenConnections,
+			Aggregation: view.LastValue(),
+			TagKeys:     []tag.Key{dbTag},
+		},
+		&view.View{
+			Measure:     DBMeasures.Errors,
+			Aggregation: view.Sum(),
+			TagKeys:     []tag.Key{dbTag},
+		},
+	)
+	err := prometheus.Register(DBMeasures.Waits)
+	if err != nil {
+		panic(err)
+	}
+
+	err = prometheus.Register(DBMeasures.WhichHost)
+	if err != nil {
+		panic(err)
+	}
+}

lib/harmony/harmonydb/sql/20230706.sql

@@ -0,0 +1,6 @@
+CREATE TABLE itest_scratch (
+    id SERIAL PRIMARY KEY,
+	content TEXT,
+    some_int INTEGER,
+	update_time TIMESTAMP DEFAULT current_timestamp
+)

lib/harmony/harmonydb/userfuncs.go

@@ -0,0 +1,149 @@
+package harmonydb
+
+import (
+	"context"
+
+	"github.com/georgysavva/scany/v2/pgxscan"
+	"github.com/jackc/pgx/v5"
+)
+
+type Intf interface {
+	ITestDeleteAll()
+	Exec(ctx context.Context, sql rawStringOnly, arguments ...any) error
+	Query(ctx context.Context, sql rawStringOnly, arguments ...any) (*Query, error)
+	QueryRow(ctx context.Context, sql rawStringOnly, arguments ...any) Row
+	Select(ctx context.Context, sliceOfStructPtr any, sql rawStringOnly, arguments ...any) error
+	BeginTransaction(ctx context.Context, f func(t *Transaction) (commit bool)) (retErr error)
+}
+
+// rawStringOnly is _intentionally_private_ to force only basic strings in SQL queries.
+// In any package, raw strings will satisfy compilation.  Ex:
+//
+//	harmonydb.Exec("INSERT INTO version (number) VALUES (1)")
+//
+// This prevents SQL injection attacks where the input contains query fragments.
+type rawStringOnly string
+
+// Exec executes changes (INSERT, DELETE,  or UPDATE).
+// Note, for CREATE & DROP please keep these permanent and express
+// them in the ./sql/ files (next number).
+func (db *DB) Exec(ctx context.Context, sql rawStringOnly, arguments ...any) error {
+	_, err := db.pgx.Exec(ctx, string(sql), arguments...)
+	return err
+}
+
+type Qry interface {
+	Next() bool
+	Err() error
+	Close()
+	Scan(...any) error
+	Values() ([]any, error)
+}
+
+// Query offers Next/Err/Close/Scan/Values/StructScan
+type Query struct {
+	Qry
+}
+
+// Query allows iterating returned values to save memory consumption
+// with the downside of needing to `defer q.Close()`. For a simpler interface,
+// try Select()
+// Next() must be called to advance the row cursor, including the first time:
+// Ex:
+// q, err := db.Query(ctx, "SELECT id, name FROM users")
+// handleError(err)
+// defer q.Close()
+//
+//	for q.Next() {
+//		  var id int
+//	   var name string
+//	   handleError(q.Scan(&id, &name))
+//	   fmt.Println(id, name)
+//	}
+func (db *DB) Query(ctx context.Context, sql rawStringOnly, arguments ...any) (*Query, error) {
+	q, err := db.pgx.Query(ctx, string(sql), arguments...)
+	return &Query{q}, err
+}
+func (q *Query) StructScan(s any) error {
+	return pgxscan.ScanRow(s, q.Qry.(pgx.Rows))
+}
+
+type Row interface {
+	Scan(...any) error
+}
+
+// QueryRow gets 1 row using column order matching.
+// This is a timesaver for the special case of wanting the first row returned only.
+// EX:
+//
+//	var name, pet string
+//	var ID = 123
+//	err := db.QueryRow(ctx, "SELECT name, pet FROM users WHERE ID=?", ID).Scan(&name, &pet)
+func (db *DB) QueryRow(ctx context.Context, sql rawStringOnly, arguments ...any) Row {
+	return db.pgx.QueryRow(ctx, string(sql), arguments...)
+}
+
+/*
+Select multiple rows into a slice using name matching
+Ex:
+
+	type user struct {
+		Name string
+		ID int
+		Number string `db:"tel_no"`
+	}
+
+	var users []user
+	pet := "cat"
+	err := db.Select(ctx, &users, "SELECT name, id, tel_no FROM customers WHERE pet=?", pet)
+*/
+func (db *DB) Select(ctx context.Context, sliceOfStructPtr any, sql rawStringOnly, arguments ...any) error {
+	return pgxscan.Select(ctx, db.pgx, sliceOfStructPtr, string(sql), arguments...)
+}
+
+type Transaction struct {
+	pgx.Tx
+}
+
+// BeginTransaction is how you can access transactions using this library.
+// The entire transaction happens in the function passed in.
+// The return must be true or a rollback will occur.
+func (db *DB) BeginTransaction(ctx context.Context, f func(t *Transaction) (commit bool)) (retErr error) {
+	tx, err := db.pgx.BeginTx(ctx, pgx.TxOptions{})
+	if err != nil {
+		return err
+	}
+	var commit bool
+	defer func() { // Panic clean-up.
+		if !commit {
+			retErr = tx.Rollback(ctx)
+		}
+	}()
+	commit = f(&Transaction{tx})
+	if commit {
+		return tx.Commit(ctx)
+	}
+	return nil
+}
+
+// Exec in a transaction.
+func (t *Transaction) Exec(ctx context.Context, sql rawStringOnly, arguments ...any) error {
+	_, err := t.Tx.Exec(ctx, string(sql), arguments...)
+	return err
+}
+
+// Query in a transaction.
+func (t *Transaction) Query(ctx context.Context, sql rawStringOnly, arguments ...any) (*Query, error) {
+	q, err := t.Tx.Query(ctx, string(sql), arguments...)
+	return &Query{q}, err
+}
+
+// QueryRow in a transaction.
+func (t *Transaction) QueryRow(ctx context.Context, sql rawStringOnly, arguments ...any) Row {
+	return t.Tx.QueryRow(ctx, string(sql), arguments...)
+}
+
+// Select in a transaction.
+func (t *Transaction) Select(ctx context.Context, sliceOfStructPtr any, sql rawStringOnly, arguments ...any) error {
+	return pgxscan.Select(ctx, t.Tx, sliceOfStructPtr, string(sql), arguments...)
+}