Fix internal metrics namespace for certain components (#2803) (#2804)

Signed-off-by: Paschalis Tsilias <[email protected]>
Co-authored-by: Paschalis Tsilias <[email protected]>

Author: thampiotr

CHANGELOG.md

@@ -7,6 +7,15 @@ This document contains a historical list of changes between releases. Only
 changes that impact end-user behavior are listed; changes to documentation or
 internal API changes are not present.
 
+v1.7.0-rc.3
+-----------------
+
+### Bugfixes
+
+- Fixed a bug where `loki.source.awsfirehose` and `loki.source.gcplog` could
+  not be used from within a module. (@tpaschalis)
+
+
 v1.7.0-rc.2
 -----------------
 

internal/component/loki/source/aws_firehose/component.go

@@ -155,7 +155,8 @@ func (c *Component) Update(args component.Arguments) error {
 		c.rbs = newRelabels
 	}
 
-	jobName := strings.Replace(c.opts.ID, ".", "_", -1)
+	r := strings.NewReplacer(".", "_", "/", "_")
+	jobName := r.Replace(c.opts.ID)
 
 	registry := prometheus.NewRegistry()
 	c.serverMetrics.SetCollector(registry)

internal/component/loki/source/aws_firehose/component_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
 
 	"github.com/grafana/alloy/internal/component"
 	"github.com/grafana/alloy/internal/component/common/loki"
@@ -56,6 +57,45 @@ func (r *receiver) run(ctx context.Context) {
 	}
 }
 
+func TestComponentFromNestedController(t *testing.T) {
+	goleak.VerifyNone(t, goleak.IgnoreTopFunction("go.opencensus.io/stats/view.(*worker).start"))
+
+	opts := component.Options{
+		ID:            "foo/loki.source.awsfirehose.default",
+		Logger:        util.TestAlloyLogger(t),
+		Registerer:    prometheus.NewRegistry(),
+		OnStateChange: func(e component.Exports) {},
+	}
+	ch1, ch2 := loki.NewLogsReceiver(), loki.NewLogsReceiver()
+	r1, r2 := newReceiver(ch1.Chan()), newReceiver(ch2.Chan())
+
+	// call cancelReceivers to terminate them
+	receiverContext, cancelReceivers := context.WithCancel(context.Background())
+	go r1.run(receiverContext)
+	go r2.run(receiverContext)
+
+	args := Arguments{}
+
+	port, err := freeport.GetFreePort()
+	require.NoError(t, err)
+	args.Server = &fnet.ServerConfig{
+		HTTP: &fnet.HTTPConfig{
+			ListenAddress: "localhost",
+			ListenPort:    port,
+		},
+		// assign random grpc port
+		GRPC: &fnet.GRPCConfig{ListenPort: 0},
+	}
+	args.ForwardTo = []loki.LogsReceiver{ch1, ch2}
+
+	// Create and run the component.
+	c, err := New(opts, args)
+	require.NoError(t, err)
+	require.NotNil(t, c)
+
+	cancelReceivers()
+}
+
 func TestComponent(t *testing.T) {
 	opts := component.Options{
 		ID:            "loki.source.awsfirehose",

internal/component/loki/source/gcplog/gcplog.go

@@ -132,7 +132,8 @@ func (c *Component) Update(args component.Arguments) error {
 		}
 	}
 	entryHandler := loki.NewEntryHandler(c.handler.Chan(), func() {})
-	jobName := strings.Replace(c.opts.ID, ".", "_", -1)
+	r := strings.NewReplacer(".", "_", "/", "_")
+	jobName := r.Replace(c.opts.ID)
 
 	if newArgs.PullTarget != nil {
 		// TODO(@tpaschalis) Are there any options from "google.golang.org/api/option"

internal/component/loki/source/gcplog/gcplog_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
 
 	"github.com/grafana/alloy/internal/component"
 	"github.com/grafana/alloy/internal/component/common/loki"
@@ -26,6 +27,43 @@ import (
 // the mock PubSub client inside the component, but we'll find a workaround.
 func TestPull(t *testing.T) {}
 
+func TestPushFromNestedController(t *testing.T) {
+	goleak.VerifyNone(t, goleak.IgnoreTopFunction("go.opencensus.io/stats/view.(*worker).start"))
+
+	opts := component.Options{
+		ID:            "foo/loki.source.gcplog.default",
+		Logger:        util.TestAlloyLogger(t),
+		Registerer:    prometheus.NewRegistry(),
+		OnStateChange: func(e component.Exports) {},
+	}
+
+	ch1, ch2 := loki.NewLogsReceiver(), loki.NewLogsReceiver()
+	args := Arguments{}
+
+	port, err := freeport.GetFreePort()
+	require.NoError(t, err)
+	args.PushTarget = &gcptypes.PushConfig{
+		Server: &fnet.ServerConfig{
+			HTTP: &fnet.HTTPConfig{
+				ListenAddress: "localhost",
+				ListenPort:    port,
+			},
+			// assign random grpc port
+			GRPC: &fnet.GRPCConfig{ListenPort: 0},
+		},
+		Labels: map[string]string{
+			"foo": "bar",
+		},
+	}
+	args.ForwardTo = []loki.LogsReceiver{ch1, ch2}
+	args.RelabelRules = exportedRules
+
+	// Create and run the component.
+	c, err := New(opts, args)
+	require.NoError(t, err)
+	require.NotNil(t, c)
+}
+
 func TestPush(t *testing.T) {
 	opts := component.Options{
 		Logger:        util.TestAlloyLogger(t),