backgroundtasks

README

background_tasks

• Project README
• Release/Build Status
• Changelog
• License
• Contribution Guidelines
• Code of Conduct
• Secure Coding Guidelines

Table of Contents

1. Description
2. Structure and Organisation
3. Class Diagram
4. Functionality
5. Data Types
6. Testing
7. Proposed Functionality/Requirements
8. References

Specification

1. Description

The background_tasks package is an internal package responsible for managing background jobs within DMS. It contains a scheduler that registers tasks and run them according to the schedule defined by the task definition.

proposed Other packages that have their own background tasks register through this package:

1. Registration
   1. The task itself, the arguments it needs
   2. priority
   3. event (time period or other event to trigger task)
2. Start , Stop, Resume
3. Algorithm that accounts for the event and priority of the task (not yet clear)
4. Monitor resource usage of tasks (not yet clear)

2. Structure and Organisation

Here is quick overview of the contents of this pacakge:

• README: Current file which is aimed towards developers who wish to use and modify the package functionality.

• init: This file initializes OpenTelemetry-based Zap logger.

• scheduler: This file This file defines a background task scheduler that manages task execution based on triggers, priority, and retry policies.

• task: This file contains background task structs and their properties.

• trigger: This file defines various trigger types (PeriodicTrigger, EventTrigger, OneTimeTrigger) for background tasks, allowing execution based on time intervals, cron expressions, or external events

Files with *_test.go naming convention contain unit tests of the functionality in corresponding file.

3. Class Diagram

Source

background_tasks class diagram

Rendered from source file

!$rootUrlGitlab = "https://gitlab.com/nunet/device-management-service/-/raw/main"
!$packageRelativePath = "/internal/background_tasks"
!$packageUrlGitlab = $rootUrlGitlab + $packageRelativePath
 
!include $packageUrlGitlab/specs/class_diagram.puml

4. Functionality

NewScheduler

• signature: NewScheduler(maxRunningTasks int) *Scheduler
  

• input: maximum no of running tasks
  

• output: internal.background_tasks.Scheduler

NewScheduler function creates a new scheduler which takes maxRunningTasks argument to limit the maximum number of tasks to run at a time.

Scheduler methods

Scheduler struct is the orchestrator that manages and runs the tasks. If the Scheduler task queue is full, remaining tasks that are triggered will wait until there is a slot available in the scheduler.

It has the following methods:

AddTask

• signature: AddTask(task *Task) *Task
  

• input: internal.background_tasks.Task
  

• output: internal.background_tasks.Task

AddTask registers a task to be run when triggered.

RemoveTask

• signature: RemoveTask(taskID int)
  

• input: identifier of the Task
  

• output: None

RemoveTask removes a task from the scheduler. Tasks with only OneTimeTrigger will be removed automatically once run.

Start

• signature: Start()
  

• input: None
  

• output: None

Start starts the scheduler to monitor tasks.

Stop

• signature: Stop()
  

• input: None
  

• output: None

Stop stops the scheduler.

runTask

• signature: runTask(taskID int)
  

• input: identifier of the Task
  

• output: None

runTask executes a task and manages its lifecycle and retry policy.

runTasks

• signature: runTasks()
  

• input: None
  

• output: None

runTasks checks and runs tasks based on their triggers and priority.

runningTasksCount

• signature: runningTasksCount() int
  

• input: None
  

• output: number of running tasks

runningTasksCount returns the count of running tasks.

Trigger Interface

type Trigger interface {
	IsReady() bool // Returns true if the trigger condition is met.
	Reset()        // Resets the trigger state.
}

Its methods are explained below:

IsReady

• signature: IsReady() bool
  

• input: None
  

• output: bool

IsReady should return true if the task should be run.

Reset

• signature: Reset()
  

• input: None
  

• output: None

Reset resets the trigger until the next event happens.

There are different implementations for the Trigger interface.

• PeriodicTrigger: Defines a trigger based on a duration interval or a cron expression.

• EventTrigger: Defines a trigger that is set by a trigger channel.

• OneTimeTrigger: A trigger that is only triggered once after a set delay.

5. Data Types

• internal.background_tasks.Scheduler

// Scheduler orchestrates the execution of tasks based on their triggers and priority.
type Scheduler struct {
	tasks           map[int]internal.background_tasks.Task // Map of tasks by their ID.
	runningTasks    map[int]bool  // Map to keep track of running tasks.
	ticker          *time.Ticker  // Ticker for periodic checks of task triggers.
	stopChan        chan struct{} // Channel to signal stopping the scheduler.
	maxRunningTasks int           // Maximum number of tasks that can run concurrently.
	lastTaskID      int           // Counter for assigning unique IDs to tasks.
	mu              sync.Mutex    // Mutex to protect access to task maps.
}

• internal.background_tasks.RetryPolicy

// RetryPolicy defines the policy for retrying tasks on failure.
type RetryPolicy struct {
	MaxRetries int           // Maximum number of retries.
	Delay      time.Duration // Delay between retries.
}

• internal.background_tasks.Execution

// Execution records the execution details of a task.
type Execution struct {
	StartedAt time.Time   // Start time of the execution.
	EndedAt   time.Time   // End time of the execution.
	Status    string      // Status of the execution (e.g., "SUCCESS", "FAILED").
	Error     string      // Error message if the execution failed.
	Event     interface{} // Event associated with the execution.
	Results   interface{} // Results of the execution.
}

• internal.background_tasks.Task

Task is a struct that defines a job. It includes the task's ID, Name, the function that is going to be run, the arguments for the function, the triggers that trigger the task to run, retry policy, etc.

// Task represents a schedulable task.
type Task struct {
	ID            int                          // Unique identifier for the task.
	Name          string                       // Name of the task.
	Description   string                       // Description of the task.
	Triggers      []internal.background_tasks.Trigger                    // List of triggers for the task.
	Function      func(args interface{}) error // Function to execute as the task.
	Args          []interface{}                // Arguments for the task function.
	RetryPolicy   internal.background_tasks.RetryPolicy                  // Retry policy for the task.
	Enabled       bool                         // Flag indicating if the task is enabled.
	Priority      int                          // Priority of the task for scheduling.
	ExecutionHist []internal.background_tasks.Execution                  // History of task executions.
}

• internal.background_tasks.PeriodicTrigger

// PeriodicTrigger triggers at regular intervals or based on a cron expression.
type PeriodicTrigger struct {
	Interval      time.Duration // Interval for periodic triggering.
	CronExpr      string        // Cron expression for triggering.
	lastTriggered time.Time     // Last time the trigger was activated.
}

• internal.background_tasks.EventTrigger

// EventTrigger triggers based on an external event signaled through a channel
type EventTrigger struct {
	Trigger chan bool // Channel to signal an event.
}

• internal.background_tasks.OneTimeTrigger

// OneTimeTrigger triggers once after a specified delay.
type OneTimeTrigger struct {
	Delay        time.Duration // The delay after which to trigger.
	registeredAt time.Time     // Time when the trigger was set.
}

6. Testing

Unit tests for each functionality are defined in files with *_test.go naming convention.

7. Proposed Functionality / Requirements

List of issues

All issues that are related to the implementation of internal package can be found below. These include any proposals for modifications to the package or new functionality needed to cover the requirements of other packages.

• internal package implementation TBD

8. References

Documentation

Index

• type EventTrigger
  • func (t *EventTrigger) IsReady() bool
  • func (t *EventTrigger) Reset()
• type Execution
• type OneTimeTrigger
  • func (t *OneTimeTrigger) IsReady() bool
  • func (t *OneTimeTrigger) Reset()
• type PeriodicTrigger
  • func (t *PeriodicTrigger) IsReady() bool
  • func (t *PeriodicTrigger) Reset()
• type PeriodicTriggerWithJitter
  • func (t *PeriodicTriggerWithJitter) IsReady() bool
  • func (t *PeriodicTriggerWithJitter) Reset()
• type RetryPolicy
• type Scheduler
  • func NewScheduler(maxRunningTasks int) *Scheduler
  • func (s *Scheduler) AddTask(task *Task) *Task
  • func (s *Scheduler) RemoveTask(taskID int)
  • func (s *Scheduler) Start()
  • func (s *Scheduler) Stop()
• type Task
• type Trigger

Types

type EventTrigger

type EventTrigger struct {
	Trigger chan bool // Channel to signal an event.
}

EventTrigger triggers based on an external event signaled through a channel.

func (*EventTrigger) IsReady

func (t *EventTrigger) IsReady() bool

IsReady checks if there is a signal in the trigger channel.

func (*EventTrigger) Reset

func (t *EventTrigger) Reset()

Reset for EventTrigger does nothing as its state is managed externally.

type Execution

type Execution struct {
	StartedAt time.Time   // Start time of the execution.
	EndedAt   time.Time   // End time of the execution.
	Status    string      // Status of the execution (e.g., "SUCCESS", "FAILED").
	Error     string      // Error message if the execution failed.
	Event     interface{} // Event associated with the execution.
	Results   interface{} // Results of the execution.
}

Execution records the execution details of a task.

type OneTimeTrigger

type OneTimeTrigger struct {
	Delay time.Duration // The delay after which to trigger.
	// contains filtered or unexported fields
}

OneTimeTrigger triggers once after a specified delay.

func (*OneTimeTrigger) IsReady

func (t *OneTimeTrigger) IsReady() bool

IsReady checks if the current time has passed the delay period.

func (*OneTimeTrigger) Reset

func (t *OneTimeTrigger) Reset()

Reset sets the trigger registration time to the current time.

type PeriodicTrigger

type PeriodicTrigger struct {
	Interval time.Duration // Interval for periodic triggering.
	CronExpr string        // Cron expression for triggering.
	// contains filtered or unexported fields
}

PeriodicTrigger triggers at regular intervals or based on a cron expression.

func (*PeriodicTrigger) IsReady

func (t *PeriodicTrigger) IsReady() bool

IsReady checks if the trigger should activate based on time or cron expression.

func (*PeriodicTrigger) Reset

func (t *PeriodicTrigger) Reset()

Reset updates the last triggered time to the current time.

type PeriodicTriggerWithJitter

type PeriodicTriggerWithJitter struct {
	Interval time.Duration // Interval for periodic triggering.
	CronExpr string        // Cron expression for triggering.

	Jitter func() time.Duration
	// contains filtered or unexported fields
}

PeriodicTrigger triggers at regular intervals or based on a cron expression.

func (*PeriodicTriggerWithJitter) IsReady

func (t *PeriodicTriggerWithJitter) IsReady() bool

IsReady checks if the trigger should activate based on time or cron expression.

func (*PeriodicTriggerWithJitter) Reset

func (t *PeriodicTriggerWithJitter) Reset()

Reset updates the last triggered time to the current time.

type RetryPolicy

type RetryPolicy struct {
	MaxRetries int           // Maximum number of retries.
	Delay      time.Duration // Delay between retries.
}

RetryPolicy defines the policy for retrying tasks on failure.

type Scheduler

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

Scheduler orchestrates the execution of tasks based on their triggers and priority.

func NewScheduler

func NewScheduler(maxRunningTasks int) *Scheduler

NewScheduler creates a new Scheduler with a specified limit on running tasks.

func (*Scheduler) AddTask

func (s *Scheduler) AddTask(task *Task) *Task

AddTask adds a new task to the scheduler and initializes its state.

func (*Scheduler) RemoveTask

func (s *Scheduler) RemoveTask(taskID int)

RemoveTask removes a task from the scheduler.

func (*Scheduler) Start

func (s *Scheduler) Start()

Start begins the scheduler's task execution loop.

func (*Scheduler) Stop

func (s *Scheduler) Stop()

Stop signals the scheduler to stop running tasks.

type Task

type Task struct {
	ID            int                          // Unique identifier for the task.
	Name          string                       // Name of the task.
	Description   string                       // Description of the task.
	Triggers      []Trigger                    // List of triggers for the task.
	Function      func(args interface{}) error // Function to execute as the task.
	Args          []interface{}                // Arguments for the task function.
	RetryPolicy   RetryPolicy                  // Retry policy for the task.
	Enabled       bool                         // Flag indicating if the task is enabled.
	Priority      int                          // Priority of the task for scheduling.
	ExecutionHist []Execution                  // History of task executions.
}

Task represents a schedulable task.

type Trigger

type Trigger interface {
	IsReady() bool // Returns true if the trigger condition is met.
	Reset()        // Resets the trigger state.
}

Trigger interface defines a method to check if a trigger condition is met.