There are several way to achieve a “Hybrid Cloud Architecure”, meaning to connect Snowflake to your on-premise or private Cloud resources and also the other way around.

You can use Snowflake External Access Privatelink or you can expose an identity aware reverse proxy like Oauth2-Proxy which Snowflake functions can access using a bearer token.

However, this post is particulary about how you can setup (hands-on) an SSH tunnel from Snowflake to your companies DMZ (demilitarized zone) in order to let Snowflake access resources behind your firewall such as on-premise data lakes, databases, APIs, files or AI models etc.

These resources may also be in a VPC/ VNET in a private Cloud network.

For those of you who can not wait to see the code, here is the Github to all the code including a Docker compose file to test locally:

https://github.com/KellerKev/snowflake-ssh-tunnel.git

Use cases:

This hybrid architecture can tremendously accelerate the value which you can get out of Snowflake during your transition to the Cloud by accessing resources from the Cloud that are not yet migrated.

Hence you can use the benefits of Elastic Cloud Computing today even with legacy services.

Notably on the security front there are also huge benefts, since there are some APIs, AI models or data that will never move to the Cloud for compliance reasons.

This services as well as on-premis data can now be accessed using Snowflake and this access can be cut anytime.

Reference Architecture:

In essence the on-premise or private Cloud resources firewall will not open any ports, but merely create an outbound connection via VPN (or other type of TCP/SSH tunnels).

It may be connected to the DMZ via peering or just by being generally reachable in the same network (2).

On the DMZ the firewall will need to open an SSH port in order to be able to esablish an SSH tunnel or several SSH tunnels with Snowflake (1).

The DMZ will act as a bridge between the on-premise or private Cloud network and the the traffic connecting via the tunnel with Snowflake in order to keep all ingress ports on the firewall closed.

The SSH tunnel from Snowflake will be established in this demo using Snowflake Container Services (SPCS).

The container will use an SSH privatekey to establish the tunnel during start up.

A note on security:

In production, SSH private keys should be avoided and it is advisable to use short lived SSH certificates instead.

You can setup an automated process that spins the container up when needed, in order to retrieve a short lived SSH certificate from your backend and use Snowflake Snow-CLI to spin up the container and re-establish a tunnel using that certificate.

In the same manner, the way hard coded secrets are used in this demo in the notebooks and code, should not be done like this in production.

I would advise to setup proper credential brokering.

For example implement a Device Code OAuth Flow for the Jupyter notebooks to have users obtain OAuth tokens via SSO (Single Sign On) so that they can access the Iceberg Catalog.

Another recommendation is to use more Snowflake aligned open source Iceberg catalog such as Apache Polaris that implements role based access and policy based access controls similiar to Snowflake.

You may also want to consider other flows which will let the container obtain secrets from a secret manager in the backend, especially for the Streamlit example.

Secrets in code must be avoided in production.

The secrets provided via the environment variables (SSH key, user, host) will not be accessible from within the container by the way.

They will only be accessible when the container spins up and establishes the connection, so that part is solved in the demo.

However, as can be concluded, this demo is intented as a proof of concept and shows the technical feasability to access private resources behind a firewall using different query engines from Snowflake.

It is not a full fledged production setup, it however provides the technical key elements needed to make such an architecture work.

Hopefully saving you valuable time in the process and unlocking new use cases.

By the way, why are we using SSH and not OpenVPN or Wireguard etc. ?

Those VPN services require the NET_ADMIN privilege for a container, which are not given in Snowflake container services today.

The Demo:

We will run a container in Snowflake that has two query engines (DuckDB + Spark) available to query an Iceberg Datalake with an Iceberg Rest Catalog behind a firewall.

The query engines can be used in a provided Jupyter notebook or VSCode instance (both provided in the container), in a Streamlit application (also in the container) plus a REST API that can be called from within Snowflake as a UDF.

Let’s have a quick look at these video snippets for the things you can expect and what we will be building.

1. Query an Iceberg Datalake and Catalog which are both hosted on-premise behind a firewall from Snowflake using the PRIVATE IP addresses of the self hosted Iceberg REST Catalog and Iceberg S3 Minio Datalake storage within a Jupyter notebook hosted in Snowflake SPCS.

One notebook is using DuckDB (the first) and the second notebook is using Apache Spark as a query engine.

You can create additional dataframes querying Snowflake data and join this data with the on-premise data frames in the notebook.

Nice thing to note is, because we are using Spark or DuckDB as query engines in the container and not a Snowflake virtual warehouse there will be no result set cache in Snowflake from the on-premise data being queried.

Once the tunnel connection is cut there is no more data access to the on-premise data from Snowflake or anything cached in Snowflake’s internal stages.

a) DuckDB query engine (with PyIceberg) to query the on-premise Iceberg data lake from Snowflake via SSH tunnel in a Jupyter notebook:

b) Apache SPARK query engine to query the on-premise Iceberg data lake from Snowflake via SSH tunnel in a Jupyter notebook:

2. Query the On-Premise Data Lake directly from a Snowflake UDF:

You can create a Snowflake UDF that talks to a REST API in the container, which executes the query (using DuckDB) against the Iceberg on-premise data lake directly from within Snowflake (in the bounds of the private container network) and use the UDF to join data directly with Snowflake tables in a Snowflake workbook or notebook.

Note, that in thise case, the results of the UDF will be cached in the Snowflake internal stage result set cache.

UDFs that call container functions do not support the ablity to return tables, but only scalar results in Snowflake today.

Meaning the resulting payload can not be bigger than 16MB and needs to be normalized / flattened into a table.

I am sure that the Snowflake team will however address this limitation in the future.

You do not have any limitations joining data with Snowflake and on-premise data in the Jupyter notebook provided however.

3. Embed the on-premises Iceberg query in a Streamlit Application in the container.

Here you can also see VSCode is also running in the Snowflake container for some spontanous edits.

You can also use VSCode notebooks and the VSCode Snowflake extension by the way.

Building the Demo:

So how can we set this up?

Before we setup anything in Snowflake itself, you will need an Iceberg S3 data lake and Iceberg REST Catalog with sample data.

For that you need get yourself a Linux server that is accessible via SSH from the internet and that you can use to install the open source Iceberg Rest Catalog and S3 Minio storage using Docker compose.

If you dont have access to a Linux server you can play with you can get 2 free ones from Oracle Cloud:

Explore Oracle Cloud Infrastructure

While in this demo will be hosting the Iceberg Catalog and data lake directoy on the DMZ, with ability to get 2 free servers you can also setup the reference architecture described earlier, but using one free Linux VM as the DMZ for the SSH tunnel and the other one to install the Iceberg data lake on.

Either way, once you have a Linux VM you will need to do at least 4things:

a) Create a non-root user

b) generate an SSH keypair for the user

c) install Docker, Docker compose and run the Iceberg data lake sample

d) For the intial setup make sure you can reach ports 9000,9001,8888,8889 temporary via http for the setup. (in Oracle Cloud you have open those ports in the security group for your Linux servers).

We will close down those ports later again.

You usually also have to open the ports with the Linux iptables firewall:

sudo iptables -I INPUT 2 -p tcp --dport 8888 -j ACCEPT
sudo iptables -I INPUT 2 -p tcp --dport 8889 -j ACCEPT
sudo iptables -I INPUT 2 -p tcp --dport 9001 -j ACCEPT
sudo iptables -I INPUT 2 -p tcp --dport 9000 -j ACCEPT
sudo iptables-save > /etc/iptables/rules.v4

While for points a) and b) I am sure you can find all the information you need on the internet, but to get the Iceberg data lake example tu run for this demo you have to do the following:

Git clone from my adapted Duckberg (DuckDB to query Iceberg) Git repository fork:

git clone https://github.com/KellerKev/duckberg

Now change directory in the following folder:

cd duckberg/duckberg/playground

Now in the playground folder you can start:

docker compose up -d

Once everysthing is up and running go to this URL of your Linux VM and port to access the Jupyter notebook provided so that you can create and populate your Iceberg data lake with sample data:

http://x.x.x.x:8889/notebooks//000%20Init%20Iceberg%20data.ipynb

Where x.x.x.x is the ip address of your Linux VM.

You should see this notebook:

Now go ahead and execute every cell.

To test if everything is ok you can go to:

http://x.x.x.x:8888/notebooks/001%20-%20Duckberg%20simple%20query%20-%20REST%20Iceberg%20catalog.ipynb

And query your data lake with Duckberg locally:

You will also be able to access the s3 mino storage to see your parquet files and metadata files in the object storage:

http://x.x.x.x:9001/

Now that everything is up and running in your Linux VM you can close the ports 9000,9001,8888 and 8889.

You can do that in the security groups in Oracle Cloud (or AWS, Azure etc.) or your firewall and you can also close them down again with iptables:

sudo iptables -I INPUT 2 -p tcp --dport 8888 -j DROP
sudo iptables -I INPUT 2 -p tcp --dport 8889 -j DROP
sudo iptables -I INPUT 2 -p tcp --dport 9001 -j DROP
sudo iptables -I INPUT 2 -p tcp --dport 9000 -j DROP
sudo iptables-save > /etc/iptables/rules.v4

The only port reachable going forward should be port 22 for SSH.

With this in place we can not setup the Snowflake container to create the SSH tunnel and query the data behind the firewall.

You will need the Docker image to run in Snowflake container services that I have prepared for the demo.

You can Git clone the Dockerfile to create the Docker image from my Git repository here, or pull a readily prepared image from my Docker Hub to your server or workstation.

You will need to have Docker installed for this.

I assume you are going for the Docker pull route as it is easier and quicker:

docker pull kbkeller/snowflakesshtunnel:latest

Now setup a database, schema, warehouse and image repository in Snowflake in your Snowflake notebook:

(You can find the full SQL workbook here by the way)

use role accountadmin;

CREATE OR REPLACE WAREHOUSE spcs_tunnel_wh WITH WAREHOUSE_SIZE='X-SMALL';

create role spcs_tunnel_role;

grant usage on WAREHOUSE spcs_tunnel_wh  to role spcs_tunnel_role;
grant operate on  WAREHOUSE spcs_tunnel_wh to role spcs_tunnel_role;

grant create database on account to role spcs_tunnel_role;


grant role spcs_tunnel_role to user xxx;

use role spcs_tunnel_role;




create database spcs_tunnel_db;
create schema spcs_tunnel_schema;

--temporarily 
grant all privileges on database spcs_tunnel_db to role accountadmin;
grant all privileges on schema spcs_tunnel_schema to role accountadmin;

CREATE IMAGE REPOSITORY IF NOT EXISTS spcs_tunnel_images;

USE ROLE ACCOUNTADMIN;
alter user xxxx set mins_to_bypass_mfa=180;

SHOW IMAGE REPOSITORIES IN SCHEMA;

Then tag the image according to the naming convention of your Snowflake image repository:

docker tag kbkeller/snowflakesshtunnel xxxx.registry.snowflakecomputing.com/spcs_tunnel_db/spcs_tunnel_schema/spcs_tunnel_images/spcs_tunnel-image:latest

Make sure to disable MFA for a bit before uploading:

alter user xxx set mins_to_bypass_mfa=180;

Now login and push your image to your Snowflake repository:

docker login xxxx.registry.snowflakecomputing.com
docker push xxxx.registry.snowflakecomputing.com/spcs_tunnel_db/spcs_tunnel_schema/spcs_tunnel_images/spcs_tunnel-image:latest

Back in the Snowflake worksheet we can now setup the compute pool and prepare everything to create the service:

Note that you have replace x.x.x.x with the IP address of your DMZ / data lake server.

USE ROLE ACCOUNTADMIN;
CREATE  COMPUTE POOL IF NOT EXISTS SPCS_TUNNEL_COMPUTE_POOL
  MIN_NODES = 1
  MAX_NODES = 1
  INSTANCE_FAMILY = CPU_X64_S
  AUTO_RESUME = true;

  DESCRIBE COMPUTE POOL SPCS_TUNNEL_COMPUTE_POOL;
  drop compute pool SPCS_TUNNEL_COMPUTE_POOL;
  
grant usage on compute pool SPCS_TUNNEL_COMPUTE_POOL to role spcs_tunnel_role;
grant operate on compute pool SPCS_TUNNEL_COMPUTE_POOL to role spcs_tunnel_role;
grant monitor on compute pool SPCS_TUNNEL_COMPUTE_POOL to role spcs_tunnel_role;
grant modify on compute pool SPCS_TUNNEL_COMPUTE_POOL to role spcs_tunnel_role;


GRANT BIND SERVICE ENDPOINT ON ACCOUNT TO ROLE spcs_tunnel_role;


use database spcs_tunnel_db;
use schema spcs_tunnel_schema;
CREATE NETWORK RULE allow_tunnel_rule
  TYPE = 'HOST_PORT'
  MODE= 'EGRESS'
  VALUE_LIST = ('x.x.x.x:22','0.0.0.0:443');

use database spcs_tunnel_db;
use schema spcs_tunnel_schema;
  
CREATE EXTERNAL ACCESS INTEGRATION allow_tunnel_eai
  ALLOWED_NETWORK_RULES=(allow_tunnel_rule)
  ENABLED=TRUE;

GRANT USAGE ON INTEGRATION allow_tunnel_eai TO ROLE spcs_tunnel_role;

use role spcs_tunnel_role;

revoke all privileges on database spcs_tunnel_db from role accountadmin;
revoke all privileges on schema spcs_tunnel_schema from role accountadmin;

  DESCRIBE COMPUTE POOL SPCS_TUNNEL_COMPUTE_POOL;

   drop  COMPUTE POOL SPCS_TUNNEL_COMPUTE_POOL;
  
CREATE or replace SECRET encodedstring
  TYPE=generic_string
  SECRET_STRING='IyEvYmluL2Jhc2ggCi9ob21lL3NwY3N0dW5uZWwvbWFtYmEvYmluL3B5dGhvbiAtbSB2ZW52IGRlZmF1bHRfcHlfZW52Ci4gL2hvbWUvc3Bjc3R1bm5lbC9kZWZhdWx0X3B5X2Vudi9iaW4vYWN0aXZhdGUgJiYgcGlwIGluc3RhbGwgdXYKY2QgL2hvbWUvc3Bjc3R1bm5lbCAmJiBnaXQgY2xvbmUgaHR0cHM6Ly9naXRodWIuY29tL0tlbGxlcktldi9kdWNrYmVyZwpjZCAvaG9tZS9zcGNzdHVubmVsL2R1Y2tiZXJnICYmIC4gL2hvbWUvc3Bjc3R1bm5lbC9kZWZhdWx0X3B5X2Vudi9iaW4vYWN0aXZhdGUgJiYgdXYgcGlwIGluc3RhbGwgLWUgLgoKbWtkaXIgL2hvbWUvc3Bjc3R1bm5lbC9qdXB5dGVyX25vdGVib29rcwpjaG93biBzcGNzdHVubmVsIC9ob21lL3NwY3N0dW5uZWwvanVweXRlcl9ub3RlYm9va3MKY3AgL2hvbWUvc3Bjc3R1bm5lbC9zYW1wbGVzL2ljZWJlcmdfbWluaW8uaXB5bmIgL2hvbWUvc3Bjc3R1bm5lbC9qdXB5dGVyX25vdGVib29rcwoKLiAvaG9tZS9zcGNzdHVubmVsL2RlZmF1bHRfcHlfZW52L2Jpbi9hY3RpdmF0ZSAmJiAgdXYgcGlwIGluc3RhbGwganVweXRlcmxhYgouIC9ob21lL3NwY3N0dW5uZWwvZGVmYXVsdF9weV9lbnYvYmluL2FjdGl2YXRlICYmIHV2IHBpcCBpbnN0YWxsIGh0dHB4PT0wLjI3LjIKLiAvaG9tZS9zcGNzdHVubmVsL2RlZmF1bHRfcHlfZW52L2Jpbi9hY3RpdmF0ZSAmJiB1diBwaXAgaW5zdGFsbCBmYXN0YXBpIHV2aWNvcm4KCiNjZCAvaG9tZS9zcGNzdHVubmVsICYmIGdpdCBjbG9uZSBodHRwczovL2dpdGh1Yi5jb20vanVweXRlcmxhYi9qdXB5dGVybGFiLmdpdAojY2QgL2hvbWUvc3Bjc3R1bm5lbC9qdXB5dGVybGFiICYmIC4gL2hvbWUvc3Bjc3R1bm5lbC9kZWZhdWx0X3B5X2Vudi9iaW4vYWN0aXZhdGUgJiYgdXYgcGlwIGluc3RhbGwgLWUgLiAmCiNjZCAvaG9tZS9zcGNzdHVubmVsL2p1cHl0ZXJsYWIgJiYgL2hvbWUvc3Bjc3R1bm5lbC9kZWZhdWx0X3B5X2Vudi9iaW4vamxwbSBydW4gYnVpbGQ6Y29yZQojY2QgL2hvbWUvc3Bjc3R1bm5lbC9qdXB5dGVybGFiICYmIC9ob21lL3NwY3N0dW5uZWwvZGVmYXVsdF9weV9lbnYvYmluL2p1cHl0ZXIgbGFiIGJ1aWxkCgouIC9ob21lL3NwY3N0dW5uZWwvZGVmYXVsdF9weV9lbnYvYmluL2FjdGl2YXRlICYmIHV2IHBpcCBpbnN0YWxsIGlweWtlcm5lbAoKLiAvaG9tZS9zcGNzdHVubmVsL2RlZmF1bHRfcHlfZW52L2Jpbi9hY3RpdmF0ZSAmJiBqdXB5dGVyIGxhYiAtLWlwICcqJyAtLXBvcnQgMzAwMCAtLW5vdGVib29rLWRpcj0vaG9tZS9zcGNzdHVubmVsL2p1cHl0ZXJfbm90ZWJvb2tzIC0tbm8tYnJvd3NlciAtLUlkZW50aXR5UHJvdmlkZXIudG9rZW49IiIgJgoKCgoKCmVjaG8gJ3NvdXJjZSAvaG9tZS9zcGNzdHVubmVsL2VudgpQT1JUMT0kKHNodWYgLWkgMjAwMC02NTAwMCAtbiAxKQpQT1JUMj0kKHNodWYgLWkgMjAwMC02NTAwMCAtbiAxKQpQT1JUMz0kKHNodWYgLWkgMjAwMC02NTAwMCAtbiAxKQpQT1JUND0kKHNodWYgLWkgMjAwMC02NTAwMCAtbiAxKQpQT1JUNT0kKHNodWYgLWkgMjAwMC02NTAwMCAtbiAxKQplY2hvICRQT1JUMyA+IC9ob21lL3NwY3N0dW5uZWwvcG9ydC50eHQKZWNobyAkUE9SVDMKZWNobyAkUE9SVDUgPiAvaG9tZS9zcGNzdHVubmVsL291dHNpZGVwb3J0LnR4dAplY2hvICRQT1JUNQpERUNPREVES0VZPSIkKHNvdXJjZSAvaG9tZS9zcGNzdHVubmVsL2VudiAmJiBlY2hvICRrZXkxIHwgYmFzZTY0IC0tZGVjb2RlKSIKZWNobyAkREVDT0RFREtFWQplY2hvICRTU0hfVVNFUgojRFlOQU1JQyBQT1JUIEZPUldBUkQgRlJPTSBDT05UQUlORVIgVE8gRE1aCgpzc2gtYWdlbnQgYmFzaCAtYyAic3NoLWFkZCA8KGVjaG8gIFwiJERFQ09ERURLRVlcIikgJiYgYXV0b3NzaCAgIC1NICRQT1JUMSAtbyBTdHJpY3RIb3N0S2V5Q2hlY2tpbmc9bm8gLW8gU2VydmVyQWxpdmVJbnRlcnZhbD02MCAtbyBFeGl0T25Gb3J3YXJkRmFpbHVyZT15ZXMgLWduTlQgIC1EIDEwODAgLUMgLU4gJFNTSF9VU0VSQCRETVpfSVAiICYKCiNPUFRJT05BTExZICBQT1JUIEZPUldBUkQgMzAwMSAoU3RyZWFtbGl0KSB0byBETVoKI3NzaC1hZ2VudCBiYXNoIC1jICJzc2gtYWRkIDwoZWNobyAgXCIkREVDT0RFREtFWVwiKSAmJiAgYXV0b3NzaCAgIC1NICRQT1JUNCAtbyBTdHJpY3RIb3N0S2V5Q2hlY2tpbmc9bm8gLW8gU2VydmVyQWxpdmVJbnRlcnZhbD02MCAtbyBFeGl0T25Gb3J3YXJkRmFpbHVyZT15ZXMgLWduTlQgLUMgLU4gIC1SIGxvY2FsaG9zdDokUE9SVDU6bG9jYWxob3N0OjMwMDIgICAkU1NIX1VTRVJAJERNWl9JUCIgJgoKI09QVElPTkFMTFkgIFBPUlQgRk9SV0FSRCAyMiAoU1NIKSB0byBETVoKI3NzaC1hZ2VudCBiYXNoIC1jICJzc2gtYWRkIDwoZWNobyAgXCIkREVDT0RFREtFWVwiKSAmJiAgYXV0b3NzaCAgIC1NICRQT1JUMiAtbyBTdHJpY3RIb3N0S2V5Q2hlY2tpbmc9bm8gLW8gU2VydmVyQWxpdmVJbnRlcnZhbD02MCAtbyBFeGl0T25Gb3J3YXJkRmFpbHVyZT15ZXMgLWduTlQgLUMgLU4gIC1SIGxvY2FsaG9zdDokUE9SVDM6bG9jYWxob3N0OjIyICAgJFNTSF9VU0VSQCRETVpfSVAiICYnID4gL2hvbWUvc3Bjc3R1bm5lbC90dW5uZWwuc2gKY2htb2QgK3ggL2hvbWUvc3Bjc3R1bm5lbC90dW5uZWwuc2gKL2hvbWUvc3Bjc3R1bm5lbC90dW5uZWwuc2ggJgpzbGVlcCA1Ci4gL2hvbWUvc3Bjc3R1bm5lbC9kZWZhdWx0X3B5X2Vudi9iaW4vYWN0aXZhdGUgJiYgY2QgL2hvbWUvc3Bjc3R1bm5lbC8gJiYgdXZpY29ybiByZXN0YXBpX2ljZWJlcmc6YXBwIC0taG9zdCAwLjAuMC4wIC0tcG9ydCAzMDAzIC0tcmVsb2FkICYKCmNkIC9ob21lL3NwY3N0dW5uZWwgJiYgbWtkaXIgc3RyZWFtbGl0X2FwcHMKY3AgL2hvbWUvc3Bjc3R1bm5lbC9zYW1wbGVzL3N0cmVhbWxpdC9tYWluLnB5IC9ob21lL3NwY3N0dW5uZWwvc3RyZWFtbGl0X2FwcHMKY2QgL2hvbWUvc3Bjc3R1bm5lbC8gJiYgLiAvaG9tZS9zcGNzdHVubmVsL2RlZmF1bHRfcHlfZW52L2Jpbi9hY3RpdmF0ZSAmJiB1diBwaXAgaW5zdGFsbCBzdHJlYW1saXQgbWl0b3NoZWV0IG51bXB5CiNjZCAvaG9tZS9zcGNzdHVubmVsL3N0cmVhbWxpdF9hcHBzICYmIC4gL2hvbWUvc3Bjc3R1bm5lbC9kZWZhdWx0X3B5X2Vudi9iaW4vYWN0aXZhdGUgJiYgc3RyZWFtbGl0IHJ1biBtYWluLnB5IC0tc2VydmVyLnBvcnQgMzAwMiAtLXNlcnZlci5hZGRyZXNzIDAuMC4wLjAgLS1zZXJ2ZXIucnVuT25TYXZlIHRydWUgICBlbmFibGVYc3JmUHJvdGVjdGlvbj1mYWxzZSBlbmFibGVDT1JTPWZhbHNlICYKY2QgL2hvbWUvc3Bjc3R1bm5lbC9zdHJlYW1saXRfYXBwcyAmJiAuIC9ob21lL3NwY3N0dW5uZWwvZGVmYXVsdF9weV9lbnYvYmluL2FjdGl2YXRlICYmIHN0cmVhbWxpdCBydW4gbWFpbi5weSAtLXNlcnZlci5wb3J0IDMwMDIgLS1zZXJ2ZXIuYWRkcmVzcyAwLjAuMC4wIC0tc2VydmVyLnJ1bk9uU2F2ZSB0cnVlIGVuYWJsZUNPUlM9ZmFsc2UgJgouIC9ob21lL3NwY3N0dW5uZWwvZGVmYXVsdF9weV9lbnYvYmluL2FjdGl2YXRlICYmIHV2IHBpcCBpbnN0YWxsIC1lIC9ob21lL3NwY3N0dW5uZWwvc3RyZWFtbGl0X2FwcHMvc3BhcmstMy41LjMtYmluLWhhZG9vcDMvcHl0aG9u';

The “encodedstring” secret is a Base64 encoded shell script that is executed when the container spins up.

This is probably an un-usual way to initilaze the container, but it helps tremendously for testing as I can change the script easily and thereby changing the behaviour of the container.

We will explore this in my next blog post how this can speed up development when you create apps in container services.

For now feel free to go to https://www.base64decode.org/ and decode the Base64 encoded script to see what it does.

It’s mainly installing Jupyterlab, Spark but also creating the SSH tunnel from an environment variable.

This is the part where the magic happens:

ssh-agent bash -c "ssh-add <(echo  \"$DECODEDKEY\") && autossh   -M $PORT1 -o StrictHostKeyChecking=no -o ServerAliveInterval=60 -o ExitOnForwardFailure=yes -gnNT  -D 1080 -C -N $SSH_USER@$DMZ_IP" &

You might have noticed at this stage that we not supplied the private key for the startup script yet.

So let’s continue in the Snowflake worksheet and add those secrets.

The following SQL will create the secrets necessary for the container.

CREATE or replace SECRET servicesecrets
  TYPE=generic_string
SECRET_STRING='{
    "SSH_USER": "xxxx",
    "DMZ_IP": "x.x.x.x",
    "SSHKEYS": [
        {
            "key1": "LS0tLS1CRUdJTiBPU....."
        }
    ]
}';

Replace SSH user with the username of your Linux user of your DMZ / Iceberg data lake server and DMZ_IP with the corresponding IP.

In order to pass the SSH key to the container we need to Base64 encode it .

So take the private SSH key you created for the DMZ / Iceberg data lake server and head over to:

Base64 Encode and Decode - Online

Paste the Base64 encoded key in the “key1” json key and as the value.

Now that you have supplied the user, IP and SSH key in the “servicessecrets” Snowflake secret and we have the startup script ready, we can finally start the container in Snowflake and start playing.

Here is how you start the container now (replace the image path url):

CREATE SERVICE
  spcs_tunnel_service
  IN COMPUTE POOL SPCS_TUNNEL_COMPUTE_POOL
  EXTERNAL_ACCESS_INTEGRATIONS = (allow_tunnel_eai)
  MIN_INSTANCES=1
  MAX_INSTANCES=1
  FROM SPECIFICATION
  $$
    spec:
        containers:
         - name: "spcstunnelservice"
           image: "xxx.xxxx.registry.snowflakecomputing.com/spcs_tunnel_db/spcs_tunnel_schema/spcs_tunnel_images/spcs_tunnel-image:latest"
           secrets:
            - snowflakeSecret: encodedstring
              envVarName: ENCODEDRUNSCRIPT
            - snowflakeSecret: servicesecrets
              envVarName: SERVICESECRETS
           env:
               SSHKEY: "False"
               ENVNAME: "snowtunnel" 
               VSVERSION: "1.95.2"
                   
        endpoints:
        - name: jupyter
          port: 3000
          public: true
        - name: vscode
          port: 3001
          public: true
        - name: streamlit
          port: 3002
          public: true
        - name: restapi
          port: 3003
  $$;

This will spin up the service, install everything and build the tunnel to your DMZ.

You can follow the process with the following SQLs:

describe service spcs_tunnel_service;


--alter service spcs_tunnel_service suspend;
--alter service spcs_tunnel_service resume;

--drop service spcs_tunnel_service;

SHOW SERVICE CONTAINERS IN SERVICE spcs_tunnel_service;



CALL SYSTEM$GET_SERVICE_LOGS('spcs_tunnel_service', '0', 'spcstunnelservice', 1000);
SHOW ENDPOINTS IN SERVICE spcs_tunnel_service;

Most importantly the “SHOW ENDPOINTS” command which will show you the endpoints for the Jupyter notebook, Streamlit app and VSCode web instance.

You should now be able to easily access the Jupyter notebooks to query the data via the tunnel with DuckDB or Apache Spark:

NOTE: You will have to change the private IP address to match the one of your DMZ / Iceberg Data Lake server!

If you want to query the data through the tunnel via Snowflake UDF, here is how you can setup the UDF:

CREATE OR REPLACE FUNCTION query_taxis_udf(sql_query STRING)
returns variant
SERVICE=spcs_tunnel_service
ENDPOINT=restapi
AS '/snowflake_function';

Now you can test it:

SELECT query_taxis_udf('SELECT * FROM "nyc.taxis" LIMIT 10');

If if the UDF does not work you have to change the private IP address to match the one of your DMZ / Iceberg Data Lake server in the REST API code. You can use the VSCode provided in the container to quickly adapt the REST API code or update the container image.

Update these IPs:

If all is working let’s transform the UDF scalar results, which look a bit hard to read into a flattend table:

WITH udf_output AS (
    SELECT
        query_taxis_udf('SELECT * FROM "nyc.taxis" LIMIT 100') AS json_result
),
flattened AS (
    SELECT
        VALUE:DOLocationID::INTEGER AS DOLocationID,
        VALUE:PULocationID::INTEGER AS PULocationID,
        VALUE:RatecodeID::FLOAT AS RatecodeID,
        VALUE:VendorID::INTEGER AS VendorID,
        VALUE:airport_fee::FLOAT AS airport_fee,
        VALUE:congestion_surcharge::FLOAT AS congestion_surcharge,
        VALUE:extra::FLOAT AS extra,
        VALUE:fare_amount::FLOAT AS fare_amount,
        VALUE:improvement_surcharge::FLOAT AS improvement_surcharge,
        VALUE:mta_tax::FLOAT AS mta_tax,
        VALUE:passenger_count::FLOAT AS passenger_count,
        VALUE:payment_type::INTEGER AS payment_type,
        VALUE:store_and_fwd_flag::STRING AS store_and_fwd_flag,
        VALUE:tip_amount::FLOAT AS tip_amount,
        VALUE:tolls_amount::FLOAT AS tolls_amount,
        VALUE:total_amount::FLOAT AS total_amount,
        VALUE:tpep_dropoff_datetime::TIMESTAMP AS tpep_dropoff_datetime,
        VALUE:tpep_pickup_datetime::TIMESTAMP AS tpep_pickup_datetime,
        VALUE:trip_distance::FLOAT AS trip_distance
    FROM udf_output, LATERAL FLATTEN(INPUT => json_result)
)
SELECT * FROM flattened;

You can further simplify this by creating a view (that should be easy) or a stored procedure such as this:

CREATE OR REPLACE PROCEDURE flatten_query_taxis_sp(
    p_col_list VARCHAR,
    p_table_name VARCHAR,
    p_row_limit NUMBER
)
RETURNS TABLE (
    DOLocationID NUMBER,
    PULocationID NUMBER,
    RatecodeID NUMBER,
    VendorID NUMBER,
    airport_fee NUMBER,
    congestion_surcharge NUMBER,
    extra NUMBER,
    fare_amount NUMBER,
    improvement_surcharge NUMBER,
    mta_tax NUMBER,
    passenger_count NUMBER,
    payment_type NUMBER,
    store_and_fwd_flag VARCHAR,
    tip_amount NUMBER,
    tolls_amount NUMBER,
    total_amount NUMBER,
    tpep_dropoff_datetime TIMESTAMP_NTZ,
    tpep_pickup_datetime TIMESTAMP_NTZ,
    trip_distance NUMBER
)
LANGUAGE SQL
AS
$$
DECLARE
    result_sql STRING;
    result_set RESULTSET;
BEGIN
    -- Construct the SQL query
    result_sql :=
        'WITH udf_output AS (
            SELECT query_taxis_udf(''SELECT ' || p_col_list || ' FROM "' || p_table_name || '" LIMIT ' || p_row_limit || ''') AS json_result
        ),
        flattened AS (
            SELECT
                VALUE:DOLocationID::NUMBER,
                VALUE:PULocationID::NUMBER,
                VALUE:RatecodeID::NUMBER,
                VALUE:VendorID::NUMBER,
                VALUE:airport_fee::NUMBER,
                VALUE:congestion_surcharge::NUMBER,
                VALUE:extra::NUMBER,
                VALUE:fare_amount::NUMBER,
                VALUE:improvement_surcharge::NUMBER,
                VALUE:mta_tax::NUMBER,
                VALUE:passenger_count::NUMBER,
                VALUE:payment_type::NUMBER,
                VALUE:store_and_fwd_flag::VARCHAR,
                VALUE:tip_amount::NUMBER,
                VALUE:tolls_amount::NUMBER,
                VALUE:total_amount::NUMBER,
                VALUE:tpep_dropoff_datetime::TIMESTAMP_NTZ,
                VALUE:tpep_pickup_datetime::TIMESTAMP_NTZ,
                VALUE:trip_distance::NUMBER
            FROM udf_output, LATERAL FLATTEN(INPUT => json_result)
        )
        SELECT * FROM flattened';

    -- Execute the dynamic SQL query and store the result in a RESULTSET variable
    result_set := (EXECUTE IMMEDIATE :result_sql);

    -- Return the RESULTSET
    RETURN TABLE(result_set);
END;
$$;

And now you can execute the flattening query much simpler:

SELECT *
FROM TABLE(
    flatten_query_taxis_sp(
        'DOLocationID, PULocationID, fare_amount, trip_distance',
        'nyc.taxis',
        10
    )
);

The null values are hard to remove as the schema has to be defined up front for flattening.

Once Snowflake container services UDFs support table return types this in-convinience should be solved by itself.

Finally you can also check out the Streamlit app that you can use to query your on-premise data and make it available via Streamlit.

Head over to the Streamlit URL endpoint from the container services and check it out:

If it does not work, you can change the Streamlit app codeon the fly using the provided VSCode instance to edit code.

You probably just need to adapt the private IP address to your DMZ / Private Iceberg Data Lake server.

Paste the VSCode endpoint URL in your browser to access it and then open the Streamlit app:

Open the folder “streamlit_apps” in the container:

And check out and play with the code and adapt the IP addreses if necessary as we did with the UDF REST API previously.

You can also open a terminal and see what else is in the container.

There are some more Spark command line sample scripts and Jupyter notebooks.

Conclusion and further steps:

I have picked an Iceberg data lake as an example, but nobody can stop you from using this to access other resources like on-premise databases, APIs, AI models.. you name it..

I can even see that it should be possible to direclty conect to a z/OS mainframe 3270 terminal and execute automated commands using this library: https://github.com/IBM/tnz?tab=readme-ov-file

Although z/OS and other usual on-premise resources have plenty of options to create REST or other APIs on top of them these days.

In part 2 of this post, which I will link to from here asap, we will explore how we can reverse the process and expose services running in Snowflake container services to apps running outside of Snowflake directly.

That opens the door to do things like SFTP, FTP uploads to Snowflake or access an S3 Minio API directly in Snowflake.

If you reverse the process you can also expose Streamlit apps and other web apps to non-Snowflake users.

Ingest Opentelemetry Logs, Traces and Metrics Directly into Snowflake as an APM and SIEM Replacement or Augumentation

Let me show you one example how you can ingest traces, logs and metrics natively into Snowflake using the Opentelemetry receiver I put together that is able to receive logs, metrics and traces using the OTLP/gRPC protocol via HTTP(S).

You can find all my code here to get this to work in your own Snowflake account.

But let’s look at the big picture what we can do here.

If you consider that you can have Snowflake write into Apache Iceberg tables and you can query that data from any Iceberg compatible query engine you can create a completely open observability and security data lake without worrying about storage, storage costs or scalability.

The benefits here are that you leverage the Opentelemetry ecosystem of connectors and plug those natively into Snowflake in order to receive logs, traces and metrics so that you can build your own observability and security data lake if you also use Snowflakes ability to create and manage Iceberg tables in your own object storage.

You can also have Snowflake natively notify you or trigger automated actions for automated incident and response management.

I showed how you can do that in one of my other blog posts here.

If you check out the Github repo, you also see that there is a table that contains only the last 10 mins of logs. This is great for time senstive incident raising.

With Snowflake notebooks you can also go ahead and do some threat hunting or discover any bottle necks in your application code or archictecture.

Need some real time dashboards in Snowflake?

I have added an example Streamlit in Snowflake app (SiS) to the Github Repository here:

You can also run PromQL queries (experimental)

There is also a sample for Splunk SPL queries:

search metric_name="http_requests_total" | table @timestamp, metric.value

Need further information from other systems and partners?

Easy! Try the Snowflake marketplace:

You can also ingest information from Workday (or other HR systems) or Entra AD (or Okta or other IDPs) to enrich the context, here is a sample for Workday:

How to Load Workday Data into Snowflake Using External Network Access

Bottom line:

You can centralise SIEM & APM, logs,traces, metrics, exceptions & alerts in Snowflake.

The Opentelemetery eco-system is vast and standarized, allowing you to fetch / ingest , batch, transform and export (forward) data from many sources with ease.

You can also point Jeager and Prometheus sources to an Opentelemetry Collector.

The sweet spot of this setup is probably the augementation play in which you keep using your current SIEM and APM solution for the most recent data and augement your setup by moving data to Snowflake instead of putting it into a long term storage, so all data is always ready to be queried and does not need to be hydrated first.

You can point your current SIEM and APM solution to the Opentelemetry Collector which will deliver the data to Snowflake.

Enough talk.. let’s setup a demo.

For that we will use a Flask sample application that sends logs, traces and metrics to an Opentelemtry collector which forwards it to a container in Snowflake that receives it and writes this data into the respective tables.

We will use Iceberg tables for the final target tables as last step of this tutorial.

The final target tables will also be transformed to adhere to the Elastic Common Schema format in order to standarize all logs, metrics and traces from different sources in a standard schema. That makes it easier to work with the data and scale.

Here is the data flow of the demo:

So lets go ahead and get this running.

We will do it in 2 steps.

Step 1: Let’s run all components locally first with Docker Compose.

Step 2: We will run the Snowflake OTEL Receiver in Snowflake Container Services while only the OpenTelemetry Collector and the Flask app will stay running outside of Snowflake.

In a next blog post I will also show how you can have all containers running in Snowflake container services.

So for Step 1:

1a) Clone this Git Repo: https://github.com/KellerKev/snowflake-opentelemetry-receiver

1b) Copy and paste the contents of otel_spcs_prep.sql into your Snowflake SQL workbook.

1c) Work through the workbook until the comment “STOP Here to prepare for Step 1)”.

Make sure you replace the user with your username for this line:

grant role oteltest to user xxx;

Here are the full statements we should do for now as well if you want to quickly copy and paste from here:

use role accountadmin;

CREATE OR REPLACE WAREHOUSE otelwh WITH WAREHOUSE_SIZE='X-SMALL';

create role oteltest;

grant usage on WAREHOUSE otelwh  to role oteltest;
grant operate on  WAREHOUSE otelwh to role oteltest;

grant create database on account to role oteltest;


grant role oteltest to user xxx;

use role oteltest;



create database otel;
create schema otelschema;

CREATE or replace TABLE metrics (
      timestamp TIMESTAMP_NTZ,
      metric_name STRING,
      value DOUBLE,
      attributes VARCHAR
);

CREATE or replace TABLE logs (
      timestamp TIMESTAMP_NTZ,
      log_level STRING,
      message STRING,
      attributes varchar
);


CREATE or replace TABLE  traces (
      trace_id STRING,
      span_id STRING,
      name STRING,
      start_time TIMESTAMP_NTZ,
      end_time TIMESTAMP_NTZ,
      attributes varchar
);

select * from metrics;
select * from logs;
select * from traces;

1d) Now that you have the warehouse, the role, the database, schema and tables setup in Snowflake you can spin up the dockercompose.Yaml with Docker Compose which will run the Flask sample app, the Opentelemtry Collector as will as the Snowflake Opentelemetry Receiver all locally, but connect to your Snowflake account via the Python driver.

In order to do that we need to supply the docker-compose.yaml files with your credential first:

Make sure to put your right credentials here for account, user, password.

The rest should work as is.

If you run Docker Compose on Windows make sure remove the Linux/Mac path line and adapt the one that is prepared for Windows and point it to the OTEL config file that is required that you got from the Git Repository.

1e) Now in the Terminal in the folder of the cloned Git Repository on your PC/Mac or Linux workstation/server enter the command:

docker compose up

In the following demo video is what you should be able to see now.

The 3 containers start up, with the Opentelemetry Snowflake Receiver connecting to your Snowflake instance.

Navigate with your browser to the Flask endpoints of the Flask sample app that sends traces, logs and metrics:

http://localhost:5000/greet/kevin

You can change the name “kevin” to your name or the name of your pet or anything you like.

And also go to:

http://localhost:5000/hello

In your Snowflake account you should now be able to see that the tables logs, traces and metrics have been populated with data.

We are now ready for Step 2) to let the Snowflake Opentelemetry Receiver run in Snowflake container services.

For that we need to upload the Snowflake Opentelemetry Receiver Docker image into your Snowflake accounts image registry.

In order to do that continue with the steps in the otel_spcs_prep.sql file

2a) Let me put the relavent steps for right now here as well if you quickly want to copy and paste from here:

CREATE IMAGE REPOSITORY IF NOT EXISTS oteltestimages;

SHOW IMAGE REPOSITORIES IN SCHEMA;

--before uploading image, if you use a username/password to authenticate and MFA is enabled for your user, you better disable it
-- before uploading for a bit so you dont get locked out and receive 10000 MFA messages:

With this we have an an image repository.

2b) When you execute the command:

SHOW IMAGE REPOSITORIES IN SCHEMA;

You should be able to see how you need to tag (rename) the Docker image so we can upload it.

It should look something like this:

xxxx.registry.snowflakecomputing.com/otel/otelschema/oteltestimages

2c) Take the value from the repository_url column and copy it.

Now in your terminal on your local workstatation or server make sure you have the image of the Snowflake Opentelemetry Receiver available.

You should have it already if you have followed Step 1).

If not you can either build the image from the Dockerfile provided in the Github repository or pull it from my Dockerhub repository via:

docker pull kbkeller/snowflakeotelreceiver:latest

2d) Now lets give this image a tag suitable to upload into the Snowflake image registry:

docker tag kbkeller/snowflakeotelreceiver:latest xxxx.registry.snowflakecomputing.com/otel/otelschema/oteltestimages/otel-image:latest

Afterwards we can authenticate and upload.

Before you do the Docker push a word of warning:

Do NOT push yet, read this:

If you have MFA for the user enabled and you are using a password to auth then disable MFA RIGHT NOW. Just for a couple of minutes.

Otherwise you will get a ton of MFA prompts on your phone and you will be locked out of your Snowflake account for 15 minutes and that is all you will achieve.

So to avoid this, go back to your Snowflake account and disable MFA for 10 minutes or so for your user:

USE ROLE ACCOUNTADMIN;
alter user xxxx set mins_to_bypass_mfa=10

2e) Ok back in our terminal we can login and push the image now like so

(first login, then push)

docker login xxxx.registry.snowflakecomputing.com

docker push xxxx.registry.snowflakecomputing.com/otel/otelschema/oteltestimages/otel-image:latest

You will be asked to login with username and password and then you should be able to upload the image to the Snowflake image registry.

2f) Once the upload is done, let’s create a service / container that we run based on on the image in your Snowflake account.

Here are the following useful commands to execute now.

We will create a compute pool, give access to the compute pool to our custom role and then run the container in that compute pool.

We can also check the logs and see the status of everything:

USE ROLE ACCOUNTADMIN;

GRANT BIND SERVICE ENDPOINT ON ACCOUNT TO ROLE oteltest;

CREATE  COMPUTE POOL IF NOT EXISTS OTEL_COMPUTE_POOL
  MIN_NODES = 1
  MAX_NODES = 1
  INSTANCE_FAMILY = CPU_X64_S
  AUTO_RESUME = true;

  DESCRIBE COMPUTE POOL OTEL_COMPUTE_POOL;
  
grant usage on compute pool OTEL_COMPUTE_POOL to role oteltest;
grant operate on compute pool OTEL_COMPUTE_POOL to role oteltest;
grant monitor on compute pool OTEL_COMPUTE_POOL to role oteltest;
grant modify on compute pool OTEL_COMPUTE_POOL to role oteltest;


use role oteltest;


 CREATE SERVICE otel_service
  IN COMPUTE POOL OTEL_COMPUTE_POOL
  MIN_INSTANCES=1
  MAX_INSTANCES=1
  FROM SPECIFICATION
  $$
   spec:
     containers:
     - name: "otel"
       image: "/otel/otelschema/oteltestimages/otel-image:latest"
       env:
         SNOWFLAKE_DATABASE: "otel"
         SNOWFLAKE_WAREHOUSE: "otelwh"
         SNOWFLAKE_SCHEMA: "otelschema"
         SPCS: "True"
          
     endpoints:
     - name: otelhttp
       port: 4318
       public: true
  $$;

  
describe service otel_service;
SHOW SERVICE CONTAINERS IN SERVICE otel_service;
SHOW IMAGES IN IMAGE REPOSITORY oteltestimages;


CALL SYSTEM$GET_SERVICE_LOGS('otel_service', '0', 'otel', 1000);
SHOW ENDPOINTS IN SERVICE otel_service;

2g) After the upload you should the see the image in the repository in Snowflake with the “SHOW IMAGES..” SQL command:

2h) So the container should also run now, based on you having executed the SQL commands above in the right order. The relevant command is:

CREATE SERVICE otel_service
  IN COMPUTE POOL OTEL_COMPUTE_POOL
  MIN_INSTANCES=1
  MAX_INSTANCES=1
  FROM SPECIFICATION
  $$
   spec:
     containers:
     - name: "otel"
       image: "/otel/otelschema/oteltestimages/otel-image:latest"
       env:
         SNOWFLAKE_DATABASE: "otel"
         SNOWFLAKE_WAREHOUSE: "otelwh"
         SNOWFLAKE_SCHEMA: "otelschema"
         SPCS: "True"
          
     endpoints:
     - name: otelhttp
       port: 4318
       public: true
  $$;

2i)

You may have to be patient as compute pools need some time to start up:

Consquently the container may also still be in pending status if the pool is not started yet:

You can check the status with:

describe service otel_service;
DESCRIBE COMPUTE POOL OTEL_COMPUTE_POOL;

2j) Once everything is in running and active state you can check the logs of the Snowflake Opentelemetry receiver:

CALL SYSTEM$GET_SERVICE_LOGS('otel_service', '0', 'otel', 1000);

2k) In order to send the logs to the receiver now, we need 2 things:

One is the endpoint url of the receiver and the other is the OAuth token in order to be able to authenticate programmatically to the running container in Snowflake.

Getting the URL is easy:

SHOW ENDPOINTS IN SERVICE otel_service;

That should show you something like this:

agtubd-xxxx-up50179.snowflakecomputing.app

Which means the full url that we need for the Opentelemetry Collector running on our server or workstation in the last step is something like;

https://agtubd-xxxx-up50179.snowflakecomputing.app

2l) So lets prepare the docker-compose.yaml file and the Opentelemetry Collector config to communicate with the receiver in Snowflake.

Go ahead and open the docker-compose.yaml file in your favourite editor and lets remove the the entry to start the receiver locally:

Just comment out the lines that would start it with the hash in front of each line:

Next we need to point the Opentelemetry collector to the Snowflake container URL:

In the collector-config.yaml replace the entry point to http://otel:4318 with the endpoint you have extracted from Snowflake in the previous step. Also eanble TLS.

So instead of this:

We should now have this:

Where tls: insecure is FALSE and the endpoint is the HTTPS endpoint of the Snowflake container.

The last step in this file is to uncommend the headers section and add the right Snowflake OAuth token.

For that you need to generate a RSA keypair for the Snowflake user, assign the the public key to the user in Snowflake and then use the Python script that is part of the repository that is called “access-via-keypair.py”.

The detailed instruction can be found here.

Make SURE to write the role of your user in UPPERCASE letters.

Once you have the token add it to the Opentelemetry Collector config like so:

Replace the XXXX with your token, or better yet, put in a environment variable that can be updated. For the first test let’s just put in the token.

2m) Finally you should be able to run Docker compose again with

docker compose up

The Opentelemetry Collector is now collecting the logs, traces and metrics from the Flask sample app (remember to visit the localhost:5000/hello etc. to generate data) and is sending it straight to the Snowflake Opentelemetry Receiver running in Snowflake.

2n) Let’s transform the tables in the Elastic Common Schema into target tables via a Stored Procedure and Streams and Tasks.

-- Create Streams on the source tables
CREATE OR REPLACE STREAM logs_stream ON TABLE logs APPEND_ONLY = TRUE;
CREATE OR REPLACE STREAM metrics_stream ON TABLE metrics APPEND_ONLY = TRUE;
CREATE OR REPLACE STREAM traces_stream ON TABLE traces APPEND_ONLY = TRUE;

Create the stored procedure to transform the tables steams:

CREATE OR REPLACE PROCEDURE ecs_transform_incremental()
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = '3.11'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'main'
AS
$$
def main(session):
    from snowflake.snowpark.functions import col, parse_json, datediff, lit

    try:
        # Create the target schema if it doesn't exist
        session.sql("CREATE SCHEMA IF NOT EXISTS ecs_schema").collect()

        # ***********************
        # Process Logs
        # ***********************

        # Create the target 'logs' table if it doesn't exist
        session.sql("""
            CREATE TABLE IF NOT EXISTS ecs_schema.logs (
                "@timestamp" TIMESTAMP_NTZ,
                "message" STRING,
                "log.level" STRING,
                "attributes" VARIANT
            )
        """).collect()

        # Read new data from the logs_stream
        logs_df = session.table("logs_stream")

        # Check if there is new data to process
        if logs_df.count() > 0:
            # Transform and insert data into ecs_schema.logs
            logs_transformed = logs_df.select(
                col("timestamp").alias("@timestamp"),
                col("message"),
                col("log_level").alias("log.level"),
                parse_json(col("attributes")).alias("attributes")
            )

            logs_transformed.write.mode("append").save_as_table("ecs_schema.logs")
        else:
            print("No new logs data to process.")

        # ***********************
        # Process Metrics
        # ***********************

        # Create the target 'metrics' table if it doesn't exist
        session.sql("""
            CREATE TABLE IF NOT EXISTS ecs_schema.metrics (
                "@timestamp" TIMESTAMP_NTZ,
                "metricset.name" STRING,
                "metric.value" FLOAT,
                "attributes" VARIANT
            )
        """).collect()

        # Read new data from the metrics_stream
        metrics_df = session.table("metrics_stream")

        # Check if there is new data to process
        if metrics_df.count() > 0:
            # Transform and insert data into ecs_schema.metrics
            metrics_transformed = metrics_df.select(
                col("timestamp").alias("@timestamp"),
                col("metric_name").alias("metricset.name"),
                col("value").alias("metric.value"),
                parse_json(col("attributes")).alias("attributes")
            )

            metrics_transformed.write.mode("append").save_as_table("ecs_schema.metrics")
        else:
            print("No new metrics data to process.")

        # ***********************
        # Process Traces
        # ***********************

        # Create the target 'traces' table if it doesn't exist
        session.sql("""
            CREATE TABLE IF NOT EXISTS ecs_schema.traces (
                "trace.id" STRING,
                "span.id" STRING,
                "span.name" STRING,
                "span.start" TIMESTAMP_NTZ,
                "span.end" TIMESTAMP_NTZ,
                "span.duration" FLOAT,
                "attributes" VARIANT
            )
        """).collect()

        # Read new data from the traces_stream
        traces_df = session.table("traces_stream")

        # Check if there is new data to process
        if traces_df.count() > 0:
            # Calculate span duration
            traces_transformed = traces_df.with_column(
                "span.duration",
                datediff("milliseconds", col("start_time"), col("end_time"))
            ).select(
                col("trace_id").alias("trace.id"),
                col("span_id").alias("span.id"),
                col("name").alias("span.name"),
                col("start_time").alias("span.start"),
                col("end_time").alias("span.end"),
                col("span.duration"),
                parse_json(col("attributes")).alias("attributes")
            )

            traces_transformed.write.mode("append").save_as_table("ecs_schema.traces")
        else:
            print("No new traces data to process.")

        return 'Incremental transformation completed successfully.'
    except Exception as e:
        return 'Error: ' + str(e)
$$;

Lets create the task:

CREATE OR REPLACE TASK ecs_transform_task
WAREHOUSE = your_warehouse_name  -- Replace with your warehouse name
SCHEDULE = '1 MINUTE'
AS
CALL ecs_transform_incremental();

And enable it:

-- Enable the Task

use role accountadmin;
grant EXECUTE TASK on account to role oteltest;
use role oteltest;
ALTER TASK ecs_transform_task RESUME;

You can monitor tasks as ACCOUNTADMIN like this:

show tasks;
show alerts;

SELECT * from LOGIN_EVENTS;

-- monitor alerts & tasks

SELECT *
FROM
  TABLE(INFORMATION_SCHEMA.ALERT_HISTORY(
    SCHEDULED_TIME_RANGE_START
      =>dateadd('hour',-1,current_timestamp())))
ORDER BY SCHEDULED_TIME DESC;

SELECT *
FROM
  TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
    SCHEDULED_TIME_RANGE_START
      =>dateadd('hour',-1,current_timestamp())))
ORDER BY SCHEDULED_TIME DESC;

Now that the task is running you sould see a new schema in the database called “ECS_SCHEMA”:

Where you can now find the logs, metrics and traces table schemas modfied to comply to ECS:

These target tables could be modified to be Iceberg tables as promised so that you can get started with having all the data in an open security and observability data lake.

Just create the table as type Iceberg with Snowflake providing the catalog and setup the external stage (s3, blob or gcp).

Here is how you can do it quickly, if you have an external stage ready:

Stop the task for a bit:

use role oteltest;
ALTER TASK ecs_transform_task suspend;

Then navigate to the ECS_SCHEMA and lets drop the tables we have there for now:

drop otel.ecs_schema.logs;
drop otel.ecs_schema.metrics;
drop otel.ecs_schema.traces;

Now in the schema, lets re-create those tables as Iceberg tables (for my_ext_vol and ‘my/relative/path/from/extvol’ use the values appropiate for your setup)

create or replace ICEBERG TABLE OTEL.ECS_SCHEMA.LOGS (
 "@timestamp" TIMESTAMP_NTZ(9),
 "message" VARCHAR(16777216),
 "log.level" VARCHAR(16777216),
 "attributes" VARIANT
)
CATALOG = 'SNOWFLAKE'
  EXTERNAL_VOLUME = 'my_ext_vol'
  BASE_LOCATION = 'my/relative/path/from/extvol';

create or replace ICEBERG TABLE OTEL.ECS_SCHEMA.METRICS (
 "@timestamp" TIMESTAMP_NTZ(9),
 "metricset.name" VARCHAR(16777216),
 "metric.value" FLOAT,
 "attributes" VARIANT
)
CATALOG = 'SNOWFLAKE'
  EXTERNAL_VOLUME = 'my_ext_vol'
  BASE_LOCATION = 'my/relative/path/from/extvol';

create or replace ICEBERG TABLE OTEL.ECS_SCHEMA.TRACES (
 "trace.id" VARCHAR(16777216),
 "span.id" VARCHAR(16777216),
 "span.name" VARCHAR(16777216),
 "span.start" TIMESTAMP_NTZ(9),
 "span.end" TIMESTAMP_NTZ(9),
 "span.duration" FLOAT,
 "attributes" VARIANT
)
CATALOG = 'SNOWFLAKE'
  EXTERNAL_VOLUME = 'my_ext_vol'
  BASE_LOCATION = 'my/relative/path/from/extvol';

ALTER TASK ecs_transform_task resume;

Next steps:

In the next posts we will move all container components of this demo setup into Snowflake and also how we can use Snowflake notebooks for data crunching and threat hunting.

We will also introduce batching on the Opentelemetry Collector in order to send data more efficiently to Snowflake.

We will also add the ingestion of VECTOR and Fluentd data in a similiar manner.

Next up →

The benefit is that you can react to security related incidents in an automated and fast way without needing to export any logs from Snowflake or run, host and maintain any additional code outside of Snowflake.

Here is just a quick guide that shall serve as a sample how you can setup Snowflake to send an API request to your SIEM system’s API such as Splunk, Sentinel etc. in order to raise an incident in a timely manner in an automated way.

While it is possible to have Snowflake raise incidents on any available data apart from Snowflakes audit logs (meaning you can also raise alerts in audit logs ingested into Snowflake from 3rd party applications), this example focuses on raising an incident when a user has tried to login into Snowflake using the wrong user password.

This use case is particularily interesting for safeguarding service accounts that are forced to use a password and where OAuth or Keypair authentication is not possible.

A single failed login attempt for these users should already raise an alarm.

This is especially interesting for those service users that connect from an ever changing public IP address space to Snowflake such as PowerBI Cloud Service, where setting Network Policies in Snowflake can become challenging if you are not willing to use the PowerBI Gateway for those fully automated report creation use cases.

The process works by running a Snowflake Alert Task every minute to check for any failed login attempts using password authenticaton.

If there are any new failed password authentication attempts then the Snowflake Alert Task will create a record of that in the LOGIN_EVENTS table.

The Alert Task will check within a time range of the last 5 minutes in the Snowflake authentication logs from the point in time when the task is running.

So there is a chance that the same event will be detected multiple times if you run the Alert Task every minute, but that doesn’t matter.

The reason it does not matter is that we will only record any events that we have not logged yet previously into the LOGIN_EVENTS table given that each failed login attempt in Snowflake has a unique event ID.

Once the event is recorded another Snowflake Task, which also runs every minute will check, if there are any new failed login attempt records in the LOGIN_EVENTS table.

If so we take the information of the record(s) and send it/them via External Access function UDF (just an API call) to your SIEM or XDR system in order to raise an incident.

Finally the Task will also update the column as being “SENT=True” with a timestamp and also create a record of the SIEM’s API response.

Next time the Task will check the records, it will see that all records have already been sent and that no further action is required.

Meanwhile if an incident was raised in the SIEM by Snowflake, the SIEM system can trigger the incident response process to react to the incident as you see fit.

You can have your SIEM system connect to Snowflake and disable the user and load additional log records into the SIEM relevant for this incident.

So how can we implement this?

Now is the time to login to your Snowflake account and create a new SQL worksheet.

You will need the ACCOUNTADMIN role temporarily.

Let’s create a Snowflake Alert Task to run every minute and check if there were any failed password login attempts in the last 5 minutes and if so, record that event into the LOGIN_EVENTS table.

In order for our user to be able to do so, you will need the MONITOR privilege on every user (or just the users you want to monitor).

You can follow this guide along with the highly privileged role ACCOUNTADMIN to get through it quickly all the way, but I would highly recommend to create a bespoke role right now INSTEAD of using a highly privileged user permanently.

1. So best you switch to ACCOUNTADMIN temporarily and do the following (replace XXX with YOUR username):

use role accountadmin;
CREATE OR REPLACE ROLE SIEM_MONITOR;
grant role siem_monitor to user XXXX;
GRANT CREATE WAREHOUSE ON ACCOUNT TO ROLE siem_monitor;
GRANT MANAGE WAREHOUSES ON ACCOUNT TO ROLE siem_monitor;
GRANT CREATE DATABASE ON ACCOUNT TO ROLE siem_monitor;
GRANT CREATE ROLE ON ACCOUNT TO siem_monitor;
GRANT EXECUTE TASK ON ACCOUNT TO ROLE siem_monitor;
GRANT EXECUTE ALERT ON ACCOUNT TO ROLE siem_monitor;
GRANT CREATE INTEGRATION ON ACCOUNT TO ROLE  siem_monitor;


use role siem_monitor;

CREATE OR REPLACE WAREHOUSE siem_alert_demo WITH WAREHOUSE_SIZE='X-SMALL';
grant usage on warehouse siem_alert_demo to role SIEM_MONITOR;

CREATE OR REPLACE DATABASE SIEM_ALERT_DEMO_DB;
CREATE OR REPLACE SCHEMA SIEM_ALERT_DEMO_SCHEMA;

grant usage on database SIEM_ALERT_DEMO_DB to role accountadmin;
grant usage on schema SIEM_ALERT_DEMO_DB.SIEM_ALERT_DEMO_SCHEMA to role accountadmin;
grant usage on warehouse siem_alert_demo to role accountadmin;

CREATE OR REPLACE ROLE MONITOR_USERS;
grant role monitor_users to role siem_monitor;

CREATE OR REPLACE PROCEDURE set_monitor_privileges_on_all_users(users varchar)
RETURNS varchar
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'run'
EXECUTE AS CALLER
AS
$$
from snowflake.snowpark import Session

def run(session, users):
      users=users.replace('[','')
      users=users.replace(']','')
      users=users.split(',')
      for user in users:
              sql = """GRANT MONITOR ON USER """+user+""" TO ROLE monitor_users"""             
              result=session.sql(sql).collect()[0][0]

      return "done"
  
  
$$;

grant usage on procedure set_monitor_privileges_on_all_users(varchar) to role accountadmin;
use role accountadmin;
GRANT CREATE NETWORK RULE ON SCHEMA SIEM_ALERT_DEMO_DB.SIEM_ALERT_DEMO_SCHEMA TO ROLE siem_monitor;
call set_monitor_privileges_on_all_users(select all_user_names());

use role siem_monitor;
revoke usage on procedure set_monitor_privileges_on_all_users(varchar) from role accountadmin;
revoke usage on warehouse siem_alert_demo from role accountadmin;
revoke usage on database SIEM_ALERT_DEMO_DB from role accountadmin;
revoke usage on schema SIEM_ALERT_DEMO_DB.SIEM_ALERT_DEMO_SCHEMA from role accountadmin;

We assign the role MONITOR_USERS which now has the privileges to monitor all users to the regular user role SIEM_MONITOR that you are currently using to create this demo with.

2. Now we continue so let’s setup our login event table , a Snowflake Alert Task and a View that makes it easier to query login events:

use role siem_monitor;

Lets the create the table that will hold the information of all the suspicious / falied logins of users that have entered the wrong password.
create or replace TABLE LOGIN_EVENTS (
 EVENT_TIMESTAMP TIMESTAMP_LTZ(3),
 EVENT_ID NUMBER(38,0),
 EVENT_TYPE VARCHAR(16777216),
 USER_NAME VARCHAR(16777216),
 CLIENT_IP VARCHAR(16777216),
 REPORTED_CLIENT_TYPE VARCHAR(16777216),
 REPORTED_CLIENT_VERSION VARCHAR(16777216),
 FIRST_AUTHENTICATION_FACTOR VARCHAR(16777216),
 SECOND_AUTHENTICATION_FACTOR VARCHAR(16777216),
 IS_SUCCESS VARCHAR(3),
 ERROR_CODE NUMBER(38,0),
 ERROR_MESSAGE VARCHAR(16777216),
 RELATED_EVENT_ID NUMBER(38,0),
 CONNECTION VARCHAR(16777216),
 SENT BOOLEAN DEFAULT FALSE,
 SENT_TIME TIMESTAMP_NTZ(9),
 SIEM_RESPONSE VARCHAR(16777216)
);

create or replace view logins as select * from table(information_schema.login_history(TIME_RANGE_START => dateadd('minutes',-5,current_timestamp()),current_timestamp())) where IS_SUCCESS='NO' and FIRST_AUTHENTICATION_FACTOR='PASSWORD' order by event_timestamp;

CREATE OR REPLACE ALERT strange_login
  WAREHOUSE = siem_alert_demo
  SCHEDULE = '1 minute'
  IF( EXISTS(
      select event_id from logins
))
  THEN

INSERT INTO login_events  
SELECT *,false, NULL,NULL
FROM logins AS src
WHERE NOT EXISTS (SELECT event_id
                  FROM  login_events AS tgt
                  WHERE tgt.event_id = src.event_id);

    
ALTER ALERT STRANGE_LOGIN RESUME;

This will record any failed login attempts into the table LOGIN_EVENTS, but only if the EVENT_ID is not already in the table, so we dont’t have any duplicates.

3. We have to make sure that any event that has not been sent over to the SIEM system can now be sent.

3a)

In order to get that done we need to create the external access function to make the API call to a SIEM.

In this demo I am using a public MOCK API service that just returns an echo of whatever we are sending.

In real life you will have to setup an external access function to your SIEM system.

This will not only mean that you need to change the API URL and add things like a BEARER TOKEN, but it will also mean that you will have to adapt and change the JSON payload that needs to be sent.

Currently this payload is sent over a public IP adress space via the internet or Cloud Service Provider backbone.

This tutorial can also be adapted to use classic Snowflake external functions (instead of external access functions) which can also connect via Privatelink (on AWS today and soon on Azure).

External access function will be able to sent payload over Privatelink soon as well.

Another way is of course to create a proxy API in a DMZ until private connectivity is available, if your SIEM is not hosted in the public Cloud.

For now in this demo we keep it simple and just send the whole LOGIN_EVENT table info to the MOCK API.

CREATE OR REPLACE NETWORK RULE siem_alert_network_rule
  MODE = EGRESS
  TYPE = HOST_PORT
  VALUE_LIST = ('echo.free.beeceptor.com:443');


CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION siem_apis_access_integration
  ALLOWED_NETWORK_RULES = (siem_alert_network_rule)
  ENABLED = TRUE;



CREATE OR REPLACE FUNCTION alert_siem_login(incident variant, url string)
RETURNS variant
LANGUAGE PYTHON
RUNTIME_VERSION = 3.8
HANDLER = 'incident_alert'
EXTERNAL_ACCESS_INTEGRATIONS = (siem_apis_access_integration)
PACKAGES = ('snowflake-snowpark-python','requests')

AS
$$
import _snowflake
import requests
import json
session = requests.Session()

def incident_alert(incident, url):

    if incident != '':
        payload = incident
    else:
        payload={}
    headers = {}

    response = requests.request("POST", url, headers=headers, data=payload)

    return response.text
$$;

3b) Now that we are able to send JSON payload to an API we can create the Stored Procedure that goes through the LOGIN_EVENTS table and sents any new events to the SIEM:

CREATE OR REPLACE PROCEDURE sent_suspicious_logins()
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'run'
EXECUTE AS CALLER
AS
$$
from snowflake.snowpark import Session
import json

to_sent={}

def run(session):
  sql1 = """select to_variant(
SELECT
  ARRAY_AGG(json) AS aggregated_results
FROM
  (
    SELECT
      OBJECT_CONSTRUCT(*) AS json
    FROM
      (select * from login_events where sent=false)
  )
    )"""       

  to_sent=json.loads(session.sql(sql1).collect()[0][0])

  for siem_payload in to_sent:
    sql_siem = """UPDATE
                    login_events
                    SET
                        sent = TRUE,
                        sent_time = CURRENT_TIMESTAMP(),
                        siem_response= (select alert_siem_login('""" +json.dumps(siem_payload).replace('\"','"')+                                                       """','https://echo.free.beeceptor.com:443/siem_api/incident'))
                    WHERE
                        sent = FALSE"""
    sql_siem_result=session.sql(sql_siem).collect()[0][0]

  

  return "payload sent"
  
  
$$;

3c) With that all that is left to do is call the Stored Procedure every minute to make sure it regulary checks and sends new events to the SIEM automatically.

CREATE OR REPLACE task sent_strange_logins_to_siem
  WAREHOUSE = siem_alert_demo
  SCHEDULE = '1 minute'
  AS 
call sent_suspicious_logins();



alter task sent_strange_logins_to_siem resume;

4) In order to see if everything is working it’s good to test with a user to try to login to Snowflake using the wrong password once or twice.

After that keep running a SELECT query on the LOGIN_EVENTS table and you will see how it’s being populated with the login events and also with the SIEM_RESPONSE payload so you can see that the event was successfully sent.

SELECT * from LOGIN_EVENTS;

You should see this at the beginning of the table:

And and the end of the table you can see if the column SENT is still set to FALSE and the if other columns like SENT_TIMESTAMP and SIEM_RESPONSE are still NULL.

In that case the event was recorded in the table, but not sent to the SIEM yet.

Once the column SENT is set to TRUE and the SENT_TIMESTAMP and SIEM_RESPONSE columns have values, you can be sure that your SIEM was notified and an incident raised:

Here are some sample APIs from MS Sentinel that can be used:

Incidents - REST API (Azure Sentinel)

or SPLUNK:

https://dev.splunk.com/observability/reference/api/incidents/latest

Where can you go from here?

Many options are possible.

I will create some Snowflake dashboards and/or Streamlit dashboards so incidents can be easily visualized next. You can do that too.

Or have Snowflake itself deactivte the user first? Just add another stored procedure to the mix that does that.

Other events are also interesting to monitor e.g. maybe you want to be notified by some queries that some people should not execute?

Such as changes of NETWORK POLICIES?

Maybe you want to monitor logs from Snowflake Container services or running containers from an event table?

Or you think beyond Snowflake and ingest logs from other applications into Snowflake and build your dashboards and incident response API calls right within Snowflake.

With Snowflake Notebooks threat hunting in Python or regular SQL has never been more approachable.

Having all logs in Snowflake and the power to create automated API calls and doing threat hunting right in Snowflake means you can ingest and have as many logs available from as many systems as you like in a hot storage ready to query via Python and SQL at anytime.

Finally here is the whole SQL workbook for easy copy and pasting (make sure to replace XXX with your username before you run this!) :

use role accountadmin;
CREATE OR REPLACE ROLE SIEM_MONITOR;
grant role siem_monitor to user XXXX;
GRANT CREATE WAREHOUSE ON ACCOUNT TO ROLE siem_monitor;
GRANT MANAGE WAREHOUSES ON ACCOUNT TO ROLE siem_monitor;
GRANT CREATE DATABASE ON ACCOUNT TO ROLE siem_monitor;
GRANT CREATE ROLE ON ACCOUNT TO siem_monitor;
GRANT EXECUTE TASK ON ACCOUNT TO ROLE siem_monitor;
GRANT EXECUTE ALERT ON ACCOUNT TO ROLE siem_monitor;
GRANT CREATE INTEGRATION ON ACCOUNT TO ROLE  siem_monitor;



use role siem_monitor;

CREATE OR REPLACE WAREHOUSE siem_alert_demo WITH WAREHOUSE_SIZE='X-SMALL';
grant usage on warehouse siem_alert_demo to role SIEM_MONITOR;

CREATE OR REPLACE DATABASE SIEM_ALERT_DEMO_DB;
CREATE OR REPLACE SCHEMA SIEM_ALERT_DEMO_SCHEMA;

grant usage on database SIEM_ALERT_DEMO_DB to role accountadmin;
grant usage on schema SIEM_ALERT_DEMO_DB.SIEM_ALERT_DEMO_SCHEMA to role accountadmin;
grant usage on warehouse siem_alert_demo to role accountadmin;



CREATE OR REPLACE ROLE MONITOR_USERS;
grant role monitor_users to role siem_monitor;



CREATE OR REPLACE PROCEDURE set_monitor_privileges_on_all_users(users varchar)
RETURNS varchar
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'run'
EXECUTE AS CALLER
AS
$$
from snowflake.snowpark import Session


def run(session, users):
      users=users.replace('[','')
      users=users.replace(']','')
      users=users.split(',')
      for user in users:
              sql = """GRANT MONITOR ON USER """+user+""" TO ROLE monitor_users"""             
              result=session.sql(sql).collect()[0][0]


      return "done"
  
  
$$;

grant usage on procedure set_monitor_privileges_on_all_users(varchar) to role accountadmin;
use role accountadmin;
GRANT CREATE NETWORK RULE ON SCHEMA SIEM_ALERT_DEMO_DB.SIEM_ALERT_DEMO_SCHEMA TO ROLE siem_monitor;
call set_monitor_privileges_on_all_users(select all_user_names());



use role siem_monitor;
revoke usage on procedure set_monitor_privileges_on_all_users(varchar) from role accountadmin;
revoke usage on warehouse siem_alert_demo from role accountadmin;
revoke usage on database SIEM_ALERT_DEMO_DB from role accountadmin;
revoke usage on schema SIEM_ALERT_DEMO_DB.SIEM_ALERT_DEMO_SCHEMA from role accountadmin;



create or replace TABLE LOGIN_EVENTS (
 EVENT_TIMESTAMP TIMESTAMP_LTZ(3),
 EVENT_ID NUMBER(38,0),
 EVENT_TYPE VARCHAR(16777216),
 USER_NAME VARCHAR(16777216),
 CLIENT_IP VARCHAR(16777216),
 REPORTED_CLIENT_TYPE VARCHAR(16777216),
 REPORTED_CLIENT_VERSION VARCHAR(16777216),
 FIRST_AUTHENTICATION_FACTOR VARCHAR(16777216),
 SECOND_AUTHENTICATION_FACTOR VARCHAR(16777216),
 IS_SUCCESS VARCHAR(3),
 ERROR_CODE NUMBER(38,0),
 ERROR_MESSAGE VARCHAR(16777216),
 RELATED_EVENT_ID NUMBER(38,0),
 CONNECTION VARCHAR(16777216),
 SENT BOOLEAN DEFAULT FALSE,
 SENT_TIME TIMESTAMP_NTZ(9),
 SIEM_RESPONSE VARCHAR(16777216)
);



create or replace view logins as select * from table(information_schema.login_history(TIME_RANGE_START => dateadd('minutes',-5,current_timestamp()),current_timestamp())) where IS_SUCCESS='NO' and FIRST_AUTHENTICATION_FACTOR='PASSWORD' order by event_timestamp;

CREATE OR REPLACE ALERT strange_login
  WAREHOUSE = siem_alert_demo
  SCHEDULE = '1 minute'
  IF( EXISTS(
      select event_id from logins
))
  THEN

INSERT INTO login_events  
SELECT *,false, NULL,NULL
FROM logins AS src
WHERE NOT EXISTS (SELECT event_id
                  FROM  login_events AS tgt
                  WHERE tgt.event_id = src.event_id);

    
ALTER ALERT STRANGE_LOGIN RESUME;



CREATE OR REPLACE NETWORK RULE siem_alert_network_rule
  MODE = EGRESS
  TYPE = HOST_PORT
  VALUE_LIST = ('echo.free.beeceptor.com:443');

CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION siem_apis_access_integration
  ALLOWED_NETWORK_RULES = (siem_alert_network_rule)
  ENABLED = TRUE;



CREATE OR REPLACE FUNCTION alert_siem_login(incident string, url string)
RETURNS string
LANGUAGE PYTHON
RUNTIME_VERSION = 3.8
HANDLER = 'incident_alert'
EXTERNAL_ACCESS_INTEGRATIONS = (siem_apis_access_integration)
PACKAGES = ('snowflake-snowpark-python','requests')

AS
$$
import _snowflake
import requests
import json
session = requests.Session()

def incident_alert(incident, url):

    if incident != '':
        payload = incident
    else:
        payload={}
    headers = {}

    response = requests.request("POST", url, headers=headers, data=payload)

    return response.text
$$;


CREATE OR REPLACE PROCEDURE sent_suspicious_logins()
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'run'
EXECUTE AS CALLER
AS
$$
from snowflake.snowpark import Session
import json

to_sent={}

def run(session):
  sql1 = """select to_variant(
SELECT
  ARRAY_AGG(json) AS aggregated_results
FROM
  (
    SELECT
      OBJECT_CONSTRUCT(*) AS json
    FROM
      (select * from login_events where sent=false)
  )
    )"""       

  to_sent=json.loads(session.sql(sql1).collect()[0][0])

  for siem_payload in to_sent:
    sql_siem = """UPDATE
                    login_events
                    SET
                        sent = TRUE,
                        sent_time = CURRENT_TIMESTAMP(),
                        siem_response= (select alert_siem_login('""" +json.dumps(siem_payload).replace('\"','"')+                                                       """','https://echo.free.beeceptor.com:443/siem_api/incident'))
                    WHERE
                        sent = FALSE"""
    sql_siem_result=session.sql(sql_siem).collect()[0][0]

  

  return "payload sent"
  
  
$$;


CREATE OR REPLACE task sent_strange_logins_to_siem
  WAREHOUSE = siem_alert_demo
  SCHEDULE = '1 minute'
  AS 
call sent_suspicious_logins();



alter task sent_strange_logins_to_siem resume;

-- check 

show tasks;
show alerts;

SELECT * from LOGIN_EVENTS;

-- monitor alerts & tasks

SELECT *
FROM
  TABLE(INFORMATION_SCHEMA.ALERT_HISTORY(
    SCHEDULED_TIME_RANGE_START
      =>dateadd('hour',-1,current_timestamp())))
ORDER BY SCHEDULED_TIME DESC;

SELECT *
FROM
  TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
    SCHEDULED_TIME_RANGE_START
      =>dateadd('hour',-1,current_timestamp())))
ORDER BY SCHEDULED_TIME DESC;




-- clean up (optional)

use role siem_monitor;
drop role monitor_users;
drop warehouse siem_alert_demo;
drop database SIEM_ALERT_DEMO_DB;
drop EXTERNAL ACCESS INTEGRATION siem_apis_access_integration;