Failure retries + recovery.

Author: kishorenc

examples/alias_operations.py

@@ -7,8 +7,7 @@
   'nodes': [{
     'host': 'localhost',
     'port': '8108',
-    'protocol': 'http',
-    'api_key': 'abcd'
+    'protocol': 'http'
   }],
   'timeout_seconds': 2
 })

examples/collection_operations.py

@@ -11,8 +11,7 @@
   'nodes': [{
     'host': 'localhost',
     'port': '8108',
-    'protocol': 'http',
-    'api_key': 'abcd'
+    'protocol': 'http'
   }],
   'timeout_seconds': 2
 })

examples/curation_operations.py

@@ -7,8 +7,7 @@
   'nodes': [{
     'host': 'localhost',
     'port': '8108',
-    'protocol': 'http',
-    'api_key': 'abcd'
+    'protocol': 'http'
   }],
   'timeout_seconds': 2
 })

examples/indexing_in_loop.py

@@ -0,0 +1,54 @@
+import os
+import sys
+import json
+
+curr_dir = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(1, os.path.abspath(os.path.join(curr_dir, os.pardir)))
+
+import typesense
+
+client = typesense.Client({
+    'api_key': 'abcd',
+    'nodes': [
+        {
+            'host': 'localhost',
+            'port': '8108',
+            'protocol': 'http'
+        },
+        {
+            'host': 'localhost',
+            'port': '7108',
+            'protocol': 'http'
+        },
+        {
+            'host': 'localhost',
+            'port': '6108',
+            'protocol': 'http'
+        }
+    ],
+    'timeout_seconds': 2
+})
+
+schema = {
+    "name": "books",
+    "fields": [
+        {"name": "title", "type": "string"},
+        {"name": "authors", "type": "string[]"},
+        {"name": "authors_facet", "type": "string[]", "facet": True},
+        {"name": "publication_year", "type": "int32"},
+        {"name": "publication_year_facet", "type": "string", "facet": True},
+        {"name": "ratings_count", "type": "int32"},
+        {"name": "average_rating", "type": "float"},
+        {"name": "image_url", "type": "string"}
+    ],
+    "default_sorting_field": "ratings_count"
+}
+
+create_response = client.collections.create(schema)
+
+print(create_response)
+
+with open('/tmp/books.jsonl') as infile:
+    for json_line in infile:
+        book_document = json.loads(json_line)
+        client.collections['books'].documents.create(book_document)

typesense/api_call.py

@@ -4,7 +4,7 @@
 from .exceptions import (ObjectAlreadyExists,
                          ObjectNotFound, ObjectUnprocessable,
                          RequestMalformed, RequestUnauthorized,
-                         ServerError, TypesenseClientError)
+                         ServerError, ServiceUnavailable, TypesenseClientError)
 
 
 class ApiCall(object):
@@ -15,27 +15,31 @@ def __init__(self, config):
         self.config = config
         self.nodes = self.config.nodes
         self.node_index = 0
-        self.failed_nodes = []
-        self.last_fail_check_ts = int(time.time())
+        self.last_health_check_ts = int(time.time())
 
     def _check_failed_node(self):
         current_epoch_ts = int(time.time())
-        check_failed_node = ((current_epoch_ts - self.last_fail_check_ts) > ApiCall.CHECK_FAILED_NODE_INTERVAL_S)
-        if check_failed_node:
-            self.last_fail_check_ts = current_epoch_ts
+        check_node = ((current_epoch_ts - self.last_health_check_ts) > ApiCall.CHECK_FAILED_NODE_INTERVAL_S)
+        if check_node:
+            self.last_health_check_ts = current_epoch_ts
 
-        return check_failed_node
+        return check_node
 
     # Returns a healthy host from the pool in a round-robin fashion.
     # Might return an unhealthy host periodically to check for recovery.
     def get_node(self):
-        num_times = 0
-        while num_times < 3:
-            num_times += 1
+        i = 0
+        while i < len(self.nodes):
+            i += 1
             self.node_index = (self.node_index + 1) % len(self.nodes)
             if self.nodes[self.node_index].healthy or self._check_failed_node():
                 return self.nodes[self.node_index]
 
+        # None of the nodes are marked healthy, but some of them could have become healthy since last health check.
+        # So we will just return the next node.
+        self.node_index = (self.node_index + 1) % len(self.nodes)
+        return self.nodes[self.node_index]
+
     @staticmethod
     def get_exception(http_code):
         if http_code == 400:
@@ -50,6 +54,8 @@ def get_exception(http_code):
             return ObjectUnprocessable
         elif http_code == 500:
             return ServerError
+        elif http_code == 503:
+            return ServiceUnavailable
         else:
             return TypesenseClientError
 
@@ -59,16 +65,20 @@ def make_request(self, fn, method, endpoint, as_json, **kwargs):
         while num_tries < self.config.num_retries:
             num_tries += 1
             node = self.get_node()
+            node.healthy = False
 
             try:
                 url = node.url() + endpoint
-                print('URL is: ' + url)
                 r = fn(url, headers={ApiCall.API_KEY_HEADER_NAME: self.config.api_key}, **kwargs)
+
+                if 0 < r.status_code < 500:
+                    node.healthy = True
+
                 if (method != 'post' and r.status_code != 200) or (method == 'post' and r.status_code != 201):
                     error_message = r.json().get('message', 'API error.')
+                    # print('error_message: ' + error_message)
                     raise ApiCall.get_exception(r.status_code)(error_message)
 
-                node.healthy = True
                 return r.json() if as_json else r.text
             except requests.exceptions.Timeout:
                 pass
@@ -79,10 +89,10 @@ def make_request(self, fn, method, endpoint, as_json, **kwargs):
             except Exception as e:
                 raise e
 
-            node.healthy = False
+            # print('Failed, retrying after sleep: ' + node.port)
             time.sleep(self.config.retry_interval_seconds)
 
-        raise TypesenseClientError('All hosts are bad.')
+        raise TypesenseClientError('Retries exceeded.')
 
     def get(self, endpoint, params=None, as_json=True):
         params = params or {}

typesense/exceptions.py

@@ -33,3 +33,7 @@ class ObjectUnprocessable(TypesenseClientError):
 
 class ServerError(TypesenseClientError):
     pass
+
+
+class ServiceUnavailable(TypesenseClientError):
+    pass