From bcb345103fbe70b9248e8b2b0711cdff6dfdd179 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Sat, 16 May 2026 14:08:58 +0330 Subject: [PATCH] minio connection log --- internal/tender/service.go | 29 ++++++++++-- pkg/ai_summarizer/errors.go | 4 ++ pkg/ai_summarizer/minio_errors.go | 55 +++++++++++++++++++++ pkg/ai_summarizer/storage.go | 71 +++++++++++++++++++++++----- pkg/ai_summarizer/storage_logging.go | 45 ++++++++++++++++++ 5 files changed, 186 insertions(+), 18 deletions(-) create mode 100644 pkg/ai_summarizer/storage_logging.go diff --git a/internal/tender/service.go b/internal/tender/service.go index f81471e..8be73ae 100644 --- a/internal/tender/service.go +++ b/internal/tender/service.go @@ -187,6 +187,14 @@ func (s *tenderService) lookupSummaryForTender(ctx context.Context, t *Tender) ( if !errors.Is(err, ai_summarizer.ErrObjectNotFound) && !errors.Is(err, ai_summarizer.ErrSummaryNotReady) && !errors.Is(err, ai_summarizer.ErrNoOverallSummary) { + if errors.Is(err, ai_summarizer.ErrMinIOUnavailable) { + s.logger.Error("MinIO unavailable while fetching AI summary", map[string]interface{}{ + "tender_id": t.ID, + "contract_folder_id": t.ContractFolderID, + "notice_id": noticeID, + "error": err.Error(), + }) + } return "", noticeID, err } } @@ -889,6 +897,15 @@ func (s *tenderService) GetAISummary(ctx context.Context, id string) (*AISummary if errors.Is(err, ai_summarizer.ErrObjectNotFound) || errors.Is(err, ai_summarizer.ErrNoOverallSummary) { return emptyResp("unavailable"), nil } + if errors.Is(err, ai_summarizer.ErrMinIOUnavailable) { + s.logger.Error("Cannot retrieve AI summary: MinIO connection unavailable", map[string]interface{}{ + "tender_id": id, + "contract_folder_id": t.ContractFolderID, + "notice_id": noticeID, + "error": err.Error(), + }) + return nil, fmt.Errorf("failed to get AI summary: MinIO connection unavailable: %w", err) + } return nil, fmt.Errorf("failed to get AI summary: %w", err) } @@ -988,9 +1005,6 @@ func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AIS // It lists all tender.json files in the bucket and returns summaries for tenders // where the AI pipeline has completed processing. func (s *tenderService) GetAllAISummaries(ctx context.Context) ([]ai_summarizer.TenderSummaryItem, error) { - if s.aiSummarizerStorage == nil { - return nil, fmt.Errorf("AI summarizer storage is not configured") - } s.logger.Info("Retrieving all AI summaries from storage", map[string]interface{}{}) @@ -1007,9 +1021,14 @@ func (s *tenderService) GetAllAISummaries(ctx context.Context) ([]ai_summarizer. results, err := storage.GetAllSummaries(ctx) if err != nil { - s.logger.Error("Failed to get all AI summaries", map[string]interface{}{ + fields := map[string]interface{}{ "error": err.Error(), - }) + } + if errors.Is(err, ai_summarizer.ErrMinIOUnavailable) { + s.logger.Error("Failed to list AI summaries: MinIO connection unavailable", fields) + return nil, fmt.Errorf("failed to get all AI summaries: MinIO connection unavailable: %w", err) + } + s.logger.Error("Failed to get all AI summaries from MinIO", fields) return nil, fmt.Errorf("failed to get all AI summaries: %w", err) } diff --git a/pkg/ai_summarizer/errors.go b/pkg/ai_summarizer/errors.go index 0e48086..932b10f 100644 --- a/pkg/ai_summarizer/errors.go +++ b/pkg/ai_summarizer/errors.go @@ -27,6 +27,10 @@ var ( // ErrBucketNotFound is returned when the MinIO bucket does not exist. ErrBucketNotFound = errors.New("ai_summarizer: bucket not found") + // ErrMinIOUnavailable is returned when MinIO cannot be reached (network, + // timeout, connection refused, etc.) as opposed to a missing object key. + ErrMinIOUnavailable = errors.New("ai_summarizer: MinIO connection unavailable") + // ErrAPINonSuccess is returned when the AI API returns a non-2xx status code. ErrAPINonSuccess = errors.New("ai_summarizer: API returned non-success status") ) diff --git a/pkg/ai_summarizer/minio_errors.go b/pkg/ai_summarizer/minio_errors.go index 6dafef6..0ccd2ca 100644 --- a/pkg/ai_summarizer/minio_errors.go +++ b/pkg/ai_summarizer/minio_errors.go @@ -1,7 +1,10 @@ package ai_summarizer import ( + "context" "errors" + "fmt" + "net" "net/http" "strings" @@ -29,3 +32,55 @@ func isMinioNotFound(err error) bool { strings.Contains(msg, "nosuchkey") || strings.Contains(msg, "not found") } + +// isMinioConnectionError reports whether err indicates MinIO is unreachable. +func isMinioConnectionError(err error) bool { + if err == nil { + return false + } + if errors.Is(err, ErrMinIOUnavailable) { + return true + } + if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) { + return true + } + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return true + } + var opErr *net.OpError + if errors.As(err, &opErr) { + return true + } + msg := strings.ToLower(err.Error()) + connectionHints := []string{ + "connection refused", + "connection reset", + "no such host", + "dial tcp", + "dial udp", + "i/o timeout", + "network is unreachable", + "broken pipe", + "tls handshake", + "server gave http response to https client", + "eof", + } + for _, hint := range connectionHints { + if strings.Contains(msg, hint) { + return true + } + } + return false +} + +// wrapMinIOError attaches ErrMinIOUnavailable when the root cause is connectivity. +func wrapMinIOError(err error) error { + if err == nil || isMinioNotFound(err) { + return err + } + if isMinioConnectionError(err) { + return fmt.Errorf("%w: %v", ErrMinIOUnavailable, err) + } + return err +} diff --git a/pkg/ai_summarizer/storage.go b/pkg/ai_summarizer/storage.go index 5588bdb..1d3e7c7 100644 --- a/pkg/ai_summarizer/storage.go +++ b/pkg/ai_summarizer/storage.go @@ -184,10 +184,13 @@ func NewStorageClient(config *Config, log logger.Logger) (*StorageClient, error) func (s *StorageClient) verifyConnection(ctx context.Context) error { exists, err := s.client.BucketExists(ctx, s.config.MinioBucket) if err != nil { - return fmt.Errorf("bucket exists check for %q: %w", s.config.MinioBucket, err) + return s.logMinIOFailuref("bucket_exists", err, nil, "bucket exists check for %q", s.config.MinioBucket) } if !exists { - return fmt.Errorf("%w: %q", ErrBucketNotFound, s.config.MinioBucket) + err := fmt.Errorf("%w: %q", ErrBucketNotFound, s.config.MinioBucket) + return s.logMinIOFailure("bucket_exists", err, map[string]interface{}{ + "bucket": s.config.MinioBucket, + }) } buckets, err := s.client.ListBuckets(ctx) @@ -228,7 +231,11 @@ func (s *StorageClient) resolveNoticeStoragePrefix(ctx context.Context, contract return prefix, nil } if !isMinioNotFound(err) { - return "", fmt.Errorf("ai_summarizer: stat tender.json under %s: %w", prefix, err) + return "", s.logMinIOFailure("stat_tender_json", err, map[string]interface{}{ + "prefix": prefix, + "contract_folder_id": contractFolderID, + "notice_id": noticePublicationID, + }) } } @@ -241,7 +248,11 @@ func (s *StorageClient) resolveNoticeStoragePrefix(ctx context.Context, contract }) for object := range objectCh { if object.Err != nil { - return "", fmt.Errorf("ai_summarizer: search notice folder: %w", object.Err) + return "", s.logMinIOFailure("list_notice_folder", object.Err, map[string]interface{}{ + "notice_id": noticePublicationID, + "contract_folder_id": contractFolderID, + "notice_key": noticeKey, + }) } if idx := strings.Index(object.Key, marker); idx >= 0 { resolved = object.Key[:idx+len(marker)] @@ -290,6 +301,15 @@ func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolde prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID) if err != nil { + if !errors.Is(err, ErrObjectNotFound) { + s.logger.Warn("Failed to resolve MinIO path for tender summary", map[string]interface{}{ + "endpoint": s.config.MinioEndpoint, + "bucket": s.config.MinioBucket, + "contract_folder_id": contractFolderID, + "notice_id": noticePublicationID, + "error": err.Error(), + }) + } return "", err } return s.getSummaryFromObjectKey(ctx, prefix+"tender.json", contractFolderID, noticePublicationID) @@ -346,9 +366,13 @@ func (s *StorageClient) readTenderJSONAtKey(ctx context.Context, objectKey strin } var minioErr minio.ErrorResponse if errors.As(err, &minioErr) && minioErr.StatusCode == http.StatusForbidden { - return nil, fmt.Errorf("ai_summarizer: access denied to bucket %s: %w", s.config.MinioBucket, err) + return nil, s.logMinIOFailure("get_object", err, map[string]interface{}{ + "object_key": objectKey, + }) } - return nil, fmt.Errorf("ai_summarizer: failed to get object: %w", err) + return nil, s.logMinIOFailure("get_object", err, map[string]interface{}{ + "object_key": objectKey, + }) } defer object.Close() @@ -361,7 +385,9 @@ func (s *StorageClient) readTenderJSONAtKey(ctx context.Context, objectKey strin }) return nil, ErrObjectNotFound } - return nil, fmt.Errorf("ai_summarizer: failed to read tender.json: %w", err) + return nil, s.logMinIOFailure("read_tender_json", err, map[string]interface{}{ + "object_key": objectKey, + }) } var tenderJSON TenderJSON @@ -547,10 +573,16 @@ func (s *StorageClient) DownloadDocument(ctx context.Context, objectName string) // under the PROC_ prefix; if none match, it falls back to a full-bucket recursive // scan (slower on large buckets) so alternate key layouts still work. func (s *StorageClient) GetAllSummaries(ctx context.Context) ([]TenderSummaryItem, error) { + s.logger.Info("Listing all tender summaries from MinIO", map[string]interface{}{ + "endpoint": s.config.MinioEndpoint, + "bucket": s.config.MinioBucket, + }) + results := make([]TenderSummaryItem, 0) seen := make(map[string]struct{}) + var fatalErr error - appendFromPrefix := func(prefix string) { + appendFromPrefix := func(prefix string) bool { s.logger.Debug("Listing tender.json files from MinIO", map[string]interface{}{ "bucket": s.config.MinioBucket, "prefix": prefix, @@ -563,11 +595,10 @@ func (s *StorageClient) GetAllSummaries(ctx context.Context) ([]TenderSummaryIte for object := range objectCh { if object.Err != nil { - s.logger.Error("Error listing MinIO objects", map[string]interface{}{ + fatalErr = s.logMinIOFailure("list_tender_json", object.Err, map[string]interface{}{ "prefix": prefix, - "error": object.Err.Error(), }) - continue + return false } key := object.Key if !strings.HasSuffix(key, "/tender.json") { @@ -606,6 +637,9 @@ func (s *StorageClient) GetAllSummaries(ctx context.Context) ([]TenderSummaryIte OverallSummary: "", Done: true, }) + case errors.Is(err, ErrMinIOUnavailable): + fatalErr = err + return false default: s.logger.Debug("Skipping tender summary", map[string]interface{}{ "object_key": key, @@ -624,17 +658,28 @@ func (s *StorageClient) GetAllSummaries(ctx context.Context) ([]TenderSummaryIte Done: true, }) } + return true } - appendFromPrefix(procedurePrefix) + if !appendFromPrefix(procedurePrefix) { + return nil, fatalErr + } if len(seen) == 0 { s.logger.Info("No tender.json keys under PROC_; listing entire bucket for */tender.json (may be slower on large buckets)", map[string]interface{}{ "bucket": s.config.MinioBucket, }) - appendFromPrefix("") + if !appendFromPrefix("") { + return nil, fatalErr + } + } + + if fatalErr != nil { + return nil, fatalErr } s.logger.Info("Retrieved all summaries from storage", map[string]interface{}{ + "endpoint": s.config.MinioEndpoint, + "bucket": s.config.MinioBucket, "total_count": len(results), "completed_count": countDone(results), }) diff --git a/pkg/ai_summarizer/storage_logging.go b/pkg/ai_summarizer/storage_logging.go new file mode 100644 index 0000000..4593bef --- /dev/null +++ b/pkg/ai_summarizer/storage_logging.go @@ -0,0 +1,45 @@ +package ai_summarizer + +import ( + "errors" + "fmt" +) + +// logMinIOFailure records a structured log for a MinIO error and returns a wrapped +// error suitable for callers (connection issues include ErrMinIOUnavailable). +func (s *StorageClient) logMinIOFailure(operation string, err error, fields map[string]interface{}) error { + if err == nil { + return nil + } + if fields == nil { + fields = make(map[string]interface{}, 4) + } + fields["endpoint"] = s.config.MinioEndpoint + fields["bucket"] = s.config.MinioBucket + fields["operation"] = operation + fields["error"] = err.Error() + + switch { + case isMinioConnectionError(err): + fields["failure_type"] = "connection" + s.logger.Error("MinIO connection unavailable", fields) + return wrapMinIOError(err) + case errors.Is(err, ErrBucketNotFound): + fields["failure_type"] = "bucket_not_found" + s.logger.Error("MinIO bucket not found", fields) + return err + default: + fields["failure_type"] = "storage" + s.logger.Error("MinIO operation failed", fields) + return err + } +} + +// logMinIOFailuref is like logMinIOFailure with a formatted message prefix on the error. +func (s *StorageClient) logMinIOFailuref(operation string, err error, fields map[string]interface{}, format string, args ...interface{}) error { + if err == nil { + return nil + } + wrapped := fmt.Errorf(format+": %w", append(args, err)...) + return s.logMinIOFailure(operation, wrapped, fields) +}