Move the input blocks to an inputs map

docs/resources/fleet_integration_policy.md

@@ -103,7 +103,7 @@ resource "elasticstack_fleet_integration_policy" "sample" {
 - `description` (String) The description of the integration policy.
 - `enabled` (Boolean) Enable the integration policy.
 - `force` (Boolean) Force operations, such as creation and deletion, to occur.
-- `input` (Block List) Integration inputs. (see [below for nested schema](#nestedblock--input))
+- `inputs` (Attributes Map) Integration inputs mapped by input ID. (see [below for nested schema](#nestedatt--inputs))
 - `output_id` (String) The ID of the output to send data to. When not specified, the default output of the agent policy will be used.
 - `policy_id` (String) Unique identifier of the integration policy.
 - `space_ids` (Set of String) The Kibana space IDs where this integration policy is available. When set, must match the space_ids of the referenced agent policy. If not set, will be inherited from the agent policy. Note: The order of space IDs does not matter as this is a set.
@@ -113,18 +113,22 @@ resource "elasticstack_fleet_integration_policy" "sample" {
 
 - `id` (String) The ID of this resource.
 
-<a id="nestedblock--input"></a>
-### Nested Schema for `input`
+<a id="nestedatt--inputs"></a>
+### Nested Schema for `inputs`
 
-Required:
+Optional:
+
+- `enabled` (Boolean) Enable the input.
+- `streams` (Attributes Map) Input streams mapped by stream ID. (see [below for nested schema](#nestedatt--inputs--streams))
+- `vars` (String, Sensitive) Input-level variables as JSON.
 
-- `input_id` (String) The identifier of the input.
+<a id="nestedatt--inputs--streams"></a>
+### Nested Schema for `inputs.streams`
 
 Optional:
 
-- `enabled` (Boolean) Enable the input.
-- `streams_json` (String, Sensitive) Input streams as JSON.
-- `vars_json` (String, Sensitive) Input variables as JSON.
+- `enabled` (Boolean) Enable the stream.
+- `vars` (String, Sensitive) Stream-level variables as JSON.
 
 ## Import
 

internal/fleet/integration/acc_test.go

@@ -199,22 +199,23 @@ resource "elasticstack_fleet_integration_policy" "sample" {
   integration_name    = elasticstack_fleet_integration.test_integration.name
   integration_version = elasticstack_fleet_integration.test_integration.version
 
-  input {
-    input_id = "tcp-tcp"
-    streams_json = jsonencode({
-      "tcp.generic" : {
-        "enabled" : true,
-        "vars" : {
-          "listen_address" : "localhost",
-          "listen_port" : 8080,
-          "data_stream.dataset" : "tcp.generic",
-          "tags" : [],
-          "syslog_options" : "field: message\n#format: auto\n#timezone: Local\n",
-          "ssl" : "#certificate: |\n#    -----BEGIN CERTIFICATE-----\n#    ...\n#    -----END CERTIFICATE-----\n#key: |\n#    -----BEGIN PRIVATE KEY-----\n#    ...\n#    -----END PRIVATE KEY-----\n",
-          "custom" : ""
-        }
+  inputs = {
+    "tcp-tcp" = {
+	  streams = {
+	    "tcp.generic" = {
+	      enabled = true,
+		  vars = jsonencode({ 
+		    "listen_address" : "localhost",
+			"listen_port" : 8080,
+		  	"data_stream.dataset" : "tcp.generic",
+			"tags" : [],
+			"syslog_options" : "field: message\n#format: auto\n#timezone: Local\n",
+			"ssl" : "#certificate: |\n#    -----BEGIN CERTIFICATE-----\n#    ...\n#    -----END CERTIFICATE-----\n#key: |\n#    -----BEGIN PRIVATE KEY-----\n#    ...\n#    -----END PRIVATE KEY-----\n",
+			"custom" : ""
+		  })
+		}
       }
-    })
+    }
   }
 }
 `, version, policyName, policyName)

internal/fleet/integration_policy/create.go

@@ -65,7 +65,7 @@ func (r *integrationPolicyResource) Create(ctx context.Context, req resource.Cre
 	}
 
 	// Remember if the user configured input in the plan
-	planHadInput := utils.IsKnown(planModel.Input) && !planModel.Input.IsNull() && len(planModel.Input.Elements()) > 0
+	planHadInput := utils.IsKnown(planModel.Inputs) && !planModel.Inputs.IsNull() && len(planModel.Inputs.Elements()) > 0
 
 	diags = planModel.populateFromAPI(ctx, policy)
 	resp.Diagnostics.Append(diags...)
@@ -76,7 +76,7 @@ func (r *integrationPolicyResource) Create(ctx context.Context, req resource.Cre
 	// If plan didn't have input configured, ensure we don't add it now
 	// This prevents "Provider produced inconsistent result" errors
 	if !planHadInput {
-		planModel.Input = types.ListNull(getInputTypeV1())
+		planModel.Inputs = types.MapNull(getInputsTypes())
 	}
 
 	diags = resp.State.Set(ctx, planModel)

internal/fleet/integration_policy/models.go

@@ -3,7 +3,6 @@ package integration_policy
 import (
 	"context"
 	"fmt"
-	"sort"
 
 	"github.com/elastic/terraform-provider-elasticstack/generated/kbapi"
 	"github.com/elastic/terraform-provider-elasticstack/internal/utils"
@@ -31,16 +30,20 @@ type integrationPolicyModel struct {
 	IntegrationName    types.String         `tfsdk:"integration_name"`
 	IntegrationVersion types.String         `tfsdk:"integration_version"`
 	OutputID           types.String         `tfsdk:"output_id"`
-	Input              types.List           `tfsdk:"input"` //> integrationPolicyInputModel
+	Inputs             types.Map            `tfsdk:"inputs"` //> integrationPolicyInputsModel
 	VarsJson           jsontypes.Normalized `tfsdk:"vars_json"`
 	SpaceIds           types.Set            `tfsdk:"space_ids"`
 }
 
-type integrationPolicyInputModel struct {
-	InputID     types.String         `tfsdk:"input_id"`
-	Enabled     types.Bool           `tfsdk:"enabled"`
-	StreamsJson jsontypes.Normalized `tfsdk:"streams_json"`
-	VarsJson    jsontypes.Normalized `tfsdk:"vars_json"`
+type integrationPolicyInputsModel struct {
+	Enabled types.Bool           `tfsdk:"enabled"`
+	Vars    jsontypes.Normalized `tfsdk:"vars"`
+	Streams types.Map            `tfsdk:"streams"` //> integrationPolicyInputStreamModel
+}
+
+type integrationPolicyInputStreamModel struct {
+	Enabled types.Bool           `tfsdk:"enabled"`
+	Vars    jsontypes.Normalized `tfsdk:"vars"`
 }
 
 func (model *integrationPolicyModel) populateFromAPI(ctx context.Context, data *kbapi.PackagePolicy) diag.Diagnostics {
@@ -107,20 +110,19 @@ func (model *integrationPolicyModel) populateFromAPI(ctx context.Context, data *
 		model.SpaceIds = types.SetNull(types.StringType)
 	}
 	// If originally set but API didn't return it, keep the original value
-
-	model.populateInputFromAPI(ctx, data.Inputs, &diags)
+	model.populateInputsFromAPI(ctx, data.Inputs, &diags)
 
 	return diags
 }
 
-func (model *integrationPolicyModel) populateInputFromAPI(ctx context.Context, inputs map[string]kbapi.PackagePolicyInput, diags *diag.Diagnostics) {
+func (model *integrationPolicyModel) populateInputsFromAPI(ctx context.Context, inputs map[string]kbapi.PackagePolicyInput, diags *diag.Diagnostics) {
 	// Handle input population based on context:
 	// 1. If model.Input is unknown: we're importing or reading fresh state → populate from API
 	// 2. If model.Input is known and null/empty: user explicitly didn't configure inputs → don't populate (avoid inconsistent state)
 	// 3. If model.Input is known and has values: user configured inputs → populate from API
 
-	isInputKnown := utils.IsKnown(model.Input)
-	isInputNullOrEmpty := model.Input.IsNull() || (isInputKnown && len(model.Input.Elements()) == 0)
+	isInputKnown := utils.IsKnown(model.Inputs)
+	isInputNullOrEmpty := model.Inputs.IsNull() || (isInputKnown && len(model.Inputs.Elements()) == 0)
 
 	// Case 1: Unknown (import/fresh read) - always populate
 	if !isInputKnown {
@@ -129,32 +131,43 @@ func (model *integrationPolicyModel) populateInputFromAPI(ctx context.Context, i
 	} else if isInputNullOrEmpty {
 		// Case 2: Known and null/empty - user explicitly didn't configure inputs
 		// Don't populate to avoid "Provider produced inconsistent result" error
-		model.Input = types.ListNull(getInputTypeV1())
+		model.Inputs = types.MapNull(getInputsTypes())
 		return
 	}
 	// Case 3: Known and not null/empty - user configured inputs, populate from API (continue below)
 
-	newInputs := utils.TransformMapToSlice(ctx, inputs, path.Root("input"), diags,
-		func(inputData kbapi.PackagePolicyInput, meta utils.MapMeta) integrationPolicyInputModel {
-			return integrationPolicyInputModel{
-				InputID:     types.StringValue(meta.Key),
-				Enabled:     types.BoolPointerValue(inputData.Enabled),
-				StreamsJson: utils.MapToNormalizedType(utils.Deref(inputData.Streams), meta.Path.AtName("streams_json"), diags),
-				VarsJson:    utils.MapToNormalizedType(utils.Deref(inputData.Vars), meta.Path.AtName("vars_json"), diags),
-			}
-		})
-	if newInputs == nil {
-		model.Input = types.ListNull(getInputTypeV1())
-	} else {
-		oldInputs := utils.ListTypeAs[integrationPolicyInputModel](ctx, model.Input, path.Root("input"), diags)
+	newInputs := make(map[string]integrationPolicyInputsModel)
+	for inputID, inputData := range inputs {
+		inputModel := integrationPolicyInputsModel{
+			Enabled: types.BoolPointerValue(inputData.Enabled),
+			Vars:    utils.MapToNormalizedType(utils.Deref(inputData.Vars), path.Root("inputs").AtMapKey(inputID).AtName("vars"), diags),
+		}
 
-		sortInputs(newInputs, oldInputs)
+		// Populate streams
+		if inputData.Streams != nil && len(*inputData.Streams) > 0 {
+			streams := make(map[string]integrationPolicyInputStreamModel)
+			for streamID, streamData := range *inputData.Streams {
+				streamModel := integrationPolicyInputStreamModel{
+					Enabled: types.BoolPointerValue(streamData.Enabled),
+					Vars:    utils.MapToNormalizedType(utils.Deref(streamData.Vars), path.Root("inputs").AtMapKey(inputID).AtName("streams").AtMapKey(streamID).AtName("vars"), diags),
+				}
 
-		inputList, d := types.ListValueFrom(ctx, getInputTypeV1(), newInputs)
-		diags.Append(d...)
+				streams[streamID] = streamModel
+			}
+
+			streamsMap, d := types.MapValueFrom(ctx, getInputStreamType(), streams)
+			diags.Append(d...)
+			inputModel.Streams = streamsMap
+		} else {
+			inputModel.Streams = types.MapNull(getInputStreamType())
+		}
 
-		model.Input = inputList
+		newInputs[inputID] = inputModel
 	}
+
+	inputsMap, d := types.MapValueFrom(ctx, getInputsTypes(), newInputs)
+	diags.Append(d...)
+	model.Inputs = inputsMap
 }
 
 func (model integrationPolicyModel) toAPIModel(ctx context.Context, isUpdate bool, feat features) (kbapi.PackagePolicyRequest, diag.Diagnostics) {
@@ -218,51 +231,53 @@ func (model integrationPolicyModel) toAPIModel(ctx context.Context, isUpdate boo
 		body.Id = model.ID.ValueStringPointer()
 	}
 
-	body.Inputs = utils.MapRef(utils.ListTypeToMap(ctx, model.Input, path.Root("input"), &diags,
-		func(inputModel integrationPolicyInputModel, meta utils.ListMeta) (string, kbapi.PackagePolicyRequestInput) {
-			return inputModel.InputID.ValueString(), kbapi.PackagePolicyRequestInput{
-				Enabled: inputModel.Enabled.ValueBoolPointer(),
-				Streams: utils.MapRef(utils.NormalizedTypeToMap[kbapi.PackagePolicyRequestInputStream](inputModel.StreamsJson, meta.Path.AtName("streams_json"), &diags)),
-				Vars:    utils.MapRef(utils.NormalizedTypeToMap[any](inputModel.VarsJson, meta.Path.AtName("vars_json"), &diags)),
-			}
-		}))
+	if utils.IsKnown(model.Inputs) && len(model.Inputs.Elements()) > 0 {
+		// Use the 'inputs' attribute (v2 format)
+		body.Inputs = utils.MapRef(model.toAPIInputsFromInputsAttribute(ctx, &diags))
+	}
 
 	// Note: space_ids is read-only for integration policies and inherited from the agent policy
 
 	return body, diags
 }
 
-// sortInputs will sort the 'incoming' list of input definitions based on
-// the order of inputs defined in the 'existing' list. Inputs not present in
-// 'existing' will be placed at the end of the list. Inputs are identified by
-// their ID ('input_id'). The 'incoming' slice will be sorted in-place.
-func sortInputs(incoming []integrationPolicyInputModel, existing []integrationPolicyInputModel) {
-	if len(existing) == 0 {
-		sort.Slice(incoming, func(i, j int) bool {
-			return incoming[i].InputID.ValueString() < incoming[j].InputID.ValueString()
-		})
-		return
+// toAPIInputsFromInputsAttribute converts the 'inputs' attribute to the API model format
+func (model integrationPolicyModel) toAPIInputsFromInputsAttribute(ctx context.Context, diags *diag.Diagnostics) map[string]kbapi.PackagePolicyRequestInput {
+	if !utils.IsKnown(model.Inputs) {
+		return nil
 	}
 
-	idToIndex := make(map[string]int, len(existing))
-	for index, inputData := range existing {
-		inputID := inputData.InputID.ValueString()
-		idToIndex[inputID] = index
+	inputsMap := utils.MapTypeAs[integrationPolicyInputsModel](ctx, model.Inputs, path.Root("inputs"), diags)
+	if inputsMap == nil {
+		return nil
 	}
 
-	sort.Slice(incoming, func(i, j int) bool {
-		iID := incoming[i].InputID.ValueString()
-		iIdx, ok := idToIndex[iID]
-		if !ok {
-			return false
+	result := make(map[string]kbapi.PackagePolicyRequestInput, len(inputsMap))
+	for inputID, inputModel := range inputsMap {
+		inputPath := path.Root("inputs").AtMapKey(inputID)
+
+		apiInput := kbapi.PackagePolicyRequestInput{
+			Enabled: inputModel.Enabled.ValueBoolPointer(),
+			Vars:    utils.MapRef(utils.NormalizedTypeToMap[any](inputModel.Vars, inputPath.AtName("vars"), diags)),
 		}
 
-		jID := incoming[j].InputID.ValueString()
-		jIdx, ok := idToIndex[jID]
-		if !ok {
-			return true
+		// Convert streams if present
+		if utils.IsKnown(inputModel.Streams) && len(inputModel.Streams.Elements()) > 0 {
+			streamsMap := utils.MapTypeAs[integrationPolicyInputStreamModel](ctx, inputModel.Streams, inputPath.AtName("streams"), diags)
+			if streamsMap != nil {
+				streams := make(map[string]kbapi.PackagePolicyRequestInputStream, len(streamsMap))
+				for streamID, streamModel := range streamsMap {
+					streams[streamID] = kbapi.PackagePolicyRequestInputStream{
+						Enabled: streamModel.Enabled.ValueBoolPointer(),
+						Vars:    utils.MapRef(utils.NormalizedTypeToMap[any](streamModel.Vars, inputPath.AtName("streams").AtMapKey(streamID).AtName("vars"), diags)),
+					}
+				}
+				apiInput.Streams = &streams
+			}
 		}
 
-		return iIdx < jIdx
-	})
+		result[inputID] = apiInput
+	}
+
+	return result
 }

internal/fleet/integration_policy/models_test.go

@@ -9,62 +9,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func Test_SortInputs(t *testing.T) {
-	t.Run("WithExisting", func(t *testing.T) {
-		existing := []integrationPolicyInputModel{
-			{InputID: types.StringValue("A"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("B"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("C"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("D"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("E"), Enabled: types.BoolValue(true)},
-		}
-
-		incoming := []integrationPolicyInputModel{
-			{InputID: types.StringValue("G"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("F"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("B"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("E"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("C"), Enabled: types.BoolValue(true)},
-		}
-
-		want := []integrationPolicyInputModel{
-			{InputID: types.StringValue("B"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("C"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("E"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("G"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("F"), Enabled: types.BoolValue(true)},
-		}
-
-		sortInputs(incoming, existing)
-
-		require.Equal(t, want, incoming)
-	})
-
-	t.Run("WithEmpty", func(t *testing.T) {
-		var existing []integrationPolicyInputModel
-
-		incoming := []integrationPolicyInputModel{
-			{InputID: types.StringValue("G"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("F"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("B"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("E"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("C"), Enabled: types.BoolValue(true)},
-		}
-
-		want := []integrationPolicyInputModel{
-			{InputID: types.StringValue("B"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("C"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("E"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("F"), Enabled: types.BoolValue(true)},
-			{InputID: types.StringValue("G"), Enabled: types.BoolValue(true)},
-		}
-
-		sortInputs(incoming, existing)
-
-		require.Equal(t, want, incoming)
-	})
-}
-
 func TestOutputIdHandling(t *testing.T) {
 	t.Run("populateFromAPI", func(t *testing.T) {
 		model := &integrationPolicyModel{}

internal/fleet/integration_policy/read.go

@@ -54,7 +54,7 @@ func (r *integrationPolicyResource) Read(ctx context.Context, req resource.ReadR
 	}
 
 	// Remember if the state had input configured
-	stateHadInput := utils.IsKnown(stateModel.Input) && !stateModel.Input.IsNull() && len(stateModel.Input.Elements()) > 0
+	stateHadInput := utils.IsKnown(stateModel.Inputs) && !stateModel.Inputs.IsNull() && len(stateModel.Inputs.Elements()) > 0
 
 	// Check if this is an import operation (PolicyID is the only field set)
 	isImport := stateModel.PolicyID.ValueString() != "" &&
@@ -70,7 +70,7 @@ func (r *integrationPolicyResource) Read(ctx context.Context, req resource.ReadR
 	// This prevents "Provider produced inconsistent result" errors during refresh
 	// However, during import we should always populate inputs from the API
 	if !stateHadInput && !isImport {
-		stateModel.Input = types.ListNull(getInputTypeV1())
+		stateModel.Inputs = types.MapNull(getInputsTypes())
 	}
 
 	diags = resp.State.Set(ctx, stateModel)

internal/fleet/integration_policy/resource.go

@@ -49,7 +49,8 @@ func (r *integrationPolicyResource) ImportState(ctx context.Context, req resourc
 
 func (r *integrationPolicyResource) UpgradeState(context.Context) map[int64]resource.StateUpgrader {
 	return map[int64]resource.StateUpgrader{
-		0: {PriorSchema: getSchemaV0(), StateUpgrader: upgradeV0},
+		0: {PriorSchema: getSchemaV0(), StateUpgrader: upgradeV0ToV2},
+		1: {PriorSchema: getSchemaV1(), StateUpgrader: upgradeV1ToV2},
 	}
 }