# Copyright 2020 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""The workflow to manipulate an AlertGroup.

We want to separate the workflow from data model. So workflow
is used to implement all AlertGroup's state transitions.

A typical workflow includes these steps:
- Associate anomalies to an AlertGroup
- Update related issues
- Trigger auto-triage if necessary
- Trigger auto-bisection if necessary
- Manage an AlertGroup's lifecycle

`AlertGroupWorkflow(group).Process()` is enough for most of use cases.
But it provides the ability to mock any input and any service, which makes
testing easier and we can have a more predictable behaviour.
"""
from __future__ import print_function
from __future__ import division
from __future__ import absolute_import

import collections
import datetime
import itertools
import jinja2
import logging
import os
import six

from google.appengine.ext import ndb

from dashboard import pinpoint_request
from dashboard import sheriff_config_client
from dashboard import revision_info_client
from dashboard.common import file_bug
from dashboard.common import utils
from dashboard.models import alert_group
from dashboard.models import anomaly
from dashboard.models import subscription
from dashboard.services import crrev_service
from dashboard.services import gitiles_service
from dashboard.services import issue_tracker_service
from dashboard.services import pinpoint_service

# Templates used for rendering issue contents
_TEMPLATE_LOADER = jinja2.FileSystemLoader(
    searchpath=os.path.join(os.path.dirname(os.path.realpath(__file__))))
_TEMPLATE_ENV = jinja2.Environment(loader=_TEMPLATE_LOADER)
_TEMPLATE_ISSUE_TITLE = jinja2.Template(
    '[{{ group.subscription_name }}]: '
    '{{ regressions|length }} regressions in {{ group.name }}')
_TEMPLATE_ISSUE_CONTENT = _TEMPLATE_ENV.get_template(
    'alert_groups_bug_description.j2')
_TEMPLATE_ISSUE_COMMENT = _TEMPLATE_ENV.get_template(
    'alert_groups_bug_comment.j2')
_TEMPLATE_REOPEN_COMMENT = _TEMPLATE_ENV.get_template('reopen_issue_comment.j2')
_TEMPLATE_AUTO_BISECT_COMMENT = _TEMPLATE_ENV.get_template(
    'auto_bisect_comment.j2')
_TEMPLATE_GROUP_WAS_MERGED = _TEMPLATE_ENV.get_template(
    'alert_groups_merge_bug_comment.j2')

# Waiting 7 days to gather more potential alerts. Just choose a long
# enough time and all alerts arrive after archived shouldn't be silent
# merged.
_ALERT_GROUP_ACTIVE_WINDOW = datetime.timedelta(days=7)

# (2020-05-01) Only ~62% issues' alerts are triggered in one hour.
# But we don't want to wait all these long tail alerts finished.
# 20 minutes are enough for a single bot.
#
# SELECT APPROX_QUANTILES(diff, 100) as percentiles
# FROM (
#   SELECT TIMESTAMP_DIFF(MAX(timestamp), MIN(timestamp), MINUTE) as diff
#   FROM chromeperf.chromeperf_dashboard_data.anomalies
#   WHERE 'Chromium Perf Sheriff' IN UNNEST(subscription_names)
#         AND bug_id IS NOT NULL AND timestamp > '2020-03-01'
#   GROUP BY bug_id
# )
_ALERT_GROUP_TRIAGE_DELAY = datetime.timedelta(minutes=20)

# The score is based on overall 60% reproduction rate of pinpoint bisection.
_ALERT_GROUP_DEFAULT_SIGNAL_QUALITY_SCORE = 0.6


class SignalQualityScore(ndb.Model):
  score = ndb.FloatProperty()
  updated_time = ndb.DateTimeProperty()


class InvalidPinpointRequest(Exception):
  pass


class AlertGroupWorkflow(object):
  """Workflow used to manipulate the AlertGroup.

  Workflow will assume the group passed from caller is same as the group in
  datastore. It may update the group in datastore multiple times during the
  process.
  """

  class Config(
      collections.namedtuple('WorkflowConfig',
                             ('active_window', 'triage_delay'))):
    __slots__ = ()

  class GroupUpdate(
      collections.namedtuple('GroupUpdate',
                             ('now', 'anomalies', 'issue', 'canonical_group'))):
    __slots__ = ()

    def __new__(cls, now, anomalies, issue, canonical_group=None):
      return super(AlertGroupWorkflow.GroupUpdate,
                   cls).__new__(cls, now, anomalies, issue, canonical_group)

  class BenchmarkDetails(
      collections.namedtuple('BenchmarkDetails',
                             ('name', 'owners', 'regressions', 'info_blurb'))):
    __slots__ = ()

  class BugUpdateDetails(
      collections.namedtuple('BugUpdateDetails',
                             ('components', 'cc', 'labels'))):
    __slots__ = ()

  def __init__(
      self,
      group,
      config=None,
      sheriff_config=None,
      issue_tracker=None,
      pinpoint=None,
      crrev=None,
      gitiles=None,
      revision_info=None,
      service_account=None,
  ):
    self._group = group
    self._config = config or self.Config(
        active_window=_ALERT_GROUP_ACTIVE_WINDOW,
        triage_delay=_ALERT_GROUP_TRIAGE_DELAY,
    )
    self._sheriff_config = (
        sheriff_config or sheriff_config_client.GetSheriffConfigClient())
    self._issue_tracker = issue_tracker or _IssueTracker()
    self._pinpoint = pinpoint or pinpoint_service
    self._crrev = crrev or crrev_service
    self._gitiles = gitiles or gitiles_service
    self._revision_info = revision_info or revision_info_client
    self._service_account = service_account or utils.ServiceAccountEmail

  def _FindCanonicalGroup(self, issue):
    """Finds the canonical issue group if any.

    Args:
      issue: Monorail API issue json. If the issue has any comments the json
        should contain additional 'comments' key with the list of Monorail API
        comments jsons.

    Returns:
      AlertGroup object or None if the issue is not duplicate, canonical issue
      has no corresponding group or duplicate chain forms a loop.
    """
    if issue.get('status') != issue_tracker_service.STATUS_DUPLICATE:
      return None

    merged_into = None
    latest_id = 0
    for comment in issue.get('comments', []):
      if comment['updates'].get('mergedInto') and comment['id'] >= latest_id:
        merged_into = int(comment['updates'].get('mergedInto'))
        latest_id = comment['id']
    if not merged_into:
      return None
    logging.info('Found canonical issue for the groups\' issue: %d',
                 merged_into)

    query = alert_group.AlertGroup.query(
        alert_group.AlertGroup.active == True,
        # It is impossible to merge bugs from different projects in monorail.
        # So the canonical group bug is guarandeed to have the same project.
        alert_group.AlertGroup.bug.project == self._group.bug.project,
        alert_group.AlertGroup.bug.bug_id == merged_into)
    query_result = query.fetch(limit=1)
    if not query_result:
      return None

    canonical_group = query_result[0]
    visited = set()
    while canonical_group.canonical_group:
      visited.add(canonical_group.key)
      next_group_key = canonical_group.canonical_group
      # Visited check is just precaution.
      # If it is true - the system previously failed to prevent loop creation.
      if next_group_key == self._group.key or next_group_key in visited:
        logging.warning(
            'Alert group auto merge failed. Found a loop while '
            'searching for a canonical group for %r', self._group)
        return None
      canonical_group = next_group_key.get()

    logging.info('Found canonical group: %s', canonical_group.key.string_id())
    return canonical_group

  def _FindDuplicateGroups(self):
    query = alert_group.AlertGroup.query(
        alert_group.AlertGroup.active == True,
        alert_group.AlertGroup.canonical_group == self._group.key)
    return query.fetch()

  def _FindRelatedAnomalies(self, groups):
    query = anomaly.Anomaly.query(
        anomaly.Anomaly.groups.IN([g.key for g in groups]))
    return query.fetch()

  def _PrepareGroupUpdate(self):
    """Prepares default input for the workflow Process

    Returns:
      GroupUpdate object that contains list of related anomalies,
      Monorail API issue json and canonical AlertGroup if any.
    """
    duplicate_groups = self._FindDuplicateGroups()
    anomalies = self._FindRelatedAnomalies([self._group] + duplicate_groups)
    now = datetime.datetime.utcnow()
    issue = None
    canonical_group = None
    if self._group.status in {
        self._group.Status.triaged, self._group.Status.bisected,
        self._group.Status.closed
    }:
      issue = self._issue_tracker.GetIssue(
          self._group.bug.bug_id, project=self._group.bug.project)
      # GetIssueComments doesn't work with empty project id so we have to
      # manually replace it with 'chromium'.
      issue['comments'] = self._issue_tracker.GetIssueComments(
          self._group.bug.bug_id, project=self._group.bug.project or 'chromium')
      canonical_group = self._FindCanonicalGroup(issue)
    return self.GroupUpdate(now, anomalies, issue, canonical_group)

  def Process(self, update=None):
    """Process the workflow.

    The workflow promises to only depend on the provided update and injected
    dependencies. The workflow steps will always be reproducible if all the
    inputs are the same.

    Process will always update the group and store once the steps have
    completed.

    The update argument can be a prepared GroupUpdate instance or None (if
    None, then Process will prepare the update itself).

    Returns the key for the associated group when the workflow was
    initialized."""

    logging.info('Processing workflow for group %s', self._group.key)
    update = update or self._PrepareGroupUpdate()
    logging.info('%d anomalies', len(update.anomalies))

    # TODO(crbug.com/1240370): understand why Datastore query may return empty
    # anomalies list.
    if (not update.anomalies and self._group.anomalies
        and self._group.group_type != alert_group.AlertGroup.Type.reserved):
      logging.error('No anomailes detected. Skipping this run.')
      return self._group.key

    # Process input before we start processing the group.
    for a in update.anomalies:
      subscriptions, _ = self._sheriff_config.Match(
          a.test.string_id(), check=True)
      a.subscriptions = subscriptions
      matching_subs = [
          s for s in subscriptions if s.name == self._group.subscription_name
      ]
      a.auto_triage_enable = any(s.auto_triage_enable for s in matching_subs)
      if a.auto_triage_enable:
        logging.info('auto_triage_enable for %s due to subscription: %s',
                     a.test.string_id(),
                     [s.name for s in matching_subs if s.auto_triage_enable])

      a.auto_merge_enable = any(s.auto_merge_enable for s in matching_subs)

      if a.auto_merge_enable:
        logging.info('auto_merge_enable for %s due to subscription: %s',
                     a.test.string_id(),
                     [s.name for s in matching_subs if s.auto_merge_enable])

      a.auto_bisect_enable = any(s.auto_bisect_enable for s in matching_subs)
      a.relative_delta = (
          abs(a.absolute_delta / float(a.median_before_anomaly))
          if a.median_before_anomaly != 0. else float('Inf'))

    added = self._UpdateAnomalies(update.anomalies)

    if update.issue:
      group_merged = self._UpdateCanonicalGroup(update.anomalies,
                                                update.canonical_group)
      self._UpdateStatus(update.issue)
      self._UpdateAnomaliesIssues(update.anomalies, update.canonical_group)

      # Current group is a duplicate.
      if self._group.canonical_group is not None:
        if group_merged:
          logging.info('Merged group %s into group %s',
                       self._group.key.string_id(),
                       update.canonical_group.key.string_id())
          self._FileDuplicatedNotification(update.canonical_group)
        self._UpdateDuplicateIssue(update.anomalies, added)
        assert (self._group.status == self._group.Status.closed), (
            'The issue is closed as duplicate (\'state\' is \'closed\'). '
            'However the groups\' status doesn\'t match the issue status')

      elif self._UpdateIssue(update.issue, update.anomalies, added):
        # Only operate on alert group if nothing updated to prevent flooding
        # monorail if some operations keep failing.
        return self._CommitGroup()

    group = self._group
    if group.updated + self._config.active_window <= update.now:
      self._Archive()
    elif group.created + self._config.triage_delay <= update.now and (
        group.status in {group.Status.untriaged}):
      logging.info('created: %s, triage_delay: %s", now: %s, status: %s',
                   group.created, self._config.triage_delay, update.now,
                   group.status)
      self._TryTriage(update.now, update.anomalies)
    elif group.status in {group.Status.triaged}:
      self._TryBisect(update)
    return self._CommitGroup()

  def _CommitGroup(self):
    return self._group.put()

  def _UpdateAnomalies(self, anomalies):
    added = [a for a in anomalies if a.key not in self._group.anomalies]
    self._group.anomalies = [a.key for a in anomalies]
    return added

  def _UpdateStatus(self, issue):
    if issue.get('state') == 'closed':
      self._group.status = self._group.Status.closed
    elif self._group.status == self._group.Status.closed:
      self._group.status = self._group.Status.triaged

  def _UpdateCanonicalGroup(self, anomalies, canonical_group):
    # If canonical_group is None, self._group will be separated from its'
    # canonical group. Since we only rely on _group.canonical_group for
    # determining duplicate status, setting canonical_group to None will
    # separate the groups. Anomalies that were added to the canonical group
    # during merged perios can't be removed.
    if canonical_group is None:
      self._group.canonical_group = None
      return False
    # Only merge groups if there is at least one anomaly that allows merge.
    if (self._group.canonical_group != canonical_group.key
        and any(a.auto_merge_enable for a in anomalies)):
      self._group.canonical_group = canonical_group.key
      return True
    return False

  def _UpdateAnomaliesIssues(self, anomalies, canonical_group):
    for a in anomalies:
      if not a.auto_triage_enable:
        continue
      if canonical_group is not None and a.auto_merge_enable:
        a.project_id = canonical_group.project_id
        a.bug_id = canonical_group.bug.bug_id
      elif a.bug_id is None:
        a.project_id = self._group.project_id
        a.bug_id = self._group.bug.bug_id

    # Write back bug_id to anomalies. We can't do it when anomaly is
    # found because group may be updating at the same time.
    ndb.put_multi(anomalies)

  def _UpdateIssue(self, issue, anomalies, added):
    """Update the status of the monorail issue.

    Returns True if the issue was changed.
    """
    # Check whether all the anomalies associated have been marked recovered.
    if all(a.recovered for a in anomalies if not a.is_improvement):
      if issue.get('state') == 'open':
        self._CloseBecauseRecovered()
      return True

    new_regressions, subscriptions = self._GetRegressions(added)
    all_regressions, _ = self._GetRegressions(anomalies)

    # Only update issue if there is at least one regression
    if not new_regressions:
      return False

    closed_by_pinpoint = False
    for c in sorted(
        issue.get('comments') or [], key=lambda c: c["id"], reverse=True):
      if c.get('updates', {}).get('status') in ('WontFix', 'Fixed', 'Verified',
                                                'Invalid', 'Duplicate', 'Done'):
        closed_by_pinpoint = (c.get('author') == self._service_account())