nsc logs: label unlabeled log lines

internal/cli/cmd/cluster/logs.go

@@ -13,6 +13,7 @@ import (
 	"os"
 	"slices"
 	"sort"
+	"strings"
 	"time"
 
 	"github.com/spf13/cobra"
@@ -247,6 +248,7 @@ type logOutput interface {
 const (
 	namespaceLogLabel  = "namespace"
 	k8sPodNameLogLabel = "kubernetes_pod_name"
+	systemLogLabel     = "system"
 )
 
 var defaultNamespaces = []string{"", "default"}
@@ -263,7 +265,29 @@ func newLogPrinter(useStdout bool) *plainLogPrinter {
 	}
 }
 
-func (lp *plainLogPrinter) writer(ctx context.Context, labels map[string]string, stream string) io.Writer {
+func visibleLogLabel(labels map[string]string, stream, source string) string {
+	if pod, ok := labels[k8sPodNameLogLabel]; ok && pod != "" {
+		label := pod
+
+		if ns, ok := labels[namespaceLogLabel]; ok && !slices.Contains(defaultNamespaces, ns) {
+			label = fmt.Sprintf("%s/%s", ns, label)
+		}
+
+		return label
+	}
+
+	if stream != "" {
+		return stream
+	}
+
+	if system, ok := labels[systemLogLabel]; ok && system != "" {
+		return system
+	}
+
+	return source
+}
+
+func (lp *plainLogPrinter) writer(ctx context.Context, labels map[string]string, stream, source string) io.Writer {
 	if lp.useStdout {
 		return console.Stdout(ctx)
 	}
@@ -272,7 +296,7 @@ func (lp *plainLogPrinter) writer(ctx context.Context, labels map[string]string,
 	keys := maps.Keys(labels)
 	sort.Strings(keys)
 
-	key := stream
+	key := fmt.Sprintf("%s/%s", source, stream)
 	for _, k := range keys {
 		key = fmt.Sprintf("%s/%s:%s", key, k, labels[k])
 	}
@@ -281,16 +305,11 @@ func (lp *plainLogPrinter) writer(ctx context.Context, labels map[string]string,
 		return out
 	}
 
-	// Only use namespace and pod name in user visible label since console space is limited
-	var label string
-	if pod, ok := labels[k8sPodNameLogLabel]; ok {
-		label = pod
-
-		if ns, ok := labels[namespaceLogLabel]; ok && !slices.Contains(defaultNamespaces, ns) {
-			label = fmt.Sprintf("%s/%s", ns, label)
-		}
-	} else {
-		label = stream
+	label := visibleLogLabel(labels, stream, source)
+	if label == "" {
+		out := console.Stdout(ctx)
+		lp.outs[key] = out
+		return out
 	}
 
 	out := console.Output(ctx, label)
@@ -300,18 +319,40 @@ func (lp *plainLogPrinter) writer(ctx context.Context, labels map[string]string,
 
 func (lp *plainLogPrinter) PrintBlock(ctx context.Context, lb api.LogBlock) error {
 	for _, l := range lb.Line {
-		out := lp.writer(ctx, lb.Labels, l.Stream)
-		fmt.Fprintf(out, "%s %s\n", l.Timestamp.Format(time.RFC3339), l.Content)
+		out := lp.writer(ctx, lb.Labels, l.Stream, l.Source)
+		printLogContent(out, l.Timestamp, l.Content)
 	}
 	return nil
 }
 
 func (lp *plainLogPrinter) PrintLine(ctx context.Context, l api.LogLine) error {
-	out := lp.writer(ctx, l.Labels, l.Stream)
-	fmt.Fprintf(out, "%s %s\n", l.Timestamp.Format(time.RFC3339), l.Content)
+	out := lp.writer(ctx, l.Labels, l.Stream, l.Source)
+	printLogContent(out, l.Timestamp, l.Content)
 	return nil
 }
 
+func printLogContent(out io.Writer, ts time.Time, content string) {
+	for _, line := range logicalLogLines(content) {
+		fmt.Fprintf(out, "%s %s\n", ts.Format(time.RFC3339), line)
+	}
+}
+
+func logicalLogLines(content string) []string {
+	if content == "" {
+		return []string{""}
+	}
+
+	normalized := strings.ReplaceAll(content, "\r\n", "\n")
+	normalized = strings.ReplaceAll(normalized, "\r", "\n")
+
+	parts := strings.Split(normalized, "\n")
+	for len(parts) > 1 && parts[len(parts)-1] == "" {
+		parts = parts[:len(parts)-1]
+	}
+
+	return parts
+}
+
 // jsonLogPrinter outputs each log line as a JSON object (JSONL format).
 
 type jsonLogPrinter struct {

internal/cli/cmd/cluster/logs_test.go

@@ -0,0 +1,96 @@
+// Copyright 2022 Namespace Labs Inc; All rights reserved.
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+
+package cluster
+
+import "testing"
+
+func TestVisibleLogLabel(t *testing.T) {
+	tests := []struct {
+		name   string
+		labels map[string]string
+		stream string
+		source string
+		want   string
+	}{
+		{
+			name: "pod label keeps non-default namespace",
+			labels: map[string]string{
+				namespaceLogLabel:  "buildkite",
+				k8sPodNameLogLabel: "agent-0",
+			},
+			want: "buildkite/agent-0",
+		},
+		{
+			name:   "stream fallback",
+			stream: "stderr",
+			want:   "stderr",
+		},
+		{
+			name: "system fallback",
+			labels: map[string]string{
+				systemLogLabel: "kernel",
+			},
+			source: "kmsg",
+			want:   "kernel",
+		},
+		{
+			name:   "source fallback",
+			source: "containers",
+			want:   "containers",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := visibleLogLabel(tt.labels, tt.stream, tt.source); got != tt.want {
+				t.Fatalf("visibleLogLabel(...) = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestLogicalLogLines(t *testing.T) {
+	tests := []struct {
+		name    string
+		content string
+		want    []string
+	}{
+		{
+			name:    "empty",
+			content: "",
+			want:    []string{""},
+		},
+		{
+			name:    "carriage returns become new lines",
+			content: "Receiving 10%\rReceiving 20%\rDone\r",
+			want:    []string{"Receiving 10%", "Receiving 20%", "Done"},
+		},
+		{
+			name:    "mixed newline styles",
+			content: "a\r\nb\nc",
+			want:    []string{"a", "b", "c"},
+		},
+		{
+			name:    "preserve empty interior lines",
+			content: "a\n\nb",
+			want:    []string{"a", "", "b"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := logicalLogLines(tt.content)
+			if len(got) != len(tt.want) {
+				t.Fatalf("logicalLogLines(%q) len=%d, want %d (%q)", tt.content, len(got), len(tt.want), got)
+			}
+
+			for i := range got {
+				if got[i] != tt.want[i] {
+					t.Fatalf("logicalLogLines(%q)[%d] = %q, want %q", tt.content, i, got[i], tt.want[i])
+				}
+			}
+		})
+	}
+}