diff --git a/api.go b/api.go index 6149c6aee..34155ee32 100644 --- a/api.go +++ b/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") } diff --git a/client/client.go b/client/client.go index a023f9c29..7e5796d74 100644 --- a/client/client.go +++ b/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 diff --git a/client/client_it_test.go b/client/client_it_test.go index ff8614d78..a58622bc8 100644 --- a/client/client_it_test.go +++ b/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) { diff --git a/client/cluster.go b/client/cluster.go index 0f1230583..cda424905 100644 --- a/client/cluster.go +++ b/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() diff --git a/client/cluster_test.go b/client/cluster_test.go index 36790b7f8..53d63222a 100644 --- a/client/cluster_test.go +++ b/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") - } -} diff --git a/client/error.go b/client/error.go index 3c6685b62..f0fbb873f 100644 --- a/client/error.go +++ b/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") diff --git a/client/ingest_api_batch_test.go b/client/ingest_api_batch_test.go index 9abfa15c6..bc6858c12 100644 --- a/client/ingest_api_batch_test.go +++ b/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() diff --git a/http_handler.go b/http_handler.go index 4756b0fab..3fc47cad4 100644 --- a/http_handler.go +++ b/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()