New modules import http (#6414)

* add the import source "import.http"

import.http uses the component remote.http to fetch a module from a http server

* add tests

Author: wildum

pkg/flow/import_test.go

@@ -5,6 +5,7 @@ import (
 	"io/fs"
 	"os"
 	"path"
+	"path/filepath"
 	"strings"
 	"sync"
 	"testing"
@@ -109,6 +110,17 @@ func TestImportString(t *testing.T) {
 	}
 }
 
+func TestImportHTTP(t *testing.T) {
+	directory := "./testdata/import_http"
+	for _, file := range getTestFiles(directory, t) {
+		archive, err := txtar.ParseFile(filepath.Join(directory, file.Name()))
+		require.NoError(t, err)
+		t.Run(archive.Files[0].Name, func(t *testing.T) {
+			testConfig(t, string(archive.Files[0].Data), "", nil)
+		})
+	}
+}
+
 type testImportError struct {
 	description   string
 	main          string

pkg/flow/internal/controller/node_config.go

@@ -27,7 +27,7 @@ func NewConfigNode(block *ast.BlockStmt, globals ComponentGlobals) (BlockNode, d
 		return NewLoggingConfigNode(block, globals), nil
 	case tracingBlockID:
 		return NewTracingConfigNode(block, globals), nil
-	case importsource.BlockImportFile, importsource.BlockImportString:
+	case importsource.BlockImportFile, importsource.BlockImportString, importsource.BlockImportHTTP:
 		return NewImportConfigNode(block, globals, importsource.GetSourceType(block.GetBlockName())), nil
 	default:
 		var diags diag.Diagnostics

pkg/flow/internal/controller/node_config_import.go

@@ -248,7 +248,7 @@ func (cn *ImportConfigNode) processImportedContent(content *ast.File) error {
 		switch componentName {
 		case declareType:
 			cn.processDeclareBlock(blockStmt)
-		case importsource.BlockImportFile, importsource.BlockImportString: // TODO: add other import sources
+		case importsource.BlockImportFile, importsource.BlockImportString, importsource.BlockImportHTTP:
 			err := cn.processImportBlock(blockStmt, componentName)
 			if err != nil {
 				return err

pkg/flow/internal/importsource/import_file.go

@@ -32,8 +32,8 @@ func NewImportFile(managedOpts component.Options, eval *vm.Evaluator, onContentC
 	}
 }
 
-// Arguments holds values which are used to configure the local.file component.
-type Arguments struct {
+// FileArguments holds values which are used to configure the local.file component.
+type FileArguments struct {
 	// Filename indicates the file to watch.
 	Filename string `river:"filename,attr"`
 	// Type indicates how to detect changes to the file.
@@ -42,18 +42,18 @@ type Arguments struct {
 	PollFrequency time.Duration `river:"poll_frequency,attr,optional"`
 }
 
-var DefaultArguments = Arguments{
+var DefaultFileArguments = FileArguments{
 	Type:          file.DetectorFSNotify,
 	PollFrequency: time.Minute,
 }
 
 type importFileConfigBlock struct {
-	LocalFileArguments Arguments `river:",squash"`
+	LocalFileArguments FileArguments `river:",squash"`
 }
 
 // SetToDefault implements river.Defaulter.
 func (a *importFileConfigBlock) SetToDefault() {
-	a.LocalFileArguments = DefaultArguments
+	a.LocalFileArguments = DefaultFileArguments
 }
 
 func (im *ImportFile) Evaluate(scope *vm.Scope) error {

pkg/flow/internal/importsource/import_http.go

@@ -0,0 +1,103 @@
+package importsource
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"reflect"
+	"time"
+
+	"github.com/grafana/agent/component"
+	common_config "github.com/grafana/agent/component/common/config"
+	remote_http "github.com/grafana/agent/component/remote/http"
+	"github.com/grafana/river/vm"
+)
+
+// ImportHTTP imports a module from a HTTP server via the remote.http component.
+type ImportHTTP struct {
+	managedRemoteHTTP *remote_http.Component
+	arguments         component.Arguments
+	managedOpts       component.Options
+	eval              *vm.Evaluator
+}
+
+var _ ImportSource = (*ImportHTTP)(nil)
+
+func NewImportHTTP(managedOpts component.Options, eval *vm.Evaluator, onContentChange func(string)) *ImportHTTP {
+	opts := managedOpts
+	opts.OnStateChange = func(e component.Exports) {
+		onContentChange(e.(remote_http.Exports).Content.Value)
+	}
+	return &ImportHTTP{
+		managedOpts: opts,
+		eval:        eval,
+	}
+}
+
+// HTTPArguments holds values which are used to configure the remote.http component.
+type HTTPArguments struct {
+	URL           string        `river:"url,attr"`
+	PollFrequency time.Duration `river:"poll_frequency,attr,optional"`
+	PollTimeout   time.Duration `river:"poll_timeout,attr,optional"`
+
+	Method  string            `river:"method,attr,optional"`
+	Headers map[string]string `river:"headers,attr,optional"`
+	Body    string            `river:"body,attr,optional"`
+
+	Client common_config.HTTPClientConfig `river:"client,block,optional"`
+}
+
+// DefaultHTTPArguments holds default settings for HTTPArguments.
+var DefaultHTTPArguments = HTTPArguments{
+	PollFrequency: 1 * time.Minute,
+	PollTimeout:   10 * time.Second,
+	Client:        common_config.DefaultHTTPClientConfig,
+	Method:        http.MethodGet,
+}
+
+// SetToDefault implements river.Defaulter.
+func (args *HTTPArguments) SetToDefault() {
+	*args = DefaultHTTPArguments
+}
+
+func (im *ImportHTTP) Evaluate(scope *vm.Scope) error {
+	var arguments HTTPArguments
+	if err := im.eval.Evaluate(scope, &arguments); err != nil {
+		return fmt.Errorf("decoding River: %w", err)
+	}
+	if im.managedRemoteHTTP == nil {
+		var err error
+		im.managedRemoteHTTP, err = remote_http.New(im.managedOpts, remote_http.Arguments{
+			URL:           arguments.URL,
+			PollFrequency: arguments.PollFrequency,
+			PollTimeout:   arguments.PollTimeout,
+			Method:        arguments.Method,
+			Headers:       arguments.Headers,
+			Body:          arguments.Body,
+			Client:        arguments.Client,
+		})
+		if err != nil {
+			return fmt.Errorf("creating http component: %w", err)
+		}
+		im.arguments = arguments
+	}
+
+	if reflect.DeepEqual(im.arguments, arguments) {
+		return nil
+	}
+
+	// Update the existing managed component
+	if err := im.managedRemoteHTTP.Update(arguments); err != nil {
+		return fmt.Errorf("updating component: %w", err)
+	}
+	im.arguments = arguments
+	return nil
+}
+
+func (im *ImportHTTP) Run(ctx context.Context) error {
+	return im.managedRemoteHTTP.Run(ctx)
+}
+
+func (im *ImportHTTP) CurrentHealth() component.Health {
+	return im.managedRemoteHTTP.CurrentHealth()
+}

pkg/flow/internal/importsource/import_source.go

@@ -13,11 +13,13 @@ type SourceType int
 const (
 	File SourceType = iota
 	String
+	HTTP
 )
 
 const (
 	BlockImportFile   = "import.file"
 	BlockImportString = "import.string"
+	BlockImportHTTP   = "import.http"
 )
 
 // ImportSource retrieves a module from a source.
@@ -38,6 +40,8 @@ func NewImportSource(sourceType SourceType, managedOpts component.Options, eval
 		return NewImportFile(managedOpts, eval, onContentChange)
 	case String:
 		return NewImportString(eval, onContentChange)
+	case HTTP:
+		return NewImportHTTP(managedOpts, eval, onContentChange)
 	}
 	panic(fmt.Errorf("unsupported source type: %v", sourceType))
 }
@@ -49,6 +53,8 @@ func GetSourceType(fullName string) SourceType {
 		return File
 	case BlockImportString:
 		return String
+	case BlockImportHTTP:
+		return HTTP
 	}
 	panic(fmt.Errorf("name does not map to a known source type: %v", fullName))
 }

pkg/flow/module_eval_test.go

@@ -259,5 +259,6 @@ func verifyNoGoroutineLeaks(t *testing.T) {
 		t,
 		goleak.IgnoreTopFunction("go.opencensus.io/stats/view.(*worker).start"),
 		goleak.IgnoreTopFunction("go.opentelemetry.io/otel/sdk/trace.(*batchSpanProcessor).processQueue"),
+		goleak.IgnoreTopFunction("internal/poll.runtime_pollWait"), // related to TCP keep alive
 	)
 }

pkg/flow/source.go

@@ -75,7 +75,7 @@ func sourceFromBody(body ast.Body) (*Source, error) {
 			switch fullName {
 			case "declare":
 				declares = append(declares, stmt)
-			case "logging", "tracing", "argument", "export", "import.file", "import.string":
+			case "logging", "tracing", "argument", "export", "import.file", "import.string", "import.http":
 				configs = append(configs, stmt)
 			default:
 				components = append(components, stmt)

pkg/flow/testdata/import_http/import_http_1.txtar

@@ -0,0 +1,19 @@
+Import passthrough module.
+
+-- main.river --
+testcomponents.count "inc" {
+	frequency = "10ms"
+	max = 10
+}
+
+import.http "testImport" {
+	url = "https://raw.githubusercontent.com/wildum/module/master/module_passthrough.river"
+}
+
+testImport.a "cc" {
+	input = testcomponents.count.inc.count
+}
+
+testcomponents.summation "sum" {
+	input = testImport.a.cc.output
+}