Send comprehensive troubleshooting report with root cause and relevant details to Slack channel 'demo'

import json from datetime import datetime from slack_sdk import WebClient from slack_sdk.errors import SlackApiError import logging logging.basicConfig(level=logging.DEBUG) report_header = "OTEL Demo App RCA" slack_channel_name = "demo" # Initialize output variables slack_message_sent = False slack_response_status = "not_attempted" message_content = "" print(f"Preparing comprehensive troubleshooting report for Slack") print(f"Timestamp: {datetime.now()}") try: # Get Slack API token slack_token = getEnvVar('SLACK_API_TOKEN') # Initialize Slack client client = WebClient(token=slack_token) # Format the comprehensive troubleshooting report # Create formatted message message_blocks = [ { "type": "header", "text": { "type": "plain_text", "text": f"{report_header}" } }, { "type": "section", "text": { "type": "mrkdwn", "text": f"*⚠️ Original Problem*\n{problem_statement}" } }, { "type": "section", "text": { "type": "mrkdwn", "text": f"*🔍 Root Cause Identified*\n`{root_cause_identified.replace('_', ' ').title()}`" } } ] # Add root cause evidence if root_cause_evidence: evidence_text = "*📋 Root Cause Evidence*\n" for i, evidence in enumerate(root_cause_evidence[:8], 1): # Limit to 8 items for readability evidence_text += f"• {evidence[:150]}{'...' if len(evidence) > 150 else ''}\n" if len(root_cause_evidence) > 8: evidence_text += f"• ... and {len(root_cause_evidence) - 8} more evidence items\n" message_blocks.append({ "type": "section", "text": { "type": "mrkdwn", "text": evidence_text } }) url = context.url(agent=True) if url.startswith("http://"): url = "https://" + url[len("http://"):] print("Runbook URL: ", url) # Add the buttons message_blocks.extend([ { "type": "divider" }, { "type": "actions", "elements": [ { "type": "button", "text": { "type": "plain_text", "text": "View on DagKnows" }, "url": url }, { "type": "static_select", "action_id": "upvote", "placeholder": { "type": "plain_text", "text": "👍 Upvote" }, "options": [ { "text": { "type": "plain_text", "text": "Forever" }, "value": json.dumps({ "duration": "forever", "runbook_task_id": context.runbook_task_id, "job_id": context.job_id, }) }, { "text": { "type": "plain_text", "text": "3 Months" }, "value": json.dumps({ "duration": "3 months", "runbook_task_id": context.runbook_task_id, "job_id": context.job_id, }) }, { "text": { "type": "plain_text", "text": "6 Months" }, "value": json.dumps({ "duration": "6 months", "runbook_task_id": context.runbook_task_id, "job_id": context.job_id, }) }, { "text": { "type": "plain_text", "text": "1 Year" }, "value": json.dumps({ "duration": "1 year", "runbook_task_id": context.runbook_task_id, "job_id": context.job_id, }) } ] }, { "type": "static_select", "action_id": "downvote_menu", "placeholder": { "type": "plain_text", "text": "👎 Downvote" }, "options": [ { "text": { "type": "plain_text", "text": "Forever" }, "value": json.dumps({ "duration": "forever", "runbook_task_id": context.runbook_task_id, "job_id": context.job_id, }) }, { "text": { "type": "plain_text", "text": "3 Months" }, "value": json.dumps({ "duration": "3 months", "runbook_task_id": context.runbook_task_id, "job_id": context.job_id, }) }, { "text": { "type": "plain_text", "text": "6 Months" }, "value": json.dumps({ "duration": "6 months", "runbook_task_id": context.runbook_task_id, "job_id": context.job_id, }) }, { "text": { "type": "plain_text", "text": "1 Year" }, "value": json.dumps({ "duration": "1 year", "runbook_task_id": context.runbook_task_id, "job_id": context.job_id, }) } ] } ] } ]) # Add footer message_blocks.extend([ { "type": "divider" }, { "type": "context", "elements": [ { "type": "mrkdwn", "text": "🤖 *Automated Troubleshooting Report* | Generated by DagKnows AI Assistant" } ] } ]) # Create fallback text for notifications fallback_text = f"""{report_header} Problem: {problem_statement} Root Cause: {root_cause_identified.replace('_', ' ').title()} Evidence: {len(root_cause_evidence)} items found Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S UTC')}""" # Store message content for output message_content = fallback_text print(f"Sending troubleshooting report to Slack channel '{slack_channel_name}'...") # Send message to Slack response = client.chat_postMessage( channel=slack_channel_name, text=fallback_text, blocks=message_blocks, unfurl_links=False, unfurl_media=False ) if response["ok"]: slack_message_sent = True slack_response_status = "success" print("✅ Troubleshooting report successfully sent to Slack channel 'demo'") print(f"Message timestamp: {response.get('ts', 'unknown')}") else: slack_message_sent = False slack_response_status = f"failed: {response.get('error', 'unknown error')}" print(f"❌ Failed to send message to Slack: {response.get('error', 'unknown error')}") except SlackApiError as e: slack_message_sent = False slack_response_status = f"slack_api_error: {e.response['error']}" error_msg = f"Slack API error: {e.response['error']}" print(f"❌ {error_msg}") print(e.response["error"]) if "response_metadata" in e.response and "messages" in e.response["response_metadata"]: print(e.response["response_metadata"]["messages"]) # Handle specific Slack errors if e.response['error'] == 'channel_not_found': print("💡 The 'demo' channel may not exist or the bot may not have access to it") elif e.response['error'] == 'not_in_channel': print("💡 The bot needs to be invited to the 'demo' channel") elif e.response['error'] == 'invalid_auth': print("💡 Check if the Slack API token is valid and has the required permissions") except Exception as e: slack_message_sent = False slack_response_status = f"error: {str(e)}" error_msg = f"Error sending Slack message: {str(e)}" print(f"❌ {error_msg}") # Print final results print("\n" + "="*60) print("SLACK NOTIFICATION SUMMARY") print("="*60) print(f"Slack Message Sent: {json.dumps(slack_message_sent, indent=2, default=str)}") print(f"Slack Response Status: {json.dumps(slack_response_status, indent=2, default=str)}") print(f"Message Content: {json.dumps(message_content[:500] + '...' if len(message_content) > 500 else message_content, indent=2, default=str)}") if slack_message_sent: print("\n🎉 Troubleshooting report successfully delivered to the team via Slack!") else: print(f"\n⚠️ Failed to send troubleshooting report to Slack: {slack_response_status}")