chore: Add server abstration for loki source components (#5895)

### Pull Request Details
After a couple of pr:s our loki source components that exposes http
server are almost identical except for how entries are parsed.

So I created an abstraction around http server for loki sources and
ported over `loki.source.api` to it. I moved all life-cycle related test
to the server and moved/re-wrote handler specific tests to
`routes_test.go`.

### Issue(s) fixed by this Pull Request

Part of: #5803

### Notes to the Reviewer

After this I will refactor the other components we have to use the same
abstration.

### PR Checklist

<!-- Remove items that do not apply. For completed items, change [ ] to
[x]. -->

- [ ] Documentation added
- [x] Tests updated
- [ ] Config converters updated

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Denys Sedchenko <9203548+x1unix@users.noreply.github.com>

Author: 3 people

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

@@ -3,7 +3,6 @@ package api
 import (
 	"context"
 	"fmt"
-	"reflect"
 	"sync"
 
 	"github.com/alecthomas/units"
@@ -14,7 +13,7 @@ import (
 	"github.com/grafana/alloy/internal/component/common/loki"
 	fnet "github.com/grafana/alloy/internal/component/common/net"
 	"github.com/grafana/alloy/internal/component/common/relabel"
-	"github.com/grafana/alloy/internal/component/loki/source/api/internal/lokipush"
+	"github.com/grafana/alloy/internal/component/loki/source"
 	"github.com/grafana/alloy/internal/featuregate"
 	"github.com/grafana/alloy/internal/util"
 )
@@ -61,7 +60,7 @@ type Component struct {
 	uncheckedCollector *util.UncheckedCollector
 
 	serverMut sync.Mutex
-	server    *lokipush.PushAPIServer
+	server    *source.Server
 
 	fanout *loki.Fanout
 }
@@ -120,8 +119,8 @@ func (c *Component) Update(args component.Arguments) error {
 
 	c.serverMut.Lock()
 	defer c.serverMut.Unlock()
-	serverNeedsRestarting := c.server == nil || !reflect.DeepEqual(c.server.ServerConfig(), *newArgs.Server)
-	if serverNeedsRestarting {
+
+	if c.server.NeedsRestart(newArgs.Server) {
 		if c.server != nil {
 			c.server.Shutdown()
 		}
@@ -130,23 +129,39 @@ func (c *Component) Update(args component.Arguments) error {
 		// avoid issues with re-registering metrics with the same name, we create a
 		// new registry for the server every time we create one, and pass it to an
 		// unchecked collector to bypass uniqueness checking.
-		serverRegistry := prometheus.NewRegistry()
-		c.uncheckedCollector.SetCollector(serverRegistry)
+		reg := prometheus.NewRegistry()
+		c.uncheckedCollector.SetCollector(reg)
 
 		var err error
-		c.server, err = lokipush.NewPushAPIServer(c.opts.Logger, newArgs.Server, c.handler, serverRegistry, int64(newArgs.MaxSendMessageSize))
+		c.server, err = source.NewServer(c.opts.Logger, reg, c.handler, source.ServerConfig{
+			Namespace: "loki_source_api",
+			NetConfig: newArgs.Server,
+			LogsConfig: &source.LogsConfig{
+				FixedLabels:          newArgs.labelSet(),
+				RelabelRules:         relabel.ComponentToPromRelabelConfigs(newArgs.RelabelRules),
+				UseIncomingTimestamp: newArgs.UseIncomingTimestamp,
+			},
+		})
+
 		if err != nil {
 			return fmt.Errorf("failed to create embedded server: %v", err)
 		}
 
-		if err = c.server.Run(); err != nil {
+		logsRoutes, handlerRoutes := newRoutes(int(newArgs.MaxSendMessageSize))
+		if err = c.server.Run(logsRoutes, handlerRoutes); err != nil {
 			return fmt.Errorf("failed to run embedded server: %v", err)
 		}
+
+		return nil
 	}
 
-	c.server.SetLabels(newArgs.labelSet())
-	c.server.SetRelabelRules(newArgs.RelabelRules)
-	c.server.SetKeepTimestamp(newArgs.UseIncomingTimestamp)
+	if c.server != nil {
+		c.server.Update(&source.LogsConfig{
+			FixedLabels:          newArgs.labelSet(),
+			RelabelRules:         relabel.ComponentToPromRelabelConfigs(newArgs.RelabelRules),
+			UseIncomingTimestamp: newArgs.UseIncomingTimestamp,
+		})
+	}
 
 	return nil
 }

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

@@ -348,7 +348,7 @@ func TestComponent_detectsWhenUpdateRequiresARestart(t *testing.T) {
 			assert.Equal(t, restarted, tc.restartRequired)
 
 			// in order to cleanly shutdown, we want to make sure the server is running first.
-			waitForServerToBeReady(t, comp)
+			waitForServerToBeReady(t, comp, tc.args.Server.HTTP.TLSConfig != nil)
 		})
 	}
 }
@@ -376,7 +376,7 @@ func TestLokiSourceAPI_TLS(t *testing.T) {
 	c := startTestComponent(t, opts, args, ctx)
 
 	// Create TLS-enabled Loki client
-	lokiClient := newTestLokiClientTLS(t, c.server.HTTPListenAddress(), opts)
+	lokiClient := newTestLokiClientTLS(t, c.server.HTTPAddr(), opts)
 	defer lokiClient.Stop()
 
 	now := time.Now()
@@ -449,12 +449,12 @@ func TestDefaultServerConfig(t *testing.T) {
 	require.NoError(t, err)
 
 	require.Eventuallyf(t, func() bool {
-		resp, err := http.Get(fmt.Sprintf(
-			"http://%v:%d/wrong/url",
-			"localhost",
-			fnet.DefaultHTTPPort,
-		))
-		return err == nil && resp.StatusCode == 404
+		resp, err := http.Get(fmt.Sprintf("http://%s/ready", comp.server.HTTPAddr()))
+		if err != nil {
+			return false
+		}
+		require.NoError(t, err)
+		return resp.StatusCode == 200
 	}, 5*time.Second, 20*time.Millisecond, "server failed to start before timeout")
 }
 
@@ -472,7 +472,7 @@ func startTestComponent(
 		require.NoError(t, err)
 	}()
 
-	waitForServerToBeReady(t, comp)
+	waitForServerToBeReady(t, comp, args.Server.HTTP.TLSConfig != nil)
 	return comp
 }
 
@@ -493,16 +493,16 @@ func TestShutdown(t *testing.T) {
 		require.NoError(t, err)
 	}()
 
-	waitForServerToBeReady(t, comp)
+	waitForServerToBeReady(t, comp, args.Server.HTTP.TLSConfig != nil)
 
 	// First request should be forwarded on channel
-	_, err = http.DefaultClient.Do(newRequest(t, context.Background(), comp.server.HTTPListenAddress()))
+	_, err = http.DefaultClient.Do(newRequest(t, context.Background(), comp.server.HTTPAddr()))
 	require.NoError(t, err)
 
 	codes := make(chan int)
 	for range 5 {
 		go func() {
-			res, err := http.DefaultClient.Do(newRequest(t, context.Background(), comp.server.HTTPListenAddress()))
+			res, err := http.DefaultClient.Do(newRequest(t, context.Background(), comp.server.HTTPAddr()))
 			if err != nil || res == nil {
 				// This should not happen but if it does we return -1 here so test will fail.
 				codes <- -1
@@ -546,18 +546,18 @@ func TestCancelRequest(t *testing.T) {
 		require.NoError(t, err)
 	}()
 
-	waitForServerToBeReady(t, comp)
+	waitForServerToBeReady(t, comp, args.Server.HTTP.TLSConfig != nil)
 
 	// First request should be forwarded on channel
-	_, err = http.DefaultClient.Do(newRequest(t, context.Background(), comp.server.HTTPListenAddress()))
+	_, err = http.DefaultClient.Do(newRequest(t, context.Background(), comp.server.HTTPAddr()))
 	require.NoError(t, err)
 
 	var wg sync.WaitGroup
 	for range 5 {
 		wg.Go(func() {
 			ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
 			defer cancel()
-			res, err := http.DefaultClient.Do(newRequest(t, ctx, comp.server.HTTPListenAddress()))
+			res, err := http.DefaultClient.Do(newRequest(t, ctx, comp.server.HTTPAddr()))
 			require.ErrorIs(t, err, context.DeadlineExceeded)
 			require.Nil(t, res)
 		})
@@ -576,13 +576,12 @@ func newRequest(t *testing.T, ctx context.Context, httpListendAddress string) *h
 	return req
 }
 
-func waitForServerToBeReady(t *testing.T, comp *Component) {
+func waitForServerToBeReady(t *testing.T, comp *Component, useTls bool) {
 	// Determine if TLS is enabled to choose the right protocol
 	protocol := "http"
 	var tlsConfig *tls.Config
 
-	serverConfig := comp.server.ServerConfig()
-	if serverConfig.HTTP.TLSConfig != nil {
+	if useTls {
 		protocol = "https"
 		tlsConfig = &tls.Config{
 			InsecureSkipVerify: true,
@@ -592,7 +591,7 @@ func waitForServerToBeReady(t *testing.T, comp *Component) {
 	url := fmt.Sprintf(
 		"%s://%s/wrong/url",
 		protocol,
-		comp.server.HTTPListenAddress(),
+		comp.server.HTTPAddr(),
 	)
 
 	client := &http.Client{Timeout: 1 * time.Second}
@@ -604,8 +603,12 @@ func waitForServerToBeReady(t *testing.T, comp *Component) {
 
 	require.Eventuallyf(t, func() bool {
 		resp, err := client.Get(url)
+		if err != nil {
+			return false
+		}
 
-		return err == nil && resp != nil && resp.StatusCode == 404
+		require.NoError(t, resp.Body.Close())
+		return resp.StatusCode == 404
 	}, 5*time.Second, 20*time.Millisecond, "server failed to start before timeout")
 }