153 lines
5.6 KiB
Go
153 lines
5.6 KiB
Go
package lexicomai
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// Manager — одиночный диспетчер состояния процесса: источник конфига
|
|
// (live-reload) и блокировки по машинам. Потокобезопасен: агенты вызывают
|
|
// инструменты конкурентно.
|
|
//
|
|
// SSH-клиенты не кэшируются (каждый вызов — отдельный процесс, см. ssh.go),
|
|
// поэтому неизменяемого состояния, кроме конфига и блокировок, нет.
|
|
type Manager struct {
|
|
loader *Loader
|
|
|
|
// mu защищает locks. RWMutex не нужен, т.к. locks меняется редко и коротко,
|
|
// а сами блокировки захватываются отдельно.
|
|
mu sync.Mutex
|
|
locks map[string]*sync.Mutex
|
|
|
|
http *http.Client
|
|
sessions *installSessions
|
|
}
|
|
|
|
// NewManager создаёт диспетчер поверх загрузчика конфига.
|
|
func NewManager(loader *Loader) *Manager {
|
|
return &Manager{
|
|
loader: loader,
|
|
locks: make(map[string]*sync.Mutex),
|
|
sessions: newInstallSessions(),
|
|
http: &http.Client{
|
|
// Каждый запрос дополнительно ограничивается context (см. probeHTTP).
|
|
Timeout: 10 * time.Second,
|
|
},
|
|
}
|
|
}
|
|
|
|
// Close освобождает ресурсы. Вызывается из main через defer.
|
|
func (m *Manager) Close() {
|
|
m.http.CloseIdleConnections()
|
|
}
|
|
|
|
// resolve возвращает актуальный конфиг (live-reload применяется на лету).
|
|
func (m *Manager) resolve() (*Config, error) {
|
|
return m.loader.Get()
|
|
}
|
|
|
|
// timeout возвращает таймаут одной SSH-операции из актуального конфига.
|
|
func (m *Manager) timeout() time.Duration {
|
|
cfg, err := m.resolve()
|
|
if err != nil {
|
|
return DefaultTimeoutSeconds * time.Second
|
|
}
|
|
return cfg.Timeout()
|
|
}
|
|
|
|
// probeHTTP выполняет read-only HTTP GET к продукту на машине (host:port) и
|
|
// возвращает статус и (обрезанное) тело. Используется для /openapi.json
|
|
// (живость/версия) и /license/check (валидность лицензии).
|
|
func (m *Manager) probeHTTP(ctx context.Context, target *Target, port int, path string) (int, string, error) {
|
|
if port <= 0 {
|
|
return 0, "", fmt.Errorf("port not known for %s", target.Alias)
|
|
}
|
|
url := fmt.Sprintf("http://%s:%d%s", target.SSH.Host, port, path)
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
|
if err != nil {
|
|
return 0, "", err
|
|
}
|
|
req.Header.Set("Accept", "application/json")
|
|
resp, err := m.http.Do(req)
|
|
if err != nil {
|
|
return 0, "", err
|
|
}
|
|
defer resp.Body.Close()
|
|
body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
|
return resp.StatusCode, string(body), nil
|
|
}
|
|
|
|
// lock возвращает эксклюзивную блокировку машины (сериализация мутаций/
|
|
// установок на один таргет). Конкурентные agents не должны гонять установки
|
|
// на одну машину параллельно.
|
|
func (m *Manager) lock(alias string) *sync.Mutex {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
l, ok := m.locks[alias]
|
|
if !ok {
|
|
l = &sync.Mutex{}
|
|
m.locks[alias] = l
|
|
}
|
|
return l
|
|
}
|
|
|
|
// loadConfigFor alias возвращает конфиг и таргет; alias="" → значение по умолчанию.
|
|
func (m *Manager) loadConfigFor(alias string) (*Config, *Target, error) {
|
|
cfg, err := m.resolve()
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
t, err := cfg.Target(alias)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
return cfg, t, nil
|
|
}
|
|
|
|
// CurrentConfig возвращает актуальный конфиг (для проверки политики в tools).
|
|
func (m *Manager) CurrentConfig() (*Config, error) {
|
|
return m.resolve()
|
|
}
|
|
|
|
// Target возвращает машину по alias (пустой alias — «по умолчанию»).
|
|
func (m *Manager) Target(alias string) (*Target, error) {
|
|
_, t, err := m.loadConfigFor(alias)
|
|
return t, err
|
|
}
|
|
|
|
// Session возвращает инстал-сессию по id (для service_install_status).
|
|
func (m *Manager) Session(id string) (*InstallSession, bool) {
|
|
return m.sessions.get(id)
|
|
}
|
|
|
|
// ensureMutationAllowed — доменный гейт мутаций: они возможны только при
|
|
// read_only=false глобально И allow_mutate=true на конкретной машине
|
|
// (fail-closed: любое из условий нарушено — отказ). Независимый слой поверх
|
|
// approval человека (approval держит ядро/agent.yaml).
|
|
func (m *Manager) ensureMutationAllowed(cfg *Config, target *Target) error {
|
|
if cfg.IsReadOnly() {
|
|
return fmt.Errorf("module is read-only (read_only=true) — mutations disabled")
|
|
}
|
|
if !target.AllowMutate {
|
|
return fmt.Errorf("target %q has allow_mutate=false — mutations disabled", target.Alias)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// mutableTarget резолвит машину и проверяет разрешение на мутацию
|
|
// (используется доменными методами записи, чтобы гейт был защищён на месте).
|
|
func (m *Manager) mutableTarget(alias string) (*Config, *Target, error) {
|
|
cfg, t, err := m.loadConfigFor(alias)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
if err := m.ensureMutationAllowed(cfg, t); err != nil {
|
|
return nil, nil, err
|
|
}
|
|
return cfg, t, nil
|
|
}
|