#!/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)