fix(reports): don't log user errors and state change has errors (#12277)

superset/reports/commands/exceptions.py

@@ -181,3 +181,7 @@ class ReportScheduleUnexpectedError(CommandException):
 
 class ReportScheduleForbiddenError(ForbiddenError):
     message = _("Changing this report is forbidden")
+
+
+class ReportSchedulePruneLogError(CommandException):
+    message = _("An error occurred while pruning logs ")

superset/reports/commands/log_prune.py

@@ -18,7 +18,9 @@ import logging
 from datetime import datetime, timedelta
 
 from superset.commands.base import BaseCommand
+from superset.dao.exceptions import DAODeleteFailedError
 from superset.models.reports import ReportSchedule
+from superset.reports.commands.exceptions import ReportSchedulePruneLogError
 from superset.reports.dao import ReportScheduleDAO
 from superset.utils.celery import session_scope
 
@@ -36,14 +38,26 @@ class AsyncPruneReportScheduleLogCommand(BaseCommand):
     def run(self) -> None:
         with session_scope(nullpool=True) as session:
             self.validate()
+            prune_errors = []
+
             for report_schedule in session.query(ReportSchedule).all():
                 if report_schedule.log_retention is not None:
                     from_date = datetime.utcnow() - timedelta(
                         days=report_schedule.log_retention
                     )
-                    ReportScheduleDAO.bulk_delete_logs(
-                        report_schedule, from_date, session=session, commit=False
-                    )
+                    try:
+                        row_count = ReportScheduleDAO.bulk_delete_logs(
+                            report_schedule, from_date, session=session, commit=False
+                        )
+                        logger.info(
+                            "Deleted %s logs for %s",
+                            str(row_count),
+                            ReportSchedule.name,
+                        )
+                    except DAODeleteFailedError as ex:
+                        prune_errors.append(str(ex))
+            if prune_errors:
+                raise ReportSchedulePruneLogError(";".join(prune_errors))
 
     def validate(self) -> None:
         pass

superset/reports/dao.py

@@ -104,10 +104,10 @@ class ReportScheduleDAO(BaseDAO):
                 db.session.delete(report_schedule)
             if commit:
                 db.session.commit()
-        except SQLAlchemyError:
+        except SQLAlchemyError as ex:
             if commit:
                 db.session.rollback()
-            raise DAODeleteFailedError()
+            raise DAODeleteFailedError(str(ex))
 
     @staticmethod
     def validate_update_uniqueness(
@@ -156,9 +156,9 @@ class ReportScheduleDAO(BaseDAO):
             if commit:
                 db.session.commit()
             return model
-        except SQLAlchemyError:
+        except SQLAlchemyError as ex:
             db.session.rollback()
-            raise DAOCreateFailedError
+            raise DAOCreateFailedError(str(ex))
 
     @classmethod
     def update(
@@ -190,9 +190,9 @@ class ReportScheduleDAO(BaseDAO):
             if commit:
                 db.session.commit()
             return model
-        except SQLAlchemyError:
+        except SQLAlchemyError as ex:
             db.session.rollback()
-            raise DAOCreateFailedError
+            raise DAOCreateFailedError(str(ex))
 
     @staticmethod
     def find_active(session: Optional[Session] = None) -> List[ReportSchedule]:
@@ -229,16 +229,21 @@ class ReportScheduleDAO(BaseDAO):
         from_date: datetime,
         session: Optional[Session] = None,
         commit: bool = True,
-    ) -> None:
+    ) -> Optional[int]:
         session = session or db.session
         try:
-            session.query(ReportExecutionLog).filter(
-                ReportExecutionLog.report_schedule == model,
-                ReportExecutionLog.end_dttm < from_date,
-            ).delete(synchronize_session="fetch")
+            row_count = (
+                session.query(ReportExecutionLog)
+                .filter(
+                    ReportExecutionLog.report_schedule == model,
+                    ReportExecutionLog.end_dttm < from_date,
+                )
+                .delete(synchronize_session="fetch")
+            )
             if commit:
                 session.commit()
+            return row_count
         except SQLAlchemyError as ex:
             if commit:
                 session.rollback()
-            raise ex
+            raise DAODeleteFailedError(str(ex))

superset/tasks/scheduler.py

@@ -23,6 +23,7 @@ import croniter
 from superset import app
 from superset.commands.exceptions import CommandException
 from superset.extensions import celery_app
+from superset.reports.commands.exceptions import ReportScheduleUnexpectedError
 from superset.reports.commands.execute import AsyncExecuteReportScheduleCommand
 from superset.reports.commands.log_prune import AsyncPruneReportScheduleLogCommand
 from superset.reports.dao import ReportScheduleDAO
@@ -62,8 +63,10 @@ def scheduler() -> None:
 def execute(report_schedule_id: int, scheduled_dttm: datetime) -> None:
     try:
         AsyncExecuteReportScheduleCommand(report_schedule_id, scheduled_dttm).run()
+    except ReportScheduleUnexpectedError as ex:
+        logger.error("An unexpected occurred while executing the report: %s", ex)
     except CommandException as ex:
-        logger.error("An exception occurred while executing the report: %s", ex)
+        logger.info("Report state: %s", ex)
 
 
 @celery_app.task(name="reports.prune_log")