[opendistro destination] Fix datasource get in odfe>=1.11.0

Author: phillbaker

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Changelog
 ## Unreleased
 ### Changed
+- [opendistro destination] Use API to get in odfe>=1.11.0 (#158)
 
 ### Added
 

es/data_source_elasticsearch_opendistro_destination.go

@@ -8,6 +8,7 @@ import (
 	"log"
 
 	"github.com/hashicorp/terraform-plugin-sdk/helper/schema"
+	"github.com/olivere/elastic/uritemplates"
 
 	elastic7 "github.com/olivere/elastic/v7"
 	elastic6 "gopkg.in/olivere/elastic.v6"
@@ -47,19 +48,26 @@ func dataSourceElasticsearchOpenDistroDestination() *schema.Resource {
 func dataSourceElasticsearchOpenDistroDestinationRead(d *schema.ResourceData, m interface{}) error {
 	destinationName := d.Get("name").(string)
 
-	// See https://github.com/opendistro-for-elasticsearch/alerting/issues/70, no tags or API endpoint for searching destination
 	var id string
-	var body *json.RawMessage
+	var destination map[string]interface{}
 	var err error
 	esClient, err := getClient(m.(*ProviderConf))
 	if err != nil {
 		return err
 	}
 	switch client := esClient.(type) {
 	case *elastic7.Client:
-		id, body, err = destinationElasticsearch7Search(client, DESTINATION_INDEX, destinationName)
+		// See https://github.com/opendistro-for-elasticsearch/alerting/issues/70,
+		// no tags or API endpoint for searching destination. In ODFE >= 1.11.0,
+		// the index has become a "system index", so it cannot be searched:
+		// https://opendistro.github.io/for-elasticsearch-docs/docs/alerting/settings/#alerting-indices
+		// instead we paginate through all destinations to find the first name match :|
+		id, destination, err = destinationElasticsearch7GetAll(client, destinationName)
+		if err != nil {
+			id, destination, err = destinationElasticsearch7Search(client, DESTINATION_INDEX, destinationName)
+		}
 	case *elastic6.Client:
-		id, body, err = destinationElasticsearch6Search(client, DESTINATION_INDEX, destinationName)
+		id, destination, err = destinationElasticsearch6Search(client, DESTINATION_INDEX, destinationName)
 	default:
 		err = errors.New("destination resource not implemented prior to Elastic v6")
 	}
@@ -71,17 +79,13 @@ func dataSourceElasticsearchOpenDistroDestinationRead(d *schema.ResourceData, m
 		return nil
 	}
 
-	destination := make(map[string]interface{})
-	if err := json.Unmarshal(*body, &destination); err != nil {
-		return fmt.Errorf("error unmarshalling destination body: %+v: %+v", err, body)
-	}
-
 	d.SetId(id)
 
 	// we get a non-uniform map[string]interface{} back for the body, terraform
-	// only accepts a mapping of string to primitive values
+	// only accepts a mapping of string to primitive values. We want to save
+	// this as a map so that attributes are accessible
 	simplifiedBody := map[string]string{}
-	for key, value := range destination["destination"].(map[string]interface{}) {
+	for key, value := range destination {
 		if stringified, ok := value.(string); ok {
 			simplifiedBody[key] = stringified
 		} else {
@@ -92,40 +96,99 @@ func dataSourceElasticsearchOpenDistroDestinationRead(d *schema.ResourceData, m
 	return err
 }
 
-func destinationElasticsearch7Search(client *elastic7.Client, index string, name string) (string, *json.RawMessage, error) {
+func destinationElasticsearch7Search(client *elastic7.Client, index string, name string) (string, map[string]interface{}, error) {
 	termQuery := elastic7.NewTermQuery(DESTINATION_NAME_FIELD, name)
 	result, err := client.Search().
 		Index(index).
 		Query(termQuery).
 		Do(context.TODO())
 
+	destination := make(map[string]interface{})
 	if err != nil {
-		return "", nil, err
+		return "", destination, err
 	}
 	if result.TotalHits() == 1 {
-		return result.Hits.Hits[0].Id, &result.Hits.Hits[0].Source, nil
+		if err := json.Unmarshal(result.Hits.Hits[0].Source, &destination); err != nil {
+			return "", destination, fmt.Errorf("error unmarshalling destination body: %+v", err)
+		}
+
+		return result.Hits.Hits[0].Id, destination["destination"].(map[string]interface{}), nil
 	} else if result.TotalHits() < 1 {
-		return "", nil, err
+		return "", destination, err
 	} else {
-		return "", nil, fmt.Errorf("1 result expected, found %d.", result.TotalHits())
+		return "", destination, fmt.Errorf("1 result expected, found %d.", result.TotalHits())
 	}
 }
 
-func destinationElasticsearch6Search(client *elastic6.Client, index string, name string) (string, *json.RawMessage, error) {
+func destinationElasticsearch6Search(client *elastic6.Client, index string, name string) (string, map[string]interface{}, error) {
 	termQuery := elastic6.NewTermQuery(DESTINATION_NAME_FIELD, name)
 	result, err := client.Search().
 		Index(index).
 		Query(termQuery).
 		Do(context.TODO())
 
+	destination := make(map[string]interface{})
 	if err != nil {
-		return "", nil, err
+		return "", destination, err
 	}
 	if result.TotalHits() == 1 {
-		return result.Hits.Hits[0].Id, result.Hits.Hits[0].Source, nil
+		if err := json.Unmarshal(*result.Hits.Hits[0].Source, &destination); err != nil {
+			return "", destination, fmt.Errorf("error unmarshalling destination body: %+v", err)
+		}
+
+		return result.Hits.Hits[0].Id, destination["destination"].(map[string]interface{}), nil
 	} else if result.TotalHits() < 1 {
-		return "", nil, err
+		return "", destination, err
 	} else {
-		return "", nil, fmt.Errorf("1 result expected, found %d.", result.TotalHits())
+		return "", destination, fmt.Errorf("1 result expected, found %d.", result.TotalHits())
 	}
 }
+
+func destinationElasticsearch7GetAll(client *elastic7.Client, name string) (string, map[string]interface{}, error) {
+	offset := 0
+	pageSize := 1000
+	destination := make(map[string]interface{})
+	for {
+		path, err := uritemplates.Expand("/_opendistro/_alerting/destinations?startIndex={startIndex}&size={size}", map[string]string{
+			"startIndex": fmt.Sprint(offset),
+			"size":       fmt.Sprint(pageSize),
+		})
+		if err != nil {
+			return "", destination, fmt.Errorf("error building URL path for destination: %+v", err)
+		}
+
+		httpResponse, err := client.PerformRequest(context.TODO(), elastic7.PerformRequestOptions{
+			Method: "GET",
+			Path:   path,
+		})
+		if err != nil {
+			return "", destination, err
+		}
+
+		var drg destinationResponseGet
+		if err := json.Unmarshal(httpResponse.Body, &drg); err != nil {
+			return "", destination, fmt.Errorf("error unmarshalling destination body: %+v", err)
+		}
+
+		for _, d := range drg.Destinations {
+			if d.Name == name {
+				j, err := json.Marshal(d)
+				if err != nil {
+					return "", destination, fmt.Errorf("error marshalling destination: %+v", err)
+				}
+				if err := json.Unmarshal(j, &destination); err != nil {
+					return "", destination, fmt.Errorf("error unmarshalling destination body: %+v", err)
+				}
+				return d.ID, destination, nil
+			}
+		}
+
+		if drg.Total > offset {
+			offset += pageSize
+		} else {
+			break
+		}
+	}
+
+	return "", destination, fmt.Errorf("destination not found")
+}