Merge pull request #1944 from molecula/bug-primary-ingest

[FB-1082] retry request on primary host

api.go

@@ -1039,6 +1039,164 @@ func (api *API) ApplySchema(ctx context.Context, s *Schema, remote bool) error {
 	return nil
 }
 
+// applyOneIngestSchema applies a single ingestSpec, which specifies operations on
+// a single index and possibly fields. If it is successful, it returns the name
+// of the index and an empty slice (if it created the index), or the name of the
+// index and a slice of the fields within that index that it created. If it
+// is unsuccessful, it tries to delete whatever it created.
+//
+// The intended idiom is that if the returned list of fields isn't empty, the index
+// already existed and only those fields need to be cleaned up in the event of
+// a later error, but if the list of fields is empty, the entire index was new,
+// and should be cleaned up, in which case there's no need to track or delete
+// the specific fields separately.
+func (api *API) ApplyOneIngestSchema(ctx context.Context, schema *ingestSpec) (index *Index, returnedFields []string, err error) {
+	if api.PrimaryNode().ID != api.NodeID() {
+		return nil, nil, RedirectError{
+			HostPort: api.PrimaryNode().URI.Normalize(),
+			error:    "request made to non-primary node",
+		}
+	}
+
+	// create index
+	indexName := schema.IndexName
+	var createdFields []string
+	var useKeys bool
+	switch schema.PrimaryKeyType {
+	case "string":
+		useKeys = true
+	case "uint":
+		useKeys = false
+	default:
+		return nil, nil, fmt.Errorf("invalid primary key type %q", schema.PrimaryKeyType)
+	}
+	opts := IndexOptions{
+		Keys:           useKeys,
+		TrackExistence: true,
+	}
+	createdIndex := false
+
+	// We check this up here because, if there's at least one field but we don't know what to do with
+	// it, we will necessarily fail, which means we'd delete the index anyway, so there's no point in
+	// trying to create it. We don't care about this if there's no fields specified.
+	if len(schema.Fields) > 0 {
+		switch schema.FieldAction {
+		case "create", "ensure", "require":
+			// do nothing
+		case "":
+			schema.FieldAction = schema.IndexAction
+		default:
+			return nil, nil, fmt.Errorf("invalid field-action %q, expecting create/ensure/require", schema.FieldAction)
+		}
+	}
+
+	switch schema.IndexAction {
+	case "ensure", "require":
+		index, err = api.Index(ctx, indexName)
+		if err != nil {
+			if _, ok := err.(NotFoundError); !ok {
+				return nil, nil, fmt.Errorf("checking for existing index %q: %w", indexName, err)
+			} else {
+				err = nil
+			}
+		}
+		if index != nil {
+			existingOpts := index.Options()
+			if existingOpts != opts {
+				return nil, nil, fmt.Errorf("index %q options mismatch: schema %#v, existing %#v", indexName, opts, existingOpts)
+			}
+			break
+		}
+		if schema.IndexAction == "require" {
+			return nil, nil, fmt.Errorf("index %q does not exist", indexName)
+		}
+		fallthrough
+	case "create":
+		index, err = api.CreateIndex(ctx, indexName, opts)
+		if err != nil {
+			return nil, nil, err
+		}
+		createdIndex = true
+	default:
+		return nil, nil, fmt.Errorf("invalid index-action %q, need create/ensure/require", schema.IndexAction)
+	}
+
+	// Now we might have an index, so we need our cleanup code.
+	defer func() {
+		if err == nil {
+			return
+		}
+		if createdIndex {
+			err := api.DeleteIndex(ctx, indexName)
+			if err != nil {
+
+				api.server.logger.Printf("trying to undo failed index %q creation: %v", indexName, err)
+			}
+			return
+		}
+		for _, field := range createdFields {
+			err := api.DeleteField(ctx, indexName, field)
+			if err != nil {
+				api.server.logger.Printf("trying to undo failed field %q creation in index %q: %v", field, indexName, err)
+			}
+		}
+	}()
+
+	// create all the fields specified in the index
+	for _, fSpec := range schema.Fields {
+		fieldName := fSpec.FieldName
+		opt := fieldSpecToFieldOption(fSpec)
+		err = opt.validate()
+		if err != nil {
+			return nil, nil, err
+		}
+		switch schema.FieldAction {
+		case "ensure", "require":
+			field, schemaErr := api.Field(ctx, indexName, fieldName)
+			if schemaErr != nil {
+				// NotFoundError is fine
+				if _, ok := schemaErr.(NotFoundError); !ok {
+					return nil, nil, fmt.Errorf("checking for existing field %q in %q: %w", fieldName, indexName, err)
+				}
+			}
+			if field != nil {
+				existing := field.Options()
+				if opt.Type != existing.Type {
+					return nil, nil, fmt.Errorf("existing field %q is %q, not %q", fieldName, existing.Type, opt.Type)
+				}
+				if ((opt.Keys != nil) && *opt.Keys) != existing.Keys {
+					if existing.Keys {
+						return nil, nil, fmt.Errorf("existing field %q in %q uses keys", fieldName, indexName)
+					} else {
+						return nil, nil, fmt.Errorf("existing field %q in %q doesn't use keys", fieldName, indexName)
+					}
+				}
+				// TODO: verify compatibility of other field opts, this is sorta hard
+				break
+			}
+			if schema.FieldAction == "require" {
+				return nil, nil, fmt.Errorf("field %q does not exist in %q", fieldName, indexName)
+			}
+			fallthrough
+		case "create":
+			fos := fieldOptionsToFunctionalOpts(opt)
+			_, err = api.CreateField(ctx, indexName, fieldName, fos...)
+			if err != nil {
+				return nil, nil, fmt.Errorf("creating field %q in %q: %v", fieldName, indexName, err)
+			}
+			createdFields = append(createdFields, fieldName)
+		}
+	}
+
+	// we don't report the fields back, so we can distinguish "created index"
+	// from "created fields within index"
+	if createdIndex {
+		createdFields = nil
+	}
+
+	return index, createdFields, nil
+}
+
 // Views returns the views in the given field.
 func (api *API) Views(ctx context.Context, indexName string, fieldName string) ([]*view, error) {
 	span, _ := tracing.StartSpanFromContext(ctx, "API.Views")
@@ -1618,6 +1776,13 @@ func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string
 	span, _ := tracing.StartSpanFromContext(ctx, "API.IngestOperations")
 	defer span.Finish()
 
+	if api.PrimaryNode().ID != api.NodeID() {
+		return RedirectError{
+			HostPort: api.PrimaryNode().URI.Normalize(),
+			error:    "request made to non-primary node",
+		}
+	}
+
 	if err := api.validate(apiIngestOperations); err != nil {
 		return errors.Wrap(err, "validating api method")
 	}

client/client.go

@@ -832,31 +832,24 @@ func (c *Client) httpRequest(method string, path string, data []byte, headers ma
 		body   []byte
 		err    error
 	)
-	// try at most maxHosts non-failed hosts; protect against broken cluster.removeHost
-	for i := 0; i < maxHosts; i++ {
+	// try request on host, if it fails, try again on primary
+	for i := 0; i <= 1; i++ {
 		host, herr := c.host(usePrimary)
 		if herr != nil {
 			return status, nil, errors.Wrapf(herr, "getting host, previous err: %v", err)
 		}
 		// doRequest implements expotential backoff
 		status, body, err = c.doRequest(host, method, path, c.augmentHeaders(headers), data)
-		if err == nil {
+		// conditions when primary should not be tried
+		if err == nil || usePrimary || path == "/status" {
 			break
 		}
-		if c.manualServerURI == nil {
-			if usePrimary {
-				c.primaryLock.Lock()
-				c.primaryURI = nil
-				c.primaryLock.Unlock()
-			} else {
-				c.logger.Printf("removing host (%s) due to '%v'\n", host.Normalize(), err)
-				c.cluster.RemoveHost(host)
-			}
-		}
+
+		usePrimary = true
 	}
 
 	if err != nil {
-		err = errors.Wrap(err, ErrTriedMaxHosts.Error())
+		err = errors.Wrap(err, ErrHTTPRequest.Error())
 	}
 
 	return status, body, err

client/client_it_test.go

@@ -497,7 +497,7 @@ func TestClientAgainstCluster(t *testing.T) {
 				tmpcli, _ := NewClient(NewClusterWithHost(uri, uri, uri, uri), OptClientRetries(0))
 
 				_, err := tmpcli.Query(testIndex.All())
-				require.Error(t, err, ErrTriedMaxHosts)
+				require.Error(t, err, ErrHTTPRequest)
 			})
 
 			t.Run("InvalidQuery", func(t *testing.T) {

client/cluster.go

@@ -65,18 +65,6 @@ func (c *Cluster) Host() *pnet.URI {
 	return host
 }
 
-// RemoveHost black lists the host with the given pnet.URI from the cluster.
-func (c *Cluster) RemoveHost(address *pnet.URI) {
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
-	for i, uri := range c.hosts {
-		if uri.Equals(address) {
-			c.okList[i] = false
-			break
-		}
-	}
-}
-
 // Hosts returns all available hosts in the cluster.
 func (c *Cluster) Hosts() []pnet.URI {
 	c.mutex.RLock()

client/cluster_test.go

@@ -49,22 +49,3 @@ func TestHosts(t *testing.T) {
 		t.Fatalf("Host should return a value if there are hosts in the cluster")
 	}
 }
-
-func TestRemoveHost(t *testing.T) {
-	uri, err := pnet.NewURIFromAddress("index1.pilosa.com:9999")
-	if err != nil {
-		t.Fatal(err)
-	}
-	c := NewClusterWithHost(uri)
-	if len(c.hosts) != 1 {
-		t.Fatalf("The cluster should contain the host")
-	}
-	uri, err = pnet.NewURIFromAddress("index1.pilosa.com:9999")
-	if err != nil {
-		t.Fatal(err)
-	}
-	c.RemoveHost(uri)
-	if len(c.Hosts()) != 0 {
-		t.Fatalf("The cluster should not contain the host")
-	}
-}

client/error.go

@@ -12,7 +12,7 @@ var (
 	ErrInvalidFieldName            = errors.New("Invalid field name")
 	ErrInvalidLabel                = errors.New("Invalid label")
 	ErrInvalidKey                  = errors.New("Invalid key")
-	ErrTriedMaxHosts               = errors.New("Tried max hosts, still failing")
+	ErrHTTPRequest                 = errors.New("Failed all HTTP retries")
 	ErrAddrURIClusterExpected      = errors.New("Addresses, URIs or a cluster is expected")
 	ErrInvalidQueryOption          = errors.New("Invalid query option")
 	ErrInvalidIndexOption          = errors.New("Invalid index option")

client/ingest_api_batch_test.go

@@ -133,7 +133,6 @@ func TestIngestAPIBatchAdd(t *testing.T) {
 }
 
 func TestIngestAPIBatch(t *testing.T) {
-	t.Skip("causing sporadic CI failures... on my list to debug, but this code doesn't affect anyone's production anyhow (jaffee)")
 	c := test.MustRunCluster(t, 3)
 	defer c.Close()
 

http_handler.go

@@ -1672,14 +1672,18 @@ func (h *Handler) handlePostIngestData(w http.ResponseWriter, r *http.Request) {
 
 	qcx := h.api.Txf().NewQcx()
 	err := h.api.IngestOperations(r.Context(), qcx, indexName, r.Body)
-	if err == nil {
-		err = qcx.Finish()
-		if err != nil {
-			http.Error(w, fmt.Sprintf("ingesting: %v", err), http.StatusInternalServerError)
+	if err != nil {
+		qcx.Abort()
+		switch e := err.(type) {
+		case RedirectError:
+			http.Redirect(w, r, e.HostPort+r.URL.Path, http.StatusPermanentRedirect)
 			return
 		}
-	} else {
-		qcx.Abort()
+	}
+	err = qcx.Finish()
+	if err != nil {
+		http.Error(w, fmt.Sprintf("ingesting: %v", err), http.StatusInternalServerError)
+		return
 	}
 
 	resp := successResponse{h: h, Name: indexName}
@@ -1747,155 +1751,6 @@ func fieldSpecToFieldOption(fSpec fieldSpec) fieldOptions {
 	return opt
 }
 
-// applyOneIngestSchema applies a single ingestSpec, which specifies operations on
-// a single index and possibly fields. If it is successful, it returns the name
-// of the index and an empty slice (if it created the index), or the name of the
-// index and a slice of the fields within that index that it created. If it
-// is unsuccessful, it tries to delete whatever it created.
-//
-// The intended idiom is that if the returned list of fields isn't empty, the index
-// already existed and only those fields need to be cleaned up in the event of
-// a later error, but if the list of fields is empty, the entire index was new,
-// and should be cleaned up, in which case there's no need to track or delete
-// the specific fields separately.
-func (h *Handler) applyOneIngestSchema(ctx context.Context, schema *ingestSpec) (index *Index, returnedFields []string, err error) {
-	// create index
-	indexName := schema.IndexName
-	var createdFields []string
-	var useKeys bool
-	switch schema.PrimaryKeyType {
-	case "string":
-		useKeys = true
-	case "uint":
-		useKeys = false
-	default:
-		return nil, nil, fmt.Errorf("invalid primary key type %q", schema.PrimaryKeyType)
-	}
-	opts := IndexOptions{
-		Keys:           useKeys,
-		TrackExistence: true,
-	}
-	createdIndex := false
-
-	// We check this up here because, if there's at least one field but we don't know what to do with
-	// it, we will necessarily fail, which means we'd delete the index anyway, so there's no point in
-	// trying to create it. We don't care about this if there's no fields specified.
-	if len(schema.Fields) > 0 {
-		switch schema.FieldAction {
-		case "create", "ensure", "require":
-			// do nothing
-		case "":
-			schema.FieldAction = schema.IndexAction
-		default:
-			return nil, nil, fmt.Errorf("invalid field-action %q, expecting create/ensure/require", schema.FieldAction)
-		}
-	}
-
-	switch schema.IndexAction {
-	case "ensure", "require":
-		index, err = h.api.Index(ctx, indexName)
-		if err != nil {
-			if _, ok := err.(NotFoundError); !ok {
-				return nil, nil, fmt.Errorf("checking for existing index %q: %w", indexName, err)
-			} else {
-				err = nil
-			}
-		}
-		if index != nil {
-			existingOpts := index.Options()
-			if existingOpts != opts {
-				return nil, nil, fmt.Errorf("index %q options mismatch: schema %#v, existing %#v", indexName, opts, existingOpts)
-			}
-			break
-		}
-		if schema.IndexAction == "require" {
-			return nil, nil, fmt.Errorf("index %q does not exist", indexName)
-		}
-		fallthrough
-	case "create":
-		index, err = h.api.CreateIndex(ctx, indexName, opts)
-		if err != nil {
-			return nil, nil, err
-		}
-		createdIndex = true
-	default:
-		return nil, nil, fmt.Errorf("invalid index-action %q, need create/ensure/require", schema.IndexAction)
-	}
-
-	// Now we might have an index, so we need our cleanup code.
-	defer func() {
-		if err == nil {
-			return
-		}
-		if createdIndex {
-			err := h.api.DeleteIndex(ctx, indexName)
-			if err != nil {
-				h.logger.Printf("trying to undo failed index %q creation: %v", indexName, err)
-			}
-			return
-		}
-		for _, field := range createdFields {
-			err := h.api.DeleteField(ctx, indexName, field)
-			if err != nil {
-				h.logger.Printf("trying to undo failed field %q creation in index %q: %v", field, indexName, err)
-			}
-		}
-	}()
-
-	// create all the fields specified in the index
-	for _, fSpec := range schema.Fields {
-		fieldName := fSpec.FieldName
-		opt := fieldSpecToFieldOption(fSpec)
-		err = opt.validate()
-		if err != nil {
-			return nil, nil, err
-		}
-		switch schema.FieldAction {
-		case "ensure", "require":
-			field, schemaErr := h.api.Field(ctx, indexName, fieldName)
-			if schemaErr != nil {
-				// NotFoundError is fine
-				if _, ok := schemaErr.(NotFoundError); !ok {
-					return nil, nil, fmt.Errorf("checking for existing field %q in %q: %w", fieldName, indexName, err)
-				}
-			}
-			if field != nil {
-				existing := field.Options()
-				if opt.Type != existing.Type {
-					return nil, nil, fmt.Errorf("existing field %q is %q, not %q", fieldName, existing.Type, opt.Type)
-				}
-				if ((opt.Keys != nil) && *opt.Keys) != existing.Keys {
-					if existing.Keys {
-						return nil, nil, fmt.Errorf("existing field %q in %q uses keys", fieldName, indexName)
-					} else {
-						return nil, nil, fmt.Errorf("existing field %q in %q doesn't use keys", fieldName, indexName)
-					}
-				}
-				// TODO: verify compatibility of other field opts, this is sorta hard
-				break
-			}
-			if schema.FieldAction == "require" {
-				return nil, nil, fmt.Errorf("field %q does not exist in %q", fieldName, indexName)
-			}
-			fallthrough
-		case "create":
-			fos := fieldOptionsToFunctionalOpts(opt)
-			_, err = h.api.CreateField(ctx, indexName, fieldName, fos...)
-			if err != nil {
-				return nil, nil, fmt.Errorf("creating field %q in %q: %v", fieldName, indexName, err)
-			}
-			createdFields = append(createdFields, fieldName)
-		}
-	}
-	// we don't report the fields back, so we can distinguish "created index"
-	// from "created fields within index"
-	if createdIndex {
-		createdFields = nil
-	}
-
-	return index, createdFields, nil
-}
-
 func (h *Handler) handleIngestSchema(w http.ResponseWriter, r *http.Request) {
 	if !validHeaderAcceptJSON(r.Header) {
 		http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
@@ -1938,12 +1793,18 @@ func (h *Handler) handleIngestSchema(w http.ResponseWriter, r *http.Request) {
 			resp.write(w, err)
 			return
 		}
-		index, fields, err := h.applyOneIngestSchema(r.Context(), &schema)
+		index, fields, err := h.api.ApplyOneIngestSchema(r.Context(), &schema)
 		if err != nil {
-			// if a previous schema created things, clean them up...
-			schemaErr = err
-			resp.write(w, err)
-			return
+			switch e := err.(type) {
+			case RedirectError:
+				http.Redirect(w, r, e.HostPort+r.URL.Path, http.StatusPermanentRedirect)
+				return
+			default:
+				// if a previous schema created things, clean them up...
+				schemaErr = err
+				resp.write(w, err)
+				return
+			}
 		}
 		// we only have one slot to report these, sorry.
 		resp.Name = index.Name()