<jupyter_start><jupyter_text>Our problem In this article we are going to use [psutil](https://psutil.readthedocs.io/en/latest/) and [gputil](https://github.com/anderskm/gputil) to monitor our system usage and then send slack notifications based on simple rules we create In some cases you may want to look at real time monitoring statistics but for the purposes of this demonstration we will just use historical monitoring data with the following rules for slack notifications:    Rules:    * average CPU or memory utilization over past 36 hours is greater than 90% send red alert    * average CPU or memory utilization over past 36 hours is between 80 and 90% send orange alert  Getting and Storing the data First we'll import all the libraries we need<jupyter_code>import psutil
from datetime import datetime
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import time
import os
import os.path
import csv<jupyter_output><empty_output><jupyter_text>This cell block illustrates the tabular format that we are going to store our data into parquet files later.<jupyter_code>timestamp = datetime.now()
cpu_percent = psutil.cpu_percent(interval=0)
memory_percent = psutil.virtual_memory()[2]
disk_percent = psutil.disk_usage('/')[3]

df = pd.DataFrame({'timestamp':[timestamp],
                   'cpu_percent': [cpu_percent],
                   'memory_percent': [memory_percent],
                   'disk_percent': [disk_percent]})

df.head()

#df.to_csv('example.parquet', engine='auto', compression='snappy')<jupyter_output><empty_output><jupyter_text>Let's generate the file structure for our parquet files now which will follow the format of a Parquet dataset that goes down to year directories to month directories and then each day is a separate parquet file. In our script we will be appending to a csv file that gets converted to a parquet file at the end of every day<jupyter_code>#define some basic functions
def root_path():
    """
    OS independent way to return root path
    """
    return os.path.abspath(os.sep)

    

def create_parquet_directory(dataset_name,base_dir):
    
    """
    generates base folder of Parquet dataset
    """
    
    if not os.path.exists(os.path.join(base_dir,dataset_name)):
        os.makedirs(os.path.join(base_dir,dataset_name))
        
def create_csv_dump_dir(dataset_name,base_dir):
    
    """
    generates base folder for csv dump file
    """
    
    if not os.path.exists(os.path.join(base_dir,dataset_name)):
        os.makedirs(os.path.join(base_dir,dataset_name))<jupyter_output><empty_output><jupyter_text>Here we just set some global variables and call these functions we just made to create parquet dataset and csv dump directories<jupyter_code>PARQUET_DATASET = 'Monitoring'
CSV_DUMP_DIR = 'csv_dump'
BASE_DIR = root_path()
CSV_NAME = 'daily_dump.csv'
CSV_FILENAME = BASE_DIR + CSV_DUMP_DIR + '\\'+CSV_NAME

#create directories if they dont exist
create_parquet_directory(PARQUET_DATASET,BASE_DIR)
create_csv_dump_dir(CSV_DUMP_DIR,BASE_DIR)<jupyter_output><empty_output><jupyter_text>When we are actually writing out the data, we'll use the csv DictWriter to write out since that's easier for writing out single rows of data at a time. So here I jsut show how the data will be structured for writing out to the daily dump csv file<jupyter_code>timestamp = datetime.now()
cpu_percent = psutil.cpu_percent(interval=0)
memory_percent = psutil.virtual_memory()[2]
disk_percent = psutil.disk_usage('/')[3]

dic = {'timestamp':timestamp,
 'cpu_percent': cpu_percent,
 'memory_percent': memory_percent,
 'disk_percent': disk_percent}

dic<jupyter_output><empty_output><jupyter_text>For testing we can create some sample rows of data in a test csv file to make sure everything is working. The cell below should create the csv file and write out to it the current monitoring stats we got from above<jupyter_code>file_exists = os.path.isfile(CSV_FILENAME)

with open(CSV_FILENAME, 'a') as csvfile:
    headers = ['timestamp', 'cpu_percent', 'memory_percent','disk_percent']
    writer = csv.DictWriter(csvfile, delimiter=',', lineterminator='\n',fieldnames=headers)

    if not file_exists:
        writer.writeheader()  # file doesn't exist yet, write a header

    writer.writerow({'timestamp': dic['timestamp'], 'cpu_percent': dic['cpu_percent'], 
                     'memory_percent': dic['memory_percent'], 'disk_percent': dic['disk_percent']})<jupyter_output><empty_output><jupyter_text>This function I've created below will take the daily dump csv file and convert it to a csv file. We call the function at the bottom of this cell block to check that it's working as intended<jupyter_code>def daily_csv_to_parquet(timestamp,base_dir,parquet_dataset,return_filepath=False):
    
    """
    function that gets triggered daily to dump previous day's csv to 
    a parquet file
    
    Args:
        timestamp: timestamp taken when process triggered
        base_dir: base directory of dataset
        parquet_dataset: parquet dataset name
        return_filepath: can be set to true to return filepath of file that was created
    """
    #get year month and day
    year = str(timestamp.year)
    month = str(timestamp.month)
    day = str(timestamp.day)
    
    #get directory path 
    year_path = base_dir + parquet_dataset + '\\' + 'Year='+year
    month_path = year_path + '\\'+'Month='+month
    day_path = month_path+'\\'+day+'.parquet'
    
    #create directories for month and year if they dont exist
    if not os.path.exists(year_path):
        os.makedirs(year_path)
        
    if not os.path.exists(month_path):
        os.makedirs(month_path)
    
    #convert csv file to parquet file
    csv_dump_df = pd.read_csv(CSV_FILENAME)
    table = pa.Table.from_pandas(csv_dump_df)
    pq.write_table(table, day_path,compression ='snappy')
    
    if return_filepath:
        return day_path
    
filepath = daily_csv_to_parquet(timestamp,BASE_DIR,PARQUET_DATASET,return_filepath=True)
filepath<jupyter_output><empty_output><jupyter_text>For our final check, we can try to read in the parquet file we just converted<jupyter_code>table = pq.read_table(filepath)
table.to_pandas()<jupyter_output><empty_output><jupyter_text>This is our final loop below, if you run this it will continually run and collect the usage metrics on your PC, dumping them to the csv file during the day, and then at the end of each day, convert the csv file to a parquet file and delete the csv file.You could have this code in a script use a scheduler to set how often you want the process to trigger or set the process to run and control how often it triggers by using time.sleep. Of course I'd recommend the scheduler method but for demonstration purposes I'm just using sleep here.<jupyter_code>try:
    while True:
        #set how often you want this to run
        time.sleep(2)
        #get current timestamp
        timestamp = datetime.now()
        time_str = timestamp.strftime("%Y-%m-%d %H:%M")
        cpu_percent = psutil.cpu_percent(interval=0)
        memory_percent = psutil.virtual_memory()[2]
        disk_percent = psutil.disk_usage('/')[3]

        dic = {'timestamp':timestamp,
         'cpu_percent': cpu_percent,
         'memory_percent': memory_percent,
         'disk_percent': disk_percent}

        file_exists = os.path.isfile(CSV_FILENAME)

        with open(CSV_FILENAME, 'a') as csvfile:
            headers = ['timestamp', 'cpu_percent', 'memory_percent','disk_percent']
            writer = csv.DictWriter(csvfile, delimiter=',', lineterminator='\n',fieldnames=headers)

            if not file_exists:
                writer.writeheader()  # file doesn't exist yet, write a header

            writer.writerow({'timestamp': dic['timestamp'], 'cpu_percent': dic['cpu_percent'], 
                             'memory_percent': dic['memory_percent'], 'disk_percent': dic['disk_percent']})
            
        #check if it's next day, if so dump current file to parquet file and reset csv file
        if time_str[-5:] == '00:00':
            #convert day's csv file to parquet
            daily_csv_to_parquet(timestamp,BASE_DIR,PARQUET_DATASET)
            #remove daily dump file so new day file starts
            os.remove(CSV_FILENAME)
        
except KeyboardInterrupt:
    print('interrupted!')<jupyter_output><empty_output><jupyter_text>Reading in and Summarizing the data So let's assume now we have many days worth of monitoring data to look at and we want to create a script to read in this data, analyze it, and post slack notifications about consistently high usageWe can read in our entire parquet dataset very easily and convert it to a pandas dataframe, this is shown below<jupyter_code>#read in parquet dataset
dataset = pq.ParquetDataset(BASE_DIR+PARQUET_DATASET)
dataset_table = dataset.read(nthreads=4)
full_df = dataset_table.to_pandas()
#convert timestamp to datetime index
full_df['timestamp'] = pd.to_datetime(full_df.timestamp)
full_df = full_df.set_index('timestamp')
full_df.head()<jupyter_output><empty_output><jupyter_text>Now whatever level of granularity we want our time in, we can resample our dataframe to get that. In this case we will resample our dataframe by the hour and use the average of our values as the aggregation metric<jupyter_code>full_dataset_df = full_df.resample('H').mean()
full_dataset_df.head()<jupyter_output><empty_output><jupyter_text>let's recap the logic we need to create:    Rules:    * average CPU or memory utilization over past 36 hours is greater than 90% send red alert    * average CPU or memory utilization over past 36 hours is between 80 and 90% send orange alert    So now that we have the average of our utilization metrics by the hour which is the data format we need in this case, we can implement our logic and post the slack notification accordingly right?But wait, we are forgetting something. Because of the way we designed our daily csv dump logic, it only gets converted to a parquet file at the end of every day. So if we want the most up-to-date data (for ex. past hour data), we will need to read in the csv file as well and merge it with this parquet data for the historical data past 24 hour window<jupyter_code>cur_day_df = pd.read_csv(CSV_FILENAME)
cur_day_df['timestamp'] = pd.to_datetime(cur_day_df.timestamp)
cur_day_df = cur_day_df.set_index('timestamp')
cur_day_df = cur_day_df.resample('H').mean()
cur_day_df.head()<jupyter_output><empty_output><jupyter_text>Now we can just combine our historical parquet file dataset with the current day csv file to get all of our data in one pandas dataframe.However, based on our rules we onl<jupyter_code>final_df = full_dataset_df.append(cur_day_df)
final_df<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.07976146105106224, 'fraction_numerical': 0.0035408125232948193, 'mean_word_length': 4.220330739299611, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 53, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '13', 'dir': 'jupyter-structured-clean-dedup', 'id': '519676', 'license': 'MIT', 'n_tokens_mistral': 3240, 'n_tokens_neox': 3127, 'n_words': 1126, 'path': '_drafts/How to make a Basic Slack Application in Python.ipynb', 'repo_name': 'DavisTownsend/http-DavisTownsend.github.io', 'size': '10732'}