Add Prometheus metrics for artifact uploads

internal/artifact/BUILD.bazel

@@ -95,6 +95,7 @@ go_test(
         "azure_blob_test.go",
         "bk_uploader_test.go",
         "download_test.go",
+        "metrics_test.go",
         "downloader_test.go",
         "gs_uploader_test.go",
         "s3_downloader_test.go",
@@ -112,6 +113,7 @@ go_test(
         "@com_github_aws_aws_sdk_go_v2_service_s3//types",
         "@com_github_google_go_cmp//cmp",
         "@com_github_google_go_cmp//cmp/cmpopts",
+        "@com_github_prometheus_client_golang//prometheus/testutil",
         "@com_github_stretchr_testify//assert",
     ],
 )

internal/artifact/metrics.go

@@ -0,0 +1,29 @@
+package artifact
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+const metricsNamespace = "buildkite_agent"
+
+var (
+	artifactsUploaded = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: metricsNamespace,
+		Subsystem: "artifacts",
+		Name:      "uploaded_total",
+		Help:      "Count of artifacts uploaded",
+	})
+	artifactBytesUploaded = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: metricsNamespace,
+		Subsystem: "artifacts",
+		Name:      "bytes_uploaded_total",
+		Help:      "Count of artifact bytes uploaded",
+	})
+	artifactUploadFailures = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: metricsNamespace,
+		Subsystem: "artifacts",
+		Name:      "uploads_failed_total",
+		Help:      "Count of artifact uploads that failed",
+	})
+)

internal/artifact/metrics_test.go

@@ -0,0 +1,131 @@
+package artifact
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/buildkite/agent/v3/api"
+	"github.com/buildkite/agent/v3/logger"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+)
+
+type artifactMetricsTestAPIClient struct{}
+
+func (artifactMetricsTestAPIClient) CreateArtifacts(context.Context, string, *api.ArtifactBatch) (*api.ArtifactBatchCreateResponse, *api.Response, error) {
+	return nil, nil, nil
+}
+
+func (artifactMetricsTestAPIClient) SearchArtifacts(context.Context, string, *api.ArtifactSearchOptions) ([]*api.Artifact, *api.Response, error) {
+	return nil, nil, nil
+}
+
+func (artifactMetricsTestAPIClient) UpdateArtifacts(context.Context, string, []api.ArtifactState) (*api.Response, error) {
+	return nil, nil
+}
+
+type artifactMetricsTestWorkUnit struct {
+	artifact *api.Artifact
+}
+
+func (w artifactMetricsTestWorkUnit) Artifact() *api.Artifact { return w.artifact }
+func (artifactMetricsTestWorkUnit) Description() string       { return "test work unit" }
+func (artifactMetricsTestWorkUnit) DoWork(context.Context) (*api.ArtifactPartETag, error) {
+	return nil, nil
+}
+
+func TestArtifactMetricsIncrementOnSuccessfulUpload(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	beforeUploaded := testutil.ToFloat64(artifactsUploaded)
+	beforeBytes := testutil.ToFloat64(artifactBytesUploaded)
+	beforeFailed := testutil.ToFloat64(artifactUploadFailures)
+
+	artifact := &api.Artifact{ID: "a1", FileSize: 1234}
+	worker := &artifactUploadWorker{
+		Uploader: &Uploader{
+			logger:    logger.Discard,
+			apiClient: artifactMetricsTestAPIClient{},
+			conf:      UploaderConfig{JobID: "job-1"},
+		},
+		trackers: map[*api.Artifact]*artifactTracker{
+			artifact: {
+				pendingWork: 1,
+				ArtifactState: api.ArtifactState{
+					ID: artifact.ID,
+				},
+			},
+		},
+	}
+
+	resultsCh := make(chan workUnitResult, 1)
+	errCh := make(chan error, 1)
+	go worker.stateUpdater(ctx, resultsCh, errCh)
+
+	resultsCh <- workUnitResult{workUnit: artifactMetricsTestWorkUnit{artifact: artifact}}
+	close(resultsCh)
+
+	if err := <-errCh; err != nil {
+		t.Fatalf("stateUpdater() error = %v", err)
+	}
+
+	if got := testutil.ToFloat64(artifactsUploaded) - beforeUploaded; got != 1 {
+		t.Fatalf("artifactsUploaded delta = %v, want 1", got)
+	}
+	if got := testutil.ToFloat64(artifactBytesUploaded) - beforeBytes; got != 1234 {
+		t.Fatalf("artifactBytesUploaded delta = %v, want 1234", got)
+	}
+	if got := testutil.ToFloat64(artifactUploadFailures) - beforeFailed; got != 0 {
+		t.Fatalf("artifactUploadFailures delta = %v, want 0", got)
+	}
+}
+
+func TestArtifactMetricsIncrementOnFailedUpload(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	beforeUploaded := testutil.ToFloat64(artifactsUploaded)
+	beforeBytes := testutil.ToFloat64(artifactBytesUploaded)
+	beforeFailed := testutil.ToFloat64(artifactUploadFailures)
+
+	artifact := &api.Artifact{ID: "a2", FileSize: 4321}
+	worker := &artifactUploadWorker{
+		Uploader: &Uploader{
+			logger:    logger.Discard,
+			apiClient: artifactMetricsTestAPIClient{},
+			conf:      UploaderConfig{JobID: "job-2"},
+		},
+		trackers: map[*api.Artifact]*artifactTracker{
+			artifact: {
+				pendingWork: 2,
+				ArtifactState: api.ArtifactState{
+					ID: artifact.ID,
+				},
+			},
+		},
+	}
+
+	resultsCh := make(chan workUnitResult, 2)
+	errCh := make(chan error, 1)
+	go worker.stateUpdater(ctx, resultsCh, errCh)
+
+	u := artifactMetricsTestWorkUnit{artifact: artifact}
+	resultsCh <- workUnitResult{workUnit: u, err: errors.New("first failure")}
+	resultsCh <- workUnitResult{workUnit: u, err: errors.New("second failure")}
+	close(resultsCh)
+
+	if err := <-errCh; err == nil {
+		t.Fatal("stateUpdater() error = nil, want error")
+	}
+
+	if got := testutil.ToFloat64(artifactUploadFailures) - beforeFailed; got != 1 {
+		t.Fatalf("artifactUploadFailures delta = %v, want 1", got)
+	}
+	if got := testutil.ToFloat64(artifactsUploaded) - beforeUploaded; got != 0 {
+		t.Fatalf("artifactsUploaded delta = %v, want 0", got)
+	}
+	if got := testutil.ToFloat64(artifactBytesUploaded) - beforeBytes; got != 0 {
+		t.Fatalf("artifactBytesUploaded delta = %v, want 0", got)
+	}
+}

internal/artifact/uploader.go

@@ -680,6 +680,9 @@ selectLoop:
 			if result.err != nil {
 				// The work unit failed, so the whole artifact upload has failed.
 				errs = append(errs, result.err)
+				if tracker.State != "error" {
+					artifactUploadFailures.Inc()
+				}
 				tracker.State = "error"
 				a.logger.Debug("Artifact %s has entered state %s", tracker.ID, tracker.State)
 				continue
@@ -698,6 +701,8 @@ selectLoop:
 			// No pending units remain, so the whole artifact is complete.
 			// Add it to the next batch of states to upload.
 			tracker.State = "finished"
+			artifactsUploaded.Inc()
+			artifactBytesUploaded.Add(float64(artifact.FileSize))
 			a.logger.Debug("Artifact %s has entered state %s", tracker.ID, tracker.State)
 		}
 	}