server/app/plugins/jirastats.py

#!/usr/bin/env python3
# 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.
"""ASF Infrastructure Reporting Dashboard - Jira Statistics Tasks"""
import asyncio

from ..lib import config
from .. import plugins
import aiohttp
import time
import re
import asfpy.pubsub
import datetime

DEFAULT_SCAN_INTERVAL = 900  # Always run a scan every 15 minutes
DEFAULT_DISCOUNT_DELTA = 600  # Calculate weekend discounts in 10 min increments
DEFAULT_RETENTION = 120  # Only return tickets that are still open, or were updated in the last 120 days
DEFAULT_SCAN_DAYS = 90  # Scan last 90 days in a full scan. This should be, at max, 500, usually ~375 issues.
DEFAULT_SLA = {  # Default (fallback) SLA
    "respond": 48,  # 48h to respond
    "resolve": 120,  # 120h to resolve
}

_cache: dict = {}
_stats: dict = {}
_scan_schedule: list = []


class JiraTicket:
    def __init__(self, data):
        self._data = data
        self.assignee = data["fields"]["assignee"]["name"] if data["fields"]["assignee"] else None
        self.status = data["fields"]["status"]["name"]
        self.closed = self.status == "Closed"
        self.reopened = False
        self.key = data["key"]
        self.project = self.key.split("-")[0]
        self.url = config.reporting.jira["ticket_url"].format(**data)
        self.summary = data["fields"]["summary"]
        self.created_at = self.get_time(data["fields"]["created"])
        self.updated_at = self.get_time(data["fields"]["updated"])
        self.priority = data["fields"]["priority"]["name"]
        self.author = data["fields"]["creator"] and data["fields"]["creator"]["name"] or "(nobody)"  # May not exist!
        self.issuetype = data["fields"]["issuetype"]["name"]
        self.sla = config.reporting.jira["slas"].get(self.priority, DEFAULT_SLA)

        # SLA stuff
        self.first_response = 0
        self.response_time = 0
        self.resolve_time = 0
        self.closed_at = 0
        self.sla_met_respond = None  # True/False if responded to at all
        self.sla_met_resolve = None
        self.sla_time_counted = 0
        self.statuses = []
        self.changelog = []
        self.paused = self.issuetype in config.reporting.jira.get("no_slas", [])

        # Scan all changelog entries
        for changelog_entry in data.get("changelog", {}).get("histories", []):
            changelog_author = (
                "author" in changelog_entry and changelog_entry["author"]["name"] or "nobody"
            )  # May have been deleted
            changelog_epoch = self.get_time(changelog_entry["created"])
            self.changelog.append((changelog_author, changelog_epoch))
            for item in changelog_entry.get("items", []):
                field = item["field"]
                if field == "assignee":  # Ticket (re)assigned
                    #  self.set_fr(changelog_epoch)
                    pass  # Should not count as a response
                elif field == "resolution":  # Ticket resolved
                    self.set_fr(changelog_epoch)
                    self.closed_at = changelog_epoch
                elif field == "status":  # Status change
                    if (
                        self.closed_at
                    ):  # if we already logged a close, but there are new status changes, it's been reopened
                        self.reopened = True
                    if not self.statuses:  # First status change, log initial status from this
                        self.statuses.append((item["fromString"].lower(), self.created_at))
                    self.statuses.append((item["toString"].lower(), changelog_epoch))  # Note change to status at time

        # Scan all comments, looking for a response earlier than changelog entries
        for comment in data["fields"].get("comment", {}).get("comments", []):
            comment_author = comment["author"]["name"]
            comment_epoch = self.get_time(comment["created"])
            self.changelog.append((comment_author, comment_epoch))
            if comment_author != self.author:  # Comment by someone other than the ticket author
                self.set_fr(comment_epoch)
                break  # Only need to find the first (earliest) occurrence

        # Calculate time spent in WFI
        times_in_wfi = []

        if not self.statuses:  # No status changes, WFI is assumed to be entire duration
            if self.closed:
                times_in_wfi.append((self.created_at, self.closed_at))  # Ticket is closed, use closed_at
            else:
                times_in_wfi.append((self.created_at, int(time.time())))  # Ticket is open, use $now
        else:
            sla_statuses_lower = [x.lower() for x in config.reporting.jira.get("sla_apply_statuses")]
            previous_ts = 0
            previous_is_sla = False
            for status in self.statuses:
                if previous_ts and previous_is_sla:
                    times_in_wfi.append((previous_ts, status[1]))  # From previous TS to this one
                previous_ts = status[1]
                previous_is_sla = status[0] in sla_statuses_lower

            # Not in WFI mode? pause if not paused
            if not self.closed and self.statuses[-1][0] not in sla_statuses_lower:
                self.paused = True

        for spans in times_in_wfi:
            self.sla_time_counted += self.calc_sla_duration(*spans)
        if self.first_response:
            self.response_time = self.calc_sla_duration(self.created_at, self.first_response)
        if self.closed_at:
            self.resolve_time = self.calc_sla_duration(self.created_at, self.closed_at)

        # If closed or responded to, check if the duration met the SLA guides
        # If not closed or responded to, check if time spent in WFI surpasses SLA guides
        if self.closed:
            self.sla_met_resolve = self.resolve_time <= (self.sla["resolve"] * 3600)
        elif self.sla_time_counted > (self.sla["resolve"] * 3600):
            self.sla_met_resolve = False
        if self.first_response:
            self.sla_met_respond = self.response_time <= (self.sla["respond"] * 3600)
        elif self.sla_time_counted > (self.sla["respond"] * 3600):
            self.sla_met_respond = False

    @property
    def as_dict(self):
        return {k: v for k, v in self.__dict__.items() if not k.startswith("_")}

    @staticmethod
    def calc_sla_duration(from_epoch, to_epoch):
        """Calculates the active SLA time (in seconds) between two durations, discounting weekends"""
        should_discount = config.reporting.jira.get("sla_discount_weekend")
        seconds_spent = to_epoch - from_epoch  # Add seconds between the two transitions
        if should_discount:
            dt_start = datetime.datetime.utcfromtimestamp(from_epoch)
            dt_end = datetime.datetime.utcfromtimestamp(to_epoch)
            total_discount = 0
            dt_temp = dt_start
            while dt_temp < dt_end and total_discount < seconds_spent:
                dt_temp += datetime.timedelta(seconds=DEFAULT_DISCOUNT_DELTA)
                if (
                    dt_temp.weekday() in [5, 6]  # Sat, Sun
                    or (dt_temp.weekday() == 4 and dt_temp.hour > 20)  # Fri after 8pm UTC
                    or (dt_temp.weekday() == 0 and dt_temp.hour < 8)  # Mon before 8am UTC
                ):
                    total_discount += DEFAULT_DISCOUNT_DELTA
            seconds_spent -= min(seconds_spent, total_discount)
        return seconds_spent

    def set_fr(self, epoch):
        if self.first_response:
            self.first_response = min(self.first_response, epoch)
        else:
            self.first_response = epoch

    @staticmethod
    def get_time(string):
        """Converts a jira ISO timestamp to unix epoch"""
        ts = time.strptime(re.sub(r"\..*", "", str(string)), "%Y-%m-%dT%H:%M:%S")
        epoch_local = time.mktime(ts)
        # Account for UTC offset when computer has a local TZ
        utc_offset = (datetime.datetime.fromtimestamp(epoch_local) - datetime.datetime.utcfromtimestamp(epoch_local)).total_seconds()
        return int(epoch_local + utc_offset)


def process_cache(issues):
    if issues:
        _stats.clear()  # Clear stats cache if we have data, so as to remove deleted tickets
    for issue in issues:
        key = issue["key"]
        _cache[key] = issue
        ticket = JiraTicket(issue)
        _stats[key] = ticket
    return len(issues)


def get_issues():
    deadline = time.time() - (DEFAULT_RETENTION * 86400)
    return [x.as_dict for x in _stats.values() if x.closed is False or x.updated_at >= deadline]


async def jira_scan_full(days=DEFAULT_SCAN_DAYS):
    """Performs a full scan of Jira activity in the past [days] days"""
    jira_scan_url = config.reporting.jira["api_url"] + "search"
    jira_project = config.reporting.jira["project"]
    jira_token = config.reporting.jira["token"]

    params = {
        "fields": "key,created,summary,status,assignee,priority,comment,creator,updated,issuetype",
        "expand": "changelog",
        "maxResults": "1000",
        "jql": f"""project={jira_project} and (updated>=-{days}d or status!=closed)""",
    }

    async with aiohttp.ClientSession(headers={"Authorization": f"Bearer: {jira_token}"}) as hc:
        async with hc.get(jira_scan_url, params=params) as req:
            if req.status == 200:
                jira_json = await req.json()
                processed = process_cache(jira_json.get("issues", []))
                return processed


async def scan_loop():
    while True:
        if _scan_schedule:  # Things are scheduled for a scan
            now = time.time()
            print("Starting Jira scan")
            processed = await jira_scan_full()
            print(f"Processed {processed} tickets in {int(time.time()-now)} seconds")
            _scan_schedule.pop()  # pop an item, freeing up space to allocate a new scan
        await asyncio.sleep(60)  # Always wait 60 secs between scan checks


async def poll_loop():
    """Schedules a scan every DEFAULT_SCAN_INTERVAL seconds, plus when a pubsub event happens.
    No more than two events can be scheduled at any given time (iow if a scan is running, and
    we get a pubsub event, we can add one more scan to be done in the future."""
    loop = asyncio.get_running_loop()

    def maybe_timeout(duration):
        "Use asyncio.timeout() for Py3.11; stub out for lower versions."
        if hasattr(asyncio, 'timeout'):
            return asyncio.timeout(duration)
        import contextlib
        class StubTimeout:
            def reschedule(self, t):
                pass
        @contextlib.asynccontextmanager
        async def gen_stub():
            yield StubTimeout()
        return gen_stub()

    pubsub_url = config.reporting.jira.get("pubsub_url")
    while True:
        _scan_schedule.append(time.time())  # Schedule a scan
        if pubsub_url:
            try:
                async with maybe_timeout(60) as to:
                    async for payload in asfpy.pubsub.listen(pubsub_url):
                        to.reschedule(loop.time() + 60)  # Got a response, pubsub works, reschedule timeout
                        if "stillalive" not in payload:  # Not a ping
                            if len(_scan_schedule) < 2:
                                _scan_schedule.append(time.time())  # add scan to schedule
            except TimeoutError:
                print("PubSub connection timed out, re-establishing")
                continue
        else:
            await asyncio.sleep(DEFAULT_SCAN_INTERVAL)


plugins.root.register(poll_loop, scan_loop, slug="jira", title="Jira Tickets", icon="bi-bug-fill", private=True)