refactor(ui): 1 websocket for the whole pipeline (#5893)

Author: sguiheux

cli/cdsctl/workflow_log.go

@@ -431,14 +431,19 @@ func workflowLogStreamRun(v cli.Values) error {
 	})
 
 	buf, err := json.Marshal(sdk.CDNStreamFilter{
-		ItemType: link.ItemType,
-		APIRef:   link.APIRef,
+		JobRunID: log.jobID,
 	})
 	if err != nil {
 		return cli.WrapError(err, "unable to marshal streamFilter")
 	}
 	chanMessageToSend <- buf
 
+	type logBlock struct {
+		Number     int64  `json:"number"`
+		Value      string `json:"value"`
+		ApiRefHash string `json:"api_ref_hash"`
+	}
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -449,6 +454,13 @@ func workflowLogStreamRun(v cli.Values) error {
 			}
 			fmt.Printf("Error: %s\n", err)
 		case m := <-chanMsgReceived:
+			var line logBlock
+			if err := json.Unmarshal(m, &line); err != nil {
+				return err
+			}
+			if line.ApiRefHash != link.APIRef {
+				continue
+			}
 			fmt.Printf("%s", string(m))
 		}
 	}

engine/cdn/cdn_item.go

@@ -291,7 +291,7 @@ func (s *Service) getItemLogValue(ctx context.Context, t sdk.CDNItemType, apiRef
 	}
 
 	log.Debug(ctx, "getItemLogValue> Getting logs from cache")
-	return it, int64(linesCount), s.LogCache.NewReader(it.ID, opts.format, opts.from, opts.size, opts.sort), filename, nil
+	return it, int64(linesCount), s.LogCache.NewReader(*it, opts.format, opts.from, opts.size, opts.sort), filename, nil
 }
 
 func (s *Service) pushItemLogIntoCache(ctx context.Context, it sdk.CDNItem, unitName string) error {

engine/cdn/cdn_log_store.go

@@ -84,7 +84,7 @@ func (s *Service) storeLogs(ctx context.Context, itemType sdk.CDNItemType, signa
 
 	// Send an event in WS broker to refresh streams on current item
 	s.GoRoutines.Exec(ctx, "storeLogsPublishWSEvent", func(ctx context.Context) {
-		s.publishWSEvent(*it)
+		s.publishWSEvent(*iu)
 	})
 
 	// If we have all lines or buffer is full and we received the last line

engine/cdn/item/dao.go

@@ -182,6 +182,19 @@ func LoadWorkerCacheItemsByProjectAndCacheTag(ctx context.Context, m *gorpmapper
 	return getItems(ctx, m, db, query)
 }
 
+// LoadByJobRunID load an item by his job id and type
+func LoadByJobRunID(ctx context.Context, m *gorpmapper.Mapper, db gorp.SqlExecutor, jobRunId int64, itemTypes []string, opts ...gorpmapper.GetOptionFunc) ([]sdk.CDNItem, error) {
+	query := gorpmapper.NewQuery(`
+		SELECT *
+		FROM item
+		WHERE api_ref->>'node_run_job_id' = $1
+		AND type = ANY($2)
+		AND to_delete = false
+		ORDER BY created DESC
+	`).Args(strconv.FormatInt(jobRunId, 10), pq.StringArray(itemTypes))
+	return getItems(ctx, m, db, query, opts...)
+}
+
 // LoadByAPIRefHashAndType load an item by his job id, step order and type
 func LoadByAPIRefHashAndType(ctx context.Context, m *gorpmapper.Mapper, db gorp.SqlExecutor, hash string, itemType sdk.CDNItemType, opts ...gorpmapper.GetOptionFunc) (*sdk.CDNItem, error) {
 	query := gorpmapper.NewQuery(`

engine/cdn/item_logs_handler.go

@@ -12,14 +12,20 @@ import (
 	"github.com/gorilla/mux"
 	"github.com/rockbears/log"
 
-	"github.com/ovh/cds/engine/cdn/item"
 	"github.com/ovh/cds/engine/cdn/redis"
 	"github.com/ovh/cds/engine/cdn/storage"
 	"github.com/ovh/cds/engine/service"
 	"github.com/ovh/cds/engine/websocket"
 	"github.com/ovh/cds/sdk"
 )
 
+type WSLine struct {
+	Number     int64  `json:"number"`
+	Value      string `json:"value"`
+	Since      int64  `json:"since,omitempty"`
+	ApiRefHash string `json:"api_ref_hash"`
+}
+
 func (s *Service) getItemLogsStreamHandler() service.Handler {
 	return func(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
 		c, err := websocket.Upgrader.Upgrade(w, r, nil)
@@ -29,8 +35,8 @@ func (s *Service) getItemLogsStreamHandler() service.Handler {
 		}
 		defer c.Close() //nolint
 
-		jwt := ctx.Value(service.ContextJWT).(*jwt.Token)
-		claims := jwt.Claims.(*sdk.AuthSessionJWTClaims)
+		jwtToken := ctx.Value(service.ContextJWT).(*jwt.Token)
+		claims := jwtToken.Claims.(*sdk.AuthSessionJWTClaims)
 		sessionID := claims.StandardClaims.Id
 
 		wsClient := websocket.NewClient(c)
@@ -39,7 +45,23 @@ func (s *Service) getItemLogsStreamHandler() service.Handler {
 		defer s.WSServer.RemoveClient(wsClient.UUID())
 
 		wsClient.OnMessage(func(m []byte) {
-			if err := wsClientData.UpdateFilter(m); err != nil {
+			var filter sdk.CDNStreamFilter
+			if err := sdk.JSONUnmarshal(m, &filter); err != nil {
+				ctx = sdk.ContextWithStacktrace(ctx, err)
+				log.Warn(ctx, err.Error())
+				return
+			}
+
+			// Load last running step
+			var iuID string
+			if filter.JobRunID > 0 {
+				iu, _ := storage.LoadLastItemUnitByJobUnitType(ctx, s.Mapper, s.mustDBWithCtx(ctx), s.Units.LogsBuffer().ID(), filter.JobRunID, sdk.CDNTypeItemStepLog)
+				if iu != nil {
+					iuID = iu.ID
+				}
+			}
+
+			if err := wsClientData.UpdateFilter(filter, iuID); err != nil {
 				ctx = sdk.ContextWithStacktrace(ctx, err)
 				log.Warn(ctx, err.Error())
 				return
@@ -92,38 +114,45 @@ func (s *Service) sendLogsToWSClient(ctx context.Context, wsClient websocket.Cli
 		return nil
 	}
 
-	if wsClientData.itemUnit == nil {
-		it, err := item.LoadByAPIRefHashAndType(ctx, s.Mapper, s.mustDBWithCtx(ctx), wsClientData.itemFilter.APIRef, wsClientData.itemFilter.ItemType)
-		if err != nil {
-			// Catch not found error as the item can be created after the client stream subscription
-			if sdk.ErrorIs(err, sdk.ErrNotFound) {
-				log.Debug(ctx, "sendLogsToWSClient> can't found item with type %s and ref %s for client %s: %+v", wsClientData.itemFilter.ItemType, wsClientData.itemFilter.APIRef, wsClient.UUID(), err)
-				return nil
+	if wsClientData.itemUnitsData == nil {
+		wsClientData.itemUnitsData = make(map[string]ItemUnitClientData)
+	}
+
+	for k := range wsClientData.itemUnitsData {
+		if wsClientData.itemUnitsData[k].itemUnit == nil {
+			iu, err := storage.LoadItemUnitByID(ctx, s.Mapper, s.mustDBWithCtx(ctx), k)
+			if err != nil {
+				return err
 			}
-			return nil
-		}
 
-		if err := s.itemAccessCheck(ctx, *it); err != nil {
-			var projectKey, workflow string
-			logRef, has := it.GetCDNLogApiRef()
-			if has {
-				projectKey = logRef.ProjectKey
-				workflow = logRef.WorkflowName
+			if err := s.itemAccessCheck(ctx, *iu.Item); err != nil {
+				var projectKey, workflow string
+				logRef, has := iu.Item.GetCDNLogApiRef()
+				if has {
+					projectKey = logRef.ProjectKey
+					workflow = logRef.WorkflowName
+				}
+				return sdk.WrapError(err, "client %s can't access logs for workflow %s/%s", wsClient.UUID(), projectKey, workflow)
+			}
+			wsClientData.itemUnitsData[k] = ItemUnitClientData{
+				itemUnit:            iu,
+				scoreNextLineToSend: wsClientData.itemUnitsData[k].scoreNextLineToSend,
 			}
-			return sdk.WrapError(err, "client %s can't access logs for workflow %s/%s", wsClient.UUID(), projectKey, workflow)
 		}
 
-		iu, err := storage.LoadItemUnitByUnit(ctx, s.Mapper, s.mustDBWithCtx(ctx), s.Units.LogsBuffer().ID(), it.ID)
-		if err != nil {
+		if err := s.sendStepLog(ctx, wsClient, wsClientData, k); err != nil {
 			return err
 		}
 
-		wsClientData.itemUnit = iu
 	}
+	return nil
+}
 
-	log.Debug(ctx, "getItemLogsStreamHandler> send log to client %s from %d", wsClient.UUID(), wsClientData.scoreNextLineToSend)
+func (s *Service) sendStepLog(ctx context.Context, wsClient websocket.Client, wsClientData *websocketClientData, mapIndex string) error {
+	data := wsClientData.itemUnitsData[mapIndex]
 
-	rc, err := s.Units.LogsBuffer().NewAdvancedReader(ctx, *wsClientData.itemUnit, sdk.CDNReaderFormatJSON, wsClientData.scoreNextLineToSend, 100, 0)
+	log.Debug(ctx, "getItemLogsStreamHandler> send log to client %s from %d", wsClient.UUID(), data.scoreNextLineToSend)
+	rc, err := s.Units.LogsBuffer().NewAdvancedReader(ctx, *data.itemUnit, sdk.CDNReaderFormatJSON, data.scoreNextLineToSend, 100, 0)
 	if err != nil {
 		return err
 	}
@@ -138,26 +167,31 @@ func (s *Service) sendLogsToWSClient(ctx context.Context, wsClient websocket.Cli
 	}
 
 	log.Debug(ctx, "getItemLogsStreamHandler> iterate over %d lines to send for client %s", len(lines), wsClient.UUID())
-	oldNextLineToSend := wsClientData.scoreNextLineToSend
+	oldNextLineToSend := data.scoreNextLineToSend
 	for i := range lines {
-		if wsClientData.scoreNextLineToSend > 0 && wsClientData.scoreNextLineToSend != lines[i].Number {
+		if data.scoreNextLineToSend > 0 && data.scoreNextLineToSend != lines[i].Number {
 			break
 		}
-		if err := wsClient.Send(lines[i]); err != nil {
+
+		if err := wsClient.Send(WSLine{
+			Number:     lines[i].Number,
+			Value:      lines[i].Value,
+			Since:      lines[i].Since,
+			ApiRefHash: data.itemUnit.Item.APIRefHash,
+		}); err != nil {
 			return err
 		}
-		if wsClientData.scoreNextLineToSend < 0 {
-			wsClientData.scoreNextLineToSend = lines[i].Number + 1
+		if data.scoreNextLineToSend < 0 {
+			data.scoreNextLineToSend = lines[i].Number + 1
 		} else {
-			wsClientData.scoreNextLineToSend++
+			data.scoreNextLineToSend++
 		}
 	}
-
+	wsClientData.itemUnitsData[mapIndex] = data
 	// If all the lines were sent, we can trigger another update, if only one line was send do not trigger an update wait for next event from broker
-	if len(lines) > 1 && (oldNextLineToSend > 0 || int(wsClientData.scoreNextLineToSend-oldNextLineToSend) == len(lines)) {
+	if len(lines) > 1 && (oldNextLineToSend > 0 || int(data.scoreNextLineToSend-oldNextLineToSend) == len(lines)) {
 		wsClientData.TriggerUpdate()
 	}
-
 	return nil
 }
 

engine/cdn/item_logs_handler_test.go

@@ -292,10 +292,15 @@ func TestGetItemLogsStreamHandler(t *testing.T) {
 	require.NoError(t, s.initWebsocket())
 	ts := httptest.NewServer(s.Router.Mux)
 
+	_, err := db.Exec("DELETE FROM storage_unit_item")
+	require.NoError(t, err)
+	_, err = db.Exec("DELETE FROM ITEM")
+	require.NoError(t, err)
+
 	s.Client = cdsclient.New(cdsclient.Config{Host: "http://lolcat.api", InsecureSkipVerifyTLS: false})
 	gock.InterceptClient(s.Client.(cdsclient.Raw).HTTPClient())
 	t.Cleanup(gock.Off)
-	gock.New("http://lolcat.api").Get("/project/" + projectKey + "/workflows/1/type/step-log/access").Reply(http.StatusOK).JSON(nil)
+	gock.New("http://lolcat.api").Get("/project/" + projectKey + "/workflows/1/type/step-log/access").Times(1).Reply(http.StatusOK).JSON(nil)
 
 	ctx, cancel := context.WithCancel(context.TODO())
 	t.Cleanup(cancel)
@@ -309,7 +314,7 @@ func TestGetItemLogsStreamHandler(t *testing.T) {
 		NodeRunID:    1,
 		NodeRunName:  "MyPipeline",
 		JobName:      "MyJob",
-		JobID:        1,
+		JobID:        123456789,
 		Worker: &cdn.SignatureWorker{
 			StepName:  "script1",
 			StepOrder: 1,
@@ -332,22 +337,6 @@ func TestGetItemLogsStreamHandler(t *testing.T) {
 	jwtTokenRaw, err := signer.SignJWT(jwtToken)
 	require.NoError(t, err)
 
-	apiRef := sdk.CDNLogAPIRef{
-		ProjectKey:     signature.ProjectKey,
-		WorkflowName:   signature.WorkflowName,
-		WorkflowID:     signature.WorkflowID,
-		RunID:          signature.RunID,
-		NodeRunName:    signature.NodeRunName,
-		NodeRunID:      signature.NodeRunID,
-		NodeRunJobName: signature.JobName,
-		NodeRunJobID:   signature.JobID,
-		StepName:       signature.Worker.StepName,
-		StepOrder:      signature.Worker.StepOrder,
-	}
-	apiRefHashU, err := hashstructure.Hash(apiRef, nil)
-	require.NoError(t, err)
-	apiRefHash := strconv.FormatUint(apiRefHashU, 10)
-
 	var messageCounter int64
 	sendMessage := func() {
 		hm := handledMessage{
@@ -385,9 +374,7 @@ func TestGetItemLogsStreamHandler(t *testing.T) {
 		chanErrorReceived <- client.RequestWebsocket(ctx, sdk.NewGoRoutines(ctx), uri, chanMsgToSend, chanMsgReceived, chanErrorReceived)
 	}()
 	buf, err := json.Marshal(sdk.CDNStreamFilter{
-		ItemType: sdk.CDNTypeItemStepLog,
-		APIRef:   apiRefHash,
-		Offset:   0,
+		JobRunID: signature.JobID,
 	})
 	require.NoError(t, err)
 	chanMsgToSend <- buf
@@ -443,15 +430,13 @@ func TestGetItemLogsStreamHandler(t *testing.T) {
 		chanErrorReceived <- client.RequestWebsocket(ctx, sdk.NewGoRoutines(ctx), uri, chanMsgToSend, chanMsgReceived, chanErrorReceived)
 	}()
 	buf, err = json.Marshal(sdk.CDNStreamFilter{
-		ItemType: sdk.CDNTypeItemStepLog,
-		APIRef:   apiRefHash,
-		Offset:   15,
+		JobRunID: signature.JobID,
 	})
 	require.NoError(t, err)
 	chanMsgToSend <- buf
 
 	lines = make([]redis.Line, 0)
-	for ctx.Err() == nil && len(lines) < 5 {
+	for ctx.Err() == nil && len(lines) < 10 {
 		select {
 		case <-ctx.Done():
 			break
@@ -465,9 +450,9 @@ func TestGetItemLogsStreamHandler(t *testing.T) {
 		}
 	}
 
-	require.Len(t, lines, 5)
-	require.Equal(t, "message 15\n", lines[0].Value)
-	require.Equal(t, int64(15), lines[0].Number)
-	require.Equal(t, "message 19\n", lines[4].Value)
-	require.Equal(t, int64(19), lines[4].Number)
+	require.Len(t, lines, 10)
+	require.Equal(t, "message 10\n", lines[0].Value)
+	require.Equal(t, int64(10), lines[0].Number)
+	require.Equal(t, "message 19\n", lines[9].Value)
+	require.Equal(t, int64(19), lines[9].Number)
 }

engine/cdn/lru/redis.go

@@ -132,15 +132,16 @@ func (r *Redis) NewWriter(itemID string) io.WriteCloser {
 }
 
 // NewReader instanciates a new reader
-func (r *Redis) NewReader(itemID string, format sdk.CDNReaderFormat, from int64, size uint, sort int64) io.ReadCloser {
+func (r *Redis) NewReader(item sdk.CDNItem, format sdk.CDNReaderFormat, from int64, size uint, sort int64) io.ReadCloser {
 	return &redis.Reader{
-		Store:     r.store,
-		ItemID:    itemID,
-		PrefixKey: redisLruItemCacheKey,
-		Size:      size,
-		From:      from,
-		Format:    format,
-		Sort:      sort,
+		Store:      r.store,
+		ItemID:     item.ID,
+		ApiRefHash: item.APIRefHash,
+		PrefixKey:  redisLruItemCacheKey,
+		Size:       size,
+		From:       from,
+		Format:     format,
+		Sort:       sort,
 	}
 }
 

engine/cdn/lru/redis_test.go

@@ -66,13 +66,13 @@ func TestRedisLRU(t *testing.T) {
 	require.Equal(t, int64(45), size)
 
 	// Get first item
-	reader1 := r.NewReader(item1.ID, sdk.CDNReaderFormatText, 0, 2, 0)
+	reader1 := r.NewReader(item1, sdk.CDNReaderFormatText, 0, 2, 0)
 	buf1 := new(strings.Builder)
 	_, err = io.Copy(buf1, reader1)
 	require.NoError(t, reader1.Close())
 	require.NoError(t, err)
 	require.Equal(t, "this is the first line\nthis is the second line\n", buf1.String())
-	reader2 := r.NewReader(item1.ID, sdk.CDNReaderFormatText, 0, 0, 0)
+	reader2 := r.NewReader(item1, sdk.CDNReaderFormatText, 0, 0, 0)
 	buf2 := new(strings.Builder)
 	_, err = io.Copy(buf2, reader2)
 	require.NoError(t, reader2.Close())
@@ -83,15 +83,15 @@ func TestRedisLRU(t *testing.T) {
 	writer2 := r.NewWriter(item2.ID)
 	_, err = io.Copy(writer2, strings.NewReader("this is the first line\nthis is the second line"))
 	require.NoError(t, err)
-	reader3 := r.NewReader(item2.ID, sdk.CDNReaderFormatText, 0, 0, 0)
+	reader3 := r.NewReader(item2, sdk.CDNReaderFormatText, 0, 0, 0)
 	buf3 := new(strings.Builder)
 	_, err = io.Copy(buf3, reader3)
 	require.NoError(t, reader3.Close())
 	require.NoError(t, err)
 	require.Equal(t, "this is the first line\n", buf3.String())
 	// close the writer should add the last line
 	require.NoError(t, writer2.Close())
-	reader4 := r.NewReader(item2.ID, sdk.CDNReaderFormatText, 0, 0, 0)
+	reader4 := r.NewReader(item2, sdk.CDNReaderFormatText, 0, 0, 0)
 	buf4 := new(strings.Builder)
 	_, err = io.Copy(buf4, reader4)
 	require.NoError(t, reader4.Close())