Skip to content

Commit d6c1337

Browse files
committed
Add polymorphic Status field to task response structs
Add Status field (json.RawMessage) to TasksGetResp, TasksListTask, and TaskCancelInfo for polymorphic task status data. Add typed status structs matching the OpenSearch API specification: BulkByScrollTaskStatus, ReplicationTaskStatus, ResyncTaskStatus, PersistentTaskStatus. Add Parse* helpers and BulkByScrollTaskStatusOrException for sliced task status. Ref: opensearch-project#803 Signed-off-by: Sean Chittenden <sean.chittenden@crowdstrike.com>
1 parent 0b1f8c8 commit d6c1337

9 files changed

Lines changed: 973 additions & 0 deletions

File tree

CHANGELOG.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
1212
- Add per-attempt `RequestTimeout` to bound individual HTTP round-trips, preventing indefinite hangs on stalled connections ([#786](https://github.com/opensearch-project/opensearch-go/issues/786))
1313
- Add `opensearchutil/shardhash` package with exported `Hash` and `ForRouting` functions for computing OpenSearch shard routing
1414
- Enhanced cluster readiness checking for improved test reliability: `testutil.NewClient()` now includes readiness validation (health + cluster state + nodes info)
15+
- Add `Status` field (`json.RawMessage`) to `TasksGetResp`, `TasksListTask`, and `TaskCancelInfo` for polymorphic task status data; add typed status structs matching the OpenSearch API specification: `BulkByScrollTaskStatus`, `ReplicationTaskStatus`, `ResyncTaskStatus`, `PersistentTaskStatus`; add `Parse*` helpers and `BulkByScrollTaskStatusOrException` for sliced task status ([#788](https://github.com/opensearch-project/opensearch-go/issues/788))
1516
- Test parallelization support via TEST_PARALLEL environment variable (default: CPU cores - 1, minimum 1)
1617
- opensearchapi/testutil package with test suite, client helpers, and JSON comparison utilities
1718
- opensearchtransport/testutil package with PollUntil helper for eventual consistency testing (ISM policies, index readiness, cluster state changes)

_samples/tasks.go

Lines changed: 143 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,143 @@
1+
// SPDX-License-Identifier: Apache-2.0
2+
//
3+
// The OpenSearch Contributors require contributions made to
4+
// this file be licensed under the Apache-2.0 license or a
5+
// compatible open source license.
6+
package main
7+
8+
import (
9+
"context"
10+
"encoding/json"
11+
"fmt"
12+
"os"
13+
"strings"
14+
"time"
15+
16+
"github.com/opensearch-project/opensearch-go/v4"
17+
"github.com/opensearch-project/opensearch-go/v4/opensearchapi"
18+
)
19+
20+
func main() {
21+
if err := example(); err != nil {
22+
fmt.Println(fmt.Sprintf("Error: %s", err))
23+
os.Exit(1)
24+
}
25+
}
26+
27+
func example() error {
28+
// Initialize the client with SSL/TLS enabled.
29+
client, err := opensearchapi.NewClient(
30+
opensearchapi.Config{
31+
Client: opensearch.Config{
32+
InsecureSkipVerify: true, // For testing only. Use certificate for validation.
33+
Addresses: []string{"https://localhost:9200"},
34+
Username: "admin", // For testing only. Don't store credentials in code.
35+
Password: "myStrongPassword123!",
36+
},
37+
},
38+
)
39+
if err != nil {
40+
return err
41+
}
42+
43+
ctx := context.Background()
44+
45+
sourceIndex := "task-source"
46+
destIndex := "task-dest"
47+
48+
// Create source index with test data.
49+
_, err = client.Indices.Create(ctx, opensearchapi.IndicesCreateReq{
50+
Index: sourceIndex,
51+
Body: strings.NewReader(`{"settings": {"number_of_shards": 1, "number_of_replicas": 0}}`),
52+
})
53+
if err != nil {
54+
return err
55+
}
56+
57+
_, err = client.Index(ctx, opensearchapi.IndexReq{
58+
Index: sourceIndex,
59+
Body: strings.NewReader(`{"title": "Test Document", "year": 2024}`),
60+
Params: opensearchapi.IndexParams{
61+
Refresh: "true",
62+
},
63+
})
64+
if err != nil {
65+
return err
66+
}
67+
68+
// Submit an async reindex task.
69+
reindexResp, err := client.Reindex(ctx, opensearchapi.ReindexReq{
70+
Body: strings.NewReader(fmt.Sprintf(
71+
`{"source":{"index":"%s"},"dest":{"index":"%s"}}`,
72+
sourceIndex, destIndex,
73+
)),
74+
Params: opensearchapi.ReindexParams{
75+
WaitForCompletion: opensearchapi.ToPointer(false),
76+
},
77+
})
78+
if err != nil {
79+
return err
80+
}
81+
taskID := reindexResp.Task
82+
fmt.Printf("Task submitted: %s\n", taskID)
83+
84+
// Poll for completion.
85+
var taskResp *opensearchapi.TasksGetResp
86+
for {
87+
taskResp, err = client.Tasks.Get(ctx, opensearchapi.TasksGetReq{TaskID: taskID})
88+
if err != nil {
89+
return err
90+
}
91+
if taskResp.Completed {
92+
break
93+
}
94+
time.Sleep(500 * time.Millisecond)
95+
}
96+
fmt.Printf("Task completed: action=%s\n", taskResp.Task.Action)
97+
98+
// Parse the BulkByScroll status.
99+
status, err := opensearchapi.ParseBulkByScrollTaskStatus(taskResp.Task.Status)
100+
if err != nil {
101+
return err
102+
}
103+
104+
fmt.Printf("Total: %d\n", status.Total)
105+
fmt.Printf("Created: %d\n", status.Created)
106+
fmt.Printf("Updated: %d\n", status.Updated)
107+
fmt.Printf("Deleted: %d\n", status.Deleted)
108+
fmt.Printf("Batches: %d\n", status.Batches)
109+
fmt.Printf("Version conflicts: %d\n", status.VersionConflicts)
110+
fmt.Printf("Noops: %d\n", status.Noops)
111+
fmt.Printf("Retries (bulk): %d\n", status.Retries.Bulk)
112+
fmt.Printf("Retries (search): %d\n", status.Retries.Search)
113+
114+
// For tasks without a dedicated type, unmarshal as generic JSON.
115+
if taskResp.Task.Status != nil {
116+
var raw map[string]any
117+
if err := json.Unmarshal(taskResp.Task.Status, &raw); err != nil {
118+
return err
119+
}
120+
fmt.Printf("Raw status: %v\n", raw)
121+
}
122+
123+
// List all running tasks.
124+
listResp, err := client.Tasks.List(ctx, nil)
125+
if err != nil {
126+
return err
127+
}
128+
for nodeID, node := range listResp.Nodes {
129+
fmt.Printf("Node %s (%s): %d tasks\n", node.Name, nodeID, len(node.Tasks))
130+
}
131+
132+
// Cleanup.
133+
delResp, err := client.Indices.Delete(ctx, opensearchapi.IndicesDeleteReq{
134+
Indices: []string{sourceIndex, destIndex},
135+
Params: opensearchapi.IndicesDeleteParams{IgnoreUnavailable: opensearchapi.ToPointer(true)},
136+
})
137+
if err != nil {
138+
return err
139+
}
140+
fmt.Printf("Deleted: %t\n", delResp.Acknowledged)
141+
142+
return nil
143+
}

0 commit comments

Comments
 (0)