Merge pull request #78 from appbaseio/feat/index-alias-map

feat: index alias mapping

Author: bietkul

middleware/classify/indices.go

@@ -2,6 +2,8 @@ package classify
 
 import (
 	"net/http"
+	"regexp"
+	"strings"
 
 	"github.com/appbaseio/arc/middleware"
 	"github.com/appbaseio/arc/model/index"
@@ -16,7 +18,50 @@ func Indices() middleware.Middleware {
 func indices(h http.HandlerFunc) http.HandlerFunc {
 	return func(w http.ResponseWriter, req *http.Request) {
 		indices := util.IndicesFromRequest(req)
+		currentCache := GetIndexAliasCache()
 
+		for _, index := range indices {
+			// '*' in case of all indices put alias in context
+			if index == "*" {
+				for cachedItem := range currentCache {
+					alias := GetIndexAlias(cachedItem)
+					if alias != "" {
+						indices = append(indices, alias)
+					}
+				}
+				break
+			} else if strings.Contains(index, "*") {
+				// in case of regex check if string contains '*' in naming pattern, if contains and doesn't have '.*' and replace '*' with '.*' because golang regex can match in that pattern. Next match regex patters with existing index names in cache and add those alias to context.
+				regex := index
+
+				if !strings.Contains(index, ".*") {
+					regex = strings.Replace(regex, "*", ".*", -1)
+				}
+				r, _ := regexp.Compile(regex)
+				cachedIndices := []string{}
+
+				for cachedItem := range currentCache {
+					cachedIndices = append(cachedIndices, cachedItem)
+				}
+				for _, val := range cachedIndices {
+					if r.MatchString(val) {
+						alias := GetIndexAlias(val)
+						if alias != "" {
+							indices = append(indices, alias)
+						}
+						break
+					}
+				}
+
+			} else {
+				// get alias for index and put in context
+				alias := GetIndexAlias(index)
+				if alias != "" {
+					indices = append(indices, alias)
+				}
+				break
+			}
+		}
 		ctx := index.NewContext(req.Context(), indices)
 		req = req.WithContext(ctx)
 

middleware/classify/util.go

@@ -0,0 +1,24 @@
+package classify
+
+// IndexAliasCache cache to store index alias map
+var IndexAliasCache = make(map[string]string)
+
+// GetIndexAliasCache get whole cache
+func GetIndexAliasCache() map[string]string {
+	return IndexAliasCache
+}
+
+// GetIndexAlias get alias for specific index
+func GetIndexAlias(index string) string {
+	alias, ok := IndexAliasCache[index]
+
+	if !ok {
+		return ""
+	}
+	return alias
+}
+
+// SetIndexAlias set alias for specific index
+func SetIndexAlias(index, alias string) {
+	IndexAliasCache[index] = alias
+}

plugins/reindexer/dao.go

@@ -4,9 +4,11 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"regexp"
 
 	log "github.com/sirupsen/logrus"
 
+	"github.com/appbaseio/arc/middleware/classify"
 	"github.com/appbaseio/arc/util"
 	es7 "github.com/olivere/elastic/v7"
 )
@@ -114,11 +116,18 @@ func reindex(ctx context.Context, sourceIndex string, config *reindexConfig, wai
 
 		if destinationIndex == "" {
 			// Fetch all the aliases of old index
-			aliases, err := aliasesOf(ctx, sourceIndex)
+			alias, err := aliasesOf(ctx, sourceIndex)
+
+			var aliases = []string{}
 			if err != nil {
 				return nil, fmt.Errorf(`error fetching aliases of index "%s": %v`, sourceIndex, err)
 			}
-			aliases = append(aliases, sourceIndex)
+
+			if alias == "" {
+				aliases = append(aliases, sourceIndex)
+			} else {
+				aliases = append(aliases, alias)
+			}
 
 			// Delete old index
 			err = deleteIndex(ctx, sourceIndex)
@@ -210,22 +219,28 @@ func settingsOf(ctx context.Context, indexName string) (map[string]interface{},
 	return settings, nil
 }
 
-func aliasesOf(ctx context.Context, indexName string) ([]string, error) {
+func aliasesOf(ctx context.Context, indexName string) (string, error) {
 	response, err := util.GetClient7().CatAliases().
 		Pretty(true).
 		Do(ctx)
 	if err != nil {
-		return nil, err
+		return "", err
 	}
 
-	var aliases []string
+	var alias = ""
+
+	// set alias for original index name only.
+	regex := ".*reindexed_[0-9]+"
+	r, _ := regexp.Compile(regex)
+
 	for _, row := range response {
-		if row.Index == indexName {
-			aliases = append(aliases, row.Alias)
+		// r.MatchString(indexName) this condition is added to handle existing alias which are created incorrectly
+		if row.Index == indexName && r.MatchString(indexName) {
+			alias = row.Alias
 		}
 	}
 
-	return aliases, nil
+	return alias, nil
 }
 
 func createIndex(ctx context.Context, indexName string, body map[string]interface{}) error {
@@ -276,6 +291,8 @@ func setAlias(ctx context.Context, indexName string, aliases ...string) error {
 		return fmt.Errorf(`unable to set aliases "%v" for index "%s"`, aliases, indexName)
 	}
 
+	// We only have one alias per index.
+	classify.SetIndexAlias(indexName, aliases[0])
 	return nil
 }
 
@@ -288,3 +305,52 @@ func getIndicesByAlias(ctx context.Context, alias string) ([]string, error) {
 	}
 	return response.IndicesByAlias(alias), nil
 }
+
+func getAliasedIndices(ctx context.Context) ([]AliasedIndices, error) {
+	var indicesList []AliasedIndices
+	indices, err := util.GetClient7().CatIndices().
+		Do(ctx)
+	if err != nil {
+		return indicesList, err
+	}
+
+	aliases, err := util.GetClient7().CatAliases().
+		Pretty(true).
+		Do(ctx)
+	if err != nil {
+		return indicesList, err
+	}
+
+	for _, index := range indices {
+		var indexStruct = AliasedIndices{
+			Health:       index.Health,
+			Status:       index.Status,
+			Index:        index.Index,
+			UUID:         index.UUID,
+			Pri:          index.Pri,
+			Rep:          index.Rep,
+			DocsCount:    index.DocsCount,
+			DocsDeleted:  index.DocsDeleted,
+			StoreSize:    index.StoreSize,
+			PriStoreSize: index.PriStoreSize,
+		}
+		var alias string
+		regex := ".*reindexed_[0-9]+"
+		r, _ := regexp.Compile(regex)
+
+		for _, row := range aliases {
+			if row.Index == index.Index && r.MatchString(index.Index) {
+				alias = row.Alias
+				break
+			}
+		}
+		if err == nil && alias != "" {
+			indexStruct.Alias = alias
+		}
+
+		indicesList = append(indicesList, indexStruct)
+
+	}
+
+	return indicesList, nil
+}

plugins/reindexer/handlers.go

@@ -60,6 +60,19 @@ func (rx *reindexer) reindexSrcToDest() http.HandlerFunc {
 	}
 }
 
+func (rx *reindexer) aliasedIndices() http.HandlerFunc {
+	return func(w http.ResponseWriter, req *http.Request) {
+		res, err := getAliasedIndices(req.Context())
+		if err != nil {
+			util.WriteBackError(w, "Unable to get aliased indices.\n"+err.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		response, err := json.Marshal(res)
+		errorHandler(nil, w, response)
+	}
+}
+
 func errorHandler(err error, w http.ResponseWriter, response []byte) {
 	if err != nil {
 		log.Errorln(logTag, ":", err)

plugins/reindexer/reindexer.go

@@ -32,6 +32,7 @@ func (rx *reindexer) Name() string {
 }
 
 func (rx *reindexer) InitFunc() error {
+	InitIndexAliasCache()
 	return nil
 }
 

plugins/reindexer/routes.go

@@ -23,6 +23,13 @@ func (rx *reindexer) routes() []plugins.Route {
 			HandlerFunc: middleware(rx.reindex()),
 			Description: "Reindexes a single index with the given mappings, settings and types.",
 		},
+		{
+			Name:        "AliasedIndexes",
+			Methods:     []string{http.MethodGet},
+			Path:        "/_aliasedindices",
+			HandlerFunc: middleware(rx.aliasedIndices()),
+			Description: "Get map of indices and aliases",
+		},
 	}
 	return routes
 }

plugins/reindexer/util.go

@@ -1,14 +1,31 @@
 package reindexer
 
 import (
+	"context"
 	"fmt"
 	"regexp"
 	"strconv"
 	"strings"
 
+	"github.com/appbaseio/arc/middleware/classify"
 	log "github.com/sirupsen/logrus"
 )
 
+// AliasedIndices struct
+type AliasedIndices struct {
+	Alias        string `json:"alias"`
+	Health       string `json:"health"`
+	Status       string `json:"status"`
+	Index        string `json:"index"`
+	UUID         string `json:"uuid"`
+	Pri          int    `json:"pri"`
+	Rep          int    `json:"rep"`
+	DocsCount    int    `json:"docs.count"`
+	DocsDeleted  int    `json:"docs.deleted"`
+	StoreSize    string `json:"store.size"`
+	PriStoreSize string `json:"pri.store.size"`
+}
+
 // reindexedName calculates from the name the number of times an index has been
 // reindexed to generate the successive name for the index. For example: for an
 // index named "twitter", the funtion returns "twitter_reindexed_1", and for an
@@ -38,3 +55,16 @@ func reindexedName(indexName string) (string, error) {
 
 	return indexName, nil
 }
+
+// InitIndexAliasCache to set cache on arc initialization
+func InitIndexAliasCache() {
+	ctx := context.Background()
+	aliasedIndexes, _ := getAliasedIndices(ctx)
+
+	for _, aliasIndex := range aliasedIndexes {
+		if aliasIndex.Alias != "" {
+			classify.SetIndexAlias(aliasIndex.Index, aliasIndex.Alias)
+		}
+	}
+	log.Println(logTag, "=> Alias Index Cache", classify.GetIndexAliasCache())
+}