pgengine

Documentation

Index

• Constants
• Variables
• func CanProceedChainExecution(chainConfigID int, maxInstances int) bool
• func CheckNeedMigrateDb()
• func DeleteChainConfig(chainConfigID int) bool
• func ExecuteSQLCommand(tx *sqlx.Tx, script string, paramValues []string) error
• func FinalizeConfigDBConnection()
• func FinalizeRemoteDBConnection(remoteDb *sqlx.DB)
• func FixSchedulerCrash()
• func GetChainElements(tx *sqlx.Tx, chains interface{}, chainID int) bool
• func GetChainParamValues(tx *sqlx.Tx, paramValues interface{}, chainElemExec *ChainElementExecution) bool
• func GetConnectionString(databaseConnection sql.NullString) (connectionString string)
• func GetLogPrefix(level string) string
• func GetLogPrefixLn(level string) string
• func GetRemoteDBTransaction(connectionString string) (*sqlx.DB, *sqlx.Tx)
• func InitAndTestConfigDBConnection()
• func InsertChainRunStatus(tx *sqlx.Tx, chainConfigID int, chainID int) int
• func LogChainElementExecution(chainElemExec *ChainElementExecution, retCode int)
• func LogToDB(level string, msg ...interface{})
• func MigrateDb()
• func MustCommitTransaction(tx *sqlx.Tx)
• func MustRollbackTransaction(tx *sqlx.Tx)
• func ReconnectDbAndFixLeftovers()
• func ResetRole(tx *sqlx.Tx)
• func SetRole(tx *sqlx.Tx, runUID sql.NullString)
• func SetupCloseHandler()
• func StartTransaction() *sqlx.Tx
• func TryLockClientName() (res bool)
• func UpdateChainRunStatus(chainElemExec *ChainElementExecution, runStatusID int, status string)
• type ChainElementExecution
  • func (chainElem ChainElementExecution) String() string

Constants

const AppID = 0x204F04EE

AppID used as a key for obtaining locks on the server, it's Adler32 hash of 'pg_timetable' string

const InvalidOid = 0

InvalidOid specifies value for non-existent objects

Variables

var ClientName string

ClientName is unique ifentifier of the scheduler application running

var ConfigDb *sqlx.DB

ConfigDb is the global database object

var DbName string = "timetable"

DbName is used to reconnect to data base

var Host string = "localhost"

Host is used to reconnect to data base

var Password string = "somestrong"

Password is used to Reconnect Data base

var Port string = "5432"

Port is used to reconnect to data base

var SSLMode string = "disable"

SSLMode parameter determines whether or with what priority a secure SSL TCP/IP connection will be negotiated with the server

var Upgrade bool

Upgrade parameter specifies if database should be upgraded to latest version

var User string = "scheduler"

User is used to reconnect to data base

var VerboseLogLevel = true

VerboseLogLevel specifies if log messages with level LOG should be logged

Functions

func CanProceedChainExecution

func CanProceedChainExecution(chainConfigID int, maxInstances int) bool

CanProceedChainExecution checks if particular chain can be exeuted in parallel

func CheckNeedMigrateDb

func CheckNeedMigrateDb()

CheckNeedMigrateDb checks need of upgrading database and throws error if that's true

func DeleteChainConfig

func DeleteChainConfig(chainConfigID int) bool

DeleteChainConfig delete chaing configuration for self destructive chains

func ExecuteSQLCommand

func ExecuteSQLCommand(tx *sqlx.Tx, script string, paramValues []string) error

ExecuteSQLCommand executes chain script with parameters inside transaction

func FinalizeConfigDBConnection

func FinalizeConfigDBConnection()

FinalizeConfigDBConnection closes session

func FinalizeRemoteDBConnection

func FinalizeRemoteDBConnection(remoteDb *sqlx.DB)

FinalizeRemoteDBConnection closes session

func FixSchedulerCrash

func FixSchedulerCrash()

FixSchedulerCrash make sure that task chains which are not complete due to a scheduler crash are "fixed" and marked as stopped at a certain point

func GetChainElements

func GetChainElements(tx *sqlx.Tx, chains interface{}, chainID int) bool

GetChainElements returns all elements for a given chain

func GetChainParamValues

func GetChainParamValues(tx *sqlx.Tx, paramValues interface{}, chainElemExec *ChainElementExecution) bool

GetChainParamValues returns parameter values to pass for task being executed

func GetConnectionString

func GetConnectionString(databaseConnection sql.NullString) (connectionString string)

GetConnectionString of database_connection

func GetLogPrefix

func GetLogPrefix(level string) string

GetLogPrefix perform formatted logging

func GetLogPrefixLn

func GetLogPrefixLn(level string) string

GetLogPrefixLn perform formatted logging with new line at the end

func GetRemoteDBTransaction

func GetRemoteDBTransaction(connectionString string) (*sqlx.DB, *sqlx.Tx)

GetRemoteDBTransaction create a remote db connection and returns transaction object

func InitAndTestConfigDBConnection

func InitAndTestConfigDBConnection()

InitAndTestConfigDBConnection opens connection and creates schema

func InsertChainRunStatus

func InsertChainRunStatus(tx *sqlx.Tx, chainConfigID int, chainID int) int

InsertChainRunStatus inits the execution run log, which will be use to effectively control scheduler concurrency

func LogChainElementExecution

func LogChainElementExecution(chainElemExec *ChainElementExecution, retCode int)

LogChainElementExecution will log current chain element execution status including retcode

func LogToDB

func LogToDB(level string, msg ...interface{})

LogToDB performs logging to configuration database ConfigDB initiated during bootstrap

func MigrateDb

func MigrateDb()

MigrateDb upgrades database with all migrations

func MustCommitTransaction

func MustCommitTransaction(tx *sqlx.Tx)

MustCommitTransaction commits transaction and log error in the case of error

func MustRollbackTransaction

func MustRollbackTransaction(tx *sqlx.Tx)

MustRollbackTransaction rollbacks transaction and log error in the case of error

func ReconnectDbAndFixLeftovers

func ReconnectDbAndFixLeftovers()

ReconnectDbAndFixLeftovers keeps trying reconnecting every `waitTime` seconds till connection established

func ResetRole

func ResetRole(tx *sqlx.Tx)

ResetRole - RESET forms reset the current user identifier to be the current session user identifier

func SetRole

func SetRole(tx *sqlx.Tx, runUID sql.NullString)

SetRole - set the current user identifier of the current session

func SetupCloseHandler

func SetupCloseHandler()

SetupCloseHandler creates a 'listener' on a new goroutine which will notify the program if it receives an interrupt from the OS. We then handle this by calling our clean up procedure and exiting the program.

func StartTransaction

func StartTransaction() *sqlx.Tx

StartTransaction return transaction object and panic in the case of error

func TryLockClientName

func TryLockClientName() (res bool)

TryLockClientName obtains lock on the server to prevent another client with the same name

func UpdateChainRunStatus

func UpdateChainRunStatus(chainElemExec *ChainElementExecution, runStatusID int, status string)

UpdateChainRunStatus inserts status information about running chain elements

Types

type ChainElementExecution

type ChainElementExecution struct {
	ChainConfig        int            `db:"chain_config"`
	ChainID            int            `db:"chain_id"`
	TaskID             int            `db:"task_id"`
	TaskName           string         `db:"task_name"`
	Script             string         `db:"script"`
	Kind               string         `db:"kind"`
	RunUID             sql.NullString `db:"run_uid"`
	IgnoreError        bool           `db:"ignore_error"`
	DatabaseConnection sql.NullString `db:"database_connection"`
	ConnectString      sql.NullString `db:"connect_string"`
	StartedAt          time.Time
	Duration           int64 // in microseconds
}

ChainElementExecution structure describes each chain execution process

func (ChainElementExecution) String

func (chainElem ChainElementExecution) String() string