openflow-gdrive-deploy

name	openflow-gdrive-deploy
description	Deploy the Google Drive CDC connector on OpenFlow. Handles runtime selection, EAI setup, PAT creation, and NiFi flow deployment.
parent_skill	openflow-gdrive-demo

OpenFlow Google Drive -- Deploy

End-to-end deployment of the Google Drive connector with Cortex Search.

When to Load

Parent SKILL.md routes here on: "deploy", "run", "start", "add connector"

Prerequisites

Before starting, the router SKILL.md has already verified:

• OpenFlow runtime is ACTIVE
• Google Shared Drive confirmed (with documents)
• GCP service account credential exists

Load config using the pattern in references/config-resolution.md.

Workflow

Step 1: Read Runtime Metadata

No user prompts needed -- the bootstrap skill has already run $openflow setup.

1. Read from $openflow cache:

   cat ~/.snowflake/cortex/memory/openflow_infrastructure_${CONNECTION}.json | jq '{
     runtime: .deployments[0].runtimes[0].runtime_name,
     url: .deployments[0].runtimes[0].url,
     role: .deployments[0].runtimes[0].execute_as_role,
     profile: .deployments[0].runtimes[0].nipyapi_profile
   }'
   

   Extract: runtime_name, url, execute_as_role, nipyapi_profile

   Fallback (no $openflow cache):

   If the cache file does not exist, read execute_as_role from the manifest:

   grep execute_as_role .sfutils/manifest.toml
   

   If not in manifest either, ask the user:

   • Use ask_user_question with type: "text", defaultValue: "OPENFLOW_DEMOS_ROLE"
   • Question: "What role does the runtime execute as? (This is the execute_as_role from your deployment.)"

   Persist to manifest: [openflow].execute_as_role

2. Confirm role with user:

   • Present: "Runtime executes as role {execute_as_role}. Use this for schema ownership and flow execution?"
   • User can override.

3. Ask about resource prefix (if not in manifest):

   • Ask via ask_user_question: "Do you want a prefix for Snowflake resources (e.g., MYPROJECT_)? Leave empty for no prefix."
   • Use type: "text" with defaultValue: "" (empty = no prefix)
   • If user provides one, use it; if empty, resources use bare names (e.g., GDRIVE_DEMO_RUNNER, EGRESS_RULE)
   • Persist to [project].resource_prefix

Step 2: Warehouse Discovery

STOP-AND-ASK: This step MUST prompt the user.

1. List warehouses available to the inferred role:

   snow sql -q "SHOW WAREHOUSES" -c $CONNECTION --role $ROLE --format json
   

2. Filter to Small or X-Small warehouses.

3. Present top options via ask_user_question:

   • Show 3-5 suitable warehouses
   • If none suitable: offer "Create {PREFIX}_WH (X-Small) using admin role"
   • STOP: Wait for user selection.

4. If user chose "create":

   snow sql -q "CREATE WAREHOUSE IF NOT EXISTS {PREFIX}_WH WAREHOUSE_SIZE = 'X-SMALL' AUTO_SUSPEND = 60 AUTO_RESUME = TRUE" -c $CONNECTION --role ACCOUNTADMIN
   snow sql -q "GRANT USAGE ON WAREHOUSE {PREFIX}_WH TO ROLE $ROLE" -c $CONNECTION --role ACCOUNTADMIN
   

Step 3: Confirm Drive Parameters

STOP-AND-ASK: This step MUST prompt the user.

The router already confirmed the Shared Drive exists. Here, confirm the specific values:

Google Shared Drive:
  Drive ID:    0AFVVJ5XhjQQCUk9PVA  (from env OPENFLOW_GDRIVE_ID)
  Folder Name: Festival Operations   (from env OPENFLOW_GDRIVE_FOLDER_NAME)

Confirm these values are correct?

Use ask_user_question with resolved values as defaults. User can correct them.

Also verify GCP credential exists:

test -f "$(eval echo $GCP_CRED)" && echo "OK" || echo "MISSING"

If MISSING: Stop and ask user to place key file at the expected path.

Step 4: Database and Schema

STOP-AND-ASK if not in manifest.

If DATABASE or SCHEMA are empty:

• For DATABASE: suggest the database where the runtime lives (from the FQN)
• For SCHEMA: ask user to name the destination schema

Present via ask_user_question and confirm.

Step 5: Verify EAI and PAT (created by bootstrap)

The bootstrap already created the PAT and EAI via $sfutils:programmatic-access-token and $sfutils:network-rule. Read from manifest to verify:

grep -E "sa_user|sa_role|status|eai|network_rule" $PROJECT_DIR/.sfutils/manifest.toml

Check:

• [pat.openflow-runner].status = COMPLETE
• [openflow].eai is set
• [openflow].network_rule is set

If any are missing, the bootstrap didn't complete. Tell user to re-run Gate 0 (bootstrap).

Grant execute_as_role to PAT user (idempotent -- safe if already granted):

snow sql -q "GRANT ROLE {execute_as_role} TO USER {SA_USER}" -c $CONNECTION --role ACCOUNTADMIN

Without this, NiFi data operations fail (PutSnowpipeStreaming needs the runtime role).

Step 6: Verify nipyapi Profile

The bootstrap already created the nipyapi profile (Step 10). Verify it's working:

nipyapi --profile {RUNTIME_KEY} canvas get_root_pg_id

If 401: PAT may have expired or IP changed. Load update-ip/SKILL.md.

If profile not found: Bootstrap didn't complete. Run the bootstrap again (Gate 0 in router SKILL.md).

Step 7: Create Destination Schema and Grant Access

snow sql -q "CREATE SCHEMA IF NOT EXISTS $DATABASE.$SCHEMA" -c $CONNECTION --role ACCOUNTADMIN
snow sql -q "GRANT OWNERSHIP ON SCHEMA $DATABASE.$SCHEMA TO ROLE $ROLE COPY CURRENT GRANTS" -c $CONNECTION --role ACCOUNTADMIN

Ensure the execute_as_role ($ROLE) has access to the database, warehouse, and schema. The runtime's managed Snowflake connection authenticates as this role for ExecuteSQLStatement, EnrichAttributes, and PutSnowpipeStreaming:

snow sql -q "GRANT USAGE ON DATABASE $DATABASE TO ROLE $ROLE" -c $CONNECTION --role ACCOUNTADMIN
snow sql -q "GRANT USAGE ON WAREHOUSE $WAREHOUSE TO ROLE $ROLE" -c $CONNECTION --role ACCOUNTADMIN
snow sql -q "GRANT ALL ON SCHEMA $DATABASE.$SCHEMA TO ROLE $ROLE" -c $CONNECTION --role ACCOUNTADMIN
snow sql -q "GRANT CREATE TABLE ON SCHEMA $DATABASE.$SCHEMA TO ROLE $ROLE" -c $CONNECTION --role ACCOUNTADMIN
snow sql -q "GRANT CREATE STAGE ON SCHEMA $DATABASE.$SCHEMA TO ROLE $ROLE" -c $CONNECTION --role ACCOUNTADMIN
snow sql -q "GRANT CREATE CORTEX SEARCH SERVICE ON SCHEMA $DATABASE.$SCHEMA TO ROLE $ROLE" -c $CONNECTION --role ACCOUNTADMIN

These are idempotent -- safe to run even if grants already exist from setup.sql.

Step 8: Deploy and Start Connector

The connector is pre-installed in the runtime's built-in ConnectorFlowRegistryClient registry (bucket: connectors).

8a. Deploy the flow from the built-in registry:

nipyapi --profile {NIPYAPI_PROFILE} ci deploy_flow \
  --registry_client "ConnectorFlowRegistryClient" \
  --bucket "connectors" \
  --flow "unstructured-google-drive-cdc-no-dwd"

If registry_client by name fails, discover the ID first:

curl -sk -H "Authorization: Bearer $TOKEN" "{ENDPOINT}/controller/registry-clients" | python3 -m json.tool

Use the ID from the response for --registry_client.

8b. Configure non-sensitive parameters:

IMPORTANT: {ROLE} below MUST be the runtime's execute_as_role (resolved in Step 1), NOT OPENFLOW_ADMIN. OPENFLOW_ADMIN is for NiFi canvas access (PAT). The runtime's managed Snowflake connection can only USE ROLE the execute_as_role. Setting OPENFLOW_ADMIN here causes "Role not granted to this user" errors on ExecuteSQLStatement, EnrichAttributes, and PutSnowpipeStreaming.

NIFI_PROCESS_GROUP_ID="{PG_ID}" nipyapi --profile {NIPYAPI_PROFILE} ci configure_params \
  --parameters '{
    "Google Drive ID": "{DRIVE_ID}",
    "Google Folder Name": "{FOLDER_NAME}",
    "Destination Database": "{DATABASE}",
    "Destination Schema": "{SCHEMA}",
    "Snowflake Warehouse": "{WAREHOUSE}",
    "Snowflake Role": "{ROLE}",
    "Snowflake Cortex Search Service User Role": "{ROLE}"
  }'

8c. Configure sensitive parameters (GCP credential):

The GCP Service Account JSON lives in an inherited parameter context ("Source Parameters"). Use configure_inherited_params (not configure_params or upload_asset):

NIFI_PROCESS_GROUP_ID="{PG_ID}" nipyapi --profile {NIPYAPI_PROFILE} ci configure_inherited_params \
  --parameters "{\"GCP Service Account JSON\": $(python3 -c "import json; print(json.dumps(open('{GCP_CRED_PATH}').read()))")}"

8d. Start the flow:

NIFI_PROCESS_GROUP_ID="{PG_ID}" nipyapi --profile {NIPYAPI_PROFILE} ci start_flow

Post-deploy tuning:

Processor	Concurrent Tasks
PutSnowpipeStreaming	8
Fetch Google Drive	4

Step 9: Validate Data Flow

Wait 90-120 seconds, then:

snow sql -q "SELECT TABLE_NAME, ROW_COUNT FROM $DATABASE.INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = '$SCHEMA' ORDER BY TABLE_NAME" -c $CONNECTION --role $ROLE --format json

Expected tables: DOCS_CHUNKS, DOCS_LOCKS, FILE_HASHES

• rows > 0 in FILE_HASHES: connector is ingesting successfully
• rows = 0 after 2 minutes:
  • Check NiFi bulletins: nipyapi --profile {RUNTIME_KEY} canvas get_bulletins
  • Verify EAI is attached: DESCRIBE OPENFLOW RUNTIME <FQN>
  • Verify GCP credential has access to the Shared Drive
  • Verify the Drive ID is a Shared Drive (not a regular folder)

Step 9b: Validate Cortex Search

Once DOCS_CHUNKS has rows, verify the Cortex Search Service is active and queryable:

snow sql -q "SHOW CORTEX SEARCH SERVICES IN SCHEMA $DATABASE.$SCHEMA" -c $CONNECTION --role $ROLE --format json

Expected: CORTEX_SEARCH_SERVICE with indexing_state = ACTIVE and serving_state = ACTIVE.

If the service exists and is active, run a demo search query using the SQL execution tool (not snow sql) so results appear inline in the chat conversation:

WITH search_results AS (
  SELECT PARSE_JSON(
    SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
      '{DATABASE}.{SCHEMA}.CORTEX_SEARCH_SERVICE',
      '{
        "query": "What are the key operational details?",
        "columns": ["FULL_NAME", "CHUNK"],
        "limit": 5
      }'
    )
  ) AS response
)
SELECT
  r.value:"FULL_NAME"::VARCHAR AS document_name,
  LEFT(r.value:"CHUNK"::VARCHAR, 200) AS chunk_preview
FROM search_results,
  LATERAL FLATTEN(input => response['results']) r;

Execute this via snowflake_sql_execute (the CoCo SQL tool), NOT via snow sql bash command. This ensures the query results display directly in the conversation for the user to see.

If the service does not yet exist, wait another 60 seconds and re-check. The connector's Init process group creates it automatically after the first batch of chunks lands.

Step 10: Persist State

Write deployment state to .sfutils/manifest.toml and mark complete:

python3 $(dirname "$(realpath "$0")")/../scripts/gate.py \
  --step deploy --action complete --manifest .sfutils/manifest.toml

If the skill directory is known, use the absolute path:

python3 <SKILL_DIR>/scripts/gate.py --step deploy --action complete --manifest .sfutils/manifest.toml

Stopping Points

• After Step 1: runtime selection
• After Step 2: warehouse selection
• After Step 3: drive parameter confirmation
• After Step 5: if EAI/PAT setup fails
• After Step 9: validation results

Output

Deployment complete!
  Runtime:    {RUNTIME_NAME} (ACTIVE)
  Database:   {DATABASE}.{SCHEMA}
  Connector:  unstructured-google-drive-cdc-no-dwd (RUNNING)
  Tables:     DOCS_CHUNKS, FILE_HASHES populated
  Cortex Search: auto-created after first chunks ingested