Add ISM Policy (opensearch-project#524)

* ci/opensearch: add ism settings

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

* plugins/ism: add base

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

* plugins/ism: add policies

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

* plugins/ism: add 'Add' function

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

* plugins/ism: add change function

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

* plugins/ism: add explain func

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

* plugins/ism: add remove func

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

* plugins/ism: add retry func

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

* plugins/ism: add api tests

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

* add changelog

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

* makefile: adjust test-integ to work for unrealesed opensearch

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

* opensearchapi: adjust scroll test to match all indices for scroll

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

* opensearchapi: nodes stats add missing IO usage field

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

---------

Signed-off-by: Jakob Hahn <jakob.hahn@hetzner.com>

.ci/opensearch/docker-compose.yml

@@ -17,6 +17,7 @@ services:
       - bootstrap.memory_lock=true
       - path.repo=/usr/share/opensearch/mnt
       - OPENSEARCH_INITIAL_ADMIN_PASSWORD=myStrongPassword123!
+      - plugins.index_state_management.job_interval=1
     ports:
       - "9200:9200"
     user: opensearch

.github/workflows/test-integration-unreleased.yml

@@ -65,7 +65,7 @@ jobs:
 
       - name: Integration test
         working-directory: go-client
-        run: make test-integ race=true
+        run: make test-integ race=true unreleased=true
         env:
           OPENSEARCH_GO_SKIP_JSON_COMPARE: true
 

CHANGELOG.md

@@ -14,7 +14,8 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Adds security plugin ([#507](https://github.com/opensearch-project/opensearch-go/pull/507))
 - Adds security settings to container for security testing ([#507](https://github.com/opensearch-project/opensearch-go/pull/507))
 - Adds cluster.get-certs to copy admin certs out of the container ([#507](https://github.com/opensearch-project/opensearch-go/pull/507))
-- Add the `Fields` field containing stored fields to the `DocumentGetResp` struct (#526)[https://github.com/opensearch-project/opensearch-go/pull/526]
+- Adds the `Fields` field containing stored fields to the `DocumentGetResp` struct (#526)[https://github.com/opensearch-project/opensearch-go/pull/526]
+- Adds ism plugin ([#524](https://github.com/opensearch-project/opensearch-go/pull/524))
 
 ### Changed
 - Uses docker compose v2 instead of v1 ([#506](https://github.com/opensearch-project/opensearch-go/pull/506))

Makefile

@@ -22,13 +22,17 @@ test: test-unit
 test-integ:  ## Run integration tests
 	@printf "\033[2m→ Running integration tests...\033[0m\n"
 	$(eval testintegtags += "integration")
+	$(eval testintegpath = "./...")
 ifdef multinode
 	$(eval testintegtags += "multinode")
 endif
 ifdef race
 	$(eval testintegargs += "-race")
 endif
-	$(eval testintegargs += "-cover" "-tags=$(testintegtags)" "-timeout=1h" "./..." "-args" "-test.gocoverdir=$(PWD)/tmp/integration")
+ifdef unreleased
+	$(eval testintegpath = "./opensearchapi/...")
+endif
+	$(eval testintegargs += "-cover" "-tags=$(testintegtags)" "-timeout=1h" "$(testintegpath)" "-args" "-test.gocoverdir=$(PWD)/tmp/integration")
 	@mkdir -p $(PWD)/tmp/integration
 	@echo "go test -v" $(testintegargs); \
 	go test -v $(testintegargs);

opensearchapi/api_nodes-stats.go

@@ -691,6 +691,9 @@ type NodesStatsResourceUsageStats struct {
 	Timestamp                int64  `json:"timestamp"`
 	CPUUtilizationPercent    string `json:"cpu_utilization_percent"`
 	MemoryUtilizationPercent string `json:"memory_utilization_percent"`
+	IOUsageStats             struct {
+		MaxIOUtilizationPercent string `json:"max_io_utilization_percent"`
+	} `json:"io_usage_stats"`
 }
 
 // NodesStatsSegmentReplicationBackpressure is a sub type of NodesStats containing information about segment replication backpressure

opensearchapi/api_scroll_test.go

@@ -22,19 +22,19 @@ import (
 
 func TestScrollClient(t *testing.T) {
 	client, err := ostest.NewClient()
-	require.Nil(t, err)
+	require.NoError(t, err)
 	failingClient, err := osapitest.CreateFailingClient()
-	require.Nil(t, err)
+	require.NoError(t, err)
 
 	search, err := client.Search(
 		nil,
 		&opensearchapi.SearchReq{
-			Indices: []string{"test*"},
+			Indices: []string{"*"},
 			Params:  opensearchapi.SearchParams{Scroll: 5 * time.Minute},
 		},
 	)
-	require.Nil(t, err)
-	require.NotNil(t, search.ScrollID)
+	require.NoError(t, err)
+	require.NotNil(t, search.ScrollID, "ScrollID is nil")
 
 	type scrollTests struct {
 		Name    string
@@ -86,11 +86,11 @@ func TestScrollClient(t *testing.T) {
 				t.Run(testCase.Name, func(t *testing.T) {
 					res, err := testCase.Results()
 					if testCase.Name == "inspect" {
-						assert.NotNil(t, err)
+						assert.Error(t, err)
 						assert.NotNil(t, res)
 						osapitest.VerifyInspect(t, res.Inspect())
 					} else {
-						require.Nil(t, err)
+						require.NoError(t, err)
 						require.NotNil(t, res)
 						assert.NotNil(t, res.Inspect().Response)
 						ostest.CompareRawJSONwithParsedJSON(t, res, res.Inspect().Response)

plugins/ism/api.go

@@ -0,0 +1,69 @@
+// SPDX-License-Identifier: Apache-2.0
+//
+// The OpenSearch Contributors require contributions made to
+// this file be licensed under the Apache-2.0 license or a
+// compatible open source license.
+
+package ism
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/opensearch-project/opensearch-go/v3"
+)
+
+// Config represents the client configuration
+type Config struct {
+	Client opensearch.Config
+}
+
+// Client represents the ism Client summarizing all API calls
+type Client struct {
+	Client   *opensearch.Client
+	Policies policiesClient
+}
+
+// clientInit inits the Client with all sub clients
+func clientInit(rootClient *opensearch.Client) *Client {
+	client := &Client{
+		Client: rootClient,
+	}
+	client.Policies = policiesClient{apiClient: client}
+	return client
+}
+
+// NewClient returns a ism client
+func NewClient(config Config) (*Client, error) {
+	rootClient, err := opensearch.NewClient(config.Client)
+	if err != nil {
+		return nil, err
+	}
+
+	return clientInit(rootClient), nil
+}
+
+// do calls the opensearch.Client.Do() and checks the response for response errors
+func (c *Client) do(ctx context.Context, req opensearch.Request, dataPointer any) (*opensearch.Response, error) {
+	resp, err := c.Client.Do(ctx, req, dataPointer)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.IsError() {
+		if dataPointer != nil {
+			return resp, opensearch.ParseError(resp)
+		} else {
+			return resp, fmt.Errorf("status: %s", resp.Status())
+		}
+	}
+
+	return resp, nil
+}
+
+// FailedIndex contains information about fieled actions
+type FailedIndex struct {
+	IndexName string `json:"index_name"`
+	IndexUUID string `json:"index_uuid"`
+	Reason    string `json:"reason"`
+}

plugins/ism/api_add.go

@@ -0,0 +1,81 @@
+// SPDX-License-Identifier: Apache-2.0
+//
+// The OpenSearch Contributors require contributions made to
+// this file be licensed under the Apache-2.0 license or a
+// compatible open source license.
+
+package ism
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"net/http"
+	"strings"
+
+	"github.com/opensearch-project/opensearch-go/v3"
+)
+
+// Add executes a add policy request with the required AddReq
+func (c Client) Add(ctx context.Context, req AddReq) (AddResp, error) {
+	var (
+		data AddResp
+		err  error
+	)
+	if data.response, err = c.do(ctx, req, &data); err != nil {
+		return data, err
+	}
+
+	return data, nil
+}
+
+// AddReq represents possible options for the add policy request
+type AddReq struct {
+	Indices []string
+	Body    AddBody
+
+	Header http.Header
+}
+
+// GetRequest returns the *http.Request that gets executed by the client
+func (r AddReq) GetRequest() (*http.Request, error) {
+	body, err := json.Marshal(r.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	indices := strings.Join(r.Indices, ",")
+	var path strings.Builder
+	path.Grow(len("/_plugins/_ism/add/") + len(indices))
+	path.WriteString("/_plugins/_ism/add")
+	if len(r.Indices) > 0 {
+		path.WriteString("/")
+		path.WriteString(indices)
+	}
+
+	return opensearch.BuildRequest(
+		http.MethodPost,
+		path.String(),
+		bytes.NewReader(body),
+		make(map[string]string),
+		r.Header,
+	)
+}
+
+// AddResp represents the returned struct of the add policy response
+type AddResp struct {
+	UpdatedIndices int           `json:"updated_indices"`
+	Failures       bool          `json:"failures"`
+	FailedIndices  []FailedIndex `json:"failed_indices"`
+	response       *opensearch.Response
+}
+
+// Inspect returns the Inspect type containing the raw *opensearch.Reponse
+func (r AddResp) Inspect() Inspect {
+	return Inspect{Response: r.response}
+}
+
+// AddBody represents the request body for the add policy request
+type AddBody struct {
+	PolicyID string `json:"policy_id"`
+}

plugins/ism/api_change.go

@@ -0,0 +1,88 @@
+// SPDX-License-Identifier: Apache-2.0
+//
+// The OpenSearch Contributors require contributions made to
+// this file be licensed under the Apache-2.0 license or a
+// compatible open source license.
+
+package ism
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"net/http"
+	"strings"
+
+	"github.com/opensearch-project/opensearch-go/v3"
+)
+
+// Change executes a change policy request with the required ChangeReq
+func (c Client) Change(ctx context.Context, req ChangeReq) (ChangeResp, error) {
+	var (
+		data ChangeResp
+		err  error
+	)
+	if data.response, err = c.do(ctx, req, &data); err != nil {
+		return data, err
+	}
+
+	return data, nil
+}
+
+// ChangeReq represents possible options for the change policy request
+type ChangeReq struct {
+	Indices []string
+	Body    ChangeBody
+
+	Header http.Header
+}
+
+// GetRequest returns the *http.Request that gets executed by the client
+func (r ChangeReq) GetRequest() (*http.Request, error) {
+	body, err := json.Marshal(r.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	indices := strings.Join(r.Indices, ",")
+	var path strings.Builder
+	path.Grow(len("/_plugins/_ism/change_policy/") + len(indices))
+	path.WriteString("/_plugins/_ism/change_policy")
+	if len(r.Indices) > 0 {
+		path.WriteString("/")
+		path.WriteString(indices)
+	}
+
+	return opensearch.BuildRequest(
+		http.MethodPost,
+		path.String(),
+		bytes.NewReader(body),
+		make(map[string]string),
+		r.Header,
+	)
+}
+
+// ChangeResp represents the returned struct of the change policy response
+type ChangeResp struct {
+	UpdatedIndices int           `json:"updated_indices"`
+	Failures       bool          `json:"failures"`
+	FailedIndices  []FailedIndex `json:"failed_indices"`
+	response       *opensearch.Response
+}
+
+// Inspect returns the Inspect type containing the raw *opensearch.Reponse
+func (r ChangeResp) Inspect() Inspect {
+	return Inspect{Response: r.response}
+}
+
+// ChangeBody represents the request body for the change policy request
+type ChangeBody struct {
+	PolicyID string              `json:"policy_id"`
+	State    string              `json:"state"`
+	Include  []ChangeBodyInclude `json:"include,omitempty"`
+}
+
+// ChangeBodyInclude is a sub type of ChangeBody containing the state information
+type ChangeBodyInclude struct {
+	State string `json:"state"`
+}

plugins/ism/api_explain-params.go

@@ -0,0 +1,27 @@
+// SPDX-License-Identifier: Apache-2.0
+//
+// The OpenSearch Contributors require contributions made to
+// this file be licensed under the Apache-2.0 license or a
+// compatible open source license.
+
+package ism
+
+// ExplainParams represents possible parameters for the ExplainReq
+type ExplainParams struct {
+	ShowPolicy     bool
+	ValidateAction bool
+}
+
+func (r ExplainParams) get() map[string]string {
+	params := make(map[string]string)
+
+	if r.ShowPolicy {
+		params["show_policy"] = "true"
+	}
+
+	if r.ValidateAction {
+		params["validate_action"] = "true"
+	}
+
+	return params
+}