From ea44877e73ee4be6e1aa257a30e14231e70e980e Mon Sep 17 00:00:00 2001
From: Amund Isaksen <31344542+Teddy-1000@users.noreply.github.com>
Date: Wed, 2 Oct 2024 10:23:23 +0200
Subject: [PATCH] Add retry config (#189)

* Add retry config

* Change max retry timeout

* Add UNAVAILABLE to retryableStatusCodes

* Update retryPolicy rules

* Update to apply to all methods

* Add option to enable retry
---
 ingest/api/grpc_putter.py | 27 +++++++++++++++++++++++++--
 1 file changed, 25 insertions(+), 2 deletions(-)

diff --git a/ingest/api/grpc_putter.py b/ingest/api/grpc_putter.py
index f0bf5fa1..5fcf973f 100644
--- a/ingest/api/grpc_putter.py
+++ b/ingest/api/grpc_putter.py
@@ -1,5 +1,6 @@
 import logging
 import os
+import json
 
 from functools import cache
 
@@ -12,7 +13,29 @@
 
 @cache
 def get_grpc_stub():
-    channel = grpc.aio.insecure_channel(f"{os.getenv('DSHOST', 'store')}:{os.getenv('DSPORT', '50050')}")
+    options = []
+    grpc_config = json.dumps(
+        {
+            "methodConfig": [
+                {
+                    "name": [{}],
+                    "retryPolicy": {
+                        "maxAttempts": 5,
+                        "initialBackoff": "0.5s",
+                        "maxBackoff": "8s",
+                        "backoffMultiplier": 2,
+                        "retryableStatusCodes": ["INTERNAL", "UNAVAILABLE", "OUT_OF_RANGE"],
+                    },
+                }
+            ]
+        }
+    )
+    options.append(("grpc.enable_retries", 1))
+    options.append(("grpc.service_config", grpc_config))
+    channel = grpc.aio.insecure_channel(
+        f"{os.getenv('DSHOST', 'store')}:{os.getenv('DSPORT', '50050')}", options=options
+    )
+
     return dstore_grpc.DatastoreStub(channel)
 
 
@@ -23,4 +46,4 @@ async def putObsRequest(put_obs_request):
         logger.debug("RPC call succeeded.")
     except grpc.aio.AioRpcError as grpc_error:
         logger.critical(f"RPC call failed: {grpc_error.code()}\n{grpc_error.details()}")
-        raise HTTPException(detail=grpc_error.details(), status_code=400)
+        raise HTTPException(detail=f"GRPC_ERROR:{grpc_error.details()}", status_code=400)