dcae98b02a
Related to https://github.com/grafana/oncall/issues/96 --------- Co-authored-by: Ravishankar <ravishankar.gnanaprakasam@gmail.com>
47 lines
1.6 KiB
Python
47 lines
1.6 KiB
Python
import enum
|
|
import json
|
|
import logging
|
|
import typing
|
|
|
|
from .insight_logs_enabled_check import is_insight_logs_enabled
|
|
|
|
if typing.TYPE_CHECKING:
|
|
from apps.user_management.models import User
|
|
|
|
insight_logger = logging.getLogger("insight_logger")
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ChatOpsEvent(enum.Enum):
|
|
WORKSPACE_CONNECTED = "workspace_connected"
|
|
WORKSPACE_DISCONNECTED = "workspace_disconnected"
|
|
CHANNEL_CONNECTED = "channel_connected"
|
|
CHANNEL_DISCONNECTED = "channel_disconnected"
|
|
USER_LINKED = "user_linked"
|
|
USER_UNLINKED = "used_unlinked"
|
|
DEFAULT_CHANNEL_CHANGED = "default_channel_changed"
|
|
|
|
|
|
class ChatOpsTypePlug(enum.Enum):
|
|
# ChatOpsTypePlug provides backend_id string for chatops integration not supporting messaging_backends.
|
|
SLACK = "slack"
|
|
TELEGRAM = "telegram"
|
|
MATTERMOST = "mattermost"
|
|
|
|
|
|
def write_chatops_insight_log(author: "User", event_name: ChatOpsEvent, chatops_type: str, **kwargs):
|
|
try:
|
|
organization = author.organization
|
|
|
|
if is_insight_logs_enabled(organization):
|
|
tenant_id = organization.stack_id
|
|
user_id = author.public_primary_key
|
|
username = json.dumps(author.username)
|
|
|
|
log_line = f"tenant_id={tenant_id} author_id={user_id} author={username} action_type=chat_ops action_name={event_name.value} chat_ops_type={chatops_type.lower()}" # noqa
|
|
for k, v in kwargs.items():
|
|
log_line += f" {k}={json.dumps(v)}"
|
|
|
|
insight_logger.info(log_line)
|
|
except Exception as e:
|
|
logger.warning(f"insight_log.failed_to_write_chatops_insight_log exception={e}")
|