magos/ptaf_processor.go

501 lines
21 KiB
Go
Executable file
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package main
import (
"database/sql"
"fmt"
"log"
"net"
"os"
"path/filepath"
"strings"
"time"
)
// processPTAFClient обрабатывает клиента для PTAF vendor
// Вся логика из processClient() перенесена сюда без изменений
func processPTAFClient(
db *sql.DB,
config Config,
hostname, instanceName string,
clientInfo ClientInfo,
appsSettingsList []AppsSettings,
portAllocator *PortAllocator,
) bool {
log.Printf("Количество контейнеров: %d", clientInfo.ContainersCount)
if clientInfo.PTAFConfig.Valid {
log.Printf("PTAF Config: установлен (скрыт для безопасности)")
} else {
log.Printf("PTAF Config: не установлен")
}
if clientInfo.FluentBitPort.Valid {
log.Printf("Fluent Bit Port: %d", clientInfo.FluentBitPort.Int64)
}
// Проверяем принадлежность хоста к instance клиента
belongsToInstance := false
if clientInfo.WAFInstance != "" {
var err error
belongsToInstance, err = checkHostBelongsToInstance(db, hostname, clientInfo.WAFInstance)
if err != nil {
log.Printf("⚠ Ошибка проверки принадлежности хоста для %s: %v", clientInfo.ClientTitle, err)
return false
}
}
// Если хост НЕ принадлежит instance — инициируем миграцию
if clientInfo.WAFInstance != "" && !belongsToInstance {
handleClientMigration(clientInfo, hostname)
return false
}
// Хост принадлежит instance — обрабатываем
if clientInfo.WAFInstance == "" {
log.Printf(" Клиент %s не привязан к конкретному instance, обрабатываю", clientInfo.ClientTitle)
} else {
log.Printf("✓ Клиент %s принадлежит instance '%s', обрабатываю", clientInfo.ClientTitle, clientInfo.WAFInstance)
}
// Если контейнеров 0 — помечаем все существующие для drain и выходим
if clientInfo.ContainersCount == 0 {
log.Printf("⚠ Количество контейнеров равно 0 для %s — помечаю существующие контейнеры для drain", clientInfo.ClientTitle)
processDrainingContainers(clientInfo.ClientTitle, 0)
angieConfigPath := fmt.Sprintf("/etc/angie/http.d/angie-ptaf-%s.conf", clientInfo.ClientTitle)
if _, err := os.Stat(angieConfigPath); err == nil {
log.Printf(" → Помечаю конфиг Angie для удаления (контейнеров 0)")
if err := markAngieConfigForDrain(clientInfo.ClientTitle, false); err != nil {
log.Printf(" ⚠ Ошибка пометки конфига Angie: %v", err)
} else {
log.Printf(" ✓ Конфиг Angie помечен для удаления")
}
}
return false
}
// Автоочистка устаревших маркеров drain
cleanStaleDrainMarkers(clientInfo, hostname)
// Получение данных ресурсов
resourcesData := fetchPTAFResourcesData(db, config, appsSettingsList)
if len(resourcesData) == 0 {
log.Printf("⚠ Нет ресурсов с данными для клиента %s, пропускаем", clientInfo.ClientTitle)
return false
}
// Если one_container = false — разбиваем SID по блокам и обрабатываем каждый блок отдельно
if !clientInfo.OneContainer && clientInfo.SidBlock.Valid && clientInfo.SidBlock.String != "" {
// Переход true → false: дренируем обычные контейнеры
cleanupOneContainerContainers(clientInfo.ClientTitle, hostname)
return processPTAFClientBlocks(db, config, hostname, instanceName, clientInfo, resourcesData, portAllocator)
}
// Переход false → true: дренируем блочные контейнеры
cleanupBlockContainers(clientInfo.ClientTitle, hostname)
// Выделение портов
resourcePortMap, err := resolvePortsForClient(clientInfo, resourcesData, portAllocator)
if err != nil {
log.Printf("Ошибка выделения портов для клиента %s: %v", clientInfo.ClientTitle, err)
return false
}
// ════════════════════════════════════════════════════════════
// КРИТИЧЕСКОЕ ИСПРАВЛЕНИЕ: Изменен порядок операций
// БЫЛО: generateAngieConfigs (с reload) → setupContainers
// СТАЛО: generateAngieConfigs (БЕЗ reload) → setupContainers → проверка портов → reload Angie
// ════════════════════════════════════════════════════════════
// 1. Генерация конфигурации Angie БЕЗ автоматического reload
angieChanged, err := generateAngieConfigsWithoutReload(config, clientInfo.ClientTitle, clientInfo, resourcesData, resourcePortMap, clientInfo.Debug)
if err != nil {
log.Printf("Ошибка генерации конфигов Angie для %s: %v", clientInfo.ClientTitle, err)
return false
}
// 2. Создание директорий и конфигов для контейнеров (пересоздание если нужно)
err = setupContainers(config, clientInfo.ClientTitle, &clientInfo, resourcesData, resourcePortMap, hostname)
if err != nil {
log.Printf("Ошибка настройки контейнеров для %s: %v", clientInfo.ClientTitle, err)
return false
}
// 3. Reload Angie только ПОСЛЕ того как все контейнеры готовы
if angieChanged {
log.Printf(" → Конфиг Angie изменился, проверяем готовность контейнеров перед reload")
// Проверяем что все порты контейнеров слушают
allPortsReady := true
totalPorts := 0
readyPorts := 0
for _, res := range resourcesData {
containerPorts, ok := resourcePortMap[res.L7ResourceID]
if !ok {
continue
}
for _, ports := range containerPorts {
// Проверяем HTTPS порты
for customPort, dockerPort := range ports.HTTPSPorts {
totalPorts++
if checkPortListening("127.0.0.1", dockerPort) {
readyPorts++
} else {
log.Printf(" ⚠ Порт %d (custom %d) для l7ResourceID %d не слушает!",
dockerPort, customPort, res.L7ResourceID)
allPortsReady = false
}
}
// Проверяем HTTP порты
for customPort, dockerPort := range ports.HTTPPorts {
totalPorts++
if checkPortListening("127.0.0.1", dockerPort) {
readyPorts++
} else {
log.Printf(" ⚠ Порт %d (custom %d) для l7ResourceID %d не слушает!",
dockerPort, customPort, res.L7ResourceID)
allPortsReady = false
}
}
}
}
log.Printf(" → Портов готово: %d/%d", readyPorts, totalPorts)
if !allPortsReady {
log.Printf("⚠ Не все порты готовы, пропускаем reload Angie для безопасности")
log.Printf(" → Angie продолжит работать со старой конфигурацией")
log.Printf(" → При следующем запуске программы будет повторена попытка")
} else {
log.Printf(" ✓ Все порты слушают, безопасно перезагрузить Angie")
reloadAngie() // Функция из containers_part2.go (не возвращает error)
}
} else {
log.Printf(" Конфиг Angie не изменился, reload не требуется")
}
// ════════════════════════════════════════════════════════════
log.Printf("Клиент %s успешно обработан", clientInfo.ClientTitle)
return true
}
// fetchPTAFResourcesData получает данные для каждого ресурса из кеша или API
// Проверяет что waf_vendor = 'ptaf'
func fetchPTAFResourcesData(db *sql.DB, config Config, appsSettingsList []AppsSettings) []ResourceData {
var resourcesData []ResourceData
for _, appsSettings := range appsSettingsList {
if !appsSettings.WAFEnabled.Valid || !appsSettings.WAFEnabled.Bool {
log.Printf("Пропускаем l7ResourceID %d: waf_enabled = false", appsSettings.L7ResourceID)
continue
}
// Проверяем waf_vendor - обрабатываем только ресурсы PTAF
if appsSettings.WAFVendor.Valid && strings.ToLower(strings.TrimSpace(appsSettings.WAFVendor.String)) != "ptaf" {
log.Printf("Пропускаем l7ResourceID %d: waf_vendor = '%s' (требуется 'ptaf')", appsSettings.L7ResourceID, appsSettings.WAFVendor.String)
continue
}
if !appsSettings.WAFVendor.Valid || appsSettings.WAFVendor.String == "" {
log.Printf("Пропускаем l7ResourceID %d: waf_vendor не указан (требуется 'ptaf')", appsSettings.L7ResourceID)
continue
}
mode := "auto"
if appsSettings.Mode.Valid {
mode = strings.ToLower(strings.TrimSpace(appsSettings.Mode.String))
}
if mode != "auto" && mode != "manual" {
log.Printf("Пропускаем l7ResourceID %d: mode = '%s' (требуется 'auto' или 'manual')", appsSettings.L7ResourceID, mode)
continue
}
if mode == "manual" {
log.Printf("l7ResourceID %d: mode = 'manual', waf_vendor = 'ptaf' ✓", appsSettings.L7ResourceID)
} else if appsSettings.Mode.Valid {
log.Printf("l7ResourceID %d: mode = 'auto', waf_vendor = 'ptaf' ✓", appsSettings.L7ResourceID)
} else {
log.Printf("l7ResourceID %d: mode = NULL (обрабатываем как 'auto'), waf_vendor = 'ptaf' ✓", appsSettings.L7ResourceID)
}
log.Printf("Получение данных для l7ResourceID: %d", appsSettings.L7ResourceID)
var serverName string
var aliases []string
var origins []OriginItem
var err error
if mode == "manual" {
serverName, aliases, origins, err = getResourceDataFromManualInfo(db, config, appsSettings.L7ResourceID)
if err != nil {
log.Printf("Ошибка получения данных из manual_info для l7ResourceID %d: %v", appsSettings.L7ResourceID, err)
continue
}
log.Printf(" ✓ Данные получены из manual_info (waf_vendor: ptaf)")
} else {
var found bool
serverName, aliases, origins, found, err = getResourceDataFromCache(db, config, appsSettings.L7ResourceID)
if found && err == nil {
log.Printf(" Server Name: %s", serverName)
log.Printf(" Aliases: %v", aliases)
log.Printf(" Origins: %d шт.", len(origins))
} else {
if err != nil {
log.Printf(" ⚠ Ошибка чтения кеша: %v, запрашиваю API", err)
} else {
log.Printf(" Данных нет в кеше, запрашиваю API")
}
serverName, err = getServerName(config, appsSettings.L7ResourceID)
if err != nil {
log.Printf("Ошибка получения server_name для l7ResourceID %d: %v", appsSettings.L7ResourceID, err)
continue
}
log.Printf(" Server Name: %s", serverName)
aliases, err = getAliases(config, appsSettings.L7ResourceID)
if err != nil {
log.Printf("Ошибка получения aliases для l7ResourceID %d: %v", appsSettings.L7ResourceID, err)
continue
}
log.Printf(" Aliases: %v", aliases)
origins, err = getOrigins(config, appsSettings.L7ResourceID)
if err != nil {
log.Printf("Ошибка получения origins для l7ResourceID %d: %v", appsSettings.L7ResourceID, err)
continue
}
log.Printf(" Origins: %d шт.", len(origins))
// Сохранение в кеш (опционально, можно закомментировать если функция отсутствует)
// if err := saveResourceDataToCache(db, appsSettings.L7ResourceID, serverName, aliases, origins); err != nil {
// log.Printf(" ⚠ Ошибка сохранения в кеш: %v", err)
// }
}
}
resourcesData = append(resourcesData, ResourceData{
L7ResourceID: appsSettings.L7ResourceID,
ServerName: serverName,
Aliases: aliases,
Origins: origins,
AppsSettings: appsSettings,
})
}
return resourcesData
}
// checkPortListening проверяет что порт слушает на указанном адресе
// Функция уже есть в main.go, но для совместимости оставим здесь тоже
// (Go позволяет дублирование функций в одном package если они приватные)
func checkPortListeningPTAF(host string, port int) bool {
conn, err := net.DialTimeout("tcp", fmt.Sprintf("%s:%d", host, port), 2*time.Second)
if err != nil {
return false
}
conn.Close()
return true
}
// processPTAFClientBlocks обрабатывает клиента с разбивкой SID по блокам контейнеров
// Используется когда one_container = false
func processPTAFClientBlocks(
db *sql.DB,
config Config,
hostname, instanceName string,
clientInfo ClientInfo,
resourcesData []ResourceData,
portAllocator *PortAllocator,
) bool {
blocks, err := parseSidBlocks(clientInfo.SidBlock.String)
if err != nil {
log.Printf("⚠ Ошибка парсинга sid_block для %s: %v", clientInfo.ClientTitle, err)
return false
}
if len(blocks) == 0 {
log.Printf("⚠ sid_block пустой для %s", clientInfo.ClientTitle)
return false
}
log.Printf(" → Режим блоков: %d блоков SID", len(blocks))
// Строим map SID → ResourceData для быстрого поиска
sidToResource := make(map[int]ResourceData)
for _, res := range resourcesData {
sidToResource[res.L7ResourceID] = res
}
success := true
activeBlocks := make(map[string]bool)
for blockIdx, sidList := range blocks {
blockNum := blockIdx + 1
blockName := fmt.Sprintf("block%d", blockNum)
activeBlocks[blockName] = true
log.Printf(" → Обработка блока %d: SID %v", blockNum, sidList)
// Собираем ресурсы для этого блока
var blockResources []ResourceData
for _, sid := range sidList {
if res, ok := sidToResource[sid]; ok {
blockResources = append(blockResources, res)
} else {
log.Printf(" ⚠ SID %d не найден в данных ресурсов", sid)
}
}
if len(blockResources) == 0 {
log.Printf(" ⚠ Нет данных для блока %d, пропускаем", blockNum)
continue
}
// Создаём clientInfo для этого блока с модифицированным именем
blockClientInfo := clientInfo
blockClientInfo.ClientTitle = fmt.Sprintf("%s_%s", clientInfo.ClientTitle, blockName)
// Выделяем порты для блока
blockPortMap, err := resolvePortsForClient(blockClientInfo, blockResources, portAllocator)
if err != nil {
log.Printf(" ⚠ Ошибка выделения портов для блока %d: %v", blockNum, err)
success = false
continue
}
// Генерируем конфиги Angie для блока
angieChanged, err := generateAngieConfigsWithoutReload(config, blockClientInfo.ClientTitle, blockClientInfo, blockResources, blockPortMap, blockClientInfo.Debug)
if err != nil {
log.Printf(" ⚠ Ошибка генерации конфигов Angie для блока %d: %v", blockNum, err)
success = false
continue
}
// Настраиваем контейнеры для блока
err = setupContainers(config, blockClientInfo.ClientTitle, &blockClientInfo, blockResources, blockPortMap, hostname)
if err != nil {
log.Printf(" ⚠ Ошибка настройки контейнеров для блока %d: %v", blockNum, err)
success = false
continue
}
// Reload Angie если нужно
if angieChanged {
allPortsReady := true
for _, res := range blockResources {
containerPorts, ok := blockPortMap[res.L7ResourceID]
if !ok {
continue
}
for _, ports := range containerPorts {
for _, dockerPort := range ports.HTTPSPorts {
if !checkPortListening("127.0.0.1", dockerPort) {
allPortsReady = false
break
}
}
if !allPortsReady {
break
}
}
if !allPortsReady {
break
}
}
if allPortsReady {
reloadAngie()
} else {
log.Printf(" ⚠ Не все порты готовы для блока %d, пропускаем reload Angie", blockNum)
}
}
log.Printf(" ✓ Блок %d обработан успешно", blockNum)
}
// Очистка лишних блоков — drain контейнеров и удаление Angie конфигов
// которые больше не входят в sid_block
cleanupStaleBlocks(clientInfo.ClientTitle, activeBlocks, hostname)
return success
}
// cleanupStaleBlocks дренирует контейнеры и удаляет Angie конфиги
// для блоков которые больше не входят в sid_block
func cleanupStaleBlocks(clientTitle string, activeBlocks map[string]bool, hostname string) {
// Ищем существующие блочные Angie конфиги
pattern := fmt.Sprintf("/etc/angie/http.d/angie-ptaf-%s_block*.conf", clientTitle)
matches, _ := filepath.Glob(pattern)
for _, confFile := range matches {
// Извлекаем blockName из имени файла: angie-ptaf-{clientTitle}_{blockName}.conf
base := filepath.Base(confFile)
base = strings.TrimSuffix(base, ".conf")
base = strings.TrimPrefix(base, fmt.Sprintf("angie-ptaf-%s_", clientTitle))
blockName := base // например "block1"
if activeBlocks[blockName] {
continue
}
// Блок устарел — дренируем контейнеры
log.Printf(" → Удаление устаревшего блока %s для %s", blockName, clientTitle)
blockClientTitle := fmt.Sprintf("%s_%s", clientTitle, blockName)
// Помечаем контейнеры на drain
actualCount := getActualContainerCount(blockClientTitle)
for containerNum := 1; containerNum <= actualCount; containerNum++ {
containerFullName := buildContainerFullName(blockClientTitle, containerNum, hostname)
exists, _ := checkContainerStatus(containerFullName)
if exists {
if err := markContainerForDrain(containerFullName, false); err != nil {
log.Printf(" ⚠ Ошибка пометки контейнера %s: %v", containerFullName, err)
} else {
log.Printf(" ✓ Контейнер %s помечен для drain", containerFullName)
}
}
}
// Помечаем Angie конфиг для drain
if err := markAngieConfigForDrain(blockClientTitle, false); err != nil {
log.Printf(" ⚠ Ошибка пометки Angie конфига для %s: %v", blockClientTitle, err)
} else {
log.Printf(" ✓ Angie конфиг для блока %s помечен для drain", blockName)
}
}
}
// cleanupBlockContainers дренирует все блочные контейнеры клиента
// Используется при переходе one_container: false → true
func cleanupBlockContainers(clientTitle string, hostname string) {
pattern := fmt.Sprintf("/etc/angie/http.d/angie-ptaf-%s_block*.conf", clientTitle)
matches, _ := filepath.Glob(pattern)
if len(matches) == 0 {
return
}
log.Printf(" → Обнаружены блочные конфиги при one_container=true, инициируем drain")
cleanupStaleBlocks(clientTitle, map[string]bool{}, hostname)
}
// cleanupOneContainerContainers дренирует обычный (не блочный) контейнер клиента
// Используется при переходе one_container: true → false
func cleanupOneContainerContainers(clientTitle string, hostname string) {
actualCount := getActualContainerCount(clientTitle)
hasNonBlock := false
for containerNum := 1; containerNum <= actualCount; containerNum++ {
containerFullName := buildContainerFullName(clientTitle, containerNum, hostname)
exists, _ := checkContainerStatus(containerFullName)
if exists {
hasNonBlock = true
if err := markContainerForDrain(containerFullName, false); err != nil {
log.Printf(" ⚠ Ошибка пометки контейнера %s: %v", containerFullName, err)
} else {
log.Printf(" ✓ Контейнер %s помечен для drain (переход в блочный режим)", containerFullName)
}
}
}
if hasNonBlock {
if err := markAngieConfigForDrain(clientTitle, false); err != nil {
log.Printf(" ⚠ Ошибка пометки Angie конфига: %v", err)
}
}
}