Snowflake Openflow Connector for Oracle を SPCS で動かし、PrivateLink経由でオンプレのOracle Databaseに接続してみた

データ事業本部の笠原です。

今回は、SnowflakeのOpenflow Connector for Oracleを使って、Oracle Databaseとの連携を試してみました。
AWSとPrivateLink経由で接続し、Oracle DatabaseのデータをCDC (Change Data Capture) でSnowflakeに転送できるか確認しました。

以前、Openflow Connector for PostgreSQLを使ってSPCS / BYOCの各デプロイモデルでの導入と動作確認を行いましたので、今回はその応用です。

なお、今回は実際のお客様の環境下での構築を元にブログ化しております。

事前の確認

• 標準のコネクタ利用規約以外の追加利用規約も適用されるため、ORGADMIN権限を持つユーザにて追加の利用規約を許諾する必要があります。
• Openflow Connector for OracleにはOracle XStreamサービスの有償ライセンスが必要です。以下2つのライセンスモデルが利用可能です。
  • 組み込みライセンス (Snowflake提供)
    • Oracle XStreamテクノロジのライセンスをSnowflakeの契約を通じて直接購入したい方向け
    • ソースのOracleDBのプロセッサコアの数に基づいて、Snowflakeを通じて請求される。キャンセルできない36ヶ月間の契約が含まれる。サポートおよびメンテナンスサービスの料金も請求される。
      • さらに、標準的なストレージ・コンピューティングコストが適用される
    • コネクタパラメータにOracle DBのCPUコア数とプロセッサ乗数係数を入力する必要がある
    • 最大16個のライセンス済コアに対する60日間の無料トライアルが含まれる。請求は61日目から自動的に開始される。
  • 独立ライセンス (Bring Your Own License - BYOL)
    • すでにOracle GoldenGateライセンス、またはXStreamの資格を提供する別のOracle契約を持っている方向け
    • SnowflakeからのOracle XStreamサービスに対する追加のライセンス、サポート、メンテナンス料金は無し。
      • 標準的なストレージ・コンピューティングコストが適用される
    • CPUコア情報をSnowflakeに提供する必要は無し
    • トライアル期間は無し
• Openflowランタイム要件は以下の通りです
  • ランタイムのサイズは Medium 以上である必要があります
  • マルチノードのOracleランタイムをサポートしていないため、コネクタのランタイムの Min nodes および Max nodes を 1 に設定して構成します
• 次のOracle Databaseのバージョンとプラットフォームがサポートされます
  • Oracle Database 12cR1 以降
  • オンプレミスサーバ
  • Oracle Exadata
  • OCI VM / ベアメタル
  • Oracle向けカスタムAWS RDS
  • Oracle向け標準シングルテナントAWS RDS
• Standard Editionデータベースに対して利用する場合、以下のドキュメント記載内容の通り、Oracleライセンス契約を確認してXStreamの使用が許可されていることを確認してください。

警告

コネクタには、技術的にOracle Database Standard Edition（SE /SE2）と互換性があります。ただし、Oracleのドキュメントでは、「Oracle Database Enterprise Editionへのライセンスは、Oracle XStream をライセンスして使用するための前提条件です」と記載されています。Standard Editionデータベースに対してコネクタを展開する前に、Oracleライセンス契約を確認して、 XStream の使用が許可されていることを確認してください。Oracleライセンスに準拠する責任はお客様にのみあります。

詳細は以下のドキュメントを参照してください。

今回は、BYOLのライセンスモデルを利用してOpenflow Connector for Oracleを利用します。
Oracle Database 19c Enterprise Editionのオンプレミスデータベースに対して接続します。
XStreamの利用は、事前にお客様にて確認済の状態から構築を進めます。

今回の構成

今回の構成は以下のようにしました。

OpenflowはSPCS版で稼働させることにします。
OpenflowからAWSへはPrivateLink接続を行います。今回の場合は、OpenflowからAWSへのOutbound接続としてPrivateLinkを設定します。

AWS側では、VPCエンドポイントサービスとNLBを経由して、オンプレのOracle Databaseに接続します。
AWSとオンプレ間の接続はすでにTrangit Gatewayで構築済となっており、VPC内からオンプレのOracle DatabaseのIPアドレスへ疎通できる状態ですので、NLBのTargetIPでOracle DatabaseのIPアドレスを指定します。

構築手順

基本的には、以下の記事と同じように、データソースとなるDBの設定、Snowflake側の設定を実施し、Openflowの設定を行う流れで構築できます。

1. ネットワーク環境

今回はパブリックサブネットは存在せず、プライベートサブネットのみ存在します。
SSMセッションマネージャー経由でOracleDBに接続するための踏み台EC2インスタンスも作成します。

他、Transit Gateway用のサブネットが別途存在しますが、その部分は今回の範囲外なので割愛します。

AWSTemplateFormatVersion: '2010-09-09'

Parameters:
  ProjectName:
    Type: String
    Default: openflow-ora-pl

  VpcCidr:
    Type: String
    Default: 10.1.0.0/16

  PrivateSubnet1Cidr:
    Type: String
    Default: 10.1.10.0/24

  PrivateSubnet2Cidr:
    Type: String
    Default: 10.1.11.0/24

Resources:
  Vpc:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: !Ref VpcCidr
      EnableDnsSupport: true
      EnableDnsHostnames: true
      Tags:
        - Key: Name
          Value: !Sub '${ProjectName}-vpc'

  PrivateSubnet1:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref Vpc
      CidrBlock: !Ref PrivateSubnet1Cidr
      AvailabilityZone: !Select [0, !GetAZs '']
      Tags:
        - Key: Name
          Value: !Sub '${ProjectName}-private-1'

  PrivateSubnet2:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref Vpc
      CidrBlock: !Ref PrivateSubnet2Cidr
      AvailabilityZone: !Select [1, !GetAZs '']
      Tags:
        - Key: Name
          Value: !Sub '${ProjectName}-private-2'

  PrivateRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref Vpc
      Tags:
        - Key: Name
          Value: !Sub '${ProjectName}-rt-private'

  PrivateSubnet1RtAssoc:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref PrivateSubnet1
      RouteTableId: !Ref PrivateRouteTable

  PrivateSubnet2RtAssoc:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref PrivateSubnet2
      RouteTableId: !Ref PrivateRouteTable

  NlbSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      VpcId: !Ref Vpc
      SecurityGroupIngress:
        - IpProtocol: tcp
          FromPort: 1521
          ToPort: 1521
          CidrIp: !Ref VpcCidr
          Description: Oracle listener from in-VPC clients (bastion / validation).
      SecurityGroupEgress:
        - IpProtocol: -1
          CidrIp: 0.0.0.0/0
          Description: >-
            Allow all outbound so the NLB forwards to the on-prem Oracle target
            (via Transit Gateway) and performs health checks.
      Tags:
        - Key: Name
          Value: !Sub '${ProjectName}-nlb-sg'

  BastionSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: Bastion host - outbound only (SSM Session Manager, dnf, psql to RDS).
      VpcId: !Ref Vpc
      SecurityGroupEgress:
        - IpProtocol: -1
          CidrIp: 0.0.0.0/0
          Description: Allow all outbound.
      Tags:
        - Key: Name
          Value: !Sub '${ProjectName}-bastion-sg'

  BastionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: ec2.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore
      Tags:
        - Key: Name
          Value: !Sub '${ProjectName}-bastion-role'

  BastionInstanceProfile:
    Type: AWS::IAM::InstanceProfile
    Properties:
      Roles:
        - !Ref BastionRole

  BastionInstance:
    Type: AWS::EC2::Instance
    Properties:
      InstanceType: !Ref BastionInstanceType
      ImageId: !Ref LatestAmiId
      IamInstanceProfile: !Ref BastionInstanceProfile
      SubnetId: !Ref PrivateSubnet1
      SecurityGroupIds:
        - !Ref BastionSecurityGroup
      Tags:
        - Key: Name
          Value: !Sub '${ProjectName}-bastion'

Outputs:
  VpcId:
    Value: !Ref Vpc
    Export:
      Name: !Sub '${ProjectName}-VpcId'
  PrivateSubnet1Id:
    Value: !Ref PrivateSubnet1
    Export:
      Name: !Sub '${ProjectName}-PrivateSubnet1Id'
  PrivateSubnet2Id:
    Value: !Ref PrivateSubnet2
    Export:
      Name: !Sub '${ProjectName}-PrivateSubnet2Id'
  NlbSecurityGroupId:
    Value: !Ref NlbSecurityGroup
    Export:
      Name: !Sub '${ProjectName}-NlbSecurityGroupId'
  BastionSecurityGroupId:
    Value: !Ref BastionSecurityGroup
    Export:
      Name: !Sub '${ProjectName}-BastionSecurityGroupId'
  BastionInstanceId:
    Description: Connect with `aws ssm start-session --target <id>`.
    Value: !Ref BastionInstance
    Export:
      Name: !Sub '${ProjectName}-BastionInstanceId'

2. データベース環境 (Oracle Database) 設定

今回は、私の方で実際のOracle Databaseの設定をしていませんが、以下の内容を事前に設定する必要があります。

• アーカイブログモードの有効化
• GoldenGate / XStream replicationの有効化
• Stream Pool Sizeを2.5GB以上空ける (推奨)

また、Snowflakeへ連携するテーブルは、主キー / 一意制約 / 一意索引 / 論理キー宣言 のいずれかが必要になります。

以下は、今回の連携に必要なXStream Outboundサーバの作成と、その管理用ユーザの作成、Snowflakeから接続するためのユーザの作成を示しています。
XStream Outboundサーバ作成時に、連携対象となるOracle Database側のスキーマ・テーブルを指定します。

ALTER SESSION SET CONTAINER = CDB$ROOT;

-- Dedicated tablespace for XStream administrator metadata.
CREATE TABLESPACE xstream_adm_tbs
  DATAFILE '<path_or_omit_on_OMF>/xstream_adm_tbs.dbf'
  SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

CREATE USER c##xstreamadmin IDENTIFIED BY "<STRONG_PASSWORD>"
  DEFAULT TABLESPACE xstream_adm_tbs
  QUOTA UNLIMITED ON xstream_adm_tbs
  CONTAINER = ALL;

GRANT CREATE SESSION TO c##xstreamadmin CONTAINER = ALL;

-- Oracle 21c and EARLIER:
BEGIN
  DBMS_XSTREAM_AUTH.GRANT_ADMIN_PRIVILEGE(
    grantee                 => 'c##xstreamadmin',
    privilege_type          => 'CAPTURE',
    grant_select_privileges => TRUE,
    container               => 'ALL');
END;
/

ALTER SESSION SET CONTAINER = CDB$ROOT;

CREATE USER c##connectuser IDENTIFIED BY "<STRONG_PASSWORD>"
  DEFAULT TABLESPACE users
  CONTAINER = ALL;

GRANT CREATE SESSION       TO c##connectuser CONTAINER = ALL;
GRANT SELECT_CATALOG_ROLE  TO c##connectuser CONTAINER = ALL;
GRANT SELECT ANY TABLE     TO c##connectuser CONTAINER = ALL;

ALTER SESSION SET CONTAINER = CDB$ROOT;

DECLARE
  tables  DBMS_UTILITY.UNCL_ARRAY;
  schemas DBMS_UTILITY.UNCL_ARRAY;
BEGIN
  tables(1)  := NULL;  -- 連携対象となるテーブルを指定する
  schemas(1) := NULL;  -- 連携対象となるスキーマを指定する
  DBMS_XSTREAM_ADM.CREATE_OUTBOUND(
    server_name  => 'XOUT1',
    table_names  => tables,
    schema_names => schemas,
    include_ddl  => TRUE);
END;
/

-- Bind the least-privilege connect user to the outbound server.
BEGIN
  DBMS_XSTREAM_ADM.ALTER_OUTBOUND(
    server_name  => 'XOUT1',
    connect_user => 'c##connectuser');
END;
/

ちなみに、連携対象を後から追加したい場合は以下のようにします。

-- テーブル単位で追加
BEGIN
DBMS_XSTREAM_ADM.ALTER_OUTBOUND(
    server_name    => 'XOUT1',
    table_names    => 'SCH0.TB1, SCH1.TB2',  -- カンマ区切りで複数可
    add            => TRUE,
    inclusion_rule => TRUE);
END;
/

-- スキーマ単位で追加
BEGIN
DBMS_XSTREAM_ADM.ALTER_OUTBOUND(
    server_name    => 'XOUT1',
    schema_names   => 'SCH1',
    add            => TRUE,
    inclusion_rule => TRUE);
END;
/

また、除外方法は以下の通りです。

-- add => FALSE でルール削除
BEGIN
DBMS_XSTREAM_ADM.ALTER_OUTBOUND(
    server_name => 'XOUT1',
    table_names => 'SCHEMA.OLD_TABLE',
    add => FALSE);
END;
/

3. Snowflake側設定 & Principal確認

続いて、Snowflake側の設定に移ります。

Snowsight上のワークシートから、以下のSQLを実行します。

この際、クエリ内の <YOUR_USER> は実際にSnowflakeを作業するユーザ名に置き換えてください。

USE ROLE ACCOUNTADMIN;

-- Admin role for managing OpenFlow deployments and runtimes.
CREATE ROLE IF NOT EXISTS OPENFLOW_ADMIN;

GRANT CREATE OPENFLOW DATA PLANE INTEGRATION ON ACCOUNT TO ROLE OPENFLOW_ADMIN;
GRANT CREATE OPENFLOW RUNTIME INTEGRATION    ON ACCOUNT TO ROLE OPENFLOW_ADMIN;
GRANT CREATE COMPUTE POOL                    ON ACCOUNT TO ROLE OPENFLOW_ADMIN;

CREATE WAREHOUSE IF NOT EXISTS OPENFLOW_WH
  WAREHOUSE_SIZE      = 'XSMALL'
  AUTO_SUSPEND        = 60
  AUTO_RESUME         = TRUE
  INITIALLY_SUSPENDED = TRUE;

-- OpenFlow control database: NETWORKING (network rules) + TELEMETRY (event table).
CREATE DATABASE IF NOT EXISTS OPENFLOW_DB;
CREATE SCHEMA   IF NOT EXISTS OPENFLOW_DB.NETWORKING;
CREATE SCHEMA   IF NOT EXISTS OPENFLOW_DB.TELEMETRY;
CREATE EVENT TABLE IF NOT EXISTS OPENFLOW_DB.TELEMETRY.EVENTS;

-- Give the operator the admin role.
GRANT ROLE OPENFLOW_ADMIN TO USER <YOUR_USER>;

-- IMPORTANT: OpenFlow users must NOT have ACCOUNTADMIN as their default role, or
-- they cannot log into runtimes. Set a non-ACCOUNTADMIN default + secondary roles.
-- Uncomment and adjust:
ALTER USER <YOUR_USER> SET DEFAULT_ROLE = OPENFLOW_ADMIN;
ALTER USER <YOUR_USER> SET DEFAULT_SECONDARY_ROLES = ('ALL');

SELECT SYSTEM$GET_PRIVATELINK_CONFIG();

最後の SELECT SYSTEM$GET_PRIVATELINK_CONFIG(); の結果から privatelink-account-principal のARNをメモしておきます。
このARNは後続の SnowflakePrincipalArn にて設定し、 VpcEndpointServicePermissions にて許可パーミッションに利用します。

4. NLB環境

今回利用するNLBと、NLBのターゲットグループを作成します。
NLBのターゲットグループに指定するIPアドレスは、今回はオンプレにあるデータベースですので、直接指定しています。
また、今回はSnowflakeからPrivateLinkで接続するため、VPCエンドポイントサービスも作成します。

AWSTemplateFormatVersion: '2010-09-09'

Parameters:
  ProjectName:
    Type: String
    Default: openflow-ora-pl

  VpcId:
    Type: AWS::EC2::VPC::Id

  PrivateSubnetIds:
    Type: List<AWS::EC2::Subnet::Id>

  OnPremOracleIp:
    Type: String
    AllowedPattern: '^(10\.\d{1,3}\.\d{1,3}\.\d{1,3}|172\.(1[6-9]|2[0-9]|3[01])\.\d{1,3}\.\d{1,3}|192\.168\.\d{1,3}\.\d{1,3}|100\.(6[4-9]|[7-9][0-9]|1[0-1][0-9]|12[0-7])\.\d{1,3}\.\d{1,3})$'
    ConstraintDescription: Must be an RFC 1918 or RFC 6598 private IPv4 address.

  OraclePort:
    Type: Number
    Default: 1521

  SnowflakePrincipalArn:
    Type: String

  AcceptanceRequired:
    Type: String
    Default: 'true'
    AllowedValues: ['true', 'false']

Resources:
  Nlb:
    Type: AWS::ElasticLoadBalancingV2::LoadBalancer
    Properties:
      Name: !Sub '${ProjectName}-nlb'
      Type: network
      Scheme: internal          # Not internet-facing: reached only via the endpoint service.
      IpAddressType: ipv4
      Subnets: !Ref PrivateSubnetIds
      LoadBalancerAttributes:
        - Key: load_balancing.cross_zone.enabled
          Value: 'true'
        - Key: deletion_protection.enabled
          Value: 'false'
      Tags:
        - Key: Name
          Value: !Sub '${ProjectName}-nlb'

  TargetGroup:
    Type: AWS::ElasticLoadBalancingV2::TargetGroup
    Properties:
      Name: !Sub '${ProjectName}-tg-onprem'
      TargetType: ip
      Protocol: TCP
      Port: !Ref OraclePort
      VpcId: !Ref VpcId
      Targets:
        - Id: !Ref OnPremOracleIp
          Port: !Ref OraclePort
          AvailabilityZone: all
      HealthCheckProtocol: TCP
      HealthCheckPort: traffic-port
      HealthCheckIntervalSeconds: 30
      HealthyThresholdCount: 3
      UnhealthyThresholdCount: 3
      TargetGroupAttributes:
        - Key: preserve_client_ip.enabled
          Value: 'false'
        - Key: deregistration_delay.timeout_seconds
          Value: '30'
      Tags:
        - Key: Name
          Value: !Sub '${ProjectName}-tg-onprem'

  Listener:
    Type: AWS::ElasticLoadBalancingV2::Listener
    Properties:
      LoadBalancerArn: !Ref Nlb
      Protocol: TCP
      Port: !Ref OraclePort
      DefaultActions:
        - Type: forward
          TargetGroupArn: !Ref TargetGroup

  VpcEndpointService:
    Type: AWS::EC2::VPCEndpointService
    Properties:
      NetworkLoadBalancerArns:
        - !Ref Nlb
      AcceptanceRequired: !Ref AcceptanceRequired

  VpcEndpointServicePermissions:
    Type: AWS::EC2::VPCEndpointServicePermissions
    Properties:
      ServiceId: !Ref VpcEndpointService
      AllowedPrincipals:
        - !Ref SnowflakePrincipalArn

Outputs:
  ServiceName:
    Value: !Sub 'com.amazonaws.vpce.${AWS::Region}.${VpcEndpointService}'
    Export:
      Name: !Sub '${ProjectName}-EndpointServiceName'
  EndpointServiceId:
    Value: !Ref VpcEndpointService
    Export:
      Name: !Sub '${ProjectName}-EndpointServiceId'
  NlbArn:
    Description: Imported by 03-phase2-rds-backend.yaml to add the RDS listener.
    Value: !Ref Nlb
    Export:
      Name: !Sub '${ProjectName}-NlbArn'
  NlbDnsName:
    Value: !GetAtt Nlb.DNSName
    Export:
      Name: !Sub '${ProjectName}-NlbDnsName'
  OnPremTargetGroupArn:
    Value: !Ref TargetGroup
    Export:
      Name: !Sub '${ProjectName}-OnPremTargetGroupArn'

5. Snowflake 側の 実行ロール作成

続いて、Snowflake側に戻り、Openflowランタイムの実行ロールの作成を行います。

USE ROLE ACCOUNTADMIN;

-- Openflow Runtime Role
CREATE ROLE IF NOT EXISTS OPENFLOW_RUNTIME_ROLE_ORAONPREM;
GRANT USAGE, OPERATE ON WAREHOUSE OPENFLOW_WH TO ROLE OPENFLOW_RUNTIME_ROLE_ORAONPREM;
GRANT ROLE OPENFLOW_RUNTIME_ROLE_ORAONPREM TO ROLE OPENFLOW_ADMIN;

6. Snowflake 側の egress（PrivateLink）設定

さらに、Snowflake側のPrivateLink設定を行います。

まずは以下のを実行して、Snowflake側のPrivateLinkのエンドポイントを作成します。
この際、先ほど作成したAWS側のVPCエンドポイントサービス名と、ELBのDNS名を指定します。

USE ROLE ACCOUNTADMIN;

-- 作成したVPCエンドポイントサービス名と、Oracle DBへのFQDN (今回はELBのDNS名) を指定
SELECT SYSTEM$PROVISION_PRIVATELINK_ENDPOINT(
  'com.amazonaws.vpce.ap-northeast-1.vpce-svc-XXXXXXXXXXXXXXXXX',
  'XXXXXXXXXXXXXXXXXXXX.elb.ap-northeast-1.amazonaws.com');

AWS側のマネジメントコンソールで、VPCエンドポイントサービスへの接続を許可します。
その後、以下の確認を行います。

-- 接続確認: `available` になるまで待つ
SELECT SYSTEM$GET_PRIVATELINK_ENDPOINTS_INFO();

-- Egressネットワークルールを作成
-- VALUE_LISTには `Oracle DBへのFQDN:ポート名` (ELBのDNS名:ポート名) を指定
CREATE OR REPLACE NETWORK RULE OPENFLOW_DB.NETWORKING.ORACLE_XSTREAM_EGRESS
  MODE       = EGRESS
  TYPE       = PRIVATE_HOST_PORT
  VALUE_LIST = ('XXXXXXXXXXXXXXXXXXXX.elb.ap-northeast-1.amazonaws.com:1521');

-- 外部アクセス統合(EAI)を作成
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION OPENFLOW_ORACLE_PL_EAI
  ALLOWED_NETWORK_RULES = (OPENFLOW_DB.NETWORKING.ORACLE_XSTREAM_EGRESS)
  ENABLED = TRUE;

-- EAI利用の権限設定
GRANT USAGE ON INTEGRATION OPENFLOW_ORACLE_PL_EAI TO ROLE OPENFLOW_RUNTIME_ROLE_ORAONPREM;
GRANT USAGE ON INTEGRATION OPENFLOW_ORACLE_PL_EAI TO ROLE OPENFLOW_ADMIN;

7. Snowflake 側の 宛先DBの作成

Snowflake側の格納先となるデータベースも作成しておきます。

USE ROLE ACCOUNTADMIN;

CREATE DATABASE IF NOT EXISTS ORACLE_ONPREM_DEST;
GRANT USAGE         ON DATABASE ORACLE_ONPREM_DEST TO ROLE OPENFLOW_RUNTIME_ROLE_ORAONPREM;
GRANT CREATE SCHEMA ON DATABASE ORACLE_ONPREM_DEST TO ROLE OPENFLOW_RUNTIME_ROLE_ORAONPREM;

-- Optional: let OPENFLOW_ADMIN inspect the destination during verification.
GRANT USAGE ON DATABASE ORACLE_ONPREM_DEST TO ROLE OPENFLOW_ADMIN;

8. Snowflake Openflow 設定

ここで、Openflowの設定を行います。SnowsightのUIで設定します。

8.1 追加の利用規約の承認

ORGADMIN権限を持つユーザにて、OracleのOpenflowコネクタ利用の際に追加で必要な利用規約の承認を事前に行います。
この承認設定は組織で1回行えば問題ないです。

Snowsight上の左側メニュー下でロールをORGADMINに切替し、左側メニューの「Admin(管理者)」»「Terms(規約)」を選択し、「Openflow」の「Oracleライセンス規約」を確認し、全て「Accept」してください。

8.2 Openflowデプロイメント作成

作業ユーザを OPENFLOW_ADMIN ロールに切り替えて実施します。

Snowsight上から左側メニューの「Ingestion(取り込み)」»「Openflow」を選択し、Openflowを起動します。
デプロイメント作成から、Locationは「Snowflake (SPCS)」を選択します。

Snowsight上の左メニューから「取り込み」>「Openflow」をクリックし、「Openflowを起動」をクリックします。

Openflowの画面が表示されたら、「Create a deployment」をクリックします。

「Prerequesites」の画面では「Next」をクリックし、「Deployment location」の画面では「Snowflake」を選択して、「Name」欄に適宜デプロイメント名を入力して「Next」をクリックします。

「Deployment configuration」では、今回は特に何も入力せずそのまま「Create deployment」をクリックします。

OpenflowデプロイメントのステータスがActiveになったら、次に進みましょう。

8.3 Openflowランタイム作成

次にOpenflow Runtimeを設定します。

Openflowの画面から「Create a runtime」をクリックします。

「Create runtime」の画面では以下の設定を行い、「Create」をクリックします。

• Deployment
  • 作成したOpenflowデプロイメント名を選択
• Runtime Name
  • 適宜入力する
  • ここでは ORAONPREM とした
• Node size
  • Medium 以上を選択
  • ここでは Medium を選択した
• Min nodes
  • 1
• Max nodes
  • 1
  • Oracle Databaseコネクタ要件により、マルチノードをサポートしていないため、単一ノードとなるようMin nodes/Max nodesとも 1 を指定します
• Execute-as role
  • OPENFLOW_RUNTIME_ROLE_ORAONPREM を選択
• External Access Integrations
  • OPENFLOW_ORACLE_PL_EAI を選択

これでランタイムが作成されます。
作成には2分程度かかります。ランタイムのステータスが「Active」になるまで待ちましょう。

8.4 Openflowコネクタ導入

Openflow Runtimeが作成されたら、Oracle Database用のコネクタを導入します。

Oracle Database用のコネクタは2種類あります。各ライセンスモデルに対応してます。

• Oracle Embedded License
• Oracle Independent License

今回は「独立ライセンス (BYOL)」で利用するので、「Oracle Independent License」を選択します。
選択後「Install」をクリックし、対象となるランタイムを選択して、コネクタをインストールします。

なお、「Install」のボタンが表示されず、以下のように表示された場合は 8.1 に戻って不足している利用規約を承認しましょう。
承認後はOpenflowランタイムは再作成する必要があるので、再作成後に改めてOpenflowコネクタをインストールしましょう。

成功すると、キャンバスにコネクタのプロセスグループが表示されます。

8.5 コネクタ設定

キャンバス上では、以下の設定を行います。

まず、Oracleと書かれたボックスを右クリックして「Parameters」をクリックします。
Source / Destination / Ingestion の各値を以下のようにし、Source / Destination / Ingestion 毎に「Apply」をクリックして適用します。

「Parameters」クリック時には、Source / Destination / Ingestion パラメータ設定画面のいずれかがポップアップ表示されますので、適宜設定したら別のパラメータ設定に移りましょう。

Source (Oracle Databaseへの接続設定)

Destination (Snowflakeへの接続設定)

Ingestion (取り込み設定)

8.6 フローの開始

キャンバスのボックスでない部分を右クリックし、「Enable all Controller Services」をクリックします。
続いて、キャンバスのボックスを右クリックし、「Start」をクリックします。

上記の順に実行するとコネクタが起動し、初回スナップショット → 増分（CDC）の順で取り込みを開始します。

動作確認

コネクタが起動したら、動作確認をします。

初回スナップショット確認

まずは初回スナップショットが成功している確認しましょう。
初回スナップショットは、コネクタ接続が成功している場合、順次Snowflakeに反映されます。

CDC (差分転送)

次に、ソース側のOracle Databaseにてデータ更新を行い、その内容がSnowflakeへ反映されているか確認します。

注意点としては、Oracle側にて変更内容がアーカイブログに出力されないとSnowflake側に反映されない点です。
Oracle側のアーカイブログ出力のタイミングを事前に確認しておきましょう。
アーカイブログ出力が行われると、Snowflake側のジャーナルテーブルに一旦反映された後、Openflowコネクタの Merge Task Schedule のタイミングでSnowflake側の実際の連携先テーブルに反映されます。

データソースの追加

すでに連携済のOracle Database内のスキーマ・テーブルを追加したい場合は、以下の手順を踏みます。

1. Oracle XStream Outbound Serverに対象スキーマ・テーブルを追加
2. Snowflake Openflowコネクタの Ingestion の Included Table Names にて、対象スキーマ・テーブルを追加
3. Snowflake Openflowコネクタを再起動 (「Stop」して「Start」)

他に、例えば、以下のように別のAWSアカウントのVPC上にあるデータベースを追加で接続する場合を考えます。

この場合は、NLB側にてポートを多重化し、例えば追加接続先は 15521 番ポートで接続するように構成します。
こうすることで、Snowflake側のPrivateLinkの作成個数を抑えることができます。

この場合、Snowflake側の EgressネットワークルールのVALUE_LIST に 15521 番ポートの接続先を追加し、
NLBにて追加の接続先をターゲットに設定する必要があります。
その上で、Openflowにてランタイムを追加して 15521 番ポートの接続設定を行います。

トラブルシューティング

Openflow Connector for Oracleを利用するにあたり、いくつかトラブルが発生したので、その内容をまとめてみました。

全体像

まずは、下記のどこで止まっているかを切り分けるのが基本かと思います。

1. Capture (LCR生成)
2. Outbound Server (LCRをコネクタへ送信)
3. Connector (Snowflake側にて snapshot / journal テーブル / merge で宛先テーブルへ書込)

• 1. が止まる
  • v$xstream_capture が WAITING FOR REDO
  • bytes_of_redo_mined=0
  • captured_scn に変化なし
• 2. が止まる
  • v$xstream_outbound_server が IDLE
  • total_messages_sent=0
• 3. が止まる
  • Snowflake側のjournalテーブルに来ているのに宛先に反映されない／CREATE 失敗等

Oracle 接続 / XStream attach

ORA-26827: Insufficient privileges to attach to XStream outbound server

• 原因
  • コネクタの接続ユーザーが outbound の connect_user と不一致。
  • connect_user 以外は attach 不可（DBA でも不可）。
• 対処
  • SELECT connect_user FROM dba_xstream_outbound; の値を、コネクタの Username に設定。

ORA-17002: Connection reset

• Stop 時に出るのは無害
  • capture 再起動等でセッションが切れた後始末
• Running 中に頻発する場合
  • NLB の TCP アイドルタイムアウト（~350 秒）でアイドル接続が切られている可能性が高い
    • Oracle SQLNET.EXPIRE_TIME 等で対処。CDC が定常的に流れれば起きにくい。

ORA-26808 / ORA-01688（capture/apply が SYSAUX 満杯で ABORTED）

• 原因
  • 消費されない LCR が SYS.STREAMS$_APPLY_SPILL_MSGS_PART（SYSAUX表領域）に溜まっていき、SYSAUXの容量が満杯で異常終了。
• 対処
  • コネクタを接続させて、LCRを消費させ続ける（放置しない）
  • SYSAUX 表領域の容量に余裕を持たせる、かつ、autoextendを有効化し、表領域の使用率を監視
• 参考: XStream Out Concepts §3.3.6。

実際の例: CDC が流れない（初回スナップショットは来るが差分が来ない）

症状

• 宛先テーブルの件数がスナップショット時のまま／journal 表 0 件／v$xstream_outbound_server が IDLE・送信 0／v$xstream_capture が WAITING FOR REDO・bytes_of_redo_mined=0。

切り分け

-- ① capture が前進しているか（複数回計測。scn_behind が縮む／bytes が増えるか）
SELECT c.capture_name, c.status, c.captured_scn, d.current_scn,
       (d.current_scn - c.captured_scn) AS scn_behind
FROM   dba_capture c CROSS JOIN v$database d;
SELECT capture_name, state, bytes_of_redo_mined, total_messages_captured FROM v$xstream_capture;

-- ② outbound がコネクタへ送信しているか
SELECT server_name, state, total_transactions_sent, total_messages_sent FROM v$xstream_outbound_server;

-- ③ Snowflake：journal（中間）に来ているか／ログ
SHOW TABLES IN SCHEMA <DEST_DB>.<DEST_SCHEMA>;          -- *_JOURNAL_* の行数
SELECT TIMESTAMP, VALUE FROM OPENFLOW_DB.TELEMETRY.EVENTS
WHERE RECORD_TYPE='LOG' AND (VALUE ILIKE '%ORA-%' OR VALUE ILIKE '%FAILED%' OR VALUE ILIKE '%<TABLE>%')
ORDER BY TIMESTAMP DESC LIMIT 200;

判定

今回の主な原因：capture 停滞（消費されない期間 ＋ 24h アーカイブログ物理削除）

• XStream の capture は消費クライアント（コネクタ）が繋がって初めて健全に前進する。XStream作成後にOpenflow Connectorによる接続まで数日空くと大きく遅延し、今回の環境では24時間アーカイブログが物理削除される環境だったので、必要な redo が消え、resume 不能になる。
• 判定は状態文字列でなく scn_behind の推移と bytes_of_redo_mined が増えるかで確認する。今回は、横ばい/増えない＝停滞であった。
• ただし v$xstream_capture.state の次の2つは正反対なので混同しないこと。

  state	意味	判定
  WAITING FOR REDO	必要な redo（アーカイブログ）が手に入らず待機	異常状態（本セクションの事象）
  WAITING FOR TRANSACTION	redo は読み切り、次のトランザクション発生を待機	正常なアイドル状態

• 注意
  • v$archived_log が DELETED=NO／STATUS=A でも、RMAN CROSSCHECK 未実施だとカタログ上存在すると判定されていても実体が無いことがある。
    • bytes_of_redo_mined=0 が続くなら物理削除を疑う

対処

必要な redo（capture の required_checkpoint_scn 以降）を取得できるか？
├─ RESTORE できる（RMAN バックアップに残存）
│    → 復元 → capture が自動で追いつく → データ欠落なし
└─ 復元不可（物理削除・バックアップ無し）
     → XStream Outbound Server を現在 SCN で再構築 → 再スナップショット

今回は復元不可だったので、XStream Outbound Serverを再作成して、Snowflake側で接続してスナップショットを取得しました。
なお、Snowflake側にテーブルが残った状態で再スナップショットを取得すると内部で CREATE TABLE を実行するようでエラーになります。事前に対象テーブルをDropしてから際スナップショットを取得しましょう。

Incremental Load が動かない

症状

• Oracle 側
  • dba_xstream_outbound.status が DETACHED のまま、v$xstream_outbound_server が0件。
• Openflow canvas
  • 「Incremental Load」グループの集計は Running 31 / Stopped 0 / Invalid 0 と正常に見えるが、グループを開いて個々のプロセッサを見ると、5分間 In/Out/Read-write が全てゼロ。
• コネクタのプロセスグループ全体を何度 Stop→Start しても変化なし。NLBアイドルタイムアウト対策やController ServiceのDisable/Enableを試しても無関係だった。

見分け方

「Incremental Load」グループの中から、Oracle へ直接接続する担当プロセッサ Read Oracle CDC Stream （プロセッサ種別 CaptureChangeOracle、com.snowflake.openflow.runtime）を探し、右クリックメニューを見る。

• メニューに Start が無く Terminate のみ表示される
• 本当に「Stopped（完全に停止済み）」なら Start が出るはずが、なぜか表示されていない。
  • これは「Stopの指示は送られたが、内部で実行中のスレッドが実際には終了していない（ハングしている）」ことと推測。

原因

プロセッサ内部でブロッキング読み込み中だったスレッドが応答不能のまま固まった可能性がある。
通常のStop操作は「新規実行の停止要求」を送るだけで、既にブロックされているスレッドを強制終了しない。
そのため、プロセスグループ単位のStop/Startを何度実行しても、実体は変わらず。

対処

1. Read Oracle CDC Stream を右クリック → Terminate をクリックして、ハングしたスレッドを強制終了
2. 数秒後、再度右クリック → 今度は Start が表示されるので選択
3. Openflow canvas で「Incremental Load」グループの Read/write バイト数が動き出すか確認
4. 【Oracle側で実行】

   SELECT server_name, connect_user, capture_name, status FROM dba_xstream_outbound;   -- ATTACHED に戻るか
   SELECT server_name, state, TO_CHAR(startup_time,'YYYY-MM-DD HH24:MI:SS') AS startup_time FROM v$xstream_outbound_server;  -- 新しい startup_time の行
   

5. 再接続後、5分空けて2回、バックログが解消しているか確認：

   SELECT c.capture_name, c.status, TO_CHAR(c.captured_scn) AS captured_scn, TO_CHAR(d.current_scn) AS current_scn, (d.current_scn - c.captured_scn) AS scn_behind FROM dba_capture c CROSS JOIN v$database d;
   

教訓

• グループ単位の集計（Running X / Stopped 0 / Invalid 0）は、個々のプロセッサの真の健全性を保証しない。
  • スレッドハングは「Stopped」でも「Invalid」でもなく「Running」のまま表示される。疑わしい時は必ずグループの中に入り、個々のプロセッサの5分間統計（In/Out/Read-write）を確認する。
• 「右クリックメニューに Start が無い」は、スレッドハングを示す最も分かりやすい外部シグナル。Stopボタンやプロセスグループ操作だけでは検出できない。

まとめ

いかがでしたか。

SnowflakeのOpenflowを使って、オンプレにあるOracle DatabaseのデータをSnowflakeに転送してみました。
Oracle Databaseの事例はあまりないので、お客様に確認して検証してみました。

トラブルシューティングは他にも細かい内容のものもありましたが、結構ハマったものを代表的に書き出してみました。

この記事がお役に立てれば幸いです。