Fix k8 errors (dgarnitz#94)

* fixed embed bug

* made minio service more secure

* added rabbit mq fix to hf app

* uncommented code

---------

Co-authored-by: David Garnitz <[email protected]>

README.md

@@ -114,7 +114,9 @@ Note that the Sentence Transformer models can be large and take several minutes
 
 ## Using VectorFlow
 
-To use VectorFlow in a live system, make an HTTP request to your API's URL at port 8000 - for example, `localhost:8000` from your development machine, or `vectorflow_api:8000` from within another docker container.
+The best way to use VectorFlow is with the python client. 
+
+To use VectorFlow for development, make an HTTP request to your API's URL - for example, `localhost:8000` from your development machine, or `vectorflow_api:8000` from within another docker container.
 
 ### Request & Response Payload
 

kube/hugging_face_deployment.yaml

@@ -0,0 +1,76 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: hugging-face
+  namespace: vectorflow
+  labels:
+    app: hugging-face
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: hugging-face
+  strategy: {}
+  template:
+    metadata:
+        labels:
+          io.kompose.network/vectorflow: "true"
+          app: hugging-face
+    spec:
+      containers:
+        - args:
+            - bash
+            - /wait-for-it.sh
+            - rabbitmq:5672
+            - --
+            - python
+            - app.py 
+            - --model_name # this is currently failing, not read in due to weird way the script runs
+            - "BAAI/bge-small-en"
+          env:
+          - name: VDB_UPLOAD_QUEUE
+            valueFrom:
+              configMapKeyRef:
+                key: VDB_UPLOAD_QUEUE
+                name: config-map
+          - name: POSTGRES_DB
+            valueFrom:
+              configMapKeyRef:
+                key: POSTGRES_DB
+                name: config-map
+          - name: POSTGRES_HOST
+            valueFrom:
+              configMapKeyRef:
+                key: POSTGRES_HOST
+                name: config-map
+          - name: POSTGRES_PASSWORD
+            valueFrom:
+              configMapKeyRef:
+                key: POSTGRES_PASSWORD
+                name: config-map
+          - name: POSTGRES_USERNAME
+            valueFrom:
+              configMapKeyRef:
+                key: POSTGRES_USERNAME
+                name: config-map
+          - name: RABBITMQ_HOST
+            valueFrom:
+              configMapKeyRef:
+                key: RABBITMQ_HOST
+                name: config-map
+          - name: RABBITMQ_PASSWORD
+            valueFrom:
+              configMapKeyRef:
+                key: RABBITMQ_PASSWORD
+                name: config-map
+          - name: RABBITMQ_USERNAME
+            valueFrom:
+              configMapKeyRef:
+                key: RABBITMQ_USERNAME
+                name: config-map
+          image: vectorflow_hf
+          name: hugging-face
+          imagePullPolicy: IfNotPresent
+          resources: {}
+      restartPolicy: Always
+status: {}

kube/minio-service.yaml

@@ -9,7 +9,6 @@ metadata:
   name: minio
   namespace: vectorflow
 spec:
-  type: LoadBalancer
   ports:
     - name: "9000"
       port: 9000

kube/scripts/deploy-local-k8s.sh

@@ -30,18 +30,18 @@ echo "Creating PVCS..."
 kubectl apply -f kube/postgres-pvc.yaml
 kubectl apply -f kube/minio-pvc.yaml
 
-echo "Deploying initial Deployments..."
-kubectl apply -f kube/postgres-deployment.yaml
-kubectl apply -f kube/rabbitmq-deployment.yaml
-kubectl apply -f kube/minio-deployment.yaml
-kubectl apply -f kube/qdrant-deployment.yaml
-
 echo "Deploying initial Services"
 kubectl apply -f kube/postgres-service.yaml
 kubectl apply -f kube/rabbitmq-service.yaml
 kubectl apply -f kube/minio-service.yaml
 kubectl apply -f kube/qdrant-service.yaml
 
+echo "Deploying initial Deployments..."
+kubectl apply -f kube/postgres-deployment.yaml
+kubectl apply -f kube/rabbitmq-deployment.yaml
+kubectl apply -f kube/minio-deployment.yaml
+kubectl apply -f kube/qdrant-deployment.yaml
+
 echo "Deploying resources with init containers..."
 kubectl apply -f kube/db-init.yaml
 kubectl apply -f kube/qdrant-init.yaml

src/api/app.py

@@ -278,7 +278,8 @@ def process_file(file, vectorflow_request, job_id):
     batch_count = create_batches(file_content, job_id, vectorflow_request)
     return batch_count
 
-def create_batches(file_content, job_id, vectorflow_request):
+def create_batches(file_content, job_id, vectorflow_request_original):
+    vectorflow_request = copy.deepcopy(vectorflow_request_original)
     chunks = [chunk for chunk in split_file(file_content, vectorflow_request.lines_per_batch)]
 
     batches = [Batch(job_id=job_id, embeddings_metadata=vectorflow_request.embeddings_metadata, vector_db_metadata=vectorflow_request.vector_db_metadata) for _ in chunks]

src/hugging_face/Dockerfile

@@ -17,7 +17,7 @@ COPY services /app/services
 # install requirements
 RUN pip install --no-cache-dir -r /app/hugging_face/requirements.txt
 
-# Set the working directory to /app/worker, as 'worker.py' is here
+# Set the working directory to /app/hugging_face, as 'app.py' is here
 WORKDIR /app/hugging_face
 
 # Run worker.py when the container launches

src/hugging_face/app.py

@@ -18,6 +18,8 @@
 from services.database import job_service
 from shared.utils import send_embeddings_to_webhook
 from shared.job_status import JobStatus
+from services.rabbitmq.rabbit_service import create_connection_params
+from pika.exceptions import AMQPConnectionError
 
 model = None
 publish_channel = None
@@ -120,27 +122,13 @@ def callback(ch, method, properties, body):
 
     ch.basic_ack(delivery_tag=method.delivery_tag)
 
-def start_connection():
+def start_connection(max_retries=5, retry_delay=5):
     global publish_channel
     global model_name
 
-    while True:
+    for attempt in range(max_retries):
         try:
-            credentials = pika.PlainCredentials(os.getenv('RABBITMQ_USERNAME'), os.getenv('RABBITMQ_PASSWORD'))
-
-            connection_params = pika.ConnectionParameters(
-                host=os.getenv('RABBITMQ_HOST'),
-                credentials=credentials,
-                port=os.getenv('RABBITMQ_PORT'),
-                heartbeat=600,
-                ssl_options=pika.SSLOptions(ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)),
-                virtual_host="/"
-            ) if os.getenv('RABBITMQ_PORT') == "5671" else pika.ConnectionParameters(
-                host=os.getenv('RABBITMQ_HOST'),
-                credentials=credentials,
-                heartbeat=600,
-            )
-
+            connection_params = create_connection_params()
             connection = pika.BlockingConnection(connection_params)
             consume_channel = connection.channel()
             publish_channel = connection.channel() 
@@ -155,13 +143,28 @@ def start_connection():
 
             logging.info('Waiting for messages.')
             consume_channel.start_consuming()
+            return  # If successful, exit the function
 
+        except AMQPConnectionError as e:
+            logging.error('AMQP Connection Error: %s', e)
         except Exception as e:
-            logging.error('ERROR connecting to RabbitMQ, retrying now. See exception: %s', e)
-            time.sleep(10) # Wait before retrying
+            logging.error('Unexpected error: %s', e)
+        finally:
+            if connection and not connection.is_closed:
+                connection.close()
+
+        logging.info('Retrying to connect in %s seconds (Attempt %s/%s)', retry_delay, attempt + 1, max_retries)
+        time.sleep(retry_delay)
 
 if __name__ == "__main__": 
     args = get_args()
     model_name = args.model_name
     model = SentenceTransformer(model_name)
-    start_connection()
+    while True:
+        try:
+            start_connection()
+
+        except Exception as e:
+            logging.error('Error in start_connection: %s', e)
+            logging.info('Restarting start_connection after encountering an error.')
+            time.sleep(10)

testing-clients/hugging_face_upload_client.py

@@ -5,7 +5,7 @@
 #####################
 # Testing Variables #
 #####################
-filepath = './api/tests/fixtures/test_medium_text.txt'
+filepath = 'src/api/tests/fixtures/test_medium_text.txt'
 url = "http://localhost:8000/embed"
 embedding_key = os.getenv("OPEN_AI_KEY")
 embedding_type="HUGGING_FACE"