Merge pull request #403 from cybertec-postgresql/394_started_at_columns

[+] add `started_at` column to `active_session` and `active_chain tables`, closes #394

internal/pgengine/migration.go

@@ -96,6 +96,12 @@ var Migrations func() migrator.Option = func() migrator.Option {
 				return ExecuteMigrationScript(ctx, tx, "00381.sql")
 			},
 		},
+		&migrator.Migration{
+			Name: "00394 Add started_at column to active_session and active_chain tables",
+			Func: func(ctx context.Context, tx pgx.Tx) error {
+				return ExecuteMigrationScript(ctx, tx, "00394.sql")
+			},
+		},
 		// adding new migration here, update "timetable"."migration" in "sql/ddl.sql"
 		// and "dbapi" variable in main.go!
 

internal/pgengine/sql/ddl.sql

@@ -18,7 +18,8 @@ VALUES
     (2, '00323 Append timetable.delete_job function'),
     (3, '00329 Migration required for some new added functions'),
     (4, '00334 Refactor timetable.task as plain schema without tree-like dependencies'),
-    (5, '00381 Rewrite active chain handling');
+    (5, '00381 Rewrite active chain handling'),
+    (6, '00394 Add started_at column to active_session and active_chain tables');
 
 CREATE DOMAIN timetable.cron AS TEXT CHECK(
     substr(VALUE, 1, 6) IN ('@every', '@after') AND (substr(VALUE, 7) :: INTERVAL) IS NOT NULL
@@ -105,7 +106,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
+    server_pid  BIGINT  NOT NULL,
+    started_at  TIMESTAMPTZ DEFAULT now()
 );
 
 COMMENT ON TABLE timetable.active_session IS
@@ -150,7 +152,8 @@ COMMENT ON TABLE timetable.execution_log IS
 
 CREATE UNLOGGED TABLE timetable.active_chain(
     chain_id    BIGINT  NOT NULL,
-    client_name TEXT    NOT NULL
+    client_name TEXT    NOT NULL,
+    started_at  TIMESTAMPTZ DEFAULT now()
 );
 
 COMMENT ON TABLE timetable.active_chain IS
@@ -189,7 +192,7 @@ BEGIN
         RETURN FALSE;
     END IF;
     -- insert current session information
-    INSERT INTO timetable.active_session VALUES (worker_pid, worker_name, pg_backend_pid());
+    INSERT INTO timetable.active_session(client_pid, client_name, server_pid) VALUES (worker_pid, worker_name, pg_backend_pid());
     RETURN TRUE;
 END;
 $CODE$

internal/pgengine/sql/migrations/00394.sql

@@ -0,0 +1,45 @@
+ALTER TABLE timetable.active_session
+    ADD COLUMN started_at  TIMESTAMPTZ DEFAULT now();
+
+ALTER TABLE timetable.active_chain
+    ADD COLUMN started_at  TIMESTAMPTZ DEFAULT now();
+
+CREATE OR REPLACE FUNCTION timetable.try_lock_client_name(worker_pid BIGINT, worker_name TEXT)
+RETURNS bool AS
+$CODE$
+BEGIN
+    IF pg_is_in_recovery() THEN
+        RAISE NOTICE 'Cannot obtain lock on a replica. Please, use the primary node';
+        RETURN FALSE;
+    END IF;
+    -- remove disconnected sessions
+    DELETE
+        FROM timetable.active_session
+        WHERE server_pid NOT IN (
+            SELECT pid
+            FROM pg_catalog.pg_stat_activity
+            WHERE application_name = 'pg_timetable'
+        );
+    DELETE 
+        FROM timetable.active_chain 
+        WHERE client_name NOT IN (
+            SELECT client_name FROM timetable.active_session
+        );
+    -- check if there any active sessions with the client name but different client pid
+    PERFORM 1
+        FROM timetable.active_session s
+        WHERE
+            s.client_pid <> worker_pid
+            AND s.client_name = worker_name
+        LIMIT 1;
+    IF FOUND THEN
+        RAISE NOTICE 'Another client is already connected to server with name: %', worker_name;
+        RETURN FALSE;
+    END IF;
+    -- insert current session information
+    INSERT INTO timetable.active_session(client_pid, client_name, server_pid) VALUES (worker_pid, worker_name, pg_backend_pid());
+    RETURN TRUE;
+END;
+$CODE$
+STRICT
+LANGUAGE plpgsql;

main.go

@@ -47,7 +47,7 @@ var (
 	commit  string = "000000"
 	version string = "master"
 	date    string = "unknown"
-	dbapi   string = "00381"
+	dbapi   string = "00394"
 )
 
 func printVersion() {