feat: tool to build custom Telegraf builds (#11524)

.gitignore

@@ -6,6 +6,7 @@
 /tools/package_lxd_test/package_lxd_test
 /tools/license_checker/license_checker*
 /tools/readme_config_includer/generator*
+/tools/custom_builder/custom_builder*
 /vendor
 .DS_Store
 process.yml

.golangci.yml

@@ -126,7 +126,7 @@ issues:
     - path: _test\.go
       text: "parameter.*seems to be a control flag, avoid control coupling"
 
-    - path: (^agent/|^cmd/|^config/|^filter/|^internal/|^logger/|^metric/|^models/|^selfstat/|^testutil/|^plugins/serializers/|^plugins/inputs/zipkin/cmd)
+    - path: (^agent/|^cmd/|^config/|^filter/|^internal/|^logger/|^metric/|^models/|^selfstat/|^testutil/|^tools|^plugins/serializers/|^plugins/inputs/zipkin/cmd)
       text: "imports-blacklist: should not use the following blacklisted import: \"log\""
       linters:
         - revive

Makefile

@@ -114,6 +114,7 @@ versioninfo:
 	go generate cmd/telegraf/telegraf_windows.go; \
 
 build_tools:
+	$(HOSTGO) build -o ./tools/custom_builder/custom_builder$(EXEEXT) ./tools/custom_builder
 	$(HOSTGO) build -o ./tools/license_checker/license_checker$(EXEEXT) ./tools/license_checker
 	$(HOSTGO) build -o ./tools/readme_config_includer/generator$(EXEEXT) ./tools/readme_config_includer/generator.go
 
@@ -223,6 +224,8 @@ clean:
 	rm -f telegraf
 	rm -f telegraf.exe
 	rm -rf build
+	rm -rf tools/custom_builder/custom_builder
+	rm -rf tools/custom_builder/custom_builder.exe
 	rm -rf tools/readme_config_includer/generator
 	rm -rf tools/readme_config_includer/generator.exe
 	rm -rf tools/package_lxd_test/package_lxd_test

cmd/telegraf/telegraf.go

@@ -27,12 +27,15 @@ import (
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/internal/goplugin"
 	"github.com/influxdata/telegraf/logger"
+	"github.com/influxdata/telegraf/plugins/aggregators"
 	_ "github.com/influxdata/telegraf/plugins/aggregators/all"
 	"github.com/influxdata/telegraf/plugins/inputs"
 	_ "github.com/influxdata/telegraf/plugins/inputs/all"
 	"github.com/influxdata/telegraf/plugins/outputs"
 	_ "github.com/influxdata/telegraf/plugins/outputs/all"
+	"github.com/influxdata/telegraf/plugins/parsers"
 	_ "github.com/influxdata/telegraf/plugins/parsers/all"
+	"github.com/influxdata/telegraf/plugins/processors"
 	_ "github.com/influxdata/telegraf/plugins/processors/all"
 	"gopkg.in/tomb.v1"
 )
@@ -271,7 +274,14 @@ func runAgent(ctx context.Context,
 
 	logger.SetupLogging(logConfig)
 
-	log.Printf("I! Starting Telegraf %s", internal.Version())
+	log.Printf("I! Starting Telegraf %s%s", internal.Version(), internal.Customized)
+	log.Printf("I! Available plugins: %d inputs, %d aggregators, %d processors, %d parsers, %d outputs",
+		len(inputs.Inputs),
+		len(aggregators.Aggregators),
+		len(processors.Processors),
+		len(parsers.Parsers),
+		len(outputs.Outputs),
+	)
 	log.Printf("I! Loaded inputs: %s", strings.Join(c.InputNames(), " "))
 	log.Printf("I! Loaded aggregators: %s", strings.Join(c.AggregatorNames(), " "))
 	log.Printf("I! Loaded processors: %s", strings.Join(c.ProcessorNames(), " "))

config/config.go

@@ -386,7 +386,7 @@ func (c *Config) LoadConfig(path string) error {
 			return err
 		}
 	}
-	data, err := loadConfig(path)
+	data, err := LoadConfigFile(path)
 	if err != nil {
 		return fmt.Errorf("Error loading config file %s: %w", path, err)
 	}
@@ -565,7 +565,7 @@ func escapeEnv(value string) string {
 	return envVarEscaper.Replace(value)
 }
 
-func loadConfig(config string) ([]byte, error) {
+func LoadConfigFile(config string) ([]byte, error) {
 	if fetchURLRe.MatchString(config) {
 		u, err := url.Parse(config)
 		if err != nil {

docs/CUSTOMIZATION.md

@@ -17,13 +17,13 @@ build as otherwise _all_ plugins will be selected regardless of other tags.
 ## Via make
 
 When using the project's makefile, the build can be customized via the
-`BUILDTAGS` environment variable containing a __space-separated__ list of the
+`BUILDTAGS` environment variable containing a __comma-separated__ list of the
 selected plugins (or categories) __and__ the `custom` tag.
 
 For example
 
 ```shell
-BUILDTAGS="custom inputs outputs.influxdb_v2 parsers.json" make
+BUILDTAGS="custom,inputs,outputs.influxdb_v2,parsers.json" make
 ```
 
 will build a customized Telegraf including _all_ `inputs`, the InfluxDB v2
@@ -32,13 +32,13 @@ will build a customized Telegraf including _all_ `inputs`, the InfluxDB v2
 ## Via `go build`
 
 If you wish to build Telegraf using native go tools, you can use the `go build`
-command with the `-tags` option. Specify  a __space-separated__ list of the
+command with the `-tags` option. Specify  a __comma-separated__ list of the
 selected plugins (or categories) __and__ the `custom` tag as argument.
 
 For example
 
 ```shell
-go build -tags "custom inputs outputs.influxdb_v2 parsers.json" ./cmd/telegraf
+go build -tags "custom,inputs,outputs.influxdb_v2,parsers.json" ./cmd/telegraf
 ```
 
 will build a customized Telegraf including _all_ `inputs`, the InfluxDB v2

filter/filter.go

@@ -45,6 +45,14 @@ func Compile(filters []string) (Filter, error) {
 	}
 }
 
+func MustCompile(filters []string) Filter {
+	f, err := Compile(filters)
+	if err != nil {
+		panic(err)
+	}
+	return f
+}
+
 // hasMeta reports whether path contains any magic glob characters.
 func hasMeta(s string) bool {
 	return strings.ContainsAny(s, "*?[")

internal/customized_no.go

@@ -0,0 +1,5 @@
+//go:build !custom
+
+package internal
+
+const Customized = ""

internal/customized_yes.go

@@ -0,0 +1,5 @@
+//go:build custom
+
+package internal
+
+const Customized = " (customized)"

tools/custom_builder/README.md

@@ -0,0 +1,81 @@
+# Telegraf customization tool
+
+Telegraf's `custom_builder` is a tool to select the plugins compiled into the
+Telegraf binary. By doing so, Telegraf can become smaller, saving both disk
+space and memory if only a sub-set of plugins is selected.
+
+## Building
+
+To build `custom_builder` run the following command:
+
+```shell
+# make build_tools
+```
+
+The resulting binary is located in the `tools/custom_builder` folder.
+
+## Running
+
+The easiest way of building a customized Telegraf is to use your
+Telegraf configuration file(s). Assuming your configuration is
+in `/etc/telegraf/telegraf.conf` you can run
+
+```shell
+# ./tools/custom_builder/custom_builder --config /etc/telegraf/telegraf.conf
+```
+
+to build a Telegraf binary tailored to your configuration.
+You can also specify a configuration directory similar to
+Telegraf itself. To additionally use the configurations in
+`/etc/telegraf/telegraf.d` run
+
+```shell
+# ./tools/custom_builder/custom_builder                      \
+    --config     /etc/telegraf/telegraf.conf \
+    --config-dir /etc/telegraf/telegraf.d
+```
+
+Configurations can also be retrieved from remote locations just
+like for Telegraf.
+
+```shell
+# ./tools/custom_builder/custom_builder --config http://myserver/telegraf.conf
+```
+
+will download the configuration from `myserver`.
+
+The `--config` and `--config-dir` option can be used multiple times.
+In case you want to deploy Telegraf to multiple systems with
+different configurations, simply specify the super-set of all
+configurations you have. `custom_builder` will figure out the list
+for you
+
+```shell
+# ./tools/custom_builder/custom_builder             \
+    --config system1/telegraf.conf  \
+    --config system2/telegraf.conf  \
+    --config ...                    \
+    --config systemN/telegraf.conf  \
+    --config-dir system1/telegraf.d \
+    --config-dir system2/telegraf.d \
+    --config-dir ...                \
+    --config-dir systemN/telegraf.d
+```
+
+The Telegraf customization uses
+[Golang's build-tags](https://pkg.go.dev/go/build#hdr-Build_Constraints) to
+select the set of plugins. To see which tags are set use the `--tags` flag.
+
+To get more help run
+
+```shell
+# ./tools/custom_builder/custom_builder --help
+```
+
+## Notes
+
+Please make sure to include all `parsers` you intend to use and check the
+enabled-plugins list.
+
+Additional plugins can potentially be enabled automatically due to
+dependencies without being shown in the enabled-plugins list.

tools/custom_builder/config.go

@@ -0,0 +1,153 @@
+package main
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/influxdata/telegraf/config"
+	"github.com/influxdata/toml"
+	"github.com/influxdata/toml/ast"
+)
+
+type pluginState map[string]bool
+type selection map[string]pluginState
+
+func ImportConfigurations(files, dirs []string) (*selection, int, error) {
+	sel := selection(make(map[string]pluginState))
+
+	// Initialize the categories
+	for _, category := range categories {
+		sel[category] = make(map[string]bool)
+	}
+
+	// Gather all configuration files
+	var filenames []string
+	filenames = append(filenames, files...)
+
+	for _, dir := range dirs {
+		// Walk the directory and get the packages
+		elements, err := os.ReadDir(dir)
+		if err != nil {
+			return nil, 0, fmt.Errorf("reading directory %q failed: %w", dir, err)
+		}
+
+		for _, element := range elements {
+			if element.IsDir() || filepath.Ext(element.Name()) != ".conf" {
+				continue
+			}
+
+			filenames = append(filenames, filepath.Join(dir, element.Name()))
+		}
+	}
+	if len(filenames) == 0 {
+		return &sel, 0, errors.New("no configuration files given or found")
+	}
+
+	// Do the actual import
+	err := sel.importFiles(filenames)
+	return &sel, len(filenames), err
+}
+
+func (s *selection) Filter(p packageCollection) (*packageCollection, error) {
+	enabled := packageCollection{
+		packages: map[string][]packageInfo{},
+	}
+
+	for category, pkgs := range p.packages {
+		var categoryEnabledPackages []packageInfo
+		settings := (*s)[category]
+		for _, pkg := range pkgs {
+			if _, found := settings[pkg.Plugin]; found {
+				categoryEnabledPackages = append(categoryEnabledPackages, pkg)
+			}
+		}
+		enabled.packages[category] = categoryEnabledPackages
+	}
+
+	// Make sure we update the list of default parsers used by
+	// the remaining packages
+	enabled.FillDefaultParsers()
+
+	// If the user did not configure any parser, we want to include
+	// the default parsers if any to preserve a functional set of
+	// plugins.
+	if len(enabled.packages["parsers"]) == 0 && len(enabled.defaultParsers) > 0 {
+		var parsers []packageInfo
+		for _, pkg := range p.packages["parsers"] {
+			for _, name := range enabled.defaultParsers {
+				if pkg.Plugin == name {
+					parsers = append(parsers, pkg)
+					break
+				}
+			}
+		}
+		enabled.packages["parsers"] = parsers
+	}
+
+	return &enabled, nil
+}
+
+func (s *selection) importFiles(configurations []string) error {
+	for _, cfg := range configurations {
+		buf, err := config.LoadConfigFile(cfg)
+		if err != nil {
+			return fmt.Errorf("reading %q failed: %v", cfg, err)
+		}
+
+		if err := s.extractPluginsFromConfig(buf); err != nil {
+			return fmt.Errorf("extracting plugins from %q failed: %v", cfg, err)
+		}
+	}
+
+	return nil
+}
+
+func (s *selection) extractPluginsFromConfig(buf []byte) error {
+	table, err := toml.Parse(trimBOM(buf))
+	if err != nil {
+		return fmt.Errorf("parsing TOML failed: %w", err)
+	}
+
+	for category, subtbl := range table.Fields {
+		categoryTbl, ok := subtbl.(*ast.Table)
+		if !ok {
+			continue
+		}
+
+		if _, found := (*s)[category]; !found {
+			continue
+		}
+
+		for name, data := range categoryTbl.Fields {
+			(*s)[category][name] = true
+
+			// We need to check the data_format field to get all required parsers
+			switch category {
+			case "inputs", "processors":
+				pluginTables, ok := data.([]*ast.Table)
+				if !ok {
+					continue
+				}
+				for _, subsubtbl := range pluginTables {
+					for field, fieldData := range subsubtbl.Fields {
+						if field != "data_format" {
+							continue
+						}
+						kv := fieldData.(*ast.KeyValue)
+						name := kv.Value.(*ast.String)
+						(*s)["parsers"][name.Value] = true
+					}
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+func trimBOM(f []byte) []byte {
+	return bytes.TrimPrefix(f, []byte("\xef\xbb\xbf"))
+}

tools/custom_builder/main.go

@@ -0,0 +1,155 @@
+package main
+
+import (
+	"bytes"
+	"flag"
+	"fmt"
+	"log"
+	"os"
+	"os/exec"
+	"strings"
+)
+
+var buildTargets = []string{"build"}
+
+var categories = []string{
+	"aggregators",
+	"inputs",
+	"outputs",
+	"parsers",
+	"processors",
+}
+
+const description = `
+This is a tool build Telegraf with a custom set of plugins. The plugins are
+select according to the specified Telegraf configuration files. This allows
+to shrink the binary size by only selecting the plugins you really need.
+A more detailed documentation is available at
+http://github.com/influxdata/telegraf/tools/custom_builder/README.md
+`
+
+const examples = `
+The following command with customize Telegraf to fit the configuration found
+at the default locations
+
+  custom_builder --config /etc/telegraf/telegraf.conf --config-dir /etc/telegraf/telegraf.d
+
+You can the --config and --config-dir multiple times
+
+  custom_builder --config global.conf --config myinputs.conf --config myoutputs.conf
+
+or use one or more remote address(es) to load the config
+
+  custom_builder --config global.conf --config http://myserver/plugins.conf
+
+Combinations of local and remote config as well as config directories are
+possible.
+`
+
+func usage() {
+	_, _ = fmt.Fprint(flag.CommandLine.Output(), description)
+	_, _ = fmt.Fprintln(flag.CommandLine.Output(), "")
+	_, _ = fmt.Fprintln(flag.CommandLine.Output(), "Usage:")
+	_, _ = fmt.Fprintln(flag.CommandLine.Output(), "  custom_builder [flags]")
+	_, _ = fmt.Fprintln(flag.CommandLine.Output(), "")
+	_, _ = fmt.Fprintln(flag.CommandLine.Output(), "Flags:")
+	flag.PrintDefaults()
+	_, _ = fmt.Fprintln(flag.CommandLine.Output(), "")
+	_, _ = fmt.Fprintln(flag.CommandLine.Output(), "Examples:")
+	_, _ = fmt.Fprint(flag.CommandLine.Output(), examples)
+	_, _ = fmt.Fprintln(flag.CommandLine.Output(), "")
+}
+
+func main() {
+	var dryrun, showtags, quiet bool
+	var configFiles, configDirs []string
+
+	flag.Func("config",
+		"Import plugins from configuration file (can be used multiple times)",
+		func(s string) error {
+			configFiles = append(configFiles, s)
+			return nil
+		},
+	)
+	flag.Func("config-dir",
+		"Import plugins from configs in the given directory (can be used multiple times)",
+		func(s string) error {
+			configDirs = append(configDirs, s)
+			return nil
+		},
+	)
+	flag.BoolVar(&dryrun, "dry-run", false, "Skip the actual building step")
+	flag.BoolVar(&quiet, "quiet", false, "Print fewer log messages")
+	flag.BoolVar(&showtags, "tags", false, "Show build-tags used")
+
+	flag.Usage = usage
+	flag.Parse()
+
+	// Check configuration options
+	if len(configFiles) == 0 && len(configDirs) == 0 {
+		log.Fatalln("No configuration specified!")
+	}
+
+	// Import the plugin list from Telegraf configuration files
+	log.Println("Importing configuration file(s)...")
+	cfg, nfiles, err := ImportConfigurations(configFiles, configDirs)
+	if err != nil {
+		log.Fatalf("Importing configuration(s) failed: %v", err)
+	}
+	if !quiet {
+		log.Printf("Found %d configuration files...", nfiles)
+	}
+
+	// Check if we do have a config
+	if nfiles == 0 {
+		log.Fatalln("No configuration files loaded!")
+	}
+
+	// Collect all available plugins
+	packages := packageCollection{}
+	if err := packages.CollectAvailable(); err != nil {
+		log.Fatalf("Collecting plugins failed: %v", err)
+	}
+
+	// Process the plugin list with the given config. This will
+	// only keep the plugins that adhere to the filtering criteria.
+	enabled, err := cfg.Filter(packages)
+	if err != nil {
+		log.Fatalf("Filtering plugins failed: %v", err)
+	}
+	if !quiet {
+		enabled.Print()
+	}
+
+	// Extract the build-tags
+	tagset := enabled.ExtractTags()
+	if len(tagset) == 0 {
+		log.Fatalln("Nothing selected!")
+	}
+	tags := "custom," + strings.Join(tagset, ",")
+	if showtags {
+		fmt.Printf("Build tags: %s\n", tags)
+	}
+
+	if !dryrun {
+		// Perform the build
+		var out bytes.Buffer
+		makeCmd := exec.Command("make", buildTargets...)
+		makeCmd.Env = append(os.Environ(), "BUILDTAGS="+tags)
+		makeCmd.Stdout = &out
+		makeCmd.Stderr = &out
+
+		if !quiet {
+			log.Println("Running build...")
+		}
+		if err := makeCmd.Run(); err != nil {
+			fmt.Println(out.String())
+			log.Fatalf("Running make failed: %v", err)
+		}
+		if !quiet {
+			fmt.Println(out.String())
+		}
+	} else if !quiet {
+		log.Println("DRY-RUN: Skipping build.")
+	}
+}

tools/custom_builder/packages.go

@@ -0,0 +1,354 @@
+package main
+
+import (
+	"bufio"
+	"errors"
+	"fmt"
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"io/fs"
+	"log"
+	"os"
+	"path/filepath"
+	"regexp"
+	"sort"
+	"strings"
+
+	"github.com/influxdata/telegraf/filter"
+)
+
+// Define the categories we can handle and package filters
+var packageFilter = filter.MustCompile([]string{
+	"*/all",
+	"*/*_test",
+	"inputs/example",
+	"inputs/main",
+})
+
+type packageInfo struct {
+	Category      string
+	Plugin        string
+	Path          string
+	Tag           string
+	DefaultParser string
+}
+
+type packageCollection struct {
+	packages       map[string][]packageInfo
+	defaultParsers []string
+}
+
+// Define the package exceptions
+var exceptions = map[string][]packageInfo{
+	"parsers": {
+		{
+			Category: "parsers",
+			Plugin:   "influx_upstream",
+			Path:     "plugins/parsers/influx/influx_upstream",
+			Tag:      "parsers.influx",
+		},
+	},
+	"processors": {
+		{
+			Category: "processors",
+			Plugin:   "aws_ec2",
+			Path:     "plugins/processors/aws/ec2",
+			Tag:      "processors.aws_ec2",
+		},
+	},
+}
+
+func (p *packageCollection) collectPackagesForCategory(category string) error {
+	var entries []packageInfo
+	pluginDir := filepath.Join("plugins", category)
+
+	// Add exceptional packages if any
+	if pkgs, found := exceptions[category]; found {
+		entries = append(entries, pkgs...)
+	}
+
+	// Walk the directory and get the packages
+	elements, err := os.ReadDir(pluginDir)
+	if err != nil {
+		return err
+	}
+
+	for _, element := range elements {
+		path := filepath.Join(pluginDir, element.Name())
+		if !element.IsDir() {
+			continue
+		}
+
+		var fset token.FileSet
+		pkgs, err := parser.ParseDir(&fset, path, sourceFileFilter, parser.ParseComments)
+		if err != nil {
+			log.Printf("parsing directory %q failed: %v", path, err)
+			continue
+		}
+
+		for name, pkg := range pkgs {
+			if packageFilter.Match(category + "/" + name) {
+				continue
+			}
+
+			// Extract the names of the plugins registered by this package
+			registeredNames := extractRegisteredNames(pkg, category)
+			if len(registeredNames) == 0 {
+				log.Printf("WARN: Could not extract information from package %q", name)
+				continue
+			}
+
+			// Extract potential default parsers for input and processor packages
+			var defaultParser string
+			switch category {
+			case "inputs", "processors":
+				var err error
+				defaultParser, err = extractDefaultParser(path)
+				if err != nil {
+					log.Printf("Getting default parser for %s.%s failed: %v", category, name, err)
+				}
+			}
+
+			for _, plugin := range registeredNames {
+				path := filepath.Join("plugins", category, element.Name())
+				tag := category + "." + element.Name()
+				entries = append(entries, packageInfo{
+					Category:      category,
+					Plugin:        plugin,
+					Path:          filepath.ToSlash(path),
+					Tag:           tag,
+					DefaultParser: defaultParser,
+				})
+			}
+		}
+	}
+	p.packages[category] = entries
+
+	return nil
+}
+
+func (p *packageCollection) FillDefaultParsers() {
+	// Make sure we ignore all empty-named parsers which indicate
+	// that there is no parser used by the plugin.
+	parsers := map[string]bool{"": true}
+
+	// Iterate over all plugins that may have parsers and collect
+	// the defaults
+	p.defaultParsers = make([]string, 0)
+	for _, category := range []string{"inputs", "processors"} {
+		for _, pkg := range p.packages[category] {
+			name := pkg.DefaultParser
+			if seen := parsers[name]; seen {
+				continue
+			}
+			p.defaultParsers = append(p.defaultParsers, name)
+			parsers[name] = true
+		}
+	}
+}
+
+func (p *packageCollection) CollectAvailable() error {
+	p.packages = make(map[string][]packageInfo)
+
+	for _, category := range categories {
+		if err := p.collectPackagesForCategory(category); err != nil {
+			return err
+		}
+	}
+
+	p.FillDefaultParsers()
+
+	return nil
+}
+
+func (p *packageCollection) ExtractTags() []string {
+	var tags []string
+	for category, pkgs := range p.packages {
+		_ = category
+		for _, pkg := range pkgs {
+			tags = append(tags, pkg.Tag)
+		}
+	}
+	sort.Strings(tags)
+
+	return tags
+}
+
+func (p *packageCollection) Print() {
+	fmt.Println("-------------------------------------------------------------------------------")
+	fmt.Println("Enabled plugins:")
+	fmt.Println("-------------------------------------------------------------------------------")
+	for _, category := range categories {
+		pkgs := p.packages[category]
+		sort.Slice(pkgs, func(i, j int) bool { return pkgs[i].Plugin < pkgs[j].Plugin })
+
+		fmt.Printf("%s (%d):\n", category, len(pkgs))
+		for _, pkg := range pkgs {
+			fmt.Printf("  %-30s  %s\n", pkg.Plugin, pkg.Path)
+		}
+		fmt.Println("-------------------------------------------------------------------------------")
+	}
+}
+
+func sourceFileFilter(d fs.FileInfo) bool {
+	return strings.HasSuffix(d.Name(), ".go") && !strings.HasSuffix(d.Name(), "_test.go")
+}
+
+func findFunctionDecl(file *ast.File, name string) *ast.FuncDecl {
+	for _, decl := range file.Decls {
+		d, ok := decl.(*ast.FuncDecl)
+		if !ok {
+			continue
+		}
+		if d.Name.Name == name && d.Recv == nil {
+			return d
+		}
+	}
+	return nil
+}
+
+func findAddStatements(decl *ast.FuncDecl, pluginType string) []*ast.CallExpr {
+	var statements []*ast.CallExpr
+	for _, stmt := range decl.Body.List {
+		s, ok := stmt.(*ast.ExprStmt)
+		if !ok {
+			continue
+		}
+		call, ok := s.X.(*ast.CallExpr)
+		if !ok {
+			continue
+		}
+		fun, ok := call.Fun.(*ast.SelectorExpr)
+		if !ok {
+			continue
+		}
+		e, ok := fun.X.(*ast.Ident)
+		if !ok {
+			continue
+		}
+		if e.Name == pluginType && (fun.Sel.Name == "Add" || fun.Sel.Name == "AddStreaming") {
+			statements = append(statements, call)
+		}
+	}
+
+	return statements
+}
+
+func extractPluginInfo(file *ast.File, pluginType string, declarations map[string]string) ([]string, error) {
+	var registeredNames []string
+
+	decl := findFunctionDecl(file, "init")
+	if decl == nil {
+		return nil, nil
+	}
+	calls := findAddStatements(decl, pluginType)
+	if len(calls) == 0 {
+		return nil, nil
+	}
+	for _, call := range calls {
+		switch arg := call.Args[0].(type) {
+		case *ast.Ident:
+			resval, found := declarations[arg.Name]
+			if !found {
+				return nil, fmt.Errorf("cannot resolve registered name variable %q", arg.Name)
+			}
+			registeredNames = append(registeredNames, strings.Trim(resval, "\""))
+		case *ast.BasicLit:
+			if arg.Kind != token.STRING {
+				return nil, errors.New("registered name is not a string")
+			}
+			registeredNames = append(registeredNames, strings.Trim(arg.Value, "\""))
+		default:
+			return nil, fmt.Errorf("unhandled argument type: %v (%T)", arg, arg)
+		}
+	}
+	return registeredNames, nil
+}
+
+func extractPackageDeclarations(pkg *ast.Package) map[string]string {
+	declarations := make(map[string]string)
+
+	for _, file := range pkg.Files {
+		for _, d := range file.Decls {
+			gendecl, ok := d.(*ast.GenDecl)
+			if !ok {
+				continue
+			}
+			for _, spec := range gendecl.Specs {
+				spec, ok := spec.(*ast.ValueSpec)
+				if !ok {
+					continue
+				}
+				for _, id := range spec.Names {
+					valspec, ok := id.Obj.Decl.(*ast.ValueSpec)
+					if !ok || len(valspec.Values) != 1 {
+						continue
+					}
+					valdecl, ok := valspec.Values[0].(*ast.BasicLit)
+					if !ok || valdecl.Kind != token.STRING {
+						continue
+					}
+					declarations[id.Name] = strings.Trim(valdecl.Value, "\"")
+				}
+			}
+		}
+	}
+	return declarations
+}
+
+func extractRegisteredNames(pkg *ast.Package, pluginType string) []string {
+	var registeredNames []string
+
+	// Extract all declared variables of all files. This might be necessary when
+	// using references across multiple files
+	declarations := extractPackageDeclarations(pkg)
+
+	// Find the registry Add statement and extract all registered names
+	for fn, file := range pkg.Files {
+		names, err := extractPluginInfo(file, pluginType, declarations)
+		if err != nil {
+			log.Printf("%q error: %v", fn, err)
+			continue
+		}
+		registeredNames = append(registeredNames, names...)
+	}
+	return registeredNames
+}
+
+func extractDefaultParser(pluginDir string) (string, error) {
+	re := regexp.MustCompile(`^\s*#?\s*data_format\s*=\s*"(.*)"\s*$`)
+
+	// Exception for exec which uses JSON by default
+	if filepath.Base(pluginDir) == "exec" {
+		return "json", nil
+	}
+
+	// Walk all config files in the package directory
+	elements, err := os.ReadDir(pluginDir)
+	if err != nil {
+		return "", err
+	}
+
+	for _, element := range elements {
+		path := filepath.Join(pluginDir, element.Name())
+		if element.IsDir() || filepath.Ext(element.Name()) != ".conf" {
+			continue
+		}
+
+		// Read the config and search for a "data_format" entry
+		file, err := os.Open(path)
+		if err != nil {
+			return "", err
+		}
+		scanner := bufio.NewScanner(file)
+		for scanner.Scan() {
+			match := re.FindStringSubmatch(scanner.Text())
+			if len(match) == 2 {
+				return match[1], nil
+			}
+		}
+	}
+
+	return "", nil
+}