Merge pull request #1249 from jaffee/1242-backport

Fix field "createdAt" race by sending schema changes to coordinator (backport 2.1)

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 {

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")

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)
+				}
+			}
+
+		})
+	}
+}