Add output_id to the schema (#1445)

* Add output_id to the schema

* CHANGELOG.md

Author: tobio

CHANGELOG.md

@@ -7,6 +7,7 @@
 - Add back missing import support for `elasticstack_elasticsearch_security_role_mapping` ([#1441](https://github.com/elastic/terraform-provider-elasticstack/pull/1441))
 - Add new `elasticstack_elasticsearch_ml_job_state` resource ([#1337](https://github.com/elastic/terraform-provider-elasticstack/pull/1337))
 - Add new `elasticstack_elasticsearch_ml_datafeed_state` resource ([#1422](https://github.com/elastic/terraform-provider-elasticstack/pull/1422))
+- Add `output_id` to `elasticstack_fleet_integration_policy` resource ([#1445](https://github.com/elastic/terraform-provider-elasticstack/pull/1445))
 - Make `hosts` attribute required in `elasticstack_fleet_output` resource ([#1450](https://github.com/elastic/terraform-provider-elasticstack/pull/1450/files)) 
 
 ## [0.12.1] - 2025-10-22

docs/resources/fleet_integration_policy.md

@@ -63,6 +63,8 @@ resource "elasticstack_fleet_integration_policy" "sample" {
   agent_policy_id     = elasticstack_fleet_agent_policy.sample.policy_id
   integration_name    = elasticstack_fleet_integration.sample.name
   integration_version = elasticstack_fleet_integration.sample.version
+  // Optional: specify a custom output to send data to
+  // output_id           = "my-custom-output-id"
 
   input {
     input_id = "tcp-tcp"
@@ -102,6 +104,7 @@ resource "elasticstack_fleet_integration_policy" "sample" {
 - `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))
+- `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.
 - `vars_json` (String, Sensitive) Integration-level variables as JSON.

examples/resources/elasticstack_fleet_integration_policy/resource.tf

@@ -32,6 +32,8 @@ resource "elasticstack_fleet_integration_policy" "sample" {
   agent_policy_id     = elasticstack_fleet_agent_policy.sample.policy_id
   integration_name    = elasticstack_fleet_integration.sample.name
   integration_version = elasticstack_fleet_integration.sample.version
+  // Optional: specify a custom output to send data to
+  // output_id           = "my-custom-output-id"
 
   input {
     input_id = "tcp-tcp"

internal/fleet/integration_policy/acc_test.go

@@ -25,6 +25,7 @@ import (
 var (
 	minVersionIntegrationPolicy    = version.Must(version.NewVersion("8.10.0"))
 	minVersionIntegrationPolicyIds = version.Must(version.NewVersion("8.15.0"))
+	minVersionOutputId             = version.Must(version.NewVersion("8.16.0"))
 	minVersionSqlIntegration       = version.Must(version.NewVersion("9.1.0"))
 )
 
@@ -62,6 +63,56 @@ func TestAccResourceIntegrationPolicyMultipleAgentPolicies(t *testing.T) {
 	})
 }
 
+func TestAccResourceIntegrationPolicyWithOutput(t *testing.T) {
+	policyName := sdkacctest.RandStringFromCharSet(22, sdkacctest.CharSetAlphaNum)
+
+	resource.Test(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t) },
+		CheckDestroy: checkResourceIntegrationPolicyDestroy,
+		Steps: []resource.TestStep{
+			{
+				ProtoV6ProviderFactories: acctest.Providers,
+				SkipFunc:                 versionutils.CheckIfVersionIsUnsupported(minVersionOutputId),
+				ConfigDirectory:          acctest.NamedTestCaseDirectory("create"),
+				ConfigVariables: config.Variables{
+					"policy_name": config.StringVariable(policyName),
+					"output_name": config.StringVariable(fmt.Sprintf("Test Output %s", policyName)),
+				},
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "name", policyName),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "description", "IntegrationPolicyTest Policy with Output"),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "integration_name", "tcp"),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "integration_version", "1.16.0"),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "output_id", fmt.Sprintf("%s-test-output", policyName)),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "input.0.input_id", "tcp-tcp"),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "input.0.enabled", "true"),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "input.0.streams_json", `{"tcp.generic":{"enabled":true,"vars":{"custom":"","data_stream.dataset":"tcp.generic","listen_address":"localhost","listen_port":8080,"ssl":"","syslog_options":"field: message","tags":[]}}}`),
+				),
+			},
+			{
+				ProtoV6ProviderFactories: acctest.Providers,
+				SkipFunc:                 versionutils.CheckIfVersionIsUnsupported(minVersionOutputId),
+				ConfigDirectory:          acctest.NamedTestCaseDirectory("update"),
+				ConfigVariables: config.Variables{
+					"policy_name":         config.StringVariable(policyName),
+					"output_name":         config.StringVariable(fmt.Sprintf("Test Output %s", policyName)),
+					"updated_output_name": config.StringVariable(fmt.Sprintf("Updated Test Output %s", policyName)),
+				},
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "name", policyName),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "description", "Updated Integration Policy with Output"),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "integration_name", "tcp"),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "integration_version", "1.16.0"),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "output_id", fmt.Sprintf("%s-updated-output", policyName)),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "input.0.input_id", "tcp-tcp"),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "input.0.enabled", "false"),
+					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "input.0.streams_json", `{"tcp.generic":{"enabled":false,"vars":{"custom":"","data_stream.dataset":"tcp.generic","listen_address":"localhost","listen_port":8085,"ssl":"","syslog_options":"field: message","tags":[]}}}`),
+				),
+			},
+		},
+	})
+}
+
 func TestAccResourceIntegrationPolicy(t *testing.T) {
 	policyName := sdkacctest.RandStringFromCharSet(22, sdkacctest.CharSetAlphaNum)
 
@@ -82,6 +133,7 @@ func TestAccResourceIntegrationPolicy(t *testing.T) {
 					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "integration_name", "tcp"),
 					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "integration_version", "1.16.0"),
 					resource.TestCheckNoResourceAttr("elasticstack_fleet_integration_policy.test_policy", "vars_json"),
+					resource.TestCheckNoResourceAttr("elasticstack_fleet_integration_policy.test_policy", "output_id"),
 					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "input.0.input_id", "tcp-tcp"),
 					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "input.0.enabled", "true"),
 					resource.TestCheckNoResourceAttr("elasticstack_fleet_integration_policy.test_policy", "input.0.vars_json"),
@@ -101,6 +153,7 @@ func TestAccResourceIntegrationPolicy(t *testing.T) {
 					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "integration_name", "tcp"),
 					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "integration_version", "1.16.0"),
 					resource.TestCheckNoResourceAttr("elasticstack_fleet_integration_policy.test_policy", "vars_json"),
+					resource.TestCheckNoResourceAttr("elasticstack_fleet_integration_policy.test_policy", "output_id"),
 					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "input.0.input_id", "tcp-tcp"),
 					resource.TestCheckResourceAttr("elasticstack_fleet_integration_policy.test_policy", "input.0.enabled", "false"),
 					resource.TestCheckNoResourceAttr("elasticstack_fleet_integration_policy.test_policy", "input.0.vars_json"),

internal/fleet/integration_policy/models.go

@@ -15,6 +15,7 @@ import (
 
 type features struct {
 	SupportsPolicyIds bool
+	SupportsOutputId  bool
 }
 
 type integrationPolicyModel struct {
@@ -29,6 +30,7 @@ type integrationPolicyModel struct {
 	Force              types.Bool           `tfsdk:"force"`
 	IntegrationName    types.String         `tfsdk:"integration_name"`
 	IntegrationVersion types.String         `tfsdk:"integration_version"`
+	OutputID           types.String         `tfsdk:"output_id"`
 	Input              types.List           `tfsdk:"input"` //> integrationPolicyInputModel
 	VarsJson           jsontypes.Normalized `tfsdk:"vars_json"`
 	SpaceIds           types.Set            `tfsdk:"space_ids"`
@@ -90,6 +92,7 @@ func (model *integrationPolicyModel) populateFromAPI(ctx context.Context, data *
 	model.Enabled = types.BoolValue(data.Enabled)
 	model.IntegrationName = types.StringValue(data.Package.Name)
 	model.IntegrationVersion = types.StringValue(data.Package.Version)
+	model.OutputID = types.StringPointerValue(data.OutputId)
 	model.VarsJson = utils.MapToNormalizedType(utils.Deref(data.Vars), path.Root("vars_json"), &diags)
 
 	// Preserve space_ids if it was originally set in the plan/state
@@ -170,11 +173,25 @@ func (model integrationPolicyModel) toAPIModel(ctx context.Context, isUpdate boo
 		}
 	}
 
+	// Check if output_id is configured and version supports it
+	if utils.IsKnown(model.OutputID) {
+		if !feat.SupportsOutputId {
+			return kbapi.PackagePolicyRequest{}, diag.Diagnostics{
+				diag.NewAttributeErrorDiagnostic(
+					path.Root("output_id"),
+					"Unsupported Elasticsearch version",
+					fmt.Sprintf("Output ID is only supported in Elastic Stack %s and above", MinVersionOutputId),
+				),
+			}
+		}
+	}
+
 	body := kbapi.PackagePolicyRequest{
 		Description: model.Description.ValueStringPointer(),
 		Force:       model.Force.ValueBoolPointer(),
 		Name:        model.Name.ValueString(),
 		Namespace:   model.Namespace.ValueStringPointer(),
+		OutputId:    model.OutputID.ValueStringPointer(),
 		Package: kbapi.PackagePolicyRequestPackage{
 			Name:    model.IntegrationName.ValueString(),
 			Version: model.IntegrationVersion.ValueString(),

internal/fleet/integration_policy/models_test.go

@@ -1,8 +1,10 @@
 package integration_policy
 
 import (
+	"context"
 	"testing"
 
+	"github.com/elastic/terraform-provider-elasticstack/generated/kbapi"
 	"github.com/hashicorp/terraform-plugin-framework/types"
 	"github.com/stretchr/testify/require"
 )
@@ -62,3 +64,77 @@ func Test_SortInputs(t *testing.T) {
 		require.Equal(t, want, incoming)
 	})
 }
+
+func TestOutputIdHandling(t *testing.T) {
+	t.Run("populateFromAPI", func(t *testing.T) {
+		model := &integrationPolicyModel{}
+		outputId := "test-output-id"
+		data := &kbapi.PackagePolicy{
+			Id:      "test-id",
+			Name:    "test-policy",
+			Enabled: true,
+			Package: &struct {
+				ExperimentalDataStreamFeatures *[]struct {
+					DataStream string `json:"data_stream"`
+					Features   struct {
+						DocValueOnlyNumeric *bool `json:"doc_value_only_numeric,omitempty"`
+						DocValueOnlyOther   *bool `json:"doc_value_only_other,omitempty"`
+						SyntheticSource     *bool `json:"synthetic_source,omitempty"`
+						Tsdb                *bool `json:"tsdb,omitempty"`
+					} `json:"features"`
+				} `json:"experimental_data_stream_features,omitempty"`
+				FipsCompatible *bool   `json:"fips_compatible,omitempty"`
+				Name           string  `json:"name"`
+				RequiresRoot   *bool   `json:"requires_root,omitempty"`
+				Title          *string `json:"title,omitempty"`
+				Version        string  `json:"version"`
+			}{
+				Name:    "test-integration",
+				Version: "1.0.0",
+			},
+			OutputId: &outputId,
+		}
+
+		diags := model.populateFromAPI(context.Background(), data)
+		require.Empty(t, diags)
+		require.Equal(t, "test-output-id", model.OutputID.ValueString())
+	})
+
+	t.Run("toAPIModel", func(t *testing.T) {
+		model := integrationPolicyModel{
+			Name:               types.StringValue("test-policy"),
+			IntegrationName:    types.StringValue("test-integration"),
+			IntegrationVersion: types.StringValue("1.0.0"),
+			OutputID:           types.StringValue("test-output-id"),
+		}
+
+		feat := features{
+			SupportsPolicyIds: true,
+			SupportsOutputId:  true,
+		}
+
+		result, diags := model.toAPIModel(context.Background(), false, feat)
+		require.Empty(t, diags)
+		require.NotNil(t, result.OutputId)
+		require.Equal(t, "test-output-id", *result.OutputId)
+	})
+
+	t.Run("toAPIModel_unsupported_version", func(t *testing.T) {
+		model := integrationPolicyModel{
+			Name:               types.StringValue("test-policy"),
+			IntegrationName:    types.StringValue("test-integration"),
+			IntegrationVersion: types.StringValue("1.0.0"),
+			OutputID:           types.StringValue("test-output-id"),
+		}
+
+		feat := features{
+			SupportsPolicyIds: true,
+			SupportsOutputId:  false, // Simulate unsupported version
+		}
+
+		_, diags := model.toAPIModel(context.Background(), false, feat)
+		require.Len(t, diags, 1)
+		require.Equal(t, "Unsupported Elasticsearch version", diags[0].Summary())
+		require.Contains(t, diags[0].Detail(), "Output ID is only supported in Elastic Stack")
+	})
+}

internal/fleet/integration_policy/resource.go

@@ -21,6 +21,7 @@ var (
 
 var (
 	MinVersionPolicyIds = version.Must(version.NewVersion("8.15.0"))
+	MinVersionOutputId  = version.Must(version.NewVersion("8.16.0"))
 )
 
 // NewResource is a helper function to simplify the provider implementation.
@@ -58,7 +59,13 @@ func (r *integrationPolicyResource) buildFeatures(ctx context.Context) (features
 		return features{}, diagutil.FrameworkDiagsFromSDK(diags)
 	}
 
+	supportsOutputId, outputIdDiags := r.client.EnforceMinVersion(ctx, MinVersionOutputId)
+	if outputIdDiags.HasError() {
+		return features{}, diagutil.FrameworkDiagsFromSDK(outputIdDiags)
+	}
+
 	return features{
 		SupportsPolicyIds: supportsPolicyIds,
+		SupportsOutputId:  supportsOutputId,
 	}, nil
 }

internal/fleet/integration_policy/schema.go

@@ -92,6 +92,10 @@ func getSchemaV1() schema.Schema {
 				Description: "The version of the integration package.",
 				Required:    true,
 			},
+			"output_id": schema.StringAttribute{
+				Description: "The ID of the output to send data to. When not specified, the default output of the agent policy will be used.",
+				Optional:    true,
+			},
 			"vars_json": schema.StringAttribute{
 				Description: "Integration-level variables as JSON.",
 				CustomType:  jsontypes.NormalizedType{},

internal/fleet/integration_policy/testdata/TestAccResourceIntegrationPolicyWithOutput/create/integration_policy.tf

@@ -0,0 +1,88 @@
+variable "policy_name" {
+  description = "The integration policy name"
+  type        = string
+}
+
+variable "output_name" {
+  description = "The output name"
+  type        = string
+}
+
+variable "integration_name" {
+  description = "The integration name"
+  type        = string
+  default     = "tcp"
+}
+
+variable "integration_version" {
+  description = "The integration version"
+  type        = string
+  default     = "1.16.0"
+}
+
+provider "elasticstack" {
+  elasticsearch {}
+  kibana {}
+}
+
+resource "elasticstack_fleet_integration" "test_policy" {
+  name    = var.integration_name
+  version = var.integration_version
+  force   = true
+}
+
+resource "elasticstack_fleet_agent_policy" "test_policy" {
+  name            = "${var.policy_name} Agent Policy"
+  namespace       = "default"
+  description     = "IntegrationPolicyTest Agent Policy"
+  monitor_logs    = true
+  monitor_metrics = true
+  skip_destroy    = false
+}
+
+resource "elasticstack_fleet_output" "test_output" {
+  name      = var.output_name
+  output_id = "${var.policy_name}-test-output"
+  type      = "elasticsearch"
+  config_yaml = yamlencode({
+    "ssl.verification_mode" : "none"
+  })
+  default_integrations = false
+  default_monitoring   = false
+  hosts = [
+    "https://elasticsearch:9200"
+  ]
+}
+
+data "elasticstack_fleet_enrollment_tokens" "test_policy" {
+  policy_id = elasticstack_fleet_agent_policy.test_policy.policy_id
+}
+
+resource "elasticstack_fleet_integration_policy" "test_policy" {
+  name                = var.policy_name
+  namespace           = "default"
+  description         = "IntegrationPolicyTest Policy with Output"
+  agent_policy_id     = elasticstack_fleet_agent_policy.test_policy.policy_id
+  integration_name    = elasticstack_fleet_integration.test_policy.name
+  integration_version = elasticstack_fleet_integration.test_policy.version
+  output_id           = elasticstack_fleet_output.test_output.output_id
+
+  input {
+    input_id = "tcp-tcp"
+    enabled  = true
+    streams_json = jsonencode({
+      "tcp.generic" : {
+        "enabled" : true
+        "vars" : {
+          "listen_address" : "localhost"
+          "listen_port" : 8080
+          "data_stream.dataset" : "tcp.generic"
+          "tags" : []
+          "syslog_options" : "field: message"
+          "ssl" : ""
+          "custom" : ""
+        }
+      }
+    })
+  }
+}