elasticsearch_client.diff

diff --git a/elasticsearch/client.go b/elasticsearch/client.go
index e7df0cde..712a1ba3 100644
--- a/elasticsearch/client.go
+++ b/elasticsearch/client.go
@@ -18,15 +18,9 @@
 package elasticsearch
 
 import (
-	"bytes"
 	"context"
-	"encoding/json"
-	"errors"
 	"io"
-	"io/ioutil"
 	"net/http"
-	"net/url"
-	"strings"
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/version"
@@ -38,26 +32,26 @@ import (
 
 // Client is an interface designed to abstract away version differences between elasticsearch clients
 type Client interface {
+	// Perform satisfies esapi.Transport.
+	Perform(*http.Request) (*http.Response, error)
+
 	// TODO: deprecate
 	// Search performs a query against the given index with the given body
 	Search(index string, body io.Reader) (int, io.ReadCloser, error)
-	// Makes a request with application/json Content-Type and Accept headers by default
-	// pass/overwrite headers with "key: value" format
-	JSONRequest(method, path string, body interface{}, headers ...string) JSONResponse
 }
 
 type clientV8 struct {
-	v8 *esv8.Client
+	*esv8.Client
 }
 
 // Search satisfies the Client interface for version 8
 func (c clientV8) Search(index string, body io.Reader) (int, io.ReadCloser, error) {
-	response, err := c.v8.Search(
-		c.v8.Search.WithContext(context.Background()),
-		c.v8.Search.WithIndex(index),
-		c.v8.Search.WithBody(body),
-		c.v8.Search.WithTrackTotalHits(true),
-		c.v8.Search.WithPretty(),
+	response, err := c.Client.Search(
+		c.Client.Search.WithContext(context.Background()),
+		c.Client.Search.WithIndex(index),
+		c.Client.Search.WithBody(body),
+		c.Client.Search.WithTrackTotalHits(true),
+		c.Client.Search.WithPretty(),
 	)
 	if err != nil {
 		return 0, nil, err
@@ -65,26 +59,18 @@ func (c clientV8) Search(index string, body io.Reader) (int, io.ReadCloser, erro
 	return response.StatusCode, response.Body, nil
 }
 
-func (c clientV8) JSONRequest(method, path string, body interface{}, headers ...string) JSONResponse {
-	req, err := makeRequest(method, path, body, headers...)
-	if err != nil {
-		return JSONResponse{nil, err}
-	}
-	return parseResponse(c.v8.Perform(req))
-}
-
 type clientV7 struct {
-	v7 *esv7.Client
+	*esv7.Client
 }
 
 // Search satisfies the Client interface for version 7
 func (c clientV7) Search(index string, body io.Reader) (int, io.ReadCloser, error) {
-	response, err := c.v7.Search(
-		c.v7.Search.WithContext(context.Background()),
-		c.v7.Search.WithIndex(index),
-		c.v7.Search.WithBody(body),
-		c.v7.Search.WithTrackTotalHits(true),
-		c.v7.Search.WithPretty(),
+	response, err := c.Client.Search(
+		c.Client.Search.WithContext(context.Background()),
+		c.Client.Search.WithIndex(index),
+		c.Client.Search.WithBody(body),
+		c.Client.Search.WithTrackTotalHits(true),
+		c.Client.Search.WithPretty(),
 	)
 	if err != nil {
 		return 0, nil, err
@@ -92,14 +78,6 @@ func (c clientV7) Search(index string, body io.Reader) (int, io.ReadCloser, erro
 	return response.StatusCode, response.Body, nil
 }
 
-func (c clientV7) JSONRequest(method, path string, body interface{}, headers ...string) JSONResponse {
-	req, err := makeRequest(method, path, body, headers...)
-	if err != nil {
-		return JSONResponse{nil, err}
-	}
-	return parseResponse(c.v7.Perform(req))
-}
-
 // NewClient parses the given config and returns  a version-aware client as an interface
 func NewClient(config *Config) (Client, error) {
 	if config == nil {
@@ -142,59 +120,3 @@ func newV8Client(apikey, user, pwd string, addresses []string, transport http.Ro
 		Transport: transport,
 	})
 }
-
-type JSONResponse struct {
-	content io.ReadCloser
-	err     error
-}
-
-func (r JSONResponse) DecodeTo(i interface{}) error {
-	if r.err != nil {
-		return r.err
-	}
-	defer r.content.Close()
-	err := json.NewDecoder(r.content).Decode(&i)
-	return err
-}
-
-// each header has the format "key: value"
-func makeRequest(method, path string, body interface{}, headers ...string) (*http.Request, error) {
-	header := http.Header{
-		"Content-Type": []string{"application/json"},
-		"Accept":       []string{"application/json"},
-	}
-	for _, h := range headers {
-		kv := strings.Split(h, ":")
-		if len(kv) == 2 {
-			header[kv[0]] = strings.Split(kv[1], ",")
-		}
-	}
-	u, _ := url.Parse(path)
-	req := &http.Request{
-		Method: method,
-		URL:    u,
-		Header: header,
-	}
-	bs, err := json.Marshal(body)
-	if err != nil {
-		return nil, err
-	}
-	if body != nil {
-		req.Body = ioutil.NopCloser(bytes.NewReader(bs))
-		req.ContentLength = int64(len(bs))
-	}
-	return req, nil
-}
-
-func parseResponse(resp *http.Response, err error) JSONResponse {
-	if err != nil {
-		return JSONResponse{nil, err}
-	}
-	body := resp.Body
-	if resp.StatusCode >= http.StatusMultipleChoices {
-		buf := new(bytes.Buffer)
-		buf.ReadFrom(body)
-		return JSONResponse{nil, errors.New(buf.String())}
-	}
-	return JSONResponse{body, nil}
-}
diff --git a/elasticsearch/security_api.go b/elasticsearch/security_api.go
index cccbda42..0fc0edf6 100644
--- a/elasticsearch/security_api.go
+++ b/elasticsearch/security_api.go
@@ -18,75 +18,82 @@
 package elasticsearch
 
 import (
+	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
-	"net/url"
-	"strconv"
+
+	"github.com/elastic/go-elasticsearch/v7/esapi"
+	"github.com/elastic/go-elasticsearch/v7/esutil"
 )
 
+func doRequest(transport esapi.Transport, out interface{}, req esapi.Request) error {
+	resp, err := req.Do(context.TODO(), transport)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.IsError() {
+		// TODO include response body
+		return fmt.Errorf("request failed with status code %d", resp.StatusCode)
+	}
+	if out != nil {
+		return json.NewDecoder(resp.Body).Decode(out)
+	}
+	return nil
+}
+
 // CreateAPIKey requires manage_security cluster privilege
 func CreateAPIKey(client Client, apikeyReq CreateApiKeyRequest) (CreateApiKeyResponse, error) {
-	response := client.JSONRequest(http.MethodPut, "/_security/api_key", apikeyReq)
-
-	var apikey CreateApiKeyResponse
-	err := response.DecodeTo(&apikey)
-	return apikey, err
+	var resp CreateApiKeyResponse
+	err := doRequest(client, &resp, esapi.SecurityCreateAPIKeyRequest{Body: esutil.NewJSONReader(apikeyReq)})
+	return resp, err
 }
 
 // GetAPIKeys requires manage_security cluster privilege
 func GetAPIKeys(client Client, apikeyReq GetApiKeyRequest) (GetApiKeyResponse, error) {
-	u := url.URL{Path: "/_security/api_key"}
-	params := url.Values{}
-	params.Set("owner", strconv.FormatBool(apikeyReq.Owner))
+	req := esapi.SecurityGetAPIKeyRequest{Owner: &apikeyReq.Owner}
 	if apikeyReq.Id != nil {
-		params.Set("id", *apikeyReq.Id)
-	} else if apikeyReq.Name != nil {
-		params.Set("name", *apikeyReq.Name)
+		req.ID = *apikeyReq.Id
 	}
-	u.RawQuery = params.Encode()
-
-	response := client.JSONRequest(http.MethodGet, u.String(), nil)
-
-	var apikey GetApiKeyResponse
-	err := response.DecodeTo(&apikey)
-	return apikey, err
+	if apikeyReq.Name != nil {
+		req.Name = *apikeyReq.Name
+	}
+	var resp GetApiKeyResponse
+	err := doRequest(client, &resp, req)
+	return resp, err
 }
 
 // CreatePrivileges requires manage_security cluster privilege
 func CreatePrivileges(client Client, privilegesReq CreatePrivilegesRequest) (CreatePrivilegesResponse, error) {
-	response := client.JSONRequest(http.MethodPut, "/_security/privilege", privilegesReq)
-
-	var privileges CreatePrivilegesResponse
-	err := response.DecodeTo(&privileges)
-	return privileges, err
+	var resp CreatePrivilegesResponse
+	err := doRequest(client, &resp, esapi.SecurityPutPrivilegesRequest{Body: esutil.NewJSONReader(privilegesReq)})
+	return resp, err
 }
 
 // InvalidateAPIKey requires manage_security cluster privilege
 func InvalidateAPIKey(client Client, apikeyReq InvalidateApiKeyRequest) (InvalidateApiKeyResponse, error) {
-	response := client.JSONRequest(http.MethodDelete, "/_security/api_key", apikeyReq)
-
-	var confirmation InvalidateApiKeyResponse
-	err := response.DecodeTo(&confirmation)
-	return confirmation, err
+	var resp InvalidateApiKeyResponse
+	err := doRequest(client, &resp, esapi.SecurityInvalidateAPIKeyRequest{Body: esutil.NewJSONReader(apikeyReq)})
+	return resp, err
 }
 
 // DeletePrivileges requires manage_security cluster privilege
 func DeletePrivileges(client Client, privilegesReq DeletePrivilegeRequest) (DeletePrivilegeResponse, error) {
-	path := fmt.Sprintf("/_security/privilege/%v/%v", privilegesReq.Application, privilegesReq.Privilege)
-	response := client.JSONRequest(http.MethodDelete, path, nil)
-
-	var confirmation DeletePrivilegeResponse
-	err := response.DecodeTo(&confirmation)
-	return confirmation, err
+	var resp DeletePrivilegeResponse
+	err := doRequest(client, &resp, esapi.SecurityDeletePrivilegesRequest{
+		Application: string(privilegesReq.Application),
+		Name:        string(privilegesReq.Privilege),
+	})
+	return resp, err
 }
 
 func HasPrivileges(client Client, privileges HasPrivilegesRequest, credentials string) (HasPrivilegesResponse, error) {
-	h := fmt.Sprintf("Authorization: ApiKey %s", credentials)
-	response := client.JSONRequest(http.MethodGet, "/_security/user/_has_privileges", privileges, h)
-
-	var info HasPrivilegesResponse
-	err := response.DecodeTo(&info)
-	return info, err
+	var resp HasPrivilegesResponse
+	header := make(http.Header)
+	header.Set("Authorization", "ApiKey "+credentials)
+	err := doRequest(client, &resp, esapi.SecurityHasPrivilegesRequest{Header: header})
+	return resp, err
 }
 
 type CreateApiKeyRequest struct {