feat(dataobj-explorer): Add stream distribution info to dataobj explorer UI (#16525)

Author: benclive

pkg/dataobj/explorer/inspect.go

@@ -11,12 +11,14 @@ import (
 	"github.com/thanos-io/objstore"
 	"golang.org/x/sync/errgroup"
 
+	"github.com/grafana/loki/v3/pkg/dataobj/internal/dataset"
 	"github.com/grafana/loki/v3/pkg/dataobj/internal/encoding"
 	"github.com/grafana/loki/v3/pkg/dataobj/internal/metadata/datasetmd"
 	"github.com/grafana/loki/v3/pkg/dataobj/internal/metadata/filemd"
 	"github.com/grafana/loki/v3/pkg/dataobj/internal/metadata/logsmd"
 	"github.com/grafana/loki/v3/pkg/dataobj/internal/metadata/streamsmd"
 	"github.com/grafana/loki/v3/pkg/dataobj/internal/result"
+	"github.com/grafana/loki/v3/pkg/dataobj/internal/sections/streams"
 )
 
 type FileMetadata struct {
@@ -68,6 +70,9 @@ type SectionMetadata struct {
 	TotalUncompressedSize uint64            `json:"totalUncompressedSize"`
 	ColumnCount           int               `json:"columnCount"`
 	Columns               []ColumnWithPages `json:"columns"`
+	Distribution          []uint64          `json:"distribution"`
+	MinTimestamp          time.Time         `json:"minTimestamp"`
+	MaxTimestamp          time.Time         `json:"maxTimestamp"`
 }
 
 func (s *Service) handleInspect(w http.ResponseWriter, r *http.Request) {
@@ -90,6 +95,10 @@ func (s *Service) handleInspect(w http.ResponseWriter, r *http.Request) {
 
 	metadata := inspectFile(r.Context(), s.bucket, filename)
 	metadata.LastModified = attrs.LastModified.UTC()
+	for _, section := range metadata.Sections {
+		section.MinTimestamp = section.MinTimestamp.UTC().Truncate(time.Second)
+		section.MaxTimestamp = section.MaxTimestamp.UTC().Truncate(time.Second)
+	}
 
 	w.Header().Set("Content-Type", "application/json")
 	if err := json.NewEncoder(w).Encode(metadata); err != nil {
@@ -147,14 +156,16 @@ func inspectFile(ctx context.Context, bucket objstore.BucketReader, path string)
 			sectionMeta, err = inspectLogsSection(ctx, reader, section)
 			if err != nil {
 				return FileMetadata{
-					Error: fmt.Sprintf("failed to inspect logs section: %v", err),
+					Sections: make([]SectionMetadata, 0, len(sections)),
+					Error:    fmt.Sprintf("failed to inspect logs section: %v", err),
 				}
 			}
 		case filemd.SECTION_TYPE_STREAMS:
 			sectionMeta, err = inspectStreamsSection(ctx, reader, section)
 			if err != nil {
 				return FileMetadata{
-					Error: fmt.Sprintf("failed to inspect streams section: %v", err),
+					Sections: make([]SectionMetadata, 0, len(sections)),
+					Error:    fmt.Sprintf("failed to inspect streams section: %v", err),
 				}
 			}
 		}
@@ -254,8 +265,10 @@ func inspectStreamsSection(ctx context.Context, reader encoding.Decoder, section
 	meta.ColumnCount = len(cols)
 
 	// Create error group for parallel execution
-	g, ctx := errgroup.WithContext(ctx)
+	g, _ := errgroup.WithContext(ctx)
 
+	globalMaxTimestamp := time.Time{}
+	globalMinTimestamp := time.Time{}
 	// Process each column in parallel
 	for i, col := range cols {
 		meta.TotalCompressedSize += col.Info.CompressedSize
@@ -268,6 +281,18 @@ func inspectStreamsSection(ctx context.Context, reader encoding.Decoder, section
 				return err
 			}
 
+			if col.Type == streamsmd.COLUMN_TYPE_MAX_TIMESTAMP && col.Info.Statistics != nil {
+				var ts dataset.Value
+				_ = ts.UnmarshalBinary(col.Info.Statistics.MaxValue)
+				globalMaxTimestamp = time.Unix(0, ts.Int64()).UTC()
+			}
+
+			if col.Type == streamsmd.COLUMN_TYPE_MIN_TIMESTAMP && col.Info.Statistics != nil {
+				var ts dataset.Value
+				_ = ts.UnmarshalBinary(col.Info.Statistics.MinValue)
+				globalMinTimestamp = time.Unix(0, ts.Int64()).UTC()
+			}
+
 			var pageInfos []PageInfo
 			for _, pages := range pageSets {
 				for _, page := range pages {
@@ -309,5 +334,27 @@ func inspectStreamsSection(ctx context.Context, reader encoding.Decoder, section
 		return meta, err
 	}
 
+	if globalMaxTimestamp.IsZero() || globalMinTimestamp.IsZero() {
+		// Short circuit if we don't have any timestamps
+		return meta, nil
+	}
+
+	width := int(globalMaxTimestamp.Add(1 * time.Hour).Truncate(time.Hour).Sub(globalMinTimestamp.Truncate(time.Hour)).Hours())
+	counts := make([]uint64, width)
+	for streamVal := range streams.Iter(ctx, reader) {
+		stream, err := streamVal.Value()
+		if err != nil {
+			return meta, err
+		}
+		for i := stream.MinTimestamp; !i.After(stream.MaxTimestamp); i = i.Add(time.Hour) {
+			hoursBeforeMax := int(globalMaxTimestamp.Sub(i).Hours())
+			counts[hoursBeforeMax]++
+		}
+	}
+
+	meta.MinTimestamp = globalMinTimestamp
+	meta.MaxTimestamp = globalMaxTimestamp
+	meta.Distribution = counts
+
 	return meta, nil
 }

pkg/ui/frontend/src/components/explorer/file-metadata.tsx

@@ -207,6 +207,7 @@ function HeadlineStats({
           </div>
         </div>
       )}
+
       {logCount && (
         <div className="rounded-lg bg-muted/50 p-6 shadow-sm">
           <div className="text-sm text-muted-foreground mb-2">Log Count</div>
@@ -339,6 +340,43 @@ function SectionStats({ section }: SectionStatsProps) {
           </Badge>
         </div>
       </div>
+      {section.distribution && (
+        <div className="rounded-lg bg-muted/50 p-6 shadow-sm">
+          <div className="text-sm text-muted-foreground mb-2">Stream age distribution</div>
+          <div className="space-y-2">
+
+            <div className="text-sm">
+              Spanning {Math.ceil((new Date(section.maxTimestamp).getTime() - new Date(section.minTimestamp).getTime()) / (1000 * 60 * 60)+1)} hours
+              <span className="text-sm text-muted-foreground">
+                <span> from </span><DateHover date={new Date(section.minTimestamp)} /> to <DateHover date={new Date(section.maxTimestamp)} />
+              </span>
+            </div>
+
+            <div className="text-sm">
+              Age within object
+            </div>
+            <div className="mt-4 space-y-1">
+              {section.distribution.map((count, i) => {
+                const maxCount = Math.max(...section.distribution);
+                const percentage = (count / maxCount) * 100;
+                const hours = i+1;
+                return (
+                  <div key={i} className="flex items-center gap-2">
+                    <div className="w-26 text-xs text-right">{`${hours}h`}</div>
+                    <div className="flex-1 h-4 bg-muted rounded-full overflow-hidden">
+                      <div
+                        className="h-full bg-primary/50 rounded-full transition-all"
+                        style={{width: `${percentage}%`}}
+                      />
+                    </div>
+                    <div className="w-20 text-xs">{count.toLocaleString()} streams</div>
+                  </div>
+                );
+              })}
+            </div>
+          </div>
+        </div>
+      )}
     </div>
   );
 }

pkg/ui/frontend/src/types/explorer.ts

@@ -43,12 +43,18 @@ export interface SectionMetadata {
   totalUncompressedSize: number;
   columnCount: number;
   columns: ColumnInfo[];
+  maxTimestamp: string;
+  minTimestamp: string;
+  distribution: number[];
 }
 
 export interface FileMetadataResponse {
   sections: SectionMetadata[];
   error?: string;
   lastModified: string;
+  minTimestamp: string;
+  maxTimestamp: string;
+  distribution: number[];
 }
 
 interface ColumnStatistics {