diff options
Diffstat (limited to 'vendor/github.com/olivere/elastic/reindex.go')
| -rw-r--r-- | vendor/github.com/olivere/elastic/reindex.go | 695 |
1 files changed, 0 insertions, 695 deletions
diff --git a/vendor/github.com/olivere/elastic/reindex.go b/vendor/github.com/olivere/elastic/reindex.go deleted file mode 100644 index 9cdd50a68..000000000 --- a/vendor/github.com/olivere/elastic/reindex.go +++ /dev/null @@ -1,695 +0,0 @@ -// Copyright 2012-present Oliver Eilhard. All rights reserved. -// Use of this source code is governed by a MIT-license. -// See http://olivere.mit-license.org/license.txt for details. - -package elastic - -import ( - "context" - "fmt" - "net/url" -) - -// ReindexService is a method to copy documents from one index to another. -// It is documented at https://www.elastic.co/guide/en/elasticsearch/reference/6.0/docs-reindex.html. -type ReindexService struct { - client *Client - pretty bool - refresh string - timeout string - waitForActiveShards string - waitForCompletion *bool - requestsPerSecond *int - slices *int - body interface{} - source *ReindexSource - destination *ReindexDestination - conflicts string - size *int - script *Script -} - -// NewReindexService creates a new ReindexService. -func NewReindexService(client *Client) *ReindexService { - return &ReindexService{ - client: client, - } -} - -// WaitForActiveShards sets the number of shard copies that must be active before -// proceeding with the reindex operation. Defaults to 1, meaning the primary shard only. -// Set to `all` for all shard copies, otherwise set to any non-negative value less than or -// equal to the total number of copies for the shard (number of replicas + 1). -func (s *ReindexService) WaitForActiveShards(waitForActiveShards string) *ReindexService { - s.waitForActiveShards = waitForActiveShards - return s -} - -// RequestsPerSecond specifies the throttle to set on this request in sub-requests per second. -// -1 means set no throttle as does "unlimited" which is the only non-float this accepts. -func (s *ReindexService) RequestsPerSecond(requestsPerSecond int) *ReindexService { - s.requestsPerSecond = &requestsPerSecond - return s -} - -// Slices specifies the number of slices this task should be divided into. Defaults to 1. -func (s *ReindexService) Slices(slices int) *ReindexService { - s.slices = &slices - return s -} - -// Refresh indicates whether Elasticsearch should refresh the effected indexes -// immediately. -func (s *ReindexService) Refresh(refresh string) *ReindexService { - s.refresh = refresh - return s -} - -// Timeout is the time each individual bulk request should wait for shards -// that are unavailable. -func (s *ReindexService) Timeout(timeout string) *ReindexService { - s.timeout = timeout - return s -} - -// WaitForCompletion indicates whether Elasticsearch should block until the -// reindex is complete. -func (s *ReindexService) WaitForCompletion(waitForCompletion bool) *ReindexService { - s.waitForCompletion = &waitForCompletion - return s -} - -// Pretty indicates that the JSON response be indented and human readable. -func (s *ReindexService) Pretty(pretty bool) *ReindexService { - s.pretty = pretty - return s -} - -// Source specifies the source of the reindexing process. -func (s *ReindexService) Source(source *ReindexSource) *ReindexService { - s.source = source - return s -} - -// SourceIndex specifies the source index of the reindexing process. -func (s *ReindexService) SourceIndex(index string) *ReindexService { - if s.source == nil { - s.source = NewReindexSource() - } - s.source = s.source.Index(index) - return s -} - -// Destination specifies the destination of the reindexing process. -func (s *ReindexService) Destination(destination *ReindexDestination) *ReindexService { - s.destination = destination - return s -} - -// DestinationIndex specifies the destination index of the reindexing process. -func (s *ReindexService) DestinationIndex(index string) *ReindexService { - if s.destination == nil { - s.destination = NewReindexDestination() - } - s.destination = s.destination.Index(index) - return s -} - -// DestinationIndexAndType specifies both the destination index and type -// of the reindexing process. -func (s *ReindexService) DestinationIndexAndType(index, typ string) *ReindexService { - if s.destination == nil { - s.destination = NewReindexDestination() - } - s.destination = s.destination.Index(index) - s.destination = s.destination.Type(typ) - return s -} - -// Conflicts indicates what to do when the process detects version conflicts. -// Possible values are "proceed" and "abort". -func (s *ReindexService) Conflicts(conflicts string) *ReindexService { - s.conflicts = conflicts - return s -} - -// AbortOnVersionConflict aborts the request on version conflicts. -// It is an alias to setting Conflicts("abort"). -func (s *ReindexService) AbortOnVersionConflict() *ReindexService { - s.conflicts = "abort" - return s -} - -// ProceedOnVersionConflict aborts the request on version conflicts. -// It is an alias to setting Conflicts("proceed"). -func (s *ReindexService) ProceedOnVersionConflict() *ReindexService { - s.conflicts = "proceed" - return s -} - -// Size sets an upper limit for the number of processed documents. -func (s *ReindexService) Size(size int) *ReindexService { - s.size = &size - return s -} - -// Script allows for modification of the documents as they are reindexed -// from source to destination. -func (s *ReindexService) Script(script *Script) *ReindexService { - s.script = script - return s -} - -// Body specifies the body of the request to send to Elasticsearch. -// It overrides settings specified with other setters, e.g. Query. -func (s *ReindexService) Body(body interface{}) *ReindexService { - s.body = body - return s -} - -// buildURL builds the URL for the operation. -func (s *ReindexService) buildURL() (string, url.Values, error) { - // Build URL path - path := "/_reindex" - - // Add query string parameters - params := url.Values{} - if s.pretty { - params.Set("pretty", "true") - } - if s.refresh != "" { - params.Set("refresh", s.refresh) - } - if s.timeout != "" { - params.Set("timeout", s.timeout) - } - if s.requestsPerSecond != nil { - params.Set("requests_per_second", fmt.Sprintf("%v", *s.requestsPerSecond)) - } - if s.slices != nil { - params.Set("slices", fmt.Sprintf("%v", *s.slices)) - } - if s.waitForActiveShards != "" { - params.Set("wait_for_active_shards", s.waitForActiveShards) - } - if s.waitForCompletion != nil { - params.Set("wait_for_completion", fmt.Sprintf("%v", *s.waitForCompletion)) - } - return path, params, nil -} - -// Validate checks if the operation is valid. -func (s *ReindexService) Validate() error { - var invalid []string - if s.body != nil { - return nil - } - if s.source == nil { - invalid = append(invalid, "Source") - } else { - if len(s.source.indices) == 0 { - invalid = append(invalid, "Source.Index") - } - } - if s.destination == nil { - invalid = append(invalid, "Destination") - } - if len(invalid) > 0 { - return fmt.Errorf("missing required fields: %v", invalid) - } - return nil -} - -// getBody returns the body part of the document request. -func (s *ReindexService) getBody() (interface{}, error) { - if s.body != nil { - return s.body, nil - } - - body := make(map[string]interface{}) - - if s.conflicts != "" { - body["conflicts"] = s.conflicts - } - if s.size != nil { - body["size"] = *s.size - } - if s.script != nil { - out, err := s.script.Source() - if err != nil { - return nil, err - } - body["script"] = out - } - - src, err := s.source.Source() - if err != nil { - return nil, err - } - body["source"] = src - - dst, err := s.destination.Source() - if err != nil { - return nil, err - } - body["dest"] = dst - - return body, nil -} - -// Do executes the operation. -func (s *ReindexService) Do(ctx context.Context) (*BulkIndexByScrollResponse, error) { - // Check pre-conditions - if err := s.Validate(); err != nil { - return nil, err - } - - // Get URL for request - path, params, err := s.buildURL() - if err != nil { - return nil, err - } - - // Setup HTTP request body - body, err := s.getBody() - if err != nil { - return nil, err - } - - // Get HTTP response - res, err := s.client.PerformRequest(ctx, PerformRequestOptions{ - Method: "POST", - Path: path, - Params: params, - Body: body, - }) - if err != nil { - return nil, err - } - - // Return operation response - ret := new(BulkIndexByScrollResponse) - if err := s.client.decoder.Decode(res.Body, ret); err != nil { - return nil, err - } - return ret, nil -} - -// DoAsync executes the reindexing operation asynchronously by starting a new task. -// Callers need to use the Task Management API to watch the outcome of the reindexing -// operation. -func (s *ReindexService) DoAsync(ctx context.Context) (*StartTaskResult, error) { - // Check pre-conditions - if err := s.Validate(); err != nil { - return nil, err - } - - // DoAsync only makes sense with WaitForCompletion set to true - if s.waitForCompletion != nil && *s.waitForCompletion { - return nil, fmt.Errorf("cannot start a task with WaitForCompletion set to true") - } - f := false - s.waitForCompletion = &f - - // Get URL for request - path, params, err := s.buildURL() - if err != nil { - return nil, err - } - - // Setup HTTP request body - body, err := s.getBody() - if err != nil { - return nil, err - } - - // Get HTTP response - res, err := s.client.PerformRequest(ctx, PerformRequestOptions{ - Method: "POST", - Path: path, - Params: params, - Body: body, - }) - if err != nil { - return nil, err - } - - // Return operation response - ret := new(StartTaskResult) - if err := s.client.decoder.Decode(res.Body, ret); err != nil { - return nil, err - } - return ret, nil -} - -// -- Source of Reindex -- - -// ReindexSource specifies the source of a Reindex process. -type ReindexSource struct { - searchType string // default in ES is "query_then_fetch" - indices []string - types []string - routing *string - preference *string - requestCache *bool - scroll string - query Query - sorts []SortInfo - sorters []Sorter - searchSource *SearchSource - remoteInfo *ReindexRemoteInfo -} - -// NewReindexSource creates a new ReindexSource. -func NewReindexSource() *ReindexSource { - return &ReindexSource{} -} - -// SearchType is the search operation type. Possible values are -// "query_then_fetch" and "dfs_query_then_fetch". -func (r *ReindexSource) SearchType(searchType string) *ReindexSource { - r.searchType = searchType - return r -} - -func (r *ReindexSource) SearchTypeDfsQueryThenFetch() *ReindexSource { - return r.SearchType("dfs_query_then_fetch") -} - -func (r *ReindexSource) SearchTypeQueryThenFetch() *ReindexSource { - return r.SearchType("query_then_fetch") -} - -func (r *ReindexSource) Index(indices ...string) *ReindexSource { - r.indices = append(r.indices, indices...) - return r -} - -func (r *ReindexSource) Type(types ...string) *ReindexSource { - r.types = append(r.types, types...) - return r -} - -func (r *ReindexSource) Preference(preference string) *ReindexSource { - r.preference = &preference - return r -} - -func (r *ReindexSource) RequestCache(requestCache bool) *ReindexSource { - r.requestCache = &requestCache - return r -} - -func (r *ReindexSource) Scroll(scroll string) *ReindexSource { - r.scroll = scroll - return r -} - -func (r *ReindexSource) Query(query Query) *ReindexSource { - r.query = query - return r -} - -// Sort adds a sort order. -func (s *ReindexSource) Sort(field string, ascending bool) *ReindexSource { - s.sorts = append(s.sorts, SortInfo{Field: field, Ascending: ascending}) - return s -} - -// SortWithInfo adds a sort order. -func (s *ReindexSource) SortWithInfo(info SortInfo) *ReindexSource { - s.sorts = append(s.sorts, info) - return s -} - -// SortBy adds a sort order. -func (s *ReindexSource) SortBy(sorter ...Sorter) *ReindexSource { - s.sorters = append(s.sorters, sorter...) - return s -} - -// RemoteInfo sets up reindexing from a remote cluster. -func (s *ReindexSource) RemoteInfo(ri *ReindexRemoteInfo) *ReindexSource { - s.remoteInfo = ri - return s -} - -// Source returns a serializable JSON request for the request. -func (r *ReindexSource) Source() (interface{}, error) { - source := make(map[string]interface{}) - - if r.query != nil { - src, err := r.query.Source() - if err != nil { - return nil, err - } - source["query"] = src - } else if r.searchSource != nil { - src, err := r.searchSource.Source() - if err != nil { - return nil, err - } - source["source"] = src - } - - if r.searchType != "" { - source["search_type"] = r.searchType - } - - switch len(r.indices) { - case 0: - case 1: - source["index"] = r.indices[0] - default: - source["index"] = r.indices - } - - switch len(r.types) { - case 0: - case 1: - source["type"] = r.types[0] - default: - source["type"] = r.types - } - - if r.preference != nil && *r.preference != "" { - source["preference"] = *r.preference - } - - if r.requestCache != nil { - source["request_cache"] = fmt.Sprintf("%v", *r.requestCache) - } - - if r.scroll != "" { - source["scroll"] = r.scroll - } - - if r.remoteInfo != nil { - src, err := r.remoteInfo.Source() - if err != nil { - return nil, err - } - source["remote"] = src - } - - if len(r.sorters) > 0 { - var sortarr []interface{} - for _, sorter := range r.sorters { - src, err := sorter.Source() - if err != nil { - return nil, err - } - sortarr = append(sortarr, src) - } - source["sort"] = sortarr - } else if len(r.sorts) > 0 { - var sortarr []interface{} - for _, sort := range r.sorts { - src, err := sort.Source() - if err != nil { - return nil, err - } - sortarr = append(sortarr, src) - } - source["sort"] = sortarr - } - - return source, nil -} - -// ReindexRemoteInfo contains information for reindexing from a remote cluster. -type ReindexRemoteInfo struct { - host string - username string - password string - socketTimeout string // e.g. "1m" or "30s" - connectTimeout string // e.g. "1m" or "30s" -} - -// NewReindexRemoteInfo creates a new ReindexRemoteInfo. -func NewReindexRemoteInfo() *ReindexRemoteInfo { - return &ReindexRemoteInfo{} -} - -// Host sets the host information of the remote cluster. -// It must be of the form "http(s)://<hostname>:<port>" -func (ri *ReindexRemoteInfo) Host(host string) *ReindexRemoteInfo { - ri.host = host - return ri -} - -// Username sets the username to authenticate with the remote cluster. -func (ri *ReindexRemoteInfo) Username(username string) *ReindexRemoteInfo { - ri.username = username - return ri -} - -// Password sets the password to authenticate with the remote cluster. -func (ri *ReindexRemoteInfo) Password(password string) *ReindexRemoteInfo { - ri.password = password - return ri -} - -// SocketTimeout sets the socket timeout to connect with the remote cluster. -// Use ES compatible values like e.g. "30s" or "1m". -func (ri *ReindexRemoteInfo) SocketTimeout(timeout string) *ReindexRemoteInfo { - ri.socketTimeout = timeout - return ri -} - -// ConnectTimeout sets the connection timeout to connect with the remote cluster. -// Use ES compatible values like e.g. "30s" or "1m". -func (ri *ReindexRemoteInfo) ConnectTimeout(timeout string) *ReindexRemoteInfo { - ri.connectTimeout = timeout - return ri -} - -// Source returns the serializable JSON data for the request. -func (ri *ReindexRemoteInfo) Source() (interface{}, error) { - res := make(map[string]interface{}) - res["host"] = ri.host - if len(ri.username) > 0 { - res["username"] = ri.username - } - if len(ri.password) > 0 { - res["password"] = ri.password - } - if len(ri.socketTimeout) > 0 { - res["socket_timeout"] = ri.socketTimeout - } - if len(ri.connectTimeout) > 0 { - res["connect_timeout"] = ri.connectTimeout - } - return res, nil -} - -// -source Destination of Reindex -- - -// ReindexDestination is the destination of a Reindex API call. -// It is basically the meta data of a BulkIndexRequest. -// -// See https://www.elastic.co/guide/en/elasticsearch/reference/6.0/docs-reindex.html -// fsourcer details. -type ReindexDestination struct { - index string - typ string - routing string - parent string - opType string - version int64 // default is MATCH_ANY - versionType string // default is "internal" -} - -// NewReindexDestination returns a new ReindexDestination. -func NewReindexDestination() *ReindexDestination { - return &ReindexDestination{} -} - -// Index specifies name of the Elasticsearch index to use as the destination -// of a reindexing process. -func (r *ReindexDestination) Index(index string) *ReindexDestination { - r.index = index - return r -} - -// Type specifies the Elasticsearch type to use for reindexing. -func (r *ReindexDestination) Type(typ string) *ReindexDestination { - r.typ = typ - return r -} - -// Routing specifies a routing value for the reindexing request. -// It can be "keep", "discard", or start with "=". The latter specifies -// the routing on the bulk request. -func (r *ReindexDestination) Routing(routing string) *ReindexDestination { - r.routing = routing - return r -} - -// Keep sets the routing on the bulk request sent for each match to the routing -// of the match (the default). -func (r *ReindexDestination) Keep() *ReindexDestination { - r.routing = "keep" - return r -} - -// Discard sets the routing on the bulk request sent for each match to null. -func (r *ReindexDestination) Discard() *ReindexDestination { - r.routing = "discard" - return r -} - -// Parent specifies the identifier of the parent document (if available). -func (r *ReindexDestination) Parent(parent string) *ReindexDestination { - r.parent = parent - return r -} - -// OpType specifies if this request should follow create-only or upsert -// behavior. This follows the OpType of the standard document index API. -// See https://www.elastic.co/guide/en/elasticsearch/reference/6.0/docs-index_.html#operation-type -// for details. -func (r *ReindexDestination) OpType(opType string) *ReindexDestination { - r.opType = opType - return r -} - -// Version indicates the version of the document as part of an optimistic -// concurrency model. -func (r *ReindexDestination) Version(version int64) *ReindexDestination { - r.version = version - return r -} - -// VersionType specifies how versions are created. -func (r *ReindexDestination) VersionType(versionType string) *ReindexDestination { - r.versionType = versionType - return r -} - -// Source returns a serializable JSON request for the request. -func (r *ReindexDestination) Source() (interface{}, error) { - source := make(map[string]interface{}) - if r.index != "" { - source["index"] = r.index - } - if r.typ != "" { - source["type"] = r.typ - } - if r.routing != "" { - source["routing"] = r.routing - } - if r.opType != "" { - source["op_type"] = r.opType - } - if r.parent != "" { - source["parent"] = r.parent - } - if r.version > 0 { - source["version"] = r.version - } - if r.versionType != "" { - source["version_type"] = r.versionType - } - return source, nil -} |