Skip to content

Commit 4428fa3

Browse files
richardltrichardlt
authored
feat(ui,cdn): display logs from cdn using stream (ovh#5518)
1 parent 15764d1 commit 4428fa3

55 files changed

Lines changed: 1696 additions & 580 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎cli/cdsctl/workflow_log.go‎

Lines changed: 22 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@ import (
55
"encoding/json"
66
"fmt"
77
"io/ioutil"
8-
"net/http"
98
"regexp"
109
"strings"
1110
"time"
@@ -181,6 +180,9 @@ func workflowLogProcess(wr *sdk.WorkflowRun) []workflowLogDetail {
181180
jobNames := map[string]int64{}
182181
for _, runJob := range stage.RunJobs {
183182
jobName := slug.Convert(runJob.Job.Job.Action.Name)
183+
if runJob.Job.Job.Action.StepName != "" {
184+
jobName = slug.Convert(runJob.Job.Job.Action.StepName)
185+
}
184186
countUsageJobName, ok := jobNames[jobName]
185187
if !ok {
186188
jobNames[jobName] = 1
@@ -312,7 +314,7 @@ func workflowLogDownloadRun(v cli.Values) error {
312314

313315
var data []byte
314316
if link != nil {
315-
data, _, _, err = client.Request(context.Background(), http.MethodGet, link.CDNURL+link.DownloadPath, nil)
317+
data, err = client.WorkflowLogDownload(context.Background(), *link)
316318
if err != nil {
317319
return err
318320
}
@@ -405,10 +407,14 @@ func workflowLogStreamRun(v cli.Values) error {
405407

406408
mJob := make(map[string][]workflowLogDetail)
407409
for i := range logs {
408-
if _, ok := mJob[logs[i].jobName]; !ok {
409-
mJob[logs[i].jobName] = nil
410+
key := logs[i].jobName
411+
if logs[i].countUsageJobName > 0 {
412+
key = fmt.Sprintf("%s-%d", key, logs[i].countUsageJobName)
413+
}
414+
if _, ok := mJob[key]; !ok {
415+
mJob[key] = nil
410416
}
411-
mJob[logs[i].jobName] = append(mJob[logs[i].jobName], logs[i])
417+
mJob[key] = append(mJob[key], logs[i])
412418
}
413419
jobNames := make([]string, 0, len(mJob))
414420
for k := range mJob {
@@ -435,24 +441,31 @@ func workflowLogStreamRun(v cli.Values) error {
435441
if err != nil {
436442
return err
437443
}
438-
if link.StreamPath == "" {
439-
return sdk.WithStack(fmt.Errorf("can't stream log for given job"))
440-
}
441444

442445
ctx := context.Background()
446+
chanMessageToSend := make(chan json.RawMessage)
443447
chanMsgReceived := make(chan json.RawMessage)
444448
chanErrorReceived := make(chan error)
445449

446450
goRoutines := sdk.NewGoRoutines()
447451
goRoutines.Exec(ctx, "WebsocketEventsListenCmd", func(ctx context.Context) {
448452
for ctx.Err() == nil {
449-
if err := client.RequestWebsocket(ctx, goRoutines, link.CDNURL+link.StreamPath, nil, chanMsgReceived, chanErrorReceived); err != nil {
453+
if err := client.RequestWebsocket(ctx, goRoutines, fmt.Sprintf("%s/item/stream", link.CDNURL), chanMessageToSend, chanMsgReceived, chanErrorReceived); err != nil {
450454
fmt.Printf("Error: %s\n", err)
451455
}
452456
time.Sleep(1 * time.Second)
453457
}
454458
})
455459

460+
buf, err := json.Marshal(sdk.CDNStreamFilter{
461+
ItemType: link.ItemType,
462+
APIRef: link.APIRef,
463+
})
464+
if err != nil {
465+
return sdk.WithStack(err)
466+
}
467+
chanMessageToSend <- buf
468+
456469
for {
457470
select {
458471
case <-ctx.Done():

‎cli/cdsctl/workflow_run_interactive.go‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@ package main
33
import (
44
"context"
55
"fmt"
6-
"net/http"
76
"strings"
87
"time"
98

@@ -63,17 +62,17 @@ func workflowRunInteractive(v cli.Values, w *sdk.WorkflowRun, baseURL string) er
6362
newOutput += fmt.Sprintf("\n")
6463

6564
for _, step := range job.Job.StepStatus {
66-
var access *sdk.CDNLogLink
65+
var link *sdk.CDNLogLink
6766
if feature.Enabled {
68-
access, err = client.WorkflowNodeRunJobStepLink(context.Background(), projectKey, workflowName, wnr.ID, job.ID, int64(step.StepOrder))
67+
link, err = client.WorkflowNodeRunJobStepLink(context.Background(), projectKey, workflowName, wnr.ID, job.ID, int64(step.StepOrder))
6968
if err != nil {
7069
return err
7170
}
7271
}
7372

7473
var data string
75-
if access != nil {
76-
buf, _, _, err := client.Request(context.Background(), http.MethodGet, access.CDNURL+access.DownloadPath, nil)
74+
if link != nil {
75+
buf, err := client.WorkflowLogDownload(context.Background(), *link)
7776
if err != nil {
7877
return err
7978
}

‎engine/api/ascode.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,7 @@ func (api *API) postImportAsCodeHandler() service.Handler {
9898
fields["stack_trace"] = fmt.Sprintf("%+v", err)
9999
}
100100
log.ErrorWithFields(ctx, fields, "%s", err)
101+
return
101102
}
102103

103104
if ope.Status == sdk.OperationStatusError {

‎engine/api/websocket.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -268,6 +268,9 @@ func (a *API) websocketOnMessage(e sdk.Event) {
268268
// Send the event to the client websocket within a goroutine
269269
a.GoRoutines.Exec(context.Background(), "websocket-"+clientID, func(ctx context.Context) {
270270
c := a.WSServer.GetClientData(clientID)
271+
if c == nil {
272+
return
273+
}
271274

272275
found, needCheckPermission := c.filters.HasOneKey(eventKeys...)
273276
if !found {

‎engine/api/websocket_test.go‎

Lines changed: 14 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -324,7 +324,17 @@ func TestWebsocketNoEventLoose(t *testing.T) {
324324
chan2MessageToSend <- []sdk.WebsocketFilter{filterGlobal}
325325

326326
// Waiting websocket to update filter
327-
time.Sleep(time.Second)
327+
for {
328+
clientIDs := api.WSServer.server.ClientIDs()
329+
if len(clientIDs) == 2 {
330+
data1 := api.WSServer.GetClientData(clientIDs[0])
331+
data2 := api.WSServer.GetClientData(clientIDs[1])
332+
if data1 != nil && data2 != nil && len(data1.filters) == 1 && len(data2.filters) == 1 {
333+
break
334+
}
335+
}
336+
time.Sleep(time.Second)
337+
}
328338

329339
// Send events
330340
countEvent := int64(100)
@@ -334,19 +344,11 @@ func TestWebsocketNoEventLoose(t *testing.T) {
334344
}
335345

336346
// Waiting client to receive all events
337-
ctx, cancel = context.WithTimeout(context.TODO(), time.Second*20)
338-
t.Cleanup(cancel)
339-
loop:
340347
for {
341-
select {
342-
case <-ctx.Done():
343-
break loop
344-
default:
345-
if client1EventCount == countEvent && client2EventCount == countEvent {
346-
break loop
347-
}
348-
time.Sleep(time.Second)
348+
if client1EventCount == countEvent && client2EventCount == countEvent {
349+
break
349350
}
351+
time.Sleep(time.Second)
350352
}
351353

352354
// Let 1 second for clients to consume events

‎engine/api/workflow_run_log.go‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@ package api
22

33
import (
44
"context"
5-
"fmt"
65
"net/http"
76
"strconv"
87

@@ -209,6 +208,9 @@ func (api *API) getWorkflowNodeRunJobLogLinkHandler(ctx context.Context, w http.
209208
stepOrder, runJobID, nodeRunID, workflowName, projectKey)
210209
}
211210
apiRef.StepName = runJob.Job.Action.Actions[int64(ss.StepOrder)].Name
211+
if runJob.Job.Action.Actions[int64(ss.StepOrder)].StepName != "" {
212+
apiRef.StepName = runJob.Job.Action.Actions[int64(ss.StepOrder)].StepName
213+
}
212214
apiRef.StepOrder = int64(ss.StepOrder)
213215
}
214216

@@ -219,9 +221,9 @@ func (api *API) getWorkflowNodeRunJobLogLinkHandler(ctx context.Context, w http.
219221
apiRefHash := strconv.FormatUint(apiRefHashU, 10)
220222

221223
return service.WriteJSON(w, sdk.CDNLogLink{
222-
DownloadPath: fmt.Sprintf("/item/%s/%s/download", itemType, apiRefHash),
223-
StreamPath: fmt.Sprintf("/item/%s/%s/stream", itemType, apiRefHash),
224-
CDNURL: httpURL,
224+
CDNURL: httpURL,
225+
ItemType: itemType,
226+
APIRef: apiRefHash,
225227
}, http.StatusOK)
226228
}
227229

‎engine/api/workflow_run_log_test.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -100,5 +100,6 @@ func Test_getWorkflowNodeRunJobLinkHandler(t *testing.T) {
100100
var link sdk.CDNLogLink
101101
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &link))
102102
require.Equal(t, "http://cdn.net:8080", link.CDNURL)
103-
require.NotEmpty(t, link.DownloadPath)
103+
require.Equal(t, sdk.CDNTypeItemStepLog, link.ItemType)
104+
require.NotEmpty(t, link.APIRef)
104105
}

‎engine/cdn/auth.go‎

Lines changed: 23 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,14 @@ func (s *Service) jwtMiddleware(ctx context.Context, w http.ResponseWriter, req
2626
return service.JWTMiddleware(ctx, w, req, rc, v.VerifyJWT)
2727
}
2828

29+
func (s *Service) validJWTMiddleware(ctx context.Context, w http.ResponseWriter, req *http.Request, rc *service.HandlerConfig) (context.Context, error) {
30+
// Check for valid session based on jwt from context
31+
if _, ok := ctx.Value(service.ContextJWT).(*jwt.Token); !ok {
32+
return ctx, sdk.WithStack(sdk.ErrUnauthorized)
33+
}
34+
return ctx, nil
35+
}
36+
2937
func (s *Service) itemAccessMiddleware(ctx context.Context, w http.ResponseWriter, req *http.Request, rc *service.HandlerConfig) (context.Context, error) {
3038
ctx, end := telemetry.Span(ctx, "router.itemAccessMiddleware")
3139
defer end()
@@ -44,13 +52,6 @@ func (s *Service) itemAccessMiddleware(ctx context.Context, w http.ResponseWrite
4452
return ctx, sdk.WithStack(sdk.ErrUnauthorized)
4553
}
4654

47-
return s.itemAccessCheck(ctx, itemType, apiRef)
48-
}
49-
50-
func (s *Service) itemAccessCheck(ctx context.Context, itemType sdk.CDNItemType, apiRef string) (context.Context, error) {
51-
ctx, end := telemetry.Span(ctx, "router.itemAccessCheck")
52-
defer end()
53-
5455
// Check for session based on jwt from context
5556
jwt, ok := ctx.Value(service.ContextJWT).(*jwt.Token)
5657
if !ok {
@@ -59,28 +60,32 @@ func (s *Service) itemAccessCheck(ctx context.Context, itemType sdk.CDNItemType,
5960
claims := jwt.Claims.(*sdk.AuthSessionJWTClaims)
6061
sessionID := claims.StandardClaims.Id
6162

62-
keyWorkflowPermissionForSession := cache.Key(keyPermission, apiRef, sessionID)
63+
item, err := item.LoadByAPIRefHashAndType(ctx, s.Mapper, s.mustDBWithCtx(ctx), apiRef, itemType)
64+
if err != nil {
65+
return ctx, sdk.NewErrorWithStack(err, sdk.ErrNotFound)
66+
}
67+
68+
return ctx, s.itemAccessCheck(ctx, *item, sessionID)
69+
}
70+
71+
func (s *Service) itemAccessCheck(ctx context.Context, item sdk.CDNItem, sessionID string) error {
72+
keyWorkflowPermissionForSession := cache.Key(keyPermission, string(item.Type), item.APIRefHash, sessionID)
6373

6474
exists, err := s.Cache.Exist(keyWorkflowPermissionForSession)
6575
if err != nil {
66-
return ctx, sdk.NewErrorWithStack(sdk.WrapError(err, "unable to check if permission %s exists", keyWorkflowPermissionForSession), sdk.ErrUnauthorized)
76+
return sdk.NewErrorWithStack(sdk.WrapError(err, "unable to check if permission %s exists", keyWorkflowPermissionForSession), sdk.ErrUnauthorized)
6777
}
6878
if exists {
69-
return ctx, nil
70-
}
71-
72-
item, err := item.LoadByAPIRefHashAndType(ctx, s.Mapper, s.mustDBWithCtx(ctx), apiRef, itemType)
73-
if err != nil {
74-
return ctx, sdk.NewErrorWithStack(err, sdk.ErrNotFound)
79+
return nil
7580
}
7681

7782
if err := s.Client.WorkflowLogAccess(ctx, item.APIRef.ProjectKey, item.APIRef.WorkflowName, sessionID); err != nil {
78-
return ctx, sdk.NewErrorWithStack(err, sdk.ErrNotFound)
83+
return sdk.NewErrorWithStack(err, sdk.ErrNotFound)
7984
}
8085

8186
if err := s.Cache.SetWithTTL(keyWorkflowPermissionForSession, true, 3600); err != nil {
82-
return ctx, sdk.NewErrorWithStack(sdk.WrapError(err, "unable to store permission %s", keyWorkflowPermissionForSession), sdk.ErrUnauthorized)
87+
return sdk.NewErrorWithStack(sdk.WrapError(err, "unable to store permission %s", keyWorkflowPermissionForSession), sdk.ErrUnauthorized)
8388
}
8489

85-
return ctx, nil
90+
return nil
8691
}

‎engine/cdn/auth_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -77,13 +77,13 @@ func Test_itemAccessMiddleware(t *testing.T) {
7777
w = httptest.NewRecorder()
7878
ctx, err = s.jwtMiddleware(context.TODO(), w, req, config)
7979
require.NoError(t, err)
80-
ctx, err = s.itemAccessCheck(ctx, sdk.CDNTypeItemStepLog, myItem.APIRefHash)
80+
err = s.itemAccessCheck(ctx, myItem, sessionID)
8181
assert.NoError(t, err, "no error should be returned because a valid jwt was given")
8282

8383
req = assets.NewJWTAuthentifiedRequest(t, jwtTokenRaw, http.MethodGet, "", nil)
8484
w = httptest.NewRecorder()
8585
ctx, err = s.jwtMiddleware(context.TODO(), w, req, config)
8686
require.NoError(t, err)
87-
ctx, err = s.itemAccessCheck(ctx, sdk.CDNTypeItemStepLog, myItem.APIRefHash)
87+
err = s.itemAccessCheck(ctx, myItem, sessionID)
8888
assert.NoError(t, err, "no error should be returned because a valid jwt was given and permission validated from cache")
8989
}

‎engine/cdn/cdn_gc_test.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ func TestCleanSynchronizedItem(t *testing.T) {
3939
Cache: cache,
4040
Mapper: m,
4141
}
42+
s.GoRoutines = sdk.NewGoRoutines()
4243

4344
tmpDir, err := ioutil.TempDir("", t.Name()+"-cdn-1-*")
4445
require.NoError(t, err)
@@ -197,6 +198,7 @@ func TestCleanWaitingItem(t *testing.T) {
197198
Cache: cache,
198199
Mapper: m,
199200
}
201+
s.GoRoutines = sdk.NewGoRoutines()
200202

201203
ctx, cancel := context.WithCancel(context.TODO())
202204
t.Cleanup(cancel)
@@ -245,6 +247,7 @@ func TestPurgeItem(t *testing.T) {
245247
Cache: cache,
246248
Mapper: m,
247249
}
250+
s.GoRoutines = sdk.NewGoRoutines()
248251

249252
var err error
250253
cfg := test.LoadTestingConf(t, sdk.TypeCDN)

0 commit comments

Comments
 (0)