diff --git a/CHANGELOG.md b/CHANGELOG.md
index ddac5b7bb..d6082b13e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,7 @@
 ## [Unreleased]
 ### Added
 - New resource `elasticstack_elasticsearch_data_stream` to manage Elasticsearch [data streams](https://www.elastic.co/guide/en/elasticsearch/reference/current/data-streams.html) ([#45](https://github.com/elastic/terraform-provider-elasticstack/pull/45))
+- New resource `elasticstack_elasticsearch_ingest_pipeline` to manage Elasticsearch [ingest pipelines](https://www.elastic.co/guide/en/elasticsearch/reference/7.16/ingest.html) ([#56](https://github.com/elastic/terraform-provider-elasticstack/issues/56))
 
 ### Fixed
 - Update only changed index settings ([#52](https://github.com/elastic/terraform-provider-elasticstack/issues/52))
diff --git a/docs/resources/elasticsearch_ingest_pipeline.md b/docs/resources/elasticsearch_ingest_pipeline.md
new file mode 100644
index 000000000..c20ed74bf
--- /dev/null
+++ b/docs/resources/elasticsearch_ingest_pipeline.md
@@ -0,0 +1,82 @@
+---
+subcategory: "Ingest"
+layout: ""
+page_title: "Elasticstack: elasticstack_elasticsearch_ingest_pipeline Resource"
+description: |-
+  Manages Ingest Pipelines
+---
+
+# Resource: elasticstack_elasticsearch_ingest_pipeline
+
+Use ingest APIs to manage tasks and resources related to ingest pipelines and processors. See: https://www.elastic.co/guide/en/elasticsearch/reference/current/ingest-apis.html
+
+## Example Usage
+
+```terraform
+provider "elasticstack" {
+  elasticsearch {}
+}
+
+resource "elasticstack_elasticsearch_ingest_pipeline" "my_ingest_pipeline" {
+  name        = "my_ingest_pipeline"
+  description = "My first ingest pipeline managed by Terraform"
+
+  // processors can be defined in different way
+  processors = [
+    // using the jsonencode function, which is the recomended way if you want to provide JSON object by yourself
+    jsonencode({
+      set = {
+        description = "My set processor descirption"
+        field       = "_meta"
+        value       = "indexed"
+      }
+    }),
+    // or use the HERE DOC construct to provide the processor definition
+    <<EOF
+    {"json": {
+      "field": "data",
+      "target_field": "parsed_data"
+    }}
+EOF
+    ,
+  ]
+}
+```
+
+<!-- schema generated by tfplugindocs -->
+## Schema
+
+### Required
+
+- **name** (String) The name of the ingest pipeline.
+- **processors** (List of String) Processors used to perform transformations on documents before indexing. Processors run sequentially in the order specified. See: https://www.elastic.co/guide/en/elasticsearch/reference/current/processors.html. Each record must be a valid JSON document.
+
+### Optional
+
+- **description** (String) Description of the ingest pipeline.
+- **elasticsearch_connection** (Block List, Max: 1) Used to establish connection to Elasticsearch server. Overrides environment variables if present. (see [below for nested schema](#nestedblock--elasticsearch_connection))
+- **metadata** (String) Optional user metadata about the index template.
+- **on_failure** (List of String) Processors to run immediately after a processor failure. Each processor supports a processor-level `on_failure` value. If a processor without an `on_failure` value fails, Elasticsearch uses this pipeline-level parameter as a fallback. The processors in this parameter run sequentially in the order specified. Elasticsearch will not attempt to run the pipeline’s remaining processors. See: https://www.elastic.co/guide/en/elasticsearch/reference/current/processors.html. Each record must be a valid JSON document
+
+### Read-Only
+
+- **id** (String) Internal identifier of the resource
+
+<a id="nestedblock--elasticsearch_connection"></a>
+### Nested Schema for `elasticsearch_connection`
+
+Optional:
+
+- **ca_file** (String) Path to a custom Certificate Authority certificate
+- **endpoints** (List of String, Sensitive) A list of endpoints the Terraform provider will point to. They must include the http(s) schema and port number.
+- **insecure** (Boolean) Disable TLS certificate validation
+- **password** (String, Sensitive) A password to use for API authentication to Elasticsearch.
+- **username** (String) A username to use for API authentication to Elasticsearch.
+
+## Import
+
+Import is supported using the following syntax:
+
+```shell
+terraform import elasticstack_elasticsearch_ingest_pipeline.my_ingest_pipeline <cluster_uuid>/<ingest pipeline name>
+```
diff --git a/examples/resources/elasticstack_elasticsearch_ingest_pipeline/import.sh b/examples/resources/elasticstack_elasticsearch_ingest_pipeline/import.sh
new file mode 100644
index 000000000..77374a73c
--- /dev/null
+++ b/examples/resources/elasticstack_elasticsearch_ingest_pipeline/import.sh
@@ -0,0 +1 @@
+terraform import elasticstack_elasticsearch_ingest_pipeline.my_ingest_pipeline <cluster_uuid>/<ingest pipeline name>
diff --git a/examples/resources/elasticstack_elasticsearch_ingest_pipeline/resource.tf b/examples/resources/elasticstack_elasticsearch_ingest_pipeline/resource.tf
new file mode 100644
index 000000000..ea3f12147
--- /dev/null
+++ b/examples/resources/elasticstack_elasticsearch_ingest_pipeline/resource.tf
@@ -0,0 +1,28 @@
+provider "elasticstack" {
+  elasticsearch {}
+}
+
+resource "elasticstack_elasticsearch_ingest_pipeline" "my_ingest_pipeline" {
+  name        = "my_ingest_pipeline"
+  description = "My first ingest pipeline managed by Terraform"
+
+  // processors can be defined in different way
+  processors = [
+    // using the jsonencode function, which is the recomended way if you want to provide JSON object by yourself
+    jsonencode({
+      set = {
+        description = "My set processor descirption"
+        field       = "_meta"
+        value       = "indexed"
+      }
+    }),
+    // or use the HERE DOC construct to provide the processor definition
+    <<EOF
+    {"json": {
+      "field": "data",
+      "target_field": "parsed_data"
+    }}
+EOF
+    ,
+  ]
+}
diff --git a/internal/clients/index.go b/internal/clients/index.go
index d90ee6be8..61dab0b00 100644
--- a/internal/clients/index.go
+++ b/internal/clients/index.go
@@ -327,3 +327,63 @@ func (a *ApiClient) DeleteElasticsearchDataStream(dataStreamName string) diag.Di
 
 	return diags
 }
+
+func (a *ApiClient) PutElasticsearchIngestPipeline(pipeline *models.IngestPipeline) diag.Diagnostics {
+	var diags diag.Diagnostics
+	pipelineBytes, err := json.Marshal(pipeline)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+	log.Printf("[TRACE] creating ingest pipeline %s: %s", pipeline.Name, pipelineBytes)
+
+	res, err := a.es.Ingest.PutPipeline(pipeline.Name, bytes.NewReader(pipelineBytes))
+	if err != nil {
+		return diag.FromErr(err)
+	}
+	defer res.Body.Close()
+	if diags := utils.CheckError(res, fmt.Sprintf("Unable to create or update ingest pipeline: %s", pipeline.Name)); diags.HasError() {
+		return diags
+	}
+
+	return diags
+}
+
+func (a *ApiClient) GetElasticsearchIngestPipeline(name *string) (*models.IngestPipeline, diag.Diagnostics) {
+	var diags diag.Diagnostics
+	req := a.es.Ingest.GetPipeline.WithPipelineID(*name)
+	res, err := a.es.Ingest.GetPipeline(req)
+	if err != nil {
+		return nil, diag.FromErr(err)
+	}
+	defer res.Body.Close()
+	if res.StatusCode == http.StatusNotFound {
+		return nil, nil
+	}
+	if diags := utils.CheckError(res, fmt.Sprintf("Unable to get requested ingest pipeline: %s", *name)); diags.HasError() {
+		return nil, diags
+	}
+
+	pipelines := make(map[string]models.IngestPipeline)
+	if err := json.NewDecoder(res.Body).Decode(&pipelines); err != nil {
+		return nil, diag.FromErr(err)
+	}
+	pipeline := pipelines[*name]
+	pipeline.Name = *name
+	log.Printf("[TRACE] get ingest pipeline %s from ES API: %#+v", *name, pipeline)
+
+	return &pipeline, diags
+}
+
+func (a *ApiClient) DeleteElasticsearchIngestPipeline(name *string) diag.Diagnostics {
+	var diags diag.Diagnostics
+
+	res, err := a.es.Ingest.DeletePipeline(*name)
+	if err != nil {
+		return diags
+	}
+	defer res.Body.Close()
+	if diags := utils.CheckError(res, fmt.Sprintf("Unable to delete ingest pipeline: %s", *name)); diags.HasError() {
+		return diags
+	}
+	return diags
+}
diff --git a/internal/elasticsearch/ingest/pipeline.go b/internal/elasticsearch/ingest/pipeline.go
new file mode 100644
index 000000000..cdf37e21d
--- /dev/null
+++ b/internal/elasticsearch/ingest/pipeline.go
@@ -0,0 +1,223 @@
+package ingest
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+
+	"github.com/elastic/terraform-provider-elasticstack/internal/clients"
+	"github.com/elastic/terraform-provider-elasticstack/internal/models"
+	"github.com/elastic/terraform-provider-elasticstack/internal/utils"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/validation"
+)
+
+func ResourceIngestPipeline() *schema.Resource {
+	pipelineSchema := map[string]*schema.Schema{
+		"id": {
+			Description: "Internal identifier of the resource",
+			Type:        schema.TypeString,
+			Computed:    true,
+		},
+		"name": {
+			Description: "The name of the ingest pipeline.",
+			Type:        schema.TypeString,
+			Required:    true,
+			ForceNew:    true,
+		},
+		"description": {
+			Description: "Description of the ingest pipeline.",
+			Type:        schema.TypeString,
+			Optional:    true,
+		},
+		"on_failure": {
+			Description: "Processors to run immediately after a processor failure. Each processor supports a processor-level `on_failure` value. If a processor without an `on_failure` value fails, Elasticsearch uses this pipeline-level parameter as a fallback. The processors in this parameter run sequentially in the order specified. Elasticsearch will not attempt to run the pipeline’s remaining processors. See: https://www.elastic.co/guide/en/elasticsearch/reference/current/processors.html. Each record must be a valid JSON document",
+			Type:        schema.TypeList,
+			Optional:    true,
+			MinItems:    1,
+			Elem: &schema.Schema{
+				Type:             schema.TypeString,
+				ValidateFunc:     validation.StringIsJSON,
+				DiffSuppressFunc: utils.DiffJsonSuppress,
+			},
+		},
+		"processors": {
+			Description: "Processors used to perform transformations on documents before indexing. Processors run sequentially in the order specified. See: https://www.elastic.co/guide/en/elasticsearch/reference/current/processors.html. Each record must be a valid JSON document.",
+			Type:        schema.TypeList,
+			Required:    true,
+			MinItems:    1,
+			Elem: &schema.Schema{
+				Type:             schema.TypeString,
+				ValidateFunc:     validation.StringIsJSON,
+				DiffSuppressFunc: utils.DiffJsonSuppress,
+			},
+		},
+		"metadata": {
+			Description:      "Optional user metadata about the index template.",
+			Type:             schema.TypeString,
+			Optional:         true,
+			ValidateFunc:     validation.StringIsJSON,
+			DiffSuppressFunc: utils.DiffJsonSuppress,
+		},
+	}
+
+	utils.AddConnectionSchema(pipelineSchema)
+
+	return &schema.Resource{
+		Description: "Manages tasks and resources related to ingest pipelines and processors. See: https://www.elastic.co/guide/en/elasticsearch/reference/current/ingest-apis.html",
+
+		CreateContext: resourceIngestPipelineTemplatePut,
+		UpdateContext: resourceIngestPipelineTemplatePut,
+		ReadContext:   resourceIngestPipelineTemplateRead,
+		DeleteContext: resourceIngestPipelineTemplateDelete,
+
+		Importer: &schema.ResourceImporter{
+			StateContext: schema.ImportStatePassthroughContext,
+		},
+
+		Schema: pipelineSchema,
+	}
+}
+
+func resourceIngestPipelineTemplatePut(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	client, err := clients.NewApiClient(d, meta)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+	pipelineId := d.Get("name").(string)
+	id, diags := client.ID(pipelineId)
+	if diags.HasError() {
+		return diags
+	}
+	var pipeline models.IngestPipeline
+	pipeline.Name = pipelineId
+	if v, ok := d.GetOk("description"); ok {
+		r := v.(string)
+		pipeline.Description = &r
+	}
+	if v, ok := d.GetOk("on_failure"); ok {
+		onFailure := make([]map[string]interface{}, len(v.([]interface{})))
+		for i, f := range v.([]interface{}) {
+			item := make(map[string]interface{})
+			if err := json.NewDecoder(strings.NewReader(f.(string))).Decode(&item); err != nil {
+				return diag.FromErr(err)
+			}
+			onFailure[i] = item
+		}
+		pipeline.OnFailure = onFailure
+	}
+	if v, ok := d.GetOk("processors"); ok {
+		procs := make([]map[string]interface{}, len(v.([]interface{})))
+		for i, f := range v.([]interface{}) {
+			item := make(map[string]interface{})
+			if err := json.NewDecoder(strings.NewReader(f.(string))).Decode(&item); err != nil {
+				return diag.FromErr(err)
+			}
+			procs[i] = item
+		}
+		pipeline.Processors = procs
+	}
+	if v, ok := d.GetOk("metadata"); ok {
+		metadata := make(map[string]interface{})
+		if err := json.NewDecoder(strings.NewReader(v.(string))).Decode(&metadata); err != nil {
+			return diag.FromErr(err)
+		}
+		pipeline.Metadata = metadata
+	}
+
+	if diags := client.PutElasticsearchIngestPipeline(&pipeline); diags.HasError() {
+		return diags
+	}
+
+	d.SetId(id.String())
+	return resourceIngestPipelineTemplateRead(ctx, d, meta)
+}
+
+func resourceIngestPipelineTemplateRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	var diags diag.Diagnostics
+	client, err := clients.NewApiClient(d, meta)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+	id := d.Id()
+	compId, diags := clients.CompositeIdFromStr(id)
+	if diags.HasError() {
+		return diags
+	}
+
+	pipeline, diags := client.GetElasticsearchIngestPipeline(&compId.ResourceId)
+	if pipeline == nil && diags == nil {
+		d.SetId("")
+		return diags
+	}
+	if diags.HasError() {
+		return diags
+	}
+	if err := d.Set("name", pipeline.Name); err != nil {
+		return diag.FromErr(err)
+	}
+	if desc := pipeline.Description; desc != nil {
+		if err := d.Set("description", desc); err != nil {
+			return diag.FromErr(err)
+		}
+	}
+	if onFailure := pipeline.OnFailure; onFailure != nil {
+		fProcs := make([]string, len(onFailure))
+		for i, v := range onFailure {
+			res, err := json.Marshal(v)
+			if err != nil {
+				return diag.FromErr(err)
+			}
+			fProcs[i] = string(res)
+		}
+
+		if err := d.Set("on_failure", fProcs); err != nil {
+			return diag.FromErr(err)
+		}
+	}
+	procs := make([]string, len(pipeline.Processors))
+	for i, v := range pipeline.Processors {
+		res, err := json.Marshal(v)
+		if err != nil {
+			return diag.FromErr(err)
+		}
+		procs[i] = string(res)
+	}
+
+	if err := d.Set("processors", procs); err != nil {
+		return diag.FromErr(err)
+	}
+
+	if meta := pipeline.Metadata; meta != nil {
+		meta, err := json.Marshal(meta)
+		if err != nil {
+			return diag.FromErr(err)
+		}
+		if err := d.Set("metadata", string(meta)); err != nil {
+			return diag.FromErr(err)
+		}
+	}
+
+	return diags
+}
+
+func resourceIngestPipelineTemplateDelete(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	var diags diag.Diagnostics
+	client, err := clients.NewApiClient(d, meta)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+	id := d.Id()
+	compId, diags := clients.CompositeIdFromStr(id)
+	if diags.HasError() {
+		return diags
+	}
+
+	if diags := client.DeleteElasticsearchIngestPipeline(&compId.ResourceId); diags.HasError() {
+		return diags
+	}
+
+	d.SetId("")
+	return diags
+}
diff --git a/internal/elasticsearch/ingest/pipeline_test.go b/internal/elasticsearch/ingest/pipeline_test.go
new file mode 100644
index 000000000..994536ebe
--- /dev/null
+++ b/internal/elasticsearch/ingest/pipeline_test.go
@@ -0,0 +1,114 @@
+package ingest_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/elastic/terraform-provider-elasticstack/internal/acctest"
+	"github.com/elastic/terraform-provider-elasticstack/internal/clients"
+	sdkacctest "github.com/hashicorp/terraform-plugin-sdk/v2/helper/acctest"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/terraform"
+)
+
+func TestAccResourceIngestPipeline(t *testing.T) {
+	pipelineName := sdkacctest.RandStringFromCharSet(22, sdkacctest.CharSetAlphaNum)
+
+	resource.UnitTest(t, resource.TestCase{
+		PreCheck:          func() { acctest.PreCheck(t) },
+		CheckDestroy:      checkResourceIngestPipelineDestroy,
+		ProviderFactories: acctest.Providers,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccResourceIngestPipelineCreate(pipelineName),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("elasticstack_elasticsearch_ingest_pipeline.test_pipeline", "name", pipelineName),
+					resource.TestCheckResourceAttr("elasticstack_elasticsearch_ingest_pipeline.test_pipeline", "description", "Test Pipeline"),
+					resource.TestCheckResourceAttr("elasticstack_elasticsearch_ingest_pipeline.test_pipeline", "processors.#", "2"),
+				),
+			},
+			{
+				Config: testAccResourceIngestPipelineUpdate(pipelineName),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("elasticstack_elasticsearch_ingest_pipeline.test_pipeline", "name", pipelineName),
+					resource.TestCheckResourceAttr("elasticstack_elasticsearch_ingest_pipeline.test_pipeline", "description", "Test Pipeline"),
+					resource.TestCheckResourceAttr("elasticstack_elasticsearch_ingest_pipeline.test_pipeline", "processors.#", "1"),
+				),
+			},
+		},
+	})
+}
+
+func testAccResourceIngestPipelineCreate(name string) string {
+	return fmt.Sprintf(`
+provider "elasticstack" {
+  elasticsearch {}
+}
+
+resource "elasticstack_elasticsearch_ingest_pipeline" "test_pipeline" {
+  name        = "%s"
+  description = "Test Pipeline"
+
+  processors = [
+    jsonencode({
+      set = {
+        description = "My set processor description"
+        field       = "_meta"
+        value       = "indexed"
+      }
+    }),
+    <<EOF
+    {"json": {
+      "field": "data",
+      "target_field": "parsed_data"
+    }}
+EOF
+    ,
+  ]
+}
+	`, name)
+}
+
+func testAccResourceIngestPipelineUpdate(name string) string {
+	return fmt.Sprintf(`
+provider "elasticstack" {
+  elasticsearch {}
+}
+
+resource "elasticstack_elasticsearch_ingest_pipeline" "test_pipeline" {
+  name        = "%s"
+  description = "Test Pipeline"
+
+  processors = [
+    jsonencode({
+      set = {
+        description = "My set processor description"
+        field       = "_meta"
+        value       = "indexed"
+      }
+    })
+  ]
+}
+	`, name)
+}
+
+func checkResourceIngestPipelineDestroy(s *terraform.State) error {
+	client := acctest.Provider.Meta().(*clients.ApiClient)
+
+	for _, rs := range s.RootModule().Resources {
+		if rs.Type != "elasticstack_elasticsearch_ingest_pipeline" {
+			continue
+		}
+		compId, _ := clients.CompositeIdFromStr(rs.Primary.ID)
+
+		res, err := client.GetESClient().Indices.Get([]string{compId.ResourceId})
+		if err != nil {
+			return err
+		}
+
+		if res.StatusCode != 404 {
+			return fmt.Errorf("Ingest pipeline (%s) still exists", compId.ResourceId)
+		}
+	}
+	return nil
+}
diff --git a/internal/models/models.go b/internal/models/models.go
index 978340816..a75ac8254 100644
--- a/internal/models/models.go
+++ b/internal/models/models.go
@@ -156,3 +156,11 @@ type DataStreamIndex struct {
 type TimestampField struct {
 	Name string `json:"name"`
 }
+
+type IngestPipeline struct {
+	Name        string                   `json:"-"`
+	Description *string                  `json:"description,omitempty"`
+	OnFailure   []map[string]interface{} `json:"on_failure,omitempty"`
+	Processors  []map[string]interface{} `json:"processors"`
+	Metadata    map[string]interface{}   `json:"_meta,omitempty"`
+}
diff --git a/internal/provider/provider.go b/internal/provider/provider.go
index 9747eeabd..f6ecb80d9 100644
--- a/internal/provider/provider.go
+++ b/internal/provider/provider.go
@@ -4,6 +4,7 @@ import (
 	"github.com/elastic/terraform-provider-elasticstack/internal/clients"
 	"github.com/elastic/terraform-provider-elasticstack/internal/elasticsearch/cluster"
 	"github.com/elastic/terraform-provider-elasticstack/internal/elasticsearch/index"
+	"github.com/elastic/terraform-provider-elasticstack/internal/elasticsearch/ingest"
 	"github.com/elastic/terraform-provider-elasticstack/internal/elasticsearch/security"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 )
@@ -73,6 +74,7 @@ func New(version string) func() *schema.Provider {
 				"elasticstack_elasticsearch_index":               index.ResourceIndex(),
 				"elasticstack_elasticsearch_index_lifecycle":     index.ResourceIlm(),
 				"elasticstack_elasticsearch_index_template":      index.ResourceTemplate(),
+				"elasticstack_elasticsearch_ingest_pipeline":     ingest.ResourceIngestPipeline(),
 				"elasticstack_elasticsearch_security_role":       security.ResourceRole(),
 				"elasticstack_elasticsearch_security_user":       security.ResourceUser(),
 				"elasticstack_elasticsearch_snapshot_lifecycle":  cluster.ResourceSlm(),
diff --git a/templates/resources/elasticsearch_ingest_pipeline.md.tmpl b/templates/resources/elasticsearch_ingest_pipeline.md.tmpl
new file mode 100644
index 000000000..477ee6ae0
--- /dev/null
+++ b/templates/resources/elasticsearch_ingest_pipeline.md.tmpl
@@ -0,0 +1,23 @@
+---
+subcategory: "Ingest"
+layout: ""
+page_title: "Elasticstack: elasticstack_elasticsearch_ingest_pipeline Resource"
+description: |-
+  Manages Ingest Pipelines
+---
+
+# Resource: elasticstack_elasticsearch_ingest_pipeline
+
+Use ingest APIs to manage tasks and resources related to ingest pipelines and processors. See: https://www.elastic.co/guide/en/elasticsearch/reference/current/ingest-apis.html
+
+## Example Usage
+
+{{ tffile "examples/resources/elasticstack_elasticsearch_ingest_pipeline/resource.tf" }}
+
+{{ .SchemaMarkdown | trimspace }}
+
+## Import
+
+Import is supported using the following syntax:
+
+{{ codefile "shell" "examples/resources/elasticstack_elasticsearch_ingest_pipeline/import.sh" }}