More cfgmgmt_actions diagnostics fixes

Signed-off-by: Daniel DeLeo <dan@chef.io>

Author: danielsdeleo

components/automate-cli/pkg/diagnostics/integration/cfgmgmt_actions.go

@@ -83,15 +83,16 @@ func CreateCfgmgmtActionsDiagnostic() diagnostics.Diagnostic {
 					return err
 				}
 
-				save.CreatedEntities = append(save.CreatedEntities, cfgmgmtActionsEntity{
+				e := cfgmgmtActionsEntity{
 					EntityName:       entityName,
 					RecordedAtMillis: day.UnixNano() / int64(time.Millisecond),
-				})
+				}
+				save.CreatedEntities = append(save.CreatedEntities, e)
 
 				tstCtx.SetValue("cfgmgmt-actions", save)
 
 				resp, err := tstCtx.DoLBRequest(
-					"/api/v0/events/data-collector",
+					"/api/v0/events/data-collector?z=cfgmgmt-actions",
 					lbrequest.WithMethod("POST"),
 					lbrequest.WithJSONBody(buf.String()),
 				)
@@ -123,7 +124,7 @@ func CreateCfgmgmtActionsDiagnostic() diagnostics.Diagnostic {
 			maxTries := 5
 
 			for _, entity := range loaded.CreatedEntities {
-				reqPath := fmt.Sprintf("/api/v0/eventfeed?collapse=true&page_size=100&start=%d&end=%d", entity.RecordedAtMillis-1000, entity.RecordedAtMillis+1000)
+				reqPath := fmt.Sprintf("/api/v0/eventfeed?collapse=true&page_size=100&start=%d&end=%d", entity.RecordedAtMillis-600000, entity.RecordedAtMillis+600000)
 				found := false
 
 				type eventsFeedResp struct {
@@ -163,7 +164,7 @@ func CreateCfgmgmtActionsDiagnostic() diagnostics.Diagnostic {
 					}
 					time.Sleep(5 * time.Duration(tries) * time.Second)
 				}
-				assert.True(tstCtx, found, "Could not find entity %s in GET %s; did get %+v", entity.EntityName, reqPath, respUnmarshalled)
+				assert.True(tstCtx, found, "[%s] Could not find entity %s in GET %s; did get %+v", time.Now().UTC().String(), entity.EntityName, reqPath, respUnmarshalled)
 			}
 
 		},

components/ingest-service/pipeline/message/chef_action.go

@@ -42,11 +42,14 @@ func (chefAction ChefAction) FinishProcessing(err error) {
 	if err == nil {
 		// Adding new metric; Time for a message to go through the pipeline
 		log.WithFields(log.Fields{
-			"message_id": chefAction.ID,
-			"message":    "ChefAction",
-			"metric":     "pipeline",
-			"type":       "ingest_time",
-			"ms":         chefAction.ClockProcessingTime(),
+			"message_id":  chefAction.ID,
+			"message":     "ChefAction",
+			"metric":      "pipeline",
+			"type":        "ingest_time",
+			"ms":          chefAction.ClockProcessingTime(),
+			"entity_name": chefAction.InternalChefAction.EntityName,
+			"entity_type": chefAction.InternalChefAction.EntityType,
+			"entity_task": chefAction.InternalChefAction.Task,
 		}).Info("Message ingested successfully")
 	}
 }

components/ingest-service/pipeline/publisher/chef_action.go

@@ -34,6 +34,13 @@ func ChefAction(in <-chan message.ChefAction, client backend.Client, out chan<-
 
 			err := client.InsertAction(msg.Ctx, msg.InternalChefAction)
 			if err != nil {
+				log.WithError(err).WithFields(log.Fields{
+					"publisher_id":    number,
+					"message_id":      msg.ID,
+					"buffer_size":     len(out),
+					"internal_action": msg.InternalChefAction,
+					"entity":          msg.InternalChefAction.EntityName,
+				}).Error("Failed to insert Chef Action")
 				msg.FinishProcessing(status.Errorf(codes.Internal, err.Error()))
 			} else {
 				out <- msg

components/ingest-service/server/chef.go

@@ -65,7 +65,7 @@ func (s *ChefIngestServer) ProcessChefRun(ctx context.Context, run *chef.Run) (*
 		err = <-errc
 
 		if err != nil {
-			log.WithError(err).Error("Message failure")
+			log.WithError(err).Error("Chef run ingestion failure")
 		}
 	} else if run.GetMessageType() == "run_start" {
 		log.WithFields(log.Fields{
@@ -95,7 +95,7 @@ func (s *ChefIngestServer) ProcessChefAction(ctx context.Context, action *chef.A
 		err := <-errc
 
 		if err != nil {
-			log.WithError(err).Error("Message failure")
+			log.WithError(err).Error("Chef Action ingestion failure")
 		}
 		return &response.ProcessChefActionResponse{}, err
 	}