Pass entriesWritten metric

Author: kalleep

internal/component/loki/source/api/api.go

@@ -56,6 +56,7 @@ func (a *Arguments) labelSet() model.LabelSet {
 
 type Component struct {
 	opts               component.Options
+	metrics            *metrics
 	handler            loki.LogsBatchReceiver
 	uncheckedCollector *util.UncheckedCollector
 
@@ -68,6 +69,7 @@ type Component struct {
 func New(opts component.Options, args Arguments) (*Component, error) {
 	c := &Component{
 		opts:               opts,
+		metrics:            newMetrics(opts.Registerer),
 		handler:            loki.NewLogsBatchReceiver(),
 		uncheckedCollector: util.NewUncheckedCollector(nil),
 
@@ -134,8 +136,9 @@ func (c *Component) Update(args component.Arguments) error {
 
 		var err error
 		c.server, err = source.NewServer(c.opts.Logger, reg, c.handler, source.ServerConfig{
-			Namespace: "loki_source_api",
-			NetConfig: newArgs.Server,
+			Namespace:      "loki_source_api",
+			EntriesWritten: c.metrics.entriesWritten,
+			NetConfig:      newArgs.Server,
 			LogsConfig: &source.LogsConfig{
 				FixedLabels:          newArgs.labelSet(),
 				RelabelRules:         relabel.ComponentToPromRelabelConfigs(newArgs.RelabelRules),

internal/component/loki/source/api/metrics.go

@@ -0,0 +1,20 @@
+package api
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/grafana/alloy/internal/util"
+)
+
+type metrics struct {
+	entriesWritten prometheus.Counter
+}
+
+func newMetrics(reg prometheus.Registerer) *metrics {
+	return &metrics{
+		entriesWritten: util.MustRegisterOrGet(reg, prometheus.NewCounter(prometheus.CounterOpts{
+			Name: "loki_source_api_entries_written",
+			Help: "Total number of entries written.",
+		})).(prometheus.Counter),
+	}
+}

internal/component/loki/source/server.go

@@ -14,13 +14,12 @@ import (
 	"github.com/grafana/alloy/internal/component/common/loki"
 	fnet "github.com/grafana/alloy/internal/component/common/net"
 	"github.com/grafana/alloy/internal/runtime/logging/level"
-	"github.com/grafana/alloy/internal/util"
 )
 
 // Server exposes HTTP routes that ingest log entries and forward them in batches.
 type Server struct {
-	logger  log.Logger
-	metrics *serverMetrics
+	logger         log.Logger
+	entriesWritten prometheus.Counter
 
 	server    *fnet.TargetServer
 	netConfig *fnet.ServerConfig
@@ -56,9 +55,10 @@ type HandlerRoute interface {
 }
 
 type ServerConfig struct {
-	Namespace  string
-	NetConfig  *fnet.ServerConfig
-	LogsConfig *LogsConfig
+	Namespace      string
+	EntriesWritten prometheus.Counter
+	NetConfig      *fnet.ServerConfig
+	LogsConfig     *LogsConfig
 }
 
 type LogsConfig struct {
@@ -74,14 +74,14 @@ func NewServer(logger log.Logger, reg prometheus.Registerer, recv loki.LogsBatch
 	}
 
 	return &Server{
-		logger:        logger,
-		metrics:       newServerMetrics(cfg.Namespace, reg),
-		server:        server,
-		netConfig:     cfg.NetConfig,
-		logsConfig:    cfg.LogsConfig,
-		recv:          recv,
-		once:          sync.Once{},
-		forceShutdown: make(chan struct{}),
+		logger:         logger,
+		entriesWritten: cfg.EntriesWritten,
+		server:         server,
+		netConfig:      cfg.NetConfig,
+		logsConfig:     cfg.LogsConfig,
+		recv:           recv,
+		once:           sync.Once{},
+		forceShutdown:  make(chan struct{}),
 	}, nil
 }
 
@@ -167,7 +167,7 @@ func (s *Server) logsHandler(logsFn func(r *http.Request, opts *LogsConfig) ([]l
 				return
 			}
 
-			s.metrics.entriesWritten.Add(float64(numEntries))
+			s.entriesWritten.Add(float64(numEntries))
 
 			if err != nil {
 				level.Warn(s.logger).Log("msg", "at least one entry failed to be processed", "err", err)
@@ -179,21 +179,3 @@ func (s *Server) logsHandler(logsFn func(r *http.Request, opts *LogsConfig) ([]l
 		w.WriteHeader(status)
 	})
 }
-
-type serverMetrics struct {
-	entriesWritten prometheus.Counter
-}
-
-func newServerMetrics(namespace string, reg prometheus.Registerer) *serverMetrics {
-	m := &serverMetrics{
-		entriesWritten: prometheus.NewCounter(prometheus.CounterOpts{
-			Namespace: namespace,
-			Name:      "entries_written",
-			Help:      "Total number of entries written.",
-		}),
-	}
-
-	m.entriesWritten = util.MustRegisterOrGet(reg, m.entriesWritten).(prometheus.Counter)
-
-	return m
-}

internal/component/loki/source/server_test.go

@@ -351,8 +351,9 @@ func newTestServer(t *testing.T, recv loki.LogsBatchReceiver, cfg ServerConfig,
 
 func testServerConfig(timeout time.Duration, logsConfig *LogsConfig) ServerConfig {
 	return ServerConfig{
-		Namespace:  "test",
-		LogsConfig: logsConfig,
+		Namespace:      "test",
+		EntriesWritten: prometheus.NewCounter(prometheus.CounterOpts{}),
+		LogsConfig:     logsConfig,
 		NetConfig: &fnet.ServerConfig{
 			HTTP: &fnet.HTTPConfig{
 				ListenAddress: "127.0.0.1",