splunk inputs configuration precedence

.chloggen/splunk-inputs-precedence.yaml

@@ -0,0 +1,14 @@
+change_type: enhancement
+
+component: splunk_inputs
+
+note: "add Splunk btool config precedence across system and `splunk_ta_*` app directories"
+
+issues: [117]
+
+subtext: |
+  The `base_dir` config field (or `$SPLUNK_HOME` env var) now points to the Splunk
+  installation root. The receiver discovers all `splunk_ta_*` directories automatically
+  and merges `inputs.conf`, `transforms.conf`, and `props.conf` per TA using standard
+  btool precedence — later layers win per key. Unsupported input and output stanza
+  kinds are now skipped silently with an info log instead of returning an error.

internal/collector/collector.go

@@ -36,15 +36,16 @@ func Run(baseDir string, cfg *config.Config) (func(), error) {
 	if err != nil {
 		return nil, err
 	}
-	inputs, err := tabuilder.ReadInputs(baseDir)
+	dirs := tabuilder.ConfDirs(baseDir)
+	inputs, err := tabuilder.ReadInputs(dirs)
 	if err != nil {
 		return nil, err
 	}
-	transforms, err := tabuilder.ReadTransforms(baseDir)
+	transforms, err := tabuilder.ReadTransforms(dirs)
 	if err != nil {
 		return nil, err
 	}
-	props, err := tabuilder.ReadProps(baseDir)
+	props, err := tabuilder.ReadProps(dirs)
 	if err != nil {
 		return nil, err
 	}

internal/collector/collector_test.go

@@ -37,7 +37,7 @@ func TestRunTA(t *testing.T) {
 	defer func() {
 		_ = rcvr.Shutdown(context.Background())
 	}()
-	cancel, err := Run(filepath.Join("testdata", "ta"), &config.Config{
+	cancel, err := Run(filepath.Join("testdata", "ta_home"), &config.Config{
 		Type:     "otlp_http",
 		Endpoint: "http://localhost:1337",
 	})
@@ -60,7 +60,7 @@ func TestRunPeriodic(t *testing.T) {
 	defer func() {
 		_ = rcvr.Shutdown(context.Background())
 	}()
-	cancel, err := Run(filepath.Join("testdata", "periodic"), &config.Config{
+	cancel, err := Run(filepath.Join("testdata", "periodic_home"), &config.Config{
 		Type:     "otlp_http",
 		Endpoint: "http://localhost:1338",
 	})
@@ -101,7 +101,7 @@ func TestRunDisabled(t *testing.T) {
 	defer func() {
 		_ = rcvr.Shutdown(context.Background())
 	}()
-	cancel, err := Run(filepath.Join("testdata", "disabled"), &config.Config{
+	cancel, err := Run(filepath.Join("testdata", "disabled_home"), &config.Config{
 		Type:     "otlp_http",
 		Endpoint: "http://localhost:1339",
 	})
@@ -123,7 +123,7 @@ func TestRunDisabledInterval(t *testing.T) {
 	defer func() {
 		_ = rcvr.Shutdown(context.Background())
 	}()
-	cancel, err := Run(filepath.Join("testdata", "disabled_interval"), &config.Config{
+	cancel, err := Run(filepath.Join("testdata", "disabled_interval_home"), &config.Config{
 		Type:     "otlp_http",
 		Endpoint: "http://localhost:1340",
 	})
@@ -144,7 +144,7 @@ func TestRunScriptedInputs(t *testing.T) {
 	defer func() {
 		_ = rcvr.Shutdown(context.Background())
 	}()
-	cancel, err := Run(filepath.Join("testdata", "script"), &config.Config{
+	cancel, err := Run(filepath.Join("testdata", "script_home"), &config.Config{
 		Type:     "otlp_http",
 		Endpoint: "http://localhost:1341",
 	})
@@ -157,7 +157,7 @@ func TestRunScriptedInputs(t *testing.T) {
 }
 
 func TestUseTCP(t *testing.T) {
-	rootDir := filepath.Join("testdata", "tcp")
+	rootDir := filepath.Join("testdata", "tcp_home")
 	logsSink := &consumertest.LogsSink{}
 	cfg := otlpreceiver.NewFactory().CreateDefaultConfig().(*otlpreceiver.Config)
 	cfg.Protocols.HTTP.GetOrInsertDefault().ServerConfig.NetAddr.Endpoint = "localhost:1342"
@@ -190,7 +190,7 @@ func TestUseTCP(t *testing.T) {
 }
 
 func TestUseUDP(t *testing.T) {
-	rootDir := filepath.Join("testdata", "udp")
+	rootDir := filepath.Join("testdata", "udp_home")
 	logsSink := &consumertest.LogsSink{}
 	cfg := otlpreceiver.NewFactory().CreateDefaultConfig().(*otlpreceiver.Config)
 	cfg.Protocols.HTTP.GetOrInsertDefault().ServerConfig.NetAddr.Endpoint = "localhost:1343"
@@ -232,7 +232,7 @@ func TestRunScriptedInputsWithHEC(t *testing.T) {
 	defer func() {
 		_ = rcvr.Shutdown(context.Background())
 	}()
-	cancel, err := Run(filepath.Join("testdata", "script"), &config.Config{
+	cancel, err := Run(filepath.Join("testdata", "script_home"), &config.Config{
 		Endpoint: "http://localhost:1341",
 		Token:    "foo",
 	})

internal/conf/inputs.go

@@ -20,10 +20,17 @@ type Input struct {
 	ServerURI     string        `xml:"server_uri"`
 	SessionKey    string        `xml:"session_key"`
 	CheckpointDir string        `xml:"checkpoint_dir"`
+	AppDir        string        `xml:"-"`
 	Configuration Configuration `xml:"configuration"`
 }
 
-func ReadInput(payload []byte) ([]Input, error) {
+// IsDisabled reports whether the stanza has disabled=1.
+func (s *Stanza) IsDisabled() bool {
+	p := s.Params.Get("disabled")
+	return p != nil && p.Value == "1"
+}
+
+func ReadInput(payload []byte, appDir string) ([]Input, error) {
 	f, err := ini.Load(payload)
 	if err != nil {
 		return nil, err
@@ -35,6 +42,7 @@ func ReadInput(payload []byte) ([]Input, error) {
 			continue // disregard default section. We need a stanza per input.
 		}
 		i := Input{
+			AppDir: appDir,
 			Configuration: Configuration{
 				Stanza: Stanza{
 					Name:   section.Name(),
@@ -58,6 +66,43 @@ func ReadInput(payload []byte) ([]Input, error) {
 	return result, nil
 }
 
+// MergeInputs merges layered inputs; later layers take precedence per param key.
+func MergeInputs(layers [][]Input) []Input {
+	seen := make(map[string]int)
+	var result []Input
+	for _, layer := range layers {
+		for _, input := range layer {
+			name := input.Configuration.Stanza.Name
+			if idx, ok := seen[name]; ok {
+				result[idx] = mergeInput(result[idx], input)
+			} else {
+				seen[name] = len(result)
+				result = append(result, input)
+			}
+		}
+	}
+	return result
+}
+
+func mergeInput(base, override Input) Input {
+	merged := base
+	params := make(map[string]int, len(base.Configuration.Stanza.Params))
+	mergedParams := append([]Param{}, base.Configuration.Stanza.Params...)
+	for i, p := range mergedParams {
+		params[p.Name] = i
+	}
+	for _, p := range override.Configuration.Stanza.Params {
+		if idx, ok := params[p.Name]; ok {
+			mergedParams[idx] = p
+		} else {
+			params[p.Name] = len(mergedParams)
+			mergedParams = append(mergedParams, p)
+		}
+	}
+	merged.Configuration.Stanza.Params = mergedParams
+	return merged
+}
+
 func (i *Input) ToXML() ([]byte, error) {
 	b, err := xml.MarshalIndent(i, "", "  ")
 	return append([]byte(xmlDeclaration), b...), err

internal/conf/inputs_test.go

@@ -15,7 +15,7 @@ import (
 func TestOneInput(t *testing.T) {
 	b, err := os.ReadFile(filepath.Join("testdata", "oneinput.conf"))
 	require.NoError(t, err)
-	res, err := ReadInput(b)
+	res, err := ReadInput(b, "")
 	require.NoError(t, err)
 	assert.Equal(
 		t,
@@ -50,7 +50,7 @@ func TestOneInput(t *testing.T) {
 func TestTwoInputs(t *testing.T) {
 	b, err := os.ReadFile(filepath.Join("testdata", "twoinputs.conf"))
 	require.NoError(t, err)
-	res, err := ReadInput(b)
+	res, err := ReadInput(b, "")
 	require.NoError(t, err)
 	assert.Equal(t, []Input{{
 		ServerHost:    "",
@@ -112,6 +112,38 @@ func TestTwoInputs(t *testing.T) {
 	}}, res)
 }
 
+func TestMergeInputsPartialOverride(t *testing.T) {
+	base := []Input{{Configuration: Configuration{Stanza: Stanza{
+		Name:   "monitor:///var/log/syslog",
+		Params: Params{{Name: "sourcetype", Value: "syslog"}, {Name: "index", Value: "main"}},
+	}}}}
+	override := []Input{{Configuration: Configuration{Stanza: Stanza{
+		Name:   "monitor:///var/log/syslog",
+		Params: Params{{Name: "index", Value: "override"}},
+	}}}}
+
+	merged := MergeInputs([][]Input{base, override})
+	require.Len(t, merged, 1)
+	assert.Equal(t, "syslog", merged[0].Configuration.Stanza.Params.Get("sourcetype").Value)
+	assert.Equal(t, "override", merged[0].Configuration.Stanza.Params.Get("index").Value)
+}
+
+func TestMergeInputsFullOverride(t *testing.T) {
+	base := []Input{{Configuration: Configuration{Stanza: Stanza{
+		Name:   "monitor:///var/log/syslog",
+		Params: Params{{Name: "sourcetype", Value: "syslog"}, {Name: "index", Value: "main"}},
+	}}}}
+	override := []Input{{Configuration: Configuration{Stanza: Stanza{
+		Name:   "monitor:///var/log/syslog",
+		Params: Params{{Name: "sourcetype", Value: "sourcetype_override"}, {Name: "index", Value: "index_override"}},
+	}}}}
+
+	merged := MergeInputs([][]Input{base, override})
+	require.Len(t, merged, 1)
+	assert.Equal(t, "sourcetype_override", merged[0].Configuration.Stanza.Params.Get("sourcetype").Value)
+	assert.Equal(t, "index_override", merged[0].Configuration.Stanza.Params.Get("index").Value)
+}
+
 func TestToXML(t *testing.T) {
 	testStr := `<?xml version="1.0" encoding="UTF-8"?>
 <Input>
@@ -133,7 +165,7 @@ func TestToXML(t *testing.T) {
 </Input>`
 	b, err := os.ReadFile(filepath.Join("testdata", "oneinput.conf"))
 	require.NoError(t, err)
-	res, err := ReadInput(b)
+	res, err := ReadInput(b, "")
 	require.NoError(t, err)
 	assert.Len(t, res, 1)
 	res[0].ServerHost = "773c28971b2a"

internal/conf/props.go

@@ -60,6 +60,26 @@ func (p *Prop) Type() PropType {
 	}
 }
 
+// MergeProps merges multiple slices of props, with later slices taking
+// precedence. Stanzas are keyed by name; the last definition wins.
+// The merged result is re-ordered by specificity.
+func MergeProps(layers [][]Prop) []Prop {
+	seen := make(map[string]int)
+	var result []Prop
+	for _, layer := range layers {
+		for _, p := range layer {
+			if idx, ok := seen[p.Name]; ok {
+				result[idx] = p
+			} else {
+				seen[p.Name] = len(result)
+				result = append(result, p)
+			}
+		}
+	}
+	orderProps(result)
+	return result
+}
+
 func ReadProps(payload []byte) ([]Prop, error) {
 	f, err := ini.Load(payload)
 	if err != nil {

internal/conf/transforms.go

@@ -11,6 +11,24 @@ type Transform struct {
 	Format string
 }
 
+// MergeTransforms merges multiple slices of transforms, with later slices
+// taking precedence. Stanzas are keyed by name; the last definition wins.
+func MergeTransforms(layers [][]Transform) []Transform {
+	seen := make(map[string]int)
+	var result []Transform
+	for _, layer := range layers {
+		for _, t := range layer {
+			if idx, ok := seen[t.Name]; ok {
+				result[idx] = t
+			} else {
+				seen[t.Name] = len(result)
+				result = append(result, t)
+			}
+		}
+	}
+	return result
+}
+
 func ReadTransforms(payload []byte) ([]Transform, error) {
 	f, err := ini.Load(payload)
 	if err != nil {

internal/receiver/batchreceiver/config.go

@@ -8,9 +8,8 @@ import (
 )
 
 type Config struct {
+	Input      conf.Input       `mapstructure:"-"`
+	BaseDir    string           `mapstructure:"-"`
 	Transforms []conf.Transform `mapstructure:"-"`
 	Props      []conf.Prop      `mapstructure:"-"`
-
-	BaseDir string     `mapstructure:"-"`
-	Input   conf.Input `mapstructure:"-"`
 }

internal/receiver/monitorreceiver/config.go

@@ -8,9 +8,8 @@ import (
 )
 
 type Config struct {
+	Input      conf.Input       `mapstructure:"-"`
+	BaseDir    string           `mapstructure:"-"`
 	Transforms []conf.Transform `mapstructure:"-"`
 	Props      []conf.Prop      `mapstructure:"-"`
-
-	BaseDir string     `mapstructure:"-"`
-	Input   conf.Input `mapstructure:"-"`
 }

internal/receiver/scriptreceiver/config.go

@@ -6,8 +6,8 @@ package scriptreceiver
 import "github.com/splunk/tarunner/internal/conf"
 
 type Config struct {
+	conf.Input `mapstructure:"-"`
 	BaseDir    string           `mapstructure:"-"`
 	Props      []conf.Prop      `mapstructure:"-"`
 	Transforms []conf.Transform `mapstructure:"-"`
-	conf.Input `mapstructure:"-"`
 }

internal/receiver/tcpreceiver/config.go

@@ -9,11 +9,10 @@ import (
 )
 
 type Config struct {
+	Input      conf.Input       `mapstructure:"-"`
+	BaseDir    string           `mapstructure:"-"`
 	Transforms []conf.Transform `mapstructure:"-"`
 	Props      []conf.Prop      `mapstructure:"-"`
-
-	BaseDir string     `mapstructure:"-"`
-	Input   conf.Input `mapstructure:"-"`
 }
 
 func (cfg *Config) Validate() error {

internal/receiver/udpreceiver/config.go

@@ -9,11 +9,10 @@ import (
 )
 
 type Config struct {
+	Input      conf.Input       `mapstructure:"-"`
+	BaseDir    string           `mapstructure:"-"`
 	Transforms []conf.Transform `mapstructure:"-"`
 	Props      []conf.Prop      `mapstructure:"-"`
-
-	BaseDir string     `mapstructure:"-"`
-	Input   conf.Input `mapstructure:"-"`
 }
 
 func (cfg *Config) Validate() error {

internal/receiver/wineventlogreceiver/config.go

@@ -8,9 +8,8 @@ import (
 )
 
 type Config struct {
+	Input      conf.Input       `mapstructure:"-"`
+	BaseDir    string           `mapstructure:"-"`
 	Transforms []conf.Transform `mapstructure:"-"`
 	Props      []conf.Prop      `mapstructure:"-"`
-
-	BaseDir string     `mapstructure:"-"`
-	Input   conf.Input `mapstructure:"-"`
 }

internal/script/command.go

@@ -18,13 +18,23 @@ func DetermineCommandName(baseDir string, input conf.Input) (string, error) {
 	if err != nil {
 		return "", err
 	}
+	resolveDir := baseDir
+	if input.AppDir != "" {
+		resolveDir = input.AppDir
+	}
 	switch parsed.Kind {
 	case "monitor", "batch":
 		return parsed.Target, nil
 	case "script":
-		return GetPath(baseDir, parsed.Target)
+		if filepath.IsAbs(parsed.Target) {
+			return parsed.Target, nil
+		}
+		return GetPath(resolveDir, parsed.Target)
 	case "":
-		return GetPath(baseDir, filepath.Join("bin", fmt.Sprintf("%s_%s", runtime.GOOS, runtime.GOARCH), parsed.Target))
+		if filepath.IsAbs(parsed.Target) {
+			return parsed.Target, nil
+		}
+		return GetPath(resolveDir, filepath.Join("bin", fmt.Sprintf("%s_%s", runtime.GOOS, runtime.GOARCH), parsed.Target))
 	default:
 		return "", fmt.Errorf("unknown scheme %q", parsed.Kind)
 	}