reindex: Add ability to add HTTP headers to request

Close #1031

Author: olivere

delete_by_query.go

@@ -7,6 +7,7 @@ package elastic
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"net/url"
 	"strings"
 
@@ -691,16 +692,17 @@ func (s *DeleteByQueryService) DoAsync(ctx context.Context) (*StartTaskResult, e
 // BulkIndexByScrollResponse is the outcome of executing Do with
 // DeleteByQueryService and UpdateByQueryService.
 type BulkIndexByScrollResponse struct {
-	Took             int64  `json:"took"`
-	SliceId          *int64 `json:"slice_id,omitempty"`
-	TimedOut         bool   `json:"timed_out"`
-	Total            int64  `json:"total"`
-	Updated          int64  `json:"updated,omitempty"`
-	Created          int64  `json:"created,omitempty"`
-	Deleted          int64  `json:"deleted"`
-	Batches          int64  `json:"batches"`
-	VersionConflicts int64  `json:"version_conflicts"`
-	Noops            int64  `json:"noops"`
+	Header           http.Header `json:"-"`
+	Took             int64       `json:"took"`
+	SliceId          *int64      `json:"slice_id,omitempty"`
+	TimedOut         bool        `json:"timed_out"`
+	Total            int64       `json:"total"`
+	Updated          int64       `json:"updated,omitempty"`
+	Created          int64       `json:"created,omitempty"`
+	Deleted          int64       `json:"deleted"`
+	Batches          int64       `json:"batches"`
+	VersionConflicts int64       `json:"version_conflicts"`
+	Noops            int64       `json:"noops"`
 	Retries          struct {
 		Bulk   int64 `json:"bulk"`
 		Search int64 `json:"search"`

reindex.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http"
 	"net/url"
 )
 
@@ -28,6 +29,7 @@ type ReindexService struct {
 	conflicts           string
 	size                *int
 	script              *Script
+	headers             http.Header
 }
 
 // NewReindexService creates a new ReindexService.
@@ -87,6 +89,15 @@ func (s *ReindexService) WaitForCompletion(waitForCompletion bool) *ReindexServi
 	return s
 }
 
+// Header sets headers on the request
+func (s *ReindexService) Header(name string, value string) *ReindexService {
+	if s.headers == nil {
+		s.headers = http.Header{}
+	}
+	s.headers.Add(name, value)
+	return s
+}
+
 // Pretty indicates that the JSON response be indented and human readable.
 func (s *ReindexService) Pretty(pretty bool) *ReindexService {
 	s.pretty = pretty
@@ -286,10 +297,11 @@ func (s *ReindexService) Do(ctx context.Context) (*BulkIndexByScrollResponse, er
 
 	// Get HTTP response
 	res, err := s.client.PerformRequest(ctx, PerformRequestOptions{
-		Method: "POST",
-		Path:   path,
-		Params: params,
-		Body:   body,
+		Method:  "POST",
+		Path:    path,
+		Params:  params,
+		Body:    body,
+		Headers: s.headers,
 	})
 	if err != nil {
 		return nil, err
@@ -300,6 +312,7 @@ func (s *ReindexService) Do(ctx context.Context) (*BulkIndexByScrollResponse, er
 	if err := s.client.decoder.Decode(res.Body, ret); err != nil {
 		return nil, err
 	}
+	ret.Header = res.Header
 	return ret, nil
 }
 
@@ -333,10 +346,11 @@ func (s *ReindexService) DoAsync(ctx context.Context) (*StartTaskResult, error)
 
 	// Get HTTP response
 	res, err := s.client.PerformRequest(ctx, PerformRequestOptions{
-		Method: "POST",
-		Path:   path,
-		Params: params,
-		Body:   body,
+		Method:  "POST",
+		Path:    path,
+		Params:  params,
+		Body:    body,
+		Headers: s.headers,
 	})
 	if err != nil {
 		return nil, err
@@ -347,6 +361,7 @@ func (s *ReindexService) DoAsync(ctx context.Context) (*StartTaskResult, error)
 	if err := s.client.decoder.Decode(res.Body, ret); err != nil {
 		return nil, err
 	}
+	ret.Header = res.Header
 	return ret, nil
 }
 

reindex_test.go

@@ -302,7 +302,12 @@ func TestReindex(t *testing.T) {
 	// Simple copying
 	src := NewReindexSource().Index(testIndexName)
 	dst := NewReindexDestination().Index(testIndexName2)
-	res, err := client.Reindex().Source(src).Destination(dst).Refresh("true").Do(context.TODO())
+	res, err := client.Reindex().
+		Source(src).
+		Destination(dst).
+		Refresh("true").
+		Header("X-Opaque-Id", "987654").
+		Do(context.TODO())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -318,6 +323,9 @@ func TestReindex(t *testing.T) {
 	if res.Created != sourceCount {
 		t.Errorf("expected %d, got %d", sourceCount, res.Created)
 	}
+	if want, have := "987654", res.Header.Get("X-Opaque-Id"); want != have {
+		t.Fatalf("expected HTTP header %#v; got: %#v", want, have)
+	}
 
 	targetCount, err = client.Count(testIndexName2).Do(context.TODO())
 	if err != nil {
@@ -330,13 +338,6 @@ func TestReindex(t *testing.T) {
 
 func TestReindexAsync(t *testing.T) {
 	client := setupTestClientAndCreateIndexAndAddDocs(t) //, SetTraceLog(log.New(os.Stdout, "", 0)))
-	esversion, err := client.ElasticsearchVersion(DefaultURL)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if esversion < "2.3.0" {
-		t.Skipf("Elasticsearch %v does not support Reindex API yet", esversion)
-	}
 
 	sourceCount, err := client.Count(testIndexName).Do(context.TODO())
 	if err != nil {
@@ -361,6 +362,7 @@ func TestReindexAsync(t *testing.T) {
 		Source(src).
 		Destination(dst).
 		Slices("auto").
+		Header("X-Opaque-Id", "987654").
 		DoAsync(context.TODO())
 	if err != nil {
 		t.Fatal(err)
@@ -371,6 +373,9 @@ func TestReindexAsync(t *testing.T) {
 	if res.TaskId == "" {
 		t.Errorf("expected a task id, got %+v", res)
 	}
+	if want, have := "987654", res.Header.Get("X-Opaque-Id"); want != have {
+		t.Fatalf("expected HTTP header %#v; got: %#v", want, have)
+	}
 
 	tasksGetTask := client.TasksGetTask()
 	taskStatus, err := tasksGetTask.TaskId(res.TaskId).Do(context.TODO())

tasks_list.go

@@ -246,5 +246,6 @@ type TaskInfo struct {
 // StartTaskResult is used in cases where a task gets started asynchronously and
 // the operation simply returnes a TaskID to watch for via the Task Management API.
 type StartTaskResult struct {
-	TaskId string `json:"task"`
+	Header http.Header `json:"-"`
+	TaskId string      `json:"task"`
 }