Refactor HTTP clients to use shared utilities and proper error handling

internal/controller/documentprocessor_controller.go

@@ -39,6 +39,7 @@ import (
 	"github.com/redhat-data-and-ai/unstructured-data-controller/internal/controller/controllerutils"
 	"github.com/redhat-data-and-ai/unstructured-data-controller/pkg/docling"
 	"github.com/redhat-data-and-ai/unstructured-data-controller/pkg/filestore"
+	"github.com/redhat-data-and-ai/unstructured-data-controller/pkg/commonhttp"
 	"github.com/redhat-data-and-ai/unstructured-data-controller/pkg/unstructured"
 )
 
@@ -222,6 +223,9 @@ func (r *DocumentProcessorReconciler) reconcileJob(ctx context.Context, job oper
 
 	doclingTaskStatus, doclingResponse, err := doclingClient.GetConvertedFile(ctx, job.TaskID)
 	if err != nil {
+		if commonhttp.IsStatusUnprocessableEntity(err) {
+			logger.Error(err, "docling validation error (422)", "taskID", job.TaskID, "filePath", job.FilePath)
+		}
 		return err
 	}
 
@@ -321,11 +325,15 @@ func (r *DocumentProcessorReconciler) processDocument(ctx context.Context, rawFi
 	}
 	response, err := doclingClient.ConvertFile(ctx, fileURL, *r.doclingConfig)
 	if err != nil {
-		logger.Error(err, "failed to convert file")
 		if strings.Contains(err.Error(), docling.SemaphoreAcquireError) {
-			logger.Error(err, "failed to convert file, semaphore acquire error, will try again later")
+			logger.Info("semaphore acquire error, will try again later", "filePath", rawFilePath)
 			return nil // no error, just skip the conversion this time
 		}
+		if commonhttp.IsStatusUnprocessableEntity(err) {
+			logger.Error(err, "docling validation error (422), check docling config", "filePath", rawFilePath)
+			return err
+		}
+		logger.Error(err, "failed to convert file", "filePath", rawFilePath)
 		return err
 	}
 

internal/controller/vectorembeddingsgenerator_controller.go

@@ -21,7 +21,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"strings"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime"
@@ -38,6 +37,7 @@ import (
 	"github.com/redhat-data-and-ai/unstructured-data-controller/internal/controller/controllerutils"
 	"github.com/redhat-data-and-ai/unstructured-data-controller/pkg/embedding"
 	"github.com/redhat-data-and-ai/unstructured-data-controller/pkg/filestore"
+	"github.com/redhat-data-and-ai/unstructured-data-controller/pkg/commonhttp"
 	"github.com/redhat-data-and-ai/unstructured-data-controller/pkg/unstructured"
 )
 
@@ -224,12 +224,13 @@ func (r *VectorEmbeddingsGeneratorReconciler) processChunkedFile(ctx context.Con
 
 		logger.Info("processing batch", "batchStart", batchStart, "batchEnd", batchEnd, "batchSize", len(batch))
 		embeddingResult, err := embeddingClient.GenerateEmbeddings(ctx, batch, encodingFormat)
-		if err != nil {
-			if strings.Contains(err.Error(), "status 429") {
-				logger.Error(err, "embedding API rate limited (429), will retry on next reconciliation", "file", chunksFilePath, "batchStart", batchStart)
-			} else {
-				logger.Error(err, "failed to generate embeddings for batch", "file", chunksFilePath, "batchStart", batchStart, "batchEnd", batchEnd)
-			}
+		if err != nil && commonhttp.IsStatusTooManyRequests(err) {
+			logger.Info("rate limit exceeded (429), will retry after 5 seconds", "batchStart", batchStart, "batchEnd", batchEnd)
+			time.Sleep(5 * time.Second)
+			batchStart -= batchSize
+			continue
+		} else if err != nil {
+			logger.Error(err, "failed to generate embeddings for batch", "batchStart", batchStart, "batchEnd", batchEnd)
 			return false, err
 		}
 		allEmbeddings = append(allEmbeddings, embeddingResult.Embeddings...)

pkg/commonhttp/client.go

@@ -0,0 +1,88 @@
+/*
+Copyright 2026.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package commonhttp
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+const (
+	HTTPClientTimeout = 60 * time.Second
+)
+
+// CreateHTTPRequest creates an HTTP request with common headers and optional auth
+func CreateHTTPRequest(ctx context.Context, method, endpoint string, payload []byte, authFormat,
+	apiKey string) (*http.Request, error) {
+	var body io.Reader
+	if len(payload) > 0 {
+		body = bytes.NewReader(payload)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, method, endpoint, body)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create request: %w", err)
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+
+	if apiKey != "" {
+		if authFormat != "" {
+			req.Header.Set("Authorization", fmt.Sprintf("%s %s", authFormat, apiKey))
+		} else {
+			req.Header.Set("Authorization", apiKey)
+		}
+	}
+
+	return req, nil
+}
+
+// Do executes an HTTP request and handles error status codes
+func Do(ctx context.Context, client *http.Client, req *http.Request) (int, []byte, error) {
+	logger := log.FromContext(ctx)
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return 0, nil, fmt.Errorf("failed to send request: %w", err)
+	}
+	defer func() {
+		if err := resp.Body.Close(); err != nil {
+			logger.Error(err, "failed to close response body")
+		}
+	}()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return resp.StatusCode, nil, fmt.Errorf("failed to read response body: %w", err)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return resp.StatusCode, body, &HTTPError{
+			StatusCode: resp.StatusCode,
+			Body:       body,
+		}
+	}
+
+	return resp.StatusCode, body, nil
+}

pkg/commonhttp/errors.go

@@ -0,0 +1,69 @@
+/*
+Copyright 2026.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package commonhttp
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+)
+
+// HTTPError represents any non-200 HTTP response
+type HTTPError struct {
+	StatusCode int
+	Body       []byte
+}
+
+func (e *HTTPError) Error() string {
+	return fmt.Sprintf("HTTP %d", e.StatusCode)
+}
+
+// Status code 413 - Batch size error (embeddings)
+func IsStatusPayloadTooLarge(err error) bool {
+	var httpErr *HTTPError
+	if errors.As(err, &httpErr) {
+		return httpErr.StatusCode == http.StatusRequestEntityTooLarge
+	}
+	return false
+}
+
+// Status code 422 - Tokenization error (embeddings) OR Validation error (docling)
+func IsStatusUnprocessableEntity(err error) bool {
+	var httpErr *HTTPError
+	if errors.As(err, &httpErr) {
+		return httpErr.StatusCode == http.StatusUnprocessableEntity
+	}
+	return false
+}
+
+// Status code 424 - Embedding error / Inference failed (embeddings)
+func IsStatusFailedDependency(err error) bool {
+	var httpErr *HTTPError
+	if errors.As(err, &httpErr) {
+		return httpErr.StatusCode == http.StatusFailedDependency
+	}
+	return false
+}
+
+// Status code 429 - Rate limit / Model overloaded
+func IsStatusTooManyRequests(err error) bool {
+	var httpErr *HTTPError
+	if errors.As(err, &httpErr) {
+		return httpErr.StatusCode == http.StatusTooManyRequests
+	}
+	return false
+}