Merge pull request #7 from ssl-oyamata/workload_sampler#1

Workload sampler#1

Author: ssl-oyamata

conf/postgres_opttune.conf

@@ -16,15 +16,28 @@ pg_os_user = postgres # PostgrSQL ower user(OS user)
 # When tuning remote PostgreSQL, use an SSH connection to restart PostgreSQL or clear the cache.
 # Note: When using remote PostgreSQL, it is necessary to grant sudo permission without password to the remote os user.
 ssh_port = 22 # ssh port
-ssh_password = postgres # pg_os_user's ssh passwoed
+ssh_password = postgres # pg_os_user's ssh password
+
+[workload-sampling]
+workload_sampling_time_second = 300
+# Time (in seconds) to sample the workload running on the database in the [PostgreSQL] section
+my_workload_save_dir = ./workload_data/
+# Workload save database settings
+pghost = localhost # PostgreSQL server host
+pgport = 5432 # PostgreSQL server port
+pguser = postgres # PostgreSQL user name(Database user)
+pgpassword = postgres12 # PostgreSQL user password(Database user)
+pgdatabase = sampling # PostgreSQL Database
+# workload save directory
+
 
 [turning]
 study_name = pgbench_study # study name
 required_recovery_time_second = 0
 # The maximum recovery time allowed by the user in case of a PostgreSQL crash,
 # which is used to estimate the wax_wal_size parameter.
 # Note: The default value of 0 does not perform the estimation of the wax_wal_size parameter.
-benchmark = pgbench # Benchmark tool name('pgbench' or 'oltpbench' or 'star_schema_benchmark')
+benchmark = pgbench # Benchmark tool name('my_workload' or pgbench' or 'oltpbench' or 'star_schema_benchmark')
 parameter_json_dir = ./conf/
 number_trail = 100 # Number of benchmarks to run for turning
 data_load_interval = 10 # Specify the data load interval by the number of benchmarks
@@ -34,6 +47,10 @@ save_study_history = True # Whether to save study history
 load_study_history = True # Whether to load study history if a study name already exists.
 history_database_url = sqlite:///study-history.db   # Example PostgreSQL. postgresql://postgres@localhost/study_history
 
+[my-workload]
+my_workload_save_file = workload_data/2020-07-05_180647.531417-2020-07-05_180657.531661.pkl
+# File saved using workload_sampler.py
+
 [pgbench]
 scale_factor = 10 # pgbench scale factor
 clients = 10 # Number of clients

pgopttune/config/my_workload_config.py

@@ -0,0 +1,20 @@
+import os
+from pgopttune.config.config import Config
+
+
+class MyWorkloadConfig(Config):
+    def __init__(self, conf_path, section='my-workload'):
+        super().__init__(conf_path)
+        self.conf_path = conf_path
+        self.config_dict = dict(self.config.items(section))
+        self._check_is_exist_my_workload_save_file()
+
+    def _check_is_exist_my_workload_save_file(self):
+        if not os.path.exists(self.get_parameter_value('my_workload_save_file')):
+            raise ValueError("{} does not exist."
+                             "Check the my_workload_save_file parameter in {}."
+                             .format(self.get_parameter_value('my_workload_save_file'), self.conf_path))
+
+    @property
+    def my_workload_save_file(self):
+        return self.get_parameter_value('my_workload_save_file')

pgopttune/config/workload_sampling_config.py

@@ -0,0 +1,24 @@
+from pgopttune.config.config import Config
+from pgopttune.utils.pg_connect import get_pg_dsn
+
+
+class WorkloadSamplingConfig(Config):
+    def __init__(self, conf_path, section='workload-sampling'):
+        super().__init__(conf_path)
+        self.config_dict = dict(self.config.items(section))
+
+    @property
+    def workload_sampling_time_second(self):
+        return float(self.get_parameter_value('workload_sampling_time_second'))
+
+    @property
+    def my_workload_save_dir(self):
+        return self.get_parameter_value('my_workload_save_dir')
+
+    @property
+    def dsn(self):
+        return get_pg_dsn(pghost=self.get_parameter_value('pghost'),
+                          pgport=self.get_parameter_value('pgport'),
+                          pgdatabase=self.get_parameter_value('pgdatabase'),
+                          pguser=self.get_parameter_value('pguser'),
+                          pgpassword=self.get_parameter_value('pgpassword'))

pgopttune/log/pg_csv_log.py

@@ -11,12 +11,12 @@
 
 
 class PostgresCsvLog:
-    def __init__(self, postgres_server_config: PostgresServerConfig, csv_log_table_name="csv_log"):
+    def __init__(self, postgres_server_config: PostgresServerConfig):
         self._postgres_server_config = postgres_server_config
         self._postgres_parameter = PostgresParameter(postgres_server_config)
-        self.csv_log_table_name = csv_log_table_name
-        self.start_csv_log = None  # The time when the csv log started to be output(epoch time)
-        self.end_csv_log = None  # The time when the csv log output was finished(epoch time)
+        self._csv_log_table_name = "csv_log"
+        self.start_csv_log_unix_time = None  # The time when the csv log started to be output(epoch time)
+        self.end_csv_log_unix_time = None  # The time when the csv log output was finished(epoch time)
         self._csv_log_file_path = None
         self.csv_log_local_file_path = None  # File path when the csv log file is copied to localhost
 
@@ -26,7 +26,10 @@ def __init__(self, postgres_server_config: PostgresServerConfig, csv_log_table_n
                                     The current setting is off; to turn it on, you need to restart PostgreSQL.")
         # get current settings
         self._current_log_destination = self._get_log_destination()
-        self._log_min_duration_statement = self._get_log_min_duration_statement()
+        self._current_log_min_duration_statement = self._get_log_min_duration_statement()
+
+    def __del__(self):
+        self.disable()
 
     def enable(self):
         # set log_destination = '[current_setting],csvlog'
@@ -36,33 +39,41 @@ def enable(self):
         else:
             log_destination_value = self._current_log_destination + ',csvlog'
             self._postgres_parameter.set_parameter(param_name="log_destination", param_value=log_destination_value)
+            logger.debug("Start outputting PostgreSQL log message to CSV file.\n"
+                         "log_destination = '{}'".format(log_destination_value))
         # set log_min_duration_statement = 0
         self._postgres_parameter.set_parameter(param_name="log_min_duration_statement", param_value=0, pg_reload=True)
-        self.start_csv_log = time.time()
+        logger.debug("Changed setting to output all executed SQL to PostgreSQL log file.\n"
+                     "log_min_duration_statement = '0'")
+        self.start_csv_log_unix_time = time.time()
         self._csv_log_file_path = self._get_csv_log_file_path()
 
     def disable(self):
         # set log_destination = '[current_setting]'
         self._postgres_parameter.set_parameter(param_name="log_destination", param_value=self._current_log_destination)
+        logger.debug("PostgreSQL log message output to CSV file is disabled.\n"
+                     "log_destination = '{}'".format(self._current_log_min_duration_statement))
         # set log_min_duration_statement = current_setting
         self._postgres_parameter.set_parameter(param_name="log_min_duration_statement",
-                                               param_value=self._log_min_duration_statement, pg_reload=True)
-        self.end_csv_log = time.time()
-
-    def load_csv_to_local_database(self):
-        self._copy_csv_logfile_to_local()  # copy logfile to /tmp directory(localhost)
-        self._create_csv_log_table()
-        self._truncate_csv_log_table()  # truncate csv log table
-        with get_pg_connection(dsn=self._postgres_server_config.dsn) as conn:
+                                               param_value=self._current_log_min_duration_statement, pg_reload=True)
+        logger.debug("The value of the log_min_duration_statement parameter has been restored to its original value.\n"
+                     "log_min_duration_statement = '{}'".format(self._current_log_min_duration_statement))
+        self.end_csv_log_unix_time = time.time()
+
+    def load_csv_to_database(self, copy_dir="/tmp", dsn=None):
+        self._copy_csv_logfile_to_local(copy_dir)  # copy logfile to directory(localhost)
+        self._create_csv_log_table(dsn)
+        self._truncate_csv_log_table(dsn)  # truncate csv log table
+        with get_pg_connection(dsn=dsn) as conn:
             conn.set_session(autocommit=True)
             with conn.cursor() as cur:
                 with open(self.csv_log_local_file_path) as f:
                     # cur.copy_from(f, self.csv_log_table_name, sep=',')
-                    cur.copy_expert("copy {} from stdin (format csv)".format(self.csv_log_table_name), f)
+                    cur.copy_expert("copy {} from stdin (format csv)".format(self._csv_log_table_name), f)
 
-    def _copy_csv_logfile_to_local(self):
+    def _copy_csv_logfile_to_local(self, copy_dir="/tmp"):
         file_name = os.path.basename(self._csv_log_file_path)
-        self.csv_log_local_file_path = os.path.join("/tmp", file_name)
+        self.csv_log_local_file_path = os.path.join(copy_dir, file_name)
         ssh = SSHCommandExecutor(user=self._postgres_server_config.os_user,
                                  password=self._postgres_server_config.ssh_password,
                                  hostname=self._postgres_server_config.host,
@@ -94,7 +105,7 @@ def _get_csv_log_file_path(self):
             csv_file_path = os.path.join(self._postgres_server_config.pgdata, csv_file_path)
         return csv_file_path
 
-    def _create_csv_log_table(self):
+    def _create_csv_log_table(self, dsn):
         create_table_sql = "CREATE TABLE IF NOT EXISTS {} (" \
                            "log_time timestamp(3) with time zone," \
                            "user_name text," \
@@ -119,15 +130,15 @@ def _create_csv_log_table(self):
                            "query_pos integer," \
                            "location text," \
                            "application_name text," \
-                           "PRIMARY KEY (session_id, session_line_num));".format(self.csv_log_table_name)
-        with get_pg_connection(dsn=self._postgres_server_config.dsn) as conn:  # FIXME
+                           "PRIMARY KEY (session_id, session_line_num));".format(self._csv_log_table_name)
+        with get_pg_connection(dsn=dsn) as conn:
             conn.set_session(autocommit=True)
             with conn.cursor() as cur:
                 cur.execute(create_table_sql)
 
-    def _truncate_csv_log_table(self):
-        truncate_table_sql = "TRUNCATE {}".format(self.csv_log_table_name)
-        with get_pg_connection(dsn=self._postgres_server_config.dsn) as conn:  # FIXME
+    def _truncate_csv_log_table(self, dsn):
+        truncate_table_sql = "TRUNCATE {}".format(self._csv_log_table_name)
+        with get_pg_connection(dsn=dsn) as conn:
             conn.set_session(autocommit=True)
             with conn.cursor() as cur:
                 cur.execute(truncate_table_sql)
@@ -136,12 +147,15 @@ def _truncate_csv_log_table(self):
 if __name__ == "__main__":
     from pgopttune.config.postgres_server_config import PostgresServerConfig
 
+    logging.basicConfig(level=logging.DEBUG)
     conf_path = './conf/postgres_opttune.conf'
     postgres_server_config_test = PostgresServerConfig(conf_path)  # PostgreSQL Server config
     csv_log = PostgresCsvLog(postgres_server_config_test)
     csv_log.enable()
-    print(csv_log._csv_log_file_path)
-    print("sleep...")
-    time.sleep(300)
+    # logging.debug(csv_log._csv_log_file_path)
+    logging.debug("sleep...")
+    time.sleep(60)
     csv_log.disable()
-    csv_log.load_csv_to_local_database()
+    csv_log.load_csv_to_database()
+    logging.debug(csv_log.start_csv_log_unix_time)
+    logging.debug(csv_log.end_csv_log_unix_time)

pgopttune/objective/objective.py

@@ -37,8 +37,8 @@ def run_workload(self, trial):
         if (int(trial.number) == 0) or (int(trial.number) % self.data_load_interval == 0):
             self.workload.data_load()  # data load
         self.params.reset_database()  # cache free and database restart
-        tps = self.workload.run()  # benchmark run
-        return tps
+        objective_value = self.workload.run()  # benchmark run
+        return objective_value
 
     def reset_param(self):
         # reset parameter value(reset postgresql.auto.conf)

pgopttune/objective/objective_my_workload.py

@@ -0,0 +1,19 @@
+import logging
+from pgopttune.workload.my_workload import MyWorkload
+from pgopttune.objective.objective import Objective
+from pgopttune.config.postgres_server_config import PostgresServerConfig
+from pgopttune.config.tune_config import TuneConfig
+from pgopttune.config.my_workload_config import MyWorkloadConfig
+
+logger = logging.getLogger(__name__)
+
+
+class ObjectiveMyWorkload(Objective):
+
+    def __init__(self,
+                 postgres_server_config: PostgresServerConfig,
+                 tune_config: TuneConfig,
+                 my_workload_config: MyWorkloadConfig):
+        super().__init__(postgres_server_config, tune_config)
+        self.workload = MyWorkload.load_my_workload(my_workload_config.my_workload_save_file,
+                                                    postgres_server_config=postgres_server_config)

pgopttune/recovery/pg_recovery.py

@@ -42,7 +42,7 @@ def estimate_max_wal_size(self):
         # Two to three checkpoints are performed within the WAL size specified in the max_wal_size parameter.
         # Therefore, we divide by 2 assuming the worst case.
         estimate_max_wal_size_mb = str(math.floor(estimate_max_wal_size / (1024 * 1024))) + 'MB'
-        logging.info("The maximum value of max_wal_size estimated based on the measured values is {}.".format(
+        logger.info("The maximum value of max_wal_size estimated based on the measured values is {}.".format(
             estimate_max_wal_size_mb))
         return estimate_max_wal_size_mb
 
@@ -63,7 +63,7 @@ def measurement_recovery_time(self, measurement_rows_scale=100000, measurement_p
             self._crash_database()
             self._free_cache()
             recovery_time = self._measurement_recovery_database_time()
-            # logging.info('The wal size written after the checkpoint is {}B. '
+            # logger.info('The wal size written after the checkpoint is {}B. '
             #              'And crash recovery time is {}s'.format(recovery_wal_size, recovery_time))
             self.x_recovery_time = np.append(self.x_recovery_time, recovery_time)
             self.y_recovery_wal_size = np.append(self.y_recovery_wal_size, recovery_wal_size)
@@ -72,8 +72,8 @@ def measurement_recovery_time(self, measurement_rows_scale=100000, measurement_p
         self.reset_param()
         progress_bar.close()
         np.set_printoptions(precision=3)
-        logging.info("The wal size written after the checkpoint(Byte) : {}".format(self.y_recovery_wal_size))
-        logging.info("PostgreSQL Recovery time(Sec) : {}".format(self.x_recovery_time))
+        logger.info("The wal size written after the checkpoint(Byte) : {}".format(self.y_recovery_wal_size))
+        logger.info("PostgreSQL Recovery time(Sec) : {}".format(self.x_recovery_time))
 
     def _create_test_table(self):
         create_table_sql = "CREATE TABLE IF NOT EXISTS " + self._test_table_name + "(id INT, test TEXT)"
@@ -136,8 +136,8 @@ def _get_latest_checkpoint_lsn(self):
                 cut_res = subprocess.Popen(cut_cmd, stdout=subprocess.PIPE, stdin=grep_res.stdout)
                 latest_checkpoint_lsn = cut_res.communicate()[0].decode('utf-8')
             except ValueError:
-                logging.critical(traceback.format_exc())
-                logging.info('Failed Command: {} '.format(get_latest_checkpoint_cmd_str))
+                logger.critical(traceback.format_exc())
+                logger.info('Failed Command: {} '.format(get_latest_checkpoint_cmd_str))
                 sys.exit(1)
         else:
             recovery_database_cmd = 'sudo -i -u {} {}/pg_controldata -D {} ' \
@@ -212,4 +212,4 @@ def random_string(length):
     conf_path = '../../conf/postgres_opttune.conf'
     postgres_server_config_test = PostgresServerConfig(conf_path)  # PostgreSQL Server config
     recovery = Recovery(postgres_server_config_test, required_recovery_time_second=300)
-    print(recovery.estimate_max_wal_size())
+    # print(recovery.estimate_max_wal_size())

pgopttune/study/study.py

@@ -4,7 +4,7 @@
 logger = logging.getLogger(__name__)
 
 
-def create_study(study_name, sampler, save_study_history=False, load_study_history=False,
+def create_study(study_name, sampler, save_study_history=False, load_study_history=False, direction='minimize',
                  history_database_url='postgresql://postgres@localhost:5432/study_history'):
     """
     create study.
@@ -15,7 +15,7 @@ def create_study(study_name, sampler, save_study_history=False, load_study_histo
         db_storage = optuna.storages.RDBStorage(history_database_url)
 
     try:
-        study = optuna.create_study(study_name=study_name, sampler=sampler, direction='maximize',
+        study = optuna.create_study(study_name=study_name, sampler=sampler, direction=direction,
                                     storage=db_storage)
     except optuna.exceptions.OptunaError:
         # If a study with the same name already exists, Load past history.

pgopttune/utils/remote_command.py

@@ -1,6 +1,7 @@
 import paramiko
 import scp
 
+
 class SSHCommandExecutor:
     def __init__(self, user, password='postgres', hostname='localhost', port=22, timeout=15.0):
         self.hostname = hostname

pgopttune/workload/my_transaction.py

@@ -0,0 +1,43 @@
+import time
+import logging
+from pgopttune.utils.pg_connect import get_pg_connection
+from pgopttune.config.postgres_server_config import PostgresServerConfig
+
+logger = logging.getLogger(__name__)
+
+
+class MyTransaction:
+    def __init__(self, session_id: str, query_start_time: list, statement: list):
+        self.session_id = session_id
+        self.query_start_time = query_start_time
+        self.statement = statement
+
+    def run(self, postgres_server_config: PostgresServerConfig):
+        start_time = time.time()
+        elapsed_times = 0
+
+        # sleep until first statement start
+        self._sleep_until_statement_start_time(start_time, self.query_start_time[0])
+
+        with get_pg_connection(dsn=postgres_server_config.dsn) as conn:
+            conn.autocommit = False
+            with conn.cursor() as cur:
+                for index in range(len(self.query_start_time)):
+                    self._sleep_until_statement_start_time(start_time, self.query_start_time[index])
+                    query_start_time = time.time()
+                    if "vacuum" in self.statement[index].lower():
+                        cur.execute("END;")
+                        cur.execute(self.statement[index])
+                    # logger.info("Execute Statement : {}".format(self.statement[index]))
+                    elapsed_times += (time.time() - query_start_time)
+        return elapsed_times
+
+    @staticmethod
+    def _sleep_until_statement_start_time(start_time, query_start_time):
+        while (time.time() - start_time) < query_start_time.total_seconds():
+            sleep_time = query_start_time.total_seconds() - (time.time() - start_time)
+            # logger.debug("sleep: {0:.4f}".format(sleep_time))
+            if sleep_time > 1:
+                time.sleep(sleep_time)
+            else:
+                time.sleep(0.1) # FIXME