scheduler

package
v5.4.5 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Apr 30, 2024 License: MIT Imports: 24 Imported by: 0

Documentation

Index

Constants

View Source
const (
	Day  string = "day"
	Hour string = "hour"
)

Internal Job

View Source
const (
	KeyMetadataDependsOn   = "depends_on_metadata"
	ValueMetadataDependsOn = "depends_on_metadata_value"
	ActionSetValue         = "set"
	ActionPendingValue     = "pending"
	IDSituationDependsOn   = "id_situation_depends_on"
	IDInstanceDependsOn    = "id_instance_depends_on"
)

Variables

This section is empty.

Functions

func CalculateAndPersistFacts

func CalculateAndPersistFacts(t time.Time, factIDs []int64) (map[string]history.HistoryRecordV4, error)

func CalculateAndPersistSituations

func CalculateAndPersistSituations(localRuleEngine *ruleeng.RuleEngine, situationsToUpdate map[string]history.HistoryRecordV4) ([]tasker.TaskBatch, error)

func GetEnabledSituations

func GetEnabledSituations(fact engine.Fact, t time.Time) ([]history.HistoryRecordV4, error)

GetEnabledSituations returns all situation linked to a fact

func GetLinkedSituations

func GetLinkedSituations(fact engine.Fact) ([]history.HistoryRecordV4, error)

GetLinkedSituations returns all situation linked to a fact

func ReceiveAndPersistFacts

func ReceiveAndPersistFacts(aggregates []ExternalAggregate) (map[string]history.HistoryRecordV4, error)

ReceiveAndPersistFacts process a slice of ExternalAggregates and trigger all standard fact-situation-rule process

func ReplaceGlobalRepository

func ReplaceGlobalRepository(repository Repository) func()

ReplaceGlobalRepository affect a new repository to the global repository singleton

func ReplaceGlobals

func ReplaceGlobals(scheduler *InternalScheduler) func()

ReplaceGlobals affect a new repository to the global fact scheduler singleton

Types

type BaselineCalculationJob added in v5.1.0

type BaselineCalculationJob struct {
	BaselineIds []int64 `json:"baselineIds"`
	Debug       bool    `json:"debug"`
	ScheduleID  int64   `json:"-"`
}

BaselineCalculationJob represent a scheduler job instance which process a group of baselines, and persist the result in postgresql

func (BaselineCalculationJob) IsValid added in v5.1.0

func (job BaselineCalculationJob) IsValid() (bool, error)

IsValid checks if an internal schedule job definition is valid and has no missing mandatory fields

func (BaselineCalculationJob) Run added in v5.1.0

func (job BaselineCalculationJob) Run()

Run contains all the business logic of the job

func (*BaselineCalculationJob) UnmarshalJSON added in v5.1.0

func (job *BaselineCalculationJob) UnmarshalJSON(data []byte) error

UnmarshalJSON unmarshals a quoted json string to a valid BaselineCalculationJob struct

type CompactHistoryJob added in v5.2.3

type CompactHistoryJob struct {
	SituationID         int64             `json:"situationId"`
	SituationInstanceID int64             `json:"situationInstanceId"`
	ParameterFilters    map[string]string `json:"parameterFilters"`
	FromOffset          string            `json:"fromOffset"`
	ToOffset            string            `json:"toOffset"`
	Interval            string            `json:"interval"`
	ScheduleID          int64             `json:"-"`
}

CpmpactHistoryJob represent a scheduler job instance which process a group of compact history, and persist the result in postgresql

func (CompactHistoryJob) IsValid added in v5.2.3

func (job CompactHistoryJob) IsValid() (bool, error)

IsValid checks if an internal schedule job definition is valid and has no missing mandatory fields

func (CompactHistoryJob) Run added in v5.2.3

func (job CompactHistoryJob) Run()

Run contains all the business logic of the job

type ExternalAggregate

type ExternalAggregate struct {
	FactID              int64       `json:"factId"`
	SituationID         int64       `json:"situationId"`
	SituationInstanceID int64       `json:"situationInstanceId"`
	Time                time.Time   `json:"time"`
	Value               reader.Item `json:"value"`
}

ExternalAggregate contains all information to store a new aggregat in postgresql

type FactCalculationJob

type FactCalculationJob struct {
	FactIds        []int64 `json:"factIds"`
	From           string  `json:"from,omitempty"`
	To             string  `json:"to,omitempty"`
	LastDailyValue bool    `json:"lastDailyValue,omitempty"`
	Debug          bool    `json:"debug"`
	ScheduleID     int64   `json:"-"`
}

FactCalculationJob represent a scheduler job instance which process a group of facts, and persist the result in postgresql It also generate situations, persists them and notify the rule engine to evaluate them

func (FactCalculationJob) IsValid

func (job FactCalculationJob) IsValid() (bool, error)

IsValid checks if an internal schedule job definition is valid and has no missing mandatory fields

func (*FactCalculationJob) ResolveFromAndTo

func (job *FactCalculationJob) ResolveFromAndTo(t time.Time) (time.Time, time.Time, error)

ResolveFromAndTo resolves the expressions in parameters From and To

func (FactCalculationJob) Run

func (job FactCalculationJob) Run()

Run contains all the business logic of the job

func (*FactCalculationJob) UnmarshalJSON

func (job *FactCalculationJob) UnmarshalJSON(data []byte) error

UnmarshalJSON unmarshals a quoted json string to a valid FactCalculationJob struct

type FactRecalculationJob added in v5.0.13

type FactRecalculationJob struct {
	FactIds        []int64 `json:"factIds"`
	From           string  `json:"from,omitempty"`
	To             string  `json:"to,omitempty"`
	LastDailyValue bool    `json:"lastDailyValue,omitempty"`
	Debug          bool    `json:"debug"`
	ScheduleID     int64   `json:"-"`
}

FactRecalculationJob represent a scheduler job instance which process a group of facts, and persist the result in postgresql It also generate situations, persists them and notify the rule engine to evaluate them

func (FactRecalculationJob) FetchRecalculationData added in v5.0.13

func (job FactRecalculationJob) FetchRecalculationData(historySituations []history.HistorySituationsV4) ([]history.HistoryFactsV4, map[int64][]int64, map[int64]int64, error)

func (FactRecalculationJob) RecalculateAndUpdateFacts added in v5.0.13

func (job FactRecalculationJob) RecalculateAndUpdateFacts(factIDs []int64, facts map[int64]engine.Fact,
	mapFactSituation map[int64]int64, mapSituations map[int64]history.HistorySituationsV4, historyFacts []history.HistoryFactsV4) (map[int64]history.HistoryFactsV4, error)

func (FactRecalculationJob) RecalculateAndUpdateSituations added in v5.0.13

func (job FactRecalculationJob) RecalculateAndUpdateSituations(localRuleEngine *ruleeng.RuleEngine, s situation.Situation, mapSituationFact map[int64][]int64,
	historySituations []history.HistorySituationsV4, newFactHistory map[int64]history.HistoryFactsV4) error

func (*FactRecalculationJob) ResolveFromAndTo added in v5.0.13

func (job *FactRecalculationJob) ResolveFromAndTo(t time.Time) (time.Time, time.Time, error)

ResolveFromAndTo resolves the expressions in parameters From and To

func (FactRecalculationJob) Run added in v5.0.13

func (job FactRecalculationJob) Run()

type InternalJob

type InternalJob interface {
	cron.Job
	IsValid() (bool, error)
}

InternalJob embed the external "standard" cron job with some additionnal data

func UnmarshalInternalJob

func UnmarshalInternalJob(t string, b json.RawMessage, scheduleID int64) (InternalJob, error)

UnmarshalInternalJob unmarshal a fact from a json string

type InternalSchedule

type InternalSchedule struct {
	ID       int64       `json:"id"`
	Name     string      `json:"name"`
	CronExpr string      `json:"cronexpr" example:"0 */15 * * *"`
	JobType  string      `json:"jobtype" enums:"fact,baseline,compact,purge"`
	Job      InternalJob `json:"job"`
	Enabled  bool        `json:"enabled"`
}

InternalSchedule wrap a schedule

func (*InternalSchedule) IsValid

func (schedule *InternalSchedule) IsValid() (bool, error)

IsValid checks if an internal schedule definition is valid and has no missing mandatory fields

func (*InternalSchedule) UnmarshalJSON

func (schedule *InternalSchedule) UnmarshalJSON(data []byte) error

UnmarshalJSON unmarshals a json object as a InternalSchedule

type InternalScheduler

type InternalScheduler struct {
	C    *cron.Cron
	Jobs map[int64]cron.EntryID

	RuleEngine chan string
	// contains filtered or unexported fields
}

InternalScheduler represents an instance of a scheduler used for fact processing

func NewScheduler

func NewScheduler() *InternalScheduler

NewScheduler returns a pointer to a new instance of InternalScheduler

func S

func S() *InternalScheduler

S is used to access the global fact scheduler singleton

func (*InternalScheduler) AddJobSchedule

func (s *InternalScheduler) AddJobSchedule(schedule InternalSchedule) error

AddJobSchedule add a new schedule to the current scheduler

func (*InternalScheduler) AddRunningJob

func (s *InternalScheduler) AddRunningJob(scheduleID int64)

AddRunningJob add a job ID to the running job list

func (*InternalScheduler) ExistingRunningJob

func (s *InternalScheduler) ExistingRunningJob(scheduleID int64) bool

ExistingRunningJob check if a job is already running

func (*InternalScheduler) Init

func (s *InternalScheduler) Init() error

Init loads the job schedules from Data Base

func (*InternalScheduler) RemoveJobSchedule

func (s *InternalScheduler) RemoveJobSchedule(scheduleID int64)

RemoveJobSchedule add a new job to the current scheduler

func (*InternalScheduler) RemoveRunningJob

func (s *InternalScheduler) RemoveRunningJob(scheduleID int64)

RemoveRunningJob remove a job ID to the running job list

type PostgresRepository

type PostgresRepository struct {
	// contains filtered or unexported fields
}

PostgresRepository is a repository containing the rules based on a PSQL database and implementing the repository interface

func (*PostgresRepository) Create

func (r *PostgresRepository) Create(schedule InternalSchedule) (int64, error)

Create creates a new schedule in the repository

func (*PostgresRepository) Delete

func (r *PostgresRepository) Delete(id int64) error

Delete deletes an entry from the repository by it's ID

func (*PostgresRepository) Get

Get search and returns a job schedule from the repository by its id

func (*PostgresRepository) GetAll

func (r *PostgresRepository) GetAll() (map[int64]InternalSchedule, error)

GetAll returns all job schedules in the repository

func (*PostgresRepository) Update

func (r *PostgresRepository) Update(schedule InternalSchedule) error

Update updates a schedule in the repository by its name

type PurgeHistoryJob added in v5.2.3

type PurgeHistoryJob struct {
	SituationID         int64             `json:"situationId"`
	SituationInstanceID int64             `json:"situationInstanceId"`
	ParameterFilters    map[string]string `json:"parameterFilters"`
	DeleteBeforeTs      string            `json:"deleteBeforeTs"`
	ScheduleID          int64             `json:"-"`
}

PurgeHistoryJob represent a scheduler job instance which process a group of Purge history, and persist the result in postgresql

func (PurgeHistoryJob) IsValid added in v5.2.3

func (job PurgeHistoryJob) IsValid() (bool, error)

IsValid checks if an internal schedule job definition is valid and has no missing mandatory fields

func (PurgeHistoryJob) Run added in v5.2.3

func (job PurgeHistoryJob) Run()

Run contains all the business logic of the job

type Repository

type Repository interface {
	Create(schedule InternalSchedule) (int64, error)
	Get(id int64) (InternalSchedule, bool, error)
	Update(schedule InternalSchedule) error
	Delete(id int64) error
	GetAll() (map[int64]InternalSchedule, error)
}

Repository is a storage interface which can be implemented by multiple backend (in-memory map, sql database, in-memory cache, file system, ...) It allows standard CRUD operation on situations

func NewPostgresRepository

func NewPostgresRepository(dbClient *sqlx.DB) Repository

NewPostgresRepository returns a new instance of PostgresRulesRepository

func R

func R() Repository

R is used to access the global repository singleton

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL