Merge pull request #1764 from porter-dev/belanger/fix-provisioning-no-logs

Add a fix for sending logs and state files to log storage

provisioner/integrations/redis_stream/global.go

@@ -160,41 +160,50 @@ func GlobalStreamListener(
 				continue
 			}
 
+			config.Logger.Debug().Msg(fmt.Sprintf("pushing state and log file for %s with status %v", workspaceID, statusVal))
+
 			switch fmt.Sprintf("%v", statusVal) {
-			case "created":
-				err := handleOperationCreated(config, client, infra, operation, workspaceID)
+			case "created", "error", "destroyed":
+				err := cleanupOperation(config, client, infra, operation, workspaceID)
 
 				if err != nil {
 					config.Alerter.SendAlert(context.Background(), err, map[string]interface{}{
 						"workspace_id": workspaceID,
 					})
 				}
-			case "error":
-			case "destroyed":
 			}
 		}
 	}
 }
 
-func handleOperationCreated(config *config.Config, client *redis.Client, infra *models.Infra, operation *models.Operation, workspaceID string) error {
+func cleanupOperation(config *config.Config, client *redis.Client, infra *models.Infra, operation *models.Operation, workspaceID string) error {
+	l := config.Logger
+	l.Debug().Msg(fmt.Sprintf("pushing state for %s", workspaceID))
+
 	err := pushNewStateToStorage(config, client, infra, operation, workspaceID)
 
 	if err != nil {
 		return err
 	}
 
+	l.Debug().Msg(fmt.Sprintf("cleaning state stream for %s", workspaceID))
+
 	err = cleanupStateStream(config, client, workspaceID)
 
 	if err != nil {
 		return nil
 	}
 
+	l.Debug().Msg(fmt.Sprintf("pushing logs for %s", workspaceID))
+
 	err = pushLogsToStorage(config, client, infra, workspaceID)
 
 	if err != nil {
 		return err
 	}
 
+	l.Debug().Msg(fmt.Sprintf("cleaning logs for %s", workspaceID))
+
 	err = cleanupLogStream(config, client, infra, workspaceID)
 
 	if err != nil {

provisioner/server/grpc/store_log.go

@@ -33,13 +33,6 @@ func (s *ProvisionerServer) StoreLog(stream pb.Provisioner_StoreLogServer) error
 		tfLog, err := stream.Recv()
 
 		if err == io.EOF {
-			// push to the global stream
-			err := redis_stream.PushToGlobalStream(s.config.RedisClient, infra, operation, "created")
-
-			if err != nil {
-				return err
-			}
-
 			return stream.SendAndClose(&pb.TerraformStateMeta{})
 		} else if err != nil {
 			return err

provisioner/server/handlers/state/create_resource.go

@@ -63,6 +63,14 @@ func (c *CreateResourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request
 		return
 	}
 
+	// push to the global stream
+	err = redis_stream.PushToGlobalStream(c.Config.RedisClient, infra, operation, "created")
+
+	if err != nil {
+		apierrors.HandleAPIError(c.Config.Logger, c.Config.Alerter, w, r, apierrors.NewErrInternal(err), true)
+		return
+	}
+
 	// update the infra to indicate completion
 	infra.Status = "created"
 

provisioner/server/handlers/state/delete_resource.go

@@ -48,6 +48,14 @@ func (c *DeleteResourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request
 		return
 	}
 
+	// push to the global stream
+	err = redis_stream.PushToGlobalStream(c.Config.RedisClient, infra, operation, "destroyed")
+
+	if err != nil {
+		apierrors.HandleAPIError(c.Config.Logger, c.Config.Alerter, w, r, apierrors.NewErrInternal(err), true)
+		return
+	}
+
 	// update the infra to indicate deletion
 	infra.Status = "deleted"
 

provisioner/server/handlers/state/report_error.go

@@ -68,6 +68,14 @@ func (c *ReportErrorHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// push to the global stream
+	err = redis_stream.PushToGlobalStream(c.Config.RedisClient, infra, operation, "error")
+
+	if err != nil {
+		apierrors.HandleAPIError(c.Config.Logger, c.Config.Alerter, w, r, apierrors.NewErrInternal(err), true)
+		return
+	}
+
 	// report the error to the error alerter but don't send to client
 	apierrors.HandleAPIError(c.Config.Logger, c.Config.Alerter, w, r, apierrors.NewErrInternal(
 		fmt.Errorf(req.Error),