supports reindex

Author: po3rin

README.md

@@ -16,6 +16,8 @@ eskeeper synchronizes index and alias with configuration files while ensuring id
 
 * index
 - [x] create
+- [x] status (open or close)
+- [x] reindex (only basic parameter)
 - [ ] update mapping
 - [ ] delete
 
@@ -38,8 +40,8 @@ es.yaml is indices & aliases config file.
 
 ```yaml
 index:
-  - name: test-v1
-    mapping: testdata/test.json
+  - name: test-v1 # index name
+    mapping: testdata/test.json # index setting & mapping (json)
 
   - name: test-v2
     mapping: testdata/test.json
@@ -48,6 +50,20 @@ index:
     mapping: testdata/test.json
     status: close
 
+  # reindex test-v1 -> reindex-v1	
+  - name: reindex-v1
+    mapping: testdata/test.json
+    reindex:
+        source: test-v1 
+        slices: 3 # default=1
+        waitForCompletion: true
+
+        # 'on' field supports 2 hooks.
+        # 'reindex': only when index is created for the first time.
+        # 'always': always exec reindex.
+        on: firstCreated
+
+
 alias:
   - name: alias1
     index:
@@ -67,14 +83,16 @@ curl localhost:9200/_cat/indices
 yellow open test-v1 ... 1 1 0 0 208b 208b
 yellow open test-v2 ... 1 1 0 0 208b 208b
 yellow close close-v1 xxxxxxxxxxxx 1 1
+yellow open reindex-v1 ... 1 1 0 0 208b 208b
 
 curl localhost:9200/_cat/aliases
 alias2 test-v2 - - - -
 alias1 test-v1 - - - -
 alias2 test-v1 - - - -
 ```
 
-## :triangular_ruler: Settings
+
+## :triangular_ruler: Usage
 
 eskeeper supports flag & environment value.
 

beta.go

@@ -125,8 +125,6 @@ func (c *esclient) equalMappingsProperties(ctx context.Context, index index, map
 		return false, errors.New("get index response dose not contain index name field")
 	}
 
-	fmt.Println(v)
-
 	b, err := json.Marshal(v.Mappings)
 	if err != nil {
 		return false, fmt.Errorf("marshal properties json: %w", err)

conf.go

@@ -16,15 +16,29 @@ var status = map[string]struct{}{
 	"":      struct{}{}, // default
 }
 
+var reindexOn = map[string]struct{}{
+	"always":       struct{}{},
+	"firstCreated": struct{}{},
+	"":             struct{}{}, // default
+}
+
 type config struct {
 	Indices []index `json:"index"`
 	Aliases []alias `json:"alias"` // supports close only
 }
 
 type index struct {
-	Name    string `json:"name"`
-	Mapping string `json:"mapping"`
-	Status  string `json:"status"`
+	Name    string  `json:"name"`
+	Mapping string  `json:"mapping"`
+	Status  string  `json:"status"`
+	Reindex reindex `json:"reindex"`
+}
+
+type reindex struct {
+	Source            string `json:"source"`
+	Slices            int    `json:"slices"`
+	WaitForCompletion bool   `json:"waitForCompletion"`
+	On                string `json:"on"`
 }
 
 type alias struct {
@@ -67,6 +81,16 @@ func validateIndex(index index) error {
 		return fmt.Errorf("unsupported status %v", index.Status)
 	}
 
+	if index.Reindex.Source != "" {
+		if index.Status == "close" {
+			return errors.New("unsupported close status and reindex cannot be used together")
+		}
+		_, ok := reindexOn[index.Reindex.On]
+		if !ok {
+			return fmt.Errorf("unsupported reindex hook %v. [always or firstCreated]", index.Reindex.On)
+		}
+	}
+
 	return nil
 }
 

conf_test.go

@@ -44,6 +44,34 @@ func TestYaml2Conf(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "reindex",
+			yaml: "testdata/es.reindex.yaml",
+			want: config{
+				Indices: []index{
+					{
+						Name:    "test-v1",
+						Mapping: "testdata/test.json",
+					},
+					{
+						Name:    "test-v2",
+						Mapping: "testdata/test.json",
+						Reindex: reindex{
+							Source:            "test-v1",
+							Slices:            20,
+							WaitForCompletion: true,
+							On:                "firstCreated",
+						},
+					},
+				},
+				Aliases: []alias{
+					{
+						Name:    "alias1",
+						Indices: []string{"test-v2"},
+					},
+				},
+			},
+		},
 	}
 
 	for _, tt := range tests {

docker-compose.yaml

@@ -1,7 +1,7 @@
 version: '3'
 services:
   elasticsearch:
-    image: docker.elastic.co/elasticsearch/elasticsearch:7.9.2
+    image: docker.elastic.co/elasticsearch/elasticsearch:7.11.1
     ports:
       - "9200:9200"
       - "9300:9300"

go.mod

@@ -1,36 +1,43 @@
 module github.com/po3rin/eskeeper
 
-go 1.15
+go 1.16
 
 require (
 	github.com/Cside/jsondiff v0.0.0-20180209072652-0e50d980b458
+	github.com/Masterminds/semver/v3 v3.1.1 // indirect
 	github.com/cenkalti/backoff v2.2.1+incompatible // indirect
 	github.com/cenkalti/backoff/v3 v3.2.2 // indirect
 	github.com/containerd/continuity v0.0.0-20200928162600-f2cc35102c2a // indirect
 	github.com/creack/pty v1.1.9 // indirect
 	github.com/elastic/go-elasticsearch v0.0.0
-	github.com/elastic/go-elasticsearch/v7 v7.9.0
+	github.com/elastic/go-elasticsearch/v7 v7.11.0
 	github.com/evanphx/json-patch v4.9.0+incompatible // indirect
-	github.com/fatih/color v1.9.0 // indirect
-	github.com/goccy/go-yaml v1.8.2
-	github.com/gofrs/uuid v3.3.0+incompatible
-	github.com/google/go-cmp v0.4.0 // indirect
-	github.com/itchyny/gojq v0.11.1
-	github.com/itchyny/timefmt-go v0.1.1 // indirect
+	github.com/fatih/color v1.10.0 // indirect
+	github.com/goccy/go-yaml v1.8.9
+	github.com/gofrs/uuid v4.0.0+incompatible
+	github.com/hokaccha/go-prettyjson v0.0.0-20190818114111-108c894c2c0e // indirect
+	github.com/itchyny/astgen-go v0.0.0-20200815150004-12a293722290 // indirect
+	github.com/itchyny/gojq v0.12.2
 	github.com/kataras/pio v0.0.10
 	github.com/mattn/go-colorable v0.1.8 // indirect
+	github.com/mgutz/ansi v0.0.0-20200706080929-d51e80ef957d // indirect
+	github.com/oligot/go-mod-upgrade v0.4.1 // indirect
 	github.com/ory/dockertest v3.3.5+incompatible
 	github.com/ory/dockertest/v3 v3.6.0 // indirect
 	github.com/pkg/errors v0.9.1
-	github.com/po3rin/bmfzf v0.0.1
+	github.com/po3rin/bmfzf v0.0.2
 	github.com/sergi/go-diff v1.1.0 // indirect
 	github.com/sirupsen/logrus v1.7.0 // indirect
-	github.com/spf13/cobra v1.0.0
-	github.com/spf13/pflag v1.0.3
-	github.com/spf13/viper v1.7.0
-	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9
-	golang.org/x/sys v0.0.0-20200826173525-f9321e4c35a6 // indirect
-	golang.org/x/text v0.3.3 // indirect
+	github.com/spf13/cobra v1.1.3
+	github.com/spf13/pflag v1.0.5
+	github.com/spf13/viper v1.7.1
+	github.com/stretchr/testify v1.6.1 // indirect
+	golang.org/x/crypto v0.0.0-20210220033148-5ea612d1eb83
+	golang.org/x/sys v0.0.0-20210313110737-8e9fff1a3a18 // indirect
+	golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d // indirect
+	golang.org/x/text v0.3.5 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
-	gopkg.in/yaml.v2 v2.2.7
+	gopkg.in/go-playground/assert.v1 v1.2.1 // indirect
+	gopkg.in/go-playground/validator.v9 v9.30.0 // indirect
+	gopkg.in/yaml.v2 v2.4.0
 )

go.sum

@@ -52,11 +52,46 @@ func (c *esclient) syncIndex(ctx context.Context, index index) error {
 			}
 			return fmt.Errorf("failed to create index [index=%v, statusCode=%v, res=%v]", index.Name, res.StatusCode, string(body))
 		}
+
+		// reindex --------
+
+		if index.Reindex.Source == "" {
+			return nil
+		}
+
+		ok, err := c.existIndex(ctx, index.Reindex.Source)
+		if err != nil {
+			return fmt.Errorf("check index exists for reindex process: %w", err)
+		}
+		if !ok {
+			return fmt.Errorf("reindex (%s -> %s) conf is invalid. Make sure %s index exists", index.Reindex.Source, index.Reindex.Source, index.Name)
+		}
+
+		err = c.reindex(ctx, index.Name, index.Reindex)
+		if err != nil {
+			return fmt.Errorf("reindex (%s -> %s)", index.Reindex.Source, index.Name)
+		}
 		return nil
 	}
 
 	// index already exists.
 
+	// reindex -------
+	if index.Reindex.Source != "" && index.Reindex.On == "always" {
+		ok, err = c.existIndex(ctx, index.Reindex.Source)
+		if err != nil {
+			return fmt.Errorf("check index exists for reindex process: %w", err)
+		}
+		if !ok {
+			return fmt.Errorf("reindex (%s -> %s) conf is invalid. Make sure %s index exists", index.Reindex.Source, index.Reindex.Source, index.Name)
+		}
+		err = c.reindex(ctx, index.Name, index.Reindex)
+		if err != nil {
+			return fmt.Errorf("reindex (%s -> %s)", index.Reindex.Source, index.Name)
+		}
+		return nil
+	}
+
 	// Since downtime may occur when switching aliases, only open is processed before switching aliases.
 	// TODO: refactoring.
 	if index.Status == "close" {
@@ -143,9 +178,9 @@ func (c *esclient) openIndex(ctx context.Context, index index) error {
 	if res.StatusCode != 200 {
 		body, err := ioutil.ReadAll(res.Body)
 		if err != nil {
-			return fmt.Errorf("failed to open index [index= %v, statusCode=%v]", index, res.StatusCode)
+			return fmt.Errorf("failed to open index [index= %v, statusCode=%v]: %w", index, res.StatusCode, err)
 		}
-		return fmt.Errorf("failed to open index [index= %v, statusCode=%v, res=%v]", index, res.StatusCode, string(body))
+		return fmt.Errorf("failed to open index [index= %v, statusCode=%v, res=%v]: %w", index, res.StatusCode, string(body), err)
 	}
 	return nil
 }

index.go

@@ -91,7 +91,7 @@ func TestSyncIndices(t *testing.T) {
 			},
 			setup: func(tb testing.TB) {
 				createTmpIndexHelper(tb, "open-v1")
-				closeIndex(tb, "open-v1")
+				closeIndexHelper(tb, "open-v1")
 			},
 		},
 		{

index_test.go

@@ -22,7 +22,7 @@ func TestMain(m *testing.M) {
 
 	resource, err := pool.Run(
 		"docker.elastic.co/elasticsearch/elasticsearch",
-		"7.9.2",
+		"7.11.1",
 		[]string{
 			"ES_JAVA_OPTS=-Xms512m -Xmx512m",
 			"discovery.type=single-node",
@@ -132,7 +132,7 @@ func createTmpAliasHelper(tb testing.TB, name string, index string) {
 	}
 }
 
-func closeIndex(tb testing.TB, index string) {
+func closeIndexHelper(tb testing.TB, index string) {
 	conf := elasticsearch.Config{
 		Addresses: []string{url},
 	}
@@ -153,3 +153,33 @@ func closeIndex(tb testing.TB, index string) {
 		tb.Fatalf("failed to close index [index= %v, statusCode=%v, res=%v]", index, res.StatusCode, string(body))
 	}
 }
+
+func postDocHelper(tb testing.TB, index string) {
+	tb.Helper()
+	conf := elasticsearch.Config{
+		Addresses: []string{url},
+	}
+	es, err := elasticsearch.NewClient(conf)
+	if err != nil {
+		tb.Fatal(err)
+	}
+
+	body := strings.NewReader(`{"title":"this is title","body":"this is body"}`)
+
+	res, err := es.Index(
+		index,
+		body,
+		es.Index.WithRefresh("true"),
+	)
+	if err != nil {
+		tb.Fatal(err)
+	}
+
+	if res.StatusCode != 201 {
+		body, err := ioutil.ReadAll(res.Body)
+		if err != nil {
+			tb.Fatal(err)
+		}
+		tb.Fatalf("failed to post document [index=%v, statusCode=%v, res=%v]", index, res.StatusCode, string(body))
+	}
+}

main_test.go

@@ -0,0 +1,48 @@
+package eskeeper
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"strings"
+)
+
+func (c *esclient) reindex(ctx context.Context, dest string, reindex reindex) error {
+	ri := c.client.Reindex
+	body := strings.NewReader(
+		fmt.Sprintf(`
+{
+  "source": {
+    "index": "%s"
+  },
+  "dest": {
+    "index": "%s"
+  }
+}`,
+			reindex.Source, dest,
+		),
+	)
+
+	slices := reindex.Slices
+	if slices == 0 {
+		slices = 1
+	}
+
+	res, err := ri(
+		body,
+		ri.WithContext(ctx),
+		ri.WithSlices(slices),
+		ri.WithWaitForCompletion(reindex.WaitForCompletion),
+	)
+	if err != nil {
+		return fmt.Errorf("reindex: %w", err)
+	}
+	if res.StatusCode != 200 {
+		body, err := ioutil.ReadAll(res.Body)
+		if err != nil {
+			return err
+		}
+		return fmt.Errorf("failed to reindex [index=%v, statusCode=%v, res=%v]", reindex.Source, res.StatusCode, string(body))
+	}
+	return nil
+}

reindex.go

@@ -0,0 +1,55 @@
+package eskeeper
+
+import (
+	"context"
+	"testing"
+)
+
+func TestReindex(t *testing.T) {
+	tests := []struct {
+		name    string
+		dest    string
+		reindex reindex
+		setup   func(tb testing.TB)
+		cleanup func(tb testing.TB)
+	}{
+		{
+			name: "reindex",
+			dest: "reindex-dest",
+			reindex: reindex{
+				Source:            "reindex-src",
+				Slices:            3,
+				WaitForCompletion: true,
+				On:                "firstCreated",
+			},
+			setup: func(tb testing.TB) {
+				createTmpIndexHelper(tb, "reindex-src")
+				createTmpIndexHelper(tb, "reindex-dest")
+				postDocHelper(tb, "reindex-src")
+			},
+			cleanup: func(tb testing.TB) {
+			},
+		},
+	}
+
+	es, err := newEsClient([]string{url}, "", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
+			if tt.setup != nil {
+				tt.setup(t)
+			}
+			err := es.reindex(ctx, tt.dest, tt.reindex)
+			if err != nil {
+				t.Error(err)
+			}
+			if tt.cleanup != nil {
+				tt.cleanup(t)
+			}
+		})
+	}
+}

reindex_test.go

@@ -0,0 +1,22 @@
+index:
+  - name: test-v1
+    mapping: testdata/test.json
+
+  # reindex test-v1 -> reindex-v1	
+  - name: reindex-v1
+    mapping: testdata/test.json
+    reindex:
+        source: test-v1 
+        slices: 3 # default=1
+        waitForCompletion: true
+        # 'on' field supports 2 hooks.
+        # 'reindex': only when index is created for the first time.
+        # 'always': always exec reindex.
+        on: firstCreated
+
+
+alias:
+  - name: alias1
+    index:
+      - test-v2
+

testdata/es.reindex.yaml

@@ -1,6 +1,6 @@
 index:
-  - name: test-v1
-    mapping: testdata/test.json
+  - name: test-v1 # index name
+    mapping: testdata/test.json # index setting & mapping (json)
 
   - name: test-v2
     mapping: testdata/test.json
@@ -14,6 +14,7 @@ alias:
     index:
       - test-v1
 
+  # multi indicies
   - name: alias2
     index:
       - test-v1