Fix HTTP headers in the Tasks API

This commit fixes an issue with the Tasks API. When passing a HTTP
header in the request, it couldn't be deserialized in the HTTP response.

Furthermore, we now add a feature to retrieve the HTTP header from the
HTTP response header, not from its body as well. This is described
here:
https://www.elastic.co/guide/en/elasticsearch/reference/6.4/tasks.html#_identifying_running_tasks.

Close #906

Author: olivere

setup_test.go

@@ -17,6 +17,7 @@ const (
 	testIndexName  = "elastic-test"
 	testIndexName2 = "elastic-test2"
 	testIndexName3 = "elastic-test3"
+	testIndexName4 = "elastic-test4"
 	testMapping    = `
 {
 	"settings":{

tasks_get_task.go

@@ -3,6 +3,7 @@ package elastic
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"net/url"
 
 	"github.com/olivere/elastic/uritemplates"
@@ -17,6 +18,7 @@ type TasksGetTaskService struct {
 	pretty            bool
 	taskId            string
 	waitForCompletion *bool
+	headers           http.Header
 }
 
 // NewTasksGetTaskService creates a new TasksGetTaskService.
@@ -47,6 +49,15 @@ func (s *TasksGetTaskService) WaitForCompletion(waitForCompletion bool) *TasksGe
 	return s
 }
 
+// Header sets headers on the request
+func (s *TasksGetTaskService) Header(name string, value string) *TasksGetTaskService {
+	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 *TasksGetTaskService) Pretty(pretty bool) *TasksGetTaskService {
 	s.pretty = pretty
@@ -94,9 +105,10 @@ func (s *TasksGetTaskService) Do(ctx context.Context) (*TasksGetTaskResponse, er
 
 	// Get HTTP response
 	res, err := s.client.PerformRequest(ctx, PerformRequestOptions{
-		Method: "GET",
-		Path:   path,
-		Params: params,
+		Method:  "GET",
+		Path:    path,
+		Params:  params,
+		Headers: s.headers,
 	})
 	if err != nil {
 		return nil, err
@@ -107,10 +119,12 @@ func (s *TasksGetTaskService) Do(ctx context.Context) (*TasksGetTaskResponse, er
 	if err := s.client.decoder.Decode(res.Body, ret); err != nil {
 		return nil, err
 	}
+	ret.Header = res.Header
 	return ret, nil
 }
 
 type TasksGetTaskResponse struct {
-	Completed bool      `json:"completed"`
-	Task      *TaskInfo `json:"task,omitempty"`
+	Header    http.Header `json:"-"`
+	Completed bool        `json:"completed"`
+	Task      *TaskInfo   `json:"task,omitempty"`
 }

tasks_get_task_test.go

@@ -5,6 +5,7 @@
 package elastic
 
 import (
+	"context"
 	"testing"
 )
 
@@ -32,22 +33,48 @@ func TestTasksGetTaskBuildURL(t *testing.T) {
 	}
 }
 
-/*
 func TestTasksGetTask(t *testing.T) {
-	client := setupTestClientAndCreateIndexAndAddDocs(t)
-	esversion, err := client.ElasticsearchVersion(DefaultURL)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if esversion < "2.3.0" {
-		t.Skipf("Elasticsearch %v does not support Tasks Management API yet", esversion)
+	client := setupTestClientAndCreateIndexAndAddDocs(t) //, SetTraceLog(log.New(os.Stdout, "", 0)))
+
+	// Create a reindexing task
+	var taskID string
+	{
+		res, err := client.Reindex().
+			SourceIndex(testIndexName).
+			DestinationIndex(testIndexName4).
+			DoAsync(context.Background())
+		if err != nil {
+			t.Fatalf("unable to start reindexing task: %v", err)
+		}
+		taskID = res.TaskId
 	}
-	res, err := client.TasksGetTask().TaskId("123").Do(context.TODO())
+
+	// Get the task by ID
+	res, err := client.TasksGetTask().
+		TaskId(taskID).
+		Header("X-Opaque-Id", "987654").
+		Do(context.Background())
 	if err != nil {
 		t.Fatal(err)
 	}
 	if res == nil {
 		t.Fatal("response is nil")
 	}
+	if want, have := "987654", res.Header.Get("X-Opaque-Id"); want != have {
+		t.Fatalf("expected HTTP header %#v; got: %#v", want, have)
+	}
+	if res.Task == nil {
+		t.Fatal("task is nil")
+	}
+	// Elasticsearch <= 6.4.1 doesn't return the X-Opaque-Id in the body,
+	// only in response header.
+	/*
+		have, found := res.Task.Headers["X-Opaque-Id"]
+		if !found {
+			t.Fatalf("expected to find headers[%q]", "X-Opaque-Id")
+		}
+		if want := "987654"; want != have {
+			t.Fatalf("expected headers[%q]=%q; got: %q", "X-Opaque-Id", want, have)
+		}
+	*/
 }
-*/

tasks_list.go

@@ -190,11 +190,13 @@ func (s *TasksListService) Do(ctx context.Context) (*TasksListResponse, error) {
 	if err := s.client.decoder.Decode(res.Body, ret); err != nil {
 		return nil, err
 	}
+	ret.Header = res.Header
 	return ret, nil
 }
 
 // TasksListResponse is the response of TasksListService.Do.
 type TasksListResponse struct {
+	Header       http.Header             `json:"-"`
 	TaskFailures []*TaskOperationFailure `json:"task_failures"`
 	NodeFailures []*FailedNodeException  `json:"node_failures"`
 	// Nodes returns the tasks per node. The key is the node id.
@@ -226,19 +228,19 @@ type DiscoveryNode struct {
 
 // TaskInfo represents information about a currently running task.
 type TaskInfo struct {
-	Node               string      `json:"node"`
-	Id                 int64       `json:"id"` // the task id (yes, this is a long in the Java source)
-	Type               string      `json:"type"`
-	Action             string      `json:"action"`
-	Status             interface{} `json:"status"`      // has separate implementations of Task.Status in Java for reindexing, replication, and "RawTaskStatus"
-	Description        interface{} `json:"description"` // same as Status
-	StartTime          string      `json:"start_time"`
-	StartTimeInMillis  int64       `json:"start_time_in_millis"`
-	RunningTime        string      `json:"running_time"`
-	RunningTimeInNanos int64       `json:"running_time_in_nanos"`
-	Cancellable        bool        `json:"cancellable"`
-	ParentTaskId       string      `json:"parent_task_id"` // like "YxJnVYjwSBm_AUbzddTajQ:12356"
-	Headers            http.Header `json:"headers"`
+	Node               string            `json:"node"`
+	Id                 int64             `json:"id"` // the task id (yes, this is a long in the Java source)
+	Type               string            `json:"type"`
+	Action             string            `json:"action"`
+	Status             interface{}       `json:"status"`      // has separate implementations of Task.Status in Java for reindexing, replication, and "RawTaskStatus"
+	Description        interface{}       `json:"description"` // same as Status
+	StartTime          string            `json:"start_time"`
+	StartTimeInMillis  int64             `json:"start_time_in_millis"`
+	RunningTime        string            `json:"running_time"`
+	RunningTimeInNanos int64             `json:"running_time_in_nanos"`
+	Cancellable        bool              `json:"cancellable"`
+	ParentTaskId       string            `json:"parent_task_id"` // like "YxJnVYjwSBm_AUbzddTajQ:12356"
+	Headers            map[string]string `json:"headers"`
 }
 
 // StartTaskResult is used in cases where a task gets started asynchronously and

tasks_list_test.go

@@ -46,7 +46,11 @@ func TestTasksListBuildURL(t *testing.T) {
 
 func TestTasksList(t *testing.T) {
 	client := setupTestClientAndCreateIndexAndAddDocs(t) //, SetTraceLog(log.New(os.Stdout, "", 0)))
-	res, err := client.TasksList().Pretty(true).Human(true).Do(context.TODO())
+	res, err := client.TasksList().
+		Pretty(true).
+		Human(true).
+		Header("X-Opaque-Id", "123456").
+		Do(context.TODO())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -56,4 +60,21 @@ func TestTasksList(t *testing.T) {
 	if len(res.Nodes) == 0 {
 		t.Fatalf("expected at least 1 node; got: %d", len(res.Nodes))
 	}
+	if want, have := "123456", res.Header.Get("X-Opaque-Id"); want != have {
+		t.Fatalf("expected HTTP header %#v; got: %#v", want, have)
+	}
+	for _, node := range res.Nodes {
+		if len(node.Tasks) == 0 {
+			t.Fatalf("expected at least 1 task; got: %d", len(node.Tasks))
+		}
+		for _, task := range node.Tasks {
+			have, found := task.Headers["X-Opaque-Id"]
+			if !found {
+				t.Fatalf("expected to find headers[%q]", "X-Opaque-Id")
+			}
+			if want := "123456"; want != have {
+				t.Fatalf("expected headers[%q]=%q; got: %q", "X-Opaque-Id", want, have)
+			}
+		}
+	}
 }