store/serverservice: return errors with a bit more consistency

This ensures the worker can decide if the work needs to head back to the
queue or is an uncorrectable error.

Author: joelrebel

internal/model/model.go

@@ -1,6 +1,7 @@
 package model
 
 import (
+	"errors"
 	"net"
 
 	"github.com/bmc-toolbox/common"
@@ -20,6 +21,10 @@ type (
 	CollectorError string
 )
 
+var (
+	ErrInventoryQuery = errors.New("inventory query returned error")
+)
+
 const (
 	AppName                  = "alloy"
 	AppKindInband    AppKind = "inband"

internal/store/serverservice/errors.go

@@ -4,7 +4,6 @@ import "github.com/pkg/errors"
 
 var (
 	ErrSlugs                        = errors.New("slugs error")
-	ErrServerServiceQuery           = errors.New("error in server service API query")
 	ErrServerServiceRegisterChanges = errors.New("error in server service API register changes")
 	ErrAssetObject                  = errors.New("asset object error")
 	ErrAssetObjectConversion        = errors.New("error converting asset object")

internal/store/serverservice/serverservice.go

@@ -107,7 +107,7 @@ func (r *Store) AssetByID(ctx context.Context, id string, fetchBmcCredentials bo
 	if err != nil {
 		span.SetStatus(codes.Error, "Get() server failed")
 
-		return nil, errors.Wrap(ErrServerServiceQuery, "error querying server attributes: "+err.Error())
+		return nil, errors.Wrap(model.ErrInventoryQuery, "error querying server attributes: "+err.Error())
 	}
 
 	var credential *serverserviceapi.ServerCredential
@@ -120,7 +120,7 @@ func (r *Store) AssetByID(ctx context.Context, id string, fetchBmcCredentials bo
 		if err != nil {
 			span.SetStatus(codes.Error, "GetCredential() failed")
 
-			return nil, errors.Wrap(ErrServerServiceQuery, "error querying BMC credentials: "+err.Error())
+			return nil, errors.Wrap(model.ErrInventoryQuery, "error querying BMC credentials: "+err.Error())
 		}
 	}
 
@@ -157,7 +157,7 @@ func (r *Store) AssetsByOffsetLimit(ctx context.Context, offset, limit int) (ass
 	if err != nil {
 		span.SetStatus(codes.Error, "List() servers failed")
 
-		return nil, 0, errors.Wrap(ErrServerServiceQuery, err.Error())
+		return nil, 0, errors.Wrap(model.ErrInventoryQuery, err.Error())
 	}
 
 	assets = make([]*model.Asset, 0, len(servers))
@@ -168,7 +168,7 @@ func (r *Store) AssetsByOffsetLimit(ctx context.Context, offset, limit int) (ass
 		if err != nil {
 			span.SetStatus(codes.Error, "GetCredential() failed")
 
-			return nil, 0, errors.Wrap(ErrServerServiceQuery, err.Error())
+			return nil, 0, errors.Wrap(model.ErrInventoryQuery, err.Error())
 		}
 
 		asset, err := toAsset(server, credential, true)
@@ -213,7 +213,7 @@ func (r *Store) AssetUpdate(ctx context.Context, asset *model.Asset) error {
 		// count error
 		metrics.ServerServiceQueryErrorCount.With(stageLabel).Inc()
 
-		return errors.Wrap(ErrServerServiceQuery, err.Error())
+		return errors.Wrap(model.ErrInventoryQuery, err.Error())
 	}
 
 	if server == nil {
@@ -223,28 +223,58 @@ func (r *Store) AssetUpdate(ctx context.Context, asset *model.Asset) error {
 				"hr": hr,
 			}).Warn("server service server query returned nil object")
 
-		return errors.Wrap(ErrServerServiceQuery, "got nil Server object")
+		return errors.Wrap(model.ErrInventoryQuery, "got nil Server object")
 	}
 
 	// publish server inventory
-	if errPublish := r.publish(ctx, asset, server); errPublish != nil {
+	if errPublishInv := r.publishInventory(ctx, asset, server); errPublishInv != nil {
 		r.logger.WithFields(
 			logrus.Fields{
 				"id":  id,
 				"hr":  hr,
-				"err": errPublish.Error(),
-			}).Warn("inventory asset insert/update error")
+				"err": errPublishInv.Error(),
+			}).Warn("asset inventory insert/update error")
 
 		metricInventorized.With(prometheus.Labels{"status": "failed"}).Add(1)
+
+		return errors.Wrap(model.ErrInventoryQuery, errPublishInv.Error())
 	}
 
 	// count devices with no errors
 	metricInventorized.With(prometheus.Labels{"status": "success"}).Add(1)
 
+	if errPublishBiosCfg := r.publishBiosConfig(ctx, asset, server); errPublishBiosCfg != nil {
+		r.logger.WithFields(
+			logrus.Fields{
+				"id":  server.UUID.String(),
+				"err": errPublishBiosCfg,
+			}).Warn("asset bios configuration insert/update error")
+
+		metricBiosCfgCollected.With(prometheus.Labels{"status": "failed"}).Add(1)
+
+		return errors.Wrap(model.ErrInventoryQuery, errPublishBiosCfg.Error())
+	}
+
+	metricBiosCfgCollected.With(prometheus.Labels{"status": "success"}).Add(1)
+
 	return nil
 }
 
-func (r *Store) publish(ctx context.Context, asset *model.Asset, server *serverserviceapi.Server) error {
+func (r *Store) publishBiosConfig(ctx context.Context, asset *model.Asset, server *serverserviceapi.Server) error {
+	// Don't publish bios config if there's no bios config data made avai
+	if len(asset.BiosConfig) == 0 {
+		errNoBiosConfig := errors.New("no BIOS configuration collected")
+		return errNoBiosConfig
+	}
+
+	if asset.HasError(outofband.GetBiosConfigError) {
+		return errors.New(string(outofband.GetBiosConfigError))
+	}
+
+	return r.createUpdateServerBIOSConfiguration(ctx, server.UUID, asset.BiosConfig)
+}
+
+func (r *Store) publishInventory(ctx context.Context, asset *model.Asset, server *serverserviceapi.Server) error {
 	// create/update server bmc error attributes - for out of band data collection
 	if r.appKind == model.AppKindOutOfBand && len(asset.Errors) > 0 {
 		if err := r.createUpdateServerBMCErrorAttributes(
@@ -253,59 +283,32 @@ func (r *Store) publish(ctx context.Context, asset *model.Asset, server *servers
 			attributeByNamespace(serverBMCErrorsAttributeNS, server.Attributes),
 			asset,
 		); err != nil {
-			return errors.Wrap(ErrServerServiceQuery, "BMC error attribute create/update error: "+err.Error())
+			return errors.Wrap(model.ErrInventoryQuery, "BMC error attribute create/update error: "+err.Error())
 		}
 
 		// both inventory and BIOS configuration collection failed on a login failure
 		if asset.HasError(outofband.LoginError) {
 			return errors.New(string(outofband.LoginError))
 		}
-	}
 
-	// publish server inventory
-	if !asset.HasError(outofband.InventoryError) {
-		if errPublishInventory := r.publishInventory(ctx, server, asset); errPublishInventory != nil {
-			return errPublishInventory
+		if asset.HasError(outofband.InventoryError) {
+			return errors.New(string(outofband.InventoryError))
 		}
 	}
 
-	// Don't publish bios config if there's no bios config data
-	if len(asset.BiosConfig) == 0 {
-		errNoBiosConfig := errors.New("no BIOS configuration collected")
-		return errNoBiosConfig
-	}
-
-	if asset.HasError(outofband.GetBiosConfigError) {
-		return errors.New(string(outofband.GetBiosConfigError))
-	}
-
-	// publish bios configuration
-	err := r.createUpdateServerBIOSConfiguration(ctx, server.UUID, asset.BiosConfig)
-	if err != nil {
-		r.logger.WithFields(
-			logrus.Fields{
-				"id":  server.UUID.String(),
-				"err": err,
-			}).Warn("error in server bios configuration versioned attribute update")
-	}
-
-	return nil
-}
-
-func (r *Store) publishInventory(ctx context.Context, server *serverserviceapi.Server, asset *model.Asset) error {
 	// create/update server serial, vendor, model attributes
 	if err := r.createUpdateServerAttributes(ctx, server, asset); err != nil {
-		return errors.Wrap(ErrServerServiceQuery, "Server Vendor attribute create/update error: "+err.Error())
+		return errors.Wrap(model.ErrInventoryQuery, "Server Vendor attribute create/update error: "+err.Error())
 	}
 
 	// create update server metadata attributes
 	if err := r.createUpdateServerMetadataAttributes(ctx, server.UUID, asset); err != nil {
-		return errors.Wrap(ErrServerServiceQuery, "Server Metadata attribute create/update error: "+err.Error())
+		return errors.Wrap(model.ErrInventoryQuery, "Server Metadata attribute create/update error: "+err.Error())
 	}
 
 	// create update server component
 	if err := r.createUpdateServerComponents(ctx, server.UUID, asset); err != nil {
-		return errors.Wrap(ErrServerServiceQuery, "Server Component create/update error: "+err.Error())
+		return errors.Wrap(model.ErrInventoryQuery, "Server Component create/update error: "+err.Error())
 	}
 
 	return nil
@@ -345,7 +348,7 @@ func (r *Store) createUpdateServerComponents(ctx context.Context, serverID uuid.
 		// set span status
 		span.SetStatus(codes.Error, "GetComponents() failed")
 
-		return errors.Wrap(ErrServerServiceQuery, err.Error())
+		return errors.Wrap(model.ErrInventoryQuery, err.Error())
 	}
 
 	// For debugging and to capture test fixtures data.
@@ -372,7 +375,7 @@ func (r *Store) createUpdateServerComponents(ctx context.Context, serverID uuid.
 	// identify changes to be applied
 	add, update, remove, err := serverServiceChangeList(ctx, currentInventoryPtrSlice, newInventory)
 	if err != nil {
-		return errors.Wrap(ErrServerServiceQuery, err.Error())
+		return errors.Wrap(model.ErrInventoryQuery, err.Error())
 	}
 
 	if len(add) == 0 && len(update) == 0 && len(remove) == 0 {
@@ -409,7 +412,7 @@ func (r *Store) createUpdateServerComponents(ctx context.Context, serverID uuid.
 			// set span status
 			span.SetStatus(codes.Error, "CreateComponents() failed")
 
-			return errors.Wrap(ErrServerServiceQuery, "CreateComponents: "+err.Error())
+			return errors.Wrap(model.ErrInventoryQuery, "CreateComponents: "+err.Error())
 		}
 	}
 
@@ -433,7 +436,7 @@ func (r *Store) createUpdateServerComponents(ctx context.Context, serverID uuid.
 			// set span status
 			span.SetStatus(codes.Error, "UpdateComponents() failed")
 
-			return errors.Wrap(ErrServerServiceQuery, "UpdateComponents: "+err.Error())
+			return errors.Wrap(model.ErrInventoryQuery, "UpdateComponents: "+err.Error())
 		}
 	}
 

internal/worker/worker.go

@@ -49,8 +49,7 @@ var (
 	errTaskFirmwareParam    = errors.New("error in task firmware parameters")
 	errInitTask             = errors.New("error initializing new task from event")
 
-	errAssetNotFound  = errors.New("asset not found in inventory")
-	errInventoryQuery = errors.New("inventory query returned error")
+	errAssetNotFound = errors.New("asset not found in inventory")
 
 	errCollector = errors.New("collector error")
 )
@@ -313,67 +312,91 @@ func (w *Worker) doWork(ctx context.Context, condition *rctypes.Condition, e eve
 	// check no error
 	err = w.runTaskWithMonitor(ctx, task, e)
 	switch err {
-	case errInventoryQuery:
-		// inventory lookup failure - non 404 errors
-		task.SetState(rctypes.Failed)
-		task.Status = err.Error()
+	case nil:
+		// work completed successfully
+		task.SetState(rctypes.Succeeded)
+		task.Status = "collection completed successfully"
 
-		metrics.RegisterEventCounter(true, "nack")
-		w.eventNak(e) // have the message bus re-deliver the message
+		publisher.Publish(ctx, task)
+
+		w.eventAckComplete(e)
 
+		metrics.RegisterConditionMetrics(startTS, string(rctypes.Succeeded))
+		metrics.RegisterEventCounter(true, "ack")
 		metrics.RegisterSpanEvent(
 			span,
 			condition,
 			w.id.String(),
 			task.Parameters.AssetID.String(),
-			"sent nack: store query error",
-			err,
+			"sent ack: condition finalized",
+			nil,
 		)
 
-	case errAssetNotFound, errCollector:
-		// asset was not found
+		publisher.Publish(ctx, task)
+
+		w.logger.WithFields(logrus.Fields{
+			"serverID":    task.Parameters.AssetID.String(),
+			"conditionID": task.ID,
+			"elapsed":     time.Since(startTS).String(),
+			"state":       task.state,
+			"status":      task.Status,
+		}).Info("task for device completed")
+
+	case model.ErrInventoryQuery:
+		// inventory lookup failure - non 404 errors
 		task.SetState(rctypes.Failed)
 		task.Status = err.Error()
 
-		w.eventAckComplete(e)
+		w.eventNak(e) // have the message bus re-deliver the message
+
+		publisher.Publish(ctx, task)
 
+		metrics.RegisterEventCounter(true, "nack")
+		metrics.RegisterConditionMetrics(startTS, string(rctypes.Failed))
 		metrics.RegisterSpanEvent(
 			span,
 			condition,
 			w.id.String(),
 			task.Parameters.AssetID.String(),
-			"sent ack: error"+err.Error(),
+			"sent nack: store query error",
 			err,
 		)
 
-	case nil:
-		// work completed successfully
-		task.SetState(rctypes.Succeeded)
-		task.Status = "collection completed successfully"
+		w.logger.WithFields(logrus.Fields{
+			"serverID":    task.Parameters.AssetID.String(),
+			"conditionID": task.ID,
+			"elapsed":     time.Since(startTS).String(),
+			"state":       task.state,
+			"status":      task.Status,
+		}).Info("task for device failed and will be retried")
 
-		publisher.Publish(ctx, task)
+	default:
+		// all other error cases
+		task.SetState(rctypes.Failed)
+		task.Status = err.Error()
 
 		w.eventAckComplete(e)
 
-		metrics.RegisterConditionMetrics(startTS, string(rctypes.Succeeded))
+		publisher.Publish(ctx, task)
+
+		metrics.RegisterConditionMetrics(startTS, string(rctypes.Failed))
 		metrics.RegisterEventCounter(true, "ack")
 		metrics.RegisterSpanEvent(
 			span,
 			condition,
 			w.id.String(),
 			task.Parameters.AssetID.String(),
-			"sent ack: condition finalized",
-			nil,
+			"sent ack: error"+err.Error(),
+			err,
 		)
 
-		publisher.Publish(ctx, task)
 		w.logger.WithFields(logrus.Fields{
 			"serverID":    task.Parameters.AssetID.String(),
 			"conditionID": task.ID,
 			"elapsed":     time.Since(startTS).String(),
 			"state":       task.state,
 			"status":      task.Status,
-		}).Info("task for device completed")
+		}).Info("task for device failed")
 	}
 }
 
@@ -446,7 +469,7 @@ func (w *Worker) inventoryOutofband(ctx context.Context, task *Task, doneCh chan
 			return errors.Wrap(errAssetNotFound, err.Error())
 		}
 
-		return errors.Wrap(errInventoryQuery, err.Error())
+		return errors.Wrap(model.ErrInventoryQuery, err.Error())
 	}
 
 	c, err := collector.NewDeviceCollectorWithStore(w.repository, w.cfg.AppKind, w.logger)