From 17e8dbd318085d2d0046c846d0d5a9a3cd59ccca Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Fri, 25 Feb 2022 16:14:43 -0600 Subject: [PATCH 1/2] client side retry ingestAPI requests on primary host If non-primary host fails to process a request, retry on primary node. conditions when we should not do this: - no error - we've aleady tried the primary - we're making a status request to get the primary node...this could lead to lock contention if we allow it to happen as we are making an http request within an on going http request to discover the primary node. This also deletes the RemoveHost method and the associated test b/c it is not used anywhere anymore and updates the returned error type. --- client/client.go | 21 +++++++-------------- client/client_it_test.go | 2 +- client/cluster.go | 12 ------------ client/cluster_test.go | 19 ------------------- client/error.go | 2 +- client/ingest_api_batch_test.go | 1 - 6 files changed, 9 insertions(+), 48 deletions(-) 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() From 48997347b9bdc28632e638f9f9d03531b6ed8e60 Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Mon, 7 Mar 2022 15:39:18 -0600 Subject: [PATCH 2/2] add server side redirect to primary for ingest requests this applies to handlePostIngestData, handlePostIngestSchema also moves a helper method (ApplyOneIngestSchema) from http handler to API --- api.go | 165 +++++++++++++++++++++++++++++++++++++++++++ http_handler.go | 181 ++++++------------------------------------------ 2 files changed, 186 insertions(+), 160 deletions(-) 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/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()