Add ext interface and move info creation into sources

Author: kalleep

internal/component/loki/source/file/decompresser.go

@@ -339,8 +339,14 @@ func (d *decompressor) Key() positions.Entry {
 	return d.key
 }
 
-func (d *decompressor) IsRunning() bool {
-	return d.running.Load()
+func (d *decompressor) DebugInfo() any {
+	offset, _ := d.positions.Get(d.key.Path, d.key.Labels)
+	return sourceDebugInfo{
+		Path:       d.key.Path,
+		Labels:     d.key.Labels,
+		IsRunning:  d.running.Load(),
+		ReadOffset: offset,
+	}
 }
 
 // cleanupMetrics removes all metrics exported by this reader

internal/component/loki/source/file/file.go

@@ -386,10 +386,10 @@ func (c *Component) scheduleSources() {
 }
 
 type debugInfo struct {
-	TargetsInfo []targetInfo `alloy:"targets_info,block"`
+	TargetsInfo []any `alloy:"targets_info,block"`
 }
 
-type targetInfo struct {
+type sourceDebugInfo struct {
 	Path       string `alloy:"path,attr"`
 	Labels     string `alloy:"labels,attr"`
 	IsRunning  bool   `alloy:"is_running,attr"`
@@ -404,13 +404,8 @@ func (c *Component) DebugInfo() any {
 	defer c.mut.RUnlock()
 	var res debugInfo
 	for s := range c.scheduler.Sources() {
-		offset, _ := c.posFile.Get(s.Key().Path, s.Key().Labels)
-		res.TargetsInfo = append(res.TargetsInfo, targetInfo{
-			Path:       s.Key().Path,
-			Labels:     s.Key().Labels,
-			IsRunning:  s.IsRunning(),
-			ReadOffset: offset,
-		})
+		ds := s.(source.DebugSource[positions.Entry])
+		res.TargetsInfo = append(res.TargetsInfo, ds.DebugInfo())
 	}
 	return res
 }

internal/component/loki/source/file/tailer.go

@@ -416,10 +416,6 @@ func (t *tailer) Key() positions.Entry {
 	return t.key
 }
 
-func (t *tailer) IsRunning() bool {
-	return t.running.Load()
-}
-
 // cleanupMetrics removes all metrics exported by this tailer
 func (t *tailer) cleanupMetrics() {
 	// When we stop tailing the file, also un-export metrics related to the file
@@ -428,3 +424,13 @@ func (t *tailer) cleanupMetrics() {
 	t.metrics.readBytes.DeleteLabelValues(t.key.Path)
 	t.metrics.totalBytes.DeleteLabelValues(t.key.Path)
 }
+
+func (t *tailer) DebugInfo() any {
+	offset, _ := t.positions.Get(t.key.Path, t.key.Labels)
+	return sourceDebugInfo{
+		Path:       t.key.Path,
+		Labels:     t.key.Labels,
+		IsRunning:  t.running.Load(),
+		ReadOffset: offset,
+	}
+}

internal/component/loki/source/file/tailer_test.go

@@ -355,7 +355,7 @@ func TestTailerCorruptedPositions(t *testing.T) {
 	}()
 
 	require.EventuallyWithT(t, func(c *assert.CollectT) {
-		assert.True(c, tailer.IsRunning())
+		assert.True(c, tailer.running.Load())
 		assert.Equal(c, "16", positionsFile.GetString(logFile.Name(), labels.String()))
 	}, time.Second, 50*time.Millisecond)
 

internal/component/loki/source/scheduler.go

@@ -98,8 +98,12 @@ type Source[K comparable] interface {
 	Run(ctx context.Context)
 	// Key is used to uniquely identify the source.
 	Key() K
-	// IsRunning reports if source is still running.
-	IsRunning() bool
+}
+
+// DebugSource is an optional interface with debug information.
+type DebugSource[k comparable] interface {
+	Source[k]
+	DebugInfo() any
 }
 
 func NewSourceWithRetry[K comparable](source Source[K], config backoff.Config) *SourceWithRetry[K] {
@@ -126,10 +130,6 @@ func (s *SourceWithRetry[K]) Key() K {
 	return s.source.Key()
 }
 
-func (s *SourceWithRetry[K]) IsRunning() bool {
-	return s.source.IsRunning()
-}
-
 // scheduledSource is a source that is already scheduled.
 // to stop the scheduledSource cancel needs to be called.
 type scheduledSource[K comparable] struct {