core/oc-datacenter
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add websocket route

Browse Source
This commit is contained in:
mr
2025-06-18 11:18:12 +02:00
parent 81167f7b86
commit 4e1e3f20af
4 changed files with 45 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
infrastructure/prometheus.go
View File

@@ -16,6 +16,7 @@ import (
"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/compute_units"
"github.com/gorilla/websocket"
)
type MetricsSnapshot struct {
@@ -81,14 +82,14 @@ func (p *PrometheusService) queryPrometheus(promURL string, expr string, namespa
return metric
}
func (p *PrometheusService) Call(book *booking.Booking) (*booking.Booking, map[string]models.MetricsSnapshot) {
func (p *PrometheusService) Call(book *booking.Booking, user string, peerID string, groups []string) (*booking.Booking, map[string]models.MetricsSnapshot) {
var wg sync.WaitGroup
metrics := map[string]models.MetricsSnapshot{}
// get all booking... from executions_id == namespace typed datacenter.
cUAccess := oclib.NewRequest(oclib.LibDataEnum(oclib.COMPUTE_UNITS), "", "", []string{}, nil)
cRAccess := oclib.NewRequest(oclib.LibDataEnum(oclib.COMPUTE_RESOURCE), "", "", []string{}, nil)
cUAccess := oclib.NewRequest(oclib.LibDataEnum(oclib.COMPUTE_UNITS), user, peerID, groups, nil)
cRAccess := oclib.NewRequest(oclib.LibDataEnum(oclib.COMPUTE_RESOURCE), user, peerID, groups, nil)
rr := cRAccess.LoadOne(book.ResourceID)
if rr.Err != "" {
@@ -129,23 +130,24 @@ func (p *PrometheusService) Call(book *booking.Booking) (*booking.Booking, map[s
var LockKill = &sync.Mutex{}
func (p *PrometheusService) Stream(bookingID string, end *time.Time, interval time.Duration, flusher *http.Flusher, encoder *json.Encoder) {
f := func(e *time.Time) bool {
if end == nil {
return true
}
return time.Now().Before(*e)
}
// TODO kill procedure
func (p *PrometheusService) Stream(bookingID string, interval time.Duration, user string, peerID string, groups []string, websocket *websocket.Conn) {
max := 100
bookIDS := []string{}
mets := map[string][]models.MetricsSnapshot{}
bAccess := oclib.NewRequest(oclib.LibDataEnum(oclib.BOOKING), "", "", []string{}, nil)
bAccess := oclib.NewRequest(oclib.LibDataEnum(oclib.BOOKING), user, peerID, groups, nil)
book := bAccess.LoadOne(bookingID)
if book.Err != "" {
fmt.Errorf("stop because of empty : %s", book.Err)
}
for f(end) {
f := func(e *booking.Booking) bool {
if e.ExpectedEndDate == nil {
return true
}
return time.Now().Before(*e.ExpectedEndDate)
}
for f(book.Data.(*booking.Booking)) {
if slices.Contains(Kill, book.Data.(*booking.Booking).ExecutionsID) {
newKill := []string{}
for _, k := range Kill {
@@ -160,7 +162,7 @@ func (p *PrometheusService) Stream(bookingID string, end *time.Time, interval ti
}
go func() {
book, metrics := p.Call(book.Data.(*booking.Booking))
book, metrics := p.Call(book.Data.(*booking.Booking), user, peerID, groups)
for k, v := range metrics {
if me, ok := mets[k]; !ok {
mets[k] = []models.MetricsSnapshot{v}
@@ -170,9 +172,8 @@ func (p *PrometheusService) Stream(bookingID string, end *time.Time, interval ti
}
}
bookIDS = append(bookIDS, bookingID)
if flusher != nil {
encoder.Encode(metrics)
(*flusher).Flush()
if websocket != nil {
(*websocket).WriteJSON(metrics)
}
if len(bookIDS) != max {
return