Implemented debug artifacts for the distributor notification adapter

internal/adapters/distributor/client.go

@@ -3,6 +3,7 @@ package distributor
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net/http"
@@ -28,11 +29,26 @@ type UploadRequest struct {
 	IdempotencyKey string
 	SourcePath     string
 	BundlePath     string
+	CreatedAt      time.Time
 }
 
 type UploadResult struct {
-	RunID  string
-	Status string
+	RunID        string
+	Status       string
+	UploadStatus string
+	StatusError  string
+	RunStatus    *RunStatus
+}
+
+type RunStatus struct {
+	RunID      string
+	PipelineID string
+	Status     string
+	AcceptedAt time.Time
+	StartedAt  *time.Time
+	FinishedAt *time.Time
+	Report     json.RawMessage
+	Error      string
 }
 
 type IdempotencyConflictError struct {
@@ -57,6 +73,7 @@ type uploadClientFactory func(endpoint, token string, timeout time.Duration) (up
 
 type uploadClient interface {
 	UploadFiles(ctx context.Context, opts uploadFilesOptions) (uploadFilesResult, error)
+	Status(ctx context.Context, runID string) (runStatus, error)
 }
 
 type uploadFilesOptions struct {
@@ -64,6 +81,7 @@ type uploadFilesOptions struct {
 	IdempotencyKey string
 	SourcePath     string
 	BundlePath     string
+	CreatedAt      time.Time
 }
 
 type uploadFilesResult struct {
@@ -71,6 +89,19 @@ type uploadFilesResult struct {
 	Status string
 }
 
+type runStatus struct {
+	RunID      string
+	PipelineID string
+	Status     string
+	AcceptedAt time.Time
+	StartedAt  *time.Time
+	FinishedAt *time.Time
+	Report     json.RawMessage
+	Error      string
+}
+
+const statusPollInterval = 250 * time.Millisecond
+
 func New(cfg config.DistributorNotifyConfig) *Client {
 	return newClient(cfg, newDistributorUploadClient)
 }
@@ -138,6 +169,7 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e
 		IdempotencyKey: req.IdempotencyKey,
 		SourcePath:     req.SourcePath,
 		BundlePath:     req.BundlePath,
+		CreatedAt:      req.CreatedAt,
 	})
 	if err != nil {
 		return UploadResult{}, wrapUploadError(err, uploadErrorContext{
@@ -150,10 +182,65 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e
 		})
 	}
 
-	return UploadResult{
-		RunID:  result.RunID,
-		Status: result.Status,
-	}, nil
+	uploadResult := UploadResult{
+		RunID:        result.RunID,
+		Status:       result.Status,
+		UploadStatus: result.Status,
+	}
+	status, statusErr := waitForRunStatus(runCtx, uploadClient, result.RunID, c.Timeout > 0)
+	if status.RunID != "" || status.Status != "" {
+		uploadResult.RunStatus = &RunStatus{
+			RunID:      status.RunID,
+			PipelineID: status.PipelineID,
+			Status:     status.Status,
+			AcceptedAt: status.AcceptedAt,
+			StartedAt:  status.StartedAt,
+			FinishedAt: status.FinishedAt,
+			Report:     append(json.RawMessage(nil), status.Report...),
+			Error:      redactTokenString(status.Error, token),
+		}
+		if status.Status != "" {
+			uploadResult.Status = status.Status
+		}
+	}
+	if statusErr != nil {
+		uploadResult.StatusError = redactTokenString(statusErr.Error(), token)
+		return uploadResult, nil
+	}
+	if status.Status == "failed" {
+		return uploadResult, fmt.Errorf("distributor run %q failed: %s", status.RunID, uploadResult.RunStatus.Error)
+	}
+	return uploadResult, nil
+}
+
+func waitForRunStatus(ctx context.Context, client uploadClient, runID string, poll bool) (runStatus, error) {
+	status, err := client.Status(ctx, runID)
+	if err != nil || terminalRunStatus(status.Status) || !poll {
+		return status, err
+	}
+
+	for {
+		timer := time.NewTimer(statusPollInterval)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return status, fmt.Errorf("distributor run %q did not reach terminal status before timeout: %w", runID, ctx.Err())
+		case <-timer.C:
+		}
+
+		next, err := client.Status(ctx, runID)
+		if err != nil {
+			return status, err
+		}
+		status = next
+		if terminalRunStatus(status.Status) {
+			return status, nil
+		}
+	}
+}
+
+func terminalRunStatus(status string) bool {
+	return status == "succeeded" || status == "failed"
 }
 
 type distributorUploadClient struct {
@@ -179,6 +266,7 @@ func newDistributorUploadClient(endpoint, token string, timeout time.Duration) (
 func (c distributorUploadClient) UploadFiles(ctx context.Context, opts uploadFilesOptions) (uploadFilesResult, error) {
 	result, err := c.client.UploadFiles(ctx, distributorupload.UploadFilesOptions{
 		ID:             opts.BundleID,
+		Created:        opts.CreatedAt,
 		IdempotencyKey: opts.IdempotencyKey,
 		Files: []distributorbundle.BundleFile{
 			{SourcePath: opts.SourcePath, Path: opts.BundlePath},
@@ -193,6 +281,23 @@ func (c distributorUploadClient) UploadFiles(ctx context.Context, opts uploadFil
 	}, nil
 }
 
+func (c distributorUploadClient) Status(ctx context.Context, runID string) (runStatus, error) {
+	status, err := c.client.Status(ctx, runID)
+	if err != nil {
+		return runStatus{}, err
+	}
+	return runStatus{
+		RunID:      status.RunID,
+		PipelineID: status.PipelineID,
+		Status:     status.Status,
+		AcceptedAt: status.AcceptedAt,
+		StartedAt:  status.StartedAt,
+		FinishedAt: status.FinishedAt,
+		Report:     append(json.RawMessage(nil), status.Report...),
+		Error:      status.Error,
+	}, nil
+}
+
 type uploadErrorContext struct {
 	Endpoint       string
 	BundleID       string