Oder ein wenig anwendungsbezogene Tetris-Theorie.
Alles Neue ist gut vergessenes Altes.
Epigrafen.
Aufgabenstellung
Es ist notwendig, regelmäßig die aktuelle Protokolldatei von PostgreSQL aus der AWS-Cloud auf einen lokalen Linux-Host zu laden. Nicht in Echtzeit, aber sagen wir mal mit einer kleinen Verzögerung.
Die Ladeperiode der Protokolldatei beträgt 5 Minuten.
Die Protokolldatei in AWS wird stündlich rotiert.
Verwendete Werkzeuge
Um die Protokolldatei auf den Host herunterzuladen, wird ein Bash-Skript verwendet, das die AWS-API "».
Parameter:
- —db-instance-identifier: Name der Instanz in AWS;
- —log-file-name: Name der aktuellen generierten Protokolldatei
- —max-item: Die Gesamtanzahl der Elemente, die in den Ausgaben des Befehls zurückgegeben werden.Die Größe des heruntergeladenen Dateipakets.
- —starting-token: Starttoken des Pakets
In diesem speziellen Fall entstand die Aufgabe, die Protokolle herunterzuladen, während der Arbeiten an
Ja, und einfach — eine interessante Aufgabe, um zu üben und für Abwechslung während der Arbeitszeit zu sorgen.
Ich nehme an, dass die Aufgabe aufgrund ihrer Alltäglichkeit bereits gelöst wurde. Aber eine schnelle Google-Suche nach Lösungen hat nichts ergeben, und es gab nicht viel Lust, intensiver zu suchen. Auf jeden Fall — eine gute Übung.
Formalisation der Aufgabe
Die endgültige Protokolldatei besteht aus mehreren Zeilen variabler Länge. Grafisch kann die Protokolldatei etwa so dargestellt werden:
Erinnert es an etwas? Was hat Tetris damit zu tun? Das hier.
Wenn man die möglichen Varianten, die beim Laden der nächsten Datei auftreten, grafisch darstellt (zum Vereinfachen, in diesem Fall haben die Zeilen eine einheitliche Länge), erhält man Standard-Tetris-Figuren:
1) Die Datei wurde vollständig geladen und ist endgültig. Die Paketgröße ist größer als die Größe der endgültigen Datei:
2) Die Datei hat eine Fortsetzung. Die Paketgröße ist kleiner als die Größe der endgültigen Datei:
3) Die Datei ist eine Fortsetzung der vorherigen Datei und hat eine Fortsetzung. Die Paketgröße ist kleiner als der Rest der endgültigen Datei:
4) Die Datei ist eine Fortsetzung der vorherigen Datei und ist endgültig. Die Paketgröße ist größer als der Rest der endgültigen Datei:
Die Aufgabe besteht darin, ein Rechteck zu bilden oder auf einer neuen Ebene Tetris zu spielen.
Probleme, die während der Lösung der Aufgabe auftreten
1) Zwei Pakete zu einer Zeichenkette zusammenfügen.
Eigentlich sind keine besonderen Probleme aufgetreten. Eine Standardaufgabe aus dem Anfangskurs der Programmierung.
Optimale Paketgröße
Das ist jedoch etwas interessanter.
Leider ist es nicht möglich, eine Versatzangabe nach dem Anfangs-Token zu verwenden:
Wie Sie bereits wissen, wird die Option —starting-token verwendet, um den Startpunkt der Paginierung anzugeben. Diese Option akzeptiert String-Werte, was bedeutet, dass, wenn Sie versuchen, einen Versatzwert vor dem Next Token-String hinzuzufügen, die Option nicht als Versatz berücksichtigt wird.
Deshalb muss man in Stücken lesen.
Wenn man in großen Portionen liest, ist die Anzahl der Lesevorgänge minimal, aber das Volumen ist maximal.
Wenn man in kleinen Portionen liest, ist es umgekehrt: Die Anzahl der Lesevorgänge ist maximal, aber das Volumen ist minimal.
Deshalb musste, um den Traffic zu reduzieren und die Lösung ansprechender zu gestalten, eine gewisse Lösung entwickelt werden, die leider ein wenig wie ein Workaround wirkt.
Zur Veranschaulichung betrachten wir den Prozess des Hochladens einer Logdatei in zwei stark vereinfachten Varianten. Die Anzahl der Lesevorgänge hängt in beiden Fällen von der Portionsgröße ab.
1) Wir laden in kleinen Portionen hoch:
2) Wir laden in großen Portionen hoch:
Wie gewohnt ist die optimale Lösung die Mitte..
Die Portionsgröße ist minimal, aber während des Lesens kann die Größe erhöht werden, um die Anzahl der Lesevorgänge zu reduzieren.
Es sollte angemerkt werden, dass die vollständige Lösung des Problems der Bestimmung der optimalen Größe des zu lesenden Portions noch nicht abgeschlossen ist und eine tiefere Analyse und Bearbeitung erfordert. Vielleicht später.
Allgemeine Beschreibung der Implementierung
Verwendete Servicetabellen
CREATE TABLE endpoint
(
id SERIAL ,
host text
);
TABLE database
(
id SERIAL ,
…
last_aws_log_time text ,
last_aws_nexttoken text ,
aws_max_item_size integer
);
last_aws_log_time – Zeitstempel der zuletzt hochgeladenen Logdatei im Format YYYY-MM-DD-HH24.
last_aws_nexttoken – Textmarke der zuletzt hochgeladenen Portion.
aws_max_item_size – empirisch ermittelter initialer Portionsgröße.
Vollständiger Skripttext
download_aws_piece.sh
#!/bin/bash
#########################################################
# download_aws_piece.sh
# downloan piece of log from AWS
# version HABR
let min_item_size=1024
let max_item_size=1048576
let growth_factor=3
let growth_counter=1
let growth_counter_max=3
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh:''STARTED'
AWS_LOG_TIME=$1
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh:AWS_LOG_TIME='$AWS_LOG_TIME
database_id=$2
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh:database_id='$database_id
RESULT_FILE=$3
endpoint=`psql -h MONITOR_ENDPOINT.rds.amazonaws.com -U USER -d MONITOR_DATABASE_DATABASE -A -t -c "select e.host from endpoint e join database d on e.id = d.endpoint_id where d.id = $database_id "`
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh:endpoint='$endpoint
db_instance=`echo $endpoint | awk -F"." '{print toupper($1)}'`
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh:db_instance='$db_instance
LOG_FILE=$RESULT_FILE'.tmp_log'
TMP_FILE=$LOG_FILE'.tmp'
TMP_MIDDLE=$LOG_FILE'.tmp_mid'
TMP_MIDDLE2=$LOG_FILE'.tmp_mid2'
current_aws_log_time=`psql -h MONITOR_ENDPOINT.rds.amazonaws.com -U USER -d MONITOR_DATABASE -A -t -c "select last_aws_log_time from database where id = $database_id "`
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh:current_aws_log_time='$current_aws_log_time
if [[ $current_aws_log_time != $AWS_LOG_TIME ]];
then
is_new_log='1'
if ! psql -h MONITOR_ENDPOINT.rds.amazonaws.com -U USER -d MONITOR_DATABASE -v ON_ERROR_STOP=1 -A -t -q -c "update database set last_aws_log_time = '$AWS_LOG_TIME' where id = $database_id "
then
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: FATAL_ERROR - update database set last_aws_log_time .'
exit 1
fi
else
is_new_log='0'
fi
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh:is_new_log='$is_new_log
let last_aws_max_item_size=`psql -h MONITOR_ENDPOINT.rds.amazonaws.com -U USER -d MONITOR_DATABASE -A -t -c "select aws_max_item_size from database where id = $database_id "`
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: last_aws_max_item_size='$last_aws_max_item_size
let count=1
if [[ $is_new_log == '1' ]];
then
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: START DOWNLOADING OF NEW AWS LOG'
if ! aws rds download-db-log-file-portion
--max-items $last_aws_max_item_size
--region REGION
--db-instance-identifier $db_instance
--log-file-name error/postgresql.log.$AWS_LOG_TIME > $LOG_FILE
then
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: FATAL_ERROR - Could not get log from AWS .'
exit 2
fi
else
next_token=`psql -h MONITOR_ENDPOINT.rds.amazonaws.com -U USER -d MONITOR_DATABASE -v ON_ERROR_STOP=1 -A -t -c "select last_aws_nexttoken from database where id = $database_id "`
if [[ $next_token == '' ]];
then
next_token='0'
fi
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: CONTINUE DOWNLOADING OF AWS LOG'
if ! aws rds download-db-log-file-portion
--max-items $last_aws_max_item_size
--starting-token $next_token
--region REGION
--db-instance-identifier $db_instance
--log-file-name error/postgresql.log.$AWS_LOG_TIME > $LOG_FILE
then
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: FATAL_ERROR - Could not get log from AWS .'
exit 3
fi
line_count=`cat $LOG_FILE | wc -l`
let lines=$line_count-1
tail -$lines $LOG_FILE > $TMP_MIDDLE
mv -f $TMP_MIDDLE $LOG_FILE
fi
next_token_str=`cat $LOG_FILE | grep NEXTTOKEN`
next_token=`echo $next_token_str | awk -F" " '{ print $2}' `
grep -v NEXTTOKEN $LOG_FILE > $TMP_FILE
if [[ $next_token == '' ]];
then
cp $TMP_FILE $RESULT_FILE
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: NEXTTOKEN NOT FOUND - FINISH '
rm $LOG_FILE
rm $TMP_FILE
rm $TMP_MIDDLE
rm $TMP_MIDDLE2
exit 0
else
psql -h MONITOR_ENDPOINT.rds.amazonaws.com -U USER -d MONITOR_DATABASE -v ON_ERROR_STOP=1 -A -t -q -c "update database set last_aws_nexttoken = '$next_token' where id = $database_id "
fi
first_str=`tail -1 $TMP_FILE`
line_count=`cat $TMP_FILE | wc -l`
let lines=$line_count-1
head -$lines $TMP_FILE > $RESULT_FILE
###############################################
# MAIN CIRCLE
let count=2
while [[ $next_token != '' ]];
do
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: count='$count
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: START DOWNLOADING OF AWS LOG'
if ! aws rds download-db-log-file-portion
--max-items $last_aws_max_item_size
--starting-token $next_token
--region REGION
--db-instance-identifier $db_instance
--log-file-name error/postgresql.log.$AWS_LOG_TIME > $LOG_FILE
then
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: FATAL_ERROR - Could not get log from AWS .'
exit 4
fi
next_token_str=`cat $LOG_FILE | grep NEXTTOKEN`
next_token=`echo $next_token_str | awk -F" " '{ print $2}' `
TMP_FILE=$LOG_FILE'.tmp'
grep -v NEXTTOKEN $LOG_FILE > $TMP_FILE
last_str=`head -1 $TMP_FILE`
if [[ $next_token == '' ]];
then
concat_str=$first_str$last_str
echo $concat_str >> $RESULT_FILE
line_count=`cat $TMP_FILE | wc -l`
let lines=$line_count-1
tail -$lines $TMP_FILE >> $RESULT_FILE
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: NEXTTOKEN NOT FOUND - FINISH '
rm $LOG_FILE
rm $TMP_FILE
rm $TMP_MIDDLE
rm $TMP_MIDDLE2
exit 0
fi
if [[ $next_token != '' ]];
then
let growth_counter=$growth_counter+1
if [[ $growth_counter -gt $growth_counter_max ]];
then
let last_aws_max_item_size=$last_aws_max_item_size*$growth_factor
let growth_counter=1
fi
if [[ $last_aws_max_item_size -gt $max_item_size ]];
then
let last_aws_max_item_size=$max_item_size
fi
psql -h MONITOR_ENDPOINT.rds.amazonaws.com -U USER -d MONITOR_DATABASE -A -t -q -c "update database set last_aws_nexttoken = '$next_token' where id = $database_id "
concat_str=$first_str$last_str
echo $concat_str >> $RESULT_FILE
line_count=`cat $TMP_FILE | wc -l`
let lines=$line_count-1
#############################
#Get middle of file
head -$lines $TMP_FILE > $TMP_MIDDLE
line_count=`cat $TMP_MIDDLE | wc -l`
let lines=$line_count-1
tail -$lines $TMP_MIDDLE > $TMP_MIDDLE2
cat $TMP_MIDDLE2 >> $RESULT_FILE
first_str=`tail -1 $TMP_FILE`
fi
let count=$count+1
done
#
#################################################################
exit 0
Teile des Skripts mit einigen Erklärungen:
Eingabeparameter des Skripts:
- Zeitstempel des Namens der Logdatei im Format YYYY-MM-DD-HH24: AWS_LOG_TIME=$1
- Datenbank-ID: database_id=$2
- Name der gesammelten Logdatei: RESULT_FILE=$3
Aktuellen Zeitstempel der letzten hochgeladenen Logdatei abrufen:
current_aws_log_time=`psql -h MONITOR_ENDPOINT.rds.amazonaws.com -U USER -d MONITOR_DATABASE -A -t -c "select last_aws_log_time from database where id = $database_id "`Wenn der Zeitstempel der letzten hochgeladenen Logdatei nicht mit dem Eingabeparameter übereinstimmt, wird eine neue Logdatei hochgeladen:
Wenn [[ $current_aws_log_time != $AWS_LOG_TIME ]];
dann
is_new_log='1'
wenn ! psql -h ENDPOINT.rds.amazonaws.com -U USER -d MONITOR_DATABASE -v ON_ERROR_STOP=1 -A -t -c "update database set last_aws_log_time = '$AWS_LOG_TIME' where id = $database_id "
dann
echo '***download_aws_piece.sh -FATAL_ERROR - update database set last_aws_log_time .'
exit 1
fi
sonst
is_new_log='0'
fi
Wir erhalten den Wert des Tags nexttoken aus der heruntergeladenen Datei:
next_token_str=`cat $LOG_FILE | grep NEXTTOKEN`
next_token=`echo $next_token_str | awk -F" " '{ print $2}' `
Ein leerer Wert für nexttoken ist ein Indikator für das Ende des Uploads.
In einer Schleife zählen wir die Portionen der Datei, verbinden dabei die Zeilen und erhöhen die Größe der Portion:
Die Hauptschleife
# MAIN CIRCLE
let count=2
while [[ $next_token != '' ]];
do
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: count='$count
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: START DOWNLOADING OF AWS LOG'
if ! aws rds download-db-log-file-portion
--max-items $last_aws_max_item_size
--starting-token $next_token
--region REGION
--db-instance-identifier $db_instance
--log-file-name error/postgresql.log.$AWS_LOG_TIME > $LOG_FILE
then
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: FATAL_ERROR - Could not get log from AWS .'
exit 4
fi
next_token_str=`cat $LOG_FILE | grep NEXTTOKEN`
next_token=`echo $next_token_str | awk -F" " '{ print $2}' `
TMP_FILE=$LOG_FILE'.tmp'
grep -v NEXTTOKEN $LOG_FILE > $TMP_FILE
last_str=`head -1 $TMP_FILE`
if [[ $next_token == '' ]];
then
concat_str=$first_str$last_str
echo $concat_str >> $RESULT_FILE
line_count=`cat $TMP_FILE | wc -l`
let lines=$line_count-1
tail -$lines $TMP_FILE >> $RESULT_FILE
echo $(date +%Y%m%d%H%M)': download_aws_piece.sh: NEXTTOKEN NOT FOUND - FINISH '
rm $LOG_FILE
rm $TMP_FILE
rm $TMP_MIDDLE
rm $TMP_MIDDLE2
exit 0
fi
if [[ $next_token != '' ]];
then
let growth_counter=$growth_counter+1
if [[ $growth_counter -gt $growth_counter_max ]];
then
let last_aws_max_item_size=$last_aws_max_item_size*$growth_factor
let growth_counter=1
fi
if [[ $last_aws_max_item_size -gt $max_item_size ]];
then
let last_aws_max_item_size=$max_item_size
fi
psql -h MONITOR_ENDPOINT.rds.amazonaws.com -U USER -d MONITOR_DATABASE -A -t -q -c "update database set last_aws_nexttoken = '$next_token' where id = $database_id "
concat_str=$first_str$last_str
echo $concat_str >> $RESULT_FILE
line_count=`cat $TMP_FILE | wc -l`
let lines=$line_count-1
#############################
#Get middle of file
head -$lines $TMP_FILE > $TMP_MIDDLE
line_count=`cat $TMP_MIDDLE | wc -l`
let lines=$line_count-1
tail -$lines $TMP_MIDDLE > $TMP_MIDDLE2
cat $TMP_MIDDLE2 >> $RESULT_FILE
first_str=`tail -1 $TMP_FILE`
fi
let count=$count+1
done
Was kommt als Nächstes?
Damit ist die erste Zwischenaufgabe – 'Das Logfile aus der Cloud laden' – gelöst. Was tun wir mit dem heruntergeladenen Log?
Zunächst müssen wir das Logfile analysieren und die Anfragen herausfiltern.
Die Aufgabe ist nicht allzu kompliziert. Ein einfacher Bash-Script erledigt das problemlos.
upload_log_query.sh
#!/bin/bash
#########################################################
# upload_log_query.sh
# Upload table table from dowloaded aws file
# version HABR
###########################################################
echo 'TIMESTAMP:'$(date +%c)' Upload log_query table '
source_file=$1
echo 'source_file='$source_file
database_id=$2
echo 'database_id='$database_id
beginer=' '
first_line='1'
let "line_count=0"
sql_line=' '
sql_flag=' '
space=' '
cat $source_file | while read line
do
line="$space$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}' `
if [[ $current_beginer == $beginer ]]; then
if [[ $sql_flag == '1' ]]; then
sql_flag='0'
log_date=`echo $sql_line | awk -F" " '{ print $1}' `
log_time=`echo $sql_line | awk -F" " '{ print $2}' `
duration=`echo $sql_line | awk -F" " '{ print $5}' `
#replace ' to ''
sql_modline=`echo "$sql_line" | sed 's/'''/''''''/g'`
sql_line=' '
################
#PROCESSING OF THE SQL-SELECT IS HERE
if ! psql -h ENDPOINT.rds.amazonaws.com -U USER -d DATABASE -v ON_ERROR_STOP=1 -A -t -c "select log_query('$ip_port',$database_id , '$log_date' , '$log_time' , '$duration' , '$sql_modline' )"
then
echo 'FATAL_ERROR - log_query '
exit 1
fi
################
fi #if [[ $sql_flag == '1' ]]; then
let "line_count=line_count+1"
check=`echo $line | awk -F" " '{ print $8}' `
check_sql=${check^^}
#echo 'check_sql='$check_sql
if [[ $check_sql == 'SELECT' ]]; then
sql_flag='1'
sql_line="$sql_line$line"
ip_port=`echo $sql_line | awk -F":" '{ print $4}' `
fi
else
if [[ $sql_flag == '1' ]]; then
sql_line="$sql_line$line"
fi
fi #if [[ $current_beginer == $beginer ]]; then
done
Jetzt können wir mit der aus dem Logfile extrahierten Anfrage arbeiten.
Und es eröffnen sich mehrere nützliche Möglichkeiten.
Die extrahierten Anfragen müssen irgendwo gespeichert werden. Dazu wird eine Servicetabelle verwendet. log_query
CREATE TABLE log_query
(
id SERIAL ,
queryid bigint ,
query_md5hash text not null ,
database_id integer not null ,
timepoint timestamp without time zone not null,
duration double precision not null ,
query text not null ,
explained_plan text[],
plan_md5hash text ,
explained_plan_wo_costs text[],
plan_hash_value text ,
baseline_id integer ,
ip text ,
port text
);
ALTER TABLE log_query ADD PRIMARY KEY (id);
ALTER TABLE log_query ADD CONSTRAINT queryid_timepoint_unique_key UNIQUE (queryid, timepoint );
ALTER TABLE log_query ADD CONSTRAINT query_md5hash_timepoint_unique_key UNIQUE (query_md5hash, timepoint );
CREATE INDEX log_query_timepoint_idx ON log_query (timepoint);
CREATE INDEX log_query_queryid_idx ON log_query (queryid);
ALTER TABLE log_query ADD CONSTRAINT database_id_fk FOREIGN KEY (database_id) REFERENCES database (id) ON DELETE CASCADE ;
Die Verarbeitung der analysierten Anfrage erfolgt in plpgsql Funktion 'log_query».
log_query.sql
--log_query.sql
--version HABR
CREATE OR REPLACE FUNCTION log_query( 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 ;
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 ;
BEGIN
result = TRUE ;
RAISE NOTICE '***log_query';
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 );
SELECT e.host , d.name , d.owner_pwd
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 );
--Execution Plan Erklären--
EXECUTE 'SELECT dblink_connect(''LINK1'',''host='||database_rec.host||' dbname='||database_rec.name||' user=DATABASE 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
)
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
);
activity_string = 'Neue Abfrage wurde protokolliert '||
' database_id = '|| log_database_id ||
' query_md5hash='||log_md5hash||
' , timepoint = '||to_char(log_timepoint,'YYYYMMDD HH24:MI:SS');
RAISE NOTICE '%',activity_string;
PERFORM pg_log( log_database_id , 'log_query' , activity_string);
EXCEPTION
WHEN unique_violation THEN
RAISE NOTICE '*** unique_violation *** Abfrage wurde bereits protokolliert';
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
RAISE NOTICE 'log_query mit query_md5hash = % und timepoint = % hat bereits eine QUERYID = %',log_md5hash,log_timepoint , log_queryid ;
RETURN result;
END IF;
------------------------------------------------
RAISE NOTICE 'Update queryid';
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;
--SELECT log_query_rec.query like pg_stat_query INTO found_flag ;
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||
' für log_query mit id = '||log_query_rec.id
;
RAISE NOTICE '%',activity_string;
EXIT ;
END IF ;
END LOOP ;
RETURN result ;
END
$$ LANGUAGE plpgsql;
Bei der Verarbeitung wird eine Servicetabelle verwendet pg_stat_db_queries, die einen Snapshot der aktuellen Abfragen aus der Tabelle enthält pg_stat_history (Die Verwendung der Tabelle ist hier beschrieben — )
TABLE pg_stat_db_queries
(
database_id integer,
queryid bigint ,
query text ,
max_time double precision
);
TABLE pg_stat_history
(
…
database_id integer ,
…
queryid bigint ,
…
max_time double precision ,
…
);
Die Funktion bietet eine Reihe nützlicher Möglichkeiten zur Verarbeitung von Abfragen aus der Protokolldatei. Nämlich:
Möglichkeit Nr. 1 — Ausführungsverlauf von Abfragen
Sehr hilfreich, um mit der Lösung eines Leistungsproblems zu beginnen. Zuerst den Verlauf prüfen — wann hat die Verlangsamung begonnen?
Anschließend, klassischerweise – nach externen Ursachen suchen. Vielleicht ist die Datenbanklast plötzlich stark gestiegen und eine bestimmte Abfrage hat damit nichts zu tun.
Füge einen neuen Datensatz in die Tabelle log_query ein
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 );
SELECT e.host , d.name , d.owner_pwd
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 = to_number(duration,'99999999999999999999D9999999999');
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);
RAISE NOTICE 'log_query=%',log_query ;
log_md5hash = md5( log_query::text );
--Explain execution plan--
EXECUTE 'SELECT dblink_connect(''LINK1'',''host='||database_rec.host||' dbname='||database_rec.name||' user=DATABASE 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
)
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
);
Möglichkeit Nr. 2 — Abfrageausführungspläne speichern
An dieser Stelle könnte eine Einwendung-oder- Klarstellung-oder-Kommentar erscheinen: „Aber es gibt doch schon autoexplain“. Es gibt es, aber was nützt es, wenn der Ausführungsplan im selben Protokoll gespeichert wird und um ihn für die spätere Analyse zu speichern, müsste die Protokolldatei geparsed werden?
Ich benötigte jedoch:
erstens: den Ausführungsplan in einer Servicetabelle der Überwachungsdatenbank zu speichern;
Zweitens: Die Möglichkeit zu haben, Ausführungspläne miteinander zu vergleichen, um sofort zu sehen, dass sich der Ausführungsplan einer Abfrage geändert hat.
Eine Anfrage mit spezifischen Ausführungsparametern liegt vor. Es ist einfach, ihren Ausführungsplan zu erhalten und zu speichern, indem man EXPLAIN verwendet.
Darüber hinaus kann man das EXPLAIN (COSTS FALSE)-Ausdruck verwenden, um ein Gerüst des Plans zu erhalten, das zur Ermittlung des Hash-Werts des Plans verwendet wird, was bei der späteren Analyse der Historie der Planänderungen hilft.
Den Ausführungsplanvorlage erhalten
--Erklärung des Ausführungsplans--
EXECUTE 'SELECT dblink_connect(''LINK1'',''host='||database_rec.host||' dbname='||database_rec.name||' user=DATABASE 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');
Möglichkeit Nr. 3 - Verwendung des Anfrageprotokolls zur Überwachung
Da die Leistungskennzahlen nicht auf den Text der Anfrage, sondern auf deren ID eingestellt sind, muss man die Anfragen aus der Protokolldatei mit Anfragen verknüpfen, für die die Leistungskennzahlen eingestellt sind.
Nun, zumindest um den genauen Zeitpunkt des Leistungsproblems zu haben.
Somit wird bei Auftreten eines Leistungsproblems für die ID der Anfrage ein Verweis auf die spezifische Anfrage mit den spezifischen Parameterwerten und dem genauen Zeitpunkt der Ausführung und der Dauer der Anfrage bestehen. Diese Informationen nur über die Sicht zu erhalten - pg_stat_statements ist nicht möglich.
Den queryid der Anfrage finden und den Datensatz in der Tabelle log_query aktualisieren
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;
--SELECT log_query_rec.query like pg_stat_query INTO found_flag ;
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||
' for log_query with id = '||log_query_rec.id
;
RAISE NOTICE '%',activity_string;
EXIT ;
END IF ;
END LOOP ;
Nachwort
Die beschriebene Methodik hat letztendlich Anwendung gefunden in , was mehr Informationen für die Analyse bei der Lösung aufkommender Performance-Vorfälle ermöglicht.
Obwohl ich persönlich der Meinung bin, dass man noch an dem Algorithmus zur Auswahl und Änderung der Größe der geladenen Portion arbeiten sollte. Das Problem wurde bisher im Allgemeinen noch nicht gelöst. Es wird wahrscheinlich interessant sein.
Aber das ist schon eine ganz andere Geschichte …
Quelle: habr.com
