feat: Add new component loki.source.string

Update documentation and changelog

Author: ravishankar15

CHANGELOG.md

@@ -37,6 +37,8 @@ Main (unreleased)
 
 - Add `truncate` stage for `loki.process` to truncate log entries, label values, and structured_metadata values. (@dehaansa)
 
+- Add a new component `loki.source.string` (@ravishankar15)
+
 ### Enhancements
 
 - Add support of `tls` in components `loki.source.(awsfirehose|gcplog|heroku|api)` and `prometheus.receive_http` and `pyroscope.receive_http`. (@fgouteroux)

docs/sources/reference/components/loki/loki.source.string.md

@@ -0,0 +1,81 @@
+---
+canonical: https://grafana.com/docs/alloy/latest/reference/components/loki/loki.source.string/
+aliases:
+  - ../loki.source.string/ # /docs/alloy/latest/reference/components/loki.source.string/
+description: Learn about loki.string
+labels:
+  stage: general-availability
+  products:
+    - oss
+title: loki.source.string
+---
+
+# `loki.source.string`
+
+`loki.source.string` receives log entries as string from other components and can be ingested to any loki component.
+
+You can specify multiple `loki.source.string` components by giving them different labels.
+
+## Usage
+
+```alloy
+loki.source.string "<LABEL>" {
+    source      = <TARGET>
+    forward_to  = <RECEIVER>
+}
+```
+
+## Arguments
+
+The components consumes the value from the source and converts them to log entries passing them to other components receivers in `forward_to` 
+
+| Name                    | Type                 | Description                                                | Default | Required |
+| ----------------------- | -------------------- | ---------------------------------------------------------- | ------- | -------- |
+| `source`                | `string`             | A value pointing to a string source.                       |         | yes      |
+| `forward_to`            | `LogsReceiver`       | Receiver to send log entries to.                           |         | yes      |
+
+## Blocks
+
+The `loki.source.string` component doesn't support any blocks.
+
+## Exported fields
+
+`loki.source.string` doesn't export any fields.
+
+## Component health
+
+`loki.source.string` is only reported as unhealthy if given an invalid configuration.
+
+## Debug information
+
+`loki.source.string` doesn't expose any component-specific debug information.
+
+## Example
+
+This example creates a pipeline that reads string response from remote `/data` endpoint passes it to `loki.source.string` which converts it to [Loki `LogsReceiver`](../../../compatibility/#loki-logsreceiver-exporters) and then echo the value to stdout:
+
+```alloy
+remote.http "server" {
+  url = "http://localhost:2112/data"
+}
+loki.source.string "stringer" {
+    source      = remote.http.server.content
+    forward_to  = loki.echo.print.receiver
+}
+loki.echo "print" { }
+```
+
+<!-- START GENERATED COMPATIBLE COMPONENTS -->
+
+## Compatible components
+
+`loki.source.string` can accept arguments from the following components:
+
+- Components that export string
+
+{{< admonition type="note" >}}
+Connecting some components may not be sensible or components may require further configuration to make the connection work correctly.
+Refer to the linked documentation for more details.
+{{< /admonition >}}
+
+<!-- END GENERATED COMPATIBLE COMPONENTS -->

internal/component/all/all.go

@@ -58,6 +58,7 @@ import (
 	_ "github.com/grafana/alloy/internal/component/loki/source/kubernetes"                   // Import loki.source.kubernetes
 	_ "github.com/grafana/alloy/internal/component/loki/source/kubernetes_events"            // Import loki.source.kubernetes_events
 	_ "github.com/grafana/alloy/internal/component/loki/source/podlogs"                      // Import loki.source.podlogs
+	_ "github.com/grafana/alloy/internal/component/loki/source/string"                       // Import loki.source.string
 	_ "github.com/grafana/alloy/internal/component/loki/source/syslog"                       // Import loki.source.syslog
 	_ "github.com/grafana/alloy/internal/component/loki/source/windowsevent"                 // Import loki.source.windowsevent
 	_ "github.com/grafana/alloy/internal/component/loki/write"                               // Import loki.write

internal/component/loki/source/string/string.go

@@ -0,0 +1,94 @@
+package string
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/grafana/alloy/internal/component"
+	"github.com/grafana/alloy/internal/component/common/loki"
+	"github.com/grafana/alloy/internal/featuregate"
+	"github.com/grafana/loki/pkg/push"
+)
+
+func init() {
+	component.Register(component.Registration{
+		Name:      "loki.source.string",
+		Stability: featuregate.StabilityGenerallyAvailable,
+		Args:      Arguments{},
+
+		Build: func(opts component.Options, args component.Arguments) (component.Component, error) {
+			return New(opts, args.(Arguments))
+		},
+	})
+}
+
+type Arguments struct {
+	Source    string            `alloy:"source,attr"`
+	ForwardTo loki.LogsReceiver `alloy:"forward_to,attr"`
+}
+
+var _ component.Component = (*Component)(nil)
+
+type Component struct {
+	mut sync.RWMutex
+
+	opts           component.Options
+	args           Arguments
+	stringInActive chan bool
+	stringIn       chan string
+	receiver       loki.LogsReceiver
+}
+
+func New(o component.Options, args Arguments) (*Component, error) {
+	c := &Component{
+		opts:           o,
+		receiver:       args.ForwardTo,
+		stringInActive: make(chan bool, 1),
+		stringIn:       make(chan string),
+	}
+
+	go c.run()
+
+	<-c.stringInActive
+	if err := c.Update(args); err != nil {
+		return nil, err
+	}
+
+	return c, nil
+}
+
+func (c *Component) Run(ctx context.Context) error {
+	<-ctx.Done()
+	close(c.stringIn)
+	close(c.stringInActive)
+	return nil
+}
+
+func (c *Component) run() {
+	c.stringInActive <- true
+	for value := range c.stringIn {
+		entry := loki.Entry{
+			Entry: push.Entry{
+				Timestamp: time.Now(),
+				Line:      value,
+			},
+		}
+		c.receiver.Chan() <- entry
+	}
+}
+
+func (c *Component) Update(args component.Arguments) error {
+	newArgs := args.(Arguments)
+
+	c.mut.Lock()
+	defer c.mut.Unlock()
+	c.args = newArgs
+
+	select {
+	case c.stringIn <- c.args.Source:
+	default:
+	}
+
+	return nil
+}

internal/component/loki/source/string/string_test.go

@@ -0,0 +1,78 @@
+package string
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/go-kit/log"
+	"github.com/grafana/alloy/internal/component"
+	"github.com/grafana/alloy/internal/component/common/loki"
+	"github.com/stretchr/testify/require"
+)
+
+func TestString(t *testing.T) {
+	opts := component.Options{
+		Logger: log.NewNopLogger(),
+	}
+
+	type arguments struct {
+		expected string
+		args     Arguments
+	}
+	receiver := loki.NewLogsReceiver()
+	argsArray := []arguments{
+		{
+			expected: "{\"data\":\"pass1\"}",
+			args: Arguments{
+				Source:    "{\"data\":\"pass1\"}",
+				ForwardTo: receiver,
+			},
+		},
+		{
+			expected: "{\"key\":\"pass\",\"nestedData\":{\"nestedKey\":\"pass\"}}",
+			args: Arguments{
+				Source:    "{\"key\":\"pass\",\"nestedData\":{\"nestedKey\":\"pass\"}}",
+				ForwardTo: receiver,
+			},
+		},
+	}
+
+	initArgs := arguments{
+		expected: "{\"data\":\"init\"}",
+		args: Arguments{
+			Source:    "{\"data\":\"init\"}",
+			ForwardTo: receiver,
+		},
+	}
+
+	comp, err := New(opts, initArgs.args)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	go func() {
+		err = comp.Run(ctx)
+	}()
+
+	// Make sure the first argument is received
+	select {
+	case received := <-receiver.Chan():
+		require.Equal(t, initArgs.expected, received.Line)
+	case <-time.After(time.Second):
+		t.Fatalf("timeout waiting for log entry")
+	}
+
+	// Subsequent update should be received
+	for i, testArgs := range argsArray {
+		comp.Update(testArgs.args)
+
+		select {
+		case received := <-receiver.Chan():
+			require.Equal(t, testArgs.expected, received.Line)
+		case <-time.After(time.Second):
+			t.Fatalf("timeout waiting for log entry %d", i)
+		}
+	}
+}