From 889c76eda23199a6f0e69c2d4e8aa81819abaff5 Mon Sep 17 00:00:00 2001 From: Amrit-thapaliya Date: Sat, 16 May 2026 18:13:58 +0545 Subject: [PATCH 1/5] feat: add OpenSearch and Splunk On-Call (VictorOps) connectors Adds two new connectors to Aurora: **OpenSearch** (log analytics for RCA) - Backend connector client with cluster info, health check, search, and index listing (server/connectors/opensearch_connector/) - REST routes for connect/disconnect/status/search/indices with full RBAC enforcement (server/routes/opensearch/) - LangChain tools: search_opensearch, list_opensearch_indices (server/chat/backend/agent/tools/opensearch_tool.py) - Agent skill with RCA workflow guidance (server/chat/backend/agent/skills/integrations/opensearch/SKILL.md) - Frontend auth page and API proxy (client/src/app/opensearch/, client/src/app/api/opensearch/) - ConnectorRegistry entry, service helpers **Splunk On-Call / VictorOps** (real-time incident alerting + RCA) - VictorOps REST API client with credential validation (server/routes/victorops/victorops_helpers.py) - Routes for connect/disconnect/status/webhook with RBAC (server/routes/victorops/victorops_routes.py) - Celery task for async webhook processing (server/routes/victorops/tasks.py) - LangChain tools: get_victorops_incidents, get_victorops_teams (server/chat/backend/agent/tools/victorops_tool.py) - Agent skill with RCA workflow guidance (server/chat/backend/agent/skills/integrations/victorops/SKILL.md) - Frontend: connection wizard (API ID + Key), webhook setup step, connected view, and auth page (client/src/app/victorops/, client/src/components/victorops/) - ConnectorRegistry entry, service helpers **Shared changes** - Added 'opensearch' and 'victorops' to CONNECTOR_DIRS in providers.py - Registered both blueprints and CORS rules in main_compute.py - Extended connector_status.py, cloud_tools.py, rca_prompt_builder.py, secret_ref_utils.py with new connector support Co-authored-by: Cursor --- client/public/opensearch.svg | 11 + client/public/victorops.svg | 8 + .../connected-accounts/[provider]/route.ts | 42 +- .../src/app/api/opensearch/[...path]/route.ts | 71 +++ client/src/app/api/victorops/route.ts | 79 ++++ .../app/api/victorops/webhook-url/route.ts | 37 ++ client/src/app/opensearch/auth/page.tsx | 292 ++++++++++++ client/src/app/victorops/auth/page.tsx | 156 +++++++ .../connectors/AtlassianConnectPage.tsx | 26 +- .../connectors/ConnectorRegistry.ts | 22 + .../victorops/VictorOpsConnectedView.tsx | 65 +++ .../victorops/VictorOpsConnectionStep.tsx | 138 ++++++ .../victorops/VictorOpsWebhookStep.tsx | 209 +++++++++ client/src/lib/services/opensearch.ts | 83 ++++ client/src/lib/services/victorops.ts | 42 ++ server/celery_config.py | 7 + .../skills/integrations/opensearch/SKILL.md | 62 +++ .../skills/integrations/victorops/SKILL.md | 41 ++ .../chat/backend/agent/tools/cloud_tools.py | 79 ++++ .../backend/agent/tools/opensearch_tool.py | 158 +++++++ .../backend/agent/tools/victorops_tool.py | 110 +++++ server/chat/background/rca_prompt_builder.py | 43 ++ .../connectors/confluence_connector/client.py | 13 +- .../opensearch_connector/__init__.py | 0 .../connectors/opensearch_connector/client.py | 166 +++++++ server/main_compute.py | 17 + server/routes/atlassian/atlassian_routes.py | 12 +- server/routes/connector_status.py | 20 + server/routes/opensearch/__init__.py | 3 + server/routes/opensearch/opensearch_routes.py | 229 ++++++++++ server/routes/victorops/__init__.py | 1 + server/routes/victorops/tasks.py | 432 ++++++++++++++++++ server/routes/victorops/victorops_helpers.py | 128 ++++++ server/routes/victorops/victorops_routes.py | 194 ++++++++ server/utils/db/db_utils.py | 25 +- server/utils/providers.py | 2 + server/utils/secrets/secret_ref_utils.py | 2 + 37 files changed, 3015 insertions(+), 10 deletions(-) create mode 100644 client/public/opensearch.svg create mode 100644 client/public/victorops.svg create mode 100644 client/src/app/api/opensearch/[...path]/route.ts create mode 100644 client/src/app/api/victorops/route.ts create mode 100644 client/src/app/api/victorops/webhook-url/route.ts create mode 100644 client/src/app/opensearch/auth/page.tsx create mode 100644 client/src/app/victorops/auth/page.tsx create mode 100644 client/src/components/victorops/VictorOpsConnectedView.tsx create mode 100644 client/src/components/victorops/VictorOpsConnectionStep.tsx create mode 100644 client/src/components/victorops/VictorOpsWebhookStep.tsx create mode 100644 client/src/lib/services/opensearch.ts create mode 100644 client/src/lib/services/victorops.ts create mode 100644 server/chat/backend/agent/skills/integrations/opensearch/SKILL.md create mode 100644 server/chat/backend/agent/skills/integrations/victorops/SKILL.md create mode 100644 server/chat/backend/agent/tools/opensearch_tool.py create mode 100644 server/chat/backend/agent/tools/victorops_tool.py create mode 100644 server/connectors/opensearch_connector/__init__.py create mode 100644 server/connectors/opensearch_connector/client.py create mode 100644 server/routes/opensearch/__init__.py create mode 100644 server/routes/opensearch/opensearch_routes.py create mode 100644 server/routes/victorops/__init__.py create mode 100644 server/routes/victorops/tasks.py create mode 100644 server/routes/victorops/victorops_helpers.py create mode 100644 server/routes/victorops/victorops_routes.py diff --git a/client/public/opensearch.svg b/client/public/opensearch.svg new file mode 100644 index 000000000..b942df659 --- /dev/null +++ b/client/public/opensearch.svg @@ -0,0 +1,11 @@ + + + + + + + + + + + diff --git a/client/public/victorops.svg b/client/public/victorops.svg new file mode 100644 index 000000000..6c86cb2af --- /dev/null +++ b/client/public/victorops.svg @@ -0,0 +1,8 @@ + + + + + + + + diff --git a/client/src/app/api/connected-accounts/[provider]/route.ts b/client/src/app/api/connected-accounts/[provider]/route.ts index b0eb29ace..9c449ce52 100644 --- a/client/src/app/api/connected-accounts/[provider]/route.ts +++ b/client/src/app/api/connected-accounts/[provider]/route.ts @@ -25,7 +25,7 @@ export async function DELETE( const { provider } = await context.params // Validate provider - if (!['gcp', 'azure', 'aws', 'github', 'gitlab', 'grafana', 'datadog', 'netdata', 'ovh', 'scaleway', 'tailscale', 'slack', 'google_chat', 'splunk', 'dynatrace', 'confluence', 'jira', 'sharepoint', 'coroot', 'thousandeyes', 'jenkins', 'cloudbees', 'bigpanda', 'spinnaker', 'newrelic', 'opsgenie', 'incidentio', 'sentry'].includes(provider)) { + if (!['gcp', 'azure', 'aws', 'github', 'gitlab', 'grafana', 'datadog', 'netdata', 'ovh', 'scaleway', 'tailscale', 'slack', 'google_chat', 'splunk', 'dynatrace', 'confluence', 'jira', 'sharepoint', 'coroot', 'thousandeyes', 'jenkins', 'cloudbees', 'bigpanda', 'spinnaker', 'newrelic', 'opsgenie', 'incidentio', 'sentry', 'victorops', 'opensearch'].includes(provider)) { return NextResponse.json( { error: 'Invalid provider' }, { status: 400 } @@ -508,6 +508,46 @@ export async function DELETE( return NextResponse.json(data) } + // Special handling for OpenSearch + if (provider === 'opensearch') { + const response = await fetch(`${API_BASE_URL}/opensearch/disconnect`, { + method: 'DELETE', + headers: authHeaders, + }) + + if (!response.ok) { + const errorText = await response.text() + console.error('Backend error disconnecting OpenSearch:', errorText) + return NextResponse.json( + { error: 'Failed to disconnect OpenSearch' }, + { status: response.status } + ) + } + + const data = await response.json() + return NextResponse.json(data) + } + + // Special handling for VictorOps (Splunk On-Call) + if (provider === 'victorops') { + const response = await fetch(`${API_BASE_URL}/victorops`, { + method: 'DELETE', + headers: authHeaders, + }) + + if (!response.ok) { + const errorText = await response.text() + console.error('Backend error disconnecting VictorOps:', errorText) + return NextResponse.json( + { error: 'Failed to disconnect Splunk On-Call' }, + { status: response.status } + ) + } + + const data = await response.json() + return NextResponse.json(data) + } + // For other providers, use the general disconnect endpoint const response = await fetch(`${API_BASE_URL}/api/connected-accounts/${userId}/${provider}`, { method: 'DELETE', diff --git a/client/src/app/api/opensearch/[...path]/route.ts b/client/src/app/api/opensearch/[...path]/route.ts new file mode 100644 index 000000000..81d451dae --- /dev/null +++ b/client/src/app/api/opensearch/[...path]/route.ts @@ -0,0 +1,71 @@ +import { NextRequest, NextResponse } from 'next/server'; +import { getAuthenticatedUser } from '@/lib/auth-helper'; + +const API_BASE_URL = process.env.BACKEND_URL; + +async function handleRequest( + request: NextRequest, + { params }: { params: Promise<{ path: string[] }> }, + method: string, +) { + try { + const authResult = await getAuthenticatedUser(); + if (authResult instanceof NextResponse) return authResult; + + const { headers: authHeaders } = authResult; + const { path } = await params; + const backendPath = path.join('/'); + const qs = request.nextUrl.searchParams.toString(); + const url = qs + ? `${API_BASE_URL}/opensearch/${backendPath}?${qs}` + : `${API_BASE_URL}/opensearch/${backendPath}`; + + const options: RequestInit = { + method, + headers: authHeaders, + credentials: 'include', + }; + + if ((method === 'POST' || method === 'PUT') && request.body) { + const payload = await request.json(); + options.headers = { ...authHeaders, 'Content-Type': 'application/json' }; + options.body = JSON.stringify(payload); + } + + if (method === 'GET') options.cache = 'no-store'; + + const response = await fetch(url, options); + + if (!response.ok) { + let errorMessage = 'OpenSearch API request failed'; + try { + const text = await response.text(); + if (text) { + try { + const errorData = JSON.parse(text); + errorMessage = errorData.error || errorData.message || errorMessage; + } catch { + if (text.length < 200) errorMessage = text; + } + } + } catch { /* fall back */ } + return NextResponse.json({ error: errorMessage }, { status: response.status }); + } + + const data = await response.json(); + return NextResponse.json(data); + } catch (error) { + console.error('[api/opensearch] Error:', error); + return NextResponse.json({ error: 'OpenSearch API request failed' }, { status: 500 }); + } +} + +export async function GET(request: NextRequest, ctx: { params: Promise<{ path: string[] }> }) { + return handleRequest(request, ctx, 'GET'); +} +export async function POST(request: NextRequest, ctx: { params: Promise<{ path: string[] }> }) { + return handleRequest(request, ctx, 'POST'); +} +export async function DELETE(request: NextRequest, ctx: { params: Promise<{ path: string[] }> }) { + return handleRequest(request, ctx, 'DELETE'); +} diff --git a/client/src/app/api/victorops/route.ts b/client/src/app/api/victorops/route.ts new file mode 100644 index 000000000..3d0767ba1 --- /dev/null +++ b/client/src/app/api/victorops/route.ts @@ -0,0 +1,79 @@ +import { NextRequest, NextResponse } from 'next/server'; +import { getAuthenticatedUser } from '@/lib/auth-helper'; + +const API_BASE_URL = process.env.BACKEND_URL; + +async function handleRequest(request: NextRequest, method: string) { + try { + const authResult = await getAuthenticatedUser(); + + if (authResult instanceof NextResponse) { + return authResult; + } + + const { headers: authHeaders } = authResult; + + const options: RequestInit = { + method, + headers: authHeaders, + credentials: 'include', + }; + + if ((method === 'POST' || method === 'PATCH') && request.body) { + const payload = await request.json(); + options.headers = { + ...authHeaders, + 'Content-Type': 'application/json', + }; + options.body = JSON.stringify(payload); + } + + if (method === 'GET') { + options.cache = 'no-store'; + } + + const response = await fetch(`${API_BASE_URL}/victorops`, options); + + if (!response.ok) { + let errorMessage = 'Splunk On-Call API request failed'; + try { + const text = await response.text(); + if (text) { + try { + const errorData = JSON.parse(text); + errorMessage = errorData.message || errorData.error || errorMessage; + } catch { + if (text.length < 200) errorMessage = text; + } + } + } catch { + // fall back to default + } + return NextResponse.json( + { error: errorMessage, message: errorMessage }, + { status: response.status } + ); + } + + const data = await response.json(); + return NextResponse.json(data); + } catch (error) { + console.error('[api/victorops] Error:', error); + return NextResponse.json( + { error: 'Splunk On-Call API request failed' }, + { status: 500 } + ); + } +} + +export async function GET(request: NextRequest) { + return handleRequest(request, 'GET'); +} + +export async function POST(request: NextRequest) { + return handleRequest(request, 'POST'); +} + +export async function DELETE(request: NextRequest) { + return handleRequest(request, 'DELETE'); +} diff --git a/client/src/app/api/victorops/webhook-url/route.ts b/client/src/app/api/victorops/webhook-url/route.ts new file mode 100644 index 000000000..2e9465566 --- /dev/null +++ b/client/src/app/api/victorops/webhook-url/route.ts @@ -0,0 +1,37 @@ +import { NextRequest, NextResponse } from 'next/server'; +import { getAuthenticatedUser } from '@/lib/auth-helper'; + +const BACKEND_URL = process.env.BACKEND_URL; + +export async function GET(request: NextRequest) { + try { + const authResult = await getAuthenticatedUser(); + + if (authResult instanceof NextResponse) { + return authResult; + } + + const { headers: authHeaders } = authResult; + + const response = await fetch(`${BACKEND_URL}/victorops/webhook-url`, { + method: 'GET', + headers: authHeaders, + credentials: 'include', + cache: 'no-store', + }); + + if (!response.ok) { + const text = await response.text(); + return NextResponse.json( + { error: text || 'Failed to load webhook URL' }, + { status: response.status } + ); + } + + const data = await response.json(); + return NextResponse.json(data); + } catch (error) { + console.error('[victorops/webhook-url] Error:', error); + return NextResponse.json({ error: 'Internal server error' }, { status: 500 }); + } +} diff --git a/client/src/app/opensearch/auth/page.tsx b/client/src/app/opensearch/auth/page.tsx new file mode 100644 index 000000000..85c0a155f --- /dev/null +++ b/client/src/app/opensearch/auth/page.tsx @@ -0,0 +1,292 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { useToast } from "@/hooks/use-toast"; +import { openSearchService, OpenSearchStatus } from "@/lib/services/opensearch"; +import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; +import { Label } from "@/components/ui/label"; +import { Card, CardContent, CardDescription, CardHeader, CardTitle, CardFooter } from "@/components/ui/card"; +import { Loader2, CheckCircle, ExternalLink, Database } from "lucide-react"; +import ConnectorAuthGuard from "@/components/connectors/ConnectorAuthGuard"; + +const CACHE_KEY = "opensearch_connection_status"; + +export default function OpenSearchAuthPage() { + const { toast } = useToast(); + const [endpoint, setEndpoint] = useState(""); + const [username, setUsername] = useState(""); + const [password, setPassword] = useState(""); + const [indexPattern, setIndexPattern] = useState("*"); + const [status, setStatus] = useState(null); + const [loading, setLoading] = useState(false); + const [isCheckingStatus, setIsCheckingStatus] = useState(true); + + const loadStatus = async () => { + try { + if (typeof window !== "undefined") { + const cached = localStorage.getItem(CACHE_KEY); + if (cached) { + const parsed = JSON.parse(cached); + setStatus(parsed); + if (parsed?.connected) setEndpoint(parsed.endpoint ?? ""); + } + } + const result = await openSearchService.getStatus(); + if (result !== null) { + setStatus(result); + if (typeof window !== "undefined") { + localStorage.setItem(CACHE_KEY, JSON.stringify(result)); + if (result.connected) setEndpoint(result.endpoint ?? ""); + } + } + } catch (err) { + console.error("[OpenSearch] Failed to load status", err); + } finally { + setIsCheckingStatus(false); + } + }; + + useEffect(() => { loadStatus(); }, []); + + const handleConnect = async (e: React.FormEvent) => { + e.preventDefault(); + setLoading(true); + try { + const result = await openSearchService.connect({ + endpoint, + username, + password, + indexPattern: indexPattern || "*", + }); + setStatus(result); + if (typeof window !== "undefined") { + localStorage.setItem(CACHE_KEY, JSON.stringify(result)); + localStorage.setItem("isOpenSearchConnected", "true"); + window.dispatchEvent(new CustomEvent("providerStateChanged")); + } + toast({ title: "OpenSearch connected", description: `Cluster: ${result.clusterName ?? endpoint}` }); + setPassword(""); + } catch (err: unknown) { + const msg = err instanceof Error ? err.message : "Connection failed"; + toast({ title: "Failed to connect", description: msg, variant: "destructive" }); + } finally { + setLoading(false); + } + }; + + const handleDisconnect = async () => { + setLoading(true); + try { + const response = await fetch("/api/opensearch/disconnect", { + method: "DELETE", + credentials: "include", + }); + if (response.ok || response.status === 204) { + setStatus({ connected: false }); + setEndpoint(""); + setUsername(""); + setPassword(""); + if (typeof window !== "undefined") { + localStorage.removeItem(CACHE_KEY); + localStorage.removeItem("isOpenSearchConnected"); + window.dispatchEvent(new CustomEvent("providerStateChanged")); + } + toast({ title: "OpenSearch disconnected" }); + } else { + throw new Error("Failed to disconnect"); + } + } catch (err: unknown) { + toast({ title: "Failed to disconnect", description: err instanceof Error ? err.message : "Unknown error", variant: "destructive" }); + } finally { + setLoading(false); + } + }; + + if (isCheckingStatus) { + return ( + +
+
+

OpenSearch Integration

+

Connect your OpenSearch cluster for log search during RCA

+
+ + + + + +
+ + ); + } + + return ( + +
+
+
+ +
+
+

OpenSearch

+

Search logs and traces during incident RCA

+
+
+ + {status?.connected ? ( + + +
+
+
+ +
+
+ Connected + {status.endpoint} +
+
+ + Basic Auth + +
+ + +
+ {status.clusterName && ( +
+

Cluster

+

{status.clusterName}

+
+ )} + {status.version && ( +
+

Version

+

{status.version}

+
+ )} + {status.indexPattern && ( +
+

Index Pattern

+

{status.indexPattern}

+
+ )} +
+ + + + + + ) : ( + + + Connect to OpenSearch + + Enter your cluster endpoint and credentials. Aurora uses Basic authentication. + + + +
+
+ + setEndpoint(e.target.value)} + required + /> +

Include the full URL with port if non-standard

+
+ +
+
+ + setUsername(e.target.value)} + required + /> +
+
+ + setPassword(e.target.value)} + required + /> +
+
+ +
+ + setIndexPattern(e.target.value)} + /> +

+ Wildcards supported, e.g. logs-* or * for all +

+
+ +
+

What Aurora uses this for:

+
    +
  • Search logs by keyword or service name during RCA
    • +
  • Find error traces within the incident time window
    • +
  • Correlate log patterns with triggered alerts
    • +
+
+ + +
+ + + )} + + + + OpenSearch REST API Docs + + + + View OpenSearch API reference + + + +
+ + ); +} diff --git a/client/src/app/victorops/auth/page.tsx b/client/src/app/victorops/auth/page.tsx new file mode 100644 index 000000000..a528a50f8 --- /dev/null +++ b/client/src/app/victorops/auth/page.tsx @@ -0,0 +1,156 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { useToast } from "@/hooks/use-toast"; +import { victoropsService, VictorOpsStatus } from "@/lib/services/victorops"; +import { VictorOpsConnectionStep } from "@/components/victorops/VictorOpsConnectionStep"; +import { VictorOpsConnectedView } from "@/components/victorops/VictorOpsConnectedView"; +import { VictorOpsWebhookStep } from "@/components/victorops/VictorOpsWebhookStep"; +import { ConnectionLoadingOverlay } from "@/components/ui/connection-loading-overlay"; +import { DisconnectConfirmDialog } from "@/components/ui/disconnect-confirm-dialog"; +import { getUserFriendlyError } from "@/lib/utils"; +import ConnectorAuthGuard from "@/components/connectors/ConnectorAuthGuard"; + +const CACHE_KEY = 'victorops_connection_status'; + +export default function VictorOpsAuthPage() { + const { toast } = useToast(); + const [displayName, setDisplayName] = useState("Splunk On-Call"); + const [apiId, setApiId] = useState(""); + const [apiKey, setApiKey] = useState(""); + const [status, setStatus] = useState(null); + const [loading, setLoading] = useState(false); + const [isConnecting, setIsConnecting] = useState(false); + const [error, setError] = useState(null); + const [showDisconnectDialog, setShowDisconnectDialog] = useState(false); + + const updateLocalStorage = (connected: boolean) => { + if (typeof window === 'undefined') return; + if (connected) { + localStorage.setItem('isVictorOpsConnected', 'true'); + } else { + localStorage.removeItem('isVictorOpsConnected'); + } + window.dispatchEvent(new CustomEvent('providerStateChanged')); + }; + + const fetchAndUpdateStatus = async () => { + const result = await victoropsService.getStatus(); + setStatus(result); + if (typeof window !== 'undefined' && result) { + localStorage.setItem(CACHE_KEY, JSON.stringify(result)); + } + updateLocalStorage(result?.connected ?? false); + }; + + const loadStatus = async (skipCache = false) => { + try { + if (!skipCache && typeof window !== 'undefined') { + const cached = localStorage.getItem(CACHE_KEY); + if (cached) setStatus(JSON.parse(cached) as VictorOpsStatus); + } + await fetchAndUpdateStatus(); + } catch { + toast({ + title: 'Error', + description: 'Unable to load Splunk On-Call status', + variant: 'destructive', + }); + } + }; + + useEffect(() => { + loadStatus(); + // eslint-disable-next-line react-hooks/exhaustive-deps + }, []); + + const handleConnect = async (e: React.FormEvent) => { + e.preventDefault(); + setError(null); + setIsConnecting(true); + + try { + const result = await victoropsService.connect(apiId, apiKey, displayName); + if (result.connected) { + setStatus(result); + updateLocalStorage(true); + toast({ + title: 'Connected', + description: `Splunk On-Call connected successfully.`, + }); + setApiId(""); + setApiKey(""); + } + } catch (err: unknown) { + const message = getUserFriendlyError(err); + setError(message); + toast({ title: 'Connection Failed', description: message, variant: 'destructive' }); + } finally { + setIsConnecting(false); + } + }; + + const handleDisconnect = async () => { + setLoading(true); + try { + await victoropsService.disconnect(); + setStatus(null); + updateLocalStorage(false); + if (typeof window !== 'undefined') localStorage.removeItem(CACHE_KEY); + toast({ title: 'Disconnected', description: 'Splunk On-Call disconnected.' }); + } catch (err: unknown) { + const message = getUserFriendlyError(err); + toast({ title: 'Error', description: message, variant: 'destructive' }); + } finally { + setLoading(false); + setShowDisconnectDialog(false); + } + }; + + return ( + +
+
+

Splunk On-Call

+

+ Connect Splunk On-Call to receive real-time incident alerts and trigger automated RCA. +

+
+ + {isConnecting && ( + + )} + + {status?.connected ? ( + <> + setShowDisconnectDialog(true)} + disconnecting={loading} + /> + + + ) : ( + + )} + + { handleDisconnect(); }} + connectorName="Splunk On-Call" + /> +
+ + ); +} diff --git a/client/src/components/connectors/AtlassianConnectPage.tsx b/client/src/components/connectors/AtlassianConnectPage.tsx index 1aff526be..620c4a6b6 100644 --- a/client/src/components/connectors/AtlassianConnectPage.tsx +++ b/client/src/components/connectors/AtlassianConnectPage.tsx @@ -44,6 +44,7 @@ export function AtlassianConnectPage({ product, sibling }: AtlassianConnectPageP const [isDisconnecting, setIsDisconnecting] = useState(false); const [alsoConnectSibling, setAlsoConnectSibling] = useState(false); const [patUrl, setPatUrl] = useState(""); + const [patEmail, setPatEmail] = useState(""); const [patToken, setPatToken] = useState(""); const [isPatConnecting, setIsPatConnecting] = useState(false); const [jiraMode, setJiraMode] = useState<"full" | "comment_only">("comment_only"); @@ -127,14 +128,18 @@ export function AtlassianConnectPage({ product, sibling }: AtlassianConnectPageP } finally { setIsConnecting(false); } }; + const isCloudUrl = (url: string) => url.toLowerCase().includes(".atlassian.net"); + const handlePatConnect = async (e: React.FormEvent) => { e.preventDefault(); if (!patUrl || !patToken) return; + if (isCloudUrl(patUrl) && !patEmail) return; setIsPatConnecting(true); try { const payload: Record = { products: [product.key], authType: "pat" as const }; payload[`${product.key}BaseUrl`] = patUrl; payload[`${product.key}PatToken`] = patToken; + if (patEmail) payload[`${product.key}Email`] = patEmail; await atlassianService.connect(payload as unknown as Parameters[0]); await loadStatus(); toast({ title: `${product.name} connected via PAT` }); @@ -405,9 +410,26 @@ export function AtlassianConnectPage({ product, sibling }: AtlassianConnectPageP setPatUrl(e.target.value)} required className="h-9" /> + {isCloudUrl(patUrl) && ( +
+ + setPatEmail(e.target.value)} required className="h-9" /> +

Required for Atlassian Cloud API tokens

+
+ )}
- - setPatToken(e.target.value)} required className="h-9" /> + + setPatToken(e.target.value)} required className="h-9" /> + {isCloudUrl(patUrl) && ( +

+ Generate at{" "} + + id.atlassian.com → API tokens + +

+ )}
+ + + {status.externalUserName && ( +
+ Account + {status.externalUserName} +
+ )} + {status.validatedAt && ( +
+ Last validated + + {new Date(status.validatedAt).toLocaleString()} + +
+ )} + + + ); +} diff --git a/client/src/components/victorops/VictorOpsConnectionStep.tsx b/client/src/components/victorops/VictorOpsConnectionStep.tsx new file mode 100644 index 000000000..f73e86e04 --- /dev/null +++ b/client/src/components/victorops/VictorOpsConnectionStep.tsx @@ -0,0 +1,138 @@ +"use client"; + +import { + Card, + CardContent, + CardDescription, + CardHeader, + CardTitle, +} from "@/components/ui/card"; +import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; +import { Label } from "@/components/ui/label"; +import { Alert, AlertDescription } from "@/components/ui/alert"; +import { AlertCircle } from "lucide-react"; + +interface VictorOpsConnectionStepProps { + displayName: string; + setDisplayName: (value: string) => void; + apiId: string; + setApiId: (value: string) => void; + apiKey: string; + setApiKey: (value: string) => void; + loading: boolean; + error: string | null; + onConnect: (e: React.FormEvent) => void; +} + +export function VictorOpsConnectionStep({ + displayName, + setDisplayName, + apiId, + setApiId, + apiKey, + setApiKey, + loading, + error, + onConnect, +}: VictorOpsConnectionStepProps) { + return ( + + + Authentication + + Connect with your Splunk On-Call API ID and API Key + + + + {error && ( + + + {error} + + )} + +
+
+ + setDisplayName(e.target.value)} + disabled={loading} + /> +
+ +
+ + setApiId(e.target.value)} + required + disabled={loading} + /> +
+ +
+ + setApiKey(e.target.value)} + required + disabled={loading} + /> + +
+

+ How to get your API credentials +

+
    +
  1. Log in to your Splunk On-Call portal
    2. +
  2. + Go to{" "} + + Integrations → API + +
    3. +
  3. + Copy your{" "} + API ID and click{" "} + Create API Key +
    4. +
  4. Paste both values above
    5. +
+ + View API documentation → + +
+
+ + +
+ + + ); +} diff --git a/client/src/components/victorops/VictorOpsWebhookStep.tsx b/client/src/components/victorops/VictorOpsWebhookStep.tsx new file mode 100644 index 000000000..8f3867126 --- /dev/null +++ b/client/src/components/victorops/VictorOpsWebhookStep.tsx @@ -0,0 +1,209 @@ +"use client"; + +import { useState, useEffect, useCallback, useRef } from "react"; +import { + Card, + CardContent, + CardDescription, + CardHeader, + CardTitle, +} from "@/components/ui/card"; +import { Button } from "@/components/ui/button"; +import { Badge } from "@/components/ui/badge"; +import { Switch } from "@/components/ui/switch"; +import { Label } from "@/components/ui/label"; +import { Copy, Check, ExternalLink } from "lucide-react"; +import { useToast } from "@/hooks/use-toast"; +import { copyToClipboard } from "@/lib/utils"; + +export function VictorOpsWebhookStep() { + const { toast } = useToast(); + const [webhookUrl, setWebhookUrl] = useState(null); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + const [copied, setCopied] = useState(false); + const [rcaEnabled, setRcaEnabled] = useState(true); + const [rcaLoading, setRcaLoading] = useState(false); + const rcaToggleInProgress = useRef(false); + + useEffect(() => { + loadWebhook(); + loadRcaPreference(); + }, []); + + const loadRcaPreference = async () => { + try { + const response = await fetch('/api/user-preferences?key=automated_rca_enabled'); + if (response.ok) { + const data = await response.json(); + setRcaEnabled(data.value !== false); + } + } catch { + setRcaEnabled(true); + } + }; + + const loadWebhook = async () => { + try { + setLoading(true); + const response = await fetch('/api/victorops/webhook-url'); + if (!response.ok) throw new Error('Failed to load webhook URL'); + const data = await response.json(); + setWebhookUrl(data.webhookUrl); + } catch (err) { + setError(err instanceof Error ? err.message : 'Failed to load webhook'); + } finally { + setLoading(false); + } + }; + + const handleCopy = async (text: string) => { + try { + await copyToClipboard(text); + setCopied(true); + setTimeout(() => setCopied(false), 2000); + } catch { + toast({ + title: 'Copy failed', + description: 'Please copy the URL manually.', + variant: 'destructive', + }); + } + }; + + const handleRcaToggle = useCallback(async (checked: boolean) => { + if (rcaToggleInProgress.current || checked === rcaEnabled) return; + + rcaToggleInProgress.current = true; + const previousValue = rcaEnabled; + setRcaEnabled(checked); + setRcaLoading(true); + + try { + const response = await fetch('/api/user-preferences', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ key: 'automated_rca_enabled', value: checked }), + }); + + if (response.ok) { + toast({ + title: checked ? 'Automated RCA Enabled' : 'Automated RCA Disabled', + description: checked + ? 'Aurora will automatically analyze new Splunk On-Call incidents' + : 'Automated RCA has been disabled', + }); + } else { + throw new Error('Failed to update preference'); + } + } catch { + setRcaEnabled(previousValue); + toast({ + title: 'Error', + description: 'Failed to update automated RCA setting', + variant: 'destructive', + }); + } finally { + setRcaLoading(false); + rcaToggleInProgress.current = false; + } + }, [rcaEnabled, toast]); + + if (loading) return ( + + + Loading webhook… + + + ); + + if (error) return ( + + {error} + + ); + + if (!webhookUrl) return null; + + return ( + + + Webhook Configuration + + Configure Splunk On-Call to forward incidents to Aurora + + + +
+
+

Webhook URL

+ Per user +
+
+ + {webhookUrl} + + +
+
+ +
+

Setup Instructions:

+
    +
  1. + Log in to your Splunk On-Call portal and go to{" "} + Integrations → Outgoing Webhooks +
    2. +
  2. Click Add Webhook
    3. +
  3. + Set Event Type to{" "} + Any-Incident +
    4. +
  4. Paste the webhook URL above and save
    5. +
  5. Trigger a test alert to verify the connection
    6. +
+ + Splunk On-Call Webhook Docs + +
+ +
+

Automation Settings

+
+
+
+ +

+ Automatically investigate new incidents with AI-powered RCA. +

+
+ +
+
+
+ + + ); +} diff --git a/client/src/lib/services/opensearch.ts b/client/src/lib/services/opensearch.ts new file mode 100644 index 000000000..748c9065d --- /dev/null +++ b/client/src/lib/services/opensearch.ts @@ -0,0 +1,83 @@ +import { apiRequest } from '@/lib/services/api-client'; + +export interface OpenSearchStatus { + connected: boolean; + clusterName?: string; + version?: string; + endpoint?: string; + indexPattern?: string; + error?: string; +} + +export interface OpenSearchConnectPayload { + endpoint: string; + username: string; + password: string; + indexPattern?: string; + verifySsl?: boolean; + maxRetries?: number; +} + +export interface OpenSearchSearchPayload { + query: string; + index?: string; + startTime?: string; + endTime?: string; + size?: number; + timestampField?: string; +} + +export interface OpenSearchSearchResult { + total: number; + hits: Record[]; + index: string; + query: string; +} + +const API_BASE = '/api/opensearch'; + +export const openSearchService = { + async getStatus(): Promise { + try { + const raw = await apiRequest>(`${API_BASE}/status`, { + cache: 'no-store', + }); + if (!raw) return null; + return { + connected: Boolean(raw.connected), + clusterName: raw.clusterName as string | undefined, + version: raw.version as string | undefined, + endpoint: raw.endpoint as string | undefined, + indexPattern: raw.indexPattern as string | undefined, + error: raw.error as string | undefined, + }; + } catch (error) { + console.error('[openSearchService] Failed to fetch status:', error); + return null; + } + }, + + async connect(payload: OpenSearchConnectPayload): Promise { + const raw = await apiRequest>(`${API_BASE}/connect`, { + method: 'POST', + body: JSON.stringify(payload), + cache: 'no-store', + }); + return { + connected: Boolean(raw?.success), + clusterName: raw?.clusterName as string | undefined, + version: raw?.version as string | undefined, + endpoint: (raw?.endpoint ?? payload.endpoint) as string, + indexPattern: (raw?.indexPattern ?? payload.indexPattern ?? '*') as string, + }; + }, + + async search(payload: OpenSearchSearchPayload): Promise { + const raw = await apiRequest(`${API_BASE}/search`, { + method: 'POST', + body: JSON.stringify(payload), + cache: 'no-store', + }); + return raw ?? { total: 0, hits: [], index: '*', query: payload.query }; + }, +}; diff --git a/client/src/lib/services/victorops.ts b/client/src/lib/services/victorops.ts new file mode 100644 index 000000000..b09e6f560 --- /dev/null +++ b/client/src/lib/services/victorops.ts @@ -0,0 +1,42 @@ +'use client'; + +import { apiRequest } from '@/lib/services/api-client'; + +export interface VictorOpsStatus { + connected: boolean; + displayName?: string; + externalUserName?: string; + accountName?: string; + validatedAt?: string; + capabilities?: { + can_read_incidents: boolean; + }; +} + +const API_BASE = '/api/victorops'; + +export const victoropsService = { + async getStatus(): Promise { + try { + return await apiRequest(API_BASE, { cache: 'no-store' }); + } catch { + return null; + } + }, + + async connect( + apiId: string, + apiKey: string, + displayName = 'Splunk On-Call' + ): Promise { + return apiRequest(API_BASE, { + method: 'POST', + body: JSON.stringify({ apiId, apiKey, displayName }), + cache: 'no-store', + }); + }, + + async disconnect(): Promise { + await apiRequest(API_BASE, { method: 'DELETE', cache: 'no-store' }); + }, +}; diff --git a/server/celery_config.py b/server/celery_config.py index d1a7ba208..7b13a9a23 100644 --- a/server/celery_config.py +++ b/server/celery_config.py @@ -86,6 +86,7 @@ 'routes.dynatrace.tasks', 'routes.bigpanda.tasks', 'routes.pagerduty.tasks', + 'routes.victorops.tasks', 'routes.opsgenie.tasks', 'routes.newrelic.tasks', 'routes.sentry.tasks', @@ -180,6 +181,12 @@ except ImportError as e: logging.warning(f"Failed to import PagerDuty tasks: {e}") +try: + import routes.victorops.tasks # noqa: F401 + logging.info("Splunk On-Call (VictorOps) tasks imported successfully") +except ImportError as e: + logging.warning(f"Failed to import VictorOps tasks: {e}") + try: import routes.opsgenie.tasks # noqa: F401 logging.info("OpsGenie tasks imported successfully") diff --git a/server/chat/backend/agent/skills/integrations/opensearch/SKILL.md b/server/chat/backend/agent/skills/integrations/opensearch/SKILL.md new file mode 100644 index 000000000..8142e1c6b --- /dev/null +++ b/server/chat/backend/agent/skills/integrations/opensearch/SKILL.md @@ -0,0 +1,62 @@ +--- +name: opensearch +id: opensearch +description: "OpenSearch log search integration for querying logs by keyword, service, time range, and error pattern during RCA investigations" +category: observability +connection_check: + method: is_connected_function + module: chat.backend.agent.tools.opensearch_tool + function: is_opensearch_connected +tools: + - search_opensearch + - list_opensearch_indices +index: "Log analytics -- search OpenSearch logs by query, time range, and index pattern" +rca_priority: 3 +allowed-tools: search_opensearch, list_opensearch_indices +metadata: + author: aurora + version: "1.0" +--- + +# OpenSearch Integration + +## Overview +OpenSearch integration for querying log data during Root Cause Analysis. OpenSearch is a REMOTE service — do NOT search the local filesystem. Use ONLY the tools listed below. + +## Instructions + +### Tool Usage (use in this order) +1. `list_opensearch_indices()` — Discover available indices. Call first to understand what data exists. +2. `search_opensearch(query='error', start_time='now-1h')` — Search for logs matching a Lucene query. + +### Common Query Patterns +- Error search: `search_opensearch(query='error AND service:api', start_time='now-1h')` +- Specific service: `search_opensearch(query='kubernetes.labels.app:payment-service AND level:error', start_time='now-30m')` +- HTTP 5xx: `search_opensearch(query='http.response.status_code:>=500', start_time='now-1h')` +- Exception: `search_opensearch(query='exception OR stacktrace', start_time='now-2h', end_time='now')` +- Specific index: `search_opensearch(query='NullPointerException', index='app-logs-*', start_time='now-1h')` + +### Time Format +- Relative: `now-1h`, `now-30m`, `now-6h`, `now-1d` +- Absolute: ISO-8601 strings (`2024-01-15T10:00:00Z`) + +## RCA Investigation Workflow + +**Step 1 — Discover indices:** +`list_opensearch_indices()` — find which indices contain logs relevant to the incident. + +**Step 2 — Search for errors around the alert time:** +`search_opensearch(query='error OR exception OR fatal', start_time='now-1h', size=50)` + +**Step 3 — Narrow by service:** +`search_opensearch(query='service:payment AND level:error', start_time='now-30m')` + +**Step 4 — Correlate with the alert window:** +Use `start_time` and `end_time` to focus on the exact incident window. + +## Important Rules +- OpenSearch is a REMOTE service. Never try to access data from the local filesystem. +- Always use `list_opensearch_indices` first if you're unsure which index to search. +- Keep queries focused — use specific service names and time ranges to avoid huge result sets. +- Results are capped at 200 hits. Use a more specific query if you need narrower results. +- Query syntax is Lucene: `field:value`, `AND`, `OR`, `NOT`, `field:>=N`, wildcards with `*`. diff --git a/server/chat/backend/agent/skills/integrations/victorops/SKILL.md b/server/chat/backend/agent/skills/integrations/victorops/SKILL.md new file mode 100644 index 000000000..91ea73d2d --- /dev/null +++ b/server/chat/backend/agent/skills/integrations/victorops/SKILL.md @@ -0,0 +1,41 @@ +--- +name: victorops +id: victorops +description: "Splunk On-Call (VictorOps) integration for querying incident history, on-call schedules, and team context during RCA investigations" +category: incident_management +connection_check: + method: is_connected_function + module: chat.backend.agent.tools.victorops_tool + function: is_victorops_connected +tools: + - get_victorops_incidents + - get_victorops_teams +index: "On-call incident management -- query recent incidents, team rosters, and on-call schedules from Splunk On-Call" +rca_priority: 4 +allowed-tools: get_victorops_incidents, get_victorops_teams +metadata: + author: aurora + version: "1.0" +--- + +# Splunk On-Call Integration + +## Overview +Splunk On-Call (formerly VictorOps) is an on-call incident management platform. Use this integration during RCA to retrieve incident history, check which teams are on-call, and correlate the current alert with past incidents. This is a REMOTE API — do NOT search the local filesystem. + +## Instructions + +### Tool Usage +1. `get_victorops_incidents()` — Retrieve recent incidents from Splunk On-Call. Use during RCA to find similar past incidents or check incident history. +2. `get_victorops_teams()` — List teams and on-call schedules. Use to identify who was on-call when the incident triggered. + +### RCA Workflow +- **Read-only**: Only query incident data during RCA. Never create, acknowledge, or resolve incidents automatically. +- Use `get_victorops_incidents` to find related past incidents and identify patterns. +- Cross-reference incident timeline with metrics from connected monitoring tools (Datadog, Grafana, etc.). +- Note the routing key and escalation path to understand which service or team is responsible. + +### Context to gather +- Recent incidents for the same service/routing key +- Incident frequency and recurring patterns +- Team ownership and on-call rotation at time of incident diff --git a/server/chat/backend/agent/tools/cloud_tools.py b/server/chat/backend/agent/tools/cloud_tools.py index 77d084981..f58a358e3 100644 --- a/server/chat/backend/agent/tools/cloud_tools.py +++ b/server/chat/backend/agent/tools/cloud_tools.py @@ -131,6 +131,20 @@ QueryDatadogArgs, ) from .opsgenie_tool import query_opsgenie, is_opsgenie_connected, QueryOpsGenieArgs +from .victorops_tool import ( + get_victorops_incidents, + get_victorops_teams, + is_victorops_connected, + GetVictorOpsIncidentsArgs, + GetVictorOpsTeamsArgs, +) +from .opensearch_tool import ( + search_opensearch, + list_opensearch_indices, + is_opensearch_connected, + OpenSearchSearchArgs, + OpenSearchListIndicesArgs, +) from .newrelic_tool import ( query_newrelic, is_newrelic_connected, @@ -1917,6 +1931,42 @@ def _pinned_trigger(action_id: str = "", _pid=pinned_id, _fn=final_func, **kw): else: logging.debug(f"Splunk tools not added - user {user_id} not connected to Splunk") + # Add OpenSearch tools if connected + if user_id and is_opensearch_connected(user_id): + context_wrapped_os_search = with_user_context(search_opensearch) + notification_wrapped_os_search = with_completion_notification(context_wrapped_os_search) + final_os_search = ( + wrap_func_with_capture(notification_wrapped_os_search, "search_opensearch") + if tool_capture else notification_wrapped_os_search + ) + tools.append(StructuredTool.from_function( + func=final_os_search, + name="search_opensearch", + description=( + "Search logs in OpenSearch using a Lucene query string. " + "Use this to find error messages, stack traces, and log patterns during RCA. " + "Example: search_opensearch(query='error AND service:api', start_time='now-1h', size=50)" + ), + args_schema=OpenSearchSearchArgs, + )) + + context_wrapped_os_indices = with_user_context(list_opensearch_indices) + notification_wrapped_os_indices = with_completion_notification(context_wrapped_os_indices) + final_os_indices = ( + wrap_func_with_capture(notification_wrapped_os_indices, "list_opensearch_indices") + if tool_capture else notification_wrapped_os_indices + ) + tools.append(StructuredTool.from_function( + func=final_os_indices, + name="list_opensearch_indices", + description="List available OpenSearch indices to discover what log data is available for searching.", + args_schema=OpenSearchListIndicesArgs, + )) + + logging.info(f"Added 2 OpenSearch tools for user {user_id}") + else: + logging.debug(f"OpenSearch tools not added - user {user_id} not connected to OpenSearch") + # Add incident.io tools if connected if is_incidentio_connected(user_id): context_wrapped_list = with_user_context(list_incidentio_incidents) @@ -2093,6 +2143,35 @@ def _pinned_trigger(action_id: str = "", _pid=pinned_id, _fn=final_func, **kw): )) logging.info(f"Added {_og_label} tool for user {user_id}") + # --- Splunk On-Call (VictorOps) tools --- + if user_id and is_victorops_connected(user_id): + context_wrapped_vo_inc = with_user_context(get_victorops_incidents) + notif_wrapped_vo_inc = with_completion_notification(context_wrapped_vo_inc) + final_vo_inc = wrap_func_with_capture(notif_wrapped_vo_inc, "get_victorops_incidents") if tool_capture else notif_wrapped_vo_inc + tools.append(StructuredTool.from_function( + func=final_vo_inc, + name="get_victorops_incidents", + description=( + "Retrieve recent incidents from Splunk On-Call (VictorOps). " + "Use during RCA to find related past incidents and identify recurring patterns." + ), + args_schema=GetVictorOpsIncidentsArgs, + )) + + context_wrapped_vo_teams = with_user_context(get_victorops_teams) + notif_wrapped_vo_teams = with_completion_notification(context_wrapped_vo_teams) + final_vo_teams = wrap_func_with_capture(notif_wrapped_vo_teams, "get_victorops_teams") if tool_capture else notif_wrapped_vo_teams + tools.append(StructuredTool.from_function( + func=final_vo_teams, + name="get_victorops_teams", + description=( + "List teams and on-call schedules from Splunk On-Call (VictorOps). " + "Use to identify who was on-call when the incident triggered." + ), + args_schema=GetVictorOpsTeamsArgs, + )) + logging.info(f"Added Splunk On-Call (VictorOps) tools for user {user_id}") + # Add Bitbucket tools if connected try: from .bitbucket import is_bitbucket_connected diff --git a/server/chat/backend/agent/tools/opensearch_tool.py b/server/chat/backend/agent/tools/opensearch_tool.py new file mode 100644 index 000000000..503b12397 --- /dev/null +++ b/server/chat/backend/agent/tools/opensearch_tool.py @@ -0,0 +1,158 @@ +"""OpenSearch log search tool for RCA agent.""" + +from __future__ import annotations + +import logging +from typing import Any, Dict, Optional + +from pydantic import BaseModel, Field + +from utils.auth.token_management import get_token_data + +logger = logging.getLogger(__name__) + +MAX_HITS = 200 + + +def is_opensearch_connected(user_id: str) -> bool: + """Return True if the user has OpenSearch credentials stored.""" + try: + creds = get_token_data(user_id, "opensearch") + return bool(creds and creds.get("endpoint") and creds.get("username") and creds.get("password")) + except Exception: + return False + + +def _get_client(user_id: str): + """Build an OpenSearchClient from stored credentials.""" + from connectors.opensearch_connector.client import OpenSearchClient + + creds = get_token_data(user_id, "opensearch") + if not creds: + raise RuntimeError("OpenSearch credentials not found. Please connect OpenSearch first.") + return OpenSearchClient( + endpoint=creds["endpoint"], + username=creds["username"], + password=creds["password"], + index_pattern=creds.get("index_pattern", "*"), + verify_ssl=creds.get("verify_ssl", True), + max_retries=creds.get("max_retries", 2), + ) + + +# --------------------------------------------------------------------------- +# Pydantic arg schemas +# --------------------------------------------------------------------------- + +class OpenSearchSearchArgs(BaseModel): + query: str = Field(description="Lucene query string, e.g. 'error AND service:api'") + index: Optional[str] = Field(default=None, description="Index or pattern to search, e.g. 'logs-*'. Defaults to the configured index pattern.") + start_time: Optional[str] = Field(default="now-1h", description="Start time — relative ('now-1h', 'now-30m') or ISO-8601") + end_time: Optional[str] = Field(default=None, description="End time — relative or ISO-8601. Defaults to now.") + size: int = Field(default=50, ge=1, le=MAX_HITS, description="Max number of log entries to return (1–200)") + timestamp_field: str = Field(default="@timestamp", description="Name of the timestamp field in your index") + + +class OpenSearchListIndicesArgs(BaseModel): + pattern: Optional[str] = Field(default=None, description="Index pattern filter, e.g. 'logs-*'. Defaults to all indices.") + + +# --------------------------------------------------------------------------- +# Tool functions +# --------------------------------------------------------------------------- + +def search_opensearch( + query: str, + index: Optional[str] = None, + start_time: Optional[str] = "now-1h", + end_time: Optional[str] = None, + size: int = 50, + timestamp_field: str = "@timestamp", + user_id: Optional[str] = None, +) -> str: + """Search OpenSearch logs using a Lucene query.""" + if not user_id: + return "Error: user_id not provided" + + try: + client = _get_client(user_id) + result = client.search( + query=query, + index=index, + start_time=start_time, + end_time=end_time, + size=min(size, MAX_HITS), + timestamp_field=timestamp_field, + ) + + total = result.get("total", 0) + hits = result.get("hits", []) + + if not hits: + return f"No results found for query: {query!r} in index {result.get('index', '*')}" + + lines = [f"OpenSearch results — index: {result['index']} | query: {query!r} | total hits: {total} | showing: {len(hits)}"] + for i, doc in enumerate(hits, 1): + # Extract common log fields + ts = doc.get("@timestamp") or doc.get("timestamp") or "" + level = doc.get("level") or doc.get("log.level") or doc.get("severity") or "" + msg = doc.get("message") or doc.get("msg") or doc.get("log") or "" + svc = doc.get("service") or doc.get("service.name") or doc.get("kubernetes.labels.app") or "" + + summary_parts = [] + if ts: + summary_parts.append(f"[{ts}]") + if level: + summary_parts.append(f"[{level.upper()}]") + if svc: + summary_parts.append(f"[{svc}]") + if msg: + summary_parts.append(msg[:300]) + + if summary_parts: + lines.append(f"{i}. {' '.join(summary_parts)}") + else: + import json as _json + lines.append(f"{i}. {_json.dumps(doc)[:400]}") + + if total > len(hits): + lines.append(f"\n... {total - len(hits)} more results not shown. Use a more specific query or smaller time range.") + + return "\n".join(lines) + + except Exception as exc: + logger.warning("[OPENSEARCH TOOL] search_opensearch failed for user %s: %s", user_id, exc) + return f"OpenSearch search failed: {exc}" + + +def list_opensearch_indices( + pattern: Optional[str] = None, + user_id: Optional[str] = None, +) -> str: + """List available OpenSearch indices.""" + if not user_id: + return "Error: user_id not provided" + + try: + client = _get_client(user_id) + indices = client.list_indices(pattern=pattern) + + if not indices: + return "No indices found matching pattern: " + (pattern or "*") + + lines = [f"OpenSearch indices ({len(indices)} found):"] + for idx in indices[:100]: + name = idx.get("index", "?") + health = idx.get("health", "?") + docs = idx.get("docs.count", "?") + size = idx.get("store.size", "?") + lines.append(f" - {name} | health={health} | docs={docs} | size={size}") + + if len(indices) > 100: + lines.append(f" ... and {len(indices) - 100} more") + + return "\n".join(lines) + + except Exception as exc: + logger.warning("[OPENSEARCH TOOL] list_opensearch_indices failed for user %s: %s", user_id, exc) + return f"Failed to list OpenSearch indices: {exc}" diff --git a/server/chat/backend/agent/tools/victorops_tool.py b/server/chat/backend/agent/tools/victorops_tool.py new file mode 100644 index 000000000..2b2ed7c97 --- /dev/null +++ b/server/chat/backend/agent/tools/victorops_tool.py @@ -0,0 +1,110 @@ +"""Splunk On-Call (VictorOps) query tools for the RCA agent.""" + +import json +import logging +from typing import Optional + +from pydantic import BaseModel, Field + +logger = logging.getLogger(__name__) + +MAX_OUTPUT_SIZE = 32_000 + + +class GetVictorOpsIncidentsArgs(BaseModel): + limit: int = Field(default=20, description="Maximum number of recent incidents to return") + + +class GetVictorOpsTeamsArgs(BaseModel): + pass + + +# --------------------------------------------------------------------------- +# Connection helpers +# --------------------------------------------------------------------------- + + +def is_victorops_connected(user_id: str) -> bool: + """Check if a user has valid Splunk On-Call credentials stored.""" + try: + from utils.auth.token_management import get_token_data + + creds = get_token_data(user_id, "victorops") + return bool(creds and creds.get("api_id") and creds.get("api_key")) + except Exception: + logger.debug("VictorOps connection check failed for user %s", user_id) + return False + + +def _get_client(user_id: str): + from utils.auth.token_management import get_token_data + from routes.victorops.victorops_helpers import VictorOpsClient + + creds = get_token_data(user_id, "victorops") + if not creds: + return None + return VictorOpsClient(api_id=creds["api_id"], api_key=creds["api_key"]) + + +# --------------------------------------------------------------------------- +# Tool functions +# --------------------------------------------------------------------------- + + +def get_victorops_incidents( + limit: int = 20, + user_id: Optional[str] = None, + **kwargs, +) -> str: + """Retrieve recent incidents from Splunk On-Call.""" + if not user_id: + return json.dumps({"error": "User context not available"}) + + client = _get_client(user_id) + if not client: + return json.dumps({"error": "Splunk On-Call not connected. Please connect it first."}) + + try: + from routes.victorops.victorops_helpers import VictorOpsAPIError + + data = client.get_incidents(limit=min(limit, 100)) + incidents = data.get("incidents", [])[:limit] + + # Trim to size + results_str = json.dumps(incidents) + if len(results_str) > MAX_OUTPUT_SIZE: + incidents = incidents[: max(1, limit // 2)] + + return json.dumps({ + "success": True, + "count": len(incidents), + "results": incidents, + }) + except Exception as exc: + logger.exception("[VICTOROPS-TOOL] get_incidents failed for user=%s", user_id) + return json.dumps({"error": f"Error fetching incidents: {str(exc)[:200]}"}) + + +def get_victorops_teams( + user_id: Optional[str] = None, + **kwargs, +) -> str: + """Retrieve teams and on-call information from Splunk On-Call.""" + if not user_id: + return json.dumps({"error": "User context not available"}) + + client = _get_client(user_id) + if not client: + return json.dumps({"error": "Splunk On-Call not connected. Please connect it first."}) + + try: + data = client.get_teams() + teams = data.get("teams", []) + return json.dumps({ + "success": True, + "count": len(teams), + "results": teams, + }) + except Exception as exc: + logger.exception("[VICTOROPS-TOOL] get_teams failed for user=%s", user_id) + return json.dumps({"error": f"Error fetching teams: {str(exc)[:200]}"}) diff --git a/server/chat/background/rca_prompt_builder.py b/server/chat/background/rca_prompt_builder.py index 104d11d2d..0696bd506 100644 --- a/server/chat/background/rca_prompt_builder.py +++ b/server/chat/background/rca_prompt_builder.py @@ -435,3 +435,46 @@ def build_rca_prompt( rail_text = _extract_rail_text_from_payload(payload) return prompt, rail_text + + +def build_victorops_rca_prompt( + payload: Dict[str, Any], + providers: Optional[List[str]] = None, + user_id: Optional[str] = None, +) -> tuple[str, str]: + """Build RCA prompt from a Splunk On-Call (VictorOps) webhook payload.""" + incident_number = payload.get("INCIDENT_NUMBER", "unknown") + incident_title = ( + payload.get("INCIDENT_DISPLAY_NAME") + or payload.get("ENTITY_DISPLAY_NAME") + or payload.get("ENTITY_ID") + or "Untitled Incident" + ) + alert_phase = payload.get("CURRENT_ALERT_PHASE", "TRIGGERED") + service_name = payload.get("SERVICE") or payload.get("MONITORING_TOOL") or "unknown" + state_message = payload.get("STATE_MESSAGE", "") + incident_url = payload.get("INCIDENT_URL", "") + entity_id = payload.get("ENTITY_ID", "") + routing_key = payload.get("ROUTING_KEY", "") + + alert_details = { + "title": f"#{incident_number}: {incident_title}", + "status": alert_phase, + "message": state_message, + "labels": { + "incident_number": str(incident_number), + "alert_phase": alert_phase, + "service": service_name, + }, + "incident_url": incident_url, + "incident_id": entity_id, + } + + if routing_key: + alert_details["labels"]["routing_key"] = routing_key + if ack_user := payload.get("ACK_USER"): + alert_details["labels"]["acknowledged_by"] = ack_user + if monitoring_tool := payload.get("MONITORING_TOOL"): + alert_details["labels"]["monitoring_tool"] = monitoring_tool + + return build_rca_prompt("victorops", alert_details, providers, user_id) diff --git a/server/connectors/confluence_connector/client.py b/server/connectors/confluence_connector/client.py index b587716dd..84296b035 100644 --- a/server/connectors/confluence_connector/client.py +++ b/server/connectors/confluence_connector/client.py @@ -199,7 +199,9 @@ def __init__( auth_type: str = "oauth", timeout: int = 30, cloud_id: Optional[str] = None, + email: Optional[str] = None, ): + import base64 as _base64 self.base_url = normalize_confluence_base_url(base_url) self.cloud_id = cloud_id self.auth_type = auth_type @@ -217,8 +219,17 @@ def __init__( ) self.access_token = access_token self.timeout = timeout + + # Atlassian Cloud API tokens require Basic auth (email:token). + # Data Center PATs use Bearer tokens. + if auth_type == "pat" and email and is_confluence_cloud_url(base_url): + creds = _base64.b64encode(f"{email}:{access_token}".encode()).decode() + auth_header = f"Basic {creds}" + else: + auth_header = f"Bearer {access_token}" + self.headers = { - "Authorization": f"Bearer {access_token}", + "Authorization": auth_header, "Accept": "application/json", } if auth_type not in {"oauth", "pat"}: diff --git a/server/connectors/opensearch_connector/__init__.py b/server/connectors/opensearch_connector/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/server/connectors/opensearch_connector/client.py b/server/connectors/opensearch_connector/client.py new file mode 100644 index 000000000..661fb9f63 --- /dev/null +++ b/server/connectors/opensearch_connector/client.py @@ -0,0 +1,166 @@ +"""OpenSearch REST API client supporting Basic auth and AWS SigV4.""" + +from __future__ import annotations + +import logging +from typing import Any, Dict, List, Optional +from urllib.parse import urlparse + +import requests +from requests.auth import HTTPBasicAuth + +logger = logging.getLogger(__name__) + +OPENSEARCH_TIMEOUT = (5, 20) + + +class OpenSearchError(Exception): + """Raised for OpenSearch API errors.""" + + +class OpenSearchClient: + """Thin REST client for an OpenSearch / Elasticsearch cluster.""" + + def __init__( + self, + endpoint: str, + username: str, + password: str, + index_pattern: str = "*", + verify_ssl: bool = True, + max_retries: int = 2, + ): + self.endpoint = endpoint.rstrip("/") + self.username = username + self.index_pattern = index_pattern + self.verify_ssl = verify_ssl + self.max_retries = max_retries + self._auth = HTTPBasicAuth(username, password) + self._session = requests.Session() + self._session.auth = self._auth + self._session.headers.update({ + "Content-Type": "application/json", + "Accept": "application/json", + }) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _url(self, path: str) -> str: + return f"{self.endpoint}/{path.lstrip('/')}" + + def _request(self, method: str, path: str, **kwargs) -> Dict[str, Any]: + url = self._url(path) + last_exc: Exception = RuntimeError("unknown error") + for attempt in range(max(1, self.max_retries)): + try: + resp = self._session.request( + method, + url, + timeout=OPENSEARCH_TIMEOUT, + verify=self.verify_ssl, + **kwargs, + ) + resp.raise_for_status() + return resp.json() + except requests.exceptions.ConnectTimeout as exc: + last_exc = exc + logger.warning("[OPENSEARCH] Connect timeout on attempt %d: %s", attempt + 1, url) + except requests.exceptions.ReadTimeout as exc: + last_exc = exc + logger.warning("[OPENSEARCH] Read timeout on attempt %d: %s", attempt + 1, url) + except requests.exceptions.SSLError as exc: + raise OpenSearchError(f"SSL error — check the endpoint certificate: {exc}") from exc + except requests.exceptions.ConnectionError as exc: + raise OpenSearchError(f"Unable to connect to OpenSearch at {self.endpoint}: {exc}") from exc + except requests.HTTPError as exc: + status = exc.response.status_code if exc.response is not None else "?" + if status == 401: + raise OpenSearchError("Authentication failed — check username/password.") from exc + if status == 403: + raise OpenSearchError("Access forbidden — check cluster permissions.") from exc + raise OpenSearchError(f"HTTP {status}: {exc}") from exc + except requests.RequestException as exc: + raise OpenSearchError(str(exc)) from exc + raise OpenSearchError(f"Request failed after {self.max_retries} attempts: {last_exc}") from last_exc + + # ------------------------------------------------------------------ + # Public API methods + # ------------------------------------------------------------------ + + def health(self) -> Dict[str, Any]: + """Return cluster health — used for connection validation.""" + return self._request("GET", "/_cluster/health") + + def cluster_info(self) -> Dict[str, Any]: + """Return cluster name and version info.""" + return self._request("GET", "/") + + def list_indices(self, pattern: Optional[str] = None) -> List[Dict[str, Any]]: + """List indices matching the pattern.""" + pat = pattern or self.index_pattern + return self._request("GET", f"/_cat/indices/{pat}?format=json&h=index,health,status,docs.count,store.size") + + def search( + self, + query: str, + index: Optional[str] = None, + start_time: Optional[str] = None, + end_time: Optional[str] = None, + size: int = 50, + timestamp_field: str = "@timestamp", + ) -> Dict[str, Any]: + """ + Full-text search across logs. + + Args: + query: Lucene query string (e.g. 'error AND service:api') + index: Index or pattern to search (defaults to self.index_pattern) + start_time: ISO-8601 start time (e.g. 'now-1h') + end_time: ISO-8601 end time (e.g. 'now') + size: Max number of hits to return + timestamp_field: Name of the timestamp field + """ + idx = index or self.index_pattern + must: List[Dict] = [{"query_string": {"query": query, "default_operator": "AND"}}] + + if start_time or end_time: + time_range: Dict[str, Any] = {} + if start_time: + time_range["gte"] = start_time + if end_time: + time_range["lte"] = end_time + must.append({"range": {timestamp_field: time_range}}) + + body = { + "query": {"bool": {"must": must}}, + "size": size, + "sort": [{timestamp_field: {"order": "desc"}}], + "_source": True, + } + + result = self._request("POST", f"/{idx}/_search", json=body) + hits = result.get("hits", {}) + return { + "total": hits.get("total", {}).get("value", 0), + "hits": [h.get("_source", {}) for h in hits.get("hits", [])], + "index": idx, + "query": query, + } + + def get_field_mapping(self, index: Optional[str] = None) -> Dict[str, Any]: + """Return field mappings to discover available log fields.""" + idx = index or self.index_pattern + return self._request("GET", f"/{idx}/_mapping") + + @staticmethod + def normalize_endpoint(raw: str) -> str: + """Ensure endpoint has a scheme and no trailing slash.""" + url = raw.strip().rstrip("/") + if not url.startswith(("http://", "https://")): + url = f"https://{url}" + parsed = urlparse(url) + if not parsed.netloc: + raise ValueError(f"Invalid OpenSearch endpoint: {raw!r}") + return url diff --git a/server/main_compute.py b/server/main_compute.py index 593ebc070..2d86674ea 100644 --- a/server/main_compute.py +++ b/server/main_compute.py @@ -122,6 +122,14 @@ "allow_headers": ["Content-Type", "X-Provider", "X-Requested-With", "X-User-ID", "Authorization", "X-Provider-Preference"], "methods": ["GET", "POST", "DELETE", "OPTIONS", "PATCH"]}, + r"/victorops/*": {"origins": FRONTEND_URL, "supports_credentials": True, + "allow_headers": ["Content-Type", "X-Provider", "X-Requested-With", "X-User-ID", + "Authorization", "X-Provider-Preference"], + "methods": ["GET", "POST", "DELETE", "OPTIONS"]}, + r"/opensearch/*": {"origins": FRONTEND_URL, "supports_credentials": True, + "allow_headers": ["Content-Type", "X-Provider", "X-Requested-With", "X-User-ID", + "Authorization", "X-Provider-Preference"], + "methods": ["GET", "POST", "DELETE", "OPTIONS"]}, r"/opsgenie/*": {"origins": FRONTEND_URL, "supports_credentials": True, "allow_headers": ["Content-Type", "X-Provider", "X-Requested-With", "X-User-ID", "Authorization", "X-Provider-Preference"], @@ -222,6 +230,7 @@ "/sentry/webhook/", "/pagerduty/webhook/", "/opsgenie/webhook/", + "/victorops/webhook/", "/jenkins/webhook/", "/cloudbees/webhook/", "/spinnaker/webhook/", @@ -466,6 +475,14 @@ def enforce_user_org_binding(): from routes.pagerduty.pagerduty_routes import pagerduty_bp # noqa: F401 app.register_blueprint(pagerduty_bp, url_prefix="/pagerduty") +# --- Splunk On-Call (VictorOps) Integration Routes --- +from routes.victorops.victorops_routes import victorops_bp # noqa: F401 +app.register_blueprint(victorops_bp, url_prefix="/victorops") + +# --- OpenSearch Integration Routes --- +from routes.opensearch import opensearch_bp # noqa: F401 +app.register_blueprint(opensearch_bp, url_prefix="/opensearch") + # --- OpsGenie Integration Routes --- from routes.opsgenie import bp as opsgenie_bp # noqa: F401 import routes.opsgenie.tasks # noqa: F401 diff --git a/server/routes/atlassian/atlassian_routes.py b/server/routes/atlassian/atlassian_routes.py index 4c51eb5a0..1d8153f1a 100644 --- a/server/routes/atlassian/atlassian_routes.py +++ b/server/routes/atlassian/atlassian_routes.py @@ -70,10 +70,10 @@ def _refresh_credentials(user_id: str, creds: Dict[str, Any], provider: str) -> return updated -def _validate_confluence(access_token: str, base_url: str, auth_type: str, cloud_id: Optional[str]) -> Optional[Dict[str, Any]]: +def _validate_confluence(access_token: str, base_url: str, auth_type: str, cloud_id: Optional[str], email: Optional[str] = None) -> Optional[Dict[str, Any]]: """Validate Confluence credentials and return user info.""" base_url = normalize_confluence_base_url(base_url) - client = ConfluenceClient(base_url, access_token, auth_type=auth_type, cloud_id=cloud_id) + client = ConfluenceClient(base_url, access_token, auth_type=auth_type, cloud_id=cloud_id, email=email) try: payload = client.get_current_user() return payload @@ -145,8 +145,9 @@ def connect(user_id): results[product] = {"connected": False, "error": f"baseUrl and patToken required for {product}"} continue + pat_email = data.get(f"{product}Email") or data.get("patEmail") if product == "confluence": - user_payload = _validate_confluence(pat_token, base_url, "pat", None) + user_payload = _validate_confluence(pat_token, base_url, "pat", None, email=pat_email) elif product == "jsm_ops": cloud_id = data.get("cloudId") if not cloud_id: @@ -175,6 +176,8 @@ def connect(user_id): "base_url": base_url.rstrip("/"), "pat_token": pat_token, } + if pat_email: + token_payload["email"] = pat_email store_tokens_in_db(user_id, token_payload, product) results[product] = {"connected": True, "authType": "pat", "baseUrl": base_url} @@ -326,8 +329,9 @@ def status(user_id): result[product] = {"connected": False} continue + stored_email = creds.get("email") if auth_type == "pat" else None if product == "confluence": - user_payload = _validate_confluence(token, base_url, auth_type, cloud_id) + user_payload = _validate_confluence(token, base_url, auth_type, cloud_id, email=stored_email) elif product == "jsm_ops": user_payload = _validate_jsm_ops(token, cloud_id) else: diff --git a/server/routes/connector_status.py b/server/routes/connector_status.py index b2000794d..b5c766019 100644 --- a/server/routes/connector_status.py +++ b/server/routes/connector_status.py @@ -544,6 +544,24 @@ def _check_pagerduty(creds: Dict[str, Any]) -> Dict[str, Any]: return {"connected": False} +def _check_opensearch(creds: Dict[str, Any]) -> Dict[str, Any]: + """Check OpenSearch credentials — credential-existence only (validated at connect time).""" + if creds.get("endpoint") and creds.get("username") and creds.get("password"): + return {"connected": True, "endpoint": creds.get("endpoint")} + return {"connected": False} + + +def _check_victorops(creds: Dict[str, Any]) -> Dict[str, Any]: + """Check Splunk On-Call (VictorOps) credentials — no live API call. + + Credentials are validated at connect time; here we just confirm they + are present so the connected-accounts endpoint returns quickly. + """ + if creds.get("api_id") and creds.get("api_key"): + return {"connected": True} + return {"connected": False} + + def _check_opsgenie(creds: Dict[str, Any]) -> Dict[str, Any]: """Validate OpsGenie / JSM Operations credentials.""" auth_type = creds.get("auth_type", "opsgenie") @@ -774,6 +792,8 @@ def _check_gitlab(creds: Dict[str, Any]) -> Dict[str, Any]: "notion": _check_notion, "spinnaker": _check_spinnaker, "pagerduty": _check_pagerduty, + "victorops": _check_victorops, + "opensearch": _check_opensearch, "opsgenie": _check_opsgenie, "dynatrace": _check_dynatrace, "bigpanda": _check_bigpanda, diff --git a/server/routes/opensearch/__init__.py b/server/routes/opensearch/__init__.py new file mode 100644 index 000000000..e1e80786f --- /dev/null +++ b/server/routes/opensearch/__init__.py @@ -0,0 +1,3 @@ +from .opensearch_routes import opensearch_bp + +__all__ = ["opensearch_bp"] diff --git a/server/routes/opensearch/opensearch_routes.py b/server/routes/opensearch/opensearch_routes.py new file mode 100644 index 000000000..858fbf191 --- /dev/null +++ b/server/routes/opensearch/opensearch_routes.py @@ -0,0 +1,229 @@ +"""OpenSearch connector routes — connect, disconnect, status, search.""" + +from __future__ import annotations + +import logging +from typing import Any, Dict, Optional + +from flask import Blueprint, jsonify, request + +from connectors.opensearch_connector.client import OpenSearchClient, OpenSearchError +from utils.auth.rbac_decorators import require_permission +from utils.auth.token_management import get_token_data, store_tokens_in_db +from utils.log_sanitizer import sanitize, hash_for_log +from utils.secrets.secret_ref_utils import delete_user_secret + +logger = logging.getLogger(__name__) + +opensearch_bp = Blueprint("opensearch", __name__) + + +def _get_creds(user_id: str) -> Optional[Dict[str, Any]]: + try: + return get_token_data(user_id, "opensearch") + except Exception as exc: + logger.error("[OPENSEARCH] Failed to retrieve credentials for %s: %s", sanitize(user_id), exc) + return None + + +def _make_client(creds: Dict[str, Any]) -> OpenSearchClient: + return OpenSearchClient( + endpoint=creds["endpoint"], + username=creds["username"], + password=creds["password"], + index_pattern=creds.get("index_pattern", "*"), + verify_ssl=creds.get("verify_ssl", True), + max_retries=creds.get("max_retries", 2), + ) + + +# --------------------------------------------------------------------------- +# Connect +# --------------------------------------------------------------------------- + +@opensearch_bp.route("/connect", methods=["POST"]) +@require_permission("connectors", "write") +def connect(user_id): + """Validate and store OpenSearch credentials.""" + data = request.get_json(force=True, silent=True) or {} + + raw_endpoint = data.get("endpoint", "").strip() + username = data.get("username", "").strip() + password = data.get("password", "") + index_pattern = data.get("indexPattern", "*").strip() or "*" + verify_ssl = bool(data.get("verifySsl", True)) + max_retries = int(data.get("maxRetries", 2)) + + if not raw_endpoint: + return jsonify({"error": "endpoint is required"}), 400 + if not username or not password: + return jsonify({"error": "username and password are required"}), 400 + + try: + endpoint = OpenSearchClient.normalize_endpoint(raw_endpoint) + except ValueError as exc: + return jsonify({"error": str(exc)}), 400 + + logger.info( + "[OPENSEARCH] Connecting user %s to %s (user=%s)", + sanitize(user_id), + sanitize(endpoint), + sanitize(username), + ) + + client = OpenSearchClient( + endpoint=endpoint, + username=username, + password=password, + index_pattern=index_pattern, + verify_ssl=verify_ssl, + max_retries=max_retries, + ) + + try: + info = client.cluster_info() + health = client.health() + except OpenSearchError as exc: + logger.warning("[OPENSEARCH] Connection failed for %s: %s", sanitize(user_id), exc) + return jsonify({"error": str(exc)}), 502 + + cluster_name = info.get("cluster_name", "opensearch") + version_info = info.get("version", {}) + version = version_info.get("number", "") + distribution = version_info.get("distribution", "opensearch") + cluster_status = health.get("status", "unknown") + + token_payload: Dict[str, Any] = { + "endpoint": endpoint, + "username": username, + "password": password, + "index_pattern": index_pattern, + "verify_ssl": verify_ssl, + "max_retries": max_retries, + "cluster_name": cluster_name, + "version": version, + "distribution": distribution, + } + + try: + store_tokens_in_db(user_id, token_payload, "opensearch") + logger.info("[OPENSEARCH] Stored credentials for %s (cluster=%s)", sanitize(user_id), sanitize(cluster_name)) + except Exception as exc: + logger.exception("[OPENSEARCH] Failed to store credentials for %s: %s", sanitize(user_id), exc) + return jsonify({"error": "Failed to store credentials"}), 500 + + return jsonify({ + "success": True, + "clusterName": cluster_name, + "version": version, + "distribution": distribution, + "clusterStatus": cluster_status, + "endpoint": endpoint, + "indexPattern": index_pattern, + }) + + +# --------------------------------------------------------------------------- +# Status +# --------------------------------------------------------------------------- + +@opensearch_bp.route("/status", methods=["GET"]) +@require_permission("connectors", "read") +def status(user_id): + """Check OpenSearch connection status.""" + creds = _get_creds(user_id) + if not creds: + return jsonify({"connected": False}) + + if not creds.get("endpoint") or not creds.get("username") or not creds.get("password"): + return jsonify({"connected": False}) + + return jsonify({ + "connected": True, + "clusterName": creds.get("cluster_name"), + "version": creds.get("version"), + "endpoint": creds.get("endpoint"), + "indexPattern": creds.get("index_pattern", "*"), + }) + + +# --------------------------------------------------------------------------- +# Disconnect +# --------------------------------------------------------------------------- + +@opensearch_bp.route("/disconnect", methods=["POST", "DELETE"]) +@require_permission("connectors", "write") +def disconnect(user_id): + """Remove stored OpenSearch credentials.""" + try: + success, deleted_count = delete_user_secret(user_id, "opensearch") + if not success: + return jsonify({"success": False, "error": "Failed to delete stored credentials"}), 500 + logger.info("[OPENSEARCH] Disconnected for %s (deleted %d entries)", sanitize(user_id), deleted_count) + return jsonify({"success": True, "deleted": deleted_count}) + except Exception as exc: + logger.exception("[OPENSEARCH] Disconnect failed for %s: %s", sanitize(user_id), exc) + return jsonify({"error": "Failed to disconnect OpenSearch"}), 500 + + +# --------------------------------------------------------------------------- +# Search +# --------------------------------------------------------------------------- + +@opensearch_bp.route("/search", methods=["POST"]) +@require_permission("connectors", "read") +def search(user_id): + """Execute a query against OpenSearch and return matching log entries.""" + creds = _get_creds(user_id) + if not creds: + return jsonify({"error": "OpenSearch not connected"}), 400 + + data = request.get_json(force=True, silent=True) or {} + query = data.get("query", "").strip() + if not query: + return jsonify({"error": "query is required"}), 400 + + index = data.get("index") or creds.get("index_pattern", "*") + start_time = data.get("startTime") + end_time = data.get("endTime") + size = min(int(data.get("size", 50)), 200) + timestamp_field = data.get("timestampField", "@timestamp") + + client = _make_client(creds) + + try: + result = client.search( + query=query, + index=index, + start_time=start_time, + end_time=end_time, + size=size, + timestamp_field=timestamp_field, + ) + return jsonify(result) + except OpenSearchError as exc: + logger.warning("[OPENSEARCH] Search failed for %s: %s", sanitize(user_id), exc) + return jsonify({"error": str(exc)}), 502 + + +# --------------------------------------------------------------------------- +# Indices +# --------------------------------------------------------------------------- + +@opensearch_bp.route("/indices", methods=["GET"]) +@require_permission("connectors", "read") +def list_indices(user_id): + """List available indices in the cluster.""" + creds = _get_creds(user_id) + if not creds: + return jsonify({"error": "OpenSearch not connected"}), 400 + + pattern = request.args.get("pattern") or creds.get("index_pattern", "*") + client = _make_client(creds) + + try: + indices = client.list_indices(pattern=pattern) + return jsonify({"indices": indices}) + except OpenSearchError as exc: + logger.warning("[OPENSEARCH] List indices failed for %s: %s", sanitize(user_id), exc) + return jsonify({"error": str(exc)}), 502 diff --git a/server/routes/victorops/__init__.py b/server/routes/victorops/__init__.py new file mode 100644 index 000000000..6b264907b --- /dev/null +++ b/server/routes/victorops/__init__.py @@ -0,0 +1 @@ +"""Splunk On-Call (VictorOps) integration routes.""" diff --git a/server/routes/victorops/tasks.py b/server/routes/victorops/tasks.py new file mode 100644 index 000000000..50d13e50d --- /dev/null +++ b/server/routes/victorops/tasks.py @@ -0,0 +1,432 @@ +"""Celery tasks for Splunk On-Call (VictorOps) webhook processing.""" + +from __future__ import annotations + +import json +import logging +from datetime import datetime, timezone +from typing import Any, Dict, Optional + +from celery_config import celery_app +from services.correlation.alert_correlator import AlertCorrelator +from services.correlation import handle_correlated_alert +from utils.auth.stateless_auth import set_rls_context + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Payload normalisation helpers +# --------------------------------------------------------------------------- + +_VO_PHASE_MAP = { + "UNACKED": "TRIGGERED", + "ACKED": "ACKNOWLEDGED", + "RESOLVED": "RESOLVED", + "CRITICAL": "TRIGGERED", + "WARNING": "TRIGGERED", + "ACKNOWLEDGEMENT": "ACKNOWLEDGED", + "RECOVERY": "RESOLVED", + "INFO": "TRIGGERED", +} + + +def _normalize_phase(payload: Dict[str, Any]) -> str: + """Extract and normalise the alert phase from a VictorOps flat payload.""" + raw = ( + payload.get("STATE.CURRENT_ALERT_PHASE") + or payload.get("INCIDENT.CURRENT_PHASE") + or payload.get("ALERT.message_type") + or payload.get("CURRENT_ALERT_PHASE") + or "TRIGGERED" + ).upper() + return _VO_PHASE_MAP.get(raw, raw) + + +def _extract_severity(payload: Dict[str, Any]) -> str: + """Map Splunk On-Call entity state / message to Aurora severity.""" + entity_state = ( + payload.get("INCIDENT.ENTITY_STATE") + or payload.get("STATE.CURRENT_STATE") + or payload.get("ALERT.entity_state") + or payload.get("ENTITY_STATE") + or "" + ).lower() + message = ( + payload.get("ALERT.state_message") + or payload.get("ALERT.message") + or payload.get("STATE_MESSAGE") + or "" + ).lower() + + if entity_state == "critical" or any(k in message for k in ("critical", "sev1", "p1")): + return "critical" + if entity_state == "warning" or any(k in message for k in ("high", "sev2", "p2", "warning")): + return "high" + if any(k in message for k in ("medium", "sev3", "p3")): + return "medium" + return "medium" + + +def _extract_status(alert_phase: str) -> str: + if alert_phase.upper() == "RESOLVED": + return "resolved" + if alert_phase.upper() == "ACKNOWLEDGED": + return "acknowledged" + return "investigating" + + +def _extract_incident_number(payload: Dict[str, Any]) -> Optional[int]: + """Return a stable integer identifier for the incident.""" + for key in ("INCIDENT.INCIDENT_ID", "INCIDENT_NUMBER", "STATE.INCIDENT_NAME"): + raw = payload.get(key) + if raw is not None: + try: + return int(raw) + except (ValueError, TypeError): + pass + return None + + +# --------------------------------------------------------------------------- +# Celery task +# --------------------------------------------------------------------------- + +@celery_app.task( + bind=True, + max_retries=3, + default_retry_delay=30, + name="victorops.process_event", +) +def process_victorops_event( + self, + payload: Dict[str, Any], + metadata: Optional[Dict[str, Any]] = None, + user_id: Optional[str] = None, +) -> None: + """Background processor for Splunk On-Call outbound webhook events. + + Args: + payload: Full webhook JSON body from Splunk On-Call. + metadata: Request metadata (headers, IP, etc.). + user_id: Aurora user ID. + """ + received_at = datetime.now(timezone.utc) + + try: + if not user_id: + return + + from utils.db.connection_pool import db_pool + + alert_phase = _normalize_phase(payload) + incident_number = _extract_incident_number(payload) + incident_title = ( + payload.get("INCIDENT.ENTITY_DISPLAY_NAME") + or payload.get("ALERT.entity_display_name") + or payload.get("ALERT.title") + or payload.get("INCIDENT_DISPLAY_NAME") + or payload.get("ENTITY_DISPLAY_NAME") + or payload.get("ENTITY_ID") + or "Untitled Incident" + ) + service_name = ( + payload.get("INCIDENT.SERVICE") + or payload.get("ALERT.monitoring_tool") + or payload.get("ALERT.entity_display_name") + or payload.get("SERVICE") + or payload.get("MONITORING_TOOL") + or "unknown" + ) + entity_id = ( + payload.get("STATE.ENTITY_ID") + or payload.get("ALERT.entity_id") + or payload.get("ENTITY_ID") + or "" + ) + incident_url = payload.get("ALERT.alert_url") or payload.get("INCIDENT_URL", "") + + if not incident_number: + logger.warning( + "[VICTOROPS] No INCIDENT_NUMBER in payload for user %s, skipping", user_id + ) + return + + severity = _extract_severity(payload) + aurora_status = _extract_status(alert_phase) + + with db_pool.get_admin_connection() as conn: + with conn.cursor() as cursor: + org_id = set_rls_context(cursor, conn, user_id, log_prefix="[VICTOROPS]") + if not org_id: + return + + # Persist raw event + cursor.execute( + """ + INSERT INTO victorops_events + (user_id, org_id, alert_phase, incident_number, incident_title, + service_name, entity_id, payload, received_at) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) + RETURNING id + """, + ( + user_id, + org_id, + alert_phase, + incident_number, + incident_title, + service_name, + entity_id, + json.dumps(payload), + received_at, + ), + ) + event_result = cursor.fetchone() + event_db_id = event_result[0] if event_result else None + conn.commit() + + if not event_db_id: + return + + alert_metadata = { + "incidentNumber": incident_number, + "incidentUrl": incident_url, + "entityId": entity_id, + "alertPhase": alert_phase, + } + state_message = ( + payload.get("ALERT.state_message") + or payload.get("ALERT.message") + or payload.get("STATE_MESSAGE") + or "" + ) + if state_message: + alert_metadata["description"] = state_message[:2000] + + # Correlation check for triggered events + if alert_phase == "TRIGGERED": + try: + correlator = AlertCorrelator() + correlation_result = correlator.correlate( + cursor=cursor, + user_id=user_id, + source_type="victorops", + source_alert_id=event_db_id, + alert_title=incident_title, + alert_service=service_name, + alert_severity=severity, + alert_metadata=alert_metadata, + org_id=org_id, + ) + + if correlation_result.is_correlated: + handle_correlated_alert( + cursor=cursor, + user_id=user_id, + incident_id=correlation_result.incident_id, + source_type="victorops", + source_alert_id=event_db_id, + alert_title=incident_title, + alert_service=service_name, + alert_severity=severity, + correlation_result=correlation_result, + alert_metadata=alert_metadata, + raw_payload=payload, + org_id=org_id, + ) + conn.commit() + return + except Exception as corr_exc: + logger.warning( + "[VICTOROPS] Correlation check failed, proceeding normally: %s", + corr_exc, + ) + + # Upsert incident + cursor.execute( + """ + WITH prev AS ( + SELECT status FROM incidents + WHERE org_id = %s AND source_type = 'victorops' + AND source_alert_id = %s AND user_id = %s + ) + INSERT INTO incidents + (user_id, org_id, source_type, source_alert_id, alert_title, alert_service, + severity, status, started_at, alert_metadata, alert_fired_at) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (org_id, source_type, source_alert_id, user_id) DO UPDATE + SET updated_at = CURRENT_TIMESTAMP, + status = EXCLUDED.status, + severity = EXCLUDED.severity, + started_at = CASE + WHEN incidents.status = 'resolved' AND EXCLUDED.status != 'resolved' + THEN EXCLUDED.started_at + ELSE incidents.started_at + END, + alert_metadata = EXCLUDED.alert_metadata, + alert_fired_at = COALESCE(EXCLUDED.alert_fired_at, incidents.alert_fired_at) + RETURNING id, (xmax = 0) AS inserted, (SELECT status FROM prev) AS previous_status + """, + ( + org_id, + incident_number, + user_id, + user_id, + org_id, + "victorops", + incident_number, + incident_title, + service_name, + severity, + aurora_status, + received_at, + json.dumps(alert_metadata), + received_at, + ), + ) + incident_row = cursor.fetchone() + incident_db_id = incident_row[0] if incident_row else None + incident_was_inserted = bool(incident_row[1]) if incident_row else False + previous_status = incident_row[2] if incident_row else None + conn.commit() + + if not incident_db_id: + return + + # Record primary alert for triggered events + if alert_phase == "TRIGGERED": + try: + cursor.execute( + """INSERT INTO incident_alerts + (user_id, org_id, incident_id, source_type, source_alert_id, + alert_title, alert_service, alert_severity, + correlation_strategy, correlation_score, alert_metadata) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""", + ( + user_id, + org_id, + incident_db_id, + "victorops", + event_db_id, + incident_title, + service_name, + severity, + "primary", + 1.0, + json.dumps(alert_metadata), + ), + ) + cursor.execute( + "UPDATE incidents SET affected_services = ARRAY[%s] WHERE id = %s", + (service_name, incident_db_id), + ) + conn.commit() + except Exception as e: + logger.warning("[VICTOROPS] Failed to record primary alert: %s", e) + + # Lifecycle events + lifecycle_writes = [] + if incident_was_inserted and alert_phase == "TRIGGERED": + lifecycle_writes.append(("created", None, "investigating")) + elif previous_status is not None and previous_status != aurora_status: + ev_name = "resolved" if aurora_status == "resolved" else "status_changed" + lifecycle_writes.append((ev_name, previous_status, aurora_status)) + + for ev_type, prev_val, new_val in lifecycle_writes: + try: + cursor.execute("SAVEPOINT sp_lifecycle") + cursor.execute( + """INSERT INTO incident_lifecycle_events + (incident_id, user_id, org_id, event_type, previous_value, new_value) + VALUES (%s, %s, %s, %s, %s, %s)""", + (incident_db_id, user_id, org_id, ev_type, prev_val, new_val), + ) + cursor.execute("RELEASE SAVEPOINT sp_lifecycle") + conn.commit() + except Exception as e: + try: + cursor.execute("ROLLBACK TO SAVEPOINT sp_lifecycle") + except Exception: + pass + logger.warning( + "[VICTOROPS] Failed to record lifecycle event %s for incident %s: %s", + ev_type, incident_db_id, e, + ) + + # SSE broadcast + try: + from routes.incidents_sse import broadcast_incident_update_to_user_connections + broadcast_incident_update_to_user_connections( + user_id, + {"type": "incident_update", "incident_id": str(incident_db_id), "source": "victorops"}, + ) + except Exception as e: + logger.warning("[VICTOROPS] Failed to notify SSE: %s", e) + + # Summary + RCA only for new triggered incidents + if alert_phase == "TRIGGERED": + try: + from chat.background.summarization import generate_incident_summary + generate_incident_summary.delay( + incident_id=str(incident_db_id), + user_id=user_id, + source_type="victorops", + alert_title=incident_title or "Unknown Incident", + severity=severity, + service=service_name, + raw_payload=payload, + alert_metadata=alert_metadata, + ) + except Exception as e: + logger.warning("[VICTOROPS] Failed to schedule summary: %s", e) + + try: + from chat.background.task import ( + run_background_chat, + create_background_chat_session, + is_background_chat_allowed, + ) + from chat.background.rca_prompt_builder import build_victorops_rca_prompt + + if is_background_chat_allowed(user_id): + rca_prompt = build_victorops_rca_prompt(payload, user_id=user_id) + session_id = create_background_chat_session( + user_id=user_id, + title=f"RCA: {incident_title}", + trigger_metadata={ + "source": "victorops", + "incident_number": incident_number, + "severity": severity, + }, + incident_id=incident_db_id, + ) + task = run_background_chat.delay( + user_id=user_id, + session_id=session_id, + initial_message=rca_prompt, + trigger_metadata={ + "source": "victorops", + "incident_number": incident_number, + }, + incident_id=incident_db_id, + ) + cursor.execute( + "UPDATE incidents SET rca_celery_task_id = %s WHERE id = %s", + (task.id, incident_db_id), + ) + conn.commit() + logger.info( + "[VICTOROPS] Triggered RCA for incident %s (task=%s)", + incident_db_id, task.id, + ) + except Exception as e: + logger.warning("[VICTOROPS] Failed to trigger RCA: %s", e) + + logger.info( + "[VICTOROPS] Processed %s event for incident #%s (db_id=%s)", + alert_phase, incident_number, incident_db_id, + ) + + except Exception as exc: + raise self.retry(exc=exc) diff --git a/server/routes/victorops/victorops_helpers.py b/server/routes/victorops/victorops_helpers.py new file mode 100644 index 000000000..ffb2b0dd6 --- /dev/null +++ b/server/routes/victorops/victorops_helpers.py @@ -0,0 +1,128 @@ +"""Splunk On-Call (VictorOps) API client and helper functions.""" + +import logging +from datetime import datetime, timezone +from typing import Any, Dict, Optional + +import requests +from flask import jsonify + +logger = logging.getLogger(__name__) + +VICTOROPS_API_BASE = "https://api.victorops.com/api-public/v1" + + +class VictorOpsAPIError(Exception): + """VictorOps API error.""" + + +class VictorOpsClient: + """Splunk On-Call (VictorOps) REST API client. + + Authentication uses two headers: + X-VO-Api-Id — the API ID from the Splunk On-Call portal + X-VO-Api-Key — the API Key from the Splunk On-Call portal + """ + + def __init__(self, api_id: str, api_key: str): + self.api_id = api_id + self.api_key = api_key + self.base_url = VICTOROPS_API_BASE + + @property + def headers(self) -> Dict[str, str]: + return { + "X-VO-Api-Id": self.api_id, + "X-VO-Api-Key": self.api_key, + "Accept": "application/json", + "Content-Type": "application/json", + } + + def _request(self, method: str, path: str, **kwargs) -> requests.Response: + try: + response = requests.request( + method, + f"{self.base_url}{path}", + headers=self.headers, + # (connect_timeout, read_timeout) — fail fast on DNS/connect, generous on read + timeout=(5, 15), + **kwargs, + ) + response.raise_for_status() + return response + except requests.exceptions.ConnectTimeout: + raise VictorOpsAPIError("Connection timed out reaching Splunk On-Call API. Check your network or try again.") + except requests.exceptions.ReadTimeout: + raise VictorOpsAPIError("Splunk On-Call API took too long to respond. Please try again.") + except requests.exceptions.ConnectionError as e: + raise VictorOpsAPIError(f"Could not connect to Splunk On-Call API: {e}") + except requests.RequestException as e: + if hasattr(e, "response") and e.response is not None: + status_code = e.response.status_code + if status_code == 429: + raise VictorOpsAPIError("Rate limited by Splunk On-Call") + elif status_code == 401: + raise VictorOpsAPIError("Unauthorized: Invalid API ID or Key") + elif status_code == 403: + raise VictorOpsAPIError("Forbidden: API credentials lack required permissions") + elif status_code == 404: + raise VictorOpsAPIError("Not found: check your API ID and Key") + else: + raise VictorOpsAPIError(str(e)) + else: + raise VictorOpsAPIError(str(e)) + + def get_current_oncall(self) -> Dict[str, Any]: + """Lightweight endpoint — used only for credential validation.""" + return self._request("GET", "/oncall/current").json() + + def get_teams(self) -> Dict[str, Any]: + """Fetch all teams.""" + return self._request("GET", "/team").json() + + def get_incidents(self, limit: int = 10) -> Dict[str, Any]: + """Fetch recent incidents.""" + return self._request("GET", f"/incidents?limit={limit}").json() + + +def validate_credentials(client: VictorOpsClient) -> Dict[str, Any]: + """Validate API credentials using the lightweight /oncall/current endpoint.""" + result = { + "validated_at": datetime.now(timezone.utc).isoformat(), + "capabilities": {"can_read_incidents": True}, + } + + try: + data = client.get_current_oncall() + # teamsOnCall is present in a valid response + if isinstance(data, dict): + teams = data.get("teamsOnCall", []) + if teams and isinstance(teams, list): + first_team = teams[0] + if team_name := first_team.get("team", {}).get("name"): + result["account_name"] = team_name + except VictorOpsAPIError: + raise + + return result + + +def error_response(exc: VictorOpsAPIError): + """Convert a VictorOpsAPIError to an HTTP response.""" + msg = str(exc).lower() + + if "timed out" in msg or "too long" in msg: + return jsonify({"error": str(exc)}), 504 + if "could not connect" in msg: + return jsonify({"error": str(exc)}), 502 + if "unauthorized" in msg or "invalid api" in msg: + return jsonify({"error": "Invalid API ID or Key"}), 401 + if "forbidden" in msg: + return jsonify({"error": "API credentials lack required permissions"}), 403 + if "rate limit" in msg: + return jsonify({"error": "Rate limited by Splunk On-Call"}), 429 + if "not found" in msg: + return jsonify({"error": "Not found: check your API ID"}), 404 + + logger.error("Splunk On-Call API error: %s", exc) + return jsonify({"error": "Splunk On-Call API request failed"}), 502 diff --git a/server/routes/victorops/victorops_routes.py b/server/routes/victorops/victorops_routes.py new file mode 100644 index 000000000..361011019 --- /dev/null +++ b/server/routes/victorops/victorops_routes.py @@ -0,0 +1,194 @@ +"""Splunk On-Call (VictorOps) integration routes.""" + +import json +import logging +import os + +from flask import Blueprint, jsonify, request + +from utils.auth.rbac_decorators import require_permission +from utils.auth.token_management import get_token_data, store_tokens_in_db +from utils.log_sanitizer import sanitize +from utils.secrets.secret_ref_utils import delete_user_secret +from routes.victorops.victorops_helpers import ( + VictorOpsAPIError, + VictorOpsClient, + error_response, + validate_credentials, +) + +logger = logging.getLogger(__name__) +victorops_bp = Blueprint("victorops", __name__) + + +@victorops_bp.route("", methods=["GET"]) +@require_permission("connectors", "read") +def victorops_status(user_id): + """Get Splunk On-Call connection status.""" + creds = get_token_data(user_id, "victorops") + if not creds: + return jsonify({"connected": False}) + + return jsonify({ + "connected": True, + "displayName": creds.get("display_name", "Splunk On-Call"), + "validatedAt": creds.get("validated_at"), + "externalUserName": creds.get("external_user_name"), + "accountName": creds.get("account_name"), + "capabilities": creds.get("capabilities", {}), + }) + + +@victorops_bp.route("", methods=["POST"]) +@require_permission("connectors", "write") +def victorops_connect(user_id): + """Connect Splunk On-Call using API ID and API Key.""" + data = request.get_json(force=True, silent=True) or {} + api_id = (data.get("apiId") or "").strip() + api_key = (data.get("apiKey") or "").strip() + display_name = data.get("displayName", "Splunk On-Call") + + if not api_id or not api_key: + return jsonify({"error": "Both API ID and API Key are required"}), 400 + + logger.info("[VICTOROPS] Validating credentials for user %s", user_id) + try: + client = VictorOpsClient(api_id=api_id, api_key=api_key) + token_info = validate_credentials(client) + logger.info("[VICTOROPS] Credentials validated for user %s", user_id) + except VictorOpsAPIError as e: + logger.warning("[VICTOROPS] Credential validation failed for user %s: %s", user_id, e) + return error_response(e) + + token_data = { + "api_id": api_id, + "api_key": api_key, + "display_name": display_name, + **token_info, + } + + try: + store_tokens_in_db(user_id, token_data, "victorops") + except Exception: + logger.exception("[VICTOROPS] Failed to store credentials for user %s", user_id) + return jsonify({"error": "Storage failed"}), 500 + + return jsonify({"success": True, "connected": True, "displayName": display_name, **token_info}) + + +@victorops_bp.route("", methods=["DELETE"]) +@require_permission("connectors", "write") +def victorops_disconnect(user_id): + """Disconnect Splunk On-Call.""" + try: + success, deleted = delete_user_secret(user_id, "victorops") + if not success: + logger.warning("[VICTOROPS] Failed to clean up secrets during disconnect") + return jsonify({"success": False, "error": "Failed to delete stored credentials"}), 500 + + logger.info("[VICTOROPS] Disconnected (deleted %d token rows)", deleted) + return jsonify({"success": True, "deleted": deleted}) + except Exception: + logger.exception("[VICTOROPS] Disconnect failed") + return jsonify({"error": "Disconnect failed"}), 500 + + +@victorops_bp.route("/webhook-url", methods=["GET"]) +@require_permission("connectors", "read") +def get_webhook_url(user_id): + """Return the webhook URL to configure in Splunk On-Call.""" + ngrok_url = os.getenv("NGROK_URL", "").rstrip("/") + backend_url = os.getenv("NEXT_PUBLIC_BACKEND_URL", "").rstrip("/") + + base_url = ngrok_url if (ngrok_url and backend_url.startswith("http://localhost")) else backend_url + webhook_url = f"{base_url}/victorops/webhook/{user_id}" + + return jsonify({ + "webhookUrl": webhook_url, + "instructions": [ + "1. Log in to your Splunk On-Call (VictorOps) portal", + "2. Go to Integrations → Outgoing Webhooks", + "3. Click 'Add Webhook'", + "4. Set Event Type to: Any-Incident", + "5. Paste the webhook URL above", + "6. Save and test the webhook", + ], + }) + + +@victorops_bp.route("/webhook/", methods=["POST"]) +def webhook(user_id: str): + """Receive outbound webhook events from Splunk On-Call.""" + if not user_id: + return jsonify({"error": "user_id is required"}), 400 + + creds = get_token_data(user_id, "victorops") + if not creds: + logger.warning( + "[VICTOROPS] Webhook received for user %s with no VictorOps connection", + sanitize(user_id), + ) + return jsonify({"error": "Splunk On-Call not connected for this user"}), 404 + + payload = request.get_json(silent=True) or {} + + # VictorOps sends a flat dot-notation payload. The phase lives at + # STATE.CURRENT_ALERT_PHASE (UNACKED/ACKED/RESOLVED) or can be + # inferred from INCIDENT.CURRENT_PHASE or ALERT.message_type. + _VO_PHASE_MAP = { + "UNACKED": "TRIGGERED", + "ACKED": "ACKNOWLEDGED", + "RESOLVED": "RESOLVED", + "CRITICAL": "TRIGGERED", + "WARNING": "TRIGGERED", + "ACKNOWLEDGEMENT": "ACKNOWLEDGED", + "RECOVERY": "RESOLVED", + "INFO": "TRIGGERED", + } + raw_phase = ( + payload.get("STATE.CURRENT_ALERT_PHASE") + or payload.get("INCIDENT.CURRENT_PHASE") + or payload.get("ALERT.message_type") + or payload.get("CURRENT_ALERT_PHASE") + or "" + ).upper() + alert_phase = _VO_PHASE_MAP.get(raw_phase, raw_phase) + + entity_id = ( + payload.get("STATE.ENTITY_ID") + or payload.get("ALERT.entity_id") + or payload.get("ENTITY_ID") + or "" + ) + + logger.info( + "[VICTOROPS] Webhook received for user %s: raw_phase=%s -> phase=%s, entity=%s", + sanitize(user_id), + sanitize(raw_phase), + sanitize(alert_phase), + sanitize(entity_id), + ) + + if not alert_phase: + logger.debug("[VICTOROPS] Ignoring webhook with no recognisable alert phase") + return jsonify({"received": True, "reason": "no alert phase"}) + + if alert_phase not in ("TRIGGERED", "ACKNOWLEDGED", "RESOLVED"): + logger.debug("[VICTOROPS] Ignoring unrecognised alert phase: %s", alert_phase) + return jsonify({"received": True, "reason": "unrecognised alert phase"}) + + from routes.victorops.tasks import process_victorops_event + + metadata = {"headers": dict(request.headers), "remote_addr": request.remote_addr} + process_victorops_event.delay( + payload=payload, + metadata=metadata, + user_id=user_id, + ) + + logger.info( + "[VICTOROPS] Enqueued event for processing: user=%s, phase=%s", + sanitize(user_id), + sanitize(alert_phase), + ) + return jsonify({"received": True}) diff --git a/server/utils/db/db_utils.py b/server/utils/db/db_utils.py index 40a7bac00..5d8378f51 100644 --- a/server/utils/db/db_utils.py +++ b/server/utils/db/db_utils.py @@ -710,6 +710,26 @@ def initialize_tables(): CREATE INDEX IF NOT EXISTS idx_pagerduty_events_status ON pagerduty_events(incident_status); CREATE INDEX IF NOT EXISTS idx_pagerduty_events_received_at ON pagerduty_events(received_at DESC); """, + "victorops_events": """ + CREATE TABLE IF NOT EXISTS victorops_events ( + id SERIAL PRIMARY KEY, + user_id VARCHAR(255) NOT NULL, + org_id VARCHAR(255), + alert_phase VARCHAR(50), + incident_number INTEGER, + incident_title TEXT, + service_name VARCHAR(255), + entity_id VARCHAR(255), + payload JSONB NOT NULL, + received_at TIMESTAMP NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + + CREATE INDEX IF NOT EXISTS idx_victorops_events_user_id ON victorops_events(user_id, received_at DESC); + CREATE INDEX IF NOT EXISTS idx_victorops_events_incident_number ON victorops_events(incident_number); + CREATE INDEX IF NOT EXISTS idx_victorops_events_received_at ON victorops_events(received_at DESC); + CREATE INDEX IF NOT EXISTS idx_victorops_events_alert_phase ON victorops_events(alert_phase); + """, "opsgenie_events": """ CREATE TABLE IF NOT EXISTS opsgenie_events ( id SERIAL PRIMARY KEY, @@ -1434,6 +1454,7 @@ def initialize_tables(): rls_tables.append("newrelic_events") rls_tables.append("sentry_events") rls_tables.append("pagerduty_events") + rls_tables.append("victorops_events") # Add monitoring tables rls_tables.append("cloudwatch_alarms") @@ -2509,7 +2530,7 @@ def initialize_tables(): _org_id_tables = list(set(rls_tables + [ "users", "workspaces", "aurora_deployments", "cloud_feed_metadata", "cloud_ingestion_state", - "pagerduty_events", "opsgenie_events", "knowledge_base_memory", + "pagerduty_events", "opsgenie_events", "victorops_events", "knowledge_base_memory", "knowledge_base_documents", ])) for tbl in _org_id_tables: @@ -2685,7 +2706,7 @@ def initialize_tables(): "deployment_tasks", "deployments", "chat_sessions", "llm_usage_tracking", "cloud_feed_metadata", "cloud_ingestion_state", "grafana_alerts", "datadog_events", "netdata_alerts", - "pagerduty_events", "opsgenie_events", "incidents", "incident_alerts", + "pagerduty_events", "opsgenie_events", "victorops_events", "incidents", "incident_alerts", "rca_notification_emails", "splunk_alerts", "jenkins_deployment_events", "dynatrace_problems", "bigpanda_events", "kubectl_agent_tokens", diff --git a/server/utils/providers.py b/server/utils/providers.py index d0315e460..9489eded9 100644 --- a/server/utils/providers.py +++ b/server/utils/providers.py @@ -33,9 +33,11 @@ "netdata", "newrelic", "notion", + "opensearch", "opsgenie", "ovh", "pagerduty", + "victorops", "scaleway", "sentry", "sharepoint", diff --git a/server/utils/secrets/secret_ref_utils.py b/server/utils/secrets/secret_ref_utils.py index 899cf2878..ec1829923 100644 --- a/server/utils/secrets/secret_ref_utils.py +++ b/server/utils/secrets/secret_ref_utils.py @@ -47,6 +47,7 @@ "pagerduty", # PagerDuty connector tokens "opsgenie", # OpsGenie connector tokens "splunk", # Splunk connector tokens + "opensearch", # OpenSearch connector tokens "ovh", # OVH Cloud "scaleway", # Scaleway Cloud "tailscale", # Tailscale VPN @@ -71,6 +72,7 @@ "incidentio", # incident.io connector tokens "flyio", # Fly.io connector tokens "kubeconfig", # Kubernetes kubeconfig uploads + "victorops", # Splunk On-Call (VictorOps) connector tokens } From e2a3f453b25084f24a3c149e3f1d84d399333ccc Mon Sep 17 00:00:00 2001 From: Amrit-thapaliya Date: Sat, 16 May 2026 18:42:52 +0545 Subject: [PATCH 2/5] revert: remove unrelated Atlassian Cloud changes from connector PR Restores AtlassianConnectPage.tsx, confluence_connector/client.py, and atlassian_routes.py to their upstream/main state. The Atlassian Cloud email + API token improvements (Basic auth support for .atlassian.net URLs) were inadvertently included from the personal fork's working directory and belong in a separate PR. Co-authored-by: Cursor --- .../connectors/AtlassianConnectPage.tsx | 26 ++----------------- .../connectors/confluence_connector/client.py | 13 +--------- server/routes/atlassian/atlassian_routes.py | 12 +++------ 3 files changed, 7 insertions(+), 44 deletions(-) diff --git a/client/src/components/connectors/AtlassianConnectPage.tsx b/client/src/components/connectors/AtlassianConnectPage.tsx index 620c4a6b6..1aff526be 100644 --- a/client/src/components/connectors/AtlassianConnectPage.tsx +++ b/client/src/components/connectors/AtlassianConnectPage.tsx @@ -44,7 +44,6 @@ export function AtlassianConnectPage({ product, sibling }: AtlassianConnectPageP const [isDisconnecting, setIsDisconnecting] = useState(false); const [alsoConnectSibling, setAlsoConnectSibling] = useState(false); const [patUrl, setPatUrl] = useState(""); - const [patEmail, setPatEmail] = useState(""); const [patToken, setPatToken] = useState(""); const [isPatConnecting, setIsPatConnecting] = useState(false); const [jiraMode, setJiraMode] = useState<"full" | "comment_only">("comment_only"); @@ -128,18 +127,14 @@ export function AtlassianConnectPage({ product, sibling }: AtlassianConnectPageP } finally { setIsConnecting(false); } }; - const isCloudUrl = (url: string) => url.toLowerCase().includes(".atlassian.net"); - const handlePatConnect = async (e: React.FormEvent) => { e.preventDefault(); if (!patUrl || !patToken) return; - if (isCloudUrl(patUrl) && !patEmail) return; setIsPatConnecting(true); try { const payload: Record = { products: [product.key], authType: "pat" as const }; payload[`${product.key}BaseUrl`] = patUrl; payload[`${product.key}PatToken`] = patToken; - if (patEmail) payload[`${product.key}Email`] = patEmail; await atlassianService.connect(payload as unknown as Parameters[0]); await loadStatus(); toast({ title: `${product.name} connected via PAT` }); @@ -410,26 +405,9 @@ export function AtlassianConnectPage({ product, sibling }: AtlassianConnectPageP setPatUrl(e.target.value)} required className="h-9" /> - {isCloudUrl(patUrl) && ( -
- - setPatEmail(e.target.value)} required className="h-9" /> -

Required for Atlassian Cloud API tokens

-
- )}
- - setPatToken(e.target.value)} required className="h-9" /> - {isCloudUrl(patUrl) && ( -

- Generate at{" "} - - id.atlassian.com → API tokens - -

- )} + + setPatToken(e.target.value)} required className="h-9" />