feat: status for all instances

engine/api/services.go

@@ -99,16 +99,22 @@ func (api *API) postServiceRegisterHandler() service.Handler {
 		}
 
 		// Update or create the service
+
+		var sessionID string
+		if a := getAuthSession(ctx); a != nil {
+			sessionID = a.ID
+		}
 		if exists {
 			srv.Update(data)
-			if err := services.Update(ctx, tx, srv); err != nil {
+			if err := services.Update(ctx, tx, srv, sessionID); err != nil {
 				return err
 			}
 			log.Debug("postServiceRegisterHandler> update existing service %s(%d) registered for consumer %s", srv.Name, srv.ID, *srv.ConsumerID)
 		} else {
 			srv = &data
 			srv.ConsumerID = &consumer.ID
-			if err := services.Insert(ctx, tx, srv); err != nil {
+
+			if err := services.Insert(ctx, tx, srv, sessionID); err != nil {
 				return sdk.WithStack(err)
 			}
 			log.Debug("postServiceRegisterHandler> insert new service %s(%d) registered for consumer %s", srv.Name, srv.ID, *srv.ConsumerID)
@@ -173,7 +179,11 @@ func (api *API) postServiceHearbeatHandler() service.Handler {
 		s.LastHeartbeat = time.Now()
 		s.MonitoringStatus = mon
 
-		if err := services.Update(ctx, tx, s); err != nil {
+		var sessionID string
+		if a := getAuthSession(ctx); a != nil {
+			sessionID = a.ID
+		}
+		if err := services.Update(ctx, tx, s, sessionID); err != nil {
 			return err
 		}
 
@@ -238,14 +248,18 @@ func (api *API) serviceAPIHeartbeatUpdate(ctx context.Context, db *gorp.DbMap) {
 		return
 	}
 
+	var authSessionID string
+	if a := getAuthSession(ctx); a != nil {
+		authSessionID = a.ID
+	}
 	if exists {
 		srv.ID = old.ID
-		if err := services.Update(ctx, tx, srv); err != nil {
+		if err := services.Update(ctx, tx, srv, authSessionID); err != nil {
 			log.Error(ctx, "serviceAPIHeartbeat> Unable to update service %s: %v", srv.Name, err)
 			return
 		}
 	} else {
-		if err := services.Insert(ctx, tx, srv); err != nil {
+		if err := services.Insert(ctx, tx, srv, authSessionID); err != nil {
 			log.Error(ctx, "serviceAPIHeartbeat> Unable to insert service %s: %v", srv.Name, err)
 			return
 		}

engine/api/services/dao.go

@@ -65,6 +65,32 @@ func LoadAll(ctx context.Context, db gorp.SqlExecutor) ([]sdk.Service, error) {
 	return getAll(ctx, db, query)
 }
 
+// LoadAllWithStatus returns all services in database with status
+func LoadAllWithStatus(ctx context.Context, db gorp.SqlExecutor) ([]sdk.Service, error) {
+	query := gorpmapping.NewQuery(`SELECT * FROM service`)
+	srvs, err := getAll(ctx, db, query)
+	if err != nil {
+		return nil, err
+	}
+
+	ss, err := loadAllServiceStatus(ctx, db)
+	if err != nil {
+		return nil, err
+	}
+	for i := range srvs {
+		srv := &srvs[i]
+		srv.MonitoringStatus = sdk.MonitoringStatus{Now: time.Now()}
+		for _, status := range ss {
+			if srv.ID == status.ServiceID {
+				srv.MonitoringStatus.Lines = append(srv.MonitoringStatus.Lines, status.MonitoringStatus.Lines...)
+			}
+		}
+		srvs[i] = *srv
+	}
+
+	return srvs, nil
+}
+
 // LoadAllByType returns all services with given type.
 func LoadAllByType(ctx context.Context, db gorp.SqlExecutor, typeService string) ([]sdk.Service, error) {
 	if ss, ok := internalCache.getFromCache(typeService); ok {
@@ -74,7 +100,7 @@ func LoadAllByType(ctx context.Context, db gorp.SqlExecutor, typeService string)
 	return getAll(ctx, db, query)
 }
 
-// LoadAllByType returns all services that users can see with given type.
+// LoadAllByTypeAndUserID returns all services that users can see with given type.
 func LoadAllByTypeAndUserID(ctx context.Context, db gorp.SqlExecutor, typeService string, userID string) ([]sdk.Service, error) {
 	query := gorpmapping.NewQuery(`
 		SELECT service.*
@@ -115,22 +141,45 @@ func FindDeadServices(ctx context.Context, db gorp.SqlExecutor, t time.Duration)
 }
 
 // Insert a service in database.
-func Insert(ctx context.Context, db gorpmapper.SqlExecutorWithTx, s *sdk.Service) error {
+func Insert(ctx context.Context, db gorpmapper.SqlExecutorWithTx, s *sdk.Service, authSessionID string) error {
 	sdb := service{Service: *s}
 	if err := gorpmapping.InsertAndSign(ctx, db, &sdb); err != nil {
 		return err
 	}
 	*s = sdb.Service
-	return nil
+	return upsertStatus(db, s, authSessionID)
 }
 
 // Update a service in database.
-func Update(ctx context.Context, db gorpmapper.SqlExecutorWithTx, s *sdk.Service) error {
+func Update(ctx context.Context, db gorpmapper.SqlExecutorWithTx, s *sdk.Service, authSessionID string) error {
 	sdb := service{Service: *s}
 	if err := gorpmapping.UpdateAndSign(ctx, db, &sdb); err != nil {
 		return err
 	}
 	*s = sdb.Service
+	return upsertStatus(db, s, authSessionID)
+}
+
+// loadAllServiceStatus returns all services status
+func loadAllServiceStatus(ctx context.Context, db gorp.SqlExecutor) ([]sdk.ServiceStatus, error) {
+	query := gorpmapping.NewQuery(`SELECT * FROM service_status`)
+	ss := []sdk.ServiceStatus{}
+	if err := gorpmapping.GetAll(ctx, db, query, &ss); err != nil {
+		return nil, sdk.WrapError(err, "cannot get services")
+	}
+	return ss, nil
+}
+
+func upsertStatus(db gorpmapper.SqlExecutorWithTx, s *sdk.Service, authSessionID string) error {
+	var sessionID *string
+	if authSessionID != "" {
+		sessionID = &authSessionID
+	}
+	query := `INSERT INTO service_status(monitoring_status, service_id, auth_session_id) VALUES($1,$2, $3)
+	ON CONFLICT (service_id, auth_session_id) DO UPDATE SET monitoring_status = $1, service_id = $2, auth_session_id = $3`
+	if _, err := db.Exec(query, s.MonitoringStatus, s.ID, sessionID); err != nil {
+		return sdk.WithStack(err)
+	}
 	return nil
 }
 

engine/api/services/dao_test.go

@@ -3,8 +3,10 @@ package services_test
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/ovh/cds/engine/api/services"
 	"github.com/ovh/cds/engine/api/test"
@@ -35,7 +37,7 @@ func TestDAO(t *testing.T) {
 		},
 	}
 
-	test.NoError(t, services.Insert(context.TODO(), db, &srv))
+	test.NoError(t, services.Insert(context.TODO(), db, &srv, ""))
 
 	srv2, err := services.LoadByName(context.TODO(), db, srv.Name)
 	test.NoError(t, err)
@@ -55,3 +57,77 @@ func TestDAO(t *testing.T) {
 	_, err = services.FindDeadServices(context.TODO(), db, 0)
 	test.NoError(t, err)
 }
+
+func TestDAOWithStatus(t *testing.T) {
+	db, _ := test.SetupPG(t)
+
+	allSrv, err := services.LoadAll(context.TODO(), db)
+	for _, s := range allSrv {
+		if err := services.Delete(db, &s); err != nil {
+			t.Fatalf("unable to delete service: %v", err)
+		}
+	}
+
+	privateKey, err := jws.NewRandomRSAKey()
+	test.NoError(t, err)
+	publicKey, err := jws.ExportPublicKey(privateKey)
+	test.NoError(t, err)
+
+	theServiceName := sdk.RandomString(10)
+	var srv = sdk.Service{
+		CanonicalService: sdk.CanonicalService{
+			Name:      theServiceName,
+			Type:      "type-service-test",
+			PublicKey: publicKey,
+		},
+		MonitoringStatus: sdk.MonitoringStatus{
+			Now: time.Now(),
+			Lines: []sdk.MonitoringStatusLine{
+				{
+					Component: "backend/cds-backend/items",
+					Hostname:  "foofoo.local",
+					Service:   "cds-cdn",
+					Status:    "OK",
+					Type:      "cdn",
+					Value:     "90",
+				},
+			},
+		},
+	}
+
+	test.NoError(t, services.Insert(context.TODO(), db, &srv, ""))
+
+	srv2, err := services.LoadByName(context.TODO(), db, srv.Name)
+	test.NoError(t, err)
+
+	assert.Equal(t, srv.Name, srv2.Name)
+	assert.Equal(t, string(srv.PublicKey), string(srv2.PublicKey))
+
+	all, err := services.LoadAllByType(context.TODO(), db, srv.Type)
+	test.NoError(t, err)
+
+	assert.True(t, len(all) >= 1)
+
+	all2, err := services.LoadAllWithStatus(context.TODO(), db)
+	test.NoError(t, err)
+	var found bool
+	for _, s := range all2 {
+		if s.Name == theServiceName {
+			found = true
+			require.EqualValues(t, 1, len(s.MonitoringStatus.Lines))
+			for _, ss := range s.MonitoringStatus.Lines {
+				require.EqualValues(t, "backend/cds-backend/items", ss.Component)
+			}
+			break
+		}
+	}
+
+	require.True(t, found)
+
+	for _, s := range all {
+		test.NoError(t, services.Delete(db, &s))
+	}
+
+	_, err = services.FindDeadServices(context.TODO(), db, 0)
+	test.NoError(t, err)
+}

engine/api/services/external.go

@@ -24,6 +24,7 @@ type ExternalService struct {
 	Path        string `json:"path"`
 }
 
+// ServiceConfig return sthe serviceConfig for the current ExternalService
 func (e ExternalService) ServiceConfig() sdk.ServiceConfig {
 	b, _ := json.Marshal(e)
 	var cfg sdk.ServiceConfig
@@ -64,7 +65,7 @@ func Pings(ctx context.Context, dbFunc func() *gorp.DbMap, ss []ExternalService)
 
 func ping(ctx context.Context, db gorpmapper.SqlExecutorWithTx, s ExternalService) error {
 	// Select for update
-	serv, err := LoadByName(context.Background(), db, s.Name)
+	srv, err := LoadByName(context.Background(), db, s.Name)
 	if err != nil {
 		return sdk.WithStack(err)
 	}
@@ -101,9 +102,9 @@ func ping(ctx context.Context, db gorpmapper.SqlExecutorWithTx, s ExternalServic
 		mon.Lines[0].Value = "Health: OK"
 	}
 
-	serv.LastHeartbeat = time.Now()
-	serv.MonitoringStatus = mon
-	if err := Update(ctx, db, serv); err != nil {
+	srv.LastHeartbeat = time.Now()
+	srv.MonitoringStatus = mon
+	if err := Update(ctx, db, srv, ""); err != nil {
 		log.Warning(ctx, "services.ping> unable to update monitoring status: %v", err)
 		return err
 	}
@@ -144,13 +145,13 @@ func initExternal(ctx context.Context, db *gorp.DbMap, s ExternalService) error
 		s.Service.LastHeartbeat = old.LastHeartbeat
 		s.Service.MonitoringStatus = old.MonitoringStatus
 		s.Service.Config = s.ServiceConfig()
-		if err := Update(ctx, tx, &s.Service); err != nil {
+		if err := Update(ctx, tx, &s.Service, ""); err != nil {
 			return sdk.WrapError(err, "unable to update external service")
 		}
 	} else {
 		s.Service.LastHeartbeat = time.Now()
 		s.Service.Config = s.ServiceConfig()
-		if err := Insert(ctx, tx, &s.Service); err != nil {
+		if err := Insert(ctx, tx, &s.Service, ""); err != nil {
 			return sdk.WrapError(err, "unable to insert external service")
 		}
 	}

engine/api/services/gorp_model.go

@@ -20,5 +20,6 @@ func (s service) Canonical() gorpmapper.CanonicalForms {
 func init() {
 	gorpmapping.Register(
 		gorpmapping.New(service{}, "service", true, "id"),
+		gorpmapping.New(sdk.ServiceStatus{}, "service_status", true, "id"),
 	)
 }

engine/api/status.go

@@ -54,7 +54,7 @@ func (api *API) statusHandler() service.Handler {
 			status = http.StatusServiceUnavailable
 		}
 
-		srvs, err := services.LoadAll(ctx, api.mustDB())
+		srvs, err := services.LoadAllWithStatus(ctx, api.mustDB())
 		if err != nil {
 			return err
 		}
@@ -333,7 +333,7 @@ func (api *API) countMetricRange(ctx context.Context, status string, timerange s
 }
 
 func (api *API) processStatusMetrics(ctx context.Context) {
-	srvs, err := services.LoadAll(ctx, api.mustDB())
+	srvs, err := services.LoadAllWithStatus(ctx, api.mustDB())
 	if err != nil {
 		log.Error(ctx, "Error while getting services list: %v", err)
 		return

engine/api/test/assets/assets.go

@@ -460,7 +460,7 @@ func InsertHatchery(t *testing.T, db gorpmapper.SqlExecutorWithTx, grp sdk.Group
 		},
 	}
 
-	require.NoError(t, services.Insert(context.TODO(), db, &srv))
+	require.NoError(t, services.Insert(context.TODO(), db, &srv, ""))
 
 	session, err := authentication.NewSession(context.TODO(), db, hConsumer, 5*time.Minute, false)
 	require.NoError(t, err)
@@ -497,7 +497,7 @@ func InsertService(t *testing.T, db gorpmapper.SqlExecutorWithTx, name, serviceT
 		},
 	}
 
-	require.NoError(t, services.Insert(context.TODO(), db, &srv))
+	require.NoError(t, services.Insert(context.TODO(), db, &srv, ""))
 
 	return &srv, privateKey
 }
@@ -532,7 +532,7 @@ func InitCDNService(t *testing.T, db gorpmapper.SqlExecutorWithTx, scopes ...sdk
 		},
 	}
 
-	require.NoError(t, services.Insert(context.TODO(), db, &srv))
+	require.NoError(t, services.Insert(context.TODO(), db, &srv, ""))
 
 	return &srv, privateKey
 }

engine/service/service.go

@@ -3,6 +3,7 @@ package service
 import (
 	"context"
 	"fmt"
+	"os"
 	"time"
 
 	"github.com/ovh/cds/sdk"
@@ -15,10 +16,16 @@ import (
 // NewMonitoringStatus returns a MonitoringStatus for the current service
 func (c *Common) NewMonitoringStatus() *sdk.MonitoringStatus {
 	t := time.Now()
+	hostname, err := os.Hostname()
+	if err != nil {
+		log.Error(context.Background(), "NewMonitoringStatus: error on getting hostname")
+	}
+
 	s := &sdk.MonitoringStatus{
-		Now:         t,
-		ServiceType: c.Type(),
-		ServiceName: c.Name(),
+		Now:             t,
+		ServiceType:     c.Type(),
+		ServiceName:     c.Name(),
+		ServiceHostname: hostname,
 	}
 	s.AddLine(c.commonMonitoring(t)...)
 	return s

engine/sql/api/216_service_status.sql

@@ -0,0 +1,16 @@
+-- +migrate Up
+
+CREATE TABLE "service_status" (
+    id BIGSERIAL PRIMARY KEY,
+    auth_session_id VARCHAR(36),
+    service_id BIGSERIAL NOT NULL,
+    monitoring_status JSONB
+);
+
+SELECT create_foreign_key_idx_cascade('FK_SERVICE_STATUS_AUTH_SESSION', 'service_status', 'auth_session', 'auth_session_id', 'id');
+SELECT create_foreign_key_idx_cascade('FK_SERVICE_STATUS_SERVICE', 'service_status', 'service', 'service_id', 'id');
+SELECT create_unique_index('service_status', 'idx_service_status_uniq', 'auth_session_id,service_id');
+
+-- +migrate Down
+
+DROP table IF EXISTS "service_status";