Forum Discussion

Mpyakurel's avatar
Mpyakurel
Icon for Neophyte rankNeophyte
2 months ago
Solved

Alert/Alerts API endpoint

Hello Everyone,

I am trying to get the list of all alerts through the API. Endpoint I am using is  https://companyname.logicmonitor.com/santaba/rest/alert/alerts?v=3

Through this API I am only able to get the Uncleared Alerts.
Now, Issue is I want all alerts including the one that has been cleared for reporting purpose. 

I would really appreciate if anyone could help me with this? 

 

Thank you,

Mnaish

  • By default the API only returns active alerts. To return alerts in any state you need to send a filter of cleared with an star.

    ?filter=cleared:"*"

    Just remember tho, if you truly want "every" alert. You can't paginate more then 10 times, so your limit is 10000. You will need to break it up in smaller chunks of time. An example of that would be something like below in python. This gets epochs for every hour of every day for the past X days. Then loops through and gets the alerts for that hour. If your portal is anything like ours, you will still hit that 10000 limit occasionally but it works pretty well.

     

    def generate_hourly_epochs(days_back, start_from=None):
        """
        Generate a list of dictionaries containing start and end epoch times for each hour
        starting from midnight of the current day and going back specified number of days.
        
        Args:
            days_back (int): Number of days to look back
            start_from (datetime): Optional timestamp to start from
            
        Returns:
            list: List of dictionaries with start and end epochs for each hour
        """
        # Get current time and set it to midnight of today
        current_time = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
        start_time = current_time - timedelta(days=days_back)
        
        # If we have a start_from time, use it instead
        if start_from:
            start_time = start_from.replace(minute=0, second=0, microsecond=0)
            print(f"Resuming from: {start_time}")
        
        epochs = []
        
        # Generate epochs for each hour
        current_hour = start_time
        while current_hour <= current_time:
            hour_end = current_hour + timedelta(hours=1)
            epochs.append({
                'start_epoch': int(current_hour.timestamp()),
                'end_epoch': int(hour_end.timestamp())
            })
            current_hour = hour_end
        
        return epochs
    
    last_processed_time = get_last_processed_time()
    epochs = generate_hourly_epochs(120, start_from=last_processed_time)
    alerts_by_month = {}  # Dictionary to store alerts grouped by month
    
    try:
        for epoch in epochs:
            start_epoch = epoch['start_epoch']
            end_epoch = epoch['end_epoch']
    
            activealerts = lm_rest_api.get_alerts(fields=['id','resourceId','instanceName','type','internalId','tenant','monitorObjectName','startEpoch','severity','endEpoch'],filter=f'startEpoch>:"{str(start_epoch)}",startEpoch<:"{str(end_epoch)}",cleared:"*"')
            
            outputtime = datetime.fromtimestamp(start_epoch).strftime('%Y-%m-%d %H:%M:%S')
            month_key = datetime.fromtimestamp(start_epoch).strftime('%Y%m')  # Format: YYYYMM
            
            # Skip if no alerts (boolean response)
            if isinstance(activealerts, bool):
                print(f"No alerts for {outputtime}")
                continue
            
            # Initialize month list if not exists
            if month_key not in alerts_by_month:
                alerts_by_month[month_key] = []
            
            # Add timestamp info to each alert
            for alert in activealerts:
                if isinstance(alert, dict):
                    alert_copy = alert.copy()
                    alert_copy['interval_start'] = datetime.fromtimestamp(start_epoch).strftime('%Y-%m-%d %H:%M:%S')
                    alert_copy['interval_end'] = datetime.fromtimestamp(end_epoch).strftime('%Y-%m-%d %H:%M:%S')
                    alert_copy['alert_start_time'] = datetime.fromtimestamp(alert['startEpoch']).strftime('%Y-%m-%d %H:%M:%S')
                    alert_copy['alert_end_time'] = datetime.fromtimestamp(alert['endEpoch']).strftime('%Y-%m-%d %H:%M:%S')
                    if alert['endEpoch'] == 0:
                        alert_copy['alert_duration'] = datetime.now().timestamp() - alert['startEpoch']
                    else:
                        alert_copy['alert_duration'] = alert['endEpoch'] - alert['startEpoch']
                    alerts_by_month[month_key].append(alert_copy)
                else:
                    print(f"Skipping non-dict alert: {alert}")
            
            print(f"Processing {len(activealerts)} alerts for {outputtime}")
            
            # Save to CSV after each day to prevent data loss
            if datetime.fromtimestamp(start_epoch).hour == 23:
                for month, alerts in alerts_by_month.items():
                    if alerts:  # Only save if there are alerts
                        df = pd.DataFrame(alerts)
                        output_filename = f'alerts_{month}.csv'
                        df.to_csv(output_filename, index=False)
                        print(f"Saved {len(alerts)} alerts to {output_filename}")
    
    except KeyboardInterrupt:
        print("\nScript interrupted by user. Saving current progress...")
        # Save current progress before exiting
        for month, alerts in alerts_by_month.items():
            if alerts:
                df = pd.DataFrame(alerts)
                output_filename = f'alerts_{month}.csv'
                df.to_csv(output_filename, index=False)
                print(f"Saved {len(alerts)} alerts to {output_filename}")
        print("Progress saved. You can resume from the last processed time.")
        exit(0)
    
    # Final save for any remaining data
    for month, alerts in alerts_by_month.items():
        if alerts:  # Only save if there are alerts
            df = pd.DataFrame(alerts)
            output_filename = f'alerts_{month}.csv'
            df.to_csv(output_filename, index=False)
            print(f"Final save: {len(alerts)} alerts to {output_filename}")
    

     

4 Replies

    • Joe_Williams's avatar
      Joe_Williams
      Icon for Professor rankProfessor

      We do the same actually. We have expressed to our CSM that the programmatic approach to generating the swagger lost us the ability to easily consume the methods and the best way to approach the stuff. Honestly this is why I ended up writing my own "sdk" in python.

      But it does appear they are making strides. I have seen a lot more articles show up in the newer section, they just aren't as detailed as the older ones.

  • By default the API only returns active alerts. To return alerts in any state you need to send a filter of cleared with an star.

    ?filter=cleared:"*"

    Just remember tho, if you truly want "every" alert. You can't paginate more then 10 times, so your limit is 10000. You will need to break it up in smaller chunks of time. An example of that would be something like below in python. This gets epochs for every hour of every day for the past X days. Then loops through and gets the alerts for that hour. If your portal is anything like ours, you will still hit that 10000 limit occasionally but it works pretty well.

     

    def generate_hourly_epochs(days_back, start_from=None):
        """
        Generate a list of dictionaries containing start and end epoch times for each hour
        starting from midnight of the current day and going back specified number of days.
        
        Args:
            days_back (int): Number of days to look back
            start_from (datetime): Optional timestamp to start from
            
        Returns:
            list: List of dictionaries with start and end epochs for each hour
        """
        # Get current time and set it to midnight of today
        current_time = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
        start_time = current_time - timedelta(days=days_back)
        
        # If we have a start_from time, use it instead
        if start_from:
            start_time = start_from.replace(minute=0, second=0, microsecond=0)
            print(f"Resuming from: {start_time}")
        
        epochs = []
        
        # Generate epochs for each hour
        current_hour = start_time
        while current_hour <= current_time:
            hour_end = current_hour + timedelta(hours=1)
            epochs.append({
                'start_epoch': int(current_hour.timestamp()),
                'end_epoch': int(hour_end.timestamp())
            })
            current_hour = hour_end
        
        return epochs
    
    last_processed_time = get_last_processed_time()
    epochs = generate_hourly_epochs(120, start_from=last_processed_time)
    alerts_by_month = {}  # Dictionary to store alerts grouped by month
    
    try:
        for epoch in epochs:
            start_epoch = epoch['start_epoch']
            end_epoch = epoch['end_epoch']
    
            activealerts = lm_rest_api.get_alerts(fields=['id','resourceId','instanceName','type','internalId','tenant','monitorObjectName','startEpoch','severity','endEpoch'],filter=f'startEpoch>:"{str(start_epoch)}",startEpoch<:"{str(end_epoch)}",cleared:"*"')
            
            outputtime = datetime.fromtimestamp(start_epoch).strftime('%Y-%m-%d %H:%M:%S')
            month_key = datetime.fromtimestamp(start_epoch).strftime('%Y%m')  # Format: YYYYMM
            
            # Skip if no alerts (boolean response)
            if isinstance(activealerts, bool):
                print(f"No alerts for {outputtime}")
                continue
            
            # Initialize month list if not exists
            if month_key not in alerts_by_month:
                alerts_by_month[month_key] = []
            
            # Add timestamp info to each alert
            for alert in activealerts:
                if isinstance(alert, dict):
                    alert_copy = alert.copy()
                    alert_copy['interval_start'] = datetime.fromtimestamp(start_epoch).strftime('%Y-%m-%d %H:%M:%S')
                    alert_copy['interval_end'] = datetime.fromtimestamp(end_epoch).strftime('%Y-%m-%d %H:%M:%S')
                    alert_copy['alert_start_time'] = datetime.fromtimestamp(alert['startEpoch']).strftime('%Y-%m-%d %H:%M:%S')
                    alert_copy['alert_end_time'] = datetime.fromtimestamp(alert['endEpoch']).strftime('%Y-%m-%d %H:%M:%S')
                    if alert['endEpoch'] == 0:
                        alert_copy['alert_duration'] = datetime.now().timestamp() - alert['startEpoch']
                    else:
                        alert_copy['alert_duration'] = alert['endEpoch'] - alert['startEpoch']
                    alerts_by_month[month_key].append(alert_copy)
                else:
                    print(f"Skipping non-dict alert: {alert}")
            
            print(f"Processing {len(activealerts)} alerts for {outputtime}")
            
            # Save to CSV after each day to prevent data loss
            if datetime.fromtimestamp(start_epoch).hour == 23:
                for month, alerts in alerts_by_month.items():
                    if alerts:  # Only save if there are alerts
                        df = pd.DataFrame(alerts)
                        output_filename = f'alerts_{month}.csv'
                        df.to_csv(output_filename, index=False)
                        print(f"Saved {len(alerts)} alerts to {output_filename}")
    
    except KeyboardInterrupt:
        print("\nScript interrupted by user. Saving current progress...")
        # Save current progress before exiting
        for month, alerts in alerts_by_month.items():
            if alerts:
                df = pd.DataFrame(alerts)
                output_filename = f'alerts_{month}.csv'
                df.to_csv(output_filename, index=False)
                print(f"Saved {len(alerts)} alerts to {output_filename}")
        print("Progress saved. You can resume from the last processed time.")
        exit(0)
    
    # Final save for any remaining data
    for month, alerts in alerts_by_month.items():
        if alerts:  # Only save if there are alerts
            df = pd.DataFrame(alerts)
            output_filename = f'alerts_{month}.csv'
            df.to_csv(output_filename, index=False)
            print(f"Final save: {len(alerts)} alerts to {output_filename}")