Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add Point In Time API #1480

Merged
merged 1 commit into from Apr 12, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
10 changes: 10 additions & 0 deletions client.go
Expand Up @@ -1636,6 +1636,16 @@ func (c *Client) ClearScroll(scrollIds ...string) *ClearScrollService {
return NewClearScrollService(c).ScrollId(scrollIds...)
}

// OpenPointInTime opens a new Point in Time.
func (c *Client) OpenPointInTime(indices ...string) *OpenPointInTimeService {
return NewOpenPointInTimeService(c).Index(indices...)
}

// ClosePointInTime closes an existing Point in Time.
func (c *Client) ClosePointInTime(id string) *ClosePointInTimeService {
return NewClosePointInTimeService(c).ID(id)
}

// -- Indices APIs --

// CreateIndex returns a service to create a new index.
Expand Down
36 changes: 36 additions & 0 deletions pit.go
@@ -0,0 +1,36 @@
// 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

// PointInTime is a lightweight view into the state of the data that existed
// when initiated. It can be created with OpenPointInTime API and be used
// when searching, e.g. in Search API or with SearchSource.
type PointInTime struct {
// Id that uniquely identifies the point in time, as created with the
// OpenPointInTime API.
Id string `json:"id,omitempty"`
// KeepAlive is the time for which this specific PointInTime will be
// kept alive by Elasticsearch.
KeepAlive string `json:"keep_alive,omitempty"`
}

// NewPointInTime creates a new PointInTime.
func NewPointInTime(id, keepAlive string) *PointInTime {
return &PointInTime{
Id: id,
KeepAlive: keepAlive,
}
}

// Source generates the JSON serializable fragment for the PointInTime.
func (pit *PointInTime) Source() (interface{}, error) {
if pit == nil {
return nil, nil
}
return map[string]interface{}{
"id": pit.Id,
"keep_alive": pit.KeepAlive,
}, nil
}
176 changes: 176 additions & 0 deletions pit_close.go
@@ -0,0 +1,176 @@
// 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/http"
"net/url"
"strings"
)

// ClosePointInTimeService removes a point in time.
//
// See https://www.elastic.co/guide/en/elasticsearch/reference/7.x/point-in-time-api.html
// for details.
type ClosePointInTimeService struct {
client *Client

pretty *bool // pretty format the returned JSON response
human *bool // return human readable values for statistics
errorTrace *bool // include the stack trace of returned errors
filterPath []string // list of filters used to reduce the response
headers http.Header // custom request-level HTTP headers

id string
bodyJson interface{}
bodyString string
}

// NewClosePointInTimeService creates a new ClosePointInTimeService.
func NewClosePointInTimeService(client *Client) *ClosePointInTimeService {
return &ClosePointInTimeService{
client: client,
}
}

// Pretty tells Elasticsearch whether to return a formatted JSON response.
func (s *ClosePointInTimeService) Pretty(pretty bool) *ClosePointInTimeService {
s.pretty = &pretty
return s
}

// Human specifies whether human readable values should be returned in
// the JSON response, e.g. "7.5mb".
func (s *ClosePointInTimeService) Human(human bool) *ClosePointInTimeService {
s.human = &human
return s
}

// ErrorTrace specifies whether to include the stack trace of returned errors.
func (s *ClosePointInTimeService) ErrorTrace(errorTrace bool) *ClosePointInTimeService {
s.errorTrace = &errorTrace
return s
}

// FilterPath specifies a list of filters used to reduce the response.
func (s *ClosePointInTimeService) FilterPath(filterPath ...string) *ClosePointInTimeService {
s.filterPath = filterPath
return s
}

// Header adds a header to the request.
func (s *ClosePointInTimeService) Header(name string, value string) *ClosePointInTimeService {
if s.headers == nil {
s.headers = http.Header{}
}
s.headers.Add(name, value)
return s
}

// Headers specifies the headers of the request.
func (s *ClosePointInTimeService) Headers(headers http.Header) *ClosePointInTimeService {
s.headers = headers
return s
}

// ID to close.
func (s *ClosePointInTimeService) ID(id string) *ClosePointInTimeService {
s.id = id
return s
}

// BodyJson is the document as a serializable JSON interface.
func (s *ClosePointInTimeService) BodyJson(body interface{}) *ClosePointInTimeService {
s.bodyJson = body
return s
}

// BodyString is the document encoded as a string.
func (s *ClosePointInTimeService) BodyString(body string) *ClosePointInTimeService {
s.bodyString = body
return s
}

// buildURL builds the URL for the operation.
func (s *ClosePointInTimeService) buildURL() (string, string, url.Values, error) {
var (
method = "DELETE"
path = "/_pit"
)

// Add query string parameters
params := url.Values{}
if v := s.pretty; v != nil {
params.Set("pretty", fmt.Sprint(*v))
}
if v := s.human; v != nil {
params.Set("human", fmt.Sprint(*v))
}
if v := s.errorTrace; v != nil {
params.Set("error_trace", fmt.Sprint(*v))
}
if len(s.filterPath) > 0 {
params.Set("filter_path", strings.Join(s.filterPath, ","))
}
return method, path, params, nil
}

// Validate checks if the operation is valid.
func (s *ClosePointInTimeService) Validate() error {
return nil
}

// Do executes the operation.
func (s *ClosePointInTimeService) Do(ctx context.Context) (*ClosePointInTimeResponse, error) {
// Check pre-conditions
if err := s.Validate(); err != nil {
return nil, err
}

// Get URL for request
method, path, params, err := s.buildURL()
if err != nil {
return nil, err
}

// Setup HTTP request body
var body interface{}
if s.id != "" {
body = map[string]interface{}{
"id": s.id,
}
} else if s.bodyJson != nil {
body = s.bodyJson
} else {
body = s.bodyString
}

// Get HTTP response
res, err := s.client.PerformRequest(ctx, PerformRequestOptions{
Method: method,
Path: path,
Params: params,
Body: body,
Headers: s.headers,
})
if err != nil {
return nil, err
}

// Return operation response
ret := new(ClosePointInTimeResponse)
if err := s.client.decoder.Decode(res.Body, ret); err != nil {
return nil, err
}
return ret, nil
}

// ClosePointInTimeResponse is the result of closing a point in time.
type ClosePointInTimeResponse struct {
Succeeded bool `json:"succeeded,omitempty"`
NumFreed int `json:"num_freed,omitempty"`
}