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] 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()