scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-11 07:11:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

forward all CreateIndex/CreateField requests to coordinator

Browse source 
this should avoid a race condition with CreateField where createdAt
can get out of sync if there are multiple concurrent requests.

The client methods didn't allow specification of the URI, so I
modified the implementation to find the coordinator and send to it
explicitly.
This commit is contained in:
Matt Jaffee 2020-12-17 15:21:24 -06:00
parent 58b9418f3c
commit b8cbd54d1b
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
2 changed files with 38 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
api.go
View file

@ -193,6 +193,13 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index
return nil, errors.Wrap(err, "validating api method")
}
if !api.holder.isCoordinator() {
if err := api.server.defaultClient.CreateIndex(ctx, indexName, options); err != nil {
return nil, errors.Wrap(err, "forwarding CreateIndex to coordinator")
}
return api.holder.Index(indexName), nil
}
// Create index.
index, err := api.holder.CreateIndex(indexName, options)
if err != nil {
@ -282,6 +289,13 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str
}
}
if !api.holder.isCoordinator() {
if err := api.server.defaultClient.CreateFieldWithOptions(ctx, indexName, fieldName, fo); err != nil {
return nil, errors.Wrap(err, "forwarding CreateField to coordinator")
}
return api.holder.Field(indexName, fieldName), nil
}
// Find index.
index := api.holder.Index(indexName)
if index == nil {

26
http/client.go
View file

@ -165,6 +165,17 @@ func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt pilo
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.CreateIndex")
defer span.Finish()
// Get the coordinator node. Schema changes must go through
// coordinator to avoid weird race conditions.
nodes, err := c.Nodes(ctx)
if err != nil {
return fmt.Errorf("getting nodes: %s", err)
}
coord := getCoordinatorNode(nodes)
if coord == nil {
return fmt.Errorf("could not find the coordinator node")
}
// Encode query request.
buf, err := json.Marshal(&postIndexRequest{
Options: opt,
@ -174,7 +185,7 @@ func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt pilo
}
// Create URL & HTTP request.
u := uriPathToURL(c.defaultURI, fmt.Sprintf("/index/%s", index))
u := uriPathToURL(&coord.URI, fmt.Sprintf("/index/%s", index))
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
if err != nil {
return errors.Wrap(err, "creating request")
@ -915,8 +926,19 @@ func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, fiel
return errors.Wrap(err, "marshaling")
}
// Get the coordinator node. Schema changes must go through
// coordinator to avoid weird race conditions.
nodes, err := c.Nodes(ctx)
if err != nil {
return fmt.Errorf("getting nodes: %s", err)
}
coord := getCoordinatorNode(nodes)
if coord == nil {
return fmt.Errorf("could not find the coordinator node")
}
// Create URL & HTTP request.
u := uriPathToURL(c.defaultURI, fmt.Sprintf("/index/%s/field/%s", index, field))
u := uriPathToURL(&coord.URI, fmt.Sprintf("/index/%s/field/%s", index, field))
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
if err != nil {
return errors.Wrap(err, "creating request")