Obter informações sobre status do Replicaset MongoDB com Python de forma simples

Boa tarde pessoal, desta vez vim compartilhar como vocês um script python que facilita a visualização de informações do Replicaset MongoDB de forma simples e visualmente melhor.

O Script irá exibir os resultados conforme abaixo:

Segue script python a seguir (verify_status_replicaset.py):

# -*- coding: utf-8 -*-

# usar python 3.10 ou superior
# pip install pandas
# pip install tabulate
# pip install pymongo

#import io
import os
import pandas as pd
from tabulate import tabulate
from pymongo import MongoClient
from datetime import date, timedelta, datetime

mylist = []

# funcao que transforma string csv para lista python
def stringtolist(v_strCsv):
    listAux = v_strCsv.split(',')
    mylist.append(listAux)
    

# funcao de verificacao do status do replicaset mongodb
def replsetstatus():

    user = 'user.rsmonitor'
    pwd = 'pwd.user.rsmonitor'
    serverReplSet = '10.10.9.1:27017,10.10.9.2:27017,10.10.9.3:27017'
    client = MongoClient(f'mongodb://{user}:{pwd}@{serverReplSet}')
    db = client ['admin']
    rs_stats = db.command({'replSetGetStatus': 1})
    replSetName = rs_stats['set']

    print(f'\nReplicaSet Name: {replSetName}')

    for key in rs_stats['members'] :
            
        if key['stateStr'] == 'PRIMARY' : 

            hostnameSrv = 'WKMONGO04' + str(key['_id'])
            IpSrv = key["name"]
            health = key["health"]
            state = key["state"]
            statusRepl = key["stateStr"]
            optimeDate = key['optimeDate']
            syncSourceHost = "IsPrimary"

            strCsv = f"{hostnameSrv},{IpSrv},{statusRepl},{state},{health},{optimeDate},{syncSourceHost}"
            stringtolist(strCsv)
    
        else:

            hostnameSrv = 'WKMONGO04' + str(key['_id'])
            IpSrv = key["name"]
            health = key["health"]
            state = key["state"]
            statusRepl = key["stateStr"]
            optimeDate = key['optimeDate']
            syncSourceHost = key["syncSourceHost"]

            strCsv = f"{hostnameSrv},{IpSrv},{statusRepl},{state},{health},{optimeDate},{syncSourceHost}"
            stringtolist(strCsv)
            
    client.close()

    imprimeResultado()

    v_titleReplicaSetLag = '\n******** LAG REPLICASET (SECONDARY) ********\n'
    cmd = f'mongosh --quiet "mongodb://{serverReplSet}" -u {user} -p {pwd} --authenticationDatabase admin --eval "db.printSecondaryReplicationInfo()"'
    retornoLagMongoDB = os.popen(cmd).read()
    print(f"{v_titleReplicaSetLag}\n{retornoLagMongoDB}")

# funcao que imprime na tela os resultados obtidos
def imprimeResultado():

    pd.set_option('display.expand_frame_repr', False) #ajusta tamanho de exibicao do dataframe
    usecols=['Host','IpSrv','statusRepl','state','health','optimeDate','syncSourceHost']
    df = pd.DataFrame(mylist, columns =usecols)
    df = df.sort_values(by=['Host'], ascending=True)

    print(f'\n********     STATUS REPLICASET     ********\n')

    #imprime em formato de tabela
    #print(f'{df}')
    #print('\n======================================\n')
    #print(tabulate(df, headers=usecols, showindex=False))
    #print('\n======================================\n')
    print(tabulate(df, headers=usecols, showindex=False, tablefmt="pretty"))
    

      
## função inicial 
def main():
    print('######################################################################################################')
    replsetstatus()
    
    dataHoraExecucao = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(f'Data/Hora de execução: {dataHoraExecucao}')
    print('######################################################################################################')
    
    

#inicio da aplicacao
if __name__ == "__main__":
    main()
        

Pode-se configurar o script para obter mais itens do replicaset conforme desejado.

Bom pessoal, por hoje é isso, espero que seja útil para mais alguém além de mim! kkkk

Até a próxima.