Revert "remove some bridge related dependencies"

docker-compose/docker-compose-dev-mode.yml

@@ -36,7 +36,8 @@ services:
       - JAVA_MAX_MEM_RATIO=75
       - JAVA_INITIAL_MEM_RATIO=50
       - GC_CONTAINER_OPTIONS=-XX:+UseG1GC
-      - STARGATE_JSONAPI_OPERATIONS_DATABASE_CONFIG_CASSANDRA_END_POINTS=coordinator
+      - QUARKUS_GRPC_CLIENTS_BRIDGE_HOST=coordinator
+      - QUARKUS_GRPC_CLIENTS_BRIDGE_PORT=8091
       - QUARKUS_HTTP_ACCESS_LOG_ENABLED=${REQUESTLOG}
       - QUARKUS_LOG_LEVEL=${LOGLEVEL}
       - JAVA_OPTS_APPEND=-Dquarkus.http.host=0.0.0.0 -Djava.util.logging.manager=org.jboss.logmanager.LogManager

docker-compose/docker-compose.yml

@@ -71,7 +71,8 @@ services:
       - JAVA_MAX_MEM_RATIO=75
       - JAVA_INITIAL_MEM_RATIO=50
       - GC_CONTAINER_OPTIONS=-XX:+UseG1GC
-      - STARGATE_JSONAPI_OPERATIONS_DATABASE_CONFIG_CASSANDRA_END_POINTS=coordinator
+      - QUARKUS_GRPC_CLIENTS_BRIDGE_HOST=coordinator
+      - QUARKUS_GRPC_CLIENTS_BRIDGE_PORT=8091
       - QUARKUS_HTTP_ACCESS_LOG_ENABLED=${REQUESTLOG}
       - QUARKUS_LOG_LEVEL=${LOGLEVEL}
       - JAVA_OPTS_APPEND=-Dquarkus.http.host=0.0.0.0 -Djava.util.logging.manager=org.jboss.logmanager.LogManager

helm/README.md

@@ -1,10 +1,12 @@
 # Stargate JSON API deployment using Helm
 
 ## Pre-requisites
-### Stargate Coordinator Instance
-You'll need the Stargate Coordinator instance deployed with port 9042 accessible
 
-You can deploy the stargate coordinator node using the helm setup at https://github.com/stargate/stargate/tree/main/helm
+### Stargate GRPC Bridge Instance
+You'll need the Stargate Bridge instance deployed with port 8091 accessible
+
+#### Stargate GRPC Bridge Setup
+You can deploy the stargate bridge / coordinator node using the helm setup at https://github.com/stargate/stargate/tree/main/helm
 
 ### Autoscaling
 Autoscaling uses metrics server. Metrics server can be installed by executing the command:
@@ -39,10 +41,13 @@ helm install jsonapi jsonapi
 ```
 
 Note:
-To install with override values, you can use the `--set` option as shown below:
+- The default values in the Helm values file (`values.yaml`) are set to assume that Stargate GRPC Bridge is setup in `default` kubernetes namespace
+
+To install with overriden values, you can use the `--set` option as shown below:
 
 ```shell script
 helm install jsonapi jsonapi \
 --namespace <ENTER_NAMESPACE_HERE> \
+--set sgGrpcBridge.namespace=<SG_GRPC_BRIDGE_NAMESPACE> \
 --set replicaCount=2
 ```

helm/jsonapi/Chart.yaml

@@ -1,6 +1,6 @@
 apiVersion: v2
 name: jsonapi
-description: A Helm chart to deploy stargate JSON API
+description: A Helm chart to deploy stargate JSON API 
 
 # A chart can be either an 'application' or a 'library' chart.
 #
@@ -22,4 +22,3 @@ version: 0.1.0
 # follow Semantic Versioning. They should reflect the version the application is using.
 # It is recommended to use it with quotes.
 appVersion: "3.0.0"
-

helm/jsonapi/templates/stargate_jsonapi_deployment.yaml

@@ -18,41 +18,45 @@ spec:
       affinity:
         podAntiAffinity:
           requiredDuringSchedulingIgnoredDuringExecution:
-            - labelSelector:
-                matchLabels:
-                  app: stargate-jsonapi
-              topologyKey: {{ .Values.topologyKey }}
+          - labelSelector:
+              matchLabels:
+                app: stargate-jsonapi
+            topologyKey: {{ .Values.topologyKey }}
+      initContainers:
+      - name: wait-for-bridge
+        image: busybox
+        command: ["sh", "-c", "until nc -z {{ .Values.sgGrpcBridge.serviceName }}.{{ .Values.sgGrpcBridge.namespace }} 8091 > /dev/null; do echo Waiting for bridge.; sleep 5;done;"]
       containers:
-        - name: jsonapi
-          image: "{{ .Values.image.registry }}/{{ .Values.image.repository }}:{{ .Values.image.tag }}"
-          ports:
-            - containerPort: 8181
-          readinessProbe:
-            httpGet:
-              path: {{ .Values.probe.readiness.url }}
-              port: {{ .Values.probe.readiness.port }}
-            timeoutSeconds: {{ .Values.probe.readiness.timeoutSeconds }}
-            periodSeconds: {{ .Values.probe.readiness.periodSeconds }}
-            failureThreshold: {{ .Values.probe.readiness.failureThreshold }}
-            initialDelaySeconds: {{ .Values.probe.readiness.initialDelaySeconds }}
-          livenessProbe:
-            httpGet:
-              path: {{ .Values.probe.liveness.url }}
-              port: {{ .Values.probe.liveness.port }}
-            timeoutSeconds: {{ .Values.probe.liveness.timeoutSeconds }}
-            periodSeconds: {{ .Values.probe.liveness.periodSeconds }}
-            failureThreshold: {{ .Values.probe.liveness.failureThreshold }}
-            initialDelaySeconds: {{ .Values.probe.liveness.initialDelaySeconds }}
-          resources:
-            requests:
-              cpu: {{ .Values.cpu | default 1000 }}m
-              memory: {{ .Values.memory | default 1024 }}Mi
-          env:
-            - name: STARGATE_JSONAPI_OPERATIONS_DATABASE_CONFIG_CASSANDRA_END_POINTS
-              value: {{ .Values.cassandraEndPoints }}
-            - name: IO_STARGATE_SGV2_API_COMMON_PROPERTIES_DATASTORE_CONFIGURATION_DATASTOREPROPERTIESCONFIGURATION_FETCHSUPPORTEDFEATURES_RETRY_MAXRETRIES
-              value: "50"
-            - name: IO_STARGATE_SGV2_API_COMMON_PROPERTIES_DATASTORE_CONFIGURATION_DATASTOREPROPERTIESCONFIGURATION_FETCHSUPPORTEDFEATURES_RETRY_MAXDURATION
-              value: "300"
-            - name: IO_STARGATE_SGV2_API_COMMON_PROPERTIES_DATASTORE_CONFIGURATION_DATASTOREPROPERTIESCONFIGURATION_FETCHSUPPORTEDFEATURES_RETRY_DELAY
-              value: "10"
+      - name: jsonapi
+        image: "{{ .Values.image.registry }}/{{ .Values.image.repository }}:{{ .Values.image.tag }}"
+        ports:
+        - containerPort: 8181
+        readinessProbe:
+          httpGet:
+            path: {{ .Values.probe.readiness.url }}
+            port: {{ .Values.probe.readiness.port }}
+          timeoutSeconds: {{ .Values.probe.readiness.timeoutSeconds }}
+          periodSeconds: {{ .Values.probe.readiness.periodSeconds }}
+          failureThreshold: {{ .Values.probe.readiness.failureThreshold }}
+          initialDelaySeconds: {{ .Values.probe.readiness.initialDelaySeconds }}
+        livenessProbe:
+          httpGet:
+            path: {{ .Values.probe.liveness.url }}
+            port: {{ .Values.probe.liveness.port }}
+          timeoutSeconds: {{ .Values.probe.liveness.timeoutSeconds }}
+          periodSeconds: {{ .Values.probe.liveness.periodSeconds }}
+          failureThreshold: {{ .Values.probe.liveness.failureThreshold }}
+          initialDelaySeconds: {{ .Values.probe.liveness.initialDelaySeconds }}
+        resources:
+          requests:
+            cpu: {{ .Values.cpu | default 1000 }}m
+            memory: {{ .Values.memory | default 1024 }}Mi
+        env:
+        - name: QUARKUS_GRPC_CLIENTS_BRIDGE_HOST
+          value: "{{ .Values.sgGrpcBridge.serviceName }}.{{ .Values.sgGrpcBridge.namespace }}"
+        - name: IO_STARGATE_SGV2_API_COMMON_PROPERTIES_DATASTORE_CONFIGURATION_DATASTOREPROPERTIESCONFIGURATION_FETCHSUPPORTEDFEATURES_RETRY_MAXRETRIES
+          value: "50"
+        - name: IO_STARGATE_SGV2_API_COMMON_PROPERTIES_DATASTORE_CONFIGURATION_DATASTOREPROPERTIESCONFIGURATION_FETCHSUPPORTEDFEATURES_RETRY_MAXDURATION
+          value: "300"
+        - name: IO_STARGATE_SGV2_API_COMMON_PROPERTIES_DATASTORE_CONFIGURATION_DATASTOREPROPERTIESCONFIGURATION_FETCHSUPPORTEDFEATURES_RETRY_DELAY
+          value: "10"

helm/jsonapi/values.yaml

@@ -7,13 +7,18 @@ replicaCount: 1
 
 # Coordinator or bridge registry and image
 image:
-  registry: "docker.io"
+  registry: "gcr.io"
   repository: "stargateio/jsonapi"
   tag: "v1"
 
 # topology key for anti affinity
 topologyKey: "kubernetes.io/hostname"
 
+# Stargate gRPC bridge hostname and kubernetes namespace
+sgGrpcBridge:
+  serviceName: "stargate-bridge"
+  namespace: "default"
+
 # JSON API service image - don't change it. CPU and memory are both request
 cpu: 2000
 memory: 2048
@@ -33,23 +38,20 @@ probe:
     failureThreshold: 5
     initialDelaySeconds: 30
 
-# HorizontalPodAutoscaler for coordinator and all services. This needs metrics server installed (kubectl apply -f https://github.com/kubernetes-sigs/metrics-server/releases/latest/download/components.yaml) in the cluster.
+# HorizontalPodAutoscaler for coordinator and all services. This needs metrics server installed (kubectl apply -f https://github.com/kubernetes-sigs/metrics-server/releases/latest/download/components.yaml) in the cluster. 
 autoscaling:
   enabled: false
   minReplicas: 1
   maxReplicas: 10
   targetCPUUtilizationPercentage: 80
   # targetMemoryUtilizationPercentage: 80
 
-# Ingress is not added to deployment by default. This needs to have ingress controller installed. Default configuration uses nginx controller. (Controller installed using
-# helm upgrade --install ingress-nginx ingress-nginx \
+# Ingress is not added to deployment by default. This needs to have ingress controller installed. Default configuration uses nginx controller. (Controller installed using 
+# helm upgrade --install ingress-nginx ingress-nginx \           
 #   --repo https://kubernetes.github.io/ingress-nginx \
 #   --namespace ingress-nginx --create-namespace
 
-# To connect to cassandra, set ip for end points
-cassandraEndPoints:  "cassandra end points"
-
 ingress:
   enabled: true
   ingressClassName: nginx
-
+ 

src/main/java/io/stargate/sgv2/jsonapi/JsonapiRetriableStargateBridgeProvider.java

@@ -0,0 +1,21 @@
+package io.stargate.sgv2.jsonapi;
+
+import io.quarkus.grpc.GrpcService;
+import io.stargate.sgv2.api.common.grpc.qualifier.Retriable;
+import jakarta.annotation.Priority;
+import jakarta.enterprise.inject.Alternative;
+import jakarta.inject.Singleton;
+import jakarta.ws.rs.Produces;
+
+@Alternative
+@Priority(1)
+@GrpcService
+@Singleton
+public class JsonapiRetriableStargateBridgeProvider {
+
+  @Produces
+  @Retriable
+  io.stargate.sgv2.api.common.grpc.RetriableStargateBridge retriableStargateBridge() {
+    return null;
+  }
+}

src/main/java/io/stargate/sgv2/jsonapi/exception/ErrorCode.java

@@ -118,6 +118,9 @@ public enum ErrorCode {
 
   UNSUPPORTED_UPDATE_FOR_VECTOR("Cannot use operator with '$vector' property"),
   UNSUPPORTED_UPDATE_FOR_VECTORIZE("Cannot use operator with '$vectorize' property"),
+
+  VECTOR_SEARCH_NOT_AVAILABLE("Vector search functionality is not available in the backend"),
+
   VECTOR_SEARCH_USAGE_ERROR("Vector search can't be used with other sort clause"),
 
   VECTOR_SEARCH_NOT_SUPPORTED("Vector search is not enabled for the collection "),
@@ -132,6 +135,7 @@ public enum ErrorCode {
 
   VECTORIZE_SERVICE_TYPE_UNAVAILABLE("Vectorize service unavailable : "),
   VECTORIZE_USAGE_ERROR("Vectorize search can't be used with other sort clause"),
+
   VECTORIZECONFIG_CHECK_FAIL("Internal server error: VectorizeConfig check fail"),
 
   UNAUTHENTICATED_REQUEST("UNAUTHENTICATED: Invalid token"),

src/main/java/io/stargate/sgv2/jsonapi/grpc/retries/impl/JsonApiGrpcRetryPolicy.java

@@ -0,0 +1,70 @@
+package io.stargate.sgv2.jsonapi.grpc.retries.impl;
+
+import io.grpc.Metadata;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.grpc.protobuf.ProtoUtils;
+import io.stargate.bridge.proto.QueryOuterClass;
+import io.stargate.sgv2.api.common.grpc.retries.GrpcRetryPredicate;
+import jakarta.enterprise.context.ApplicationScoped;
+import java.util.Objects;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Default gRPC retry policy used in the project. The policy defines retries when:
+ *
+ * <ol>
+ *   <li>The received GRPC status code is <code>UNAVAILABLE</code>
+ *   <li>The received GRPC status code is <code>DEADLINE_EXCEEDED</code>, but the metadata received
+ *       contains the trailers set by the Bridge in case of a server-side read, write and CAS write
+ *       timeouts.
+ * </ol>
+ *
+ * Note that this class only defines the policy. Amount of retries and other retry properties are
+ * defined by <code>stargate.grpc.retries</code> property group.
+ */
+@ApplicationScoped
+public class JsonApiGrpcRetryPolicy implements GrpcRetryPredicate {
+  private static final Logger logger = LoggerFactory.getLogger(JsonApiGrpcRetryPolicy.class);
+  private static final Metadata.Key<QueryOuterClass.WriteTimeout> WRITE_TIMEOUT_KEY =
+      ProtoUtils.keyForProto(QueryOuterClass.WriteTimeout.getDefaultInstance());
+
+  private static final Metadata.Key<QueryOuterClass.ReadTimeout> READ_TIMEOUT_KEY =
+      ProtoUtils.keyForProto(QueryOuterClass.ReadTimeout.getDefaultInstance());
+
+  /** {@inheritDoc} */
+  @Override
+  public boolean test(StatusRuntimeException e) {
+    Status status = e.getStatus();
+    Status.Code code = status.getCode();
+
+    // always retry unavailable
+    if (Objects.equals(code, Status.Code.UNAVAILABLE)) {
+      return true;
+    }
+
+    // for timeouts, retry only server side timeouts
+    if (Objects.equals(code, Status.Code.DEADLINE_EXCEEDED)) {
+      return isValidServerSideTimeout(e.getTrailers());
+    }
+
+    // nothing else
+    return false;
+  }
+
+  // ensure we retry only server side timeouts we want
+  private boolean isValidServerSideTimeout(Metadata trailers) {
+    // if we have trailers
+    if (null != trailers) {
+      // read, write and CAS write timeouts will include one of two trailers
+      if (trailers.containsKey(READ_TIMEOUT_KEY) || trailers.containsKey(WRITE_TIMEOUT_KEY)) {
+        logger.warn("DEADLINE_EXCEEDED with read/write timeout trailers, retrying");
+        return true;
+      }
+    }
+
+    // otherwise not
+    return false;
+  }
+}

src/main/java/io/stargate/sgv2/jsonapi/service/operation/model/ReadOperation.java

@@ -11,6 +11,8 @@
 import com.google.common.collect.MinMaxPriorityQueue;
 import io.smallrye.mutiny.Multi;
 import io.smallrye.mutiny.Uni;
+import io.stargate.bridge.grpc.Values;
+import io.stargate.bridge.proto.QueryOuterClass;
 import io.stargate.sgv2.jsonapi.api.v1.metrics.JsonProcessingMetricsReporter;
 import io.stargate.sgv2.jsonapi.exception.ErrorCode;
 import io.stargate.sgv2.jsonapi.exception.JsonApiException;
@@ -333,6 +335,19 @@ default Uni<FindResponse> findOrderDocument(
             });
   }
 
+  /**
+   * Database key type is tuple<int, text>, first field is json value type and second field is text
+   *
+   * @param value
+   * @return
+   */
+  default DocumentId getDocumentId(QueryOuterClass.Value value) {
+    QueryOuterClass.Collection coll = value.getCollection();
+    int typeId = Values.tinyint(coll.getElements(0));
+    String documentIdAsText = Values.string(coll.getElements(1));
+    return DocumentId.fromDatabase(typeId, documentIdAsText);
+  }
+
   default DocumentId getDocumentId(TupleValue value) {
     int typeId = value.get(0, Byte.class);
     String documentIdAsText = value.get(1, String.class);

src/main/java/io/stargate/sgv2/jsonapi/service/operation/model/impl/DBFilterBase.java

@@ -7,11 +7,13 @@
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import io.stargate.bridge.grpc.Values;
+import io.stargate.bridge.proto.QueryOuterClass;
 import io.stargate.sgv2.api.common.cql.builder.BuiltCondition;
 import io.stargate.sgv2.api.common.cql.builder.Predicate;
 import io.stargate.sgv2.jsonapi.exception.ErrorCode;
 import io.stargate.sgv2.jsonapi.exception.JsonApiException;
 import io.stargate.sgv2.jsonapi.service.cqldriver.serializer.CQLBindValues;
+import io.stargate.sgv2.jsonapi.service.cqldriver.serializer.CustomValueSerializers;
 import io.stargate.sgv2.jsonapi.service.shredding.model.DocValueHasher;
 import io.stargate.sgv2.jsonapi.service.shredding.model.DocumentId;
 import io.stargate.sgv2.jsonapi.util.JsonUtil;
@@ -700,6 +702,21 @@ boolean canAddField() {
     }
   }
 
+  private static QueryOuterClass.Value getGrpcValue(Object value) {
+    if (value instanceof String) {
+      return Values.of((String) value);
+    } else if (value instanceof BigDecimal) {
+      return Values.of((BigDecimal) value);
+    } else if (value instanceof Byte) {
+      return Values.of((Byte) value);
+    } else if (value instanceof Integer) {
+      return Values.of((Integer) value);
+    } else if (value instanceof Date) {
+      return Values.of(((Date) value).getTime());
+    }
+    return Values.of((String) null);
+  }
+
   /**
    * Return JsonNode for a filter conditions value, used to set in new document created for upsert.
    *
@@ -748,4 +765,8 @@ private static String getHashValue(DocValueHasher hasher, String path, Object ar
   private static String getHash(DocValueHasher hasher, Object arrayValue) {
     return hasher.getHash(arrayValue).hash();
   }
+
+  private static QueryOuterClass.Value getDocumentIdValue(DocumentId value) {
+    return Values.of(CustomValueSerializers.getDocumentIdValue(value));
+  }
 }