oc-datacenter/infrastructure/monitor/vector.go

package monitor

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"sync"
	"time"

	oclib "cloud.o-forge.io/core/oc-lib"
	"cloud.o-forge.io/core/oc-lib/models/booking"
	"cloud.o-forge.io/core/oc-lib/models/common/models"
	"cloud.o-forge.io/core/oc-lib/models/live"
	"cloud.o-forge.io/core/oc-lib/models/resources"
	"github.com/gorilla/websocket"
)

// --- Structure métrique Vector ---
type VectorMetric struct {
	Name      string            `json:"name"`
	Value     float64           `json:"value"`
	Labels    map[string]string `json:"labels"`
	Timestamp int64             `json:"timestamp"`
}

// --- Service Vector ---
type VectorService struct {
	mu               sync.Mutex
	ExecutionMetrics map[string][]models.MetricsSnapshot // bookingID -> snapshots
	sessions         map[string]context.CancelFunc       // optional: WS clients
}

func NewVectorService() *VectorService {
	return &VectorService{
		ExecutionMetrics: make(map[string][]models.MetricsSnapshot),
		sessions:         make(map[string]context.CancelFunc),
	}
}

// --- Connexion à un flux Vector en WS ---
func (v *VectorService) ListenVector(ctx context.Context, b *booking.Booking, interval time.Duration, ws *websocket.Conn) error {
	max := 100
	count := 0
	mets := map[string][]models.MetricsSnapshot{}
	isActive := func() bool {
		if b.ExpectedEndDate == nil {
			return true
		}
		return time.Now().Before(*b.ExpectedEndDate)
	}

	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for isActive() {
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
		}

		bb, metrics := Call(b,
			func(dc *live.LiveDatacenter, instance *resources.ComputeResourceInstance, metrics map[string]models.MetricsSnapshot,
				wg *sync.WaitGroup, mu *sync.Mutex) {
				c, _, err := websocket.DefaultDialer.Dial(dc.MonitorPath, nil)
				if err != nil {
					return
				}
				defer c.Close()
				_, msg, err := c.ReadMessage()
				if err != nil {
					log.Println("vector ws read error:", err)
					return
				}

				var m models.Metric
				if err := json.Unmarshal(msg, &m); err != nil {
					log.Println("json unmarshal error:", err)
					return
				}
				mu.Lock()
				if mm, ok := metrics[instance.Name]; !ok {
					metrics[instance.Name] = models.MetricsSnapshot{
						From:    instance.Source,
						Metrics: []models.Metric{m},
					}
				} else {
					mm.Metrics = append(mm.Metrics, m)
				}

				mu.Unlock()
			})
		_ = bb
		for k, v := range metrics {
			mets[k] = append(mets[k], v)
		}
		count++

		if ws != nil {
			if err := ws.WriteJSON(metrics); err != nil {
				return fmt.Errorf("websocket write error: %w", err)
			}
		}

		if count < max {
			continue
		}

		if b.ExecutionMetrics == nil {
			b.ExecutionMetrics = mets
		} else {
			for kk, vv := range mets {
				b.ExecutionMetrics[kk] = append(b.ExecutionMetrics[kk], vv...)
			}
		}
		b.GetAccessor(nil).UpdateOne(b, b.GetID())
		mets = map[string][]models.MetricsSnapshot{}
		count = 0
	}

	return nil
}

// --- Permet d'ajouter un front WebSocket pour recevoir les metrics live ---
// --- Permet de récupérer le cache historique pour un booking ---
func (v *VectorService) GetCache(bookingID string) []models.MetricsSnapshot {
	v.mu.Lock()
	defer v.mu.Unlock()
	snapshots := v.ExecutionMetrics[bookingID]
	return snapshots
}

// --- Exemple d'intégration avec un booking ---
func (v *VectorService) Stream(ctx context.Context, bookingID string, interval time.Duration, ws *websocket.Conn) {
	logger := oclib.GetLogger()
	go func() {
		bAccess := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.BOOKING), nil)
		book := bAccess.LoadOne(bookingID)
		if book.Err != "" {
			logger.Err(fmt.Errorf("stop because of empty : %s", book.Err))
			return
		}
		b := book.ToBookings()
		if b == nil {
			logger.Err(fmt.Errorf("stop because of empty is not a booking"))
			return
		}
		if err := v.ListenVector(ctx, b, interval, ws); err != nil {
			log.Printf("Vector listen error for booking %s: %v\n", b.GetID(), err)
		}
	}()
}