Untitled

mail@pastecode.io avatar
unknown
plain_text
16 days ago
10 kB
3
Indexable
Never
# Imports

from opensearchpy import OpenSearch
import json
from math import inf
import random
import re
from datetime import datetime
from opensearchpy.client.snapshot import SnapshotClient
from opensearchpy.client.indices import IndicesClient
# Client intialization

host = 'default.elk.inf.use1.cwdevsandbox.cwnet.io'
port = 9200
auth = ('admin', '/hk3UlwmalS72t0dK+B6G+8Q4ZEnSNi9+mENgtWBBeo=')

client = OpenSearch(
    hosts=[{'host': host, 'port': port}],
    http_compress=True,
    http_auth=auth,
    use_ssl=True,
    verify_certs=False,
    ssl_assert_hostname=False,
    ssl_show_warn=False,
    
)

if client.ping():
    print("Client created successfully")

snapshot_client = SnapshotClient(client)
if snapshot_client:
    print("Snapshot client created successfully")

index_client = IndicesClient(client)
if index_client:
    print("Index client created successfully")
# Helper Functions

def extract_date_from_string(string):
    pattern = r'\d+\.\d+'

    match = re.search(pattern, string)

    if match:
        date_str = match.group(0)

        date_obj = datetime.strptime(date_str, '%Y.%m')

        return date_obj
    else:
        return None


def extract_date_from_snapshot(string):
    snapshot_timestamp = string.split('-')[1:]
    snapshot_timestamp = "".join(snapshot_timestamp)
    dt_object = datetime.strptime(snapshot_timestamp, '%Y%m%d%H%M%S')

    return dt_object


def create_repository(repository_name):
    snapshot_client.create_repository(
        repository=repository_name,
        body={
            'type': 's3',
                    'settings': {
                        'bucket': 'pf-elkcluster-default-essnapshotss3bucket-zrop9rmz1gbk',
                    }
        }
    )


def get_repositories():
    response = client.cat.repositories()
    temp = response.split("\n")
    repository_names = []
    for string in temp:
        repository_names.append(string.split(" ")[0])

    return repository_names[:-1]


def take_snapshot(repository_name="os-snapshot-repo"):
    snapshot_settings = {
        "indices": ["*", "-restored_*"],
    }
    today = str(datetime.now())
    year, month, day, hour, minutes, seconds = today[0:4], today[5:
                                                                 7], today[8:10], today[11:13], today[14:16], today[17:19]
    client.snapshot.create(
        repository=repository_name, snapshot=f"snapshot-{year}-{month}-{day}-{hour}-{minutes}-{seconds}", body=snapshot_settings)


def get_snapshots(repository_name):
    snapshots = client.snapshot.get(
        repository=repository_name, snapshot='_all')["snapshots"]
    snapshot_names = []
    for i in snapshots:
        snapshot_names.append(i["snapshot"])

    return snapshot_names


def get_indices_from_snapshot(repository_name, snapshot_name):
    return snapshot_client.get(repository=repository_name, snapshot=snapshot_name)["snapshots"][0]["indices"]


def milliseconds_to_datetime(millis_str):
    millis_int = int(millis_str)
    seconds = millis_int / 1000
    dt_object = datetime.fromtimestamp(seconds)
    return dt_object


def extract_creation_date_from_index(index_name):
    response = index_client.get(index=index_name)
    miliseconds = response[index_name]["settings"]["index"]["creation_date"]
    return milliseconds_to_datetime(miliseconds)


def extract_creation_date_from_restored_index(index_name):
    response = index_client.get(index=f'restored_{index_name}')
    miliseconds = response[f'restored_{index_name}']["settings"]["index"]["creation_date"]
    return milliseconds_to_datetime(miliseconds)


def extract_date_of_deleted_index(repository_name, snapshot_name, index):
    restore_snapshot(repository_name, snapshot_name, index, searchable=True)
    temp = extract_creation_date_from_restored_index(index)
    delete_index(f'restored_{index}')
    return temp


def restore_snapshot(repository_name, snapshot_name, indices, searchable=False):
    client.snapshot.restore(repository=repository_name, snapshot=snapshot_name, body={
        "indices": indices,
        'storage_type': 'remote_snapshot' if searchable else 'local',
        'rename_pattern': "(.+)" if searchable else "",
        "rename_replacement": "restored_$1" if searchable else ""
        # "ignore_unavailable": False,
        # "include_global_state": False
    })


def delete_index(index_name):
    try:
        response = index_client.delete(index=index_name,)
        return response
    except Exception as e:
        print(f"Failed to delete index '{index_name}'. Error: {str(e)}")
        return None


def convert_year_month_day_to_datetime(year, month, day):
    dt_object = datetime(year, month, day)
    return dt_object


def diff_in_days(start_date, end_date):
    return (end_date - start_date).days


def diff_in_weeks(start_date, end_date):
    return (end_date - start_date).days // 7


def formatter(obj):
    return json.dumps(obj, indent=4)


def user_input():
    # region = input("Enter the region (NA, EU, AU): ")

    # valid_regions = ['NA', 'EU', 'AU']
    # while region not in valid_regions:
    #     print("Invalid region entered. Please enter a valid region.")
    #     region = input("Enter the region (NA, EU, AU): ")

    index_name = input("Enter the index name: ")

    start_date_str = input("Enter the start date (YYYY MM DD): ")
    start_date = convert_year_month_day_to_datetime(
        int(start_date_str.split(" ")[0]), int(start_date_str.split(" ")[1]), int(start_date_str.split(" ")[2]))

    end_date_str = input("Enter the end date (YYYY MM DD): ")
    end_date = convert_year_month_day_to_datetime(
        int(end_date_str.split(" ")[0]), int(end_date_str.split(" ")[1]), int(end_date_str.split(" ")[2]))

    return index_name, start_date, end_date


def delete_indices(indices):
    print("Deleting indices:", indices)

    try:
        if type(indices)==type(""):
            client.indices.delete(index=indices, params=None, headers=None)
        else:
            for index in indices:
                client.indices.delete(index=index, params=None, headers=None)
    except Exception as e:
        print(f"An error occurred while deleting indices: {e}")

def delete_restored_indices(indices):
    print("Deleting indices:", indices)

    try:
        if type(indices)==type(""):
            client.indices.delete(index=f'restored_{indices}', params=None, headers=None)
        else:
            for index in indices:
                client.indices.delete(index=f'restored_{index}', params=None, headers=None)
        print("Indices Deleted")
    except Exception as e:
        print(f"An error occurred while deleting indices: {e}")


def get_documents_from_index(index_name, size=inf):
    return client.search({
        "size": size,
        "query": {
            "match_all": {}
        }
    }, index_name)


def filter_snapshots(snapshots, start_date, end_date):
    filtered_snapshots = []
    for snapshot in snapshots:
        snapshot_time = extract_date_from_snapshot(snapshot)
        if start_date <= snapshot_time <= end_date:
            filtered_snapshots.append(snapshot)

    return filtered_snapshots


def extract_indices_from_time_frame(repository_name, micro_service, start_date, end_date):
    res = []
    snapshots = get_snapshots(repository_name)
    filtered_snapshots = filter_snapshots(snapshots, start_date, end_date)
    for snapshot in filtered_snapshots:
        indices = get_indices_from_snapshot(repository_name, snapshot)
        res.extend(indices)
    res = list(set(res))
    filtered = []
    for item in res:
        if micro_service in item:
            filtered.append(item)
    filtered.sort()
    return filtered,filtered_snapshots
# Code

## User Input

# index_name, start_date, end_date = user_input()
## Restore Index

repository = get_repositories()[0]
print(repository)
snapshots = get_snapshots(repository)
print(snapshots)
# print(formatter(snapshots))

extracted_indices,filtered_snapshots = extract_indices_from_time_frame(
    repository, 'security-auditlog', start_date, end_date)

my_snapshot = filtered_snapshots[0]
for i in range(len(extracted_indices)):
    restore_snapshot(repository,my_snapshot,extracted_indices[i],True)
    creation_date = extract_creation_date_from_restored_index(extracted_indices[i])
    if creation_date>start_date:
        extracted_indices = extracted_indices[i:]
        break
    delete_restored_indices(extracted_indices[i])


restore_snapshot("os-snapshot-repo","snapshot-2024-07-04-12-48-26","kc-hamza-008",True)
# print(extract_creation_date_from_index("automation-engine-service-000002-restored"))
get_documents_from_index("automation-engine-service-000002-restored",50000)['took']
get_documents_from_index("restored_kc-hamza-001",500)
# client.snapshot.restore(repository="os-snapshot-repo", snapshot="snapshot-2024-07-04-12-06-18", body={
#     "indices": "kc-hamza-008",
#     'storage_type': 'remote_snapshot',
#     'rename_pattern': "(.+)" ,
#     "rename_replacement": "restored_$1"
#     # "ignore_unavailable": False,
#     # "include_global_state": False
# })

reindex_request_body = {
    "source": {
        "index": "automation-engine-service-000002-restored",
        # "query" : {'term': {'field_name': 'value'}}
        },
    "dest": {
        "index": "kc-hamza-007"
        }
}

client.reindex(reindex_request_body)

settings_response = client.indices.get_settings(index="automation-engine-service-000002-restored")
source_settings = settings_response["automation-engine-service-000002-restored"]

client.indices.put_settings(index="kc-hamza-007", body=source_settings)



def take_snapshot(repository_name="os-snapshot-repo"):
    snapshot_settings = {
        "indices": ["*", "-restored_*"],
    }
    today = str(datetime.now())
    year, month, day, hour, minutes, seconds = today[0:4], today[5:
                                                                 7], today[8:10], today[11:13], today[14:16], today[17:19]
    client.snapshot.create(
        repository=repository_name, snapshot=f"snapshot-{year}-{month}-{day}-{hour}-{minutes}-{seconds}", body=snapshot_settings)
    
take_snapshot()
Leave a Comment