_collect_metrics()Por que medir?
Em automações que alteram estado (ex.: atualizar perfis de usuários no GLPI), mensurar o antes e o depois é indispensável para comprovar o efeito da regra, auditar impactos, criar indicadores e sustentar decisões operacionais.
Na GSERV/SETIC, padronizamos a coleta com o método_collect_metrics()e duas tasks Airflow:quantificar_antesequantificar_depois.
A função _collect_metrics() retorna um dicionário de métricas que descreve o cenário atual do banco em relação a uma regra específica (no exemplo, o órgão PC e o perfil 140):
total_users: total de usuários em glpi_users.pc_users_like: quantos usuários têm user_dn compatível com o filtro (ex.: "%OU=PC_%").users_profiles_id_140: quantos usuários estão com profiles_id = 140 em glpi_users.links_profiles_users_140: quantos vínculos existem em glpi_profiles_users com profiles_id = 140.sample_pc_users: amostra de até 10 usuários do filtro, para inspeção rápida.Esses nomes são exemplos parametrizados. Em outra instituição, você troca o DN e o perfil padrão (e até adiciona métricas) sem mudar a estrutura do pipeline.
Fluxo típico (exemplo):
quantificar_antes: captura o estado inicial → referência de baseline.
aplicar_regra_atualizacao: executa a regra de negócio (UPDATE/INSERT/etc.).
quantificar_depois: mede o estado final.
comparar_e_gerar_relatorio: consolida diferenças, KPI’s e evidências.
No Airflow, as tasks @task retornam o dicionário e isso é armazenado em XCom, facilitando a comparação posterior.
def _collect_metrics():
"""
Coleta métricas (antes/depois):
- total_users
- pc_users_like (user_dn LIKE %OU=PC_%)
- users_profiles_id_140 (glpi_users.profiles_id = 140)
- links_profiles_users_140 (glpi_profiles_users com profiles_id = 140)
- sample_pc_users (amostra)
"""
metrics = {}
conn = cursor = None
try:
# 0) Conexão — usa a Connection do Airflow (BaseHook) em get_mysql_connection()
conn = get_mysql_connection()
cursor = conn.cursor(dictionary=True)
# 1) Universo: total de usuários
cursor.execute("SELECT COUNT(*) AS c FROM glpi_users;")
metrics["total_users"] = cursor.fetchone()["c"]
# 2) Segmento-alvo: usuários que pertencem ao órgão (DN)
cursor.execute(
"SELECT COUNT(*) AS c FROM glpi_users WHERE user_dn LIKE %s;",
(USER_DN_LIKE,), # ex.: "%OU=PC_%"
)
metrics["pc_users_like"] = cursor.fetchone()["c"]
# 3) Aderência na tabela principal: usuários já com o perfil padrão aplicado
cursor.execute(
"SELECT COUNT(*) AS c FROM glpi_users WHERE profiles_id = %s;",
(PROFILE_ID_PC,), # ex.: 140
)
metrics["users_profiles_id_140"] = cursor.fetchone()["c"]
# 4) Aderência na tabela de vínculos (autorizações)
cursor.execute(
"""
SELECT COUNT(*) AS c
FROM glpi_profiles_users
WHERE profiles_id = %s;
""",
(PROFILE_ID_PC,),
)
metrics["links_profiles_users_140"] = cursor.fetchone()["c"]
# 5) Amostra para inspeção (debug e auditoria)
cursor.execute(
"""
SELECT id, name, user_dn, profiles_id
FROM glpi_users
WHERE user_dn LIKE %s
LIMIT 10;
""",
(USER_DN_LIKE,),
)
metrics["sample_pc_users"] = cursor.fetchall()
return metrics
except Error as e:
# Padroniza a superfície de erro para logs e alertas
raise RuntimeError(f"[ERRO _collect_metrics] {e}")
finally:
# Evita vazamento de recursos
if cursor: cursor.close()
if conn: conn.close()
O exemplo abaixo apresenta indicadores que são fundamentais para quantificar os resultados e obter as métricas necessárias para os relatórios.
Centralize a conexão em get_mysql_connection(). Isso simplifica troca de credenciais/host e permite ligar autocommit/SSL de forma padronizada.
Define o universo. Útil para normalizar indicadores (percentuais) e identificar outliers (crescimentos anômalos entre execuções).
Mede o tamanho do segmento elegível pela regra (orgão/OU).
Índice recomendado: INDEX(user_dn) para acelerar LIKE (especialmente se o padrão começar com prefixo fixo, ex.: OU=PC_).
Indica a aderência atual da regra na tabela principal (glpi_users). Compara-se antes vs. depois para verificar o impacto real (quantos passaram a ter o perfil).
Confirma a coerência na tabela de vinculações (glpi_profiles_users). É possível ter profiles_id ajustado em glpi_users e não refletir em vínculos — essa métrica ajuda a detectar inconsistências.
Uma amostra para inspeção manual e auditoria (log, relatório HTML/CSV).
Em incidentes, essa amostra acelera a identificação de casos representativos.
Lança RuntimeError padronizado para facilitar alertas/observabilidade e fecha recursos no finally.
Este padrão foi criado para a GSERV visando:
Rastreabilidade: provar em relatório o que mudou entre antes/depois.
Qualidade de dados: cruzar a aplicação da regra entre tabelas (principal e vínculo) e detectar desvios.
Governança: unificar nomes de métricas, formato de retorno e pontos de coleta para todas as DAGs de perfis (SESAU, SEJUS, PC, etc.).
Decisão: alimentar painéis/KPIs (quantos usuários elegíveis, quantos aderentes, % de aplicação por órgão, tendências temporais).
Auditoria & rollback: com métricas e amostras, fica mais simples justificar mudanças e reverter se necessário.
from airflow import DAG
from airflow.decorators import task
from datetime import datetime
import json
@task
def quantificar_antes():
return _collect_metrics() # XCom: dict
@task
def aplicar_regra():
# UPDATEs/INSERTs conforme a regra de negócio
# ...
return "ok"
@task
def quantificar_depois():
return _collect_metrics() # XCom: dict
@task
def comparar_e_gerar_relatorio(m_before: dict, m_after: dict):
diff = {
"delta_users_profiles_id_140":
m_after["users_profiles_id_140"] - m_before["users_profiles_id_140"],
"delta_links_profiles_users_140":
m_after["links_profiles_users_140"] - m_before["links_profiles_users_140"],
}
print("[ANTES]", json.dumps(m_before, ensure_ascii=False, indent=2))
print("[DEPOIS]", json.dumps(m_after, ensure_ascii=False, indent=2))
print("[DIFERENÇAS]", json.dumps(diff, ensure_ascii=False, indent=2))
# Aqui você pode salvar CSV/HTML no /opt/airflow/output/..., enviar e-mail, etc.
return diff
with DAG(
dag_id="glpi_pc_perfil_padrao",
start_date=datetime(2024, 1, 1),
schedule="@daily",
catchup=False,
) as dag:
antes = quantificar_antes()
regra = aplicar_regra()
depois = quantificar_depois()
comparar_e_gerar_relatorio(antes, depois) # dependências implícitas via XCom
Para reuso da mesma DAG entre órgãos, parametrize:
USER_DN_LIKE: ex.: «%OU=PC_%», «%OU=SESAU%».
PROFILE_ID_PC: ex.: 140 (troque para o perfil padrão do órgão).
Airflow Variables/Params: centralize em Variable.get(«gserv_profile_id») e Variable.get(«gserv_user_dn_like»), ou use dag.params.
Exemplo:
from airflow.models import Variable
PROFILE_ID = int(Variable.get("gserv_profile_id", default_var="140"))
USER_DN_LIKE = Variable.get("gserv_user_dn_like", default_var="%OU=PC_%")
Índices:
Paginação em amostras: mantenha LIMIT pequeno.
Janela de execução: agendar em horários de carga menor do banco.
Observabilidade: logar métricas principais e salvar relatórios CSV/HTML por execução (timestamp em pasta output/…).
{
"total_users": 52731,
"pc_users_like": 8412,
"users_profiles_id_140": 7890,
"links_profiles_users_140": 7885,
"sample_pc_users": [
{"id": 101, "name": "Ana Souza", "user_dn": "CN=ana,...,OU=PC_...", "profiles_id": 140},
{"id": 102, "name": "Carlos Lima", "user_dn": "CN=carlos,...,OU=PC_...", "profiles_id": 1}
]
}
Em um de-para simples, a diferença entre pc_users_like e users_profiles_id_140 sinaliza quantos ainda faltam receber o perfil; divergência com links_profiles_users_140 aponta vínculos inconsistentes.
O padrão:
Prova o efeito da automação com números.
Eleva a qualidade (coerência entre tabelas).
Facilita auditoria e governança.
Cria base de decisão para evoluir regras e priorizar novas DAGs.
Esse é o padrão GSERV para pipelines que alteram estado no GLPI (e sistemas correlatos), e deve ser replicado/adaptado em todas as novas DAGs de automação institucional.