next time

new custom var
bug fixes

Author: Vladimir Ershov

pgpro_scheduler--1.0.sql

@@ -24,7 +24,8 @@ CREATE TABLE schedule.cron(
    max_instances integer default 1,
    start_date timestamp with time zone,
    end_date timestamp with time zone,
-   reason text
+   reason text,
+   _next_exec_time timestamp with time zone
 );
 
 CREATE TABLE schedule.at(

src/pgpro_scheduler.c

@@ -39,6 +39,7 @@ volatile sig_atomic_t got_sigterm = false;
 
 static char *sched_databases = "";
 static char *sched_nodename = "master";
+static char *sched_transaction_state = "undefined";
 static int sched_max_workers = 2;
 
 extern void
@@ -91,7 +92,7 @@ char *make_date_from_timestamp(TimestampTz ts)
 	fsec_t fsec;
 	const char *tzn;
 
-	timestamp2tm(ts, &tz, &dt, &fsec, &tzn, NULL ); /* TODO ERROR */
+	timestamp2tm(ts, &tz, &dt, &fsec, &tzn, NULL ); 
 	sprintf(str, "%04d-%02d-%02d %02d:%02d", dt.tm_year , dt.tm_mon,
 			dt.tm_mday, dt.tm_hour, dt.tm_min);
 	return str;
@@ -180,13 +181,11 @@ char_array_t *readBasesToCheck(void)
 		pfree(clean_value);
 		return result;
 	}
-	elog(LOG, "clean value: %s [%d,%d]", clean_value, cv_len, nnames);
 	names = makeCharArray();
 	for(i=0; i < cv_len + 1; i++)
 	{
 		if(clean_value[i] == 0)
 		{
-			elog(LOG, "start position: %d", start_pos);
 			ptr = clean_value + start_pos;
 			if(strlen(ptr)) pushCharArray(names, ptr);
 			start_pos = i + 1;
@@ -259,7 +258,6 @@ void parent_scheduler_main(Datum arg)
 
 	BackgroundWorkerInitializeConnection("postgres", NULL);
 	names = readBasesToCheck();
-	elog(LOG, "GOT NAMES");
 	poll = initSchedulerManagerPool(names);
 	destroyCharArray(names);
 
@@ -374,6 +372,18 @@ void _PG_init(void)
 		NULL,
 		NULL
 	);
+	DefineCustomStringVariable(
+		"schedule.transaction_state",
+		"State of scheduler executor transaction",
+		"If not under scheduler executor process the variable has no mean and has a value = 'undefined', possible values: progress, success, failure",
+		&sched_transaction_state , 
+		"undefined",
+		PGC_INTERNAL,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
 	DefineCustomIntVariable(
 		"schedule.max_workers",
 		"How much workers can serve scheduler on one database",

src/scheduler_executor.c

@@ -58,13 +58,12 @@ void executor_worker_main(Datum arg)
 	dsm_segment *seg;
 	job_t *job;
 	int i;
-	bool success = true;
 	executor_error_t EE;
 	int ret;
 	char *error = NULL;
-	bool use_pg_vars = true;
+	/* bool use_pg_vars = true; */
+	/* bool success = true; */
 	schd_executor_status_t status;
-	/* int rc = 0; */
 
 	EE.n = 0;
 	EE.errors = NULL;
@@ -127,6 +126,7 @@ void executor_worker_main(Datum arg)
 	CHECK_FOR_INTERRUPTS();
 	/* rc = WaitLatch(MyLatch, WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, 0);
 	ResetLatch(MyLatch); */
+	SetConfigOption("schedule.transaction_state", "running", PGC_INTERNAL, PGC_S_SESSION);
 
 	if(job->same_transaction)
 	{
@@ -143,8 +143,9 @@ void executor_worker_main(Datum arg)
 		ret = execute_spi(job->dosql[i], &error);
 		if(ret < 0)
 		{
-			success = false;
+			/* success = false; */
 			status = SchdExecutorError;
+			SetConfigOption("schedule.transaction_state", "failure", PGC_INTERNAL, PGC_S_SESSION);
 			if(error)
 			{
 				push_executor_error(&EE, "error on %d: %s", i+1, error);
@@ -174,13 +175,15 @@ void executor_worker_main(Datum arg)
 			STOP_SPI_SNAP();
 		}
 		status = SchdExecutorDone;
+		SetConfigOption("schedule.transaction_state", "success", PGC_INTERNAL, PGC_S_SESSION);
 	}
 	if(job->next_time_statement)
 	{
-		if(use_pg_vars)  /* may be to define custom var is better */
+/*		if(use_pg_vars)  
 		{
 			set_pg_var(success, &EE);
 		}
+*/
 		shared->next_time = get_next_excution_time(job->next_time_statement, &EE);
 	}
 	pgstat_report_activity(STATE_RUNNING, "finish job processing");
@@ -261,7 +264,7 @@ void set_shared_message(schd_executor_share_t *shared, executor_error_t *ee)
 		}
 		else
 		{
-			memcpy(ptr, ", ", 2);
+			memcpy(ptr, "; ", 2);
 			left -= 2;
 			ptr += 2;
 		}
@@ -277,7 +280,7 @@ TimestampTz get_next_excution_time(char *sql, executor_error_t *ee)
 	bool isnull;
 
 	START_SPI_SNAP();
-	pgstat_report_activity(STATE_RUNNING, "finish job processing");
+	pgstat_report_activity(STATE_RUNNING, "culc next time execution time");
 	ret = execute_spi(sql, &error);
 	if(ret < 0)
 	{
@@ -333,7 +336,6 @@ int executor_onrollback(job_t *job, executor_error_t *ee)
 	{
 		if(error)
 		{
-			elog(LOG, "EXECUTOR: onrollback error: %s", error);
 			push_executor_error(ee, "onrollback error: %s", error);
 			pfree(error);
 		}
@@ -371,7 +373,7 @@ void set_pg_var(bool result, executor_error_t *ee)
 		}
 		else
 		{
-			push_executor_error(ee, "set variable: code: %d", ret);
+			push_executor_error(ee, "set variable error code: %d", ret);
 		}
 	}
 }

src/scheduler_job.c

@@ -25,7 +25,7 @@ job_t *get_jobs_to_do(char *nodename, int *n, int *is_error)
 	int ret, got, i;
 	Oid argtypes[1] = { TEXTOID };
 	Datum values[1];
-	const char *get_job_sql = "select at.start_at, at.last_start_available, at.cron, max_run_time, cron.max_instances, cron.executor from schedule.at at, schedule.cron cron where start_at <= 'now' and not at.active and (last_start_available is NULL OR last_start_available > 'now') and at.cron = cron.id AND cron.node = $1 order by at.start_at";
+	const char *get_job_sql = "select at.start_at, at.last_start_available, at.cron, max_run_time, cron.max_instances, cron.executor, cron.next_time_statement from schedule.at at, schedule.cron cron where start_at <= 'now' and not at.active and (last_start_available is NULL OR last_start_available > 'now') and at.cron = cron.id AND cron.node = $1 order by at.start_at";
 
 	*is_error = *n = 0;
 	START_SPI_SNAP();
@@ -48,6 +48,7 @@ job_t *get_jobs_to_do(char *nodename, int *n, int *is_error)
 				jobs[i].max_instances = get_int_from_spi(i, 5, 1);
 				jobs[i].node = _copy_string(nodename);
 				jobs[i].executor = get_text_from_spi(i, 6);
+				jobs[i].next_time_statement = get_text_from_spi(i, 7);
 			}
 		}
 	}