From 5e23d0a70838eebee33354bb1e7cce307b8a65e5 Mon Sep 17 00:00:00 2001 From: Shikhar Gupta Date: Thu, 30 Apr 2026 10:34:31 +0530 Subject: [PATCH] Refactor HTTP clients to use shared utilities and proper error handling - The embedding and docling clients had a lot of duplicated code for making HTTP requests - same logic for creating requests, setting headers, handling responses. This was getting messy, especially when we needed to check for specific error codes like rate limits (429) or validation errors (422) --- .../documentprocessor_controller.go | 12 +- .../vectorembeddingsgenerator_controller.go | 15 +-- pkg/commonhttp/client.go | 88 ++++++++++++++ pkg/commonhttp/errors.go | 69 +++++++++++ pkg/docling/client.go | 107 ++++++------------ pkg/embedding/client.go | 52 +-------- 6 files changed, 216 insertions(+), 127 deletions(-) create mode 100644 pkg/commonhttp/client.go create mode 100644 pkg/commonhttp/errors.go diff --git a/internal/controller/documentprocessor_controller.go b/internal/controller/documentprocessor_controller.go index 1439a7d71..0a7fbf036 100644 --- a/internal/controller/documentprocessor_controller.go +++ b/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 } diff --git a/internal/controller/vectorembeddingsgenerator_controller.go b/internal/controller/vectorembeddingsgenerator_controller.go index 155292593..7f0fefc3c 100644 --- a/internal/controller/vectorembeddingsgenerator_controller.go +++ b/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...) diff --git a/pkg/commonhttp/client.go b/pkg/commonhttp/client.go new file mode 100644 index 000000000..590745841 --- /dev/null +++ b/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 +} diff --git a/pkg/commonhttp/errors.go b/pkg/commonhttp/errors.go new file mode 100644 index 000000000..2bd412257 --- /dev/null +++ b/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 +} diff --git a/pkg/docling/client.go b/pkg/docling/client.go index 81c46eb5b..eae3ba762 100644 --- a/pkg/docling/client.go +++ b/pkg/docling/client.go @@ -17,16 +17,14 @@ limitations under the License. package docling import ( - "bytes" "context" "encoding/json" "errors" "fmt" - "io" "net/http" "net/url" - "time" + "github.com/redhat-data-and-ai/unstructured-data-controller/pkg/commonhttp" "golang.org/x/sync/semaphore" "sigs.k8s.io/controller-runtime/pkg/log" ) @@ -95,6 +93,7 @@ type DoclingRequestPayload struct { } type Client struct { + Client *http.Client `json:"doclingclient"` ClientConfig *ClientConfig `json:"client_config"` } @@ -129,6 +128,9 @@ type TaskStatusResponse struct { func NewClientFromURL(clientConfig *ClientConfig) *Client { clientConfig.sem = semaphore.NewWeighted(clientConfig.MaxConcurrentRequests) return &Client{ + Client: &http.Client{ + Timeout: commonhttp.HTTPClientTimeout, + }, ClientConfig: clientConfig, } } @@ -154,58 +156,35 @@ func (c *Client) getTaskResultEndpoint(taskID string) (string, error) { return url.JoinPath(c.ClientConfig.URL, "/v1/result", taskID) } -func (c *Client) createHTTPRequest(ctx context.Context, method, endpoint string, payload []byte, authFormat string) ( - *http.Request, error) { - req, err := http.NewRequestWithContext(ctx, method, endpoint, bytes.NewReader(payload)) - 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 c.ClientConfig.Key != "" { - req.Header.Set("Authorization", fmt.Sprintf(authFormat, c.ClientConfig.Key)) - } - return req, nil -} - -func (c *Client) createDoclingRequest(ctx context.Context, method, endpoint string, payload []byte) ( - io.ReadCloser, error) { +func (c *Client) createDoclingRequest(ctx context.Context, method, endpoint string, payload []byte) ([]byte, error) { logger := log.FromContext(ctx) - client := &http.Client{ - Timeout: 15 * time.Second, - } - req, err := c.createHTTPRequest(ctx, method, endpoint, payload, "Bearer %s") + logger.Info("sending request to docling service", "url", endpoint) + + req, err := commonhttp.CreateHTTPRequest(ctx, method, endpoint, payload, "Bearer", c.ClientConfig.Key) if err != nil { return nil, fmt.Errorf("failed to create request: %w", err) } - logger.Info("sending request to docling service", "url", endpoint) - resp, err := client.Do(req) + statusCode, body, err := commonhttp.Do(ctx, c.Client, req) if err != nil { - return nil, fmt.Errorf("failed to send request: %w", err) - } - - if resp.StatusCode == http.StatusForbidden && c.ClientConfig.Key != "" { - req, err = c.createHTTPRequest(ctx, method, endpoint, payload, "%s") - if err != nil { - return nil, fmt.Errorf("failed to create request: %w", err) - } - resp, err = client.Do(req) - if err != nil { - return nil, fmt.Errorf("failed to send request: %w", err) + // If we get 403 with Bearer auth, try without Bearer prefix + if statusCode == http.StatusForbidden && c.ClientConfig.Key != "" { + logger.Info("retrying with raw API key auth", "url", endpoint) + req, err = commonhttp.CreateHTTPRequest(ctx, method, endpoint, payload, "", c.ClientConfig.Key) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + _, body, err = commonhttp.Do(ctx, c.Client, req) + if err != nil { + return nil, err + } + return body, nil } + return nil, err } - if resp.StatusCode != http.StatusOK { - logger.Error(errors.New("received non-200 OK response from endpoint"), - "docling request returned non-200 status", - "statusCode", resp.StatusCode, - "url", endpoint) - return nil, fmt.Errorf("failed to process request: status code %d", resp.StatusCode) - } - - return resp.Body, nil + return body, nil } func (c *Client) ConvertFile( @@ -251,19 +230,13 @@ func (c *Client) ConvertFile( logger.Info("sending request to convert file", "urlToSendRequest", convertSourceAsyncEndpoint, "sourceFileURL", baseURL) - // convert response to AsyncDoclingResponse - var asyncResponse AsyncDoclingResponse - responseBody, err := c.createDoclingRequest(ctx, http.MethodPost, convertSourceAsyncEndpoint, payload) - if err != nil { - return nil, fmt.Errorf("failed to get response body: %w", err) - } - defer func() { _ = responseBody.Close() }() - body, err := io.ReadAll(responseBody) + body, err := c.createDoclingRequest(ctx, http.MethodPost, convertSourceAsyncEndpoint, payload) if err != nil { - return nil, fmt.Errorf("failed to read response body: %w", err) + return nil, fmt.Errorf("failed to get response body: %w", err) } + var asyncResponse AsyncDoclingResponse if err = json.Unmarshal(body, &asyncResponse); err != nil { return nil, fmt.Errorf("failed to decode response: %w", err) } @@ -281,22 +254,17 @@ func (c *Client) getTaskStatus(ctx context.Context, taskID string) (bool, *TaskS } logger.Info("sending request to get status of task", "url", getTaskStatusPollEndpoint) - var taskStatusResponse TaskStatusResponse - bodyResponse, err := c.createDoclingRequest(ctx, http.MethodGet, getTaskStatusPollEndpoint, nil) + + body, err := c.createDoclingRequest(ctx, http.MethodGet, getTaskStatusPollEndpoint, nil) if err != nil { return false, nil, fmt.Errorf("failed to get response body: %w", err) } - if err := json.NewDecoder(bodyResponse).Decode(&taskStatusResponse); err != nil { + var taskStatusResponse TaskStatusResponse + if err := json.Unmarshal(body, &taskStatusResponse); err != nil { return false, nil, fmt.Errorf("failed to decode response: %w", err) } - defer func() { - if err = bodyResponse.Close(); err != nil { - err = fmt.Errorf("failed to close response body: %w", err) - } - }() - return true, &taskStatusResponse, nil } @@ -335,16 +303,15 @@ func (c *Client) GetConvertedFile(ctx context.Context, taskID string) (TaskStatu } logger.Info("sending request to get converted file", "url", taskResultURL) - var doclingResponse DoclingResponse - bodyResponse, err := c.createDoclingRequest(ctx, http.MethodGet, taskResultURL, nil) + + body, err := c.createDoclingRequest(ctx, http.MethodGet, taskResultURL, nil) if err != nil { c.safeRelease() return "", nil, fmt.Errorf("failed to get response body: %w", err) } - defer func() { _ = bodyResponse.Close() }() - if err := json.NewDecoder(bodyResponse).Decode(&doclingResponse); err != nil { - c.safeRelease() + var doclingResponse DoclingResponse + if err := json.Unmarshal(body, &doclingResponse); err != nil { return "", nil, fmt.Errorf("failed to decode response: %w", err) } @@ -369,16 +336,12 @@ func (c *Client) GetConvertedFile(ctx context.Context, taskID string) (TaskStatu return doclingResponse.Status, &doclingResponse, nil } -// getBaseURl will return url without query and fragments, only hostname and path func getBaseURL(fullURL string) (string, error) { u, err := url.Parse(fullURL) if err != nil { return "", fmt.Errorf("failed to parse URL: %w", err) } - - // set query and fragment as empty u.RawQuery = "" u.Fragment = "" - return u.String(), nil } diff --git a/pkg/embedding/client.go b/pkg/embedding/client.go index c6a2477b8..da47b9f7f 100644 --- a/pkg/embedding/client.go +++ b/pkg/embedding/client.go @@ -1,21 +1,15 @@ package embedding import ( - "bytes" "context" "encoding/json" "fmt" - "io" "net/http" - "time" + "github.com/redhat-data-and-ai/unstructured-data-controller/pkg/commonhttp" "sigs.k8s.io/controller-runtime/pkg/log" ) -const ( - HTTPClientTimeout = 60 * time.Second -) - type EmbeddingGenerator interface { GenerateEmbeddings(ctx context.Context, inputs []string, encodingFormat string) (*EmbeddingResult, error) } @@ -59,32 +53,12 @@ type HTTPClient struct { func NewHTTPClient(config *HTTPClientConfig) *HTTPClient { return &HTTPClient{ Client: &http.Client{ - Timeout: HTTPClientTimeout, + Timeout: commonhttp.HTTPClientTimeout, }, Config: config, } } -// createHTTPRequest creates an HTTP request with auth header set from format and api key. -func (c *HTTPClient) createHTTPRequest( - ctx context.Context, method, endpoint string, payload []byte, -) (*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 c.Config.APIKey != "" && c.Config.AuthFormat != "" { - req.Header.Set("Authorization", fmt.Sprintf("%s %s", c.Config.AuthFormat, c.Config.APIKey)) - } - return req, nil -} - func (c *HTTPClient) GenerateEmbeddings( ctx context.Context, inputs []string, encodingFormat string, ) (*EmbeddingResult, error) { @@ -103,30 +77,16 @@ func (c *HTTPClient) GenerateEmbeddings( return nil, fmt.Errorf("failed to marshal embedding request: %w", err) } - // TODO: Add a better log statement logger.Info("sending embedding request") - req, err := c.createHTTPRequest(ctx, http.MethodPost, c.Config.Endpoint, payload) + req, err := commonhttp.CreateHTTPRequest(ctx, http.MethodPost, + c.Config.Endpoint, payload, c.Config.AuthFormat, c.Config.APIKey) if err != nil { return nil, err } - resp, err := c.Client.Do(req) - if err != nil { - return nil, fmt.Errorf("failed to send embedding 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) + _, body, err := commonhttp.Do(ctx, c.Client, req) if err != nil { - return nil, fmt.Errorf("failed to read embedding response: %w", err) - } - - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("API returned status %d: %s", resp.StatusCode, string(body)) + return nil, err } var embResp EmbeddingResponse