atarax

README

Atarax

Atarax is a lightweight adaptive scheduler for periodic tasks in Go. It provides an elegant solution for managing recurring jobs with intelligent adaptation to system load with minimal memory footprint and CPU usage.

Install

To install the package, run:

go get github.com/pelageech/atarax

Quick start

The package provides an interface Scheduler for every schedulers, it's usage is recommended in your code.

Package workerpool contains an implementation for this interface. A simple usage:

package main

import (
	"context"
	"fmt"
	"log"
	"os/signal"
	"syscall"
	"time"

	"github.com/pelageech/atarax"
	"github.com/pelageech/atarax/workerpool"
)

func main() {
	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer cancel()

	var sched atarax.Scheduler
	sched = workerpool.NewScheduler(nil)
	go sched.Schedule(ctx)

	now := time.Now()
	jobsComplete := 0
	sum := time.Duration(0)

	// runnable implements atarax.Runnable
	runnable := atarax.RunnableFunc(func(_ context.Context) error {
		t := time.Now()
		sub := t.Sub(now)
		sum += sub
		now = t

		jobsComplete++

		fmt.Println(sub)
		return nil
	})

	timeout, interval := time.Second/2, time.Second
	job := atarax.NewJob(runnable, timeout, interval)

	if err := sched.Add(job); err != nil {
		log.Fatal(err)
	}

	<-ctx.Done()
	fmt.Printf("Jobs complete: %d, duration sum: %v\n", jobsComplete, sum)
}

Documentation

For more information, see the documentation.

License

Atarax is licensed under the MIT License. See the LICENSE file for details.

Documentation

Index

• Constants
• Variables
• func InstrumentMetrics()
• func Meter() metric.Meter
• func SetService(name string)
• func SetSystem(system string)
• func SetVersion(version string)
• func SystemAttributes() []attribute.KeyValue
• func Version() string
• func WithSystemAttributes() metric.MeasurementOption
• type Job
  • func NewJob(task Runnable, timeout, interval time.Duration, opts ...JobOpt) *Job
  • func (j *Job) Clone() *Job
  • func (j *Job) ID() JobID
  • func (j *Job) Interval() time.Duration
  • func (j *Job) Run(ctx context.Context, opts ...JobRunOpt) error
  • func (j *Job) Task() Runnable
  • func (j *Job) Timeout() time.Duration
• type JobID
• type JobOpt
• type JobRunOpt
• type Runnable
• type RunnableFunc
  • func (f RunnableFunc) Run(ctx context.Context) error
• type Scheduler

Constants

const (
	OTelScopeName = "github.com/pelageech/atarax"
	MeterPrefix   = "atarax."
)

const (
	Major = 0
	Minor = 1
	Patch = 0
)

Variables

var (
	ErrJobExists   = errors.New("job already exists")
	ErrJobNotFound = errors.New("job not found")
)

var ErrJobTimeout = errors.New("job timeout")

Functions

func InstrumentMetrics

func InstrumentMetrics()

func Meter

func Meter() metric.Meter

Meter returns an instrumented meter with the scope OTelScopeName.

func SetService

func SetService(name string)

SetService configures atarax.service attribute for atarax.info.

func SetSystem

func SetSystem(system string)

SetSystem configures atarax.system attribute for atarax.info.

func SetVersion

func SetVersion(version string)

SetVersion configures atarax.version attribute for atarax.info.

func SystemAttributes

func SystemAttributes() []attribute.KeyValue

func Version

func Version() string

func WithSystemAttributes

func WithSystemAttributes() metric.MeasurementOption

Types

type Job

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

func NewJob

func NewJob(task Runnable, timeout, interval time.Duration, opts ...JobOpt) *Job

NewJob creates a new instance of Job. If timeout is greater than interval, timeout == interval is used.

func (*Job) Clone

func (j *Job) Clone() *Job

func (*Job) ID

func (j *Job) ID() JobID

func (*Job) Interval

func (j *Job) Interval() time.Duration

func (*Job) Run

func (j *Job) Run(ctx context.Context, opts ...JobRunOpt) error

func (*Job) Task

func (j *Job) Task() Runnable

func (*Job) Timeout

func (j *Job) Timeout() time.Duration

type JobID

type JobID int64

type JobOpt

type JobOpt func(*Job)

type JobRunOpt

type JobRunOpt interface {
	// contains filtered or unexported methods
}

type Runnable

type Runnable interface {
	Run(ctx context.Context) error
}

type RunnableFunc

type RunnableFunc func(ctx context.Context) error

func (RunnableFunc) Run

func (f RunnableFunc) Run(ctx context.Context) error

type Scheduler

type Scheduler interface {
	// Jobs returns an iterator to all the jobs that the scheduler processes currently.
	Jobs() iter.Seq2[JobID, *Job]

	// Schedule starts scheduling tasks containing in the pool. It should be run synchronously.
	//
	// The function should return an error if the scheduling can't be completed anymore.
	// Context cancellation should provide graceful shutdown for all the running jobs and then
	// return an error from the context using [context.Cause].
	//
	// Task errors don't influence Scheduler error.
	Schedule(context.Context) error

	// Add adds a new job into the pool to schedule. If job's ID collides with
	// another job containing in the pool, the function should return ErrJobExists.
	Add(*Job) error

	// Remove finds the job with a given ID and pulls it from the scheduling pool.
	// If there's no the job with such ID, ErrJobNotFound is returned.
	Remove(JobID) error
}

Scheduler is the interface for schedule periodically executing jobs. It contains the necessary minimum providing the Schedule function and an adaptivity.