bkyryliuk commented on a change in pull request #9944:
URL:
https://github.com/apache/incubator-superset/pull/9944#discussion_r445756450
##########
File path: superset/charts/api.py
##########
@@ -497,12 +637,17 @@ def thumbnail(
if not chart:
return self.response_404()
if kwargs["rison"].get("force", False):
+ logger.info("Triggering thumbnail compute ASYNC")
Review comment:
maybe add chart id to the logging message
##########
File path: superset/models/alerts.py
##########
@@ -0,0 +1,102 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Models for scheduled execution of jobs"""
+from datetime import datetime
+
+from flask_appbuilder import Model
+from sqlalchemy import (
+ Boolean,
+ Column,
+ DateTime,
+ ForeignKey,
+ Integer,
+ String,
+ Table,
+ Text,
+)
+from sqlalchemy.orm import backref, relationship
+
+from superset import security_manager
+
+metadata = Model.metadata # pylint: disable=no-member
+
+
+alert_owner = Table(
+ "alert_owner",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("user_id", Integer, ForeignKey("ab_user.id")),
+ Column("alert_id", Integer, ForeignKey("alerts.id")),
+)
+
+
+class Alert(Model):
Review comment:
would be nice to have AuditMixinNullable
##########
File path: superset/charts/api.py
##########
@@ -453,6 +459,140 @@ def data(self) -> Response:
resp.headers["Content-Type"] = "application/json; charset=utf-8"
return resp
+ @expose("/<pk>/cache_screenshot/", methods=["GET"])
+ @protect()
+ @rison(screenshot_query_schema)
+ @safe
+ @statsd_metrics
+ def cache_screenshot(self, pk: int, **kwargs: Dict[str, bool]) ->
WerkzeugResponse:
+ """Get Chart screenshot
+ ---
+ get:
+ description: Compute or get already computed screenshot from cache.
+ parameters:
+ - in: path
+ schema:
+ type: integer
+ name: pk
+ - in: path
+ schema:
+ type: string
+ name: sha
+ responses:
+ 200:
+ description: Chart thumbnail image
+ content:
+ application/json:
+ schema:
+ type: object
+ properties:
+ cache_key:
+ type: string
+ chart_url:
+ type: string
+ image_url:
+ type: string
+ 302:
+ description: Redirects to the current digest
+ 400:
+ $ref: '#/components/responses/400'
+ 401:
+ $ref: '#/components/responses/401'
+ 404:
+ $ref: '#/components/responses/404'
+ 500:
+ $ref: '#/components/responses/500'
+ """
+ rison_dict = kwargs["rison"]
+ window_size = rison_dict.get("window_size") or (800, 600)
+
+ # Don't shrink the image if thumb_size is not specified
+ thumb_size = rison_dict.get("thumb_size") or window_size
+
+ chart = self.datamodel.get(pk, self._base_filters)
+ if not chart:
+ return self.response_404()
+
+ chart_url = get_url_path("Superset.slice", slice_id=chart.id,
standalone="true")
+ screenshot_obj = ChartScreenshot(chart_url, chart.digest)
+ cache_key = screenshot_obj.cache_key(window_size, thumb_size)
+ image_url = get_url_path(
+ "ChartRestApi.screenshot", pk=chart.id, digest=cache_key
+ )
+
+ def trigger_celery():
+ logger.info("Triggering screenshot ASYNC")
+ kwargs = {
+ "url": chart_url,
+ "digest": chart.digest,
+ "force": True,
+ "window_size": window_size,
+ "thumb_size": thumb_size,
+ }
+ cache_chart_thumbnail.delay(**kwargs)
+ return self.response(
+ 202, cache_key=cache_key, chart_url=chart_url,
image_url=image_url,
+ )
+
+ return trigger_celery()
+
+ @expose("/<pk>/screenshot/<digest>/", methods=["GET"])
+ @protect()
+ @rison(screenshot_query_schema)
+ @safe
+ @statsd_metrics
+ def screenshot(
Review comment:
s/screenshot/cached_screenshot?
##########
File path: superset/charts/api.py
##########
@@ -453,6 +459,140 @@ def data(self) -> Response:
resp.headers["Content-Type"] = "application/json; charset=utf-8"
return resp
+ @expose("/<pk>/cache_screenshot/", methods=["GET"])
+ @protect()
+ @rison(screenshot_query_schema)
+ @safe
+ @statsd_metrics
+ def cache_screenshot(self, pk: int, **kwargs: Dict[str, bool]) ->
WerkzeugResponse:
Review comment:
s/cache_screenshot/schedule_screenshot
##########
File path: superset/tasks/schedules.py
##########
@@ -79,11 +86,13 @@ def _get_recipients(
def _deliver_email(
- schedule: Union[DashboardEmailSchedule, SliceEmailSchedule],
+ schedule: Union[DashboardEmailSchedule, SliceEmailSchedule, Alert],
subject: str,
email: EmailContent,
) -> None:
for (to, bcc) in _get_recipients(schedule):
+ logging.info("Sending email to [%s] bcc [%s]", to, bcc)
Review comment:
maybe do it debug logging? so it is configurable if needed ?
##########
File path: superset/tasks/schedules.py
##########
@@ -475,3 +638,22 @@ def schedule_hourly() -> None:
stop_at = start_at + timedelta(seconds=3600)
schedule_window(ScheduleType.dashboard, start_at, stop_at, resolution)
schedule_window(ScheduleType.slice, start_at, stop_at, resolution)
+
+
+@celery_app.task(name="alerts.schedule_check")
+def schedule_alerts() -> None:
+ """ Celery beat job meant to be invoked every minute to check alerts """
+
+ # if not config["ENABLE_SCHEDULED_EMAIL_REPORTS"]:
Review comment:
probably just delete it, I assume this function should not be called if
it's not enabled
##########
File path: superset/charts/api.py
##########
@@ -453,6 +459,140 @@ def data(self) -> Response:
resp.headers["Content-Type"] = "application/json; charset=utf-8"
return resp
+ @expose("/<pk>/cache_screenshot/", methods=["GET"])
+ @protect()
+ @rison(screenshot_query_schema)
+ @safe
+ @statsd_metrics
+ def cache_screenshot(self, pk: int, **kwargs: Dict[str, bool]) ->
WerkzeugResponse:
+ """Get Chart screenshot
Review comment:
s/Get Chart screenshot/Schedule screenshot computation in celery
##########
File path: superset/utils/screenshots.py
##########
@@ -218,28 +229,39 @@ def get(
:param thumb_size: Override thumbnail site
"""
payload: Optional[bytes] = None
- thumb_size = thumb_size or self.thumb_size
+ cache_key = self.cache_key(self.window_size, thumb_size)
if cache:
- payload = cache.get(self.cache_key)
+ payload = cache.get(cache_key)
if not payload:
payload = self.compute_and_cache(
user=user, thumb_size=thumb_size, cache=cache
)
else:
- logger.info(f"Loaded thumbnail from cache: {self.cache_key}")
+ logger.info(f"Loaded thumbnail from cache: {cache_key}")
if payload:
return BytesIO(payload)
return None
- def get_from_cache(self, cache: "Cache") -> Optional[BytesIO]:
- payload = cache.get(self.cache_key)
+ def get_from_cache(
+ self, cache: "Cache", window_size=None, thumb_size=None,
+ ) -> Optional[BytesIO]:
+ cache_key = self.cache_key(window_size, thumb_size)
+ payload = cache.get(cache_key)
+ return self.get_from_cache_key(cache, cache_key)
+
+ @staticmethod
+ def get_from_cache_key(cache: "Cache", cache_key: str) ->
Optional[BytesIO]:
+ logger.info("Attempting to get from cache: %s", cache_key)
+ payload = cache.get(cache_key)
if payload:
return BytesIO(payload)
+ logger.info("Failed at getting from cache: %s", cache_key)
return None
def compute_and_cache( # pylint: disable=too-many-arguments
self,
user: "User" = None,
+ window_size: Optional[WindowSize] = None,
Review comment:
would be nice to have a timeout here and time to live here
e.g. for hourly alerts ttl should be 1 hour tops
##########
File path: superset/migrations/versions/2f1d15e8a6af_add_alerts.py
##########
@@ -0,0 +1,83 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""add_alerts
+
+Revision ID: 2f1d15e8a6af
+Revises: 620241d1153f
+Create Date: 2020-05-26 23:21:50.059635
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = "2f1d15e8a6af"
+down_revision = "620241d1153f"
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import mysql
+
+
+def upgrade():
+ # ### commands auto generated by Alembic - please adjust! ###
+ op.create_table(
+ "alerts",
+ sa.Column("id", sa.Integer(), nullable=False),
+ sa.Column("label", sa.String(length=150), nullable=False),
+ sa.Column("active", sa.Boolean(), nullable=True),
+ sa.Column("crontab", sa.String(length=50), nullable=True),
+ sa.Column("sql", sa.Text(), nullable=True),
+ sa.Column("alert_type", sa.String(length=50), nullable=True),
+ sa.Column("log_retention", sa.Integer(), nullable=False, default=90),
+ sa.Column("grace_period", sa.Integer(), nullable=False, default=60 *
60 * 24),
+ sa.Column("recipients", sa.Text(), nullable=True),
+ sa.Column("slice_id", sa.Integer(), nullable=True),
+ sa.Column("database_id", sa.Integer(), nullable=False),
+ sa.Column("dashboard_id", sa.Integer(), nullable=True),
+ sa.Column("last_eval_dttm", sa.DateTime(), nullable=True),
+ sa.Column("last_state", sa.String(length=10), nullable=True),
+ sa.ForeignKeyConstraint(["dashboard_id"], ["dashboards.id"],),
+ sa.ForeignKeyConstraint(["slice_id"], ["slices.id"],),
+ sa.PrimaryKeyConstraint("id"),
+ )
+ op.create_index(op.f("ix_alerts_active"), "alerts", ["active"],
unique=False)
+ op.create_table(
+ "alert_logs",
+ sa.Column("id", sa.Integer(), nullable=False),
+ sa.Column("scheduled_dttm", sa.DateTime(), nullable=True),
+ sa.Column("dttm_start", sa.DateTime(), nullable=True),
+ sa.Column("dttm_end", sa.DateTime(), nullable=True),
+ sa.Column("alert_id", sa.Integer(), nullable=True),
+ sa.Column("state", sa.String(length=10), nullable=True),
+ sa.ForeignKeyConstraint(["alert_id"], ["alerts.id"],),
+ sa.PrimaryKeyConstraint("id"),
+ )
+ op.create_table(
+ "alert_owner",
Review comment:
s/alert_owner/alert_owners
if I get it right alert can have many owners
##########
File path: superset/charts/api.py
##########
@@ -453,6 +459,140 @@ def data(self) -> Response:
resp.headers["Content-Type"] = "application/json; charset=utf-8"
return resp
+ @expose("/<pk>/cache_screenshot/", methods=["GET"])
+ @protect()
+ @rison(screenshot_query_schema)
+ @safe
+ @statsd_metrics
+ def cache_screenshot(self, pk: int, **kwargs: Dict[str, bool]) ->
WerkzeugResponse:
+ """Get Chart screenshot
+ ---
+ get:
+ description: Compute or get already computed screenshot from cache.
+ parameters:
+ - in: path
+ schema:
+ type: integer
+ name: pk
+ - in: path
+ schema:
+ type: string
+ name: sha
+ responses:
+ 200:
+ description: Chart thumbnail image
+ content:
+ application/json:
+ schema:
+ type: object
+ properties:
+ cache_key:
+ type: string
+ chart_url:
+ type: string
+ image_url:
+ type: string
+ 302:
+ description: Redirects to the current digest
+ 400:
+ $ref: '#/components/responses/400'
+ 401:
+ $ref: '#/components/responses/401'
+ 404:
+ $ref: '#/components/responses/404'
+ 500:
+ $ref: '#/components/responses/500'
+ """
+ rison_dict = kwargs["rison"]
+ window_size = rison_dict.get("window_size") or (800, 600)
+
+ # Don't shrink the image if thumb_size is not specified
+ thumb_size = rison_dict.get("thumb_size") or window_size
+
+ chart = self.datamodel.get(pk, self._base_filters)
+ if not chart:
+ return self.response_404()
+
+ chart_url = get_url_path("Superset.slice", slice_id=chart.id,
standalone="true")
+ screenshot_obj = ChartScreenshot(chart_url, chart.digest)
+ cache_key = screenshot_obj.cache_key(window_size, thumb_size)
+ image_url = get_url_path(
+ "ChartRestApi.screenshot", pk=chart.id, digest=cache_key
+ )
+
+ def trigger_celery():
+ logger.info("Triggering screenshot ASYNC")
+ kwargs = {
+ "url": chart_url,
+ "digest": chart.digest,
+ "force": True,
+ "window_size": window_size,
+ "thumb_size": thumb_size,
+ }
+ cache_chart_thumbnail.delay(**kwargs)
+ return self.response(
+ 202, cache_key=cache_key, chart_url=chart_url,
image_url=image_url,
+ )
+
+ return trigger_celery()
+
+ @expose("/<pk>/screenshot/<digest>/", methods=["GET"])
+ @protect()
+ @rison(screenshot_query_schema)
+ @safe
+ @statsd_metrics
+ def screenshot(
+ self, pk: int, digest: str = None, **kwargs: Dict[str, bool]
+ ) -> WerkzeugResponse:
+ """Get Chart screenshot
+ ---
+ get:
+ description: Get a computed screenshot from cache.
+ parameters:
+ - in: path
+ schema:
+ type: integer
+ name: pk
+ - in: path
+ schema:
+ type: string
+ name: digest
+ responses:
+ 200:
+ description: Chart thumbnail image
+ content:
+ image/*:
+ schema:
+ type: string
+ format: binary
+ 302:
+ description: Redirects to the current digest
+ 400:
+ $ref: '#/components/responses/400'
+ 401:
+ $ref: '#/components/responses/401'
+ 404:
+ $ref: '#/components/responses/404'
+ 500:
+ $ref: '#/components/responses/500'
+ """
+ chart = self.datamodel.get(pk, self._base_filters)
+
+ # Making sure the chart still exists
+ if not chart:
+ return self.response_404()
+
+ # TODO make sure the user has access to the chart
+
+ # fetch the chart screenshot using the current user and cache if set
+ img = ChartScreenshot.get_from_cache_key(thumbnail_cache, digest)
+ if img:
+ return Response(
+ FileWrapper(img), mimetype="image/png", direct_passthrough=True
+ )
+ # TODO: return an empty image
+ return None
Review comment:
maybe return 404 ?
##########
File path: superset/cli.py
##########
@@ -593,3 +593,14 @@ def sync_tags():
add_types(db.engine, metadata)
add_owners(db.engine, metadata)
add_favorites(db.engine, metadata)
+
+
+@with_appcontext
[email protected]()
Review comment:
delete me :)
it would be useful to have a dev cli for things like it
##########
File path: superset/tasks/schedules.py
##########
@@ -410,6 +419,141 @@ def schedule_email_report( # pylint:
disable=unused-argument
raise RuntimeError("Unknown report type")
+@celery_app.task(
+ name="alerts.run_query",
+ bind=True,
+ soft_time_limit=config["EMAIL_ASYNC_TIME_LIMIT_SEC"],
+)
+def schedule_alert_query( # pylint: disable=unused-argument
+ task: Task,
+ report_type: ScheduleType,
+ schedule_id: int,
+ recipients: Optional[str] = None,
+) -> None:
+ model_cls = get_scheduler_model(report_type)
+ dbsession = db.create_scoped_session()
+ schedule = dbsession.query(model_cls).get(schedule_id)
+
+ # The user may have disabled the schedule. If so, ignore this
+ if not schedule or not schedule.active:
+ logger.info("Ignoring deactivated alert")
+ return
+
+ if report_type == ScheduleType.alert:
+ if run_alert_query(schedule, dbsession):
+ # deliver_dashboard OR deliver_slice
+ return
+ else:
+ raise RuntimeError("Unknown report type")
+
+
+class AlertState:
+ ERROR = "error"
+ TRIGGER = "trigger"
+ PASS = "pass"
+
+
+def deliver_alert(alert):
+ logging.info("Triggering alert: %s", alert)
+ img_data = None
+ if alert.slice:
+
+ chart_url = get_url_path(
+ "Superset.slice", slice_id=alert.slice.id, standalone="true"
+ )
+ screenshot = ChartScreenshot(chart_url, alert.slice.digest)
+ cache_key = screenshot.cache_key()
+ image_url = get_url_path(
+ "ChartRestApi.screenshot", pk=alert.slice.id, digest=cache_key
+ )
+
+ user =
security_manager.find_user(current_app.config["THUMBNAIL_SELENIUM_USER"])
+ img_data = screenshot.compute_and_cache(
+ user=user, cache=thumbnail_cache, force=True,
+ )
+ else:
+ image_url = "https://media.giphy.com/media/dzaUX7CAG0Ihi/giphy.gif";
+
+ # generate the email
+ subject = f"[Superset] Triggered alert: {alert.label}"
+ data = None
+ images = {"screenshot": img_data}
+ body = __(
+ textwrap.dedent(
+ """\
+ <h2>Alert: %(label)s</h2>
+ <img src="cid:screenshot" alt="%(label)s" />
+ """
+ ),
+ label=alert.label,
+ image_url=image_url,
+ )
+
+ email = EmailContent(body, data, images)
+
+ # send the email
+ _deliver_email(alert, subject, email)
+
+
+def run_alert_query(alert: Alert, session: Session) -> Optional[bool]:
+ """
+ Execute alert.sql and return value if any rows are returned
+ """
+ logger.info(f"Processing alert: {alert}")
+ database = alert.database
+ if not database:
Review comment:
nit: would be nicer to raise here
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
![https://media.giphy.com/media/dzaUX7CAG0Ihi/giphy.gif"](https://media.giphy.com/media/dzaUX7CAG0Ihi/giphy.gif"){kind=link}