ProHoster > Blog > Administrimi > Ekspozimi i logut të PostgreSQL nga AWS

Ekspozimi i logut të PostgreSQL nga AWS

Ose edhe pak aplikimi i tetrisologjisë.
E reja është gjithmonë diçka e vjetër e harruar.
Epigrafët.
Ekspozimi i logut të PostgreSQL nga AWS

Vendosja e detyrës

Duhet të ngarkoni periudhshëm skedarin aktual të log-ut të PostgreSQL nga AWS cloud në një host lokal Linux. Jo në kohë reale, por le të themi, me një vonesë të vogël.
Periudha e ngarkimit të skedarit të log-ut është 5 minuta.
Skedari i log-ut në AWS rrotullohet çdo orë.

Instrumentet e përdorura

Për ngarkimin e skedarit të log-ut në host përdoret një skript bash, i cili thirret API e AWS "aws rds download-db-log-file-portion».

Parametrat:

  • —db-instance-identifier: Emri i instancĂ«s nĂ« AWS;
  • —log-file-name: emri i skedarit aktual tĂ« formuar tĂ« log-ut
  • —max-item: Numri i pĂ«rgjithshĂ«m i elementeve qĂ« kthehen nĂ« daljen e komandĂ«s.MadhĂ«sia e skedarit tĂ« ngarkuar.
  • —starting-token: Etiketa e skedarit fillestar

Në këtë rast të veçantë, detyra e ngarkimit të logëve u paraqit gjatë punës mbi monitorimin e performancës së pyetjeve PostgreSQL.

Në fakt, kjo është një detyrë interesante, për stërvitje dhe për të sjellë diversitet në kohën e punës.
Supozoi qĂ« detyra, pĂ«r shkak tĂ« zakonshmĂ«risĂ«, ishte tashmĂ« e zgjidhur. Por njĂ« kĂ«rkim i shpejtĂ« nĂ« Google nuk ofroi zgjidhje, dhe nuk kishte dĂ«shirĂ« tĂ« kĂ«rkoja mĂ« thellĂ«. Çdo rast — njĂ« stĂ«rvitje e mirĂ«.

Formalizimi i detyrës

Skedari përfundimtar i log-ut përbëhet nga shumë rreshta me gjatësi të ndryshme. Grafikisht, skedari i log-ut mund të përfaqësohet, përafërsisht kështu:
Ekspozimi i logut të PostgreSQL nga AWS

A po i ngjan diçkaje? ÇfarĂ« ka lidhje me 'tetris'? Ja, kjo.
Nëse e përfaqësoni vizualisht mundësitë që lindin gjatë ngarkimit të skedarit të ri (për thjeshtësi, në këtë rast, le të themi se rreshtat kanë të njëjtën gjatësi), do të merrni figurat standarde të tetris:

1) Skedari është ngarkuar në tërësi dhe është përfundimtar. Madhësia e skedarit është më e madhe se madhësia e skedarit përfundimtar:
Ekspozimi i logut të PostgreSQL nga AWS

2) Skedari ka vazhdim. Madhësia e skedarit është më e vogël se madhësia e skedarit përfundimtar:
Ekspozimi i logut të PostgreSQL nga AWS

3) Skedari është vazhdimi i skedarit të mëparshëm dhe ka vazhdim. Madhësia e skedarit është më e vogël se madhësia e mbetjeve të skedarit përfundimtar:
Ekspozimi i logut të PostgreSQL nga AWS

4) Skedari është vazhdimi i skedarit të mëparshëm dhe është përfundimtar. Madhësia e skedarit është më e madhe se madhësia e mbetjeve të skedarit përfundimtar:
Ekspozimi i logut të PostgreSQL nga AWS

Detyra është të ndërton një drejtkëndësh ose të luajë në tetris, në një nivel të ri.
Ekspozimi i logut të PostgreSQL nga AWS

Problemet që shfaqen gjatë zgjidhjes së detyrës

1) Bashko rreshtin nga 2 skedarë

Ekspozimi i logut të PostgreSQL nga AWS
Në përgjithësi nuk ka pasur ndonjë problem të veçantë. Një detyrë standarde nga kursi fillestar i programimit.

Madhësia optimale e skedarit

Kjo është pak më interesante.
Fatkeqësisht, nuk ka mundësi të përdorësh një ofset pas etiketes fillestare:

Ashtu siç e dini, opsioni —starting-token pĂ«rdoret pĂ«r tĂ« pĂ«rcaktuar nga ku tĂ« filloni paginimin. Ky opsion merr vlera String qĂ« do tĂ« thotĂ« se nĂ«se pĂ«rpiqeni tĂ« shtoni njĂ« vlerĂ« ofseti pĂ«rpara vargut tĂ« Tokenit TĂ« Ndjekur, opsioni nuk do tĂ« merret parasysh si njĂ« ofset.

Prandaj, duhet të lexoni në copa-porcione.
Nëse lexoni në mënyra të mëdha, atëherë numri i leximeve do të jetë minimal, por sasia do të jetë maksimale.
Nëse lexoni në mënyra të vogla, atëherë përkundrazi, numri i leximeve do të jetë maksimal, por sasia do të jetë minimale.
Prandaj, për të shkurtuar trafikun dhe për të pasuruar zgjidhjen, duhet të shpikni një zgjidhje që, fatkeqësisht, duket pak si një ndihmë.

Për ilustruar, le të shqyrtojmë procesin e ngarkimit të skedarit të log-ut në 2 varianta shumë të thjeshtuar. Numri i leximeve në të dyja rastet varet nga madhësia e skedarit.

1) Ngarko në copa të vogla:
Ekspozimi i logut të PostgreSQL nga AWS

2) Ngarko në copa të mëdha:
Ekspozimi i logut të PostgreSQL nga AWS

Si zakonisht, zgjidhja optimale është ndërmjet..
Madhësia e skedarit është minimale, por gjatë leximit, madhësia mund të rritet, për të shkurtuar numrin e leximeve.

Duhet të theksohet, se detyra e përshtatjes së madhësisë optimale të skedarit të lexuar ende nuk është e zgjidhur plotësisht dhe kërkon një përpunim dhe analizë më të thellë. Ndoshta, më vonë.

Përshkrimi i përgjithshëm i realizimit

Të dhënat e shërbimeve të përdorura

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 — markimi i fundit pĂ«r skedarin e ngarkuar tĂ« log-ut nĂ« formatin YYYY-MM-DD-HH24.
last_aws_nexttoken — etiketĂ« tekstuale pĂ«r skedarin pĂ«rfundimtar tĂ« ngarkuar.
aws_max_item_size- madhësia fillestare e skedarit e përcaktuar në mënyrë empirike.

Teksti i plotë i skriptit

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

Fragmentet e skriptit me disa shpjegime:

Parametrat hyrës të skriptit:

  • Markimi i kohĂ«s pĂ«r emrin e skedarit tĂ« log-ut nĂ« formatin YYYY-MM-DD-HH24: AWS_LOG_TIME=$1
  • ID e BazĂ«s sĂ« DhĂ«nave: database_id=$2
  • Emri i skedarit tĂ« pĂ«rmbledhur tĂ« log-ut: RESULT_FILE=$3

Merr markimin e kohës për skedarin e fundit të ngarkuar të log-ut:

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 "`

NĂ«se markimi i kohĂ«s pĂ«r skedarin e fundit tĂ« ngarkuar tĂ« log-ut nuk pĂ«rputhet me parametrin hyrĂ«s — ngarkohet njĂ« skedar i ri tĂ« log-ut:

nëse [[ $current_aws_log_time != $AWS_LOG_TIME ]];
  atëherë
    is_new_log='1'
	nëse ! 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 "
	atëherë
	  echo '***download_aws_piece.sh -FATAL_ERROR - update database set last_aws_log_time .'
	  exit 1
	fi
  tjetër
    is_new_log='0'
  fi

Marrim vlerën e etiketës nexttoken nga skedari i ngarkuar:

  next_token_str=`cat $LOG_FILE | grep NEXTTOKEN` 
  next_token=`echo $next_token_str | awk -F" " '{ print $2}' `

Një tregues për fundin e ngarkimit është një vlerë e zbrazët nexttoken.

Në cikël numërojmë porcionet e skedarit, duke bashkuar linjat dhe rritur madhësinë e porcionit:
Cikli kryesor

# 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

ÇfarĂ« do tĂ« ndodhĂ« mĂ« pas?

Pra, detyra e parĂ« ndĂ«rmjetĂ«sore — «tĂ« ngarkohet skedari i log-ut nga re» Ă«shtĂ« zgjidhur. ÇfarĂ« duhet bĂ«rĂ« me log-un e ngarkuar?
Së pari, është e nevojshme të analizojmë skedarin e log-ut dhe të nxjerrim kërkesat.
Detyra nuk është shumë e komplikuar. Një skript bash shumë i thjeshtë e bën punën.
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

Tani me kërkesën e nxjerrë nga skedari i log-ut, mund të punojmë.

Dhe hapen disa mundësi të dobishme.

Kërkesat e analizuar duhet të ruhen diku. Për këtë përdoret një tabelë shërbimi 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 ;

Përpunimi i kërkesës së analizuar bëhet në plpgsql funksionin «log_query».
log_query.sql

--log_query.sql
--verison 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 );
  
  --Shpjegoni planin e ekzekutimit--
  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 = 	'Kërkesa e re është regjistruar '||
						' 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 *** kërkesa tashmë është regjistruar';
	END;

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

	NËSE log_queryid NUK ËSHTË NULL 
	ATËHERË 
	  RAISE NOTICE 'log_query me query_md5hash = % dhe timepoint = % tashmë ka një QUERYID = %',log_md5hash,log_timepoint , log_queryid ;
	  KTHYJ rezultatin;
	KTHEH PËRRALLAT;
	
	------------------------------------------------
	RAISE NOTICE 'Përditëso kërkesën id';	
	
	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');
	
	PËR pg_stat_history_rec NË
	 SELECT 
         queryid ,
	  query 
	 FROM 
         pg_stat_db_queries 
     KUJDE  
      database_id = log_database_id AND
       queryid nuk është 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 si pg_stat_query INTO found_flag ; 
	  NËSE (log_query_text LIKE pg_stat_query_text) ATËHERË
		found_flag = TRUE ;
	  TËTJERË
		found_flag = FALSE ;
	  FUND IF;	  
	  
	  
	  NËSE found_flag ATËHERË
	    
		UPDATE log_query SET queryid = pg_stat_history_rec.queryid WHERE query_md5hash = log_md5hash AND timepoint = log_timepoint ;
		activity_string = 	' përditësuar queryid = '||pg_stat_history_rec.queryid||
		                    ' për log_query me id = '||log_query_rec.id               
		   				    ;						
	    RAISE NOTICE '%',activity_string;	
		EXIT ;
	  FUND IF ;
	  
	END LOOP ;
	
  KTHYJ rezultatin ;
END
$$ LANGUAGE plpgsql;

gjatĂ« pĂ«rpunimit pĂ«rdoret tabelĂ« shĂ«rbimi pg_stat_db_queries, qĂ« pĂ«rmban njĂ« pamje tĂ« kĂ«rkesave aktuale nga tabela pg_stat_history (PĂ«rdorimi i tabelĂ«s pĂ«rshkruhet kĂ«tu — Monitorimi i performancĂ«s sĂ« kĂ«rkesave PostgreSQL. Pjesa 1 — raportimi.)

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	 , 	


);

Kjo funksion ofron një sërë mundësish të dobishme për përpunimin e kërkesave nga skedari i log-ut. Për më shumë:

MundĂ«sia Nr. 1 — Historia e ekzekutimit tĂ« kĂ«rkesave

ËshtĂ« shumĂ« e dobishme pĂ«r tĂ« filluar zgjidhjen e njĂ« incidenti me performancĂ«n. SĂ« pari, shikoni historinĂ« — kur filloi ngadalĂ«simi?
Pastaj, sipas klasikes – kontrolloni shkaqet e jashtme. Ndoshta thjesht ndodhi njĂ« rritje e papritur e ngarkesĂ«s sĂ« bazĂ«s sĂ« tĂ« dhĂ«nave dhe njĂ« kĂ«rkesĂ« e veçantĂ« nuk ka asnjĂ« lidhje.
Shto një regjistrim të ri në tabelën 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 = 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 );
  
  --Shpjego planin e ekzekutimit--
  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		
	);

MundĂ«sia Nr. 2 — Ruani planet e ekzekutimit tĂ« kĂ«rkesave

Në këtë pikë mund të ngrihet një kundërshtim-sqarim-komentar: «Por, apo nuk ekziston autoexplain». Po, ai është aty, por çfarë vlefte ka nëse plani i ekzekutimit ruhet në të njëjtin skedar log dhe për ta ruajtur për analizë të mëvonshme, duhet të analizosh skedarin log?

Mua, më duheshin:
së pari: ruani planin e ekzekutimit në një tabelë shërbimi të bazës së të dhënave të monitorimit;
së dyti: të kem mundësi për të krahasuar planet e ekzekutimit me njëra-tjetrën, që të shikoj menjëherë nëse plani i ekzekutimit të një kërkese u ndryshua.

KĂ«rkesa me parametra tĂ« caktuara Ă«shtĂ« e pranishme. Marrja dhe ruajtja e planit tĂ« saj tĂ« ekzekutimit, duke pĂ«rdorur EXPLAIN — Ă«shtĂ« njĂ« detyrĂ« elementare.
Më shumë se kaq, duke përdorur shprehjen EXPLAIN (COSTS FALSE), mund të merrni skeletin e planit, i cili do të përdoret për të marrë vlerën hash të planit, e cila do të ndihmojë në analizën e mëvonshme të historisë së ndryshimit të planit të ekzekutimit.
Merrni një model të planit të ekzekutimit

  --Shpjego planin e ekzekutimit--
  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');

MundĂ«sia Nr. 3 — PĂ«rdorimi i log-ut tĂ« kĂ«rkesave pĂ«r monitorim

Duke qenë se metrikat e performancës nuk janë të përshtatura për tekstin e kërkesës, por për ID-në e saj, është e nevojshme të lidhen kërkesat nga skedari log me kërkesat për të cilat janë vendosur metrikat e performancës.
Të paktën për të pasur kohën e saktë të shpërthimit të incidentit të performancës.

NĂ« kĂ«tĂ« mĂ«nyrĂ«, nĂ« rastin e ndodhisĂ« sĂ« incidentit tĂ« performancĂ«s pĂ«r ID-nĂ« e kĂ«rkesĂ«s, do tĂ« jetĂ« njĂ« lidhje me kĂ«rkesĂ«n specifike me vlera specifike tĂ« parametrave dhe kohĂ«n e saktĂ« tĂ« ekzekutimit dhe gjatĂ«si tĂ« kĂ«rkesĂ«s. TĂ« merrni kĂ«tĂ« informacion duke pĂ«rdorur vetĂ«m pamjen pg_stat_statements — nuk Ă«shtĂ« e mundur.
Gjeni queryid e kërkesës dhe përditësoni regjistrimin në tabelën log_query

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 = 	' përditësuar queryid = '||pg_stat_history_rec.queryid||
		                    ' për log_query me id = '||log_query_rec.id		                    
		   				    ;						
					
	    RAISE NOTICE '%',activity_string;	
		EXIT ;
	  END IF ;
	  
	END LOOP ;

Pasthënie

Metodologjia e përshkruar në fund, gjeti aplikimin e saj në sistemin e zhvilluar për monitorimin e performancës së kërkesave PostgreSQL, duke mundësuar të ketë më shumë informacion për analizë kur zgjidhen incidentet që lidhen me performancën e kërkesave.

Megjithatë, sipas mendimit tim personal si autor, do të nevojitej më shumë punë mbi algoritmin për zgjedhjen dhe ndryshimin e madhësisë së grumbullit të ngarkuar. Problemi ende nuk është zgjidhur në rastin e përgjithshëm. Mund të jetë interesante.

Por kjo është një histori krejtësisht e ndryshme 


Burimi: habr.com

Bleni hostim tĂ« besueshĂ«m pĂ«r faqe me mbrojtje nga DDoS, serverĂ« VPS VDS đŸ”„ Bleni hostim tĂ« besueshĂ«m pĂ«r faqe me mbrojtje nga DDoS, serverĂ« VPS VDS | ProHoster