September 4, 2020

Making RDS Database Versioning & Instrumentation Easy with Liquibase & CloudFormation

Reposted with permission from Martin Whittington, Head of Platform @oddschecker.

1. CloudFormation Stack Create/Update process starts.
2. RDS instance is created/updated.
3. Lambda-backed custom resource runs.
3a. Fetch database parameters stored on SSM.
3b. Connect to RDS instance and run Liquibase changelogs.
3c. Send CFNResponse success.

Database schema changes are risky – learn how Liquibase can help make these changes in a controlled and stepped manner.

Database schema changes need to be managed carefully. Rolling out updates without fully understanding the ramifications can be catastrophic especially if, for example, a table has been nominated for “the drop”.

The team over at Liquibase has provided a wonderful database versioning tool that can be used to help make these sorts of changes in a controlled and stepped manner.

Let's say we have three tables; Orders, Products, and Sales Reps. An order can contain one or more products. An order is also usually linked to a single sales rep so that the commission can be calculated correctly. Now, let's imagine that a decision has come down from the top that commission is no longer being paid and the database needs updating to reflect this. The change would be removing the link from the orders table to the sales rep table and then eventually removing the sales rep table altogether. Each change would be a version-controlled update using changelogs:

Our master changelog contains all the files to include (run in order top to bottom). Let's assume that db.changelog-1.0.xml was used to create the initial DB schema. We now want to add a new changelog that will:

Drop the FK constraint between the order and sales_rep table.
Drop the sales_rep_id column from the order table.
Drop the sales_rep table.

Now we need to add this to our master log to roll out the update:

Once this is tested on a local non-critical DB, we can package this as part of our AWS Lambda Jar. The Lambda code looks like this:

public class RDSManager implements RequestHandler, Object> {
    
    private static enum Status {
        SUCCESS, FAILED
    }
    
    private AWSSimpleSystemsManagement ssm = AWSSimpleSystemsManagementClientBuilder.defaultClient();
    private String dbName = "whatever";
    
    @Override
    public Object handleRequest(Map input, Context context) {
        LambdaLogger logger = context.getLogger();
        String requestType = input.get("RequestType").toString();
        logger.log(String.format("[INFO] RequestType is %s\n", requestType));
        
        // If the event is equal to delete we don't need to do anything
        if (requestType.equalsIgnoreCase("Delete")) {
            logger.log(String.format("[INFO] RequestType %s -> Exiting\n", requestType));
            return sendCFNResponse(input, Status.SUCCESS, context);
        }
        
        // Get Values from the event
        @SuppressWarnings("unchecked")
        Map resourceProps = (Map)input.get("ResourceProperties");
        String dbHost = resourceProps.get("DBHost").toString();
        
        // Get Parameter values from the env vars
        logger.log("[INFO] Fetching SSM Values\n");
        String userName = null;
        String userPass = null;
        try {
            userName = this.getSSMParameter(System.getenv("DB_USER"));
            userPass = this.getSSMParameter(System.getenv("DB_PASS"));
        } catch (SdkClientException e) {
            return this.handleError(e, input, context);
        }
        
        if (userName != null && userPass != null) {
            logger.log("[INFO] Values fetched OK!\n");
            // Start the DB Connection
            logger.log(String.format("[INFO] Connecting to %s\n", dbHost));
            String url = String.format("jdbc:mysql://%s:3306/%s", dbHost, dbName);
            Connection conn = null;
            try {
                conn = DriverManager.getConnection(url, userName, userPass);
                logger.log("[INFO] Connected to DB!\n");
                Database dataBase = DatabaseFactory.getInstance().findCorrectDatabaseImplementation(new JdbcConnection(conn));
                Liquibase liquiBase = new liquibase.Liquibase("db.changelog-master.xml", new ClassLoaderResourceAccessor(), dataBase);
                liquiBase.update(new Contexts(), new LabelExpression());
            } catch (SQLException | LiquibaseException e) {
                return this.handleError(e, input, context);
            } finally {
                try {
                    if (!conn.isClosed() || conn != null) {
                        conn.close();
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                    logger.log(e.getMessage());
                }
            }
            logger.log(String.format("[INFO] %s provisioned\n", dbHost));
            return sendCFNResponse(input, Status.SUCCESS, context);
        }
        logger.log("[ERROR] Unable to get AWS Parameters\n");
        return sendCFNResponse(input, Status.FAILED, context);
    }
    
    /**
     * Displays the error in the CloudWatch logs and then returns out of the Lambda
     * @param e the current exception
     * @param input The event object
     * @param context The context Object
     * @return The Status, FAILED
     */
    private String handleError(Exception e, Map input, Context context) {
        context.getLogger().log(String.format("[ERROR] Exception - %s", e.getMessage()));
        e.printStackTrace();
        return sendCFNResponse(input, Status.FAILED, context);
    }
    
    /**
     * Gets the SSM Parameter Value from AWS using the paramName
     * @param paramName the name of the param to fetch
     * @return the string value of the param
     */
    private String getSSMParameter(String paramName) {
        if (paramName != null && !paramName.isEmpty()) {
            GetParameterRequest request = new GetParameterRequest()
                    .withName(paramName)
                    .withWithDecryption(true);
            GetParameterResult result = this.ssm.getParameter(request);
            if (result != null) {
                return result.getParameter().getValue();
            }
        }
        return null;
    }
    
    /**
     * Send a response to the CFN build process
     */
    private String sendCFNResponse(Map input, Status status, Context context) {
        LambdaLogger logger = context.getLogger();
        String responseURL = input.get("ResponseURL").toString();
        try {
            URL url = new URL(responseURL);
            HttpURLConnection connection=(HttpURLConnection)url.openConnection();
            connection.setDoOutput(true);
            connection.setRequestMethod("PUT");

            OutputStreamWriter out = new OutputStreamWriter(connection.getOutputStream());
            ObjectMapper mapper = new ObjectMapper();
            
            ObjectNode cloudFormationJsonResponse = mapper.createObjectNode();
            cloudFormationJsonResponse.put("Status", status.toString());
            cloudFormationJsonResponse.put("PhysicalResourceId", context.getLogStreamName());
            cloudFormationJsonResponse.put("StackId", input.get("StackId").toString());
            cloudFormationJsonResponse.put("RequestId", input.get("RequestId").toString());
            cloudFormationJsonResponse.put("LogicalResourceId", input.get("LogicalResourceId").toString());
            cloudFormationJsonResponse.put("Reason", "See details in CloudWatch Log StreamName " + context.getLogStreamName() +" ** GroupName: "+context.getLogGroupName());
            String cfnResp = cloudFormationJsonResponse.toString();
            logger.log("[DEBUG] CF Json repsonse "+cfnResp);
            out.write(cfnResp);
            out.close();
            int responseCode = connection.getResponseCode();
            logger.log("[INFO] Response Code "+responseCode);
        } catch (IOException e) {
            logger.log("[ERROR] Not able to send message to CF Template \nStackTrace "+e.getMessage());
        }
        return status.toString();
    }
}

In logical order, the Lambda does the following:

Checks the request type. If it is a ‘Delete’ request, we just return a success to CloudFormation.
Next, the code fetches AWS SSM Parameters to securely connect to the RDS instance.
Once the parameters have been fetched, a connection to the RDS instance is made.
Once the connection is established, Liquibase runs the changelog file to update the database schema.
Once Liquibase has completed, a success signal is sent back to CloudFormation
The connection is closed.

I’ll let you all figure out the CloudFormation code to run this, that’s the easy part!