Remove all versions of a file form the S3 bucket

upup/models/vfs.go

@@ -143,3 +143,7 @@ func (p *AssetPath) String() string {
 func (p *AssetPath) Remove() error {
 	return ReadOnlyError
 }
+
+func (p *AssetPath) RemoveAll() error {
+	return p.Remove()
+}

util/pkg/vfs/fs.go

@@ -187,6 +187,10 @@ func (p *FSPath) Remove() error {
 	return os.Remove(p.location)
 }
 
+func (p *FSPath) RemoveAll() error {
+	return p.Remove()
+}
+
 func (p *FSPath) PreferredHash() (*hashing.Hash, error) {
 	return p.Hash(hashing.HashAlgorithmSHA256)
 }

util/pkg/vfs/gsfs.go

@@ -132,6 +132,10 @@ func (p *GSPath) Remove() error {
 	}
 }
 
+func (p *GSPath) RemoveAll() error {
+	return p.Remove()
+}
+
 func (p *GSPath) Join(relativePath ...string) Path {
 	args := []string{p.key}
 	args = append(args, relativePath...)

util/pkg/vfs/k8sfs.go

@@ -68,6 +68,10 @@ func (p *KubernetesPath) Remove() error {
 	return fmt.Errorf("KubernetesPath::Remove not supported")
 }
 
+func (p *KubernetesPath) RemoveAll() error {
+	return p.Remove()
+}
+
 func (p *KubernetesPath) Join(relativePath ...string) Path {
 	args := []string{p.key}
 	args = append(args, relativePath...)

util/pkg/vfs/memfs.go

@@ -169,3 +169,7 @@ func (p *MemFSPath) Remove() error {
 	p.contents = nil
 	return nil
 }
+
+func (p *MemFSPath) RemoveAll() error {
+	return p.Remove()
+}

util/pkg/vfs/ossfs.go

@@ -217,6 +217,10 @@ func (p *OSSPath) Remove() error {
 	}
 }
 
+func (p *OSSPath) RemoveAll() error {
+	return p.Remove()
+}
+
 func (p *OSSPath) Base() string {
 	return path.Base(p.key)
 }

util/pkg/vfs/s3fs.go

@@ -106,6 +106,71 @@ func (p *S3Path) Remove() error {
 	return nil
 }
 
+func (p *S3Path) RemoveAll() error {
+	client, err := p.client()
+	if err != nil {
+		return err
+	}
+
+	klog.V(8).Infof("removing file %s", p)
+
+	request := &s3.ListObjectVersionsInput{
+		Bucket: aws.String(p.bucket),
+		Prefix: aws.String(p.key),
+	}
+
+	response, err := client.ListObjectVersions(request)
+	if err != nil {
+		return fmt.Errorf("error listing versions %s: %v", p, err)
+	}
+
+	if len(response.Versions) == 0 && len(response.DeleteMarkers) == 0 {
+		return os.ErrNotExist
+	}
+
+	// Sometimes S3 will return paginated results if there are too many versions and markers for an object.
+	// This happens at about entries 1000, so it is unlikely with current use cases.
+	if aws.BoolValue(response.IsTruncated) {
+		klog.Warningf("too many versions for %s", p)
+	}
+
+	objects := []*s3.ObjectIdentifier{}
+	for _, version := range response.Versions {
+		klog.V(8).Infof("removing file %s version %q", p, aws.StringValue(version.VersionId))
+		file := s3.ObjectIdentifier{
+			Key:       version.Key,
+			VersionId: version.VersionId,
+		}
+		objects = append(objects, &file)
+	}
+	for _, version := range response.DeleteMarkers {
+		klog.V(8).Infof("removing marker %s version %q", p, aws.StringValue(version.VersionId))
+		marker := s3.ObjectIdentifier{
+			Key:       version.Key,
+			VersionId: version.VersionId,
+		}
+		objects = append(objects, &marker)
+	}
+
+	if len(objects) > 0 {
+		klog.V(8).Infof("removing %d file/marker versions\n", len(objects))
+
+		request := &s3.DeleteObjectsInput{
+			Bucket: aws.String(p.bucket),
+			Delete: &s3.Delete{
+				Objects: objects,
+			},
+		}
+
+		_, err = client.DeleteObjects(request)
+		if err != nil {
+			return fmt.Errorf("error removing %d file/marker versions: %v", len(objects), err)
+		}
+	}
+
+	return nil
+}
+
 func (p *S3Path) Join(relativePath ...string) Path {
 	args := []string{p.key}
 	args = append(args, relativePath...)

util/pkg/vfs/sshfs.go

@@ -111,6 +111,10 @@ func (p *SSHPath) Remove() error {
 	return nil
 }
 
+func (p *SSHPath) RemoveAll() error {
+	return p.Remove()
+}
+
 func (p *SSHPath) Join(relativePath ...string) Path {
 	args := []string{p.path}
 	args = append(args, relativePath...)

util/pkg/vfs/swiftfs.go

@@ -291,6 +291,10 @@ func (p *SwiftPath) Remove() error {
 	}
 }
 
+func (p *SwiftPath) RemoveAll() error {
+	return p.Remove()
+}
+
 func (p *SwiftPath) Join(relativePath ...string) Path {
 	args := []string{p.key}
 	args = append(args, relativePath...)

util/pkg/vfs/vfs.go

@@ -59,6 +59,9 @@ type Path interface {
 	// Remove deletes the file
 	Remove() error
 
+	// RemoveAll completely deletes the file (with all its versions and markers)
+	RemoveAll() error
+
 	// Base returns the base name (last element)
 	Base() string