cluster

Documentation

Overview

Package cluster wires a service's lifecycle.

It loads configuration, builds the CLI (the run command plus optional database-migration commands) and drives a user-provided Bootstrap through Init/Start/Close with signal handling and ordered shutdown.

A service implements Bootstrap and starts with:

cluster.New[Config]("my-service", &App{}).Execute()

Example

Example shows the minimal entry point of a service: cluster handles config loading, the CLI, signal handling and ordered shutdown.

package main

import (
	"context"

	"github.com/zloevil/jet/cluster"
)

// exampleConfig is the service's typed configuration.
type exampleConfig struct {
	HTTP struct{ Port string }
}

// exampleApp implements cluster.Bootstrap.
type exampleApp struct{}

func (*exampleApp) Init(ctx context.Context, cfg any) error {
	c := cfg.(*exampleConfig) // cluster passes a *exampleConfig
	_ = c                     // build dependencies from config here
	return nil
}

func (*exampleApp) Start(ctx context.Context) error { return nil } // start servers/consumers
func (*exampleApp) Close(ctx context.Context)       {}             // release resources

// Example shows the minimal entry point of a service: cluster handles config
// loading, the CLI, signal handling and ordered shutdown.
func main() {
	svc := cluster.New[exampleConfig]("my-service", &exampleApp{})
	_ = svc.Execute()
}

Index

• Constants
• Variables
• func GetServiceRootPath(serviceName string) string
• type Bootstrap
• type DistributedKeys
  • func NewDistributedKeys() DistributedKeys
• type ServiceInstance
  • func New[TCfg any](svcCode string, bootstrap Bootstrap) *ServiceInstance[TCfg]
  • func (s *ServiceInstance[TCfg]) Execute() error
  • func (s *ServiceInstance[TCfg]) GetCode() string
  • func (s *ServiceInstance[TCfg]) GetLogger() jet.CLoggerFunc
  • func (s *ServiceInstance[TCfg]) NodeId() string
  • func (s *ServiceInstance[TCfg]) WithClickHouseMigration(getClickConfigFn func(cfg *TCfg) (any, error)) *ServiceInstance[TCfg]
  • func (s *ServiceInstance[TCfg]) WithClickHouseMigrations(getClickConfigFns ...func(cfg *TCfg) (any, error)) *ServiceInstance[TCfg]
  • func (s *ServiceInstance[TCfg]) WithConfigPathEnv(env string) *ServiceInstance[TCfg]
  • func (s *ServiceInstance[TCfg]) WithDbMigration(getDbConfigFn func(cfg *TCfg) (any, error)) *ServiceInstance[TCfg]
  • func (s *ServiceInstance[TCfg]) WithMigrationSourceEnv(env string) *ServiceInstance[TCfg]

Examples

• Package

Constants

const (
	ErrCodeConfigPathNotSpecified      = "SVS-001"
	ErrCodeMigrationSourcePathInvalid  = "SVS-002"
	ErrCodeMigrationSourceParamInvalid = "SVS-003"
	ErrCodeClickConfigInvalid          = "SVS-004"
	ErrCodePgConfigInvalid             = "SVS-005"
	ErrCodeApplyConfigReadDirFailed    = "SVS-006"
	ErrCodeApplyConfigMkdirTempFailed  = "SVS-007"
	ErrCodeApplyConfigReadFileFailed   = "SVS-008"
	ErrCodeApplyConfigWriteFileFailed  = "SVS-009"
)

Variables

var (
	ErrConfigPathNotSpecified = func() error {
		return jet.NewAppErrBuilder(ErrCodeConfigPathNotSpecified, "").Business().Err()
	}
	ErrMigrationSourcePathInvalid = func() error {
		return jet.NewAppErrBuilder(ErrCodeMigrationSourcePathInvalid, "migration source path isn't valid").Business().Err()
	}
	ErrMigrationSourceParamInvalid = func() error {
		return jet.NewAppErrBuilder(ErrCodeMigrationSourceParamInvalid, "command param isn't valid").Business().Err()
	}
	ErrClickConfigInvalid = func() error {
		return jet.NewAppErrBuilder(ErrCodeClickConfigInvalid, "click config isn't valid").Business().Err()
	}
	ErrPgConfigInvalid = func() error {
		return jet.NewAppErrBuilder(ErrCodePgConfigInvalid, "pg config isn't valid").Business().Err()
	}
	ErrApplyConfigReadDirFailed = func(err error) error {
		return jet.NewAppErrBuilder(ErrCodeApplyConfigReadDirFailed, "apply config reading directory failed").Wrap(err).Err()
	}
	ErrApplyConfigMkdirTempFailed = func(err error) error {
		return jet.NewAppErrBuilder(ErrCodeApplyConfigMkdirTempFailed, "apply config making directory failed").Wrap(err).Err()
	}
	ErrApplyConfigReadFileFailed = func(err error) error {
		return jet.NewAppErrBuilder(ErrCodeApplyConfigReadFileFailed, "apply config reading file failed").Wrap(err).Err()
	}
	ErrApplyConfigWriteFileFailed = func(err error) error {
		return jet.NewAppErrBuilder(ErrCodeApplyConfigWriteFileFailed, "apply config writing file failed").Wrap(err).Err()
	}
)

Functions

func GetServiceRootPath

func GetServiceRootPath(serviceName string) string

GetServiceRootPath detects the full path to the service root folder which corresponds a service name by checking currently running file and going up along the folder tree until the target folder reached

Types

type Bootstrap

type Bootstrap interface {
	// Init initializes the service
	Init(ctx context.Context, cfg any) error
	// Start executes all background processes
	Start(ctx context.Context) error
	// Close closes the service
	Close(ctx context.Context)
}

Bootstrap interface needs to be implemented by service instance implementation to handle service's lifecycle

type DistributedKeys

type DistributedKeys interface {
	Set(key string)
	Remove(key string)
	Check(key string) bool
}

DistributedKeys is useful when some processing happens in cluster environment (multiple replicas) some business keys might be processed on particular replicas this helps to trace which replica owns the business key

func NewDistributedKeys

func NewDistributedKeys() DistributedKeys

type ServiceInstance

type ServiceInstance[TCfg any] struct {
	// contains filtered or unexported fields
}

func New

func New[TCfg any](svcCode string, bootstrap Bootstrap) *ServiceInstance[TCfg]

func (*ServiceInstance[TCfg]) Execute

func (s *ServiceInstance[TCfg]) Execute() error

func (*ServiceInstance[TCfg]) GetCode

func (s *ServiceInstance[TCfg]) GetCode() string

func (*ServiceInstance[TCfg]) GetLogger

func (s *ServiceInstance[TCfg]) GetLogger() jet.CLoggerFunc

func (*ServiceInstance[TCfg]) NodeId

func (s *ServiceInstance[TCfg]) NodeId() string

func (*ServiceInstance[TCfg]) WithClickHouseMigration

func (s *ServiceInstance[TCfg]) WithClickHouseMigration(getClickConfigFn func(cfg *TCfg) (any, error)) *ServiceInstance[TCfg]

func (*ServiceInstance[TCfg]) WithClickHouseMigrations

func (s *ServiceInstance[TCfg]) WithClickHouseMigrations(getClickConfigFns ...func(cfg *TCfg) (any, error)) *ServiceInstance[TCfg]

func (*ServiceInstance[TCfg]) WithConfigPathEnv

func (s *ServiceInstance[TCfg]) WithConfigPathEnv(env string) *ServiceInstance[TCfg]

WithConfigPathEnv allows to specify env which is used to obtain a config path. It has priority over command flags

func (*ServiceInstance[TCfg]) WithDbMigration

func (s *ServiceInstance[TCfg]) WithDbMigration(getDbConfigFn func(cfg *TCfg) (any, error)) *ServiceInstance[TCfg]

func (*ServiceInstance[TCfg]) WithMigrationSourceEnv

func (s *ServiceInstance[TCfg]) WithMigrationSourceEnv(env string) *ServiceInstance[TCfg]

WithMigrationSourceEnv allows to specify env which is used to obtain a migration source folder path. It has priority over command flags