[-] prevent running several instances with the same clientname, fixes #475 (#476)

* `[-]` prevent running several instances with the same `clientname`, fixes #475
* `[*]` increase connection timeout in SetupTestCase
* `[-]` add `pge.Finalize()` to migrator tests

Author: pashagolub

internal/migrator/migrator_test.go

@@ -86,6 +86,8 @@ func TestBadMigrations(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer pge.Finalize()
+
 	ctx := context.Background()
 	db, err := pge.ConfigDb.Acquire(ctx)
 	if err != nil {
@@ -143,6 +145,7 @@ func TestPending(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer pge.Finalize()
 	ctx := context.Background()
 	db, err := pge.ConfigDb.Acquire(ctx)
 	if err != nil {

internal/pgengine/access.go

@@ -3,7 +3,6 @@ package pgengine
 import (
 	"context"
 	"fmt"
-	"os"
 	"strings"
 
 	"github.com/georgysavva/scany/pgxscan"
@@ -35,7 +34,7 @@ chain_id, task_id, command, kind, last_run, finished, returncode, pid, output, c
 VALUES ($1, $2, $3, $4, clock_timestamp() - $5 :: interval, clock_timestamp(), $6, $7, NULLIF($8, ''), $9, $10)`,
 		task.ChainID, task.TaskID, task.Script, task.Kind,
 		fmt.Sprintf("%f seconds", float64(task.Duration)/1000000),
-		retCode, os.Getpid(), strings.TrimSpace(output), pge.ClientName, task.Txid)
+		retCode, pge.Getpid(), strings.TrimSpace(output), pge.ClientName, task.Txid)
 	if err != nil {
 		pge.l.WithError(err).Error("Failed to log chain element execution status")
 	}
@@ -65,7 +64,7 @@ func (pge *PgEngine) RemoveChainRunStatus(ctx context.Context, chainID int) {
 	}
 }
 
-//Select live chains with proper client_name value
+// Select live chains with proper client_name value
 const sqlSelectLiveChains = `SELECT chain_id, chain_name, self_destruct, exclusive_execution, COALESCE(timeout, 0) as timeout, COALESCE(max_instances, 16) as max_instances
 FROM timetable.chain WHERE live AND (client_name = $1 or client_name IS NULL)`
 

internal/pgengine/bootstrap.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io/ioutil"
-	"os"
+	"math/rand"
 	"time"
 
 	"github.com/cybertec-postgresql/pg_timetable/internal/config"
@@ -51,6 +51,18 @@ type PgEngine struct {
 	config.CmdOptions
 	// NOTIFY messages passed verification are pushed to this channel
 	chainSignalChan chan ChainSignal
+	pid             int32
+}
+
+// Getpid returns the pseudo-random process ID to use for the session identification.
+// Previously `os.Getpid()` used but this approach is not producing unique IDs for docker containers
+// where all IDs are the same across all running containers, e.g. 1
+func (pge *PgEngine) Getpid() int32 {
+	if pge.pid == 0 {
+		rand.Seed(time.Now().UnixNano())
+		pge.pid = rand.Int31()
+	}
+	return pge.pid
 }
 
 var sqls = []string{sqlDDL, sqlJSONSchema, sqlCronFunctions, sqlJobFunctions}
@@ -61,12 +73,12 @@ func New(ctx context.Context, cmdOpts config.CmdOptions, logger log.LoggerHooker
 	var wt int = WaitTime
 	var err error
 	pge := &PgEngine{
-		logger,
-		nil,
-		cmdOpts,
-		make(chan ChainSignal, 64),
+		l:               logger,
+		ConfigDb:        nil,
+		CmdOptions:      cmdOpts,
+		chainSignalChan: make(chan ChainSignal, 64),
 	}
-	pge.l.WithField("PID", os.Getpid()).Debug("Starting new session... ")
+	pge.l.WithField("PID", pge.Getpid()).Info("Starting new session... ")
 	connctx, conncancel := context.WithTimeout(ctx, time.Duration(cmdOpts.Connection.Timeout)*time.Second)
 	defer conncancel()
 
@@ -106,7 +118,12 @@ func New(ctx context.Context, cmdOpts config.CmdOptions, logger log.LoggerHooker
 // NewDB creates pgengine instance for already opened database connection, allowing to bypass a parameters based credentials.
 // We assume here all checks for proper schema validation are done beforehannd
 func NewDB(DB PgxPoolIface, args ...string) *PgEngine {
-	return &PgEngine{log.Init(config.LoggingOpts{LogLevel: "error"}), DB, *config.NewCmdOptions(args...), make(chan ChainSignal, 64)}
+	return &PgEngine{
+		l:               log.Init(config.LoggingOpts{LogLevel: "error"}),
+		ConfigDb:        DB,
+		CmdOptions:      *config.NewCmdOptions(args...),
+		chainSignalChan: make(chan ChainSignal, 64),
+	}
 }
 
 // getPgxConnConfig transforms standard connestion string to pgx specific one with
@@ -159,7 +176,7 @@ func (pge *PgEngine) getPgxConnConfig() *pgxpool.Config {
 
 // AddLogHook adds a new pgx log hook to logrus logger
 func (pge *PgEngine) AddLogHook(ctx context.Context) {
-	pge.l.AddHook(NewHook(ctx, pge.ConfigDb, pge.ClientName, 500, pge.Logging.LogDBLevel))
+	pge.l.AddHook(NewHook(ctx, pge, pge.Logging.LogDBLevel))
 }
 
 // QueryRowIface specifies interface to use QueryRow method
@@ -182,7 +199,7 @@ func (pge *PgEngine) TryLockClientName(ctx context.Context, conn QueryRowIface)
 	}
 	var wt int = WaitTime
 	for {
-		sql := fmt.Sprintf("SELECT timetable.try_lock_client_name(%d, $worker$%s$worker$)", os.Getpid(), pge.ClientName)
+		sql := fmt.Sprintf("SELECT timetable.try_lock_client_name(%d, $worker$%s$worker$)", pge.Getpid(), pge.ClientName)
 		var locked bool
 		err = conn.QueryRow(ctx, sql).Scan(&locked)
 		if err != nil {
@@ -262,7 +279,7 @@ DELETE FROM timetable.active_session WHERE client_name = $1`
 	pge.ConfigDb = nil
 }
 
-//Reconnect keeps trying reconnecting every `waitTime` seconds till connection established
+// Reconnect keeps trying reconnecting every `waitTime` seconds till connection established
 func (pge *PgEngine) Reconnect(ctx context.Context) bool {
 	for pge.ConfigDb.Ping(ctx) != nil {
 		pge.l.Info("Connection to the server was lost. Waiting for ", WaitTime, " sec...")

internal/pgengine/log_hook.go

@@ -3,7 +3,6 @@ package pgengine
 import (
 	"context"
 	"encoding/json"
-	"os"
 	"time"
 
 	pgx "github.com/jackc/pgx/v4"
@@ -19,23 +18,24 @@ type LogHook struct {
 	input           chan logrus.Entry
 	ctx             context.Context
 	lastError       chan error
-	pid             int
+	pid             int32
 	client          string
 	level           string
 }
 
 // NewHook creates a LogHook to be added to an instance of logger
-func NewHook(ctx context.Context, db PgxPoolIface, client string, cacheLimit int, level string) *LogHook {
+func NewHook(ctx context.Context, pge *PgEngine, level string) *LogHook {
+	cacheLimit := 500
 	l := &LogHook{
 		cacheLimit:      cacheLimit,
 		cacheTimeout:    2 * time.Second,
 		highLoadTimeout: 200 * time.Millisecond,
-		db:              db,
+		db:              pge.ConfigDb,
 		input:           make(chan logrus.Entry, cacheLimit),
 		lastError:       make(chan error),
 		ctx:             ctx,
-		pid:             os.Getpid(),
-		client:          client,
+		pid:             pge.Getpid(),
+		client:          pge.ClientName,
 		level:           level,
 	}
 	go l.poll(l.input)

internal/pgengine/log_hook_test.go

@@ -49,15 +49,15 @@ func TestLogHook(t *testing.T) {
 func TestCancelledContext(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
-	h := NewHook(ctx, nil, "foo", 100, "debug")
+	h := NewHook(ctx, &PgEngine{}, "debug")
 	assert.Equal(t, h.Levels(), logrus.AllLevels)
 	assert.NoError(t, h.Fire(&logrus.Entry{}))
 }
 
 func TestFireError(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	h := NewHook(ctx, nil, "foo", 100, "debug")
+	h := NewHook(ctx, &PgEngine{}, "debug")
 	err := errors.New("fire error")
 	go func() { h.lastError <- err }()
 	<-time.After(time.Second)

internal/pgengine/pgengine_test.go

@@ -29,10 +29,10 @@ func SetupTestCaseEx(t *testing.T, fc func(c *config.CmdOptions)) func(t *testin
 	return SetupTestCase(t)
 }
 
-//SetupTestCase used to connect and to initialize test PostgreSQL database
+// SetupTestCase used to connect and to initialize test PostgreSQL database
 func SetupTestCase(t *testing.T) func(t *testing.T) {
 	t.Helper()
-	timeout := time.After(6 * time.Second)
+	timeout := time.After(30 * time.Second)
 	done := make(chan bool)
 	go func() {
 		pge, _ = pgengine.New(context.Background(), *cmdOpts, log.Init(config.LoggingOpts{LogLevel: "error"}))

internal/pgengine/sql/ddl.sql

@@ -107,8 +107,8 @@ COMMENT ON TABLE timetable.parameter IS
 
 CREATE UNLOGGED TABLE timetable.active_session(
     client_pid  BIGINT  NOT NULL,
-    client_name TEXT    NOT NULL,
     server_pid  BIGINT  NOT NULL,
+    client_name TEXT    NOT NULL,
     started_at  TIMESTAMPTZ DEFAULT now()
 );