test(e2e): stage AWS pull artifacts in S3

Makefile

@@ -249,6 +249,19 @@ gcs-publish-ci: gcloud version-dist-ci
 	echo "${GCS_URL}/${VERSION}" > ${UPLOAD}/${LATEST_FILE}
 	gcloud storage cp --cache-control="private, max-age=0, no-transform" ${UPLOAD}/${LATEST_FILE} ${GCS_LOCATION}
 
+# s3-publish-ci is the entry point for AWS CI testing.
+# Credentials reach the aws CLI through AWS_SHARED_CREDENTIALS_FILE or similar
+# pointers that are not unexported, but AWS_REGION must be restored explicitly.
+.PHONY: s3-publish-ci
+ifneq ($(AWS_REGION),)
+s3-publish-ci: export AWS_REGION := $(AWS_REGION)
+endif
+s3-publish-ci: version-dist-ci
+	@echo "== Uploading kops =="
+	${UPLOAD_CMD} ${UPLOAD}/kops/ ${UPLOAD_DEST}
+	echo "VERSION: ${VERSION}"
+	echo "$(patsubst %/,%,$(UPLOAD_DEST))/${VERSION}" > ${UPLOAD}/${LATEST_FILE}
+
 .PHONY: gen-cli-docs
 gen-cli-docs: kops # Regenerate CLI docs
 	KOPS_STATE_STORE= \

tests/e2e/go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/ec2 v1.307.0
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.97.3
 	github.com/aws/aws-sdk-go-v2/service/sts v1.41.5
+	github.com/aws/smithy-go v1.27.3
 	github.com/blang/semver/v4 v4.0.0
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510
 	github.com/spf13/pflag v1.0.10
@@ -93,7 +94,6 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/signin v1.0.4 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sso v1.30.7 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12 // indirect
-	github.com/aws/smithy-go v1.27.3 // indirect
 	github.com/awslabs/amazon-ecr-credential-helper/ecr-login v0.10.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver v3.5.1+incompatible // indirect

tests/e2e/kubetest2-kops/aws/s3.go

@@ -30,6 +30,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/aws/aws-sdk-go-v2/service/sts"
+	"github.com/aws/smithy-go"
 	"k8s.io/klog/v2"
 )
 
@@ -50,6 +51,7 @@ type BucketType string
 const (
 	BucketTypeStateStore     BucketType = "state"
 	BucketTypeDiscoveryStore BucketType = "discovery"
+	BucketTypeStagingStore   BucketType = "staging"
 )
 
 // NewAWSClient returns a new instance of awsClient configured to work in the default region (us-east-2).
@@ -68,17 +70,14 @@ func NewClient(ctx context.Context, region string) (*Client, error) {
 
 // BucketName constructs an unique bucket name using the AWS account ID in the default region (us-east-2).
 func (c Client) BucketName(ctx context.Context, bucketType BucketType) (string, error) {
-	// Construct the bucket name based on the ProwJob ID (if running in Prow) or AWS account ID (if running outside
-	// Prow) and a timestamp. When BUILD_ID is set we use it in place of time.Now() so that multiple kubetest2-kops
-	// invocations within the same CI job (e.g. upgrade tests) resolve to the same bucket name.
-	var suffix string
-	if jobID := os.Getenv("BUILD_ID"); jobID != "" {
-		if len(jobID) > 14 {
-			suffix = jobID[:14]
-		} else {
-			suffix = jobID
-		}
-	} else {
+	// Construct the bucket name based on the ProwJob ID (if running in Prow) or AWS account ID (if
+	// running outside Prow) and a timestamp. When BUILD_ID is set we use it in place of time.Now()
+	// so that multiple kubetest2-kops invocations within the same CI job (e.g. upgrade tests)
+	// resolve to the same bucket name. Use the full build ID because truncating its low-order
+	// digits can make jobs started in the same millisecond collide, allowing one job to delete
+	// another's bucket.
+	suffix := os.Getenv("BUILD_ID")
+	if suffix == "" {
 		callerIdentity, err := c.stsClient.GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{})
 		if err != nil {
 			return "", fmt.Errorf("building AWS STS presigned request: %w", err)
@@ -91,14 +90,12 @@ func (c Client) BucketName(ctx context.Context, bucketType BucketType) (string,
 	// Only allow lowercase letters, numbers, and hyphens
 	bucket = bucketNameRegex.ReplaceAllString(bucket, "")
 
-	if len(bucket) > 63 {
-		bucket = bucket[:63] // Max length is 63
-	}
-
+	// Names over the 63 character S3 limit fail bucket creation rather than being silently
+	// truncated, which could make distinct jobs collide on one bucket.
 	return bucket, nil
 }
 
-// EnsureS3Bucket creates a new S3 bucket with the given name and public read permissions.
+// EnsureS3Bucket creates an S3 bucket, optionally with public read access.
 func (c Client) EnsureS3Bucket(ctx context.Context, region, bucketName string, publicRead bool) error {
 	bucketName = strings.TrimPrefix(bucketName, "s3://")
 	klog.Infof("Creating bucket %s in region %s", bucketName, region)
@@ -112,14 +109,11 @@ func (c Client) EnsureS3Bucket(ctx context.Context, region, bucketName string, p
 	},
 	)
 	if err != nil {
-		var exists *types.BucketAlreadyExists
-		if errors.As(err, &exists) {
-			klog.Infof("Bucket %s already exists\n", bucketName)
-		} else {
-			klog.Infof("Error creating bucket %s, err: %v\n", bucketName, err)
+		var owned *types.BucketAlreadyOwnedByYou
+		if !errors.As(err, &owned) {
+			return fmt.Errorf("creating bucket %s: %w", bucketName, err)
 		}
-
-		return fmt.Errorf("creating bucket %s: %w", bucketName, err)
+		klog.Infof("Bucket %s already exists in this account", bucketName)
 	}
 
 	// Wait for the bucket to be created
@@ -134,26 +128,21 @@ func (c Client) EnsureS3Bucket(ctx context.Context, region, bucketName string, p
 		return fmt.Errorf("waiting for bucket %s to exist: %w", bucketName, err)
 	}
 
-	klog.Infof("Bucket %s created successfully", bucketName)
+	klog.Infof("Bucket %s is ready", bucketName)
 
 	if publicRead {
-		err = c.setPublicAccessBlock(ctx, bucketName)
-		if err != nil {
+		if err := c.setPublicAccessBlock(ctx, bucketName); err != nil {
 			klog.Errorf("Failed to disable public access block policies on bucket %s, err: %v", bucketName, err)
-
 			return fmt.Errorf("disabling public access block policies for bucket %s: %w", bucketName, err)
 		}
 
 		// Wait for public access block settings to propagate before setting the policy
 		time.Sleep(10 * time.Second)
 
-		err = c.setPublicReadPolicy(ctx, bucketName)
-		if err != nil {
+		if err := c.setPublicReadPolicy(ctx, bucketName); err != nil {
 			klog.Errorf("Failed to set public read policy on bucket %s, err: %v", bucketName, err)
-
 			return fmt.Errorf("setting public read policy for bucket %s: %w", bucketName, err)
 		}
-
 		klog.Infof("Public read policy set on bucket %s", bucketName)
 	}
 
@@ -162,28 +151,47 @@ func (c Client) EnsureS3Bucket(ctx context.Context, region, bucketName string, p
 
 // DeleteS3Bucket deletes a S3 bucket with the given name.
 func (c Client) DeleteS3Bucket(ctx context.Context, bucketName string) error {
+	return c.deleteS3Bucket(ctx, bucketName, false)
+}
+
+// DeleteS3BucketAndContents deletes all objects from an S3 bucket before deleting the bucket.
+func (c Client) DeleteS3BucketAndContents(ctx context.Context, bucketName string) error {
+	return c.deleteS3Bucket(ctx, bucketName, true)
+}
+
+func (c Client) deleteS3Bucket(ctx context.Context, bucketName string, empty bool) error {
 	bucketName = strings.TrimPrefix(bucketName, "s3://")
 
 	// Resolve the bucket's actual region to avoid 301 PermanentRedirect errors.
 	// During teardown the deployer may derive the S3 client region from random
 	// zones, which can differ from the region where the bucket was created.
 	bucketRegion, err := c.getBucketRegion(ctx, bucketName)
 	if err != nil {
+		if isNoSuchBucket(err) {
+			// Teardown may run without creating every bucket it knows about.
+			return nil
+		}
 		klog.Infof("Could not determine region for bucket %s: %v", bucketName, err)
 	}
 
-	regionOpt := func(o *s3.Options) {
-		if bucketRegion != "" {
-			o.Region = bucketRegion
+	client := c.s3Client
+	if bucketRegion != "" {
+		options := c.s3Client.Options()
+		options.Region = bucketRegion
+		client = s3.New(options)
+	}
+
+	if empty {
+		if err := emptyS3Bucket(ctx, client, bucketName); err != nil {
+			return err
 		}
 	}
 
-	_, err = c.s3Client.DeleteBucket(ctx, &s3.DeleteBucketInput{
+	_, err = client.DeleteBucket(ctx, &s3.DeleteBucketInput{
 		Bucket: aws.String(bucketName),
-	}, regionOpt)
+	})
 	if err != nil {
-		var noBucket *types.NoSuchBucket
-		if errors.As(err, &noBucket) {
+		if isNoSuchBucket(err) {
 			klog.Infof("Bucket %s does not exist.", bucketName)
 
 			return nil
@@ -193,7 +201,7 @@ func (c Client) DeleteS3Bucket(ctx context.Context, bucketName string) error {
 		return fmt.Errorf("deleting bucket %s: %w", bucketName, err)
 	}
 
-	err = s3.NewBucketNotExistsWaiter(c.s3Client).Wait(
+	err = s3.NewBucketNotExistsWaiter(client).Wait(
 		ctx, &s3.HeadBucketInput{
 			Bucket: aws.String(bucketName),
 		},
@@ -205,10 +213,53 @@ func (c Client) DeleteS3Bucket(ctx context.Context, bucketName string) error {
 	}
 
 	klog.Infof("Bucket %s deleted", bucketName)
-
 	return nil
 }
 
+func emptyS3Bucket(ctx context.Context, client *s3.Client, bucketName string) error {
+	for {
+		result, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
+			Bucket: aws.String(bucketName),
+		})
+		if err != nil {
+			if isNoSuchBucket(err) {
+				return nil
+			}
+			return fmt.Errorf("listing objects in bucket %s: %w", bucketName, err)
+		}
+
+		if len(result.Contents) == 0 {
+			return nil
+		}
+
+		objects := make([]types.ObjectIdentifier, len(result.Contents))
+		for i, object := range result.Contents {
+			objects[i] = types.ObjectIdentifier{Key: object.Key}
+		}
+		deleteResult, err := client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
+			Bucket: aws.String(bucketName),
+			Delete: &types.Delete{Objects: objects, Quiet: aws.Bool(true)},
+		})
+		if err != nil {
+			return fmt.Errorf("deleting objects from bucket %s: %w", bucketName, err)
+		}
+		if len(deleteResult.Errors) > 0 {
+			objectError := deleteResult.Errors[0]
+			return fmt.Errorf("deleting object %q from bucket %s: %s: %s",
+				aws.ToString(objectError.Key), bucketName, aws.ToString(objectError.Code), aws.ToString(objectError.Message))
+		}
+	}
+}
+
+func isNoSuchBucket(err error) bool {
+	var noBucket *types.NoSuchBucket
+	if errors.As(err, &noBucket) {
+		return true
+	}
+	var apiErr smithy.APIError
+	return errors.As(err, &apiErr) && (apiErr.ErrorCode() == "NoSuchBucket" || apiErr.ErrorCode() == "NotFound")
+}
+
 // getBucketRegion resolves the AWS region where the bucket resides.
 // GetBucketLocation is region-agnostic and can locate buckets in any region
 // from any endpoint. We pin to defaultRegion for consistency.

tests/e2e/kubetest2-kops/aws/s3_test.go

@@ -0,0 +1,92 @@
+/*
+Copyright 2026 The Kubernetes Authors.
+
+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 aws
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/aws/smithy-go"
+)
+
+func TestStagingBucketName(t *testing.T) {
+	t.Setenv("BUILD_ID", "12345678901234567890")
+	name, err := (Client{}).BucketName(context.Background(), BucketTypeStagingStore)
+	if err != nil {
+		t.Fatalf("building bucket name: %v", err)
+	}
+	if expected := "k8s-infra-kops-staging-12345678901234567890"; name != expected {
+		t.Errorf("expected %q, got %q", expected, name)
+	}
+	if len(name) > 63 {
+		t.Errorf("bucket name %q is longer than the 63 character limit", name)
+	}
+}
+
+// Teardown can run without --build, so a missing staging bucket must be a no-op.
+func TestDeleteMissingS3Bucket(t *testing.T) {
+	var requests []string
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests = append(requests, r.Method+" "+r.URL.RequestURI())
+		w.Header().Set("Content-Type", "application/xml")
+		w.WriteHeader(http.StatusNotFound)
+		fmt.Fprint(w, `<?xml version="1.0" encoding="UTF-8"?><Error><Code>NoSuchBucket</Code><Message>The specified bucket does not exist</Message></Error>`)
+	}))
+	defer server.Close()
+
+	c := Client{s3Client: s3.New(s3.Options{
+		Region:           defaultRegion,
+		BaseEndpoint:     aws.String(server.URL),
+		Credentials:      aws.AnonymousCredentials{},
+		UsePathStyle:     true,
+		RetryMaxAttempts: 1,
+	})}
+
+	if err := c.deleteS3Bucket(context.Background(), "does-not-exist", true); err != nil {
+		t.Errorf("deleting a bucket that does not exist: %v", err)
+	}
+	if len(requests) != 1 || !strings.Contains(requests[0], "location") {
+		t.Errorf("expected the bucket location request only, got %v", requests)
+	}
+}
+
+func TestIsNoSuchBucket(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{name: "typed", err: &types.NoSuchBucket{}, want: true},
+		{name: "generic code", err: &smithy.GenericAPIError{Code: "NoSuchBucket"}, want: true},
+		{name: "generic not found", err: &smithy.GenericAPIError{Code: "NotFound"}, want: true},
+		{name: "other", err: errors.New("other")},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := isNoSuchBucket(tc.err); got != tc.want {
+				t.Errorf("isNoSuchBucket() = %v, expected %v", got, tc.want)
+			}
+		})
+	}
+}