diff --git a/api.go b/api.go index 4483e0db9..478323afa 100644 --- a/api.go +++ b/api.go @@ -181,6 +181,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 { @@ -270,6 +277,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 { diff --git a/http/client.go b/http/client.go index 943604bf9..b8a490e68 100644 --- a/http/client.go +++ b/http/client.go @@ -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") diff --git a/server/server_test.go b/server/server_test.go index 3534fe0d0..7bf1e82be 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -1210,3 +1210,65 @@ Set("h", adec=100.22) } } + +// TestClusterCreatedAtRace is a regression test for an issue where +// creating the same field concurrently across the cluster could cause +// the createdAt value to be disagreed upon by various nodes in the +// cluster (which would cause ingest to fail). +func TestClusterCreatedAtRace(t *testing.T) { + iterations := 1 + if runStress { + iterations = 10 + } + for k := 0; k < iterations; k++ { + t.Run(fmt.Sprintf("run-%d", k), func(t *testing.T) { + cluster := test.MustRunCluster(t, 4) + defer cluster.Close() + + for _, com := range cluster { + nodes := com.API.Hosts(context.Background()) + for _, n := range nodes { + if n.State != "READY" { + t.Fatalf("unexpected node state after upping cluster: %v", nodes) + } + } + } + _, err := cluster[0].API.CreateIndex(context.Background(), "anindex", pilosa.IndexOptions{}) + if err != nil && errors.Cause(err).Error() != pilosa.ErrIndexExists.Error() { + t.Fatal(err) + } + + eg := errgroup.Group{} + for i := 0; i < 4; i++ { + for _, cmd := range cluster { + cmd := cmd + eg.Go(func() error { + _, err := cmd.API.CreateField(context.Background(), "anindex", "afield") + if err != nil && errors.Cause(err).Error() != pilosa.ErrFieldExists.Error() { + return errors.Wrap(err, "creating field") + } + return nil + }) + } + } + + err = eg.Wait() + if err != nil { + t.Fatalf("creating indices and fields concurrently: %v", err) + } + + schemas := make([]*pilosa.IndexInfo, len(cluster)) + for i, cmd := range cluster { + schemas[i] = cmd.API.Schema(context.Background())[0] + } + + createdAtField := schemas[0].Fields[0].CreatedAt + for i, schema := range schemas[1:] { + if schema.Fields[0].CreatedAt != createdAtField { + t.Fatalf("node %d doesn't match node 0 for field. 0: %d, %d: %d", i, createdAtField, i, schema.Fields[0].CreatedAt) + } + } + + }) + } +}