feat(#541): upload dataset in chunks

Upload datasets in chunks and fix max payload-size
in kubernetes-setup.

Signed-off-by: Tobias Anker <tobias.anker@kitsunemimi.moe>

Author: kitsudaiki

.github/workflows/build_test.yml

@@ -831,7 +831,7 @@ jobs:
 
 
   docker_build:
-    # if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/develop' || github.event_name == 'push' && startsWith(github.ref, 'refs/tags/') }}
+    if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/develop' || github.event_name == 'push' && startsWith(github.ref, 'refs/tags/') }}
     name: "Build Docker-images"
     needs: [ sdk_api_tests, cli_api_tests ]
     runs-on: ubuntu-24.04
@@ -912,7 +912,7 @@ jobs:
 
 
   docker_merge:
-    # if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/develop' || github.event_name == 'push' && startsWith(github.ref, 'refs/tags/') }}
+    if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/develop' || github.event_name == 'push' && startsWith(github.ref, 'refs/tags/') }}
     name: "Merge and push Docker-image"
     runs-on: ubuntu-24.04
     needs: [ docker_build ]
@@ -964,7 +964,7 @@ jobs:
 
 
   kubernetes_test:
-    # if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/develop' || github.event_name == 'push' && startsWith(github.ref, 'refs/tags/') }}
+    if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/develop' || github.event_name == 'push' && startsWith(github.ref, 'refs/tags/') }}
     needs: [ docker_merge ]
     runs-on: ubuntu-22.04
     strategy:

deploy/k8s/openhanami/templates/hanami-ingress.yaml

@@ -6,6 +6,7 @@ metadata:
     nginx.ingress.kubernetes.io/ssl-passthrough: "true"
     nginx.ingress.kubernetes.io/force-ssl-redirect: "true"
     nginx.ingress.kubernetes.io/backend-protocol: HTTPS
+    nginx.ingress.kubernetes.io/proxy-body-size: 1G
 spec:
   ingressClassName: nginx
   rules:

deploy/k8s/openhanami/templates/hanami-nginx-config.yaml

@@ -46,6 +46,7 @@ data:
             # HSTS (ngx_http_headers_module is required) (15768000 seconds = 6 months)
             add_header Strict-Transport-Security max-age=15768000;
 
+            client_max_body_size 1G;
             # OCSP Stapling ---
             # fetch OCSP records from URL in ssl_certificate and cache them
             # ssl_stapling on;

src/hanami/src/api/http_endpoints/dataset/create_dataset_v1_0.rs

@@ -108,7 +108,10 @@ pub async fn upload_binary(mut payload: Multipart, path: Path<(String, String)>,
         while let Some(chunk) = field.next().await {
             let data = match chunk {
                 Ok(value) => value,
-                Err(_) => return Err(ErrorResponse::BadRequest("Failed to read chunk.".to_string())),
+                Err(e) => {
+                    error!("{}", e);
+                    return Err(ErrorResponse::BadRequest("Failed to read chunk.".to_string()));
+                }
             };
 
             let _ = f.write_all(&data).await;

src/hanami/src/api/http_server.rs

@@ -14,6 +14,7 @@
 
 use actix_web::{App, HttpServer};
 use actix_web::middleware::{Logger, from_fn};
+use actix_web::web::PayloadConfig;
 use apistos::app::OpenApiWrapper;
 use apistos::spec::Spec;
 use apistos::web::{post, put, get, delete, resource, scope, Scope};
@@ -173,6 +174,7 @@ pub async fn run_server() -> Result<(), impl Error> {
             .document(spec)
             .wrap(from_fn(authorization_middleware))
             .wrap(Logger::default())
+            .app_data(PayloadConfig::new(1 << 30))  // 1GB max payload-size
             .service(v1alpha_routes())
             .build("/openapi.json")
     })

src/sdk/go/hanami_sdk/http_request.go

@@ -31,8 +31,11 @@ import (
 	"io"
 	"os"
 	"mime/multipart"
+	"path/filepath"
 )
 
+const chunkSize = 1024 * 1024 // 1 MiB
+
 type RequestError struct {
 	StatusCode int
 	Err        string
@@ -192,20 +195,31 @@ func UploadFiles(address, token, path string, filePaths []string, skipTlsVerific
 	writer := multipart.NewWriter(body)
 	outputMap := map[string]interface{}{}
 
-	for _, path := range filePaths {
+	// Open and stream each file into multipart writer
+	for idx, path := range filePaths {
 		file, err := os.Open(path)
 		if err != nil {
-			return outputMap, fmt.Errorf("failed to open file %s: %w", path, err)
+			return outputMap, fmt.Errorf("failed to open file %s: %v", path, err)
 		}
 		defer file.Close()
 
-		part, err := writer.CreateFormFile("file", path)
+		part, err := writer.CreateFormFile(fmt.Sprintf("file%d", idx), filepath.Base(path))
 		if err != nil {
-			return outputMap, fmt.Errorf("failed to create form file for %s: %w", path, err)
+			return outputMap, fmt.Errorf("failed to create form part for %s: %v", path, err)
 		}
 
-		if _, err := io.Copy(part, file); err != nil {
-			return outputMap, fmt.Errorf("failed to copy file data for %s: %w", path, err)
+		buf := make([]byte, chunkSize)
+		for {
+			n, err := file.Read(buf)
+			if err != nil && err != io.EOF {
+				return outputMap, fmt.Errorf("error reading file %s: %v", path, err)
+			}
+			if n == 0 {
+				break
+			}
+			if _, err := part.Write(buf[:n]); err != nil {
+				return outputMap, fmt.Errorf("error writing part for %s: %v", path, err)
+			}
 		}
 	}
 

src/sdk/python/hanami_sdk/hanami_sdk/hanami_request.py

@@ -15,6 +15,7 @@
 import requests
 import json
 import os
+from requests_toolbelt import MultipartEncoder
 
 from . import hanami_exceptions
 
@@ -95,19 +96,28 @@ def upload_files(token: str,
                  path: str,
                  file_paths,
                  verify: bool):
-    files = []
-    for file_path in file_paths:
-        file = open(file_path, 'rb')
-        files.append(('file', (os.path.basename(file_path), file)))
-
     url = f'{address}{path}'
     bearer_token = "Bearer " + token
-    headers = {'Authorization': bearer_token}
+
+    fields = {}
+    open_files = []
+
+    for i, file_path in enumerate(file_paths):
+        f = open(file_path, 'rb')
+        open_files.append(f)  # Keep open until after upload!
+        fields[f'file{i}'] = (os.path.basename(file_path), f, 'application/octet-stream')
+
+    encoder = MultipartEncoder(fields=fields)
+    headers = {
+        'Authorization': bearer_token,
+        'Content-Type': encoder.content_type
+    }
+
     try:
-        response = requests.post(url, headers=headers, files=files, verify=verify)
+        response = requests.post(url, data=encoder, headers=headers, verify=verify)
         return json.loads(_handle_response(response))
     except requests.exceptions.RequestException as e:
         raise e
     finally:
-        for _, (_, fileobj) in files:
-            fileobj.close()
+        for f in open_files:
+            f.close()

src/sdk/python/hanami_sdk/requirements.txt

@@ -3,3 +3,4 @@ protobuf==3.20.3
 requests==2.32.3
 simplejson==3.20.1
 websockets==15.0.1
+requests_toolbelt==1.0.0

src/sdk/python/hanami_sdk/setup.py

@@ -32,7 +32,8 @@ def run(self):
                       'protobuf==3.20.3',
                       'requests==2.32.3',
                       'simplejson==3.20.1',
-                      'websockets==15.0.1'],
+                      'websockets==15.0.1',
+                      'requests_toolbelt==1.0.0'],
     cmdclass={
         'install': GenerateProtobufMessages,
     },