Fixed review issues

client_cluster.go

@@ -1,8 +1,6 @@
 package ovirtclient
 
 import (
-	"fmt"
-
 	ovirtsdk4 "github.com/ovirt/go-ovirt"
 )
 
@@ -25,12 +23,12 @@ type Cluster interface {
 func convertSDKCluster(sdkCluster *ovirtsdk4.Cluster) (Cluster, error) {
 	id, ok := sdkCluster.Id()
 	if !ok {
-		return nil, fmt.Errorf("failed to fetch ID for cluster")
+		return nil, newError(EFieldMissing, "failed to fetch ID for cluster")
 	}
 
 	name, ok := sdkCluster.Name()
 	if !ok {
-		return nil, fmt.Errorf("failed to fetch name for cluster %s", id)
+		return nil, newError(EFieldMissing, "failed to fetch name for cluster %s", id)
 	}
 	return &cluster{
 		id:   id,

client_cluster_list.go

@@ -11,7 +11,7 @@ func (o *oVirtClient) ListClusters() ([]Cluster, error) {
 	}
 	sdkClusters, ok := clustersResponse.Clusters()
 	if !ok {
-		return nil, nil
+		return []Cluster{}, nil
 	}
 	clusters := make([]Cluster, len(sdkClusters.Slice()))
 	for i, sdkCluster := range sdkClusters.Slice() {

client_disk.go

@@ -2,7 +2,6 @@ package ovirtclient
 
 import (
 	"context"
-	"fmt"
 	"io"
 
 	ovirtsdk4 "github.com/ovirt/go-ovirt"
@@ -60,7 +59,7 @@ type DiskClient interface {
 	// GetDisk fetches a disk with a specific ID from the oVirt Engine.
 	GetDisk(diskID string) (Disk, error)
 	// RemoveDisk removes a disk with a specific ID.
-	RemoveDisk(diskID string) error
+	RemoveDisk(ctx context.Context, diskID string) error
 }
 
 // UploadImageResult represents the completed image upload.
@@ -113,7 +112,7 @@ const (
 func convertSDKDisk(sdkDisk *ovirtsdk4.Disk) (Disk, error) {
 	id, ok := sdkDisk.Id()
 	if !ok {
-		return nil, fmt.Errorf("disk does not contain an ID")
+		return nil, newError(EFieldMissing, "disk does not contain an ID")
 	}
 	var storageDomainID string
 	if sdkStorageDomain, ok := sdkDisk.StorageDomain(); ok {
@@ -127,19 +126,19 @@ func convertSDKDisk(sdkDisk *ovirtsdk4.Disk) (Disk, error) {
 		}
 	}
 	if storageDomainID == "" {
-		return nil, fmt.Errorf("failed to find a valid storage domain ID for disk %s", id)
+		return nil, newError(EFieldMissing, "failed to find a valid storage domain ID for disk %s", id)
 	}
 	alias, ok := sdkDisk.Alias()
 	if !ok {
-		return nil, fmt.Errorf("disk %s does not contain an alias", id)
+		return nil, newError(EFieldMissing, "disk %s does not contain an alias", id)
 	}
 	provisionedSize, ok := sdkDisk.ProvisionedSize()
 	if !ok {
-		return nil, fmt.Errorf("disk %s does not contain a provisioned size", id)
+		return nil, newError(EFieldMissing, "disk %s does not contain a provisioned size", id)
 	}
 	format, ok := sdkDisk.Format()
 	if !ok {
-		return nil, fmt.Errorf("disk %s has no format field", id)
+		return nil, newError(EFieldMissing, "disk %s has no format field", id)
 	}
 	return &disk{
 		id:              id,

client_disk_list.go

@@ -11,7 +11,7 @@ func (o *oVirtClient) ListDisks() ([]Disk, error) {
 	}
 	sdkDisks, ok := response.Disks()
 	if !ok {
-		return nil, nil
+		return []Disk{}, nil
 	}
 	result := make([]Disk, len(sdkDisks.Slice()))
 	for i, sdkDisk := range sdkDisks.Slice() {

client_disk_remove.go

@@ -1,13 +1,34 @@
 package ovirtclient
 
-func (o *oVirtClient) RemoveDisk(diskID string) error {
-	if _, err := o.conn.SystemService().DisksService().DiskService(diskID).Remove().Send(); err != nil {
-		return wrap(
+import (
+	"context"
+	"time"
+)
+
+func (o *oVirtClient) RemoveDisk(ctx context.Context, diskID string) error {
+	var lastError EngineError
+	for {
+		_, err := o.conn.SystemService().DisksService().DiskService(diskID).Remove().Send()
+		if err == nil {
+			return err
+		}
+		lastError = wrap(
 			err,
 			EUnidentified,
 			"failed to remove disk %s",
 			diskID,
 		)
+		if !lastError.CanAutoRetry() {
+			return lastError
+		}
+		select {
+		case <-ctx.Done():
+			return wrap(
+				lastError,
+				ETimeout,
+				"timeout while removing disk",
+			)
+		case <-time.After(10 * time.Second):
+		}
 	}
-	return nil
 }

client_disk_uploadimage.go

@@ -76,7 +76,12 @@ func (o *oVirtClient) createDiskForUpload(
 ) (*ovirtsdk4.Disk, error) {
 	storageDomain, err := ovirtsdk4.NewStorageDomainBuilder().Id(storageDomainID).Build()
 	if err != nil {
-		panic(fmt.Errorf("bug: failed to build storage domain object from storage domain ID: %s", storageDomainID))
+		return nil, wrap(
+			err,
+			EBug,
+			"failed to build storage domain object from storage domain ID: %s",
+			storageDomainID,
+		)
 	}
 	diskBuilder := ovirtsdk4.NewDiskBuilder().
 		Alias(alias).
@@ -112,6 +117,7 @@ func (o *oVirtClient) createProgress(
 	disk *ovirtsdk4.Disk,
 ) (UploadImageProgress, error) {
 	progress := &uploadImageProgress{
+		cli:             o,
 		correlationID:   fmt.Sprintf("image_transfer_%s", alias),
 		uploadedBytes:   0,
 		cowSize:         qcowSize,
@@ -135,6 +141,7 @@ func (o *oVirtClient) createProgress(
 }
 
 type uploadImageProgress struct {
+	cli             *oVirtClient
 	uploadedBytes   uint64
 	cowSize         uint64
 	size            uint64
@@ -150,7 +157,7 @@ type uploadImageProgress struct {
 	done chan struct{}
 	// lock is a lock that prevents race conditions during the upload process.
 	lock *sync.Mutex
-	// cancel is the cancel function for the context. Is is called to ensure that the context is properly canceled.
+	// cancel is the cancel function for the context. HasCode is called to ensure that the context is properly canceled.
 	cancel context.CancelFunc
 	// err holds the error that happened during the upload. It can be queried using the Err() method.
 	err error
@@ -276,7 +283,7 @@ func (u *uploadImageProgress) removeDisk() {
 	disk := u.disk
 	if disk != nil {
 		if id, ok := u.disk.Id(); ok {
-			_ = u.client.RemoveDisk(id)
+			_ = u.client.RemoveDisk(u.ctx, id)
 		}
 	}
 }
@@ -384,6 +391,7 @@ func (u *uploadImageProgress) setupImageTransfer(diskID string) (
 	*ovirtsdk4.ImageTransferService,
 	error,
 ) {
+	var lastError EngineError
 	imageTransfersService := u.conn.SystemService().ImageTransfersService()
 	image := ovirtsdk4.NewImageBuilder().Id(diskID).MustBuild()
 	transfer := ovirtsdk4.
@@ -402,8 +410,8 @@ func (u *uploadImageProgress) setupImageTransfer(diskID string) (
 	transferService := imageTransfersService.ImageTransferService(transfer.MustId())
 
 	for {
-		req, lastError := transferService.Get().Send()
-		if lastError == nil {
+		req, err := transferService.Get().Send()
+		if err == nil {
 			if req.MustImageTransfer().MustPhase() == ovirtsdk4.IMAGETRANSFERPHASE_TRANSFERRING {
 				break
 			} else {
@@ -413,6 +421,11 @@ func (u *uploadImageProgress) setupImageTransfer(diskID string) (
 					req.MustImageTransfer().MustPhase(),
 				)
 			}
+		} else {
+			lastError = wrap(err, EUnidentified, "failed to get image transfer for disk %s", diskID)
+			if !lastError.CanAutoRetry() {
+				return nil, nil, lastError
+			}
 		}
 		select {
 		case <-time.After(time.Second * 5):
@@ -433,7 +446,7 @@ func (u *uploadImageProgress) waitForDiskOk(diskService *ovirtsdk4.DiskService)
 				return newError(EUnsupported, "the disk was removed after upload, probably not supported")
 			}
 			if disk.MustStatus() == ovirtsdk4.DISKSTATUS_OK {
-				return nil
+				return u.cli.waitForJobFinished(u.ctx, u.correlationID)
 			} else {
 				lastError = newError(EPending, "disk status is %s, not ok", disk.MustStatus())
 			}

client_disk_uploadimage_test.go

@@ -46,7 +46,7 @@ func TestImageUploadDiskCreated(t *testing.T) {
 	if err != nil {
 		t.Fatal(fmt.Errorf("failed to fetch disk after image upload (%w)", err))
 	}
-	if err := client.RemoveDisk(disk.ID()); err != nil {
-		t.Fatal(fmt.Errorf("failed to remove disk (%w)", err))
+	if err := client.RemoveDisk(context.Background(), disk.ID()); err != nil {
+		t.Fatal(err)
 	}
 }

client_host_list.go

@@ -7,7 +7,7 @@ func (o *oVirtClient) ListHosts() ([]Host, error) {
 	}
 	sdkHosts, ok := response.Hosts()
 	if !ok {
-		return nil, nil
+		return []Host{}, nil
 	}
 	result := make([]Host, len(sdkHosts.Slice()))
 	for i, sdkHost := range sdkHosts.Slice() {

client_storagedomain_list.go

@@ -7,7 +7,7 @@ func (o *oVirtClient) ListStorageDomains() (storageDomains []StorageDomain, err
 	}
 	sdkStorageDomains, ok := response.StorageDomains()
 	if !ok {
-		return nil, nil
+		return []StorageDomain{}, nil
 	}
 	storageDomains = make([]StorageDomain, len(sdkStorageDomains.Slice()))
 	for i, sdkStorageDomain := range sdkStorageDomains.Slice() {

client_template.go

@@ -1,8 +1,6 @@
 package ovirtclient
 
 import (
-	"fmt"
-
 	ovirtsdk4 "github.com/ovirt/go-ovirt"
 )
 
@@ -28,7 +26,7 @@ func convertSDKTemplate(sdkTemplate *ovirtsdk4.Template) (Template, error) {
 	}
 	description, ok := sdkTemplate.Description()
 	if !ok {
-		return nil, fmt.Errorf("template does not contain a description")
+		return nil, newError(EFieldMissing, "template does not contain a description")
 	}
 	return &template{
 		id:          id,

client_template_get.go

@@ -11,7 +11,7 @@ func (o *oVirtClient) GetTemplate(id string) (Template, error) {
 	}
 	template, err := convertSDKTemplate(sdkTemplate)
 	if err != nil {
-		return nil, wrap(err, EUnidentified, "failed to convert template object")
+		return nil, wrap(err, EBug, "failed to convert template object")
 	}
 	return template, nil
 }

client_template_list.go

@@ -7,7 +7,7 @@ func (o *oVirtClient) ListTemplates() ([]Template, error) {
 	}
 	sdkTemplates, ok := response.Templates()
 	if !ok {
-		return nil, newError(ENotFound, "host list response didn't contain hosts")
+		return []Template{}, nil
 	}
 	result := make([]Template, len(sdkTemplates.Slice()))
 	for i, sdkTemplate := range sdkTemplates.Slice() {

client_util.go

@@ -0,0 +1,58 @@
+package ovirtclient
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	ovirtsdk "github.com/ovirt/go-ovirt"
+)
+
+// waitForJobFinished waits for a job to truly finish. This is especially important when disks are involved as their
+// status changes to OK prematurely.
+//
+// correlationID is a query parameter assigned to a job before it is sent to the ovirt engine, it must be unique and
+// under 30 chars. To set a correlationID add `Query("correlation_id", correlationID)` to the engine API call, for
+// example:
+//
+//     correlationID := fmt.Sprintf("image_transfer_%s", utilrand.String(5))
+//     conn.
+//         SystemService().
+//         DisksService().
+//         DiskService(diskId).
+//         Update().
+//         Query("correlation_id", correlationID).
+//         Send()
+func (o *oVirtClient) waitForJobFinished(ctx context.Context, correlationID string) error {
+	var lastError EngineError
+	for {
+		jobResp, err := o.conn.SystemService().JobsService().List().Search(fmt.Sprintf("correlation_id=%s", correlationID)).Send()
+		if err == nil {
+			if jobSlice, ok := jobResp.Jobs(); ok {
+				if len(jobSlice.Slice()) == 0 {
+					return nil
+				}
+				for _, job := range jobSlice.Slice() {
+					if status, _ := job.Status(); status != ovirtsdk.JOBSTATUS_STARTED {
+						return nil
+					}
+				}
+			}
+			lastError = newError(EPending, "job for correlation ID %s still pending", correlationID)
+		} else {
+			realErr := wrap(err, EUnidentified, "failed to list jobs for correlation ID %s", correlationID)
+			if !realErr.CanAutoRetry() {
+				return realErr
+			}
+			lastError = realErr
+		}
+		select {
+		case <-time.After(5 * time.Second):
+		case <-ctx.Done():
+			return wrap(
+				lastError,
+				ETimeout,
+				"timeout while waiting for job with correlation_id %s to finish", correlationID)
+		}
+	}
+}

client_vm.go

@@ -2,7 +2,6 @@ package ovirtclient
 
 import (
 	"context"
-	"fmt"
 )
 
 type VMClient interface {
@@ -19,13 +18,13 @@ type VMClient interface {
 // is 0. If the parameters are guaranteed to be non-zero MustNewVMCPUTopo should be used.
 func NewVMCPUTopo(cores uint, threads uint, sockets uint) (VMCPUTopo, error) {
 	if cores == 0 {
-		return nil, fmt.Errorf("BUG: cores cannot be zero")
+		return nil, newError(EBadArgument, "cores cannot be zero")
 	}
 	if threads == 0 {
-		return nil, fmt.Errorf("BUG: threads cannot be zero")
+		return nil, newError(EBadArgument, "threads cannot be zero")
 	}
 	if sockets == 0 {
-		return nil, fmt.Errorf("BUG: sockets cannot be zero")
+		return nil, newError(EBadArgument, "sockets cannot be zero")
 	}
 	return &vmCPUTopo{
 		cores:   cores,