NEOS-1624: add more tags to the logs page (#3178)

backend/internal/loki/loki.go

@@ -110,10 +110,24 @@ func GetStreamsFromResponseData(data *QueryResponseData) (Streams, error) {
 	return streams, nil
 }
 
-func GetEntriesFromStreams(streams Streams) []Entry {
-	entries := []Entry{}
+func GetEntriesFromStreams(streams Streams) []*LabeledEntry {
+	entries := []*LabeledEntry{}
 	for _, stream := range streams {
-		entries = append(entries, stream.Entries...)
+		for _, entry := range stream.Entries {
+			entries = append(entries, &LabeledEntry{Entry: entry, Labels: getFilteredLabels(stream.Labels, allowedLabels)})
+		}
 	}
 	return entries
 }
+
+var allowedLabels = []string{"ActivityType", "Name", "Schema", "Table", "Attempt", "metadata_Schema", "metadata_Table"}
+
+func getFilteredLabels(labels LabelSet, keepLabels []string) LabelSet {
+	filteredLabels := LabelSet{}
+	for _, label := range keepLabels {
+		if value, ok := labels[label]; ok {
+			filteredLabels[label] = value
+		}
+	}
+	return filteredLabels
+}

backend/internal/loki/loki_test.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -40,13 +41,18 @@ func Test_GetStreamsFromResponseData(t *testing.T) {
 }
 
 func Test_GetEntriesFromStreams(t *testing.T) {
+	time := time.Now()
 	actual := GetEntriesFromStreams(Streams{
-		{Entries: []Entry{{Line: "foo-line"}}},
-		{Entries: []Entry{{Line: "bar-line"}, {Line: "baz-line"}}},
+		{Entries: []Entry{{Line: "foo-line", Timestamp: time}}, Labels: LabelSet{"Attempt": "1", "willberemoved": "true"}},
+		{Entries: []Entry{{Line: "bar-line", Timestamp: time}, {Line: "baz-line", Timestamp: time}}, Labels: LabelSet{"Attempt": "2"}},
 	})
 	require.Equal(
 		t,
-		[]Entry{{Line: "foo-line"}, {Line: "bar-line"}, {Line: "baz-line"}},
+		[]*LabeledEntry{
+			{Entry: Entry{Line: "foo-line", Timestamp: time}, Labels: LabelSet{"Attempt": "1"}},
+			{Entry: Entry{Line: "bar-line", Timestamp: time}, Labels: LabelSet{"Attempt": "2"}},
+			{Entry: Entry{Line: "baz-line", Timestamp: time}, Labels: LabelSet{"Attempt": "2"}},
+		},
 		actual,
 	)
 }

backend/internal/loki/types.go

@@ -159,6 +159,11 @@ type Entry struct {
 	Line      string
 }
 
+type LabeledEntry struct {
+	Entry
+	Labels LabelSet
+}
+
 func (e *Entry) MarshalJSON() ([]byte, error) {
 	l, err := json.Marshal(e.Line)
 	if err != nil {

backend/protos/mgmt/v1alpha1/job.proto

@@ -1038,6 +1038,8 @@ message GetJobRunLogsStreamResponse {
   string log_line = 1;
   // The timestamp of the log line
   optional google.protobuf.Timestamp timestamp = 2;
+  // The labels associated with the log line
+  map<string, string> labels = 3;
 }
 
 message GetJobRunLogsRequest {
@@ -1061,6 +1063,8 @@ message GetJobRunLogsResponse {
     string log_line = 1;
     // The timestamp of the log line
     optional google.protobuf.Timestamp timestamp = 2;
+    // The labels associated with the log line
+    map<string, string> labels = 3;
   }
 }
 

backend/services/mgmt/v1alpha1/job-service/runs.go

@@ -581,7 +581,7 @@ func (s *Service) streamK8sWorkerPodLogs(
 				if logLine.Time != nil {
 					timestamp = timestamppb.New(*logLine.Time)
 				}
-				if err := stream.Send(&mgmtv1alpha1.GetJobRunLogsResponse_LogLine{LogLine: txt, Timestamp: timestamp}); err != nil {
+				if err := stream.Send(&mgmtv1alpha1.GetJobRunLogsResponse_LogLine{LogLine: txt, Labels: map[string]string{}, Timestamp: timestamp}); err != nil {
 					if err == io.EOF {
 						return nil
 					}
@@ -665,7 +665,7 @@ func (s *Service) streamLokiWorkerLogs(
 	}
 
 	for _, entry := range entries {
-		err := stream.Send(&mgmtv1alpha1.GetJobRunLogsResponse_LogLine{LogLine: entry.Line, Timestamp: timestamppb.New(entry.Timestamp)})
+		err := stream.Send(&mgmtv1alpha1.GetJobRunLogsResponse_LogLine{LogLine: entry.Line, Labels: entry.Labels.Map(), Timestamp: timestamppb.New(entry.Timestamp)})
 		if err != nil {
 			return err
 		}

docs/openapi/mgmt/v1alpha1/job.openapi.yaml

@@ -1664,8 +1664,26 @@ components:
             - $ref: '#/components/schemas/google.protobuf.Timestamp'
           title: timestamp
           description: The timestamp of the log line
+        labels:
+          type: object
+          title: labels
+          additionalProperties:
+            type: string
+            title: value
+          description: The labels associated with the log line
       title: LogLine
       additionalProperties: false
+    mgmt.v1alpha1.GetJobRunLogsResponse.LogLine.LabelsEntry:
+      type: object
+      properties:
+        key:
+          type: string
+          title: key
+        value:
+          type: string
+          title: value
+      title: LabelsEntry
+      additionalProperties: false
     mgmt.v1alpha1.GetJobRunLogsStreamRequest:
       type: object
       allOf:
@@ -1745,8 +1763,26 @@ components:
             - $ref: '#/components/schemas/google.protobuf.Timestamp'
           title: timestamp
           description: The timestamp of the log line
+        labels:
+          type: object
+          title: labels
+          additionalProperties:
+            type: string
+            title: value
+          description: The labels associated with the log line
       title: GetJobRunLogsStreamResponse
       additionalProperties: false
+    mgmt.v1alpha1.GetJobRunLogsStreamResponse.LabelsEntry:
+      type: object
+      properties:
+        key:
+          type: string
+          title: key
+        value:
+          type: string
+          title: value
+      title: LabelsEntry
+      additionalProperties: false
     mgmt.v1alpha1.GetJobRunRequest:
       type: object
       properties:

docs/openapi/neosync.mgmt.v1alpha1.yaml

@@ -9999,8 +9999,26 @@ components:
             - $ref: '#/components/schemas/google.protobuf.Timestamp'
           title: timestamp
           description: The timestamp of the log line
+        labels:
+          type: object
+          title: labels
+          additionalProperties:
+            type: string
+            title: value
+          description: The labels associated with the log line
       title: LogLine
       additionalProperties: false
+    mgmt.v1alpha1.GetJobRunLogsResponse.LogLine.LabelsEntry:
+      type: object
+      properties:
+        key:
+          type: string
+          title: key
+        value:
+          type: string
+          title: value
+      title: LabelsEntry
+      additionalProperties: false
     mgmt.v1alpha1.GetJobRunLogsStreamRequest:
       type: object
       allOf:
@@ -10084,8 +10102,26 @@ components:
             - $ref: '#/components/schemas/google.protobuf.Timestamp'
           title: timestamp
           description: The timestamp of the log line
+        labels:
+          type: object
+          title: labels
+          additionalProperties:
+            type: string
+            title: value
+          description: The labels associated with the log line
       title: GetJobRunLogsStreamResponse
       additionalProperties: false
+    mgmt.v1alpha1.GetJobRunLogsStreamResponse.LabelsEntry:
+      type: object
+      properties:
+        key:
+          type: string
+          title: key
+        value:
+          type: string
+          title: value
+      title: LabelsEntry
+      additionalProperties: false
     mgmt.v1alpha1.GetJobRunRequest:
       type: object
       properties: