From 5fd70b31c803ce7b97c3f0e38a57cc4818d1ef34 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 17 Dec 2020 15:21:24 -0600 Subject: [PATCH 1/2] forward all CreateIndex/CreateField requests to coordinator 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. --- api.go | 14 ++++++++++ http/client.go | 26 ++++++++++++++++-- server/server_test.go | 62 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 100 insertions(+), 2 deletions(-) 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..15b1706f0 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; all bits are sent to the + // primary translate store (i.e. coordinator). + 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; all bits are sent to the + // primary translate store (i.e. coordinator). + 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) + } + } + + }) + } +} From e12aa7234991e35f01ce1c3f235076b81657cbc7 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Fri, 18 Dec 2020 10:15:11 -0600 Subject: [PATCH 2/2] fix comments --- http/client.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/http/client.go b/http/client.go index 15b1706f0..b8a490e68 100644 --- a/http/client.go +++ b/http/client.go @@ -165,8 +165,8 @@ 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; all bits are sent to the - // primary translate store (i.e. coordinator). + // 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) @@ -926,8 +926,8 @@ func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, fiel return errors.Wrap(err, "marshaling") } - // Get the coordinator node; all bits are sent to the - // primary translate store (i.e. coordinator). + // 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)