# Openflow Gdrive Deploy

> Deploy the Google Drive CDC connector on OpenFlow. Handles runtime selection, EAI setup, PAT creation, and NiFi flow deployment.

- Skill: `snowflake-labs/openflow-gdrive-deploy` (Agent Skill)
- Install (CLI): `npx skillmds@latest add snowflake-labs/openflow-gdrive-deploy`
- Raw SKILL.md: https://api.skillmd.com/api/skills/snowflake-labs/openflow-gdrive-deploy/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: DevOps & Infra
- Author: snowflake-labs (https://skillmd.com/u/snowflake-labs)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/snowflake-labs/openflow-gdrive-deploy

---


# 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:**

   ```bash
   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:

   ```bash
   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:**

   ```bash
   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":

   ```bash
   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:

```bash
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:

```bash
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):

```bash
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:

```bash
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

```bash
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:

```bash
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:**

```bash
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:
```bash
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.

```bash
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`):

```bash
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:**

```bash
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:

```bash
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:

```bash
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:

```sql
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:

```bash
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:

```bash
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
```