import streamlit as st import requests import json from datetime import datetime, timedelta import pandas as pd import time from utils import ZabbixAPI, OllamaAnalysis from config import ZABBIX_API_URL, OLLAMA_API_URL # Set page config st.set_page_config( page_title="Zabbix Event Analyzer", page_icon="๐Ÿ“Š", layout="wide", initial_sidebar_state="expanded" ) # Custom CSS st.markdown(""" """, unsafe_allow_html=True) # Header st.markdown('
Zabbix Event Analyzer
', unsafe_allow_html=True) st.markdown('
Analyze Zabbix events with AI-powered insights
', unsafe_allow_html=True) # Built with anycoder link st.markdown('Built with anycoder', unsafe_allow_html=True) # Initialize session state if 'zabbix_auth_token' not in st.session_state: st.session_state.zabbix_auth_token = None if 'events_data' not in st.session_state: st.session_state.events_data = None if 'analysis_results' not in st.session_state: st.session_state.analysis_results = None # Sidebar with st.sidebar: st.header("Configuration") # Authentication method selection auth_method = st.radio( "Authentication Method", ["Username/Password", "API Token"], index=0 ) # Zabbix API Configuration - simplified to just connect button with st.expander("Zabbix API Settings", expanded=True): if auth_method == "Username/Password": zabbix_user = st.text_input("Username", value="Admin") zabbix_password = st.text_input("Password", type="password") else: zabbix_token = st.text_input("API Token", type="password") if st.button("ะŸะพะดะบะปัŽั‡ะธั‚ัŒัั"): with st.spinner("Connecting to Zabbix..."): try: zabbix = ZabbixAPI(ZABBIX_API_URL) if auth_method == "Username/Password": auth_token = zabbix.authenticate(zabbix_user, zabbix_password) else: # For API token authentication, we'll validate it by making a simple API call auth_token = zabbix_token # Test the token by making a simple API call test_result = zabbix.test_token_auth(auth_token) if not test_result: st.error("Invalid API token") auth_token = None if auth_token: st.session_state.zabbix_auth_token = auth_token st.success("Successfully connected to Zabbix!") else: st.error("Failed to authenticate with Zabbix") except Exception as e: st.error(f"Connection error: {str(e)}") # Event Filtering with st.expander("Event Filters", expanded=True): if st.session_state.zabbix_auth_token: # Get host groups zabbix = ZabbixAPI(ZABBIX_API_URL) host_groups = zabbix.get_host_groups(st.session_state.zabbix_auth_token) selected_group = st.selectbox("Host Group", [""] + [g['name'] for g in host_groups]) # Get hosts based on selected group hosts = [] if selected_group: hosts = zabbix.get_hosts_by_group(st.session_state.zabbix_auth_token, selected_group) selected_host = st.selectbox("Host", [""] + [h['name'] for h in hosts]) # Time range time_range = st.select_slider( "Time Range (hours)", options=[1, 6, 12, 24, 48, 72, 168], value=24 ) # Event severity severity_options = { "Not classified": -1, "Information": 0, "Warning": 1, "Average": 2, "High": 3, "Disaster": 4 } selected_severities = st.multiselect( "Severity Levels", options=list(severity_options.keys()), default=["Warning", "Average", "High", "Disaster"] ) if st.button("Fetch Events"): with st.spinner("Fetching events from Zabbix..."): try: end_time = int(time.time()) start_time = end_time - (time_range * 3600) events = zabbix.get_events( st.session_state.zabbix_auth_token, start_time, end_time, [severity_options[s] for s in selected_severities], selected_host ) if events: st.session_state.events_data = events st.success(f"Fetched {len(events)} events") else: st.warning("No events found matching the criteria") except Exception as e: st.error(f"Error fetching events: {str(e)}") else: st.warning("Please connect to Zabbix first") # Ollama Configuration with st.expander("Ollama Settings", expanded=True): ollama_url = st.text_input("Ollama API URL", value=OLLAMA_API_URL) model_name = st.selectbox( "Analysis Model", ["llama3", "mistral", "gemma", "phi3"], index=0 ) # Main content if st.session_state.events_data: st.header("Event Analysis") # Display events with st.expander("Raw Events Data", expanded=False): events_df = pd.DataFrame(st.session_state.events_data) st.dataframe(events_df, use_container_width=True) # Analyze events if st.button("Analyze Events with Ollama"): with st.spinner("Analyzing events with Ollama..."): try: # Prepare event summaries for analysis event_summaries = [] for event in st.session_state.events_data[:10]: # Limit to 10 events for demo summary = ( f"Event: {event.get('name', 'N/A')}\n" f"Host: {event.get('host', 'N/A')}\n" f"Severity: {event.get('severity', 'N/A')}\n" f"Time: {datetime.fromtimestamp(int(event.get('clock', 0))).strftime('%Y-%m-%d %H:%M:%S')}\n" f"Status: {event.get('value', 'N/A')}\n" f"Description: {event.get('description', 'N/A')}" ) event_summaries.append(summary) # Combine all events into a single prompt prompt = ( "Analyze the following Zabbix monitoring events and provide:\n" "1. A summary of the overall system health\n" "2. Identification of any patterns or recurring issues\n" "3. Recommendations for resolution or investigation\n\n" "Events:\n" + "\n\n".join(event_summaries) ) # Get analysis from Ollama ollama = OllamaAnalysis(ollama_url) analysis = ollama.get_analysis(model_name, prompt) if analysis: st.session_state.analysis_results = analysis else: st.error("Failed to get analysis from Ollama") except Exception as e: st.error(f"Analysis error: {str(e)}") # Display analysis results if st.session_state.analysis_results: st.subheader("Analysis Results") # Display in a nice card format st.markdown(f"""

๐Ÿ“Š System Health Summary

{st.session_state.analysis_results}

""", unsafe_allow_html=True) # Download button for analysis st.download_button( label="Download Analysis", data=st.session_state.analysis_results, file_name=f"zabbix_analysis_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt", mime="text/plain" ) else: if st.session_state.zabbix_auth_token: st.info("Connect to Zabbix and fetch events to begin analysis") else: st.info("Please configure and connect to Zabbix to begin")