-
Notifications
You must be signed in to change notification settings - Fork 291
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add responders improvements #3128
Changes from 43 commits
4d6794b
4bfd384
0d8b632
4a74bff
dc058c2
3d80dbf
25c7912
c8581b0
748791c
50bd4e7
cd64267
8146c47
0f86f12
4bb90c3
2f72920
2ba681a
0bf9323
a46aede
5d98949
3f30e76
8899845
4bcc713
8a1c17b
2019eee
fd5f5d8
2b932c8
f51c735
28f33f4
d2c2ad5
fc89fe8
0ac4521
a05e691
3808099
88fbfc6
a21492a
8d80c8e
d0b0266
7140308
9acab65
c5ee288
51fdb26
d0c5b0e
02d73c3
cc5388a
a3b69fc
3ac88ad
9855f6f
500a696
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -70,6 +70,16 @@ class LogRecordUser(typing.TypedDict): | |
avatar_full: str | ||
|
||
|
||
class PagedUser(typing.TypedDict): | ||
id: int | ||
username: str | ||
name: str | ||
pk: str | ||
avatar: str | ||
avatar_full: str | ||
important: bool | ||
|
||
|
||
class LogRecords(typing.TypedDict): | ||
time: str # humanized delta relative to now | ||
action: str # human-friendly description | ||
|
@@ -509,22 +519,57 @@ def declare_incident_link(self) -> str: | |
def happened_while_maintenance(self): | ||
return self.root_alert_group is not None and self.root_alert_group.maintenance_uuid is not None | ||
|
||
def get_paged_users(self) -> QuerySet[User]: | ||
def get_paged_users(self) -> typing.List[PagedUser]: | ||
from apps.alerts.models import AlertGroupLogRecord | ||
|
||
users_ids = set() | ||
for log_record in self.log_records.filter( | ||
user_ids: typing.Set[str] = set() | ||
users: typing.List[PagedUser] = [] | ||
|
||
log_records = self.log_records.filter( | ||
type__in=(AlertGroupLogRecord.TYPE_DIRECT_PAGING, AlertGroupLogRecord.TYPE_UNPAGE_USER) | ||
): | ||
) | ||
|
||
for log_record in log_records: | ||
# filter paging events, track still active escalations | ||
info = log_record.get_step_specific_info() | ||
user_id = info.get("user") if info else None | ||
important = info.get("important") if info else None | ||
|
||
if user_id is not None: | ||
users_ids.add( | ||
user_ids.add( | ||
user_id | ||
) if log_record.type == AlertGroupLogRecord.TYPE_DIRECT_PAGING else users_ids.discard(user_id) | ||
) if log_record.type == AlertGroupLogRecord.TYPE_DIRECT_PAGING else user_ids.discard(user_id) | ||
|
||
user_instances = User.objects.filter(public_primary_key__in=user_ids) | ||
user_map = {u.public_primary_key: u for u in user_instances} | ||
|
||
return User.objects.filter(public_primary_key__in=users_ids) | ||
# mostly doing this second loop to avoid having to query each user individually in the first loop | ||
for log_record in log_records: | ||
# filter paging events, track still active escalations | ||
info = log_record.get_step_specific_info() | ||
user_id = info.get("user") if info else None | ||
important = info.get("important") if info else False | ||
|
||
if user_id is not None and (user := user_map.get(user_id)) is not None: | ||
if log_record.type == AlertGroupLogRecord.TYPE_DIRECT_PAGING: | ||
# add the user | ||
users.append( | ||
{ | ||
"id": user.pk, | ||
"pk": user.public_primary_key, | ||
"name": user.name, | ||
"username": user.username, | ||
"avatar": user.avatar_url, | ||
"avatar_full": user.avatar_full_url, | ||
"important": important, | ||
"teams": [{"pk": t.public_primary_key, "name": t.name} for t in user.teams.all()], | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
} | ||
) | ||
else: | ||
# user was unpaged at some point, remove them | ||
users = [u for u in users if u["pk"] != user_id] | ||
|
||
return users | ||
|
||
def _get_response_time(self): | ||
"""Return response_time based on current alert group status.""" | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -245,6 +245,41 @@ def create(cls, **kwargs): | |
channel.save() | ||
return channel | ||
|
||
@classmethod | ||
def get_orgs_direct_paging_integrations( | ||
joeyorlando marked this conversation as resolved.
Show resolved
Hide resolved
|
||
cls, organization: "Organization" | ||
) -> models.QuerySet["AlertReceiveChannel"]: | ||
return cls.objects.filter( | ||
organization=organization, | ||
integration=AlertReceiveChannel.INTEGRATION_DIRECT_PAGING, | ||
) | ||
|
||
@property | ||
def is_contactable(self) -> bool: | ||
""" | ||
Returns true if: | ||
- the integration has more than one channel filter associated with it | ||
- the default channel filter has at least one notification method specified or an escalation chain associated with it | ||
""" | ||
if self.channel_filters.count() > 1: | ||
return True | ||
|
||
default_channel_filter = self.default_channel_filter | ||
if not default_channel_filter: | ||
return False | ||
|
||
notify_via_slack = self.organization.slack_is_configured and default_channel_filter.notify_in_slack | ||
notify_via_telegram = self.organization.telegram_is_configured and default_channel_filter.notify_in_telegram | ||
|
||
notify_via_chatops = notify_via_slack or notify_via_telegram | ||
custom_messaging_backend_configured = default_channel_filter.notification_backends is not None | ||
|
||
return ( | ||
default_channel_filter.escalation_chain is not None | ||
or notify_via_chatops | ||
or custom_messaging_backend_configured | ||
) | ||
|
||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this is needed to be able to (optionally) filter down teams, in the There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. So this only makes sense for direct paging integrations? Should any other integrations |
||
def delete(self): | ||
self.deleted_at = timezone.now() | ||
self.save() | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
this
important
key here is really the only reason I modified this method to returntyping.List[PagedUser]
instead of the originaltyping.List[User]
(obviously don't have the "important" context in the prior approach)