ApsaraDB RDS:Use Simple Log Service to monitor the sessions of an ApsaraDB RDS for PostgreSQL instance

Background information

In most cases, a database system is a relatively large application. If the load of the database system is heavy, a large number of memory, CPU, I/O, and network resources are consumed. ApsaraDB RDS for PostgreSQL uses process models, and each session corresponds to a background process. Session monitoring helps you understand the status of an RDS instance and troubleshoot the performance bottlenecks of the RDS instance.

pg_stat_activity

The pg_stat_activity view provides information about all sessions that are running in a database system. You can query sessions in a database system at regular intervals to retain session information. This process is equivalent to regular snapshot creation for the database system. This helps you understand the status of a database system and troubleshoot performance issues of the database system.

Session-related parameters

The following table describes the parameters in a pg_stat_activity view.

SQL statements for session collection

This section provides a sample SQL statement for session collection. The leader_id parameter is available only for RDS instances that run PostgreSQL13 or later. You can adjust the SQL statement for session collection based on the database engine version of your RDS instance.

SELECT
    (
        CASE
            WHEN leader_pid is NULL THEN pid
            ELSE leader_pid
        END
    ) AS leader_pid,
    (
        CASE
            WHEN state_change <= now() AND state != 'active' THEN extract(
                epoch
                FROM
                    state_change - query_start
            )
            ELSE extract(
                epoch
                FROM
                    now() - query_start
            )
        END
    ) AS query_duration,
    (
        CASE
            WHEN wait_event_type is NULL THEN 'CPU'
            ELSE coalesce(wait_event_type || '.' || wait_event, '')
        END
    ) AS wait_entry,
    query_id,
    (
        CASE
            WHEN state = 'active' THEN 'running'
            ELSE 'finished'
        END
    ) AS execute_state,
    query,
    datname,
    application_name,
    client_hostname,
    query_start
FROM
    pg_stat_activity
WHERE
    usename NOT IN ('aurora', 'replicator')
    AND backend_type IN ('client backend','parallel worker');

Prerequisites

• A RAM user is created and the following permissions are granted to the RAM user. For more information, see Create a RAM user and grant permissions to the RAM user.

  • AliyunRDSFullAccess: the permissions to manage all ApsaraDB RDS resources.

  • AliyunLogFullAccess: the permissions to manage all Simple Log Service resources.

  • AliyunCloudMonitorFullAccess: the permissions to manage CloudMonitor resources.

• A project and a Logstore are created. For more information, see Activate Simple Log Service.

• The username and password of the required database on your RDS instance are obtained. For more information, see Accounts and permissions.

Procedure

1. Collect session information in the pg_stat_activity system view and send the collected information to the Logstore of Simple Log Service at regular intervals.

   This section provides an example on how to use Simple Log Service SDK for Java to collect and send data at regular intervals.

   1. Install Simple Log Service SDK for Java. For more information, see Install Simple Log Service SDK for Java.

   2. Configure the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables. For more information, see Configure environment variables.

   3. Import the required dependencies into the pom.xml file. In this example, a Maven project is used, and Maven dependencies are required.

      pom.xml

              <dependency>
                  <groupId>com.aliyun.openservices</groupId>
                  <artifactId>aliyun-log</artifactId>
                  <version>0.6.75</version>
              </dependency>
              <dependency>
                  <groupId>org.postgresql</groupId>
                  <artifactId>postgresql</artifactId>
                  <version>42.2.18</version>
              </dependency>
              <dependency>
                  <groupId>com.aliyun</groupId>
                  <artifactId>tea-openapi</artifactId>
                  <version>0.3.2</version>
              </dependency>
              <dependency>
                  <groupId>com.aliyun</groupId>
                  <artifactId>tea-console</artifactId>
                  <version>0.0.1</version>
              </dependency>
              <dependency>
                  <groupId>com.aliyun</groupId>
                  <artifactId>tea-util</artifactId>
                  <version>0.2.21</version>
              </dependency>
              <!-- Add the Lombok dependency. -->
              <dependency>
                  <groupId>org.projectlombok</groupId>
                  <artifactId>lombok</artifactId>
                  <version>1.18.4</version> <!-- Make sure that the most recent version is used. -->
                  <scope>provided</scope>
              </dependency>
              <dependency>
                  <groupId>org.projectlombok</groupId>
                  <artifactId>lombok</artifactId>
                  <version>RELEASE</version>
                  <scope>provided</scope>
              </dependency>
              <dependency>
                  <groupId>ch.qos.logback</groupId>
                  <artifactId>logback-classic</artifactId>
                  <version>1.2.3</version>
              </dependency>
              <dependency>
                  <groupId>ch.qos.logback</groupId>
                  <artifactId>logback-core</artifactId>
                  <version>1.2.3</version>
              </dependency>
              <dependency>
                  <groupId>org.slf4j</groupId>
                  <artifactId>slf4j-api</artifactId>
                  <version>1.7.30</version>
              </dependency>
              <dependency>
                  <groupId>com.aliyun</groupId>
                  <artifactId>rds20140815</artifactId>
                  <version>5.0.1</version>
              </dependency>
              <dependency>
                  <groupId>com.aliyun</groupId>
                  <artifactId>alibabacloud-sls20201230</artifactId>
                  <version>4.0.7</version>
              </dependency>

   4. Run the following sample program to collect session information about the RDS instance and send the collected information to the Logstore of Simple Log Service:

      PgMonitor.java

      package org.example;
      
      import com.aliyun.openservices.log.Client;
      import com.aliyun.openservices.log.common.LogItem;
      import com.aliyun.openservices.log.exception.LogException;
      import com.aliyun.openservices.log.request.PutLogsRequest;
      
      import java.sql.*;
      import java.util.ArrayList;
      import java.util.List;
      import java.util.concurrent.Executors;
      import java.util.concurrent.ScheduledExecutorService;
      import java.util.concurrent.TimeUnit;
      
      public class PgMonitor {
      
          // The information that is used to connect to the RDS instance on which the database is connected.
          private static final String PG_URL = "<jdbc:postgresql://your-host:5432/mydb>";
          private static final String PG_USER = "<your-user>";
          private static final String PG_PASSWORD = "<your-passwd>";
      
          // Information about Alibaba Cloud Simple Log Service.
          private static final String LOG_ENDPOINT = "<your-sls-endpoint>";
          private static final String LOG_PROJECT = "<your-project>";
          private static final String LOG_STORE = "<your-logStore>";
          private static final String ACCESS_KEY_ID = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
          private static final String ACCESS_KEY_SECRET = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
      
          // The interval at which the scheduled task is executed. Unit: minutes.
          private static final int INTERVAL_MINUTES = 1;
      
          public static void main(String[] args) {
              ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
              Runnable task = () -> {
                  try {
                      // Collect activity data from the RDS instance on which the database is connected.
                      List<LogItem> logItems = fetchPgStatActivity();
      
                      // Send collected data to Alibaba Cloud Simple Log Service.
                      sendLogsToSLS(logItems);
                  } catch (Exception e) {
                      e.printStackTrace();
                  }
              };
      
              // Execute the task at regular intervals. INTERVAL_MINUTES specifies the time interval.
              scheduler.scheduleAtFixedRate(task, 0, INTERVAL_MINUTES, TimeUnit.MINUTES);
          }
      
          private static List<LogItem> fetchPgStatActivity() throws SQLException {
              List<LogItem> logItems = new ArrayList<>();
      
              Connection conn = DriverManager.getConnection(PG_URL, PG_USER, PG_PASSWORD);
              Statement stmt = conn.createStatement();
              String query = "SELECT ( CASE WHEN leader_pid is NULL THEN pid ELSE leader_pid END ) as leader_pid, ( CASE WHEN state_change <= now() AND state != 'active' THEN extract( epoch from state_change - query_start ) ELSE extract( epoch from now() - query_start ) END ) AS query_duration, ( CASE WHEN wait_event_type is NULL THEN 'CPU' ELSE coalesce(wait_event_type || '.' || wait_event, '') END ) AS wait_entry, query_id, ( CASE WHEN state = 'active' THEN 'running' ELSE 'finished' END ) AS execute_state, query, datname, application_name, client_hostname, query_start FROM pg_stat_activity WHERE usename NOT IN ('aurora', 'replicator') AND backend_type IN ('client backend','parallel worker')";
              ResultSet rs = stmt.executeQuery(query);
      
              while (rs.next()) {
                  LogItem logItem = new LogItem();
                  logItem.PushBack("leader_pid", rs.getString("leader_pid"));
                  logItem.PushBack("query_duration", rs.getString("query_duration"));
                  logItem.PushBack("wait_entry", rs.getString("wait_entry"));
                  logItem.PushBack("query", rs.getString("query"));
                  logItem.PushBack("datname", rs.getString("datname"));
                  logItem.PushBack("application_name", rs.getString("application_name"));
                  logItem.PushBack("client_hostname", rs.getString("client_hostname"));
                  logItem.PushBack("query_start", rs.getString("query_start"));
                  logItems.add(logItem);
              }
      
              rs.close();
              stmt.close();
              conn.close();
      
              return logItems;
          }
      
          private static void sendLogsToSLS(List<LogItem> logItems) throws LogException {
              Client client = new Client(LOG_ENDPOINT, ACCESS_KEY_ID, ACCESS_KEY_SECRET);
              String topic = "pg_stat_activity";
              String source = "postgresql-monitor";
      
              PutLogsRequest request = new PutLogsRequest(LOG_PROJECT, LOG_STORE, topic, source, logItems);
              client.PutLogs(request);
          }
      }

      The following table describes specific parameters. You can modify the parameters based on your business requirements.

      Parameter	Description	Example
      PG_URL	The URL that is used to connect to the RDS instance. The value must contain the name of the database that you want to monitor.	jdbc:postgresql://pgm-bp1c82mky1avip****.pg.rds.aliyuncs.com:5432/testdb01
      PG_USER	The username of the account that is used to log on to the database.	testdbuser
      PG_PASSWORD	The password of the account that is used to log on to the database.	****
      LOG_ENDPOINT	The endpoint of Simple Log Service. For more information, see Endpoints.	cn-hangzhou.log.aliyuncs.com
      LOG_PROJECT	The name of the project of Simple Log Service.	rdspg-test
      LOG_STORE	The name of the Logstore of Simple Log Service.	rdspg-sls

      After the preceding operations are complete, you can query the session logs of the RDS instance in the Logstore of Simple Log Service. For more information, see Guide to log query and analysis.

      Important

      You can query and analyze the session logs only after you create indexes in the Logstore of Simple Log Service. For more information, see Create indexes.

2. Import the session logs in Simple Log Service to CloudMonitor.

   Log on to the CloudMonitor console, create a metric, and then import logs from Simple Log Service to CloudMonitor. After logs are imported from Simple Log Service to CloudMonitor, you can create a dashboard to view the monitoring chart of the metric. For more information, see Manage metrics for logs imported from Simple Log Service.

3. Optional. Use the ApsaraDB RDS API to import slow query logs and error logs to Simple Log Service.

   You can analyze the status and performance of the RDS instance and troubleshoot performance issues based on the monitoring information about instance sessions, slow query logs, and error logs.

   This section describes how to use Simple Log Service SDK for Java together with the DescribeSlowLogRecords and DescribeErrorLogs operations of ApsaraDB RDS to obtain the slow query logs and error logs of an RDS instance and send the obtained logs to Simple Log Service at regular intervals. The following information provides the sample code.

   Note

   You can click Debug in API documentation of an operation to go to the Debugging page in OpenAPI Explorer and view how to install and use the operation.

   PgCollectLogOpt.java

   package org.example;
   
   import com.aliyun.openservices.log.Client;
   import com.aliyun.openservices.log.common.LogItem;
   import com.aliyun.openservices.log.exception.LogException;
   import com.aliyun.openservices.log.request.PutLogsRequest;
   import lombok.extern.slf4j.Slf4j;
   
   import java.time.*;
   import java.time.format.DateTimeFormatter;
   import java.time.temporal.ChronoUnit;
   import java.util.ArrayList;
   import java.util.List;
   import java.util.concurrent.Executors;
   import java.util.concurrent.ScheduledExecutorService;
   import java.util.concurrent.TimeUnit;
   
   @Slf4j
   public class PgCollectLogOpt {
       private static final String LOG_ENDPOINT = "your-endpoint";
       private static final String LOG_PROJECT = "your-project";
       private static final String LOG_STORE = "your-log-store";
       private static final String ACCESS_KEY_ID = "your-access-key-id";
       private static final String ACCESS_KEY_SECRET = "your-access-key-secret";
       private static final int INTERVAL_MINUTES = 10;
       private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm'Z'");
       private static volatile String lastEndTime = getCurrentTime();
   
       public static com.aliyun.rds20140815.Client createClient() throws Exception {
           com.aliyun.teaopenapi.models.Config config = new com.aliyun.teaopenapi.models.Config()
                   .setAccessKeyId(ACCESS_KEY_ID)
                   .setAccessKeySecret(ACCESS_KEY_SECRET);
           config.endpoint = "rds.aliyuncs.com";
           return new com.aliyun.rds20140815.Client(config);
       }
   
       public static void main(String[] args) throws Exception {
           ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
           executor.scheduleAtFixedRate(() -> {
               try {
                   collectAndSendLogs();
               } catch (Exception e) {
                   log.error("Error during log collection and sending: ", e);
               }
           }, 0, INTERVAL_MINUTES, TimeUnit.MINUTES);
       }
   
       private static void collectAndSendLogs() throws Exception {
           com.aliyun.rds20140815.Client client = createClient();
           String instanceId = "pgm-bp1nz4ed24u6679d";
           String startTime = lastEndTime;
           String endTime = getNextEndTime(startTime);
   
           try {
               List<LogItem> slowLogs = getLogs(client, instanceId, startTime, endTime, LogType.SLOW);
               if (!slowLogs.isEmpty()) sendLogsToSLS(slowLogs);
   
               List<LogItem> errorLogs = getLogs(client, instanceId, startTime, endTime, LogType.ERROR);
               if (!errorLogs.isEmpty()) sendLogsToSLS(errorLogs);
   
               lastEndTime = endTime;
           } catch (Exception e) {
               log.error("Log collection error: ", e);
           }
       }
   
       private static List<LogItem> getLogs(com.aliyun.rds20140815.Client client, String instanceId, String startTime, String endTime, LogType logType) throws Exception {
           List<LogItem> logItems = new ArrayList<>();
           int pageNumber = 1, totalPage;
   
           do {
               totalPage = fetchAndProcessLogs(client, instanceId, startTime, endTime, logType, pageNumber, logItems);
               pageNumber++;
           } while (pageNumber <= totalPage);
   
           return logItems;
       }
   
       private static int fetchAndProcessLogs(com.aliyun.rds20140815.Client client, String instanceId, String startTime, String endTime, LogType logType, int pageNumber, List<LogItem> logItems) throws Exception {
           if (logType == LogType.SLOW) {
               var request = new com.aliyun.rds20140815.models.DescribeSlowLogRecordsRequest()
                       .setDBInstanceId(instanceId)
                       .setStartTime(startTime)
                       .setEndTime(endTime)
                       .setPageNumber(pageNumber)
                       .setPageSize(100);
               var response = client.describeSlowLogRecordsWithOptions(request, new com.aliyun.teautil.models.RuntimeOptions());
               var items = response.getBody().getItems().getSQLSlowRecord();
               items.forEach(item -> logItems.add(createLogItem(item.getExecutionStartTime(), item.getSQLText(), item.getHostAddress(), item.getDBName(), item.getQueryTimes().toString(), item.getLockTimes().toString())));
               return (int) Math.ceil((double) response.getBody().getTotalRecordCount() / response.getBody().getPageRecordCount());
           } else {
               var request = new com.aliyun.rds20140815.models.DescribeErrorLogsRequest()
                       .setDBInstanceId(instanceId)
                       .setStartTime(startTime)
                       .setEndTime(endTime)
                       .setPageNumber(pageNumber)
                       .setPageSize(100);
               var response = client.describeErrorLogsWithOptions(request, new com.aliyun.teautil.models.RuntimeOptions());
               var items = response.getBody().getItems().getErrorLog();
               items.forEach(item -> logItems.add(createLogItem(item.getCreateTime(), item.getErrorInfo())));
               return (int) Math.ceil((double) response.getBody().getTotalRecordCount() / response.getBody().getPageRecordCount());
           }
       }
   
       private static LogItem createLogItem(String utcTime, String... fields) {
           LogItem logItem = new LogItem();
           String collectTime = convertToBeijingTime(utcTime);
           logItem.PushBack("collectTime", collectTime);
           String[] fieldNames = {"sql", "hostAddress", "dbName", "queryTimes", "lockTimes", "errorInfo"};
   
           for (int i = 0; i < fields.length; i++) {
               logItem.PushBack(fieldNames[i], fields[i]);
           }
   
           return logItem;
       }
   
       private static String convertToBeijingTime(String utcTime) {
           Instant instant = Instant.parse(utcTime);
           ZonedDateTime utcDateTime = instant.atZone(ZoneId.of("UTC"));
           ZonedDateTime beijingDateTime = utcDateTime.withZoneSameInstant(ZoneId.of("Asia/Shanghai"));
           return DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ssXXX").format(beijingDateTime);
       }
   
       private static void sendLogsToSLS(List<LogItem> logItems) throws LogException {
           Client slsClient = new Client(LOG_ENDPOINT, ACCESS_KEY_ID, ACCESS_KEY_SECRET);
           PutLogsRequest request = new PutLogsRequest(LOG_PROJECT, LOG_STORE, "pg_stat_activity", "postgresql-monitor", logItems);
           slsClient.PutLogs(request);
       }
   
       private static String getCurrentTime() {
           return formatToIsoInstantWithoutMillis(Instant.now().minus(Duration.ofMinutes(10)).atZone(ZoneId.of("UTC")));
       }
   
       private static String getNextEndTime(String startTime) {
           Instant startInstant = Instant.parse(startTime.replace("Z", ":00Z"));
           return formatToIsoInstantWithoutMillis(startInstant.atZone(ZoneId.of("UTC")).plusMinutes(10).plusSeconds(1));
       }
   
       private static String formatToIsoInstantWithoutMillis(ZonedDateTime zdt) {
           return DATE_FORMATTER.format(zdt.truncatedTo(ChronoUnit.MINUTES));
       }
   
       private enum LogType {
           SLOW, ERROR;
       }
   }
   

   The following table describes specific parameters. You can modify the parameters based on your business requirements.

   Parameter	Description	Example
   LOG_ENDPOINT	The endpoint of Simple Log Service. For more information, see Endpoints.	cn-hangzhou.log.aliyuncs.com
   LOG_PROJECT	The name of the project of Simple Log Service.	rdspg-test
   LOG_STORE	The name of the Logstore of Simple Log Service.	rdspg-sls
   instanceId	The instance ID.	pgm-bp1c82mky1av****

4. Create an alert rule in CloudMonitor based on the created metric. If the metric value of a resource meets the alert conditions, CloudMonitor automatically sends alert notifications. For more information, see Create an alert rule.