import pandas as pd

import requests

import warnings

from requests.packages.urllib3.exceptions import InsecureRequestWarning

warnings.simplefilter(‘ignore’,InsecureRequestWarning)

import concurrent.futures

import threading

import os

from datetime import datetime

import subprocess

import math

import re

import sys

import json

AddAuditClassPath=os.getenv(“AddAuditClassPath”)

AddHdfsStorageClassPath=os.getenv(“AddHdfsStorageClassPath”)

print(f”AddAuditClassPath path : {AddAuditClassPath}”)

print(f”path of StorageCheckClassPath path : {AddHdfsStorageClassPath}”)

sys.path.append(AddAuditClassPath)

sys.path.append(AddHdfsStorageClassPath)

#sys.path.append(r’/var/app/cudleap2/REG_DEV_2/DRIFT/DeployDir/resources/python_scripts/AddAudit’)

from Add_Audit_Entry import add_audit_entry

#Defines base URL, array of data sources, and query to execute.

FILE_NAME=os.getenv(“FILE_NAME”)

lib_path=os.getenv(“libPath”)

jar_path=os.getenv(“jarPath”)

runid=os.getenv(“runid”)

path_from_table=os.getenv(“hdfsPaths”)

running_server=os.getenv(“servername”)

print(f”FILE_NAME : {FILE_NAME}”)

print(f”lib_path : {lib_path}”)

print(f”jar_path : {jar_path}”)

print(f”runid : {runid}”)

#print(f”print path captured from table : {path_from_table}”)

print(f”server name : {running_server}”)

#to check filled storage capacity of hdfs directory

def storage_check(each_path):

    try:

        print(“inside storage_check method”)

        command = f”hdfs dfs -df {each_path}”

        hdfs_output = subprocess.check_output(command, shell=True, text=True).strip()

        if hdfs_output:

            result_line = hdfs_output.split(“n”)[1]

            get_values = result_line.split()

            total_storage = int(get_values[1])

            used_storage = int(get_values[2])

            used_percent = get_values[4]

            get_percnt_value = re.search(r’d+’, used_percent)

            percent_num = int(get_percnt_value.group())

            print(f’each_path: {each_path}’)

            print(f’total storage: {total_storage}’)

            print(f’used storage: {used_storage}’)

            print(f’used storage % from command: {used_percent}’)

            if percent_num >= 70:

                print(f”HDFS Storage {each_path} has achieved 70% capacity”)

                return True

            else:

                print(f”filled capacity is {percent_num}% of path: {each_path}, hence no cleaning needed yet on HDFS storage Space”)

                return False

        else:

            print(f”storage check command not executed on hdfs”)

    except subprocess.CalledProcessError as e:

        print(“Error in executing hdfs cmd:”, e)

    except Exception as ex:

       print(“error occured:”, ex)

# to check whether hdfs directory exists

def check_hdfs_directory_exist(directory_path):

    try:

        command = f”hdfs dfs -test -d {directory_path}”

        subprocess.run(command, shell=True, check=True)

        print(“trash directory available”)

        return True

    except subprocess.CalledProcessError:

        print(“trash directory not found, trash cleaning skipped”)

        return False

#to convert byte format into human-readable form os storage capacity

def bytes_to_human_readable(size_in_bytes):

    suffixes = [‘B’, ‘KB’, ‘MB’, ‘GB’, ‘TB’, ‘PB’]

    index = 0

    while size_in_bytes >= 1024 and index <len(suffixes) – 1:

        size_in_bytes /= 1024.0

        index += 1

    return f”{size_in_bytes:.1f} {suffixes[index]}”

path_list = json.loads(path_from_table)

for item in path_list:

    path_till_sourcefiles = item[‘Path’]

    print(f”For each individual path: {path_till_sourcefiles}”)

    print(“********Start Storage check**********”)

    get_status = storage_check(path_till_sourcefiles)

    if get_status:

        print(f”HDFS Storage {path_till_sourcefiles} has achieved 70% capacity”)

        print(“********Storage check ended*********”)

        format_path_for_trash = “/”.join(path_till_sourcefiles.split(“/”)[:3]) + “/” + “.Trash”

        print(f”cropped_path till trash folder: {format_path_for_trash}”)

        if check_hdfs_directory_exist(format_path_for_trash):

            print(f”start trash clean of path {format_path_for_trash}”)

            trash_command = f”hadoop fs -rm -r -skipTrash {format_path_for_trash}”

            trash_run = subprocess.run(trash_command, shell=True, check=True)

            if trash_run:

                print(f”trash cleaned of path {format_path_for_trash}”)

            else:

                print(“trash command not executed on server”)

        print(“add into Audit table”)

        print(f”display top 10 voluminous directories”)

        command = f”hdfs dfs -du -s  {path_till_sourcefiles}/* |sort -nr | head -n 10″

        hdfs_cmd = subprocess.run(command, capture_output =True, shell=True, text=True)

        if hdfs_cmd:

            hdfs_output = hdfs_cmd.stdout.strip()

            lines = hdfs_output.splitlines()

            for line in lines:

                parts = line.split()

                size_in_byte = int(parts[0])

                if size_in_byte != 0:

                    large_directories = parts[-1]

                    human_readable_size = bytes_to_human_readable(size_in_byte)

                    print(f”size: {human_readable_size}, directory: {large_directories}”)

                    add_audit_entry(runid,”HDFS_storage_check”,large_directories,”Large_Directory_Size_Check”,f”Large_Size_Directories of server: {running_server}”,”Pass”, f”filled_storage_size: {human_readable_size}”,lib_path,jar_path,FILE_NAME)

                    print(f”Add Audit Entry  process completed with following runid : {runid}”)

        else:

            print(f”fetching top 10 large size directory command does not executed on server”)

            add_audit_entry(runid,”HDFS_storage_check”,f”hdfs command to extract large size directories not executed on server: {running_server}”,”Large_Directory_Size_Check”,f”Large_Size_Directories of server: {running_server}”,”Fail”,”NA”,lib_path,jar_path,FILE_NAME)

            print(f”Add Audit Entry  process completed with following runid : {runid}”)