add log exporter

Author: ashwanthgoli

pkg/engine/engine.go

@@ -459,5 +459,7 @@ func exportCapture(ctx context.Context, capture *xcap.Capture, plan *physical.Pl
 		return parentID, ok
 	})
 
+	xcap.ExportLog(capture, logger)
+
 	return xcap.ExportTrace(ctx, capture, logger)
 }

pkg/xcap/aggregation.go

@@ -10,18 +10,28 @@ type AggregatedObservation struct {
 // Record aggregates a new observation into this aggregated observation.
 // It updates the value according to the statistic's aggregation type.
 func (a *AggregatedObservation) Record(obs Observation) {
-	stat := obs.statistic()
-	val := obs.value()
+	a.aggregate(obs.statistic().Aggregation(), obs.value())
+	a.Count++
+}
+
+// Merge aggregates another AggregatedObservation into this one.
+func (a *AggregatedObservation) Merge(other *AggregatedObservation) {
+	if other == nil {
+		return
+	}
+	a.aggregate(a.Statistic.Aggregation(), other.Value)
+	a.Count += other.Count
+}
 
-	switch stat.Aggregation() {
+func (a *AggregatedObservation) aggregate(aggType AggregationType, val any) {
+	switch aggType {
 	case AggregationTypeSum:
 		switch v := val.(type) {
 		case int64:
 			a.Value = a.Value.(int64) + v
 		case float64:
 			a.Value = a.Value.(float64) + v
 		}
-
 	case AggregationTypeMin:
 		switch v := val.(type) {
 		case int64:
@@ -33,7 +43,6 @@ func (a *AggregatedObservation) Record(obs Observation) {
 				a.Value = v
 			}
 		}
-
 	case AggregationTypeMax:
 		switch v := val.(type) {
 		case int64:
@@ -50,16 +59,12 @@ func (a *AggregatedObservation) Record(obs Observation) {
 				a.Value = v
 			}
 		}
-
 	case AggregationTypeLast:
-		// Last value overwrites
 		a.Value = val
-
 	case AggregationTypeFirst:
+		// Keep the first value, don't update
 		if a.Value == nil {
 			a.Value = val
 		}
 	}
-
-	a.Count++
 }

pkg/xcap/exporter.go

@@ -101,3 +101,73 @@ func observationToAttribute(key StatisticKey, obs *AggregatedObservation) attrib
 	// Fallback: convert to string
 	return attrKey.String(fmt.Sprintf("%v", obs.Value))
 }
+
+// ExportLog exports a Capture as a structured log line with aggregated statistics.
+func ExportLog(capture *Capture, logger log.Logger) {
+	if capture == nil || logger == nil {
+		return
+	}
+
+	summary := summarizeObservations(capture)
+	level.Info(logger).Log(summary.toLogValues()...)
+}
+
+// summarizeObservations collects and summarizes observations from the capture.
+func summarizeObservations(capture *Capture) *observations {
+	if capture == nil {
+		return nil
+	}
+
+	collect := newObservationCollector(capture)
+	result := newObservations()
+
+	// collect observations from all DataObjScan regions. observations from
+	// child regions are rolled-up to include dataset reader and bucket stats.
+	// streamView is excluded as it is handled separately below.
+	result.merge(
+		collect.fromRegions("DataObjScan", true, "streamsView.init").
+			filter(
+				// object store calls
+				"bucket.get", "bucket.getrange", "bucket.attributes",
+				// dataset reader stats
+				"row.max", "rows.after.pruning", "read.calls",
+				"primary.pages.downloaded", "secondary.pages.downloaded",
+				"primary.pages.compressed.bytes", "secondary.pages.compressed.bytes",
+				"primary.rows.read", "secondary.rows.read",
+				"primary.row.read.bytes", "secondary.row.read.bytes",
+				"pages.scanned", "pages.cache.hit",
+				"pages.download.requests", "pages.download.duration.ns",
+			).
+			prefix("logs_dataset_").
+			normalizeKeys(),
+	)
+
+	// metastore index and resolved section stats
+	result.merge(
+		collect.fromRegions("ObjectMetastore.Sections", true).
+			filter("metastore.index.objects", "metastore.resolved.sections").
+			normalizeKeys(),
+	)
+
+	// metastore bucket and dataset reader stats
+	result.merge(
+		collect.fromRegions("ObjectMetastore.Sections", true).
+			filter("bucket.get", "bucket.getrange", "bucket.attributes",
+				"primary.pages.downloaded", "secondary.pages.downloaded",
+				"primary.pages.compressed.bytes", "secondary.pages.compressed.bytes").
+			prefix("metastore_").
+			normalizeKeys(),
+	)
+
+	// streamsView bucket and dataset reader stats
+	result.merge(
+		collect.fromRegions("streamsView.init", true).
+			filter("bucket.get", "bucket.getrange", "bucket.attributes",
+				"primary.pages.downloaded", "secondary.pages.downloaded",
+				"primary.pages.compressed.bytes", "secondary.pages.compressed.bytes").
+			prefix("streams_").
+			normalizeKeys(),
+	)
+
+	return result
+}

pkg/xcap/summary.go

@@ -0,0 +1,231 @@
+package xcap
+
+import (
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/dustin/go-humanize"
+)
+
+// observations holds aggregated observations that can be transformed and merged.
+//
+// All transformation methods (filter, prefix, normalizeKeys) return new instances,
+// leaving the original unchanged.
+type observations struct {
+	data map[StatisticKey]*AggregatedObservation
+}
+
+// newObservations creates an empty observations.
+func newObservations() *observations {
+	return &observations{data: make(map[StatisticKey]*AggregatedObservation)}
+}
+
+// filter returns a new observations containing only entries with matching stat names.
+func (o *observations) filter(names ...string) *observations {
+	if len(names) == 0 || o == nil {
+		return o
+	}
+
+	nameSet := make(map[string]struct{}, len(names))
+	for _, n := range names {
+		nameSet[n] = struct{}{}
+	}
+
+	result := newObservations()
+	for k, obs := range o.data {
+		if _, ok := nameSet[k.Name]; ok {
+			result.data[k] = obs
+		}
+	}
+	return result
+}
+
+// prefix returns a new observations with all stat names prefixed.
+func (o *observations) prefix(p string) *observations {
+	if p == "" || o == nil {
+		return o
+	}
+
+	result := newObservations()
+	for k, obs := range o.data {
+		newKey := StatisticKey{
+			Name:        p + k.Name,
+			DataType:    k.DataType,
+			Aggregation: k.Aggregation,
+		}
+		result.data[newKey] = obs
+	}
+	return result
+}
+
+// normalizeKeys returns a new observations with dots replaced by underscores in stat names.
+func (o *observations) normalizeKeys() *observations {
+	if o == nil {
+		return o
+	}
+
+	result := newObservations()
+	for k, obs := range o.data {
+		newKey := StatisticKey{
+			Name:        strings.ReplaceAll(k.Name, ".", "_"),
+			DataType:    k.DataType,
+			Aggregation: k.Aggregation,
+		}
+		result.data[newKey] = obs
+	}
+	return result
+}
+
+// merge merges another observations into this one.
+func (o *observations) merge(other *observations) {
+	if other == nil {
+		return
+	}
+	for k, obs := range other.data {
+		if existing, ok := o.data[k]; ok {
+			existing.Merge(obs)
+		} else {
+			o.data[k] = &AggregatedObservation{
+				Statistic: obs.Statistic,
+				Value:     obs.Value,
+				Count:     obs.Count,
+			}
+		}
+	}
+}
+
+// ToLogValues converts observations to a slice suitable for go-kit/log.
+// Keys are sorted for deterministic output.
+func (o *observations) toLogValues() []any {
+	if o == nil {
+		return nil
+	}
+
+	// Collect key-value pairs for sorting by name.
+	type kv struct {
+		name  string
+		value any
+	}
+	pairs := make([]kv, 0, len(o.data))
+	for k, obs := range o.data {
+		pairs = append(pairs, kv{name: k.Name, value: obs.Value})
+	}
+	sort.Slice(pairs, func(i, j int) bool {
+		return strings.Compare(pairs[i].name, pairs[j].name) < 0
+	})
+
+	result := make([]any, 0, len(pairs)*2)
+	for _, p := range pairs {
+		value := p.value
+
+		// Format bytes values (keys ending with "_bytes")
+		if strings.HasSuffix(p.name, "_bytes") {
+			switch val := value.(type) {
+			case uint64:
+				value = humanize.Bytes(val)
+			case int64:
+				value = humanize.Bytes(uint64(val))
+			}
+		}
+
+		// Format duration values (keys ending with "duration_ns")
+		if strings.HasSuffix(p.name, "duration_ns") {
+			switch val := value.(type) {
+			case int64:
+				value = time.Duration(val).String()
+			case uint64:
+				value = time.Duration(val).String()
+			}
+		}
+
+		result = append(result, p.name, value)
+	}
+	return result
+}
+
+// observationCollector provides methods to collect observations from a Capture.
+type observationCollector struct {
+	capture     *Capture
+	childrenMap map[identifier][]*Region
+}
+
+// newObservationCollector creates a new collector for gathering observations from the given capture.
+func newObservationCollector(capture *Capture) *observationCollector {
+	if capture == nil {
+		return nil
+	}
+
+	// Build parent -> children map
+	childrenMap := make(map[identifier][]*Region)
+	for _, r := range capture.Regions() {
+		childrenMap[r.parentID] = append(childrenMap[r.parentID], r)
+	}
+
+	return &observationCollector{
+		capture:     capture,
+		childrenMap: childrenMap,
+	}
+}
+
+// fromRegions collects observations from regions with the given name.
+// If rollUp is true, each region's stats include all its descendant stats
+// aggregated according to each stat's aggregation type.
+func (c *observationCollector) fromRegions(name string, rollUp bool, excluded ...string) *observations {
+	if c == nil {
+		return newObservations()
+	}
+
+	excludedSet := make(map[string]struct{}, len(excluded))
+	for _, name := range excluded {
+		excludedSet[name] = struct{}{}
+	}
+
+	result := newObservations()
+	for _, region := range c.capture.Regions() {
+		if region.name != name {
+			continue
+		}
+
+		var obs *observations
+		if rollUp {
+			obs = c.rollUpObservations(region, excludedSet)
+		} else {
+			obs = c.getRegionObservations(region)
+		}
+
+		result.merge(obs)
+	}
+
+	return result
+}
+
+// getRegionObservations returns a copy of a region's observations.
+func (c *observationCollector) getRegionObservations(region *Region) *observations {
+	result := newObservations()
+	for k, obs := range region.observations {
+		result.data[k] = &AggregatedObservation{
+			Statistic: obs.Statistic,
+			Value:     obs.Value,
+			Count:     obs.Count,
+		}
+	}
+	return result
+}
+
+// rollUpObservations computes observations for a region including all its descendants.
+// Stats are aggregated according to their aggregation type.
+func (c *observationCollector) rollUpObservations(region *Region, excludedSet map[string]struct{}) *observations {
+	result := c.getRegionObservations(region)
+
+	// Recursively aggregate from children.
+	for _, child := range c.childrenMap[region.id] {
+		// Skip children with excluded names.
+		if _, excluded := excludedSet[child.name]; excluded {
+			continue
+		}
+		result.merge(c.rollUpObservations(child, excludedSet))
+	}
+
+	return result
+}