manager

package
v1.0.0-rc1 Latest Latest
Warning

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

 Go to latest Published: Jun 23, 2026 License: MIT Imports: 21 Imported by: 0

Documentation

Overview

Package manager содержит ядро управления кластером: менеджер (Manager), текущую реплику (Me) и удалённую (Peer). Отвечает за хранение состояния кластера, подписки на события и управление репликами.

Index

Constants

View Source 
const (
	ZeroUptime = time.Duration(0)
)

Variables

View Source 
var Component = compogo.Component{
	Name: "cluster.manager",
	Dependencies: compogo.Components{
		&runner.Component,
		&config.Component,
	},
	Init: compogo.StepFunc(func(container compogo.Container) error {

		return container.Provides(
			NewMe,
			func(m *Me) cluster.Me { return m },
			newManager,
			func(m *manager) cluster.Manager { return m },
		)
	}),
	Execute: compogo.StepFunc(func(container compogo.Container) error {

		return container.Invoke(func(manager *manager, m *Me) {
			m.onChangeRole.Subscribe(manager.replicaOnChangeRole)
		})
	}),
}

Component — компонент менеджера кластера для Compogo.

Жизненный цикл компонента:

  1. Init — регистрирует конструкторы Me и Manager в DI-контейнере.
  2. Execute — подписывает Manager на события изменения ролей Me.

Компонент является основой всего кластера — без него не работают ни Election, ни Reaper, ни Sync.

View Source 
var (
	// PeerAlreadyExistsError возникает при попытке добавить реплику,
	// которая уже существует в кластере (по Host).
	//
	// Это защита от дублирования реплик в кластере.
	PeerAlreadyExistsError = errors.New("already exists")
)

Ошибки, возвращаемые менеджером кластера.

Functions

This section is empty.

Types

type Me 

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

Me — представление текущей (локальной) реплики.

Реализует интерфейс cluster.Me и хранит всё состояние текущего узла.

Ключевые особенности:

  • Ресурсы обновляются через OnChangeResource (из пакета resource)
  • Роли меняются через AddRole/RemoveRole (при выборах)
  • Score меняется через SetScore (для динамического веса)
  • Все изменения генерируют события (OnChangeRole, OnChangeScore)

func NewMe 

func NewMe(closer compogo.Closer, compogoApp *compogo.App, config *managerConfig.Replica) *Me

NewMe создаёт новый объект Me.

Инициализация:

  • Устанавливает начальные роли: Slave (Master удаляется, если был в конфиге)
  • Устанавливает Score из конфига
  • Создаёт пустой linker для ресурсов

func (*Me) AddRole 

func (m *Me) AddRole(role domain.Role)

AddRole добавляет одну роль.

func (*Me) AddRoles 

func (m *Me) AddRoles(roles set.Set[domain.Role])

AddRoles добавляет несколько ролей.

func (*Me) GetAvailableRoles 

func (m *Me) GetAvailableRoles() set.Set[domain.Role]

GetAvailableRoles возвращает роли, которые реплика МОЖЕТ выполнять. Это статическая информация из конфигурации.

func (*Me) GetHost 

func (m *Me) GetHost() string

GetHost возвращает хост реплики.

func (*Me) GetId 

func (m *Me) GetId() uuid.UUID

GetId возвращает ID реплики.

func (*Me) GetMeta 

func (m *Me) GetMeta() *domain.Meta

GetMeta возвращает неизменяемые метаданные реплики.

func (*Me) GetName 

func (m *Me) GetName() string

GetName возвращает имя реплики.

func (*Me) GetResource 

func (m *Me) GetResource() *linker.Linker[resource.Type, *resource.Resource]

GetResource возвращает информацию об использовании ресурсов.

func (*Me) GetRoles 

func (m *Me) GetRoles() set.Set[domain.Role]

GetRoles возвращает текущие роли реплики. Это динамическая информация, которая меняется при выборах.

func (*Me) GetScore 

func (m *Me) GetScore() float32

GetScore возвращает вес реплики.

func (*Me) GetState 

func (m *Me) GetState() *domain.State

GetState возвращает текущее состояние реплики.

func (*Me) GetTimeStart 

func (m *Me) GetTimeStart() time.Time

GetTimeStart возвращает время запуска реплики.

Используется для вычисления uptime (время работы).

func (*Me) GetUptime 

func (m *Me) GetUptime() time.Duration

GetUptime возвращает время работы реплики.

func (*Me) IsMe 

func (m *Me) IsMe() bool

IsMe всегда возвращает true для текущей реплики.

func (*Me) OnChangeResource 

func (m *Me) OnChangeResource(_ context.Context, resource *resource.Resource)

OnChangeResource — обработчик изменения ресурсов.

Вызывается из пакета resource при изменении использования CPU/Memory. Обновляет информацию о ресурсах текущей реплики.

func (*Me) OnChangeRole 

func (m *Me) OnChangeRole() emitter.Emitter[cluster.Replica]

OnChangeRole возвращает emitter для событий изменения ролей.

func (*Me) OnChangeScore 

func (m *Me) OnChangeScore() emitter.Emitter[cluster.Me]

OnChangeScore возвращает emitter для событий изменения Score.

func (*Me) RemoveRole 

func (m *Me) RemoveRole(role domain.Role)

RemoveRole удаляет одну роль.

func (*Me) RemoveRoles 

func (m *Me) RemoveRoles(roles set.Set[domain.Role])

RemoveRoles удаляет несколько ролей.

func (*Me) SetScore 

func (m *Me) SetScore(score float32)

SetScore устанавливает новый вес реплики.

При изменении Score генерируется событие OnChangeScore. Score может меняться динамически (например, при изменении нагрузки).

type Peer 

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

Peer — представление удалённой реплики.

Реализует интерфейс cluster.Peer и хранит кэш состояния удалённой реплики.

Ключевые особенности:

  • Статус соединения — может быть Disconnection, Connection, Ready.
  • Метаданные (Meta) обновляются через SetMeta.
  • Состояние (State) обновляется через SetState (при получении из gRPC).
  • gRPC-соединение создаётся лениво при первом вызове GetGrpcConnection.

func NewPeer 

func NewPeer(
	host string,
	me cluster.Me,
	grpcConfig *grpcClient.Config,
	logger compogo.Logger,
	closer compogo.Closer,
) *Peer

NewPeer создаёт новый объект Peer для указанного хоста.

При создании:

  • Устанавливается Host
  • Статус = Disconnection
  • Создаётся пустой linker для ресурсов

func (*Peer) GetAvailableRoles 

func (r *Peer) GetAvailableRoles() set.Set[domain.Role]

func (*Peer) GetGrpcConnection 

func (r *Peer) GetGrpcConnection() (grpc.ClientConnInterface, error)

GetGrpcConnection возвращает gRPC-соединение к реплике.

Соединение создаётся лениво (при первом вызове) и кэшируется. Это позволяет не открывать соединения для всех Peer'ов сразу.

Настройки соединения берутся из grpcConfig (keepalive, таймауты).

func (*Peer) GetHost 

func (r *Peer) GetHost() string

func (*Peer) GetId 

func (r *Peer) GetId() uuid.UUID

func (*Peer) GetName 

func (r *Peer) GetName() string

func (*Peer) GetResource 

func (r *Peer) GetResource() *linker.Linker[resource.Type, *resource.Resource]

func (*Peer) GetRoles 

func (r *Peer) GetRoles() set.Set[domain.Role]

func (*Peer) GetScore 

func (r *Peer) GetScore() float32

func (*Peer) GetStatus 

func (r *Peer) GetStatus() domain.Status

func (*Peer) GetUptime 

func (r *Peer) GetUptime() time.Duration

func (*Peer) IsMe 

func (r *Peer) IsMe() bool

IsMe всегда возвращает false для удалённой реплики.

func (*Peer) OnChangeAvailableRoles 

func (r *Peer) OnChangeAvailableRoles() emitter.Emitter[cluster.Peer]

func (*Peer) OnChangeRole 

func (r *Peer) OnChangeRole() emitter.Emitter[cluster.Replica]

func (*Peer) OnChangeStatus 

func (r *Peer) OnChangeStatus() emitter.Emitter[cluster.Peer]

func (*Peer) SetMeta 

func (r *Peer) SetMeta(meta *domain.Meta)

SetMeta обновляет метаданные реплики.

Вызывается при получении Meta через gRPC. Генерирует событие onChangeAvailableRoles, если изменились доступные роли.

func (*Peer) SetState 

func (r *Peer) SetState(state *domain.State)

SetState обновляет состояние реплики и переводит её в статус Ready.

Вызывается при получении State через gRPC. Это сигнал, что реплика полностью синхронизировалась.

func (*Peer) SetStatus 

func (r *Peer) SetStatus(status domain.Status)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.