Replication failover example

This page shows an example of how a Sync Engine instance in one Snowflake account might transfer responsibilities to a Sync Engine instance in another account, for example in the case of a whole region outage.

In the example we set up a failover for continuous replication, but you can also get the same result with just a replication group.

Diagram

Setup process

Source side

In this scenario, we have created a sync with slug my-sync-slug, and the storage location has been set to store both the raw tables and the normalized views in PROD.RAW.

We use a database role named PROD.OMNATA_ROLE to simplify ownership.

create database PROD;
create schema PROD.RAW;
create database role PROD.OMNATA_ROLE;
grant ownership on schema PROD.RAW to database role PROD.OMNATA_ROLE;
grant usage on schema PROD.RAW to role ACCOUNTADMIN; -- or some other role
grant ownership on future tables in schema PROD.RAW 
to database role PROD.OMNATA_ROLE;

This database will contain the sync'd data, but we also want to keep track of the configuration of the sync, as well as the state of all the streams. The stream state is what allows incremental syncs to continue from where they finished in the previous run.

We cannot just replicate the sync engine metadata directly; Snowflake does not support replication of native applications, plus it's stored in a hybrid table which also cannot be replicated.

So instead, we create a separate database:

create database OMNATA_CONFIGURATION_REPLICA;

create or replace table OMNATA_CONFIGURATION_REPLICA.PUBLIC.SYNC(
    SYNC_SLUG VARCHAR(255),
    SYNC_PARAMETERS OBJECT,
    INBOUND_STREAMS_CONFIGURATION OBJECT,
    INBOUND_STORAGE_CONFIGURATION OBJECT,
    INBOUND_LATEST_STREAMS_STATE OBJECT
);

Then we create a post run hook which takes a configuration snapshot from the sync engine:

create database OMNATA_SCRIPTS;

CREATE OR REPLACE PROCEDURE OMNATA_SCRIPTS.PUBLIC.SYNC_RUN_POST_HOOK(result OBJECT)
RETURNS VARCHAR NOT NULL
LANGUAGE SQL
EXECUTE AS OWNER
AS
BEGIN
  merge into OMNATA_CONFIGURATION_REPLICA.PUBLIC.SYNC tgt
  using (
    select * from OMNATA_SYNC_ENGINE.DATA_VIEWS.SYNC
    where SYNC_SLUG=:result:"sync_slug"::varchar
  ) src
  on src.SYNC_SLUG = tgt.SYNC_SLUG
when not matched then 
    insert (SYNC_SLUG, 
            SYNC_PARAMETERS,
            INBOUND_STREAMS_CONFIGURATION,
            INBOUND_STORAGE_CONFIGURATION,
            INBOUND_LATEST_STREAMS_STATE)
    values (src.SYNC_SLUG,
        src.SYNC_PARAMETERS,
        src.INBOUND_STREAMS_CONFIGURATION,
        src.INBOUND_STORAGE_CONFIGURATION,
        src.INBOUND_LATEST_STREAMS_STATE)
when matched then 
    update set tgt.SYNC_PARAMETERS = src.SYNC_PARAMETERS,
        tgt.INBOUND_STREAMS_CONFIGURATION = src.INBOUND_STREAMS_CONFIGURATION,
        tgt.INBOUND_STORAGE_CONFIGURATION = src.INBOUND_STORAGE_CONFIGURATION,
        tgt.INBOUND_LATEST_STREAMS_STATE = src.INBOUND_LATEST_STREAMS_STATE
;
END;

-- grant access to the sync engine, it will be automatically invoked at the end of every run
grant usage on database OMNATA_SCRIPTS to application OMNATA_SYNC_ENGINE;
grant usage on schema OMNATA_SCRIPTS.PUBLIC to application OMNATA_SYNC_ENGINE;
grant usage on procedure OMNATA_SCRIPTS.PUBLIC.SYNC_RUN_POST_HOOK(OBJECT) to application OMNATA_SYNC_ENGINE;

Finally, we create a failover group to sync these two databases to the other account:

CREATE FAILOVER GROUP RAW_DATA_REPLICATION 
  OBJECT_TYPES = DATABASES 
  ALLOWED_DATABASES = PROD,OMNATA_CONFIGURATION_REPLICA
  ALLOWED_ACCOUNTS = MY_ORG.ACCOUNT_B 
  REPLICATION_SCHEDULE = '10 MINUTE';

Target side

First, the sync engine must have been installed, along with the plugin application. There will also need to be a connection created to the same environment as the primary - it is not practical to replicate secrets.

Assuming a sync run and replication has occurred since you configured the source side, the latest configuration and state should be visible:

You can use this data to create a sync on the target side:

call OMNATA_SYNC_ENGINE.API.CONFIGURE_OMNATA_INBOUND_SYNC(
        sync_slug => 'my-sync-slug',
        sync_name => 'Replicated sync',
        connection_slug => 'my-connection-slug'::variant,
        sync_parameters => (select SYNC_PARAMETERS from OMNATA_CONFIGURATION_REPLICA.PUBLIC.SYNC where SYNC_SLUG='omnata-jira-to-snowflake'),
        sync_schedule => OBJECT_CONSTRUCT(
          'mode', 'manual',
          'time_limit_mins', 240,
          'warehouse', 'COMPUTE_WH'
        ),
        stream_names => (select OBJECT_KEYS(INBOUND_STREAMS_CONFIGURATION:"included_streams")::array from OMNATA_CONFIGURATION_REPLICA.PUBLIC.SYNC where SYNC_SLUG='omnata-jira-to-snowflake'),
        sync_strategy => (select object_agg(KEY,VALUE:"sync_strategy")::variant
            from OMNATA_CONFIGURATION_REPLICA.PUBLIC.SYNC,table(flatten(INPUT => INBOUND_STREAMS_CONFIGURATION:"included_streams"))
            where SYNC_SLUG='my-sync-slug'),
        storage_behaviour => (select object_agg(KEY,VALUE:"storage_behaviour")::variant
            from OMNATA_CONFIGURATION_REPLICA.PUBLIC.SYNC,table(flatten(INPUT => INBOUND_STREAMS_CONFIGURATION:"included_streams"))
            where SYNC_SLUG='my-sync-slug'),
        stream_primary_keys => (select object_agg(KEY,VALUE:"primary_key_field")::variant
            from OMNATA_CONFIGURATION_REPLICA.PUBLIC.SYNC,table(flatten(INPUT => INBOUND_STREAMS_CONFIGURATION:"included_streams"))
            where SYNC_SLUG='my-sync-slug' and VALUE:"primary_key_field"::varchar is not null),
        stream_cursor_fields => (select object_agg(KEY,VALUE:"cursor_field")::variant
            from OMNATA_CONFIGURATION_REPLICA.PUBLIC.SYNC,table(flatten(INPUT => INBOUND_STREAMS_CONFIGURATION:"included_streams"))
            where SYNC_SLUG='my-sync-slug' and VALUE:"cursor_field"::varchar is not null),
        inbound_storage_location => (select INBOUND_STORAGE_CONFIGURATION from OMNATA_CONFIGURATION_REPLICA.PUBLIC.SYNC where SYNC_SLUG='omnata-jira-to-snowflake'),
        raise_errors => true
        );

This query picks out the sync configuration metadata from the source sync engine and applies it to the local one. This query can be ran on a schedule to ensure that the sync configuration is kept up to date.

We also want to keep the sync state up to date:

call OMNATA_SYNC_ENGINE.API.SET_INBOUND_SYNC_STATE(
        sync_slug => 'my-sync-slug',
        streams_state => (select INBOUND_LATEST_STREAMS_STATE
                from OMNATA_CONFIGURATION_REPLICA.PUBLIC.SYNC
                where SYNC_SLUG='my-sync-slug')
);

After this point, you should be able to see the sync in the UI, and it should be configured the same as on the source account, with the stream state applied (visible on the Status tab).

Failing over

If you try to run a sync on the secondary, you will get this error:

By marking the failover group as the primary:

ALTER FAILOVER GROUP RAW_DATA_REPLICATION PRIMARY

You make the PROD database writable, and syncs will start to work.

Failing back

This example doesn't cover replicating the sync state back in the other direction, but assuming that ACCOUNT_B did not act as the primary for a significant amount of time, you can just make ACCOUNT_A the primary again and it will resume from where it left off before the outage:

ALTER FAILOVER GROUP RAW_DATA_REPLICATION PRIMARY