[ISSUE-154] Add Watches

.github/workflows/test.yml

@@ -60,7 +60,9 @@ jobs:
         image: docker.elastic.co/elasticsearch/elasticsearch:${{ matrix.version }}
         env:
           discovery.type: single-node
+          xpack.license.self_generated.type: trial
           xpack.security.enabled: true
+          xpack.watcher.enabled: true
           repositories.url.allowed_urls: https://example.com/*
           path.repo: /tmp
           ELASTIC_PASSWORD: ${{ env.ELASTIC_PASSWORD }}

Makefile

@@ -86,6 +86,8 @@ docker-elasticsearch: docker-network ## Start Elasticsearch single node cluster
 		-p 9200:9200 -p 9300:9300 \
 		-e "discovery.type=single-node" \
 		-e "xpack.security.enabled=true" \
+		-e "xpack.watcher.enabled=true" \
+		-e "xpack.license.self_generated.type=trial" \
 		-e "repositories.url.allowed_urls=https://example.com/*" \
 		-e "path.repo=/tmp" \
 		-e ELASTIC_PASSWORD=$(ELASTICSEARCH_PASSWORD) \

docs/resources/elasticsearch_logstash_pipeline.md

@@ -113,5 +113,5 @@ Optional:
 Import is supported using the following syntax:
 
 ```shell
-terraform import elasticstack_elasticsearch_security_logstash_pipeline.my_pipeline <cluster_uuid>/<pipeline ID>
+terraform import elasticstack_elasticsearch_logstash_pipeline.my_pipeline <cluster_uuid>/<pipeline ID>
 ```

docs/resources/elasticsearch_watcher_watch.md

@@ -0,0 +1,80 @@
+---
+subcategory: "Watcher"
+layout: ""
+page_title: "Elasticstack: elasticstack_elasticsearch_watcher_watch Resource"
+description: |-
+  Adds and manages a Watch.
+---
+
+# Resource: elasticstack_elasticsearch_watcher_watch
+
+Adds and manages a Watch. See: https://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api.html
+
+## Example Usage
+
+```terraform
+provider "elasticstack" {
+  elasticsearch {}
+}
+
+resource "elasticstack_elasticsearch_watcher_watch" "example" {
+  watch_id = "test_watch"
+  active   = true
+
+  body = jsonencode({
+    "trigger" = {
+      "schedule" = {
+        "daily" = {
+          "at" = "noon"
+        }
+      }
+    },
+  })
+}
+
+output "watch" {
+  value = elasticstack_elasticsearch_watcher_watch.example.watch_id
+}
+```
+
+<!-- schema generated by tfplugindocs -->
+## Schema
+
+### Required
+
+- `body` (String) Configuration for the pipeline.
+- `watch_id` (String) Identifier for the watch.
+
+### Optional
+
+- `active` (Boolean) Defines whether the watch is active or inactive by default. The default value is true, which means the watch is active by default.
+- `elasticsearch_connection` (Block List, Max: 1, Deprecated) Elasticsearch connection configuration block. This property will be removed in a future provider version. Configure the Elasticsearch connection via the provider configuration instead. (see [below for nested schema](#nestedblock--elasticsearch_connection))
+
+### Read-Only
+
+- `id` (String) Internal identifier of the resource
+
+<a id="nestedblock--elasticsearch_connection"></a>
+### Nested Schema for `elasticsearch_connection`
+
+Optional:
+
+- `api_key` (String, Sensitive) API Key to use for authentication to Elasticsearch
+- `ca_data` (String) PEM-encoded custom Certificate Authority certificate
+- `ca_file` (String) Path to a custom Certificate Authority certificate
+- `cert_data` (String) PEM encoded certificate for client auth
+- `cert_file` (String) Path to a file containing the PEM encoded certificate for client auth
+- `endpoints` (List of String, Sensitive) A list of endpoints where the terraform provider will point to, this must include the http(s) schema and port number.
+- `insecure` (Boolean) Disable TLS certificate validation
+- `key_data` (String, Sensitive) PEM encoded private key for client auth
+- `key_file` (String) Path to a file containing the PEM encoded private key for client auth
+- `password` (String, Sensitive) Password to use for API authentication to Elasticsearch.
+- `username` (String) Username to use for API authentication to Elasticsearch.
+
+## Import
+
+Import is supported using the following syntax:
+
+```shell
+terraform import elasticstack_elasticsearch_watcher_watch.watch_id <cluster_uuid>/<watch ID>
+```

examples/resources/elasticstack_elasticsearch_logstash_pipeline/import.sh

@@ -1 +1 @@
-terraform import elasticstack_elasticsearch_security_logstash_pipeline.my_pipeline <cluster_uuid>/<pipeline ID>
+terraform import elasticstack_elasticsearch_logstash_pipeline.my_pipeline <cluster_uuid>/<pipeline ID>

examples/resources/elasticstack_elasticsearch_watcher_watch/import.sh

@@ -0,0 +1 @@
+terraform import elasticstack_elasticsearch_watcher_watch.watch_id <cluster_uuid>/<watch ID>

examples/resources/elasticstack_elasticsearch_watcher_watch/resource.tf

@@ -0,0 +1,22 @@
+provider "elasticstack" {
+  elasticsearch {}
+}
+
+resource "elasticstack_elasticsearch_watcher_watch" "example" {
+  watch_id = "test_watch"
+  active   = true
+
+  body = jsonencode({
+    "trigger" = {
+      "schedule" = {
+        "daily" = {
+          "at" = "noon"
+        }
+      }
+    },
+  })
+}
+
+output "watch" {
+  value = elasticstack_elasticsearch_watcher_watch.example.watch_id
+}

internal/clients/elasticsearch/watch.go

@@ -0,0 +1,111 @@
+package elasticsearch
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	"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"
+)
+
+func PutWatch(ctx context.Context, apiClient *clients.ApiClient, watch *models.Watch) diag.Diagnostics {
+	var diags diag.Diagnostics
+
+	esClient, err := apiClient.GetESClient()
+	if err != nil {
+		return diag.FromErr(err)
+	}
+
+	watchBytes, err := json.Marshal(watch.Body)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+
+	body := esClient.Watcher.PutWatch.WithBody(bytes.NewReader(watchBytes))
+
+	res, err := esClient.Watcher.PutWatch(watch.WatchID, body, esClient.Watcher.PutWatch.WithContext(ctx))
+	if err != nil {
+		return diag.FromErr(err)
+	}
+	defer res.Body.Close()
+	if diags := utils.CheckError(res, "Unable to create or update watch"); diags.HasError() {
+		return diags
+	}
+
+	if watch.Active {
+		_, err := esClient.Watcher.ActivateWatch(watch.WatchID)
+		if err != nil {
+			return diag.FromErr(err)
+		}
+	} else {
+		_, err := esClient.Watcher.DeactivateWatch(watch.WatchID)
+		if err != nil {
+			return diag.FromErr(err)
+		}
+	}
+
+	return diags
+}
+
+func GetWatch(ctx context.Context, apiClient *clients.ApiClient, watchID string) (*models.Watch, diag.Diagnostics) {
+	var diags diag.Diagnostics
+
+	esClient, err := apiClient.GetESClient()
+	if err != nil {
+		return nil, diag.FromErr(err)
+	}
+
+	res, err := esClient.Watcher.GetWatch(watchID, esClient.Watcher.GetWatch.WithContext(ctx))
+	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, "Unable to find watch on cluster."); diags.HasError() {
+		return nil, diags
+	}
+
+	watch := make(map[string]models.Watch)
+	watchBody := make(map[string]interface{})
+	if err := json.NewDecoder(res.Body).Decode(&watchBody); err != nil {
+		return nil, diag.FromErr(err)
+	}
+
+	if watch, ok := watch[watchID]; ok {
+		watch.WatchID = watchID
+		watch.Body = watchBody
+		return &watch, diags
+	}
+
+	diags = append(diags, diag.Diagnostic{
+		Severity: diag.Error,
+		Summary:  "Unable to find watch in the cluster",
+		Detail:   fmt.Sprintf(`Unable to find "%s" watch in the cluster`, watchID),
+	})
+	return nil, diags
+}
+
+func DeleteWatch(ctx context.Context, apiClient *clients.ApiClient, watchID string) diag.Diagnostics {
+	var diags diag.Diagnostics
+	esClient, err := apiClient.GetESClient()
+	if err != nil {
+		return diag.FromErr(err)
+	}
+	res, err := esClient.Watcher.DeleteWatch(watchID, esClient.Watcher.DeleteWatch.WithContext(ctx))
+
+	if err != nil && res.IsError() {
+		return diag.FromErr(err)
+	}
+	defer res.Body.Close()
+	if diags := utils.CheckError(res, "Unable to delete watch"); diags.HasError() {
+		return diags
+	}
+	return diags
+}

internal/elasticsearch/watcher/watch.go

@@ -0,0 +1,146 @@
+package watcher
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/elastic/terraform-provider-elasticstack/internal/clients"
+	"github.com/elastic/terraform-provider-elasticstack/internal/clients/elasticsearch"
+	"github.com/elastic/terraform-provider-elasticstack/internal/models"
+	"github.com/elastic/terraform-provider-elasticstack/internal/utils"
+	"github.com/hashicorp/terraform-plugin-log/tflog"
+	"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 ResourceWatch() *schema.Resource {
+	watchSchema := map[string]*schema.Schema{
+		"id": {
+			Description: "Internal identifier of the resource",
+			Type:        schema.TypeString,
+			Computed:    true,
+		},
+		"watch_id": {
+			Description: "Identifier for the watch.",
+			Type:        schema.TypeString,
+			Required:    true,
+			ForceNew:    true,
+		},
+		"active": {
+			Description: "Defines whether the watch is active or inactive by default. The default value is true, which means the watch is active by default.",
+			Type:        schema.TypeBool,
+			Optional:    true,
+			Default:     true,
+		},
+		"body": {
+			Description:      "Configuration for the pipeline.",
+			Type:             schema.TypeString,
+			ValidateFunc:     validation.StringIsJSON,
+			DiffSuppressFunc: utils.DiffJsonSuppress,
+			Required:         true,
+		},
+	}
+
+	utils.AddConnectionSchema(watchSchema)
+
+	return &schema.Resource{
+		Description: "Manage Watches. See, https://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api.html",
+
+		CreateContext: resourceWatchPut,
+		UpdateContext: resourceWatchPut,
+		ReadContext:   resourceWatchRead,
+		DeleteContext: resourceWatchDelete,
+
+		Importer: &schema.ResourceImporter{
+			StateContext: schema.ImportStatePassthroughContext,
+		},
+
+		Schema: watchSchema,
+	}
+}
+
+func resourceWatchPut(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	client, diags := clients.NewApiClient(d, meta)
+	if diags.HasError() {
+		return diags
+	}
+
+	watchID := d.Get("watch_id").(string)
+	id, diags := client.ID(ctx, watchID)
+	if diags.HasError() {
+		return diags
+	}
+
+	var watchBody map[string]interface{}
+	if err := json.Unmarshal([]byte(d.Get("body").(string)), &watchBody); err != nil {
+		return diag.FromErr(err)
+	}
+
+	watch := models.Watch{
+		WatchID: watchID,
+		Active:  d.Get("active").(bool),
+		Body:    watchBody,
+	}
+
+	if diags := elasticsearch.PutWatch(ctx, client, &watch); diags.HasError() {
+		return diags
+	}
+
+	d.SetId(id.String())
+	return resourceWatchRead(ctx, d, meta)
+}
+
+func resourceWatchRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	client, diags := clients.NewApiClient(d, meta)
+	if diags.HasError() {
+		return diags
+	}
+	resourceID, diags := clients.ResourceIDFromStr(d.Id())
+	if diags.HasError() {
+		return diags
+	}
+
+	watch, diags := elasticsearch.GetWatch(ctx, client, resourceID)
+	if watch == nil && diags == nil {
+		tflog.Warn(ctx, fmt.Sprintf(`Watch "%s" not found, removing from state`, resourceID))
+		d.SetId("")
+		return diags
+	}
+	if diags.HasError() {
+		return diags
+	}
+
+	if err := d.Set("watch_id", watch.WatchID); err != nil {
+		return diag.FromErr(err)
+	}
+	if err := d.Set("active", watch.Active); err != nil {
+		return diag.FromErr(err)
+	}
+	body, err := json.Marshal(watch.Body)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+	if err := d.Set("body", string(body)); err != nil {
+		return diag.FromErr(err)
+	}
+
+	return nil
+}
+
+func resourceWatchDelete(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	client, diags := clients.NewApiClient(d, meta)
+	if diags.HasError() {
+		return diags
+	}
+	resourceID, diags := clients.ResourceIDFromStr(d.Id())
+	if diags.HasError() {
+		return diags
+	}
+
+	if diags := elasticsearch.DeleteWatch(ctx, client, resourceID); diags.HasError() {
+		return diags
+	}
+	return nil
+}