ProHoster > Blog > Administrare > pg_stat_statements + pg_stat_activity + loq_query = pg_ash?

pg_stat_statements + pg_stat_activity + loq_query = pg_ash?

Ca un complement scurt la articol Încercarea de a crea un analog ASH pentru PostgreSQL.

Sarcină

Este necesar să corelăm istoricul vizualizărilor pg_stat_statements și pg_stat_activity. Drept urmare, folosind istoricul planurilor de execuție din tabelul de serviciu log_query, se poate obține o mulțime de informații utile pentru a fi utilizate în procesul de rezolvare a incidentelor de performanță și optimizarea interogărilor.

Atenție.

În legătură cu continuarea testării și dezvoltării, articolul nu poate pretinde descrierea unei soluții industriale gata.

Critica și comentariile referitoare la implementare sunt binevenite și așteptate.

Date de intrare

Tabelul history_pg_stat_activity

--ACTIVITY_HIST.HISTORY_PG_STAT_ACTIVITY
DROP TABLE IF EXISTS activity_hist.history_pg_stat_activity;
CREATE TABLE activity_hist.history_pg_stat_activity
(
  timepoint timestamp without time zone ,
  datid             oid  ,
  datname           name ,
  pid               integer,
  usesysid          oid    ,
  usename           name   ,
  application_name  text   ,
  client_addr       inet   ,
  client_hostname   text   ,
  client_port       integer,
  backend_start     timestamp without time zone ,
  xact_start        timestamp without time zone ,
  query_start       timestamp without time zone ,
  state_change      timestamp without time zone ,
  wait_event_type   text ,                     
  wait_event        text ,                   
  state             text ,                  
  backend_xid       xid  ,                 
  backend_xmin      xid  ,                
  query             text ,               
  backend_type      text ,
  queryid           bigint
);

Tabelul pg_stat_db_queries

CREATE TABLE pg_stat_db_queries
(
  database_id integer ,
  queryid bigint ,
  query text ,
  max_time double precision
);

Vedere materializată mvw_pg_stat_queries

CREATE MATERIALIZED VIEW public.mvw_pg_stat_queries AS
 SELECT t.queryid,
    t.max_time,
    t.query
   FROM public.dblink('LINK1'::text, 'SELECT queryid , max_time , query FROM pg_stat_statements WHERE dbid=(SELECT oid FROM  pg_database WHERE datname=current_database() ) AND max_time >= 0 '::text) t(queryid bigint, max_time double precision, query text)
  WITH NO DATA;

Tabelul log_query

CREATE TABLE log_query
(
  id  integer ,
  queryid  bigint ,
  query_md5hash  text ,
  database_id  integer ,
  timepoint  timestamp without time zone , 
  query  text ,
  explained_plan  text[] , 
  plan_md5hash  text ,
  explained_plan_wo_costs  text[] , 
  plan_hash_value  text ,
  ip  text,
  port  text , 
  pid  integer 
);

Algoritmul general

Actualizați tabela pg_stat_db_queries

Actualizați vederea materializată mvw_pg_stat_queries

CREAȚI SAU REÎNLOCUIȚI FUNCȚIA refresh_pg_stat_queries_list( database_id int) ÎNCEPE SĂ $$
DECLARE
 result BOOLEAN ;
 database_rec record ;  
BEGIN   
  SELECT *
  ÎNTO database_rec
  DIN endpoint e JOIN database d PE e.id = d.endpoint_id 
  UNDE d.id = database_id  ;  
  
  DACĂ NU database_rec.is_need_monitoring ATUNCI RAISE NOTICE 'NU ESTE NECESAR MONITORIZARE PENTRU database_id=%',database_id; return TRUE ; END IF ;
  
  EXECUTĂ 'SELECT dblink_connect(''LINK1'',''host='||database_rec.host||' port=5432 dbname='||database_rec.name||
		                                         ' user='||database_rec.s_name||' password='||database_rec.s_pass|| ' '')';
   
  REÎNNOIȚI VIZUALIZAREA MATERIALIZATĂ mvw_pg_stat_queries ;
  
  PERFORM dblink_disconnect('LINK1');  

  RETURN result;
END
$$ LIMBA plpgsql;

Completați tabela pg_stat_db_queries

CREAȚI SAU REÎNLOCUIȚI FUNCȚIA refresh_pg_stat_db_queries( ) ÎNCEPE SĂ $$
DECLARE
 result BOOLEAN ;
 database_rec record ;  
 pg_stat_rec record ;
BEGIN 
  TRUNCATE pg_stat_db_queries;
  
  
  PENTRU database_rec ÎN
  SELECT *
  DIN database d 
  LOOP
  
    DACĂ NU database_rec.is_need_monitoring ATUNCI RAISE NOTICE 'NU ESTE NECESAR MONITORIZARE PENTRU database_id=%',database_rec.id; CONTINUE ; END IF ;
   
    PERFORM refresh_pg_stat_queries_list( database_rec.id ) ; 
	
	PENTRU pg_stat_rec ÎN
	SELECT * 
	DIN mvw_pg_stat_queries 
	LOOP
	  INSERTAȚI ÎN pg_stat_db_queries
	  ( database_id , queryid , query , max_time )
	  VALORI
	  ( database_rec.id , pg_stat_rec.queryid , pg_stat_rec.query , pg_stat_rec.max_time);
	END LOOP;     
  END LOOP; 

  RETURN TRUE;
END
$$ LIMBA plpgsql;

Ca rezultat, tabela conține texte normalizate ale interogărilor, queryid, timpul maxim de execuție a interogării de până acum (folosit pentru monitorizare).

Umplerea log_query și formarea istoricului planurilor de execuție.

Textul actual al interogării este obținut din fișierul de log. Fișierul de log de pe gazda țintă pe gazda de monitorizare este copiat pe părți, folosind un script bash, prin cron. Pentru a economisi spațiu și din cauza trivialității sarcinii de copiere a unei părți din fișierul text de pe gazdă pe gazdă, scriptul nu este prezentat.

Analiza fișierului de log și extragerea textului interogării

#!/bin/bash
#########################################################
# upload_log_query.sh
# Upload table table from dowloaded aws file 
# version 12.0
###########################################################  
echo 'TIMESTAMP:'$(date +%c)' Upload log_query table '

source_file=$1
echo 'source_file='$source_file

database_id=$2
echo 'database_id='$database_id

database_name=$3
echo 'database_name='$database_name


beginer=' '
first_line='1'
let "line_count=0"
sql_line=' '
sql_flag=' '    
space=' '
cat $source_file | while read line
do
  #first line will be passed
  if [[ $line_count == '0' ]]; then 
    let "line_count++" 
    continue 
  fi
  
  line="$space$line"
  #echo 'line='$line

  if [[ $first_line == "1" ]]; then
    beginer=`echo $line | awk -F" " '{ print $1}' `
    first_line='0'
  fi

  current_beginer=`echo $line | awk -F" " '{ print $1}' `
  #echo 'current_beginer='$current_beginer
  #echo 'beginer='$beginer
  

  if [[ $current_beginer == $beginer ]]; then
   if [[ $sql_flag == '1' ]]; then
     sql_flag='0' 
     #echo 'TIMESTAMP:'$(date +%c)' Upload log_query table : SQL STATEMENT ='"$sql_line"

     log_date=`echo $sql_line | awk -F" " '{ print $1}' `
     #echo 'log_date='$log_date

     log_time=`echo $sql_line | awk -F" " '{ print $2}' `
     #echo 'log_time='$log_time
     
     duration=`echo $sql_line | awk -F" " '{ print $5}' `
     #echo 'duration='$duration

	 connect=`echo $sql_line | awk -F" " '{ print $3}' `
	 userdb=`echo $connect | awk -F":" '{ print $3}' `
	 userdb2=$userdb'@'
	 db_port_log=`echo $connect | awk -F"@" '{ print $2}' `
	 log_database_name=`echo $db_port_log | awk -F":" '{ print $1}' `
	 
	 
	 #echo 'connect='$connect
	 #echo 'userdb='$userdb
	 #echo 'userdb2='$userdb2
	 #echo 'db_port_log='$db_port_log
	 #echo 'log_database_name='$log_database_name
	 
	 if [[  "$log_database_name" != "$database_name" ]];
	 then
	   echo '*** database_name '$log_database_name' from log is not equal '$database_name' CONTINUE '
	   continue;
	 fi
	 
     #replace ' to ''
     sql_modline=`echo "$sql_line" | sed 's/'''/''''''/g'`
     sql_line=' '
	 #echo '*********************************log_query start'
     #echo 'pid_str='$pid_str 
	 #echo 'ip_port='$ip_port 
	 #echo 'database_id='$database_id 
	 #echo 'log_date='$log_date
	 #echo 'log_time='$log_time
	 #echo 'duration='$duration
	 #echo 'sql_modline='$sql_modline
     if ! psql -U monitor -d monitor -v ON_ERROR_STOP=1 -A -t -q -c "select log_query( '$pid_str' , '$ip_port' , $database_id , '$log_date' , '$log_time' , '$duration' , '$sql_modline' )"
     then
        echo 'FATAL_ERROR - log_query '
        exit 1
     fi
	 #echo '**********************************log_query finish'

    fi #if [[ $sql_flag == '1' ]]; then

    let "line_count=line_count+1"
    #echo 'line_count= '$line_count
    #echo $line

    #check=`echo $line | awk -F" " '{ print $8}' `
    #check_sql=${check^^}    

    #echo 'check_sql='$check_sql
    
	if [[ ${line^^} =~ "SELECT" ]]; 
	then 
	 if [[ $line =~ "duration:" ]];
	 then
	    test_statement=`echo $line | awk -F" " '{ print $8}'` 
		is_select=${test_statement^^}		
		
		#echo 'test_statement='$test_statement
		#echo 'is_select='$is_select
		
        if [[ $is_select == 'SELECT' ]]; 
        then		
		  sql_flag='1'    
		  sql_line="$sql_line$line"
		  ip_port=`echo $sql_line | awk -F":" '{ print $4}' `
		  pid_str=`echo $sql_line | awk -F":" '{ print $6}' `
		fi
	 fi	 
    fi
  else       
    #echo $line
    #echo 'sql_flag ='$sql_flag

    if [[ $sql_flag == '1' ]]; then
      sql_line="$sql_line$line"
    fi   
    
  fi #if [[ $current_beginer == $beginer ]]; then

done

Umplerea tabelei log_query

--log_query.sql
--inserare o nouă interogare în tabela log_query
CREATE OR REPLACE FUNCTION log_query( pid_str text , ip_port text , log_database_id integer , log_date text , log_time text , duration text , sql_line text ) RETURNS boolean AS $$
DECLARE
  result boolean ;
  log_timepoint timestamp without time zone ;
  log_duration double precision ; 
  pos integer ;
  log_query text ;
  activity_string text ;
  log_md5hash text ;
  log_explain_plan text[] ;
  
  log_planhash text ;
  log_plan_wo_costs text[] ; 
  
  database_rec record ;
  
  pg_stat_query text ; 
  test_log_query text ;
  metric_rec record;
  log_query_rec record;
  found_flag boolean;
  
  pg_stat_history_rec record ;
  port_start integer ;
  port_end integer ;
  client_ip text ;
  client_port text ;
  log_queryid bigint ;
  log_query_text text ;
  pg_stat_query_text text ; 
  current_pid_str text ;
  current_pid integer;
  pid_start_pos integer ;
  pid_finish_pos integer ;
  
BEGIN
  result = TRUE ;    
  
  IF ip_port != '[local]' THEN
    port_start = position('(' in ip_port);
    port_end = position(')' in ip_port);
    client_ip = substring( ip_port from 1 for port_start-1 );
    client_port = substring( ip_port from port_start+1 for port_end-port_start-1 );
  ELSE 
    client_ip = 'local';
    client_port = 'local';
  END IF; 
  
  pid_start_pos = position('[' in pid_str);
  pid_finish_pos = position(']' in pid_str);
  current_pid_str=substring( pid_str from 2 for pid_finish_pos - pid_start_pos -1 );
  current_pid = to_number(current_pid_str , '999999999999');
  
  SELECT e.host , d.name , d.owner_pwd , d.owner_user
  INTO database_rec
  FROM database d JOIN endpoint e ON e.id = d.endpoint_id
  WHERE d.id = log_database_id ;
  
  log_timepoint = to_timestamp(log_date||' '||log_time,'YYYY-MM-DD HH24-MI-SS');
  log_duration = duration::double precision ; 
    
  pos = position ('SELECT' in UPPER(sql_line) );
  log_query = substring( sql_line from pos for LENGTH(sql_line));
  log_query = regexp_replace(log_query,' +',' ','g');
  log_query = regexp_replace(log_query,';+','','g');
  log_query = trim(trailing ' ' from log_query);
  
  log_md5hash = md5( log_query::text );
  
  --Plan de execuție--
EXECUTE 'SELECT dblink_connect(''LINK1'',''host='||database_rec.host||' port=5432 dbname='||database_rec.name||' user='||database_rec.owner_user||' password='||database_rec.owner_pwd||' '')'; 
  
  log_explain_plan = ARRAY ( SELECT * FROM dblink('LINK1', 'EXPLAIN '||log_query ) AS t (plan text) );
  log_plan_wo_costs = ARRAY ( SELECT * FROM dblink('LINK1', 'EXPLAIN ( COSTS FALSE ) '||log_query ) AS t (plan text) );
  
  PERFORM dblink_disconnect('LINK1');
  --------------------------
  BEGIN
    INSERT INTO log_query
    (
        query_md5hash ,
        database_id , 
        timepoint ,
        duration ,
        query ,
        explained_plan ,
        plan_md5hash , 
        explained_plan_wo_costs , 
        plan_hash_value , 
        ip , 
        port ,
        pid
    ) 
    VALUES 
    (
        log_md5hash ,
        log_database_id , 
        log_timepoint , 
        log_duration , 
        log_query ,
        log_explain_plan , 
        md5(log_explain_plan::text) ,
        log_plan_wo_costs , 
        md5(log_plan_wo_costs::text),
        client_ip , 
        client_port , 
        current_pid
    );
    activity_string =  'O nouă interogare a fost înregistrată '||
                    ' database_id = '|| log_database_id ||
                    ' query_md5hash='||log_md5hash||
                    ' , timepoint = '||to_char(log_timepoint,'YYYYMMDD HH24:MI:SS');
    PERFORM pg_log( log_database_id , 'log_query' , activity_string);  

    EXCEPTION
      WHEN unique_violation THEN
        activity_string =  'EXCEPTION *** interogarea a fost deja înregistrată '||
                      ' database_id = '|| log_database_id ||
                      ' query_md5hash='||log_md5hash||
                      ' , timepoint = '||to_char(log_timepoint,'YYYYMMDD HH24:MI:SS');					 
        PERFORM pg_log( log_database_id , 'log_query' , activity_string);
    END;

    SELECT 	queryid
    INTO   	log_queryid
    FROM 	log_query 
    WHERE 	query_md5hash = log_md5hash AND
            timepoint = log_timepoint;

    IF log_queryid IS NOT NULL 
    THEN 
      RETURN result;
    END IF;
    
    ------------------------------------------------
    SELECT * 
    INTO log_query_rec
    FROM log_query
    WHERE query_md5hash = log_md5hash AND timepoint = log_timepoint ; 
    
    log_query_rec.query=regexp_replace(log_query_rec.query,';+','','g');

    FOR pg_stat_history_rec IN
    SELECT 
            queryid ,
            query 
    FROM 
        pg_stat_db_queries 
    WHERE  
        database_id = log_database_id AND
        queryid is not null 
    LOOP
        pg_stat_query = pg_stat_history_rec.query ; 
        pg_stat_query=regexp_replace(pg_stat_query,'n+',' ','g');
        pg_stat_query=regexp_replace(pg_stat_query,'t+',' ','g');
        pg_stat_query=regexp_replace(pg_stat_query,' +',' ','g');
        pg_stat_query=regexp_replace(pg_stat_query,'$.','%','g');

        log_query_text = trim(trailing ' ' from log_query_rec.query);
        pg_stat_query_text = pg_stat_query;   
        
        IF (log_query_text LIKE pg_stat_query_text) THEN
            found_flag = TRUE ;
        ELSE
            found_flag = FALSE ;
        END IF;	  

        IF found_flag THEN
            
            UPDATE log_query SET queryid = pg_stat_history_rec.queryid WHERE query_md5hash = log_md5hash AND timepoint = log_timepoint ;
            activity_string =  ' updated queryid = '||pg_stat_history_rec.queryid||
                            ' pentru log_query cu id = '||log_query_rec.id;					
            EXIT ;
        END IF ;	  
    END LOOP ;	
    RETURN result ;
END
$$ LANGUAGE plpgsql;

În rezultatul acesta, tabelul conține textul actual al interogării, planurile de execuție, hash-ul planului de execuție și hash-ul textului interogării.

Completați valoarea queryid în tabelul history_pg_stat_activity

update_history_pg_stat_activity_by_queryid.sql

--update_history_pg_stat_activity_by_queryid.sql
CREATE OR REPLACE FUNCTION update_history_pg_stat_activity_by_queryid() RETURNS boolean AS $$
DECLARE
  result boolean;
  history_pg_stat_activity_rec record;
  pg_stat_query text;
  pg_stat_query_text text;
  pg_stat_history_rec record;
  found_flag boolean;
  history_pg_stat_activity_query text;
  query_text text;
  activity_string text;

BEGIN
  RAISE NOTICE '***update_history_pg_stat_activity_by_queryid';

  result = TRUE;

  FOR history_pg_stat_activity_rec IN
  SELECT DISTINCT(query) AS query
  FROM activity_hist.history_pg_stat_activity
  WHERE queryid IS NULL
  LOOP
		history_pg_stat_activity_query = regexp_replace(history_pg_stat_activity_rec.query, 'n+', ' ', 'g');
		history_pg_stat_activity_query = regexp_replace(history_pg_stat_activity_query, 't+', ' ', 'g');
		history_pg_stat_activity_query = regexp_replace(history_pg_stat_activity_query, ' +', ' ', 'g');
		history_pg_stat_activity_query = regexp_replace(history_pg_stat_activity_query, ';', '', 'g');
		query_text = trim(trailing ' ' from history_pg_stat_activity_query);
		
		FOR pg_stat_history_rec IN
		SELECT 
			queryid,
			query 
		FROM 
			--pg_stat_history
			pg_stat_db_queries
		WHERE  
			queryid is not null 
		GROUP BY queryid, query
		LOOP
			pg_stat_query = pg_stat_history_rec.query;
			pg_stat_query = regexp_replace(pg_stat_query, 'n+', ' ', 'g');
			pg_stat_query = regexp_replace(pg_stat_query, 't+', ' ', 'g');
			pg_stat_query = regexp_replace(pg_stat_query, ' +', ' ', 'g');
			pg_stat_query = regexp_replace(pg_stat_query, '$.', '%', 'g');
			
			pg_stat_query_text = pg_stat_query;
			
			IF (query_text LIKE pg_stat_query_text) THEN
				found_flag = TRUE;
			ELSE
				found_flag = FALSE;
			END IF;
			
			IF found_flag 
			THEN
				UPDATE activity_hist.history_pg_stat_activity
				SET queryid = pg_stat_history_rec.queryid
				WHERE regexp_replace(regexp_replace(regexp_replace(regexp_replace(query, 'n+', ' ', 'g'), 't+', ' ', 'g'), ' +', ' ', 'g'), ';', '', 'g') 
				      LIKE query_text||'%';
		
				activity_string = 'history_pg_stat_activity a fost actualizată prin queryid = ' || pg_stat_history_rec.queryid;
				RAISE NOTICE '%', activity_string;
				
				PERFORM pg_log(999, 'update_history_pg_stat_activity_by_queryid', activity_string);
				
				EXIT;
				
			END IF;
		END LOOP;
		
		IF NOT found_flag 
		THEN
			activity_string = 'AVERTIZARE : Nu s-a găsit queryid pentru interogarea : ' || query_text;
			
			RAISE NOTICE '%', activity_string;
				
			PERFORM pg_log(999, 'update_history_pg_stat_activity_by_queryid', activity_string);
		END IF;
		
	RAISE NOTICE 'UPDATE log_query dacă interogarea nu a fost înregistrată în fișierul de log';
		
  END LOOP;
  
  RETURN result;
END
$$ LANGUAGE plpgsql;

În rezultatul acesta, tabelul conține valoarea queryid corespunzătoare valorii queryid a interogării.

Rezultatul

Prin corelarea pg_stat_activity, pg_stat_statements, log_query, se pot obține multe informații utile despre interogare, în special:

  • Istoricul planurilor de execuție.
  • Istoricul timpului CPU pentru interogare.
  • Istoricul așteptărilor pentru interogare.

Datele și numeroase rapoarte suplimentare vor fi descrise în articolul următor.

Dezvoltare

Corelând informațiile disponibile cu povestea prezentării pg_locks, se poate obține informația despre ce blocare specifică a așteptat interogarea și, cel mai important, care proces (interogare) a menținut această blocare.

Soluția acestei probleme va fi descrisă în următorul articol. Acum se efectuează teste și ajustări.

Sursa: habr.com

Cumpără un hosting fiabil pentru site-uri cu protecție DDoS, servere VPS VDS 🔥 Cumpără un hosting fiabil pentru site-uri cu protecție DDoS, servere VPS VDS | ProHoster