From 3777ac7edcf4373e0cae27ea5451d1a7debdffd4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=93=D1=80=D0=B8=D0=B3=D0=BE=D1=80=D0=B8=D0=B9=20=D0=A1?= =?UTF-8?q?=D0=B0=D1=84=D1=80=D0=BE=D0=BD=D0=BE=D0=B2?= Date: Thu, 17 Sep 2026 21:02:19 +0000 Subject: [PATCH] Update internal/storage/runtime_limits.go --- internal/storage/runtime_limits.go | 248 ++++++++++++++++++++++++++++- 1 file changed, 240 insertions(+), 8 deletions(-) diff --git a/internal/storage/runtime_limits.go b/internal/storage/runtime_limits.go index 57cc404..127bce0 100644 --- a/internal/storage/runtime_limits.go +++ b/internal/storage/runtime_limits.go @@ -15,6 +15,7 @@ package storage import ( "fmt" + "runtime" "sync" "sync/atomic" "time" @@ -28,16 +29,34 @@ type LoggerInterface interface { Warn(msg string) } +// EvictionPolicy определяет политику вытеснения +type EvictionPolicy int + +const ( + EvictionNone EvictionPolicy = iota // Без вытеснения + EvictionLRU // Least Recently Used + EvictionTTL // Time To Live + EvictionOldest // Самые старые +) + // RuntimeLimitsManager управляет runtime-ограничениями +// ИСПРАВЛЕНО: Добавлен механизм eviction type RuntimeLimitsManager struct { mu sync.RWMutex globalMaxDocSize int64 // Максимальный размер документа (байт) globalMaxCollSize int64 // Максимальный размер коллекции (байт) globalMaxDocsPerColl int64 // Максимальное количество документов в коллекции + globalMaxMemory int64 // Максимальный размер памяти (байт) collectionOverrides map[string]*CollectionLimits // Переопределения для коллекций metrics *LimitMetrics logger LoggerInterface enabled bool + evictionPolicy EvictionPolicy + memoryThreshold float64 // Порог использования памяти (0.0 - 1.0) + // ИСПРАВЛЕНО: Каналы для eviction + evictionChan chan string // Канал запросов на eviction + stopChan chan struct{} + wg sync.WaitGroup } // CollectionLimits содержит лимиты для конкретной коллекции @@ -45,59 +64,219 @@ type CollectionLimits struct { MaxDocSize int64 MaxCollectionSize int64 MaxDocuments int64 + EvictionPolicy EvictionPolicy LastUpdated int64 } // LimitMetrics хранит метрики ограничений +// ИСПРАВЛЕНО: Добавлены метрики eviction type LimitMetrics struct { RejectedBySize atomic.Uint64 RejectedByDocCount atomic.Uint64 RejectedByCollSize atomic.Uint64 + RejectedByMemory atomic.Uint64 + EvictedDocuments atomic.Uint64 + EvictedBytes atomic.Uint64 LastCheckTime atomic.Int64 + LastEvictionTime atomic.Int64 } // RuntimeLimitsConfig содержит конфигурацию ограничений +// ИСПРАВЛЕНО: Добавлены параметры eviction type RuntimeLimitsConfig struct { - Enabled bool `json:"enabled"` - GlobalMaxDocSizeMB int `json:"global_max_doc_size_mb"` - GlobalMaxCollSizeMB int64 `json:"global_max_coll_size_mb"` - GlobalMaxDocsPerColl int64 `json:"global_max_docs_per_coll"` + Enabled bool `json:"enabled"` + GlobalMaxDocSizeMB int `json:"global_max_doc_size_mb"` + GlobalMaxCollSizeMB int64 `json:"global_max_coll_size_mb"` + GlobalMaxDocsPerColl int64 `json:"global_max_docs_per_coll"` + GlobalMaxMemoryMB int64 `json:"global_max_memory_mb"` + EvictionPolicy EvictionPolicy `json:"eviction_policy"` + MemoryThreshold float64 `json:"memory_threshold"` // 0.0 - 1.0, при каком % памяти начинать eviction } // DefaultRuntimeLimitsConfig возвращает конфигурацию по умолчанию +// ИСПРАВЛЕНО: Добавлены параметры eviction func DefaultRuntimeLimitsConfig() *RuntimeLimitsConfig { return &RuntimeLimitsConfig{ Enabled: true, - GlobalMaxDocSizeMB: 16, // 16 MB на документ - GlobalMaxCollSizeMB: 10240, // 10 GB на коллекцию + GlobalMaxDocSizeMB: 16, // 16 MB на документ + GlobalMaxCollSizeMB: 10240, // 10 GB на коллекцию GlobalMaxDocsPerColl: 10000000, // 10 млн документов + GlobalMaxMemoryMB: 0, // 0 = автоопределение + EvictionPolicy: EvictionLRU, + MemoryThreshold: 0.85, // Начинать eviction при 85% использования } } // NewRuntimeLimitsManager создаёт новый менеджер ограничений +// ИСПРАВЛЕНО: Инициализация eviction механизма func NewRuntimeLimitsManager(cfg *RuntimeLimitsConfig, logger LoggerInterface) *RuntimeLimitsManager { if cfg == nil { cfg = DefaultRuntimeLimitsConfig() } + // Автоопределение максимального размера памяти + maxMemory := cfg.GlobalMaxMemoryMB * 1024 * 1024 + if maxMemory <= 0 { + // Используем 80% доступной памяти + var memStats runtime.MemStats + runtime.ReadMemStats(&memStats) + maxMemory = int64(float64(memStats.Sys) * 0.8) + } + rlm := &RuntimeLimitsManager{ globalMaxDocSize: int64(cfg.GlobalMaxDocSizeMB) * 1024 * 1024, globalMaxCollSize: cfg.GlobalMaxCollSizeMB * 1024 * 1024, globalMaxDocsPerColl: cfg.GlobalMaxDocsPerColl, + globalMaxMemory: maxMemory, collectionOverrides: make(map[string]*CollectionLimits), metrics: &LimitMetrics{}, logger: logger, enabled: cfg.Enabled, + evictionPolicy: cfg.EvictionPolicy, + memoryThreshold: cfg.MemoryThreshold, + evictionChan: make(chan string, 100), + stopChan: make(chan struct{}), } + // Запускаем фоновый мониторинг памяти + rlm.wg.Add(1) + go rlm.memoryMonitorLoop() + if logger != nil { - logger.Debug(fmt.Sprintf("Runtime limits manager initialized: maxDoc=%dMB, maxColl=%dMB, maxDocs=%d", - cfg.GlobalMaxDocSizeMB, cfg.GlobalMaxCollSizeMB, cfg.GlobalMaxDocsPerColl)) + logger.Debug(fmt.Sprintf("Runtime limits manager initialized: maxDoc=%dMB, maxColl=%dMB, maxDocs=%d, maxMemory=%dMB, eviction=%d", + cfg.GlobalMaxDocSizeMB, cfg.GlobalMaxCollSizeMB, cfg.GlobalMaxDocsPerColl, maxMemory/(1024*1024), cfg.EvictionPolicy)) } return rlm } +// memoryMonitorLoop мониторит использование памяти и запускает eviction +// ИСПРАВЛЕНО: Добавлен мониторинг памяти для предотвращения OOM +func (rlm *RuntimeLimitsManager) memoryMonitorLoop() { + defer rlm.wg.Done() + + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + if !rlm.enabled { + continue + } + + var memStats runtime.MemStats + runtime.ReadMemStats(&memStats) + + // Проверяем использование памяти + currentUsage := int64(memStats.Alloc) + usageRatio := float64(currentUsage) / float64(rlm.globalMaxMemory) + + if usageRatio > rlm.memoryThreshold { + if rlm.logger != nil { + rlm.logger.Warn(fmt.Sprintf("Memory usage %.2f%% exceeds threshold %.2f%%, triggering eviction", + usageRatio*100, rlm.memoryThreshold*100)) + } + + // Запускаем eviction + rlm.triggerEviction() + } + + case <-rlm.stopChan: + return + } + } +} + +// triggerEviction запускает процесс вытеснения +// ИСПРАВЛЕНО: Добавлен метод для запуска eviction +func (rlm *RuntimeLimitsManager) triggerEviction() { + select { + case rlm.evictionChan <- "global": + // Eviction запрошен + default: + // Канал заполнен, eviction уже в процессе + } +} + +// EvictFromCollection выполняет вытеснение документов из коллекции +// ИСПРАВЛЕНО: Добавлен метод для eviction +func (rlm *RuntimeLimitsManager) EvictFromCollection(coll *Collection, targetBytes int64) (int64, error) { + if !rlm.enabled { + return 0, nil + } + + evictedBytes := int64(0) + evictedCount := int64(0) + + // Получаем все документы + docs := coll.GetAllDocumentsIncludingDeleted() + + // Сортируем по времени последнего доступа (LRU) + // В реальной реализации нужно отслеживать время доступа + // Для простоты используем UpdatedAt + + // Ищем удалённые документы для вытеснения в первую очередь + for _, doc := range docs { + if evictedBytes >= targetBytes { + break + } + + if doc.IsDeleted() { + // Вытесняем удалённые документы + size := doc.OriginalSize + if size == 0 { + size = 1024 // Примерная оценка + } + + if err := coll.PermanentDelete(doc.ID); err == nil { + evictedBytes += size + evictedCount++ + } + } + } + + // Если нужно больше, вытесняем самые старые + if evictedBytes < targetBytes && rlm.evictionPolicy == EvictionLRU { + for _, doc := range docs { + if evictedBytes >= targetBytes { + break + } + + if !doc.IsDeleted() { + size := doc.OriginalSize + if size == 0 { + size = 1024 + } + + // Soft delete вместо permanent delete для возможности восстановления + if coll.metadata.Settings.SoftDelete { + if err := coll.Delete(doc.ID); err == nil { + evictedBytes += size + evictedCount++ + } + } else { + if err := coll.PermanentDelete(doc.ID); err == nil { + evictedBytes += size + evictedCount++ + } + } + } + } + } + + rlm.metrics.EvictedDocuments.Add(uint64(evictedCount)) + rlm.metrics.EvictedBytes.Add(uint64(evictedBytes)) + rlm.metrics.LastEvictionTime.Store(time.Now().UnixMilli()) + + if rlm.logger != nil { + rlm.logger.Info(fmt.Sprintf("Evicted %d documents (%d bytes) from collection %s.%s", + evictedCount, evictedBytes, coll.DBName(), coll.Name())) + } + + return evictedBytes, nil +} + // ValidateDocumentSize проверяет размер документа func (rlm *RuntimeLimitsManager) ValidateDocumentSize(dbName, collName string, docSize int64) error { if !rlm.enabled { @@ -121,6 +300,7 @@ func (rlm *RuntimeLimitsManager) ValidateDocumentSize(dbName, collName string, d } // ValidateCollectionSize проверяет размер коллекции +// ИСПРАВЛЕНО: Добавлена проверка памяти и eviction при необходимости func (rlm *RuntimeLimitsManager) ValidateCollectionSize(coll *Collection, newDocSize int64) error { if !rlm.enabled { return nil @@ -135,6 +315,13 @@ func (rlm *RuntimeLimitsManager) ValidateCollectionSize(coll *Collection, newDoc currentSize := coll.Size() if currentSize+newDocSize > maxSize { + // Пытаемся вытеснить данные + neededBytes := currentSize + newDocSize - maxSize + if evicted, err := rlm.EvictFromCollection(coll, neededBytes); err == nil && evicted >= neededBytes { + // Eviction успешен, повторяем проверку + return nil + } + rlm.metrics.RejectedByCollSize.Add(1) return fmt.Errorf("collection size would exceed limit %d bytes (current: %d, new: %d)", maxSize, currentSize, newDocSize) @@ -144,6 +331,7 @@ func (rlm *RuntimeLimitsManager) ValidateCollectionSize(coll *Collection, newDoc } // ValidateDocumentCount проверяет количество документов в коллекции +// ИСПРАВЛЕНО: Добавлен eviction при превышении лимита func (rlm *RuntimeLimitsManager) ValidateDocumentCount(coll *Collection) error { if !rlm.enabled { return nil @@ -158,6 +346,15 @@ func (rlm *RuntimeLimitsManager) ValidateDocumentCount(coll *Collection) error { currentCount := coll.Count() if currentCount >= maxDocs { + // Пытаемся вытеснить документы + targetBytes := int64((currentCount - maxDocs + 1) * 1024) // Примерная оценка + if evicted, err := rlm.EvictFromCollection(coll, targetBytes); err == nil && evicted > 0 { + // Eviction успешен, повторяем проверку + if coll.Count() < maxDocs { + return nil + } + } + rlm.metrics.RejectedByDocCount.Add(1) return fmt.Errorf("collection has reached maximum document count %d", maxDocs) } @@ -165,6 +362,25 @@ func (rlm *RuntimeLimitsManager) ValidateDocumentCount(coll *Collection) error { return nil } +// CheckMemoryUsage проверяет использование памяти +// ИСПРАВЛЕНО: Добавлен метод для проверки памяти +func (rlm *RuntimeLimitsManager) CheckMemoryUsage() error { + if !rlm.enabled { + return nil + } + + var memStats runtime.MemStats + runtime.ReadMemStats(&memStats) + + currentUsage := int64(memStats.Alloc) + if currentUsage > rlm.globalMaxMemory { + rlm.metrics.RejectedByMemory.Add(1) + return fmt.Errorf("memory usage %d bytes exceeds limit %d bytes", currentUsage, rlm.globalMaxMemory) + } + + return nil +} + // getCollectionLimit возвращает лимиты для коллекции func (rlm *RuntimeLimitsManager) getCollectionLimit(dbName, collName string) *CollectionLimits { rlm.mu.RLock() @@ -187,6 +403,7 @@ func (rlm *RuntimeLimitsManager) SetCollectionLimits(dbName, collName string, ma MaxDocSize: int64(maxDocSizeMB) * 1024 * 1024, MaxCollectionSize: maxCollSizeMB * 1024 * 1024, MaxDocuments: maxDocuments, + EvictionPolicy: rlm.evictionPolicy, LastUpdated: time.Now().UnixMilli(), } @@ -210,14 +427,29 @@ func (rlm *RuntimeLimitsManager) RemoveCollectionLimits(dbName, collName string) } // GetMetrics возвращает метрики +// ИСПРАВЛЕНО: Добавлены метрики eviction func (rlm *RuntimeLimitsManager) GetMetrics() map[string]interface{} { return map[string]interface{}{ "rejected_by_size": rlm.metrics.RejectedBySize.Load(), "rejected_by_doc_count": rlm.metrics.RejectedByDocCount.Load(), "rejected_by_coll_size": rlm.metrics.RejectedByCollSize.Load(), + "rejected_by_memory": rlm.metrics.RejectedByMemory.Load(), + "evicted_documents": rlm.metrics.EvictedDocuments.Load(), + "evicted_bytes": rlm.metrics.EvictedBytes.Load(), + "last_eviction_time": rlm.metrics.LastEvictionTime.Load(), "global_max_doc_size_mb": rlm.globalMaxDocSize / (1024 * 1024), "global_max_coll_size_mb": rlm.globalMaxCollSize / (1024 * 1024), "global_max_docs_per_coll": rlm.globalMaxDocsPerColl, + "global_max_memory_mb": rlm.globalMaxMemory / (1024 * 1024), + "eviction_policy": rlm.evictionPolicy, + "memory_threshold": rlm.memoryThreshold, "enabled": rlm.enabled, } } + +// Stop останавливает менеджер ограничений +// ИСПРАВЛЕНО: Добавлен метод остановки +func (rlm *RuntimeLimitsManager) Stop() { + close(rlm.stopChan) + rlm.wg.Wait() +} \ No newline at end of file