Preserve RAM at RAG indexing by streaming files (#4613)

At RAG indexation time, we used to read the file from disk, keep it in
memory in a buffer and send it to the network. This can be problematic
if we have plenty of parallel indexing jobs, as the RAM consumption
could skyrocket. So, we now stream the file in a Pipe by chunks of 32KB
(the default size of `io.Copy`), avoiding RAM peak consumption even with
large files.

Author: paultranvan

model/rag/index.go

@@ -221,55 +221,66 @@ func callRAGIndexer(inst *instance.Instance, doctype string, change couchdb.Chan
 			}
 		}
 
-		var requestBody bytes.Buffer
-		writer := multipart.NewWriter(&requestBody)
-		part, err := writer.CreateFormFile("file", name)
-		if err != nil {
-			return err
-		}
-		_, err = io.Copy(part, content)
-		if err != nil {
-			return err
-		}
-		// No need to add filename here, it is already set through the file form
-		meta := map[string]string{
-			"md5sum":   md5sum,
-			"datetime": datetime,
-			"doctype":  doctype,
-		}
-		ragMetadata, err := json.Marshal(meta)
-		if err != nil {
-			return err
-		}
-		_ = writer.WriteField("metadata", string(ragMetadata))
-		err = writer.Close()
-		if err != nil {
-			return err
-		}
 		u.RawQuery = url.Values{
 			"dir_id": []string{dirID},
 			"name":   []string{name},
 			"md5sum": []string{md5sum},
 		}.Encode()
-
 		u.Path = fmt.Sprintf("/indexer/partition/%s/file/%s", inst.Domain, change.DocID)
+
+		// Create pipe and writer for file streaming
+		pr, pw := io.Pipe()
+		writer := multipart.NewWriter(pw)
+
+		go func() {
+			part, err := writer.CreateFormFile("file", name)
+			if err != nil {
+				_ = pw.CloseWithError(err)
+				return
+			}
+			if _, err := io.Copy(part, content); err != nil {
+				_ = pw.CloseWithError(err)
+				return
+			}
+
+			// No need to add filename here, it is already set through the file form
+			meta := map[string]string{
+				"md5sum":   md5sum,
+				"datetime": datetime,
+				"doctype":  doctype,
+			}
+			ragMetadata, err := json.Marshal(meta)
+			if err != nil {
+				_ = pw.CloseWithError(err)
+				return
+			}
+			if err := writer.WriteField("metadata", string(ragMetadata)); err != nil {
+				_ = pw.CloseWithError(err)
+				return
+			}
+			defer pw.Close()
+			defer writer.Close()
+		}()
+
 		if isNewFile {
-			req, err = http.NewRequest(http.MethodPost, u.String(), &requestBody)
+			req, err = http.NewRequest(http.MethodPost, u.String(), pr)
 		} else {
-			req, err = http.NewRequest(http.MethodPut, u.String(), &requestBody)
+			req, err = http.NewRequest(http.MethodPut, u.String(), pr)
 		}
 		if err != nil {
 			return err
 		}
+
 		req.Header.Add(echo.HeaderAuthorization, "Bearer "+ragServer.APIKey)
 		req.Header.Add("Content-Type", writer.FormDataContentType())
 
 		res, err = http.DefaultClient.Do(req)
 		if err != nil {
 			return err
 		}
+
 		var response map[string]interface{}
-		if err = json.NewDecoder(res.Body).Decode(&response); err != nil {
+		if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
 			return err
 		}
 		defer res.Body.Close()