# Testing (SimpleIngestIT Test)
@@ -93,11 +143,13 @@ you would need to remove the following scope limits in pom.xml
- Use an unencrypted version(Only for testing) of private key while generating keys(private and public pair) using OpenSSL.
- Here is the link for documentation [Key Pair
- Generator](https://docs.snowflake.net/manuals/user-guide/python-connector-example.html#using-key-pair-authentication)
+ Generator](https://docs.snowflake.com/en/user-guide/key-pair-auth.html)
-# Google Java Format
+# Contributing to this repo
-- Download the formatter jar file from https://github.com/google/google-java-format, then run it with
-```
-java -jar ~/path-to/google-java-format-1.10.0-all-deps.jar -i $(find . -type f -name "*.java" | grep ".*/src/.*java")
-```
+Each PR must pass all required github action merge gates before approval and merge. In addition to those tests, you will need:
+
+- Formatter: run this script [`./format.sh`](https://github.com/snowflakedb/snowflake-ingest-java/blob/master/format.sh) from root
+- CLA: all contributers must sign the Snowflake CLA. This is a one time signature, please provide your email so we can work with you to get this signed after you open a PR.
+
+Thank you for contributing! We will review and approve PRs as soon as we can.
diff --git a/deploy.sh b/deploy.sh
index 0df76eedf..d3cd928fc 100755
--- a/deploy.sh
+++ b/deploy.sh
@@ -84,8 +84,4 @@ mvn ${MVN_OPTIONS[@]} \
-DstagingRepositoryId=$snowflake_repositories \
-DstagingDescription="Automated Release"
-rm $OSSRH_DEPLOY_SETTINGS_XML
-
-#white source
-chmod 755 ./scripts/run_whitesource_gh.sh
-scripts/run_whitesource_gh.sh
+rm $OSSRH_DEPLOY_SETTINGS_XML
\ No newline at end of file
diff --git a/e2e-jar-test/core/pom.xml b/e2e-jar-test/core/pom.xml
new file mode 100644
index 000000000..36d52ae5b
--- /dev/null
+++ b/e2e-jar-test/core/pom.xml
@@ -0,0 +1,38 @@
+
+ 4.0.0
+
+ net.snowflake.snowflake-ingest-java-e2e-jar-test
+ parent
+ 1.0-SNAPSHOT
+
+
+ core
+ core
+ jar
+
+
+ UTF-8
+ 1.8
+ 1.8
+
+
+
+
+
+ net.snowflake
+ snowflake-ingest-sdk
+ provided
+
+
+
+ org.slf4j
+ slf4j-simple
+
+
+
+ com.fasterxml.jackson.core
+ jackson-databind
+
+
+
diff --git a/e2e-jar-test/core/src/main/java/net/snowflake/IngestTestUtils.java b/e2e-jar-test/core/src/main/java/net/snowflake/IngestTestUtils.java
new file mode 100644
index 000000000..e7db8d1c2
--- /dev/null
+++ b/e2e-jar-test/core/src/main/java/net/snowflake/IngestTestUtils.java
@@ -0,0 +1,200 @@
+package net.snowflake;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.security.KeyFactory;
+import java.security.NoSuchAlgorithmException;
+import java.security.PrivateKey;
+import java.security.spec.InvalidKeySpecException;
+import java.security.spec.PKCS8EncodedKeySpec;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+import java.util.UUID;
+
+import net.snowflake.ingest.streaming.OpenChannelRequest;
+import net.snowflake.ingest.streaming.SnowflakeStreamingIngestChannel;
+import net.snowflake.ingest.streaming.SnowflakeStreamingIngestClient;
+import net.snowflake.ingest.streaming.SnowflakeStreamingIngestClientFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class IngestTestUtils {
+ private static final String PROFILE_PATH = "profile.json";
+
+ private final Connection connection;
+
+ private final String database;
+ private final String schema;
+ private final String table;
+
+ private final String testId;
+
+ private static final Logger logger = LoggerFactory.getLogger(IngestTestUtils.class);
+
+ private final SnowflakeStreamingIngestClient client;
+
+ private final SnowflakeStreamingIngestChannel channel;
+
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
+ private final Random random = new Random();
+
+ private final Base64.Decoder base64Decoder = Base64.getDecoder();
+
+ public IngestTestUtils(String testName)
+ throws SQLException,
+ IOException,
+ ClassNotFoundException,
+ NoSuchAlgorithmException,
+ InvalidKeySpecException {
+ testId = String.format("%s_%s", testName, UUID.randomUUID().toString().replace("-", "_"));
+ connection = getConnection();
+ database = String.format("database_%s", testId);
+ schema = String.format("schema_%s", testId);
+ table = String.format("table_%s", testId);
+
+ connection.createStatement().execute(String.format("create database %s", database));
+ connection.createStatement().execute(String.format("create schema %s", schema));
+ connection.createStatement().execute(String.format("create table %s (c1 int, c2 varchar, c3 binary)", table));
+
+ client =
+ SnowflakeStreamingIngestClientFactory.builder("TestClient01")
+ .setProperties(loadProperties())
+ .build();
+
+ channel = client.openChannel(
+ OpenChannelRequest.builder(String.format("channel_%s", this.testId))
+ .setDBName(database)
+ .setSchemaName(schema)
+ .setTableName(table)
+ .setOnErrorOption(OpenChannelRequest.OnErrorOption.CONTINUE)
+ .build());
+ }
+
+ private Properties loadProperties() throws IOException {
+ Properties props = new Properties();
+ Iterator> propIt =
+ objectMapper.readTree(new String(Files.readAllBytes(Paths.get(PROFILE_PATH)))).fields();
+ while (propIt.hasNext()) {
+ Map.Entry prop = propIt.next();
+ props.put(prop.getKey(), prop.getValue().asText());
+ }
+ return props;
+ }
+
+ private Connection getConnection()
+ throws IOException, ClassNotFoundException, SQLException, NoSuchAlgorithmException, InvalidKeySpecException {
+ Class.forName("net.snowflake.client.jdbc.SnowflakeDriver");
+
+ Properties loadedProps = loadProperties();
+
+ byte[] decoded = base64Decoder.decode(loadedProps.getProperty("private_key"));
+ KeyFactory kf = KeyFactory.getInstance("RSA");
+
+ PKCS8EncodedKeySpec keySpec = new PKCS8EncodedKeySpec(decoded);
+ PrivateKey privateKey = kf.generatePrivate(keySpec);
+
+ Properties props = new Properties();
+ props.putAll(loadedProps);
+ props.put("client_session_keep_alive", "true");
+ props.put("privateKey", privateKey);
+
+ return DriverManager.getConnection(loadedProps.getProperty("connect_string"), props);
+ }
+
+ private Map createRow() {
+ Map row = new HashMap<>();
+
+ byte[] bytes = new byte[1024];
+ random.nextBytes(bytes);
+
+ row.put("c1", random.nextInt());
+ row.put("c2", String.valueOf(random.nextInt()));
+ row.put("c3", bytes);
+
+ return row;
+ }
+
+ /**
+ * Given a channel and expected offset, this method waits up to 60 seconds until the last
+ * committed offset is equal to the passed offset
+ */
+ private void waitForOffset(SnowflakeStreamingIngestChannel channel, String expectedOffset)
+ throws InterruptedException {
+ int counter = 0;
+ String lastCommittedOffset = null;
+ while (counter < 600) {
+ String currentOffset = channel.getLatestCommittedOffsetToken();
+ if (expectedOffset.equals(currentOffset)) {
+ return;
+ }
+ System.out.printf("Waiting for offset expected=%s actual=%s%n", expectedOffset, currentOffset);
+ lastCommittedOffset = currentOffset;
+ counter++;
+ Thread.sleep(100);
+ }
+ throw new RuntimeException(
+ String.format(
+ "Timeout exceeded while waiting for offset %s. Last committed offset: %s",
+ expectedOffset, lastCommittedOffset));
+ }
+
+ public void runBasicTest() throws InterruptedException {
+ // Insert few rows one by one
+ for (int offset = 2; offset < 1000; offset++) {
+ offset++;
+ channel.insertRow(createRow(), String.valueOf(offset));
+ }
+
+ // Insert a batch of rows
+ String offset = "final-offset";
+ channel.insertRows(
+ Arrays.asList(createRow(), createRow(), createRow(), createRow(), createRow()), offset);
+
+ waitForOffset(channel, offset);
+ }
+
+ public void runLongRunningTest(Duration testDuration) throws InterruptedException {
+ final Instant testStart = Instant.now();
+ int counter = 0;
+ while(true) {
+ counter++;
+
+ channel.insertRow(createRow(), String.valueOf(counter));
+
+ if (!channel.isValid()) {
+ throw new IllegalStateException("Channel has been invalidated");
+ }
+ Thread.sleep(60000);
+
+ final Duration elapsed = Duration.between(testStart, Instant.now());
+
+ logger.info("Test loop_nr={} duration={}s/{}s committed_offset={}", counter, elapsed.get(ChronoUnit.SECONDS), testDuration.get(ChronoUnit.SECONDS), channel.getLatestCommittedOffsetToken());
+
+ if (elapsed.compareTo(testDuration) > 0) {
+ break;
+ }
+ }
+ waitForOffset(channel, String.valueOf(counter));
+ }
+
+ public void close() throws Exception {
+ connection.close();
+ channel.close().get();
+ client.close();
+ }
+}
diff --git a/e2e-jar-test/fips/pom.xml b/e2e-jar-test/fips/pom.xml
new file mode 100644
index 000000000..039af9b7a
--- /dev/null
+++ b/e2e-jar-test/fips/pom.xml
@@ -0,0 +1,53 @@
+
+ 4.0.0
+
+ net.snowflake.snowflake-ingest-java-e2e-jar-test
+ parent
+ 1.0-SNAPSHOT
+
+
+ fips
+ jar
+ fips
+
+
+ UTF-8
+ 1.8
+ 1.8
+
+
+
+
+ net.snowflake.snowflake-ingest-java-e2e-jar-test
+ core
+
+
+
+ net.snowflake
+ snowflake-ingest-sdk
+
+
+ net.snowflake
+ snowflake-jdbc
+
+
+ org.bouncycastle
+ bcpkix-jdk18on
+
+
+ org.bouncycastle
+ bcprov-jdk18on
+
+
+
+
+ net.snowflake
+ snowflake-jdbc-fips
+
+
+ junit
+ junit
+
+
+
diff --git a/e2e-jar-test/fips/src/test/java/net/snowflake/FipsIngestE2ETest.java b/e2e-jar-test/fips/src/test/java/net/snowflake/FipsIngestE2ETest.java
new file mode 100644
index 000000000..c6f9bfe33
--- /dev/null
+++ b/e2e-jar-test/fips/src/test/java/net/snowflake/FipsIngestE2ETest.java
@@ -0,0 +1,40 @@
+package net.snowflake;
+
+import org.bouncycastle.jcajce.provider.BouncyCastleFipsProvider;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.security.Security;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+
+public class FipsIngestE2ETest {
+
+ private IngestTestUtils ingestTestUtils;
+
+ @Before
+ public void setUp() throws Exception {
+ // Add FIPS provider, the SDK does not do this by default
+ Security.addProvider(new BouncyCastleFipsProvider("C:HYBRID;ENABLE{All};"));
+
+ ingestTestUtils = new IngestTestUtils("fips_ingest");
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ ingestTestUtils.close();
+ }
+
+ @Test
+ public void basicTest() throws InterruptedException {
+ ingestTestUtils.runBasicTest();
+ }
+
+ @Test
+ @Ignore("Takes too long to run")
+ public void longRunningTest() throws InterruptedException {
+ ingestTestUtils.runLongRunningTest(Duration.of(80, ChronoUnit.MINUTES));
+ }
+}
diff --git a/e2e-jar-test/pom.xml b/e2e-jar-test/pom.xml
new file mode 100644
index 000000000..e9fb43592
--- /dev/null
+++ b/e2e-jar-test/pom.xml
@@ -0,0 +1,65 @@
+
+ 4.0.0
+
+ net.snowflake.snowflake-ingest-java-e2e-jar-test
+ parent
+ 1.0-SNAPSHOT
+ pom
+
+ snowflake-ingest-sdk-e2e-test
+
+
+ standard
+ fips
+ core
+
+
+
+
+
+
+ net.snowflake.snowflake-ingest-java-e2e-jar-test
+ core
+ ${project.version}
+
+
+
+ net.snowflake
+ snowflake-ingest-sdk
+ 2.0.0
+
+
+
+ net.snowflake
+ snowflake-jdbc-fips
+ 3.13.34
+
+
+
+ org.bouncycastle
+ bc-fips
+ 1.0.2.4
+
+
+
+ org.slf4j
+ slf4j-simple
+ 1.7.36
+
+
+
+ com.fasterxml.jackson.core
+ jackson-databind
+ 2.14.0
+
+
+
+ junit
+ junit
+ 4.13.2
+ test
+
+
+
+
diff --git a/e2e-jar-test/run_e2e_jar_test.sh b/e2e-jar-test/run_e2e_jar_test.sh
new file mode 100755
index 000000000..e5e9b5c55
--- /dev/null
+++ b/e2e-jar-test/run_e2e_jar_test.sh
@@ -0,0 +1,61 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+## This script tests the SDK JARs end-to-end, i.e. not using integration tests from within the project, but from an
+## external Maven project, which depends on the SDK deployed into the local maven repository. The following SDK variants are tested:
+## 1. Shaded jar
+## 2. Unshaded jar
+## 3. FIPS-compliant jar, i.e. unshaded jar without snowflake-jdbc and bouncy castle dependencies, but with snowflake-jdbc-fips depedency
+
+maven_repo_dir=$(mvn help:evaluate -Dexpression=settings.localRepository -q -DforceStdout)
+sdk_repo_dir="${maven_repo_dir}/net/snowflake/snowflake-ingest-sdk"
+
+cp profile.json e2e-jar-test/standard
+cp profile.json e2e-jar-test/fips
+
+###################
+# TEST SHADED JAR #
+###################
+
+# Remove the SDK from local maven repository
+rm -fr $sdk_repo_dir
+
+# Prepare pom.xml for shaded JAR
+project_version=$(./scripts/get_project_info_from_pom.py pom.xml version)
+./scripts/update_project_version.py public_pom.xml $project_version > generated_public_pom.xml
+
+# Build shaded SDK
+mvn clean package -DskipTests=true --batch-mode --show-version
+
+# Install shaded SDK JARs into local maven repository
+mvn install:install-file -Dfile=target/snowflake-ingest-sdk.jar -DpomFile=generated_public_pom.xml
+
+# Run e2e tests
+(cd e2e-jar-test && mvn clean verify -pl standard -am)
+
+#####################
+# TEST UNSHADED JAR #
+#####################
+
+# Remove the SDK from local maven repository
+rm -r $sdk_repo_dir
+
+# Install unshaded SDK into local maven repository
+mvn clean install -Dnot-shadeDep -DskipTests=true --batch-mode --show-version
+
+# Run e2e tests
+(cd e2e-jar-test && mvn clean verify -pl standard -am)
+
+#############
+# TEST FIPS #
+#############
+
+# Remove the SDK from local maven repository
+rm -r $sdk_repo_dir
+
+# Install unshaded SDK into local maven repository
+mvn clean install -Dnot-shadeDep -DskipTests=true --batch-mode --show-version
+
+# Run e2e tests on the FIPS module
+(cd e2e-jar-test && mvn clean verify -pl fips -am)
\ No newline at end of file
diff --git a/e2e-jar-test/standard/pom.xml b/e2e-jar-test/standard/pom.xml
new file mode 100644
index 000000000..2fdddcddd
--- /dev/null
+++ b/e2e-jar-test/standard/pom.xml
@@ -0,0 +1,34 @@
+
+ 4.0.0
+
+ net.snowflake.snowflake-ingest-java-e2e-jar-test
+ parent
+ 1.0-SNAPSHOT
+
+
+ standard
+ jar
+ unshaded
+
+
+ UTF-8
+ 1.8
+ 1.8
+
+
+
+
+ net.snowflake.snowflake-ingest-java-e2e-jar-test
+ core
+
+
+ net.snowflake
+ snowflake-ingest-sdk
+
+
+ junit
+ junit
+
+
+
diff --git a/e2e-jar-test/standard/src/test/java/net/snowflake/StandardIngestE2ETest.java b/e2e-jar-test/standard/src/test/java/net/snowflake/StandardIngestE2ETest.java
new file mode 100644
index 000000000..211c421fc
--- /dev/null
+++ b/e2e-jar-test/standard/src/test/java/net/snowflake/StandardIngestE2ETest.java
@@ -0,0 +1,35 @@
+package net.snowflake;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+
+public class StandardIngestE2ETest {
+
+ private IngestTestUtils ingestTestUtils;
+
+ @Before
+ public void setUp() throws Exception {
+ ingestTestUtils = new IngestTestUtils("standard_ingest");
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ ingestTestUtils.close();
+ }
+
+ @Test
+ public void basicTest() throws InterruptedException {
+ ingestTestUtils.runBasicTest();
+ }
+
+ @Test
+ @Ignore("Takes too long to run")
+ public void longRunningTest() throws InterruptedException {
+ ingestTestUtils.runLongRunningTest(Duration.of(80, ChronoUnit.MINUTES));
+ }
+}
diff --git a/format.sh b/format.sh
new file mode 100755
index 000000000..cb58bd995
--- /dev/null
+++ b/format.sh
@@ -0,0 +1,24 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+cd "$SCRIPT_DIR"
+
+DOWNLOAD_URL="https://github.com/google/google-java-format/releases/download/v1.10.0/google-java-format-1.10.0-all-deps.jar"
+JAR_FILE="./.cache/google-java-format-1.10.0-all-deps.jar"
+
+if [ ! -f "${JAR_FILE}" ]; then
+ mkdir -p "$(dirname "${JAR_FILE}")"
+ echo "Downloading Google Java format to ${JAR_FILE}"
+ curl -# -L --fail "${DOWNLOAD_URL}" --output "${JAR_FILE}"
+fi
+
+if ! command -v java > /dev/null; then
+ echo "Java not installed."
+ exit 1
+fi
+echo "Running Google Java Format"
+find ./src -type f -name "*.java" -print0 | xargs -0 java -jar "${JAR_FILE}" --replace --set-exit-if-changed && echo "OK"
+
+echo "Sorting pom.xml"
+mvn com.github.ekryd.sortpom:sortpom-maven-plugin:sort
diff --git a/linkage-checker-exclusion-rules.xml b/linkage-checker-exclusion-rules.xml
new file mode 100644
index 000000000..0cb2eb38c
--- /dev/null
+++ b/linkage-checker-exclusion-rules.xml
@@ -0,0 +1,161 @@
+
+
+
+
+
+
+
+
+
+ Bouncy castle is an optional dependency of nimbus-jose-jwt
+
+
+
+
+
+
+
+
+ Google Crypto Tink is an optional dependency of nimbus-jose-jwt
+
+
+
+
+
+
+ Seems like a false positive, this class does exist on classpath.
+
+
+
+
+
+ Seems like a false positive, this class does exist on classpath.
+
+
+
+
+
+
+ Not used in the SDK
+
+
+
+
+
+
+
+
+ Relates to security/SSL pulled in by hadoop-auth, not used
+
+
+
+
+
+ We do not need Hadoop authentication
+
+
+
+
+
+
+
+ We don't explicitly exclude any dependency of the JDBC driver.
+ If some classes are missing, they know what are they doing.
+
+
+
+
+
+
+
+
+
+ Optional dependency of Snowflake JDBC driver that the SDK doesn't need
+
+
+
+
+
+
+
+
+
+
+
+
+
+ The SDK is not using mapreduce
+
+
+
+
+
+ The SDK is not using mapreduce
+
+
+
+
+
+ The SDK is not using Hadoop YARN
+
+
+
+
+
+ The SDK does not use XZ compression
+
+
+
+
+
+ Client library for Apache ZooKeeper, not used in the SDK
+
+
+
+
+
+ SSH client, not used in the SDK
+
+
+
+
+
+ The SDK is not using Avro
+
+
+
+
+
+ Not used, the SDK is using FasterXML Jackson
+
+
+
+
+
+ Comes from dnsjava, which is not used in the SDK
+
+
+
+
+
+ Not used in the SDK
+
+
+
+
+
+ The SDK does not log with log4j
+
+
+
+
+
+ The SDK does not use web services
+
+
+
+
+
+ The SDK does not use Jetty web server
+
+
diff --git a/pom.xml b/pom.xml
index 9a747959f..58cd1f485 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1,561 +1,1268 @@
-
- 4.0.0
+
+ 4.0.0
-
- net.snowflake
- snowflake-ingest-sdk
- 1.0.1-beta
- jar
- Snowflake Ingest SDK
- Snowflake Ingest SDK
- https://www.snowflake.net/
+
+ io.confluent
+ snowflake-ingest-sdk
+ 2.0.0
+ jar
+ Snowflake Ingest SDK
+ Snowflake Ingest SDK
+
+
+ 1.74
+ 1.9.13
+ 1.15
+ 3.2.2
+ 1.21
+ 2.8.0
+ 3.12.0
+ 1.2
+ 1.10.0
+ 2.14.0
+ 32.0.1-jre
+ 3.3.6
+ true
+ 0.8.5
+ ${project.build.directory}/dependency-jars
+ ${project.build.directory}/dependency-list.txt
+ ${project.build.directory}/generated-licenses-info
+ ${license.processing.resourcesRoot}/META-INF/third-party-licenses
+ 1.8
+ 1.8
+ 2.4.9
+ 4.1.94.Final
+ 9.31
+ 3.1
+ 1.13.1
+ UTF-8
+ 3.19.6
+ net.snowflake.ingest.internal
+ 1.7.36
+ 1.1.10.4
+ 3.14.5
+ 0.13.0
+
-
-
- The Apache Software License, Version 2.0
- http://www.apache.org/licenses/LICENSE-2.0.txt
-
-
-
-
-
- Snowflake Support Team
- snowflake-java@snowflake.net
- Snowflake Computing
- https://www.snowflake.net
-
-
-
-
- scm:git:git://github.com/snowflakedb/snowflake-ingest-java
- http://github.com/snowflakedb/snowflake-ingest-java/tree/master
-
-
-
-
- 1.8
- 1.8
- net.snowflake.ingest.internal
- true
- 0.8.5
-
-
-
- ${project.artifactId}
-
-
- src/main/resources
- true
-
-
-
-
- org.apache.maven.plugins
- maven-surefire-plugin
- 2.19.1
-
-
- **/TestSimpleIngestLocal.java
-
-
-
-
- maven-assembly-plugin
-
-
-
- true
- net.snowflake.ingest.SimpleIngestManager
-
-
-
- jar-with-dependencies
-
-
-
-
- org.apache.maven.plugins
- maven-compiler-plugin
- 2.5.1
- true
-
- 1.8
- 1.8
-
-
-
- org.apache.maven.plugins
- maven-javadoc-plugin
- 2.10.4
-
-
- attach-javadocs
-
- jar
-
-
- -Xdoclint:none
- 8
-
-
-
-
-
- org.apache.maven.plugins
- maven-source-plugin
- 3.0.1
-
-
- attach-sources
-
- jar
-
-
-
-
-
- org.codehaus.mojo
- exec-maven-plugin
-
-
- check-shaded-content
- verify
-
- exec
-
-
- ${basedir}/scripts/check_content.sh
-
-
-
-
-
-
-
-
-
-
- maven-deploy-plugin
-
- true
-
-
-
- org.jacoco
- jacoco-maven-plugin
- ${jacoco.version}
-
-
- pre-unit-test
-
- prepare-agent
-
-
- target/jacoco-ut.exec
-
-
-
- post-unit-test
- test
-
- report
-
-
- target/jacoco-ut.exec
- target/jacoco-ut
-
-
-
-
- ${jacoco.skip.instrument}
-
-
-
-
-
- org.apache.maven.plugins
- maven-failsafe-plugin
- 3.0.0-M5
-
-
- **/StreamingIngestIT.java
-
-
-
-
-
-
-
-
+
+
+ com.fasterxml.jackson
+ jackson-bom
+ ${fasterxml.version}
+ pom
+ import
+
+
+ com.google.guava
+ guava
+ ${guava.version}
+
+
+ com.google.protobuf
+ protobuf-java
+ ${protobuf.version}
+
+
+ com.nimbusds
+ nimbus-jose-jwt
+ ${nimbusds.version}
+
+
+ commons-codec
+ commons-codec
+ ${commonscodec.version}
+
+
+ commons-collections
+ commons-collections
+ ${commonscollections.version}
+
+
+ commons-io
+ commons-io
+ ${commonsio.version}
+
+
+ commons-logging
+ commons-logging
+ ${commonslogging.version}
+
+
+ io.netty
+ netty-buffer
+ ${netty.version}
+
+
+ io.netty
+ netty-common
+ ${netty.version}
+
+
+ net.minidev
+ json-smart
+ ${net.minidev.version}
+
+
+ net.snowflake
+ snowflake-jdbc
+ ${snowjdbc.version}
+
+
+ org.apache.commons
+ commons-compress
+ ${commonscompress.version}
+
+
+ org.apache.commons
+ commons-lang3
+ ${commonslang3.version}
+
+
+ org.apache.commons
+ commons-text
+ ${commonstext.version}
+
+
+ org.apache.hadoop
+ hadoop-common
+ ${hadoop.version}
+
+
+ ch.qos.reload4j
+ reload4j
+
+
+ com.github.pjfanning
+ jersey-json
+
+
+ com.jcraft
+ jsch
+
+
+ com.sun.jersey
+ jersey-core
+
+
+ com.sun.jersey
+ jersey-server
+
+
+ com.sun.jersey
+ jersey-servlet
+
+
+ commons-logging
+ commons-logging
+
+
+ dnsjava
+ dnsjava
+
+
+ io.dropwizard.metrics
+ metrics-core
+
+
+ javax.activation
+ activation
+
+
+ javax.servlet
+ javax.servlet-api
+
+
+ javax.servlet.jsp
+ jsp-api
+
+
+ org.apache.avro
+ avro
+
+
+ org.apache.curator
+ curator-client
+
+
+ org.apache.curator
+ curator-recipes
+
+
+ org.apache.hadoop
+ hadoop-auth
+
+
+ org.apache.httpcomponents
+ httpclient
+
+
+ org.apache.kerby
+ kerb-core
+
+
+ org.apache.zookeeper
+ zookeeper
+
+
+ org.eclipse.jetty
+ jetty-server
+
+
+ org.eclipse.jetty
+ jetty-servlet
+
+
+ org.eclipse.jetty
+ jetty-util
+
+
+ org.eclipse.jetty
+ jetty-webapp
+
+
+ org.slf4j
+ slf4j-log4j12
+
+
+ org.slf4j
+ slf4j-reload4j
+
+
+
+
+ org.apache.parquet
+ parquet-column
+ ${parquet.version}
+
+
+ org.apache.parquet
+ parquet-common
+ ${parquet.version}
+
+
+ org.apache.parquet
+ parquet-hadoop
+ ${parquet.version}
+
+
+ org.apache.yetus
+ audience-annotations
+ ${yetus.version}
+
+
+ org.bouncycastle
+ bcpkix-jdk18on
+ ${bouncycastle.version}
+
+
+ org.bouncycastle
+ bcprov-jdk18on
+ ${bouncycastle.version}
+
+
+ org.codehaus.jackson
+ jackson-core-asl
+ ${codehaus.version}
+
+
+ org.codehaus.jackson
+ jackson-jaxrs
+ ${codehaus.version}
+
+
+ org.codehaus.jackson
+ jackson-xc
+ ${codehaus.version}
+
+
+ org.codehaus.jettison
+ jettison
+ 1.5.4
+
+
+ org.codehaus.woodstox
+ stax2-api
+ 4.2.1
+
+
+ org.objenesis
+ objenesis
+ ${objenesis.version}
+
+
+ org.slf4j
+ slf4j-api
+ ${slf4j.version}
+
+
+ org.xerial.snappy
+ snappy-java
+ ${snappy.version}
+
-
-
- net.snowflake
- snowflake-jdbc
- 3.13.14
-
+
+
+ junit
+ junit
+ 4.13.2
+ test
+
+
+ net.bytebuddy
+ byte-buddy
+ 1.10.19
+ test
+
+
+ net.bytebuddy
+ byte-buddy-agent
+ 1.10.19
+ test
+
+
+ org.mockito
+ mockito-core
+ 3.7.7
+ test
+
+
+
-
-
- com.ibm.icu
- icu4j
- 70.1
-
+
+
+
+ com.fasterxml.jackson.core
+ jackson-annotations
+
+
+ com.fasterxml.jackson.core
+ jackson-core
+
+
+
+ com.fasterxml.jackson.core
+ jackson-databind
+
-
-
- com.nimbusds
- nimbus-jose-jwt
- 9.9.3
-
+
+
+
+ com.github.ben-manes.caffeine
+ caffeine
+ 2.9.3
+
+
+
+ com.google.errorprone
+ error_prone_annotations
+
+
+ org.checkerframework
+ checker-qual
+
+
+
+
+ com.google.code.findbugs
+ jsr305
+ 3.0.2
+
+
+ com.google.guava
+ guava
+
-
-
- com.fasterxml.jackson.core
- jackson-core
- 2.13.1
-
+
+
+ com.nimbusds
+ nimbus-jose-jwt
+
+
+ commons-codec
+ commons-codec
+
-
-
- com.fasterxml.jackson.core
- jackson-databind
- 2.13.2.1
-
+
+
+ io.dropwizard.metrics
+ metrics-core
+ 4.1.22
+
-
-
- org.apache.httpcomponents
- httpclient
- 4.5.13
-
-
- commons-codec
- commons-codec
-
-
-
+
+
+ io.dropwizard.metrics
+ metrics-jmx
+ 4.1.22
+
-
- commons-codec
- commons-codec
- 1.15
-
+
+
+ io.dropwizard.metrics
+ metrics-jvm
+ 4.1.22
+
+
+ io.netty
+ netty-common
+
+
+ net.bytebuddy
+ byte-buddy-agent
+
+
+
+ net.snowflake
+ snowflake-jdbc
+
+
+ org.apache.hadoop
+ hadoop-common
+
+
+
+ org.apache.parquet
+ parquet-column
+
+
+ org.apache.parquet
+ parquet-common
+
+
+ org.apache.parquet
+ parquet-hadoop
+
+
+ org.bouncycastle
+ bcpkix-jdk18on
+
+
+ org.bouncycastle
+ bcprov-jdk18on
+
+
+ org.slf4j
+ slf4j-api
+
+
+ com.github.luben
+ zstd-jni
+ 1.5.0-1
+ runtime
+
+
+ com.google.protobuf
+ protobuf-java
+ runtime
+
+
+ org.xerial.snappy
+ snappy-java
+ runtime
+
-
-
- org.apache.httpcomponents
- httpasyncclient
- 4.1.2
-
+
+
+ junit
+ junit
+ test
+
+
+ org.apache.commons
+ commons-lang3
+ test
+
+
+ org.hamcrest
+ hamcrest-core
+ 1.3
+ test
+
+
+ org.mockito
+ mockito-core
+ test
+
+
+ org.powermock
+ powermock-api-mockito2
+ 2.0.2
+ test
+
+
+ org.powermock
+ powermock-core
+ 2.0.2
+ test
+
+
+ org.powermock
+ powermock-module-junit4
+ 2.0.2
+ test
+
+
+
+ org.slf4j
+ slf4j-simple
+ ${slf4j.version}
+ test
+
+
+
+ ${project.artifactId}
+
+
+ true
+ src/main/resources
+
+
+ ${license.processing.resourcesRoot}
+
+
-
-
- org.slf4j
- slf4j-api
- 1.7.21
- provided
-
+
+
+
+
+ maven-deploy-plugin
+
+
+
+
+ org.apache.maven.plugins
+ maven-enforcer-plugin
+ 3.0.0-M3
+
+
+ com.google.cloud.tools
+ linkage-checker-enforcer-rules
+ 1.5.13
+
+
+ org.codehaus.mojo
+ extra-enforcer-rules
+ 1.3
+
+
+ org.eclipse.aether
+ aether-util
+
+
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-failsafe-plugin
+ 3.0.0-M5
+
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+ 2.19.1
+
+
+ org.jacoco
+ jacoco-maven-plugin
+ ${jacoco.version}
+
+ ${jacoco.skip.instrument}
+
+ **/*SnowflakeStreamingIngestExample*
+ **/*SnowflakeIngestBasicExample*
+ **/*IngestExampleHelper*
+
+
+
+
+ pre-unit-test
+
+ prepare-agent
+
+
+ target/jacoco-ut.exec
+
+
+
+ post-unit-test
+
+ report
+
+ test
+
+ target/jacoco-ut.exec
+ target/jacoco-ut
+
+
+
+
+
+
+
+
+ maven-assembly-plugin
+
+
+
+ true
+ net.snowflake.ingest.SimpleIngestManager
+
+
+
+ jar-with-dependencies
+
+
+
+
+ org.apache.maven.plugins
+ maven-compiler-plugin
+ 2.5.1
+ true
+
+ 1.8
+ 1.8
+
+
+
+ org.apache.maven.plugins
+ maven-dependency-plugin
+ 3.3.0
+
+
+ analyze
+
+ analyze-only
+
+
+ true
+ true
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-enforcer-plugin
+
+
+ enforce-best-practices
+
+ enforce
+
+
+
+
+
+ META-INF/versions/*/org/bouncycastle/*
+
+ true
+ true
+
+
+
+
+
+
+
+
+
+ enforce-maven
+
+ enforce
+
+
+
+
+ 3.5
+
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-javadoc-plugin
+ 2.10.4
+
+
+ attach-javadocs
+
+ jar
+
+
+ -Xdoclint:none
+ 8
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-source-plugin
+ 3.0.1
+
+
+ attach-sources
+
+ jar
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+ 2.19.1
+
+
+ **/TestSimpleIngestLocal.java
+
+
+
+
+ org.codehaus.mojo
+ license-maven-plugin
+ 2.0.1
+
+ failFast
+
+
+ Apache License 2.0
+ BSD 2-Clause License
+ 3-Clause BSD License
+ The MIT License
+ EDL 1.0
+ The Go license
+ Bouncy Castle Licence
+
+ test,provided,system
+ true
+
+ Apache License 2.0
+ |The Apache License, Version 2.0
+ |The Apache Software License, Version 2.0
+ |Apache-2.0
+ |Apache License, Version 2.0
+ |Apache 2.0
+ |Apache License V2.0
+ BSD 2-Clause License
+ |The BSD License
+ The MIT License|MIT License
+
+
+
+
+ add-third-party
+
+ add-third-party
+
+ package
+
+
+
+
+
+ https://www.snowflake.net/
-
-
- org.slf4j
- slf4j-simple
- 1.7.21
- test
-
+
+
+ The Apache Software License, Version 2.0
+ https://www.apache.org/licenses/LICENSE-2.0.txt
+
+
-
-
-
- javax.xml.bind
- jaxb-api
- 2.3.1
-
+
+
+ Snowflake Support Team
+ snowflake-java@snowflake.net
+ Snowflake Computing
+ https://www.snowflake.net
+
+
+
+ scm:git:git://github.com/snowflakedb/snowflake-ingest-java
+ https://github.com/snowflakedb/snowflake-ingest-java/tree/master
+
-
-
- junit
- junit
- 4.13.2
- test
-
-
- org.powermock
- powermock-module-junit4
- 2.0.2
- test
-
-
- org.mockito
- mockito-core
- 3.7.7
- test
-
-
- org.powermock
- powermock-api-mockito2
- 2.0.2
- test
-
-
- org.powermock
- powermock-core
- 2.0.2
- test
-
+
+
+ checkLinkageErrors
+
+
+
+ not-shadeDep
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-enforcer-plugin
+
+
+ enforce-linkage-checker
+
+ enforce
+
+ verify
+
+
+
+ true
+ ${basedir}/linkage-checker-exclusion-rules.xml
+ WARN
+
+
+
+
+
+
+
+
+
+
-
-
- org.apache.arrow
- arrow-vector
- 4.0.0
-
-
- org.apache.arrow
- arrow-memory-netty
- 4.0.0
- runtime
-
+
+
-
-
- io.dropwizard.metrics
- metrics-core
- 4.1.22
-
+
+
+
+
+ checkShadedContent
+
+ false
+
+
+
+
+ org.codehaus.mojo
+ exec-maven-plugin
+
+
+ check-shaded-content
+
+ exec
+
+ verify
+
+ ${basedir}/scripts/check_content.sh
+
+
+
+
+
+
+
+
+ shadeDep
+
+
+ !not-shadeDep
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-dependency-plugin
+
+
+ copy-dependencies
+
+ copy-dependencies
+
+ generate-resources
+
+ ${license.processing.dependencyJarsDir}
+ false
+ false
+ true
+
+
+
+
-
-
- io.dropwizard.metrics
- metrics-jvm
- 4.1.22
-
+
+
+ org.apache.maven.plugins
+ maven-dependency-plugin
+ 3.6.0
+
+ runtime
+ ${license.processing.dependencyListFile}
+
+
+
+
+ list
+
+ generate-resources
+
+
+
-
-
- io.dropwizard.metrics
- metrics-jmx
- 4.2.3
-
-
+
+
+ org.apache.maven.plugins
+ maven-shade-plugin
+ 3.4.1
+
+
+
+ net.snowflake:snowflake-jdbc
+ org.slf4j:slf4j-api
+ com.github.luben:zstd-jni
+
+
+
+
+ com.nimbusds
+ ${shadeBase}.com.nimbusds
+
+
+ org.bouncycastle
+ ${shadeBase}.org.bouncycastle
+
+
+ net.jcip
+ ${shadeBase}.net.jcip
+
+
+ net.minidev
+ ${shadeBase}.net.minidev
+
+
+ org.objectweb
+ ${shadeBase}.org.objectweb
+
+
+ com.fasterxml
+ ${shadeBase}.fasterxml
+
+
+ org.apache
+ ${shadeBase}.apache
+
+
+ org.xbill
+ ${shadeBase}.org.xbill
+
+
+ org.xerial
+ ${shadeBase}.org.xerial
+
+
+ io.netty
+ ${shadeBase}.io.netty
+
+
+ com.google
+ ${shadeBase}.com.google
+
+
+ com.github.benmanes
+ ${shadeBase}.com.github.benmanes
+
+
+
+ shaded.parquet
+ ${shadeBase}.shaded.parquet
+
+
+ org.codehaus
+ ${shadeBase}.org.codehaus
+
+
+ com.jcraft
+ ${shadeBase}.com.jcraft
+
+
+ org.eclipse
+ ${shadeBase}.org.eclipse
+
+
+ org.checkerframework
+ ${shadeBase}.org.checkerframework
+
+
+ com.codahale
+ ${shadeBase}.com.codahale
+
+
+ com.ctc
+ ${shadeBase}.com.ctc
+
+
+ com.thoughtworks
+ ${shadeBase}.com.thoughtworks
+
+
+
+ codegen
+ ${shadeBase}.codegen
+
+
+ javax.annotation
+ ${shadeBase}.javax.annotation
+
+
+ javax.activation
+ ${shadeBase}.javax.activation
+
+
+ io.airlift.compress
+ ${shadeBase}.io.airlift.compress
+
+
+
+
+ *:*
+
+ assets/org/apache/commons/math3/exception/util/LocalizedFormats_fr.properties
+ about.html
+ mozilla/public-suffix-list.txt
-
-
- shadeDep
-
-
- !not-shadeDep
-
-
-
-
-
-
- org.apache.maven.plugins
- maven-shade-plugin
- 3.1.1
-
-
-
- com.nimbusds
-
- ${shadeBase}.com.nimbusds
-
-
-
- net.jcip
- ${shadeBase}.net.jcip
-
-
- net.minidev
- ${shadeBase}.net.minidev
-
-
- org.objectweb
- ${shadeBase}.org.objectweb
-
-
- com.fasterxml
- ${shadeBase}.fasterxml
-
-
- org.apache
- ${shadeBase}.apache
-
-
-
-
- *:*
-
- META-INF/LICENSE*
- META-INF/NOTICE*
- META-INF/DEPENDENCIES
- META-INF/maven/**
- META-INF/services/com.fasterxml.*
- META-INF/*.xml
- META-INF/*.SF
- META-INF/*.DSA
- META-INF/*.RSA
-
-
-
- commons-logging:commons-logging
-
- org/apache/commons/logging/impl/AvalonLogger.class
-
-
-
- org.slf4j:slf4j-simple
-
- org/slf4j/**
-
-
-
-
-
-
-
-
-
-
-
- package
-
- shade
-
-
-
-
-
-
-
-
- ossrh-deploy
-
-
- ossrh-deploy
-
-
-
-
-
- org.apache.maven.plugins
- maven-gpg-plugin
- 1.6
-
-
- deploy
-
- sign-and-deploy-file
-
-
- target/${project.artifactId}.jar
- ossrh
- https://oss.sonatype.org/service/local/staging/deploy/maven2
- generated_public_pom.xml
- target/${project.artifactId}-javadoc.jar
- target/${project.artifactId}-sources.jar
- ${env.GPG_KEY_ID}
- ${env.GPG_KEY_PASSPHRASE}
-
-
-
-
-
-
-
-
- ghActionsIT
-
-
- ghActionsIT
-
-
-
-
-
- org.apache.maven.plugins
- maven-failsafe-plugin
-
-
-
- integration-test
-
-
-
- verify_github_actions_it
- verify
-
- verify
-
-
-
-
-
- org.jacoco
- jacoco-maven-plugin
- ${jacoco.version}
-
-
- pre-unit-test
-
- prepare-agent
-
-
- target/jacoco-ut.exec
-
-
-
- post-unit-test
- test
-
- report
-
-
- target/jacoco-ut.exec
- target/jacoco-ut
-
-
-
- pre-integration-test
- pre-integration-test
-
- prepare-agent
-
-
- target/jacoco-it.exec
-
-
-
- post-integration-test
- post-integration-test
-
- report
-
-
- target/jacoco-it.exec
- target/jacoco-it
-
-
-
-
- ${jacoco.skip.instrument}
-
-
-
-
-
-
+ META-INF/LICENSE*
+ META-INF/NOTICE*
+ META-INF/DEPENDENCIES
+ META-INF/maven/**
+ META-INF/services/com.fasterxml.*
+ META-INF/*.xml
+ META-INF/*.SF
+ META-INF/*.DSA
+ META-INF/*.RSA
+ google/protobuf/**/*.proto
+
+
+
+ commons-logging:commons-logging
+
+ org/apache/commons/logging/impl/AvalonLogger.class
+
+
+
+ org.slf4j:slf4j-simple
+
+ org/slf4j/**
+
+
+
+
+
+
+
+
+
+
+
+ shade
+
+ package
+
+
+
+
+
+ org.codehaus.mojo
+ exec-maven-plugin
+
+
+ process-third-party-licenses
+
+ exec
+
+ generate-resources
+
+ python3
+
+ ${project.basedir}/scripts/process_licenses.py
+ ${license.processing.dependencyListFile}
+ ${license.processing.dependencyJarsDir}
+ ${license.processing.targetDir}
+
+
+
+
+
+
+
+
+
+ ossrh-deploy
+
+
+ ossrh-deploy
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-gpg-plugin
+ 1.6
+
+
+
+ sign-and-deploy-file
+
+ deploy
+
+ target/${project.artifactId}.jar
+ ossrh
+ https://oss.sonatype.org/service/local/staging/deploy/maven2
+ generated_public_pom.xml
+ target/${project.artifactId}-javadoc.jar
+ target/${project.artifactId}-sources.jar
+ ${env.GPG_KEY_ID}
+ ${env.GPG_KEY_PASSPHRASE}
+
+
+
+
+
+
+
+
+ snapshot-deploy
+
+
+ snapshot-deploy
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-gpg-plugin
+ 1.6
+
+
+
+ sign-and-deploy-file
+
+ deploy
+
+ target/${project.artifactId}.jar
+ snapshot
+ https://nexus.int.snowflakecomputing.com/repository/Snapshots/
+ generated_public_pom.xml
+ target/${project.artifactId}-javadoc.jar
+ target/${project.artifactId}-sources.jar
+ ${env.GPG_KEY_ID}
+ ${env.GPG_KEY_PASSPHRASE}
+
+
+
+
+
+
+
+
+ ghActionsIT
+
+
+ ghActionsIT
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-failsafe-plugin
+
+
+
+ integration-test
+
+
+
+ verify_github_actions_it
+
+ verify
+
+ verify
+
+
+
+
+ org.jacoco
+ jacoco-maven-plugin
+ ${jacoco.version}
+
+ ${jacoco.skip.instrument}
+
+
+
+ pre-unit-test
+
+ prepare-agent
+
+
+ target/jacoco-ut.exec
+
+
+
+ post-unit-test
+
+ report
+
+ test
+
+ target/jacoco-ut.exec
+ target/jacoco-ut
+
+
+
+ pre-integration-test
+
+ prepare-agent
+
+ pre-integration-test
+
+ target/jacoco-it.exec
+
+
+
+ post-integration-test
+
+ report
+
+ post-integration-test
+
+ target/jacoco-it.exec
+ target/jacoco-it
+
+
+
+
+
+
+
+
diff --git a/profile.json.example b/profile.json.example
index 9f99ad0c1..46dbd42f0 100644
--- a/profile.json.example
+++ b/profile.json.example
@@ -1,15 +1,15 @@
{
"user": "user name",
- "url": "https://account_name.snowflakecomputing.com:443"
+ "url": "https://account_name.snowflakecomputing.com:443",
"account": "account_name",
"private_key": "PEM Private Key",
"port": 443,
"host": "account_name.snowflakecomputing.com",
"schema": "schema",
"scheme": "https",
- "database": "database name",
+ "database": "database_name",
"connect_string": "jdbc:snowflake://account_name.snowflakecomputing.com:443",
"ssl": "on",
- "warehouse": "warehouse name"
- "role": "accountadmin"
+ "warehouse": "warehouse_name",
+ "role": "role_name"
}
\ No newline at end of file
diff --git a/profile.json.gpg b/profile.json.gpg
index 11b20def7..12955621b 100644
Binary files a/profile.json.gpg and b/profile.json.gpg differ
diff --git a/profile_azure.json.gpg b/profile_azure.json.gpg
new file mode 100644
index 000000000..b388626dc
Binary files /dev/null and b/profile_azure.json.gpg differ
diff --git a/profile_gcp.json.gpg b/profile_gcp.json.gpg
new file mode 100644
index 000000000..cc45b109f
Binary files /dev/null and b/profile_gcp.json.gpg differ
diff --git a/profile_streaming.json.example b/profile_streaming.json.example
new file mode 100644
index 000000000..a94994b4b
--- /dev/null
+++ b/profile_streaming.json.example
@@ -0,0 +1,9 @@
+{
+ "user": "user name",
+ "url": "https://account_name.snowflakecomputing.com:443", // This is required, or it can be constructed from host, scheme and port
+ "private_key": "PEM Private Key",
+ "port": 443,
+ "host": "account_name.snowflakecomputing.com",
+ "scheme": "https",
+ "role": "role_name"
+}
\ No newline at end of file
diff --git a/public_pom.xml b/public_pom.xml
index 2edbfb272..fa8c7dd4a 100644
--- a/public_pom.xml
+++ b/public_pom.xml
@@ -34,4 +34,25 @@
scm:git:git://github.com/snowflakedb/snowflake-ingest-javahttp://github.com/snowflakedb/snowflake-ingest-java/tree/master
+
+
+
+ net.snowflake
+ snowflake-jdbc
+ 3.14.5
+ compile
+
+
+ org.slf4j
+ slf4j-api
+ 1.7.36
+ compile
+
+
+ com.github.luben
+ zstd-jni
+ 1.5.0-1
+ runtime
+
+
diff --git a/scripts/check_content.sh b/scripts/check_content.sh
index b25f24901..e4d3e2076 100755
--- a/scripts/check_content.sh
+++ b/scripts/check_content.sh
@@ -1,11 +1,40 @@
#!/bin/bash -e
# scripts used to check if all dependency is shaded into snowflake internal path
+# Do not add any additional exceptions into this script.
+# If this script fails the maven build, re-configure shading relocations in pom.xml.
set -o pipefail
DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null && pwd )"
-if jar tvf $DIR/../target/snowflake-ingest-sdk.jar | awk '{print $8}' | grep -v -E "^(net|com)/snowflake" | grep -v -E "(com|net)/\$" | grep -v -E "^META-INF" | grep -v -E "^mozilla" | grep -v mime.types | grep -v project.properties | grep -v -E "javax" | grep -v -E "^org/" | grep -v -E "^com/google" | grep -v -E "^com/sun" | grep -v "log4j.properties" | grep -v "git.properties" | grep -v "io/" | grep -v "codegen/" | grep -v "com/codahale/" | grep -v "com/ibm/" | grep -v "LICENSE"; then
- echo "[ERROR] Ingest SDK jar includes class not under the snowflake namespace"
+
+if jar tvf $DIR/../target/snowflake-ingest-sdk.jar | awk '{print $8}' | \
+ # Ignore directories
+ grep -v -E '/$' | \
+
+ # Snowflake top-level packages are allowed
+ grep -v -E "^net/snowflake/ingest" | \
+
+ # META-INF is allowed in the shaded JAR
+ grep -v -E "^META-INF" | \
+
+ # Files in the JAR root that are probably required
+ grep -v mime.types | \
+ grep -v project.properties | \
+ grep -v arrow-git.properties | \
+ grep -v core-default.xml | \
+ grep -v org.apache.hadoop.application-classloader.properties | \
+ grep -v common-version-info.properties | \
+ grep -v PropertyList-1.0.dtd | \
+ grep -v properties.dtd | \
+ grep -v parquet.thrift | \
+
+ # Native zstd libraries are allowed
+ grep -v -E '^darwin' | \
+ grep -v -E '^freebsd' | \
+ grep -v -E '^linux' | \
+ grep -v -E '^win' ; then
+
+ echo "[ERROR] JDBC jar includes class not under the snowflake namespace"
exit 1
fi
\ No newline at end of file
diff --git a/scripts/decrypt_secret.sh b/scripts/decrypt_secret.sh
index 05df864e6..37b923ff2 100755
--- a/scripts/decrypt_secret.sh
+++ b/scripts/decrypt_secret.sh
@@ -3,5 +3,21 @@
# Decrypt the file
# --batch to prevent interactive command --yes to assume "yes" for questions
-gpg --quiet --batch --yes --decrypt --passphrase="$DECRYPTION_PASSPHRASE" \
- --output profile.json ./profile.json.gpg
\ No newline at end of file
+# Input one argument
+snowflake_deployment=$1
+
+# Convert the input to uppercase
+uppercase_deployment=$(echo $snowflake_deployment | tr '[:lower:]' '[:upper:]')
+
+if [ $uppercase_deployment = 'AWS' ]; then
+ gpg --quiet --batch --yes --decrypt --passphrase="$DECRYPTION_PASSPHRASE" \
+ --output profile.json ./profile.json.gpg
+elif [ $uppercase_deployment = 'GCP' ]; then
+ gpg --quiet --batch --yes --decrypt --passphrase="$DECRYPTION_PASSPHRASE" \
+ --output profile.json ./profile_gcp.json.gpg
+elif [ $uppercase_deployment = 'AZURE' ]; then
+ gpg --quiet --batch --yes --decrypt --passphrase="$DECRYPTION_PASSPHRASE" \
+ --output profile.json ./profile_azure.json.gpg
+else
+ echo "Invalid deployment option. Please enter 'AWS', 'AZURE', or 'GCP'."
+fi
\ No newline at end of file
diff --git a/scripts/decrypt_secret_windows.ps1 b/scripts/decrypt_secret_windows.ps1
new file mode 100644
index 000000000..e0cd0b589
--- /dev/null
+++ b/scripts/decrypt_secret_windows.ps1
@@ -0,0 +1,25 @@
+param(
+ [string]$SnowflakeDeployment
+)
+
+# Convert the input to uppercase
+$UppercaseDeployment = $SnowflakeDeployment.ToUpper()
+
+# Decrypt the file based on the deployment option
+switch ($UppercaseDeployment) {
+ 'AWS' {
+ gpg --quiet --batch --yes --decrypt --passphrase="$env:DECRYPTION_PASSPHRASE" `
+ --output profile.json ./profile.json.gpg
+ }
+ 'GCP' {
+ gpg --quiet --batch --yes --decrypt --passphrase="$env:DECRYPTION_PASSPHRASE" `
+ --output profile.json ./profile_gcp.json.gpg
+ }
+ 'AZURE' {
+ gpg --quiet --batch --yes --decrypt --passphrase="$env:DECRYPTION_PASSPHRASE" `
+ --output profile.json ./profile_azure.json.gpg
+ }
+ default {
+ Write-Output "Invalid deployment option. Please enter 'AWS', 'AZURE', or 'GCP'."
+ }
+}
diff --git a/scripts/process_licenses.py b/scripts/process_licenses.py
new file mode 100644
index 000000000..bb43fbbf0
--- /dev/null
+++ b/scripts/process_licenses.py
@@ -0,0 +1,149 @@
+#!/usr/bin/env python
+
+# This script processes licenses of 3rd party dependencies and stores them in the JAR. The rules are:
+# 1. Dependencies, which contains a license file should be put into the shaded JAR as-is.
+# 2. Dependencies, which do not contain a license file should be mentioned in the file ADDITIONAL_LICENCES, together with the name of its license.
+#
+#
+# The script accepts the following arguments:
+# * DEPENDENCY_LIST_FILE_PATH
+# * Can be obtained by running mvn dependency:list -DincludeScope=runtime -DoutputFile=target/dependency_list.txt
+# * DEPENDENCIES_DIR
+# * Directory containging the JAR files of all SDK dependencies. Automatically generated by `mvn clean package` in target/dependency-jars
+# * TARGET_DIR
+# * Where to save all output, should be target/generated-sources/META-INF/third-party-licenses
+#
+#
+# Useful mvn commands:
+# * mvn clean license:add-third-party
+# * Generate dependency report; useful to find out licenses for dependencies that don't ship with a license file
+# * mvn dependency:list -DincludeScope=runtime -DoutputFile=target/dependency_list.txt
+# * Used as input of this script (DEPENDENCY_LIST_FILE_PATH)
+import sys
+from pathlib import Path
+from zipfile import ZipFile
+
+# License name constants
+APACHE_LICENSE = "Apache License 2.0"
+BSD_2_CLAUSE_LICENSE = "2-Clause BSD License"
+BSD_3_CLAUSE_LICENSE = "3-Clause BSD License"
+EDL_10_LICENSE = "EDL 1.0"
+MIT_LICENSE = "The MIT License"
+GO_LICENSE = "The Go license"
+BOUNCY_CASTLE_LICENSE = "Bouncy Castle Licence "
+
+# The SDK does not need to include licenses of dependencies, which aren't shaded
+IGNORED_DEPENDENCIES = {"net.snowflake:snowflake-jdbc", "org.slf4j:slf4j-api"}
+
+# List of dependencies, which don't ship with a license file.
+# Only add a new record here after verifying that the dependency JAR does not contain a license!
+ADDITIONAL_LICENSES_MAP = {
+ "com.google.code.findbugs:jsr305": APACHE_LICENSE,
+ "io.dropwizard.metrics:metrics-core": APACHE_LICENSE,
+ "io.dropwizard.metrics:metrics-jmx": APACHE_LICENSE,
+ "io.dropwizard.metrics:metrics-jvm": APACHE_LICENSE,
+ "com.google.guava:guava": APACHE_LICENSE,
+ "com.google.guava:failureaccess": APACHE_LICENSE,
+ "com.google.guava:listenablefuture": APACHE_LICENSE,
+ "com.google.errorprone:error_prone_annotations": APACHE_LICENSE,
+ "com.google.j2objc:j2objc-annotations": APACHE_LICENSE,
+ "com.nimbusds:nimbus-jose-jwt": APACHE_LICENSE,
+ "com.github.stephenc.jcip:jcip-annotations": APACHE_LICENSE,
+ "io.netty:netty-common": APACHE_LICENSE,
+ "com.google.re2j:re2j": GO_LICENSE,
+ "com.google.protobuf:protobuf-java": BSD_3_CLAUSE_LICENSE,
+ "com.google.code.gson:gson": APACHE_LICENSE,
+ "org.xerial.snappy:snappy-java": APACHE_LICENSE,
+ "org.apache.parquet:parquet-common": APACHE_LICENSE,
+ "org.apache.parquet:parquet-format-structures": APACHE_LICENSE,
+ "com.github.luben:zstd-jni": BSD_2_CLAUSE_LICENSE,
+ "io.airlift:aircompressor": APACHE_LICENSE,
+ "org.bouncycastle:bcpkix-jdk18on": BOUNCY_CASTLE_LICENSE,
+ "org.bouncycastle:bcutil-jdk18on": BOUNCY_CASTLE_LICENSE,
+ "org.bouncycastle:bcprov-jdk18on": BOUNCY_CASTLE_LICENSE,
+}
+
+
+def parse_cmdline_args():
+ if len(sys.argv) != 4:
+ raise Exception("usage: process_licenses.py DEPENDENCY_LIST_FILE_PATH DEPENDENCIES_DIR TARGET_DIR")
+ dependency_list_file_path = Path(sys.argv[1]).absolute()
+ dependencies_dir_path = Path(sys.argv[2]).absolute()
+ target_dir = Path(sys.argv[3]).absolute()
+
+ if not dependency_list_file_path.exists() or not dependency_list_file_path.is_file():
+ raise Exception(f"File {dependency_list_file_path} does not exist")
+
+ if not dependencies_dir_path.exists() or not dependencies_dir_path.is_dir():
+ raise Exception(f"Directory {dependencies_dir_path} does not exist")
+ return dependency_list_file_path, dependencies_dir_path, target_dir
+
+
+def main():
+ dependency_list_path, dependency_jars_path, target_dir = parse_cmdline_args()
+
+ dependency_count = 0
+ dependency_with_license_count = 0
+ dependency_without_license_count = 0
+ dependency_ignored_count = 0
+
+ missing_licenses_str = ""
+
+ target_dir.mkdir(parents=True, exist_ok=True)
+
+ with open(dependency_list_path, "r") as dependency_file_handle:
+ for line in dependency_file_handle.readlines():
+ line = line.strip()
+ if line == "" or line == "The following files have been resolved:":
+ continue
+ dependency_count += 1
+
+ # Line is a string like: "commons-codec:commons-codec:jar:1.15:compile -- module org.apache.commons.codec [auto]"
+ artifact_details = line.split()[0]
+ group_id, artifact_id, _, version, scope = artifact_details.split(":")
+ current_jar = Path(dependency_jars_path, f"{artifact_id}-{version}.jar")
+ if not current_jar.exists() and current_jar.is_file():
+ raise Exception(f"Expected JAR file does not exist: {current_jar}")
+ current_jar_as_zip = ZipFile(current_jar)
+
+ dependency_lookup_key = f"{group_id}:{artifact_id}"
+ if dependency_lookup_key in IGNORED_DEPENDENCIES:
+ dependency_ignored_count += 1
+ continue
+
+ license_found = False
+ for zip_info in current_jar_as_zip.infolist():
+ if zip_info.is_dir():
+ continue
+ if zip_info.filename in ("META-INF/LICENSE.txt", "META-INF/LICENSE", "META-INF/LICENSE.md"):
+ license_found = True
+ dependency_with_license_count += 1
+ # Extract license to the target directory
+ zip_info.filename = f"LICENSE_{group_id}__{artifact_id}"
+ current_jar_as_zip.extract(zip_info, target_dir)
+ break
+ if "license" in zip_info.filename.lower(): # Log potential license matches
+ print(f"Potential license match: {current_jar} {zip_info}")
+
+ if not license_found:
+ print(f"License not found {current_jar}; using value from ADDITIONAL_LICENSES_MAP")
+ license_name = ADDITIONAL_LICENSES_MAP.get(dependency_lookup_key)
+ if license_name:
+ dependency_without_license_count += 1
+ missing_licenses_str += f"{dependency_lookup_key}: {license_name}\n"
+ else:
+ raise Exception(f"The dependency {dependency_lookup_key} does not ship a license file, but neither is it not defined in ADDITIONAL_LICENSES_MAP")
+
+ with open(Path(target_dir, "ADDITIONAL_LICENCES"), "w") as additional_licenses_handle:
+ additional_licenses_handle.write(missing_licenses_str)
+
+ if dependency_count < 30:
+ raise Exception(f"Suspiciously low number of dependency JARs detected in {dependency_jars_path}: {dependency_count}")
+ print("License generation finished")
+ print(f"\tTotal dependencies: {dependency_count}")
+ print(f"\tTotal dependencies (with license): {dependency_with_license_count}")
+ print(f"\tTotal dependencies (without license): {dependency_without_license_count}")
+ print(f"\tIgnored dependencies: {dependency_ignored_count}")
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/run_gh_actions.sh b/scripts/run_gh_actions.sh
index 4ceacbe4b..49895010b 100755
--- a/scripts/run_gh_actions.sh
+++ b/scripts/run_gh_actions.sh
@@ -1,4 +1,9 @@
-mvn install -DskipTests=true --batch-mode --show-version
+#!/bin/bash -e
+
+set -o pipefail
+
+# Build and install shaded JAR first. check_content.sh runs here.
+mvn install -PcheckShadedContent -DskipTests=true --batch-mode --show-version
PARAMS=()
PARAMS+=("-DghActionsIT")
@@ -13,8 +18,3 @@ rc=$?
if [ $rc -ne 0 ] ; then
echo Could not perform mvn verify with parameters "${PARAMS[@]}", exit code [$rc]; exit $rc
fi
-
-# run whitesource
-echo ${PWD}
-chmod 755 ./scripts/run_whitesource_gh.sh
-./scripts/run_whitesource_gh.sh
\ No newline at end of file
diff --git a/scripts/run_whitesource_gh.sh b/scripts/run_whitesource_gh.sh
deleted file mode 100644
index 990352f5c..000000000
--- a/scripts/run_whitesource_gh.sh
+++ /dev/null
@@ -1,113 +0,0 @@
-#!/usr/bin/env bash
-#
-# Run whitesource for components which need versioning
-
-# SCAN_DIRECTORIES is a comma-separated list (as a string) of file paths which contain all source code and build artifacts for this project
-THIS_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
-CONNECTOR_DIR="$( dirname "${THIS_DIR}")"
-SCAN_DIRECTORIES="${CONNECTOR_DIR}"
-echo "SCAN_DIRECTORIES:"$SCAN_DIRECTORIES
-
-# If your PROD_BRANCH is not master, you can define it here based on the need
-PROD_BRANCH="master"
-
-PRODUCT_NAME="snowflake-ingest-java"
-
-PROJECT_VERSION=${GITHUB_SHA}
-
-BRANCH_OR_PR_NUMBER="$(echo "${GITHUB_REF}" | awk 'BEGIN { FS = "/" } ; { print $3 }')"
-
-echo "PROJECT_VERSION:"$PROJECT_VERSION
-echo "BRANCH_OR_PR_NUMBER:"$BRANCH_OR_PR_NUMBER
-
-# GITHUB_EVENT_NAME should either be 'push', or 'pull_request'
-if [[ "$GITHUB_EVENT_NAME" == "pull_request" ]]; then
- echo "[INFO] Pull Request"
- export PROJECT_NAME="PR-${BRANCH_OR_PR_NUMBER}"
-elif [[ "${BRANCH_OR_PR_NUMBER}" == "$PROD_BRANCH" ]]; then
- echo "[INFO] Production branch"
- export PROJECT_NAME="$PROD_BRANCH"
-else
- echo "[INFO] Non Production branch. Skipping wss..."
- export PROJECT_NAME=""
-fi
-
-echo "PROJECT_NAME:"$PROJECT_NAME
-
-[[ -z "$WHITESOURCE_API_KEY" ]] && echo "[WARNING] No WHITESOURCE_API_KEY is set. No WhiteSource scan will occurr." && exit 0
-
-#if [[ -z "${JOB_BASE_NAME}" ]]; then
-# echo "[ERROR] No JOB_BASE_NAME is set. Run this on Jenkins"
-# exit 0
-#fi
-
-# Download the latest whitesource unified agent to do the scanning if there is no existing one
-curl -LO https://github.com/whitesource/unified-agent-distribution/releases/latest/download/wss-unified-agent.jar
-if [[ ! -f "wss-unified-agent.jar" ]]; then
- echo "failed to download whitesource unified agent"
- # if you want to fail the build when failing to download whitesource scanning agent, please use exit 1
- # exit 1
-fi
-
-# whitesource will scan the folder and detect the corresponding configuration
-# configuration file wss-generated-file.config will be generated under ${SCAN_DIRECTORIES}
-# java -jar wss-unified-agent.jar -detect -d ${SCAN_DIRECTORIES}
-# SCAN_CONFIG="${SCAN_DIRECTORIES}/wss-generated-file.config"
-
-# SCAN_CONFIG is the path to your whitesource configuration file
-SCAN_CONFIG="scripts/wss-java-maven-agent.config"
-echo "SCAN_CONFIG:"$SCAN_CONFIG
-
-echo "[INFO] Running wss.sh for ${PRODUCT_NAME}-${PROJECT_NAME} under ${SCAN_DIRECTORIES}"
-
-if [[ "$PROJECT_NAME" == "$PROD_BRANCH" ]]; then
- echo "PRODUCTION"
- java -jar wss-unified-agent.jar -apiKey ${WHITESOURCE_API_KEY} \
- -c ${SCAN_CONFIG} \
- -project ${PROJECT_NAME} \
- -product ${PRODUCT_NAME} \
- -d ${SCAN_DIRECTORIES} \
- -projectVersion ${PROJECT_VERSION} \
- -offline true
- ERR=$?
- if [[ "$ERR" != "254" && "$ERR" != "0" ]]; then
- echo "failed to run wss for PROJECT_VERSION=${PROJECT_VERSION} in ${PROJECT_VERSION}..."
- exit 1
- fi
- java -jar wss-unified-agent.jar -apiKey ${WHITESOURCE_API_KEY} \
- -c ${SCAN_CONFIG} \
- -project ${PROJECT_NAME} \
- -product ${PRODUCT_NAME} \
- -projectVersion baseline \
- -requestFiles whitesource/update-request.txt
- ERR=$?
- if [[ "$ERR" != "254" && "$ERR" != "0" ]]; then
- echo "failed to run wss for PROJECT_VERSION=${PROJECT_VERSION} in baseline"
- exit 1
- fi
- java -jar wss-unified-agent.jar -apiKey ${WHITESOURCE_API_KEY} \
- -c ${SCAN_CONFIG} \
- -project ${PROJECT_NAME} \
- -product ${PRODUCT_NAME} \
- -projectVersion ${PROJECT_VERSION} \
- -requestFiles whitesource/update-request.txt
- ERR=$?
- if [[ "$ERR" != "254" && "$ERR" != "0" ]]; then
- echo "failed to run wss for PROJECT_VERSION=${PROJECT_VERSION} in ${PROJECT_VERSION}"
- exit 1
- fi
-elif [[ -n "$PROJECT_NAME" ]]; then
- echo "PR"
- java -jar wss-unified-agent.jar -apiKey ${WHITESOURCE_API_KEY} \
- -c ${SCAN_CONFIG} \
- -project ${PROJECT_NAME} \
- -product ${PRODUCT_NAME} \
- -d ${SCAN_DIRECTORIES} \
- -projectVersion ${PROJECT_VERSION}
- ERR=$?
- if [[ "$ERR" != "254" && "$ERR" != "0" ]]; then
- echo "failed to run wss for PROJECT_VERSION=${PROJECT_VERSION} in ${PROJECT_VERSION}"
- exit 1
- fi
-fi
-exit 0
diff --git a/scripts/update_project_version.py b/scripts/update_project_version.py
index 6ed3d47c1..c67d58802 100755
--- a/scripts/update_project_version.py
+++ b/scripts/update_project_version.py
@@ -1,10 +1,11 @@
-#!/usr/bin/env python
+#!/usr/bin/env python2
# -*- coding: utf-8 -*-
#
# Update project version
#
# arg1: pom.xml
# arg2: new version number
+# arg3: optional suffix for project name
#
import sys
import xml.etree.ElementTree as ET
@@ -12,6 +13,9 @@
pom_file = sys.argv[1]
version = sys.argv[2]
+suffix = None
+if len(sys.argv) > 3:
+ suffix = sys.argv[3]
def remove_namespace(doc, namespace):
"""
@@ -42,6 +46,8 @@ def handle_comment(self, data):
for e1 in root:
if type(e1.tag) is str and e1.tag.endswith('version'):
e1.text = version
+ if suffix and type(e1.tag) is str and e1.tag.endswith('artifactId'):
+ e1.text = e1.text + '-' + suffix
remove_namespace(root, u'http://maven.apache.org/POM/4.0.0')
diff --git a/scripts/wss-java-maven-agent.config b/scripts/wss-java-maven-agent.config
deleted file mode 100644
index 37c54929a..000000000
--- a/scripts/wss-java-maven-agent.config
+++ /dev/null
@@ -1,89 +0,0 @@
-###############################################################
-# WhiteSource Unified-Agent configuration file
-###############################################################
-# MAVEN SCAN MODE
-###############################################################
-
-apiKey=
-#userKey is required if WhiteSource administrator has enabled "Enforce user level access" option
-#userKey=
-#requesterEmail=user@provider.com
-
-projectName=
-projectVersion=
-projectToken=
-#projectTag= key:value
-
-productName=
-productVersion=
-productToken=
-
-#projectPerFolder=true
-#projectPerFolderIncludes=
-#projectPerFolderExcludes=
-
-#wss.connectionTimeoutMinutes=60
-wss.url=https://saas.whitesourcesoftware.com/agent
-
-############
-# Policies #
-############
-checkPolicies=true
-forceCheckAllDependencies=true
-forceUpdate=true
-forceUpdate.failBuildOnPolicyViolation=true
-#updateInventory=false
-
-###########
-# General #
-###########
-#offline=false
-#updateType=APPEND
-#ignoreSourceFiles=true
-#scanComment=
-#failErrorLevel=ALL
-#requireKnownSha1=false
-
-#generateProjectDetailsJson=true
-#generateScanReport=true
-#scanReportTimeoutMinutes=10
-#scanReportFilenameFormat=
-
-#analyzeFrameworks=true
-#analyzeFrameworksReference=
-
-#updateEmptyProject=false
-
-#log.files.level=
-#log.files.maxFileSize=
-#log.files.maxFilesCount=
-#log.files.path=
-
-########################################
-# Package Manager Dependency resolvers #
-########################################
-resolveAllDependencies=false
-
-maven.ignoredScopes=test provided
-maven.resolveDependencies=true
-maven.ignoreSourceFiles=false
-maven.aggregateModules=true
-maven.ignorePomModules=false
-maven.runPreStep=false
-maven.ignoreMvnTreeErrors=true
-maven.environmentPath=
-maven.m2RepositoryPath=
-maven.downloadMissingDependencies=false
-maven.additionalArguments=
-maven.projectNameFromDependencyFile=false
-
-###########################################################################################
-# Includes/Excludes Glob patterns - Please use only one exclude line and one include line #
-###########################################################################################
-includes=**/*.java **/*.jar
-
-#Exclude file extensions or specific directories by adding **/*. or **//**
-excludes=**/*sources.jar **/*javadoc.jar **/original-snowflake-ingest-sdk.jar
-
-case.sensitive.glob=false
-followSymbolicLinks=true
diff --git a/service.yml b/service.yml
new file mode 100644
index 000000000..e217c3de9
--- /dev/null
+++ b/service.yml
@@ -0,0 +1,24 @@
+name: snowflake-ingest-java
+lang: java
+lang_version: 8
+codeowners:
+ enable: true
+semaphore:
+ enable: true
+ pipeline_type: cp
+ branches:
+ - master
+ - main
+ - /^\d+\.\d+\.x$/
+ - /v\d+\.\d+\.\d+\-hotfix\-x/
+ - /^gh-readonly-queue.*/
+ - /.*-test-release$/
+git:
+ enable: true
+github:
+ enable: true
+code_artifact:
+ enable: true
+ package_paths:
+ - maven-snapshots/maven/io.confluent/snowflake-ingest-sdk
+ - maven-releases/maven/io.confluent/snowflake-ingest-sdk
diff --git a/snapshot_deploy.sh b/snapshot_deploy.sh
new file mode 100755
index 000000000..78a9bdc90
--- /dev/null
+++ b/snapshot_deploy.sh
@@ -0,0 +1,52 @@
+#!/bin/bash -e
+
+THIS_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+export GPG_KEY_ID="Snowflake Computing"
+export LDAP_USER="$INTERNAL_NEXUS_USERNAME"
+export LDAP_PWD="$INTERNAL_NEXUS_PASSWORD"
+
+if [ -z "$GPG_KEY_PASSPHRASE" ]; then
+ echo "[ERROR] GPG passphrase is not specified for $GPG_KEY_ID!"
+ exit 1
+fi
+
+if [ -z "$GPG_PRIVATE_KEY" ]; then
+ echo "[ERROR] GPG private key file is not specified!"
+ exit 1
+fi
+
+echo "[INFO] Import PGP Key"
+if ! gpg --list-secret-key | grep "$GPG_KEY_ID"; then
+ gpg --allow-secret-key-import --import "$GPG_PRIVATE_KEY"
+fi
+# copy the settings.xml template and inject credential information
+SNAPSHOT_DEPLOY_SETTINGS_XML="$THIS_DIR/mvn_settings_snapshot_deploy.xml"
+MVN_REPOSITORY_ID=snapshot
+
+cat > $SNAPSHOT_DEPLOY_SETTINGS_XML << SETTINGS.XML
+
+
+
+
+ $MVN_REPOSITORY_ID
+ $LDAP_USER
+ $LDAP_PWD
+
+
+
+SETTINGS.XML
+
+MVN_OPTIONS+=(
+ "--settings" "$SNAPSHOT_DEPLOY_SETTINGS_XML"
+ "--batch-mode"
+)
+
+echo "[Info] Sign package and deploy to staging area"
+project_version=$($THIS_DIR/scripts/get_project_info_from_pom.py $THIS_DIR/pom.xml version)
+$THIS_DIR/scripts/update_project_version.py public_pom.xml $project_version > generated_public_pom.xml
+
+mvn clean deploy ${MVN_OPTIONS[@]} -Dsnapshot-deploy
+
+rm $SNAPSHOT_DEPLOY_SETTINGS_XML
\ No newline at end of file
diff --git a/sonar-project.properties b/sonar-project.properties
new file mode 100644
index 000000000..c7eb73003
--- /dev/null
+++ b/sonar-project.properties
@@ -0,0 +1,9 @@
+### service-bot sonarqube plugin managed file
+sonar.coverage.exclusions=**/test/**/*,**/tests/**/*,**/mock/**/*,**/mocks/**/*,**/*mock*,**/*test*
+sonar.coverage.jacoco.xmlReportPaths=**/jacoco.xml
+sonar.cpd.exclusions=**/test/**/*,**/tests/**/*,**/mock/**/*,**/mocks/**/*,**/*mock*,**/*test*
+sonar.exclusions=**/*.pb.*,**/mk-include/**/*
+sonar.java.binaries=.
+sonar.language=java
+sonar.projectKey=snowflake-ingest-java
+sonar.sources=.
diff --git a/src/main/java/net/snowflake/ingest/SimpleIngestManager.java b/src/main/java/net/snowflake/ingest/SimpleIngestManager.java
index 1064c25ab..3a5fe71ed 100644
--- a/src/main/java/net/snowflake/ingest/SimpleIngestManager.java
+++ b/src/main/java/net/snowflake/ingest/SimpleIngestManager.java
@@ -15,27 +15,24 @@
import java.security.spec.InvalidKeySpecException;
import java.util.Collections;
import java.util.List;
-import java.util.Optional;
+import java.util.Properties;
import java.util.Set;
import java.util.UUID;
import java.util.stream.Collectors;
-import net.snowflake.ingest.connection.ClientStatusResponse;
-import net.snowflake.ingest.connection.ConfigureClientResponse;
+import net.snowflake.client.jdbc.internal.apache.http.client.methods.CloseableHttpResponse;
+import net.snowflake.client.jdbc.internal.apache.http.client.methods.HttpGet;
+import net.snowflake.client.jdbc.internal.apache.http.client.methods.HttpPost;
+import net.snowflake.client.jdbc.internal.apache.http.impl.client.CloseableHttpClient;
import net.snowflake.ingest.connection.HistoryRangeResponse;
import net.snowflake.ingest.connection.HistoryResponse;
import net.snowflake.ingest.connection.IngestResponse;
import net.snowflake.ingest.connection.IngestResponseException;
-import net.snowflake.ingest.connection.InsertFilesClientInfo;
import net.snowflake.ingest.connection.RequestBuilder;
import net.snowflake.ingest.connection.ServiceResponseHandler;
import net.snowflake.ingest.utils.BackOffException;
import net.snowflake.ingest.utils.HttpUtil;
import net.snowflake.ingest.utils.StagedFileWrapper;
import net.snowflake.ingest.utils.Utils;
-import org.apache.http.HttpResponse;
-import org.apache.http.client.HttpClient;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.client.methods.HttpPost;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -197,7 +194,8 @@ public SimpleIngestManager build() {
// logger object for this class
private static final Logger LOGGER = LoggerFactory.getLogger(SimpleIngestManager.class);
// HTTP Client that we use for sending requests to the service
- private HttpClient httpClient;
+ private CloseableHttpClient httpClient;
+ private boolean httpClientClosed = false;
// the account in which the user lives
private String account;
@@ -388,6 +386,27 @@ public SimpleIngestManager(
new RequestBuilder(account, user, keyPair, schemeName, hostName, port, userAgentSuffix);
}
+ /* Another flavor of constructor which supports userAgentSuffix and proxyProperties */
+ public SimpleIngestManager(
+ String account,
+ String user,
+ String pipe,
+ PrivateKey privateKey,
+ String schemeName,
+ String hostName,
+ int port,
+ String userAgentSuffix,
+ Properties proxyProperties)
+ throws NoSuchAlgorithmException, InvalidKeySpecException {
+ KeyPair keyPair = Utils.createKeyPairFromPrivateKey(privateKey);
+ // call our initializer method
+ init(account, user, pipe, keyPair, proxyProperties);
+
+ // make the request builder we'll use to build messages to the service
+ builder =
+ new RequestBuilder(account, user, keyPair, schemeName, hostName, port, userAgentSuffix);
+ }
+
// ========= Constructors End =========
/**
@@ -400,14 +419,43 @@ public SimpleIngestManager(
* @param keyPair the KeyPair we'll use to sign JWT tokens
*/
private void init(String account, String user, String pipe, KeyPair keyPair) {
+ init(account, user, pipe, keyPair, new Properties());
+ }
+
+ /**
+ * init - Does the basic work of constructing a SimpleIngestManager that is common across all
+ * constructors
+ *
+ * @param account The account into which we're loading
+ * @param user the user performing this load
+ * @param pipe the fully qualified name of pipe
+ * @param keyPair the KeyPair we'll use to sign JWT tokens
+ * @param proxyProperties proxy properties for HTTP client configuration
+ */
+ private void init(
+ String account, String user, String pipe, KeyPair keyPair, Properties proxyProperties) {
// set up our reference variables
this.account = account;
this.user = user;
this.pipe = pipe;
this.keyPair = keyPair;
- // make our client for sending requests
- httpClient = HttpUtil.getHttpClient();
+ // make our client for sending requests with proxy properties support
+ if (proxyProperties != null && !proxyProperties.isEmpty()) {
+ LOGGER.debug(
+ "Creating HTTP client for SimpleIngestManager with proxy properties for account: {},"
+ + " user: {}",
+ account,
+ user);
+ } else {
+ LOGGER.debug(
+ "Creating HTTP client for SimpleIngestManager without proxy properties for account: {},"
+ + " user: {}",
+ account,
+ user);
+ }
+
+ httpClient = HttpUtil.getHttpClient(account, proxyProperties);
// make the request builder we'll use to build messages to the service
}
@@ -524,53 +572,20 @@ public IngestResponse ingestFiles(List files, UUID requestId)
public IngestResponse ingestFiles(
List files, UUID requestId, boolean showSkippedFiles)
throws URISyntaxException, IOException, IngestResponseException, BackOffException {
- return ingestFiles(files, requestId, showSkippedFiles, null /* Client info is null */);
- }
-
- /**
- * ingestFiles With Client Info - synchronously sends a request to the ingest service to enqueue
- * these files along with clientSequencer and offSetToken.
- *
- *
OffsetToken will be atomically persisted on server(Snowflake) side along with files if the
- * clientSequencer added in this request matches with what Snowflake currently has.
- *
- *
If clientSequencers doesnt match, 400 response code is sent back and no files will be added.
- *
- * @param files - list of wrappers around filenames and sizes
- * @param requestId - a requestId that we'll use to label - if null, we generate one for the user
- * @param showSkippedFiles - a flag which returns the files that were skipped when set to true.
- * @param clientInfo - clientSequencer and offsetToken to pass along with files. Can be null.
- * @return an insert response from the server
- * @throws URISyntaxException - if the provided account name was illegal and caused a URI
- * construction failure
- * @throws IOException - if we have some other network failure
- * @throws IngestResponseException - if snowflake encountered error during ingest
- * @throws BackOffException - if we have a 503 response
- */
- public IngestResponse ingestFiles(
- List files,
- UUID requestId,
- boolean showSkippedFiles,
- InsertFilesClientInfo clientInfo)
- throws URISyntaxException, IOException, IngestResponseException, BackOffException {
-
// the request id we want to send with this payload
if (requestId == null || requestId.toString().isEmpty()) {
requestId = UUID.randomUUID();
}
HttpPost httpPostForIngestFile =
- builder.generateInsertRequest(
- requestId, pipe, files, showSkippedFiles, Optional.ofNullable(clientInfo));
+ builder.generateInsertRequest(requestId, pipe, files, showSkippedFiles);
// send the request and get a response....
- HttpResponse response = httpClient.execute(httpPostForIngestFile);
-
- LOGGER.info(
- "Attempting to unmarshall insert response - {}, with clientInfo - {}",
- response,
- clientInfo);
- return ServiceResponseHandler.unmarshallIngestResponse(response, requestId);
+ try (CloseableHttpResponse response = httpClient.execute(httpPostForIngestFile)) {
+ LOGGER.info("Attempting to unmarshall insert response - {}", response);
+ return ServiceResponseHandler.unmarshallIngestResponse(
+ response, requestId, httpClient, httpPostForIngestFile, builder);
+ }
}
/**
@@ -597,10 +612,11 @@ public HistoryResponse getHistory(UUID requestId, Integer recentSeconds, String
builder.generateHistoryRequest(requestId, pipe, recentSeconds, beginMark);
// send the request and get a response...
- HttpResponse response = httpClient.execute(httpGetHistory);
-
- LOGGER.info("Attempting to unmarshall history response - {}", response);
- return ServiceResponseHandler.unmarshallHistoryResponse(response, requestId);
+ try (CloseableHttpResponse response = httpClient.execute(httpGetHistory)) {
+ LOGGER.info("Attempting to unmarshall history response - {}", response);
+ return ServiceResponseHandler.unmarshallHistoryResponse(
+ response, requestId, httpClient, httpGetHistory, builder);
+ }
}
/**
@@ -626,57 +642,14 @@ public HistoryRangeResponse getHistoryRange(
requestId = UUID.randomUUID();
}
- HttpResponse response =
- httpClient.execute(
- builder.generateHistoryRangeRequest(
- requestId, pipe, startTimeInclusive, endTimeExclusive));
-
- LOGGER.info("Attempting to unmarshall history range response - {}", response);
- return ServiceResponseHandler.unmarshallHistoryRangeResponse(response, requestId);
- }
+ HttpGet request =
+ builder.generateHistoryRangeRequest(requestId, pipe, startTimeInclusive, endTimeExclusive);
- /**
- * Register a snowpipe client and returns the client sequencer
- *
- * @param requestId a UUID we use to label the request, if null, one is generated for the user
- * @return
- * @throws URISyntaxException - if the provided account name was illegal and caused a URI
- * construction failure
- * @throws IOException - if we have some other network failure
- * @throws IngestResponseException - if snowflake encountered error during ingest
- * @throws BackOffException - if we have a 503 response
- */
- public ConfigureClientResponse configureClient(UUID requestId)
- throws URISyntaxException, IOException, IngestResponseException, BackOffException {
- if (requestId == null || requestId.toString().isEmpty()) {
- requestId = UUID.randomUUID();
- }
- HttpResponse response =
- httpClient.execute(builder.generateConfigureClientRequest(requestId, pipe));
- LOGGER.info("Attempting to unmarshall configure client response - {}", response);
- return ServiceResponseHandler.unmarshallConfigureClientResponse(response, requestId);
- }
-
- /**
- * Get client status for snowpipe which contains offset token and client sequencer
- *
- * @param requestId a UUID we use to label the request, if null, one is generated for the user
- * @return
- * @throws URISyntaxException - if the provided account name was illegal and caused a URI
- * construction failure
- * @throws IOException - if we have some other network failure
- * @throws IngestResponseException - if snowflake encountered error during ingest
- * @throws BackOffException - if we have a 503 response
- */
- public ClientStatusResponse getClientStatus(UUID requestId)
- throws URISyntaxException, IOException, IngestResponseException, BackOffException {
- if (requestId == null || requestId.toString().isEmpty()) {
- requestId = UUID.randomUUID();
+ try (CloseableHttpResponse response = httpClient.execute(request)) {
+ LOGGER.info("Attempting to unmarshall history range response - {}", response);
+ return ServiceResponseHandler.unmarshallHistoryRangeResponse(
+ response, requestId, httpClient, request, builder);
}
- HttpResponse response =
- httpClient.execute(builder.generateGetClientStatusRequest(requestId, pipe));
- LOGGER.info("Attempting to unmarshall get client status response - {}", response);
- return ServiceResponseHandler.unmarshallGetClientStatus(response, requestId);
}
/**
@@ -687,6 +660,7 @@ public ClientStatusResponse getClientStatus(UUID requestId)
@Override
public void close() {
builder.closeResources();
+ HttpUtil.shutdownHttpConnectionManagerDaemonThread();
}
/* Used for testing */
diff --git a/src/main/java/net/snowflake/ingest/connection/ClientStatusResponse.java b/src/main/java/net/snowflake/ingest/connection/ClientStatusResponse.java
deleted file mode 100644
index 5a3cbdc24..000000000
--- a/src/main/java/net/snowflake/ingest/connection/ClientStatusResponse.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Copyright (c) 2021 Snowflake Computing Inc. All rights reserved.
- */
-package net.snowflake.ingest.connection;
-
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-
-/** ClientStatusResponse - response from a get client status request */
-@JsonIgnoreProperties(ignoreUnknown = true)
-public class ClientStatusResponse {
- private Long clientSequencer;
-
- private String offsetToken;
-
- @Override
- public String toString() {
- return "IngestResponse{"
- + "clientSequencer='"
- + clientSequencer
- + '\''
- + ", offsetToken='"
- + offsetToken
- + '\''
- + '}';
- }
-
- /**
- * unique identifier for the client
- *
- * @return clientSequencer
- */
- public Long getClientSequencer() {
- return clientSequencer;
- }
-
- /**
- * offset number for kafka connector
- *
- * @return offsetToken
- */
- public String getOffsetToken() {
- return offsetToken;
- }
-}
diff --git a/src/main/java/net/snowflake/ingest/connection/ConfigureClientResponse.java b/src/main/java/net/snowflake/ingest/connection/ConfigureClientResponse.java
deleted file mode 100644
index 144065c22..000000000
--- a/src/main/java/net/snowflake/ingest/connection/ConfigureClientResponse.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * Copyright (c) 2021 Snowflake Computing Inc. All rights reserved.
- */
-package net.snowflake.ingest.connection;
-
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-
-/** ConfigureClientResponse - response from a configure client request */
-@JsonIgnoreProperties(ignoreUnknown = true)
-public class ConfigureClientResponse {
- private Long clientSequencer;
-
- @Override
- public String toString() {
- return "IngestResponse{" + "clientSequencer='" + clientSequencer + '\'' + '}';
- }
-
- /**
- * unique identifier for the client
- *
- * @return clientSequencer
- */
- public Long getClientSequencer() {
- return clientSequencer;
- }
-}
diff --git a/src/main/java/net/snowflake/ingest/connection/InsertFilesClientInfo.java b/src/main/java/net/snowflake/ingest/connection/InsertFilesClientInfo.java
deleted file mode 100644
index 175f994d0..000000000
--- a/src/main/java/net/snowflake/ingest/connection/InsertFilesClientInfo.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * Copyright (c) 2021 Snowflake Computing Inc. All rights reserved.
- */
-package net.snowflake.ingest.connection;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- * Just a wrapper class which is serialised into REST request for insertFiles.
- *
- *
This is an optional field which can be passed with a required field "files" in the request
- * body.
- *
- *
Here is how the new request could look like
- *
- *
- */
-public class InsertFilesClientInfo {
- // FDB constant
- public static final int MAX_ALLOWED_OFFSET_TOKEN_BYTE_SIZE = 100000;
-
- // client sequencer which the caller thinks it currently has
- private final long clientSequencer;
-
- // offsetToken to atomically commit with files.
- private final String offsetToken;
-
- /** Constructor with both fields as required. */
- public InsertFilesClientInfo(long clientSequencer, String offsetToken) {
- if (clientSequencer < 0) {
- throw new IllegalArgumentException("ClientSequencer should be non negative.");
- }
- if (offsetToken == null || offsetToken.isEmpty()) {
- throw new IllegalArgumentException("OffsetToken can not be null or empty.");
- }
- /* Keeping some buffer */
- if (offsetToken.getBytes(StandardCharsets.UTF_8).length
- > MAX_ALLOWED_OFFSET_TOKEN_BYTE_SIZE - 1000) {
- throw new IllegalArgumentException("OffsetToken size too large.");
- }
- this.clientSequencer = clientSequencer;
- this.offsetToken = offsetToken;
- }
-
- /** Gets client Sequencer associated with this clientInfo record. */
- public long getClientSequencer() {
- return clientSequencer;
- }
-
- /** Gets offsetToken associated with this clientInfo record. */
- public String getOffsetToken() {
- return offsetToken;
- }
-
- /** Only printing clientSequencer */
- @Override
- public String toString() {
- return "InsertFilesClientInfo{" + "clientSequencer=" + clientSequencer + '}';
- }
-}
diff --git a/src/main/java/net/snowflake/ingest/connection/JWTManager.java b/src/main/java/net/snowflake/ingest/connection/JWTManager.java
new file mode 100644
index 000000000..691348a34
--- /dev/null
+++ b/src/main/java/net/snowflake/ingest/connection/JWTManager.java
@@ -0,0 +1,180 @@
+/*
+ * Copyright (c) 2012-2023 Snowflake Computing Inc. All rights reserved.
+ */
+
+package net.snowflake.ingest.connection;
+
+import com.nimbusds.jose.JOSEException;
+import com.nimbusds.jose.JWSAlgorithm;
+import com.nimbusds.jose.JWSHeader;
+import com.nimbusds.jose.JWSSigner;
+import com.nimbusds.jose.crypto.RSASSASigner;
+import com.nimbusds.jwt.JWTClaimsSet;
+import com.nimbusds.jwt.SignedJWT;
+import java.security.KeyPair;
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import net.snowflake.ingest.utils.Cryptor;
+
+/**
+ * This class manages creating and automatically renewing the JWT token
+ *
+ * @author obabarinsa
+ */
+public final class JWTManager extends SecurityManager {
+ // the token lifetime is 59 minutes
+ private static final float LIFETIME_IN_MINUTES = 59;
+
+ // the renewal time is 54 minutes
+ private static final int RENEWAL_INTERVAL_IN_MINUTES = 54;
+
+ private static final String TOKEN_TYPE = "KEYPAIR_JWT";
+
+ // The public - private key pair we're using to connect to the service
+ private final transient KeyPair keyPair;
+
+ // the token itself
+ private final AtomicReference token;
+
+ /**
+ * Creates a JWTManager entity for a given account, user and KeyPair with a specified time to
+ * renew the token
+ *
+ * @param accountName - the snowflake account name of this user
+ * @param username - the snowflake username of the current user
+ * @param keyPair - the public/private key pair we're using to connect
+ * @param timeTillRenewal - the time measure until we renew the token
+ * @param unit the unit by which timeTillRenewal is measured
+ * @param telemetryService reference to the telemetry service
+ */
+ JWTManager(
+ String accountName,
+ String username,
+ KeyPair keyPair,
+ int timeTillRenewal,
+ TimeUnit unit,
+ TelemetryService telemetryService) {
+ super(accountName, username, telemetryService);
+ // if any of our arguments are null, throw an exception
+ if (keyPair == null) {
+ throw new IllegalArgumentException();
+ }
+ token = new AtomicReference<>();
+
+ // we have to keep around the keys
+ this.keyPair = keyPair;
+
+ // generate our first token
+ refreshToken();
+
+ // schedule all future renewals
+ tokenRefresher.scheduleAtFixedRate(this::refreshToken, timeTillRenewal, timeTillRenewal, unit);
+ }
+
+ /**
+ * Creates a JWTManager entity for a given account, user and KeyPair with the default time to
+ * renew (RENEWAL_INTERVAL_IN_MINUTES minutes)
+ *
+ * @param accountName - the snowflake account name of this user
+ * @param username - the snowflake username of the current user
+ * @param keyPair - the public/private key pair we're using to connect
+ * @param telemetryService reference to the telemetry service
+ */
+ public JWTManager(
+ String accountName, String username, KeyPair keyPair, TelemetryService telemetryService) {
+ this(
+ accountName,
+ username,
+ keyPair,
+ RENEWAL_INTERVAL_IN_MINUTES,
+ TimeUnit.MINUTES,
+ telemetryService);
+ }
+
+ @Override
+ public String getToken() {
+ // if we failed to regenerate the token at some point, throw
+ if (refreshFailed.get()) {
+ LOGGER.error("getToken request failed due to token regeneration failure");
+ throw new SecurityException();
+ }
+
+ return token.get();
+ }
+
+ @Override
+ String getTokenType() {
+ return TOKEN_TYPE;
+ }
+
+ /** regenerateToken - Regenerates our Token given our current user, account and keypair */
+ @Override
+ void refreshToken() {
+ // create our JWT claim builder object
+ JWTClaimsSet.Builder builder = new JWTClaimsSet.Builder();
+
+ // get the subject to the fully qualified username
+ String subject = String.format("%s.%s", account, user);
+
+ // get the issuer
+ String publicKeyFPInJwt = calculatePublicKeyFp(keyPair);
+ String issuer = String.format("%s.%s.%s", account, user, publicKeyFPInJwt);
+
+ // iat set to now
+ Date iat = new Date(System.currentTimeMillis());
+
+ // expiration in 59 minutes
+ Date exp = new Date(iat.getTime() + 59 * 60 * 1000);
+
+ // build claim set
+ JWTClaimsSet claimsSet =
+ builder.issuer(issuer).subject(subject).issueTime(iat).expirationTime(exp).build();
+ LOGGER.debug("Creating new JWT with subject {} and issuer {}...", subject, issuer);
+
+ SignedJWT signedJWT = new SignedJWT(new JWSHeader(JWSAlgorithm.RS256), claimsSet);
+
+ JWSSigner signer = new RSASSASigner(this.keyPair.getPrivate());
+
+ String newToken;
+ try {
+ signedJWT.sign(signer);
+ newToken = signedJWT.serialize();
+ } catch (JOSEException e) {
+ refreshFailed.set(true);
+ LOGGER.error("Failed to regenerate token! Exception is as follows : {}", e.getMessage());
+ throw new SecurityException();
+ }
+
+ // atomically update the string
+ LOGGER.info("Successfully created new JWT");
+ token.set(newToken);
+
+ // Refresh the token used in the telemetry service as well
+ if (telemetryService != null) {
+ telemetryService.refreshToken(newToken);
+ }
+ }
+
+ /**
+ * Given a keypair
+ *
+ * @return the fingerprint of public key
+ *
The idea is to hash public key's raw bytes using SHA-256 and converts hash into a string
+ * using Base64 encoding.
+ */
+ private String calculatePublicKeyFp(KeyPair keyPair) {
+ // get the raw bytes of public key
+ byte[] publicKeyRawBytes = keyPair.getPublic().getEncoded();
+
+ // take sha256 on raw bytes and do base64 encode
+ publicKeyFingerPrint = String.format("SHA256:%s", Cryptor.sha256HashBase64(publicKeyRawBytes));
+ return publicKeyFingerPrint;
+ }
+
+ /** Currently, it only shuts down the instance of ExecutorService. */
+ @Override
+ public void close() {
+ tokenRefresher.shutdown();
+ }
+}
diff --git a/src/main/java/net/snowflake/ingest/connection/OAuthClient.java b/src/main/java/net/snowflake/ingest/connection/OAuthClient.java
new file mode 100644
index 000000000..bdff6c96e
--- /dev/null
+++ b/src/main/java/net/snowflake/ingest/connection/OAuthClient.java
@@ -0,0 +1,14 @@
+/*
+ * Copyright (c) 2023 Snowflake Computing Inc. All rights reserved.
+ */
+
+package net.snowflake.ingest.connection;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+/** Interface to perform token refresh request from {@link OAuthManager} */
+public interface OAuthClient {
+ AtomicReference getoAuthCredentialRef();
+
+ void refreshToken();
+}
diff --git a/src/main/java/net/snowflake/ingest/connection/OAuthCredential.java b/src/main/java/net/snowflake/ingest/connection/OAuthCredential.java
new file mode 100644
index 000000000..e222478d1
--- /dev/null
+++ b/src/main/java/net/snowflake/ingest/connection/OAuthCredential.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright (c) 2023 Snowflake Computing Inc. All rights reserved.
+ */
+
+package net.snowflake.ingest.connection;
+
+import java.util.Base64;
+
+/** This class hold credentials for OAuth authentication */
+public class OAuthCredential {
+ private static final String BASIC_AUTH_HEADER_PREFIX = "Basic ";
+ private final String authHeader;
+ private final String clientId;
+ private final String clientSecret;
+ private String accessToken;
+ private String refreshToken;
+ private int expiresIn;
+
+ public OAuthCredential(String clientId, String clientSecret, String refreshToken) {
+ this.authHeader =
+ BASIC_AUTH_HEADER_PREFIX
+ + Base64.getEncoder().encodeToString((clientId + ":" + clientSecret).getBytes());
+ this.clientId = clientId;
+ this.clientSecret = clientSecret;
+ this.refreshToken = refreshToken;
+ }
+
+ public String getAuthHeader() {
+ return authHeader;
+ }
+
+ public String getAccessToken() {
+ return accessToken;
+ }
+
+ public void setAccessToken(String accessToken) {
+ this.accessToken = accessToken;
+ }
+
+ public String getRefreshToken() {
+ return refreshToken;
+ }
+
+ public void setRefreshToken(String refreshToken) {
+ this.refreshToken = refreshToken;
+ }
+
+ public void setExpiresIn(int expiresIn) {
+ this.expiresIn = expiresIn;
+ }
+
+ public int getExpiresIn() {
+ return expiresIn;
+ }
+}
diff --git a/src/main/java/net/snowflake/ingest/connection/OAuthManager.java b/src/main/java/net/snowflake/ingest/connection/OAuthManager.java
new file mode 100644
index 000000000..fefb94299
--- /dev/null
+++ b/src/main/java/net/snowflake/ingest/connection/OAuthManager.java
@@ -0,0 +1,176 @@
+/*
+ * Copyright (c) 2012-2023 Snowflake Computing Inc. All rights reserved.
+ */
+
+package net.snowflake.ingest.connection;
+
+import java.util.concurrent.TimeUnit;
+import net.snowflake.client.jdbc.internal.apache.http.client.utils.URIBuilder;
+import net.snowflake.ingest.utils.Constants;
+import net.snowflake.ingest.utils.ErrorCode;
+import net.snowflake.ingest.utils.SFException;
+
+/** This class manages creating and automatically refresh the OAuth token */
+public final class OAuthManager extends SecurityManager {
+ private static final double DEFAULT_UPDATE_THRESHOLD_RATIO = 0.8;
+
+ // the endpoint for token request
+ private static final String TOKEN_REQUEST_ENDPOINT = "/oauth/token-request";
+
+ // Content type header to specify the encoding
+ private static final String OAUTH_CONTENT_TYPE_HEADER = "application/x-www-form-urlencoded";
+
+ // Properties for token refresh request
+ private static final String TOKEN_TYPE = "OAUTH";
+
+ // Update threshold, a floating-point value representing the ratio between the expiration time of
+ // an access token and the time needed to update it. It must be a value greater than 0 and less
+ // than 1. E.g. An access token with expires_in=600 and update_threshold_ratio=0.8 would be
+ // updated after 600*0.8 = 480.
+ private final double updateThresholdRatio;
+
+ private OAuthClient oAuthClient;
+
+ /**
+ * Creates a OAuthManager entity for a given account, user and OAuthCredential with default time
+ * to refresh the access token
+ *
+ * @param accountName - the snowflake account name of this user
+ * @param username - the snowflake username of the current user
+ * @param oAuthCredential - the OAuth credential we're using to connect
+ * @param baseURIBuilder - the uri builder with common scheme, host and port
+ * @param telemetryService reference to the telemetry service
+ */
+ OAuthManager(
+ String accountName,
+ String username,
+ OAuthCredential oAuthCredential,
+ URIBuilder baseURIBuilder,
+ TelemetryService telemetryService) {
+ this(
+ accountName,
+ username,
+ oAuthCredential,
+ baseURIBuilder,
+ DEFAULT_UPDATE_THRESHOLD_RATIO,
+ telemetryService);
+ }
+
+ /**
+ * Creates a OAuthManager entity for a given account, user and OAuthCredential with a specified
+ * time to refresh the token
+ *
+ * @param accountName - the snowflake account name of this user
+ * @param username - the snowflake username of the current user
+ * @param oAuthCredential - the OAuth credential we're using to connect
+ * @param baseURIBuilder - the uri builder with common scheme, host and port
+ * @param updateThresholdRatio - the ratio between the expiration time of a token and the time
+ * needed to refresh it.
+ * @param telemetryService reference to the telemetry service
+ */
+ OAuthManager(
+ String accountName,
+ String username,
+ OAuthCredential oAuthCredential,
+ URIBuilder baseURIBuilder,
+ double updateThresholdRatio,
+ TelemetryService telemetryService) {
+ // disable telemetry service until jdbc v3.13.34 is released
+ super(accountName, username, null);
+
+ // if any of our arguments are null, throw an exception
+ if (oAuthCredential == null || baseURIBuilder == null) {
+ throw new IllegalArgumentException();
+ }
+
+ // check if update threshold is within (0, 1)
+ if (updateThresholdRatio <= 0 || updateThresholdRatio >= 1) {
+ throw new IllegalArgumentException("updateThresholdRatio should fall in (0, 1)");
+ }
+ this.updateThresholdRatio = updateThresholdRatio;
+ this.oAuthClient = new SnowflakeOAuthClient(accountName, oAuthCredential, baseURIBuilder);
+
+ // generate our first token
+ refreshToken();
+ }
+
+ /**
+ * Creates a OAuthManager entity for a given account, user and OAuthClient with a specified time
+ * to refresh the token. Use for testing only.
+ *
+ * @param accountName - the snowflake account name of this user
+ * @param username - the snowflake username of the current user
+ * @param oAuthClient - the OAuth client to perform token refresh
+ * @param updateThresholdRatio - the ratio between the expiration time of a token and the time
+ * needed to refresh it.
+ */
+ public OAuthManager(
+ String accountName, String username, OAuthClient oAuthClient, double updateThresholdRatio) {
+ super(accountName, username, null);
+
+ this.updateThresholdRatio = updateThresholdRatio;
+ this.oAuthClient = oAuthClient;
+
+ refreshToken();
+ }
+
+ @Override
+ String getToken() {
+ if (refreshFailed.get()) {
+ throw new SecurityException("getToken request failed due to token refresh failure");
+ }
+ return oAuthClient.getoAuthCredentialRef().get().getAccessToken();
+ }
+
+ @Override
+ String getTokenType() {
+ return TOKEN_TYPE;
+ }
+
+ /**
+ * Set refresh token, this method is for refresh token renewal without requiring to restart
+ * client. This method only works when the authorization type is OAuth.
+ *
+ * @param refreshToken the new refresh token
+ */
+ void setRefreshToken(String refreshToken) {
+ oAuthClient.getoAuthCredentialRef().get().setRefreshToken(refreshToken);
+ }
+
+ /** refreshToken - Get new access token using refresh_token, client_id, client_secret */
+ @Override
+ void refreshToken() {
+ for (int retries = 0; retries < Constants.MAX_OAUTH_REFRESH_TOKEN_RETRY; retries++) {
+ try {
+ oAuthClient.refreshToken();
+
+ // Schedule next refresh
+ long nextRefreshDelay =
+ (long)
+ (oAuthClient.getoAuthCredentialRef().get().getExpiresIn()
+ * this.updateThresholdRatio);
+ tokenRefresher.schedule(this::refreshToken, nextRefreshDelay, TimeUnit.SECONDS);
+ LOGGER.debug(
+ "Refresh access token, next refresh is scheduled after {} seconds", nextRefreshDelay);
+
+ return;
+ } catch (SFException e1) {
+ // Exponential backoff retries
+ try {
+ Thread.sleep((1L << retries) * 1000L);
+ } catch (InterruptedException e2) {
+ throw new SFException(ErrorCode.OAUTH_REFRESH_TOKEN_ERROR, e2.getMessage());
+ }
+ }
+ }
+
+ refreshFailed.set(true);
+ throw new SecurityException("Fail to refresh access token");
+ }
+
+ /** Currently, it only shuts down the instance of ExecutorService. */
+ @Override
+ public void close() {
+ tokenRefresher.shutdown();
+ }
+}
diff --git a/src/main/java/net/snowflake/ingest/connection/RequestBuilder.java b/src/main/java/net/snowflake/ingest/connection/RequestBuilder.java
index 44d846b94..3fdb8a660 100644
--- a/src/main/java/net/snowflake/ingest/connection/RequestBuilder.java
+++ b/src/main/java/net/snowflake/ingest/connection/RequestBuilder.java
@@ -1,41 +1,32 @@
/*
- * Copyright (c) 2012-2017 Snowflake Computing Inc. All rights reserved.
+ * Copyright (c) 2012-2023 Snowflake Computing Inc. All rights reserved.
*/
package net.snowflake.ingest.connection;
+import static net.snowflake.ingest.utils.Constants.ENABLE_TELEMETRY_TO_SF;
import static net.snowflake.ingest.utils.Utils.isNullOrEmpty;
-import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.UncheckedIOException;
import java.net.URI;
import java.net.URISyntaxException;
-import java.net.URL;
-import java.nio.file.Files;
-import java.nio.file.Paths;
import java.security.KeyPair;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
import java.util.UUID;
-import net.snowflake.ingest.SimpleIngestManager;
+import net.snowflake.client.jdbc.internal.apache.http.HttpHeaders;
+import net.snowflake.client.jdbc.internal.apache.http.client.methods.HttpGet;
+import net.snowflake.client.jdbc.internal.apache.http.client.methods.HttpPost;
+import net.snowflake.client.jdbc.internal.apache.http.client.methods.HttpUriRequest;
+import net.snowflake.client.jdbc.internal.apache.http.client.utils.URIBuilder;
+import net.snowflake.client.jdbc.internal.apache.http.entity.ContentType;
+import net.snowflake.client.jdbc.internal.apache.http.entity.StringEntity;
+import net.snowflake.client.jdbc.internal.apache.http.impl.client.CloseableHttpClient;
import net.snowflake.ingest.utils.ErrorCode;
import net.snowflake.ingest.utils.SFException;
import net.snowflake.ingest.utils.SnowflakeURL;
import net.snowflake.ingest.utils.StagedFileWrapper;
-import org.apache.http.HttpHeaders;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.client.methods.HttpUriRequest;
-import org.apache.http.client.utils.URIBuilder;
-import org.apache.http.entity.ContentType;
-import org.apache.http.entity.StringEntity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,7 +43,7 @@ public class RequestBuilder {
/* Member variables Begin */
// the security manager which will handle token generation
- private SecurityManager securityManager;
+ private final SecurityManager securityManager;
// whatever the actual scheme is
private String scheme;
@@ -65,6 +56,9 @@ public class RequestBuilder {
private final String userAgentSuffix;
+ // Reference to the telemetry service
+ private final TelemetryService telemetryService;
+
/* Member variables End */
/* Static constants Begin */
@@ -87,13 +81,6 @@ public class RequestBuilder {
// the endpoint for history time range queries
private static final String HISTORY_RANGE_ENDPOINT_FORMAT = "/v1/data/pipes/%s/loadHistoryScan";
- // the endpoint for configure snowpipe client
- private static final String CONFIGURE_CLIENT_ENDPOINT_FORMAT =
- "/v1/data/pipes/%s/client/configure";
-
- // the endpoint for get snowpipe client status
- private static final String CLIENT_STATUS_ENDPOINT_FORMAT = "/v1/data/pipes/%s/client/status";
-
// optional number of max seconds of items to fetch(eg. in the last hour)
private static final String RECENT_HISTORY_IN_SECONDS = "recentSeconds";
@@ -118,16 +105,12 @@ public class RequestBuilder {
// and object mapper for all marshalling and unmarshalling
private static final ObjectMapper objectMapper = new ObjectMapper();
- private static final String RESOURCES_FILE = "project.properties";
-
- private static final Properties PROPERTIES = loadProperties();
-
private static final String USER_AGENT = getDefaultUserAgent();
// Don't change!
public static final String CLIENT_NAME = "SnowpipeJavaSDK";
- public static final String DEFAULT_VERSION = "1.0.1-beta";
+ public static final String DEFAULT_VERSION = "2.0.0";
public static final String JAVA_USER_AGENT = "JAVA";
@@ -145,10 +128,11 @@ public class RequestBuilder {
*
* @param accountName - the name of the Snowflake account to which we're connecting
* @param userName - the username of the entity loading files
- * @param keyPair - the Public/Private key pair we'll use to authenticate
+ * @param credential - the credential we'll use to authenticate
*/
- public RequestBuilder(String accountName, String userName, KeyPair keyPair) {
- this(accountName, userName, keyPair, DEFAULT_SCHEME, DEFAULT_HOST_SUFFIX, DEFAULT_PORT, null);
+ public RequestBuilder(String accountName, String userName, Object credential) {
+ this(
+ accountName, userName, credential, DEFAULT_SCHEME, DEFAULT_HOST_SUFFIX, DEFAULT_PORT, null);
}
/**
@@ -156,15 +140,17 @@ public RequestBuilder(String accountName, String userName, KeyPair keyPair) {
*
* @param accountName - the name of the Snowflake account to which we're connecting
* @param userName - the username of the entity loading files
- * @param keyPair - the Public/Private key pair we'll use to authenticate
+ * @param credential - the credential we'll use to authenticate
+ * @param userAgentSuffix - The suffix part of HTTP Header User-Agent
*/
public RequestBuilder(
String accountName,
String userName,
String hostName,
- KeyPair keyPair,
+ Object credential,
String userAgentSuffix) {
- this(accountName, userName, keyPair, DEFAULT_SCHEME, hostName, DEFAULT_PORT, userAgentSuffix);
+ this(
+ accountName, userName, credential, DEFAULT_SCHEME, hostName, DEFAULT_PORT, userAgentSuffix);
}
/**
@@ -173,7 +159,7 @@ public RequestBuilder(
*
* @param accountName - the account name to which we're connecting
* @param userName - for whom are we connecting?
- * @param keyPair - our auth credentials
+ * @param credential - the credential we'll use to authenticate
* @param schemeName - are we HTTP or HTTPS?
* @param hostName - the host for this snowflake instance
* @param portNum - the port number
@@ -181,11 +167,11 @@ public RequestBuilder(
public RequestBuilder(
String accountName,
String userName,
- KeyPair keyPair,
+ Object credential,
String schemeName,
String hostName,
int portNum) {
- this(accountName, userName, keyPair, schemeName, hostName, portNum, null);
+ this(accountName, userName, credential, schemeName, hostName, portNum, null);
}
/**
@@ -193,7 +179,7 @@ public RequestBuilder(
*
* @param accountName - the account name to which we're connecting
* @param userName - for whom are we connecting?
- * @param keyPair - our auth credentials
+ * @param credential - the credential we'll use to authenticate
* @param schemeName - are we HTTP or HTTPS?
* @param hostName - the host for this snowflake instance
* @param portNum - the port number
@@ -202,20 +188,91 @@ public RequestBuilder(
public RequestBuilder(
String accountName,
String userName,
- KeyPair keyPair,
+ Object credential,
String schemeName,
String hostName,
int portNum,
String userAgentSuffix) {
+ this(
+ accountName,
+ userName,
+ credential,
+ schemeName,
+ hostName,
+ portNum,
+ userAgentSuffix,
+ null,
+ null,
+ null);
+ }
+
+ /**
+ * RequestBuilder - constructor used by streaming ingest
+ *
+ * @param url - the Snowflake account to which we're connecting
+ * @param userName - the username of the entity loading files
+ * @param credential - the credential we'll use to authenticate
+ * @param httpClient - reference to the http client
+ * @param clientName - name of the client, used to uniquely identify a client if used
+ */
+ public RequestBuilder(
+ SnowflakeURL url,
+ String userName,
+ Object credential,
+ CloseableHttpClient httpClient,
+ String clientName) {
+ this(
+ url.getAccount(),
+ userName,
+ credential,
+ url.getScheme(),
+ url.getUrlWithoutPort(),
+ url.getPort(),
+ null,
+ null,
+ httpClient,
+ clientName);
+ }
+
+ /**
+ * RequestBuilder - this constructor is for testing purposes only
+ *
+ * @param accountName - the account name to which we're connecting
+ * @param userName - for whom are we connecting?
+ * @param credential - our auth credentials, either JWT key pair or OAuth credential
+ * @param schemeName - are we HTTP or HTTPS?
+ * @param hostName - the host for this snowflake instance
+ * @param portNum - the port number
+ * @param userAgentSuffix - The suffix part of HTTP Header User-Agent
+ * @param securityManager - The security manager for authentication
+ * @param httpClient - reference to the http client
+ * @param clientName - name of the client, used to uniquely identify a client if used
+ */
+ public RequestBuilder(
+ String accountName,
+ String userName,
+ Object credential,
+ String schemeName,
+ String hostName,
+ int portNum,
+ String userAgentSuffix,
+ SecurityManager securityManager,
+ CloseableHttpClient httpClient,
+ String clientName) {
// none of these arguments should be null
- if (accountName == null || userName == null || keyPair == null) {
+ if (accountName == null || userName == null || credential == null) {
throw new IllegalArgumentException();
}
- // create our security/token manager
- securityManager = new SecurityManager(accountName, userName, keyPair);
+ // Set up the telemetry service if needed
+ // TODO: SNOW-854272 Support telemetry service when using OAuth authentication
+ this.telemetryService =
+ ENABLE_TELEMETRY_TO_SF && credential instanceof KeyPair
+ ? new TelemetryService(
+ httpClient, clientName, schemeName + "://" + hostName + ":" + portNum)
+ : null;
- // stash references to the account and user name as well
+ // stash references to the account and username as well
String account = accountName.toUpperCase();
String user = userName.toUpperCase();
@@ -225,6 +282,27 @@ public RequestBuilder(
this.host = hostName;
this.userAgentSuffix = userAgentSuffix;
+ // create our security/token manager
+ if (securityManager == null) {
+ if (credential instanceof KeyPair) {
+ this.securityManager =
+ new JWTManager(accountName, userName, (KeyPair) credential, telemetryService);
+ } else if (credential instanceof OAuthCredential) {
+ this.securityManager =
+ new OAuthManager(
+ accountName,
+ userName,
+ (OAuthCredential) credential,
+ makeBaseURI(),
+ telemetryService);
+ } else {
+ throw new IllegalArgumentException(
+ "Credential should be instance of either KeyPair or OAuthCredential");
+ }
+ } else {
+ this.securityManager = securityManager;
+ }
+
LOGGER.info(
"Creating a RequestBuilder with arguments : "
+ "Account : {}, User : {}, Scheme : {}, Host : {}, Port : {}, userAgentSuffix: {}",
@@ -236,52 +314,6 @@ public RequestBuilder(
this.userAgentSuffix);
}
- /**
- * RequestBuilder - constructor used by streaming ingest
- *
- * @param url - the Snowflake account to which we're connecting
- * @param userName - the username of the entity loading files
- * @param keyPair - the Public/Private key pair we'll use to authenticate
- */
- public RequestBuilder(SnowflakeURL url, String userName, KeyPair keyPair) {
- this(
- url.getAccount(),
- userName,
- keyPair,
- url.getScheme(),
- url.getUrlWithoutPort(),
- url.getPort());
- }
-
- private static Properties loadProperties() {
- Properties properties = new Properties();
- properties.put("version", DEFAULT_VERSION);
-
- try {
- URL res = SimpleIngestManager.class.getClassLoader().getResource(RESOURCES_FILE);
- if (res == null) {
- throw new UncheckedIOException(new FileNotFoundException(RESOURCES_FILE));
- }
-
- URI uri;
- try {
- uri = res.toURI();
- } catch (URISyntaxException ex) {
- throw new IllegalArgumentException(ex);
- }
-
- try (InputStream is = Files.newInputStream(Paths.get(uri))) {
- properties.load(is);
- } catch (IOException ex) {
- throw new UncheckedIOException("Failed to load resource", ex);
- }
- } catch (Exception e) {
- LOGGER.warn("Could not read version info: " + e.toString());
- }
-
- return properties;
- }
-
/**
* Creates a string for user agent which should always be present in all requests to Snowflake
* (Snowpipe APIs)
@@ -293,8 +325,7 @@ private static Properties loadProperties() {
* @return the default agent string
*/
private static String getDefaultUserAgent() {
- final String clientVersion = PROPERTIES.getProperty("version");
- StringBuilder defaultUserAgent = new StringBuilder(CLIENT_NAME + "/" + clientVersion);
+ StringBuilder defaultUserAgent = new StringBuilder(CLIENT_NAME + "/" + DEFAULT_VERSION);
final String osInformation =
String.format(
@@ -310,48 +341,35 @@ private static String getDefaultUserAgent() {
// Add Java Version
final String javaVersion = System.getProperty("java.version");
- defaultUserAgent.append(JAVA_USER_AGENT + "/" + javaVersion);
+ defaultUserAgent.append(JAVA_USER_AGENT + "/").append(javaVersion);
+ String userAgent = defaultUserAgent.toString();
- return defaultUserAgent.toString();
+ LOGGER.info("Default user agent " + userAgent);
+ return userAgent;
}
private static String buildCustomUserAgent(String additionalUserAgentInfo) {
return USER_AGENT.trim() + " " + additionalUserAgentInfo;
}
+
/** A simple POJO for generating our POST body to the insert endpoint */
private static class IngestRequest {
// the list of files we're loading
private final List files;
- // additional info passed along with files which includes clientSequencer and offsetToken
- // can be null
- @JsonInclude(JsonInclude.Include.NON_NULL)
- private final InsertFilesClientInfo clientInfo;
-
- /** Constructor used when both files and clientInfo are passed in request */
- public IngestRequest(List files, InsertFilesClientInfo clientInfo) {
- this.files = files;
- this.clientInfo = clientInfo;
- }
-
/**
* Ctor used when only files is used in request body.
*
*
clientInfo will be defaulted to null
*/
public IngestRequest(List files) {
- this(files, null);
+ this.files = files;
}
/* Gets the list of files which were added in request */
public List getFiles() {
return files;
}
-
- /* Gets the clientInfo associated with files which was added in request */
- public InsertFilesClientInfo getClientInfo() {
- return clientInfo;
- }
}
/**
@@ -367,6 +385,17 @@ private URIBuilder makeBaseURI(UUID requestId) {
throw new IllegalArgumentException();
}
+ // construct the builder object without param
+ URIBuilder builder = makeBaseURI();
+
+ // set the request id
+ builder.setParameter(REQUEST_ID, requestId.toString());
+
+ return builder;
+ }
+
+ /** Construct a URIBuilder for common parts of request without any parameter */
+ private URIBuilder makeBaseURI() {
// construct the builder object
URIBuilder builder = new URIBuilder();
@@ -379,9 +408,6 @@ private URIBuilder makeBaseURI(UUID requestId) {
// set the port name
builder.setPort(port);
- // set the request id
- builder.setParameter(REQUEST_ID, requestId.toString());
-
return builder;
}
@@ -492,79 +518,14 @@ private URI makeHistoryRangeURI(
}
/**
- * Given a request UUID, and a fully qualified pipe name make a URI for configure snowpipe client
- * http://snowflakeURL{:PORT}/v1/data/pipes/{pipeName}/client/configure
- *
- *
Where snowflake URL can be an old url or new regionless URL.
- *
- *
Request Body is Empty
- *
- *
And our response looks like:
- *
- *
- * {
- * 'clientSequencer': LONG
- * }
- *
- *
- * @param requestId the label for this request
- * @param pipe the pipe name
- * @return configure snowpipe client URI
- * @throws URISyntaxException
- */
- private URI makeConfigureClientURI(UUID requestId, String pipe) throws URISyntaxException {
- if (pipe == null) {
- throw new IllegalArgumentException();
- }
- URIBuilder builder = makeBaseURI(requestId);
- builder.setPath(String.format(CONFIGURE_CLIENT_ENDPOINT_FORMAT, pipe));
- LOGGER.info("Final Configure Client URIBuilder - {}", builder);
- return builder.build();
- }
-
- /**
- * makeGetClientURI - Given a request UUID, and a fully qualified pipe name make a URI for getting
- * snowpipe client http://snowflakeURL{:PORT}/v1/data/pipes/{pipeName}/client/status
- *
- *
Where snowflake URL can be an old url or new regionless URL.
- *
- *
- *
- * @param requestId the label for this request
- * @param pipe the pipe name
- * @return get client URI
- * @throws URISyntaxException
- */
- private URI makeGetClientURI(UUID requestId, String pipe) throws URISyntaxException {
- if (pipe == null) {
- throw new IllegalArgumentException();
- }
- URIBuilder builder = makeBaseURI(requestId);
- builder.setPath(String.format(CLIENT_STATUS_ENDPOINT_FORMAT, pipe));
- LOGGER.info("Final Get Client URIBuilder - {}", builder);
- return builder.build();
- }
-
- /**
- * Given a list of files, and an optional clientInfo generate json string which later can be
- * passed in request body of insertFiles API
+ * Given a list of files, generate a json string which later can be passed in request body of
+ * insertFiles API
*
* @param files the list of files we want to send
- * @param clientInfo optional clientInfo which can be empty.
* @return the string json blob
* @throws IllegalArgumentException if files passed in is null
*/
- private String serializeInsertFilesRequest(
- List files, Optional clientInfo) {
+ private String serializeInsertFilesRequest(List files) {
// if the files argument is null, throw
if (files == null) {
LOGGER.info("Null files argument in RequestBuilder");
@@ -572,10 +533,7 @@ private String serializeInsertFilesRequest(
}
// create pojo
- IngestRequest ingestRequest =
- clientInfo
- .map(insertFilesClientInfo -> new IngestRequest(files, insertFilesClientInfo))
- .orElseGet(() -> new IngestRequest(files));
+ IngestRequest ingestRequest = new IngestRequest(files);
// serialize to a string
try {
@@ -605,21 +563,26 @@ private static void addUserAgent(HttpUriRequest request, String userAgentSuffix)
}
/**
- * addToken - adds a the JWT token to a request
+ * addToken - adds a token to a request
*
* @param request the URI request
- * @param token the token to add
*/
- private static void addToken(HttpUriRequest request, String token) {
- request.setHeader(HttpHeaders.AUTHORIZATION, BEARER_PARAMETER + token);
- request.setHeader(SF_HEADER_AUTHORIZATION_TOKEN_TYPE, JWT_TOKEN_TYPE);
+ public void addToken(HttpUriRequest request) {
+ request.setHeader(HttpHeaders.AUTHORIZATION, BEARER_PARAMETER + securityManager.getToken());
+ request.setHeader(SF_HEADER_AUTHORIZATION_TOKEN_TYPE, this.securityManager.getTokenType());
}
- private static void addHeaders(HttpUriRequest request, String token, String userAgentSuffix) {
+ /**
+ * addHeader - adds necessary header to a request
+ *
+ * @param request the URI request
+ * @param userAgentSuffix user agent suffix
+ */
+ private void addHeaders(HttpUriRequest request, String userAgentSuffix) {
addUserAgent(request, userAgentSuffix);
// Add the auth token
- addToken(request, token);
+ addToken(request);
// Add Accept header
request.setHeader(HttpHeaders.ACCEPT, HTTP_HEADER_CONTENT_TYPE_JSON);
@@ -639,28 +602,6 @@ private static void addHeaders(HttpUriRequest request, String token, String user
public HttpPost generateInsertRequest(
UUID requestId, String pipe, List files, boolean showSkippedFiles)
throws URISyntaxException {
- return generateInsertRequest(requestId, pipe, files, showSkippedFiles, Optional.empty());
- }
-
- /**
- * generateInsertRequest - given a pipe, list of files and clientInfo, make a request for the
- * insert endpoint
- *
- * @param requestId a UUID we will use to label this request
- * @param pipe a fully qualified pipe name
- * @param files a list of files
- * @param showSkippedFiles a boolean which returns skipped files when set to true
- * @param clientInfo
- * @return a post request with all the data we need
- * @throws URISyntaxException if the URI components provided are improper
- */
- public HttpPost generateInsertRequest(
- UUID requestId,
- String pipe,
- List files,
- boolean showSkippedFiles,
- Optional clientInfo)
- throws URISyntaxException {
// make the insert URI
URI insertURI = makeInsertURI(requestId, pipe, showSkippedFiles);
LOGGER.info("Created Insert Request : {} ", insertURI);
@@ -668,12 +609,11 @@ public HttpPost generateInsertRequest(
// Make the post request
HttpPost post = new HttpPost(insertURI);
- addHeaders(post, securityManager.getToken(), this.userAgentSuffix);
+ addHeaders(post, this.userAgentSuffix);
// the entity for the containing the json
final StringEntity entity =
- new StringEntity(
- serializeInsertFilesRequest(files, clientInfo), ContentType.APPLICATION_JSON);
+ new StringEntity(serializeInsertFilesRequest(files), ContentType.APPLICATION_JSON);
post.setEntity(entity);
return post;
@@ -697,7 +637,7 @@ public HttpGet generateHistoryRequest(
// make the get request
HttpGet get = new HttpGet(historyURI);
- addHeaders(get, securityManager.getToken(), this.userAgentSuffix);
+ addHeaders(get, this.userAgentSuffix);
return get;
}
@@ -723,7 +663,7 @@ public HttpGet generateHistoryRangeRequest(
HttpGet get = new HttpGet(historyRangeURI);
- addHeaders(get, securityManager.getToken(), this.userAgentSuffix /*User agent information*/);
+ addHeaders(get, this.userAgentSuffix /*User agent information*/);
return get;
}
@@ -751,7 +691,7 @@ public HttpPost generateStreamingIngestPostRequest(
// Make the post request
HttpPost post = new HttpPost(uri);
- addHeaders(post, securityManager.getToken(), this.userAgentSuffix /*User agent information*/);
+ addHeaders(post, this.userAgentSuffix /*User agent information*/);
// The entity for the containing the json
final StringEntity entity = new StringEntity(payload, ContentType.APPLICATION_JSON);
@@ -760,22 +700,6 @@ public HttpPost generateStreamingIngestPostRequest(
return post;
}
- /**
- * Given a requestId and a pipe, make a configure client request
- *
- * @param requestID a UUID we will use to label this request
- * @param pipe a fully qualified pipe name
- * @return configure client request
- * @throws URISyntaxException
- */
- public HttpPost generateConfigureClientRequest(UUID requestID, String pipe)
- throws URISyntaxException {
- URI configureClientURI = makeConfigureClientURI(requestID, pipe);
- HttpPost post = new HttpPost(configureClientURI);
- addHeaders(post, securityManager.getToken(), this.userAgentSuffix);
- return post;
- }
-
/**
* Generate post request for streaming ingest related APIs
*
@@ -798,19 +722,25 @@ public HttpPost generateStreamingIngestPostRequest(
}
/**
- * Given a requestId and a pipe, make a get client status request
+ * Set refresh token, this method is for refresh token renewal without requiring to restart
+ * client. This method only works when the authorization type is OAuth
*
- * @param requestID UUID
- * @param pipe a fully qualified pipe name
- * @return get client status request
- * @throws URISyntaxException
+ * @param refreshToken the new refresh token
*/
- public HttpGet generateGetClientStatusRequest(UUID requestID, String pipe)
- throws URISyntaxException {
- URI getClientStatusURI = makeGetClientURI(requestID, pipe);
- HttpGet get = new HttpGet(getClientStatusURI);
- addHeaders(get, securityManager.getToken(), this.userAgentSuffix);
- return get;
+ public void setRefreshToken(String refreshToken) {
+ if (securityManager instanceof OAuthManager) {
+ ((OAuthManager) securityManager).setRefreshToken(refreshToken);
+ }
+ }
+
+ /** Get authorization type */
+ public String getAuthType() {
+ return securityManager.getTokenType();
+ }
+
+ /** Refresh token manually */
+ public void refreshToken() {
+ securityManager.refreshToken();
}
/**
@@ -822,5 +752,13 @@ public HttpGet generateGetClientStatusRequest(UUID requestID, String pipe)
*/
public void closeResources() {
securityManager.close();
+ if (telemetryService != null) {
+ telemetryService.close();
+ }
+ }
+
+ /** Get the telemetry service */
+ public TelemetryService getTelemetryService() {
+ return telemetryService;
}
}
diff --git a/src/main/java/net/snowflake/ingest/connection/SecurityManager.java b/src/main/java/net/snowflake/ingest/connection/SecurityManager.java
index 7832ab88e..3fe89aeff 100644
--- a/src/main/java/net/snowflake/ingest/connection/SecurityManager.java
+++ b/src/main/java/net/snowflake/ingest/connection/SecurityManager.java
@@ -1,128 +1,71 @@
/*
- * Copyright (c) 2012-2017 Snowflake Computing Inc. All rights reserved.
+ * Copyright (c) 2012-2023 Snowflake Computing Inc. All rights reserved.
*/
package net.snowflake.ingest.connection;
-import com.nimbusds.jose.JOSEException;
-import com.nimbusds.jose.JWSAlgorithm;
-import com.nimbusds.jose.JWSHeader;
-import com.nimbusds.jose.JWSSigner;
-import com.nimbusds.jose.crypto.RSASSASigner;
-import com.nimbusds.jwt.JWTClaimsSet;
-import com.nimbusds.jwt.SignedJWT;
-import java.security.KeyPair;
-import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
-import net.snowflake.ingest.utils.Cryptor;
import net.snowflake.ingest.utils.ThreadFactoryUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * This class manages creating and automatically renewing the JWT token
+ * This class manages creating and automatically renewing the token
*
* @author obabarinsa
- * @since 1.8
*/
-final class SecurityManager implements AutoCloseable {
+abstract class SecurityManager implements AutoCloseable {
// the logger for SecurityManager
- private static final Logger LOGGER = LoggerFactory.getLogger(SecurityManager.class);
+ protected static final Logger LOGGER = LoggerFactory.getLogger(SecurityManager.class);
- // the token lifetime is 59 minutes
- private static final float LIFETIME = 59;
-
- // the renewal time is 54 minutes
- private static final int RENEWAL_INTERVAL = 54;
-
- // The public - private key pair we're using to connect to the service
- private transient KeyPair keyPair;
-
- // the name of the account on behalf of which we're connecting
- private String account;
+ protected final String account;
// Fingerprint of public key sent from client in jwt payload
- private String publicKeyFingerPrint;
+ protected String publicKeyFingerPrint;
// the name of the user who will be loading the files
- private String user;
+ protected final String user;
- // the token itself
- private AtomicReference token;
-
- // Did we fail to regenerate our token at some point?
- private AtomicBoolean regenFailed;
+ // Did we fail to refresh our token at some point?
+ protected final AtomicBoolean refreshFailed;
// Thread factory for daemon threads so that application can shutdown
final ThreadFactory tf = ThreadFactoryUtil.poolThreadFactory(getClass().getSimpleName(), true);
- // the thread we use for renewing all tokens
- private final ScheduledExecutorService keyRenewer = Executors.newScheduledThreadPool(1, tf);
+ // the thread we use for refresh tokens
+ protected ScheduledExecutorService tokenRefresher = Executors.newScheduledThreadPool(1, tf);
+
+ // Reference to the Telemetry Service in the client
+ protected final TelemetryService telemetryService;
/**
- * Creates a SecurityManager entity for a given account, user and KeyPair with a specified time to
- * renew the token
+ * Creates a SecurityManager entity for a given account
*
- * @param accountname - the snowflake account name of this user
+ * @param accountName - the snowflake account name of this user
* @param username - the snowflake username of the current user
- * @param keyPair - the public/private key pair we're using to connect
- * @param timeTillRenewal - the time measure until we renew the token
- * @param unit the unit by which timeTillRenewal is measured
*/
- SecurityManager(
- String accountname, String username, KeyPair keyPair, int timeTillRenewal, TimeUnit unit) {
+ SecurityManager(String accountName, String username, TelemetryService telemetryService) {
// if any of our arguments are null, throw an exception
- if (accountname == null || username == null || keyPair == null) {
+ if (accountName == null || username == null) {
throw new IllegalArgumentException();
}
- account = parseAccount(accountname);
+ account = parseAccount(accountName);
user = username.toUpperCase();
- // create our automatic reference to a string (our token)
- token = new AtomicReference<>();
-
// we haven't yet failed to regenerate our token
- regenFailed = new AtomicBoolean();
-
- // we have to keep around the keys
- this.keyPair = keyPair;
-
- // generate our first token
- regenerateToken();
+ this.refreshFailed = new AtomicBoolean();
- // schedule all future renewals
- keyRenewer.scheduleAtFixedRate(this::regenerateToken, timeTillRenewal, timeTillRenewal, unit);
+ this.telemetryService = telemetryService;
}
- /**
- * Creates a SecurityManager entity for a given account, user and KeyPair with the default time to
- * renew (54 minutes)
- *
- * @param accountname - the snowflake account name of this user
- * @param username - the snowflake username of the current user
- * @param keyPair - the public/private key pair we're using to connect
- */
- SecurityManager(String accountname, String username, KeyPair keyPair) {
- this(accountname, username, keyPair, RENEWAL_INTERVAL, TimeUnit.MINUTES);
+ /** Test only */
+ void setRefreshFailed(Boolean refreshFailed) {
+ this.refreshFailed.set(refreshFailed);
}
- /**
- * Trims an account name if it contains a "."
- *
- *
Snowflake's python connector trims an accountname if it contains a "."
- *
- * @param accountName given accountName in SimpleIngestManager's constructor
- * @return initial part of account name if originally it contained a ".", otherwise return the
- * same accountName.
- * @see Python
- * Connector
- */
private String parseAccount(final String accountName) {
String parseAccount = null;
if (accountName.contains(".")) {
@@ -136,94 +79,19 @@ private String parseAccount(final String accountName) {
return parseAccount.toUpperCase();
}
- /** regenerateToken - Regenerates our Token given our current user, account and keypair */
- private void regenerateToken() {
- // create our JWT claim builder object
- JWTClaimsSet.Builder builder = new JWTClaimsSet.Builder();
-
- // set the subject to the fully qualified username
- String subject = String.format("%s.%s", account, user);
- LOGGER.info("Creating Token with subject {}", subject);
-
- // set the issuer
- String publicKeyFPInJwt = calculatePublicKeyFp(keyPair);
- String issuer = String.format("%s.%s.%s", account, user, publicKeyFPInJwt);
- LOGGER.info("Creating Token with issuer {}", issuer);
-
- // iat set to now
- Date iat = new Date(System.currentTimeMillis());
-
- // expiration in 59 minutes
- Date exp = new Date(iat.getTime() + 59 * 60 * 1000);
-
- // build claim set
- JWTClaimsSet claimsSet =
- builder.issuer(issuer).subject(subject).issueTime(iat).expirationTime(exp).build();
-
- SignedJWT signedJWT = new SignedJWT(new JWSHeader(JWSAlgorithm.RS256), claimsSet);
-
- JWSSigner signer = new RSASSASigner(this.keyPair.getPrivate());
-
- String newToken;
- try {
- signedJWT.sign(signer);
- newToken = signedJWT.serialize();
- } catch (JOSEException e) {
- regenFailed.set(true);
- LOGGER.error("Failed to regenerate token! Exception is as follows : {}", e.getMessage());
- throw new SecurityException();
- }
-
- // atomically update the string
- LOGGER.info("Created new JWT");
- token.set(newToken);
- }
-
/**
* getToken - returns we've most recently generated
*
- * @return the string version of a valid JWT token
- * @throws SecurityException if we failed to regenerate a token since the last call
- */
- String getToken() {
- // if we failed to regenerate the token at some point, throw
- if (regenFailed.get()) {
- LOGGER.error("getToken request failed due to token regeneration failure");
- throw new SecurityException();
- }
-
- return token.get();
- }
-
- /* Only used in testing at the moment */
- String getAccount() {
- return this.account;
- }
-
- /**
- * Given a keypair
- *
- * @return the fingerprint of public key
- *
The idea is to hash public key's raw bytes using SHA-256 and converts hash into a string
- * using Base64 encoding.
+ * @return the string version of a valid token
+ * @throws SecurityException if we failed to regenerate or refresh a token since the last call
*/
- private String calculatePublicKeyFp(KeyPair keyPair) {
- // get the raw bytes of public key
- byte[] publicKeyRawBytes = keyPair.getPublic().getEncoded();
+ abstract String getToken();
- // take sha256 on raw bytes and do base64 encode
- publicKeyFingerPrint = String.format("SHA256:%s", Cryptor.sha256HashBase64(publicKeyRawBytes));
- return publicKeyFingerPrint;
- }
+ /** getTokenType - returns the token type, either "KEYPAIR_JWT" or "OAUTH" */
+ abstract String getTokenType();
- /** Only called by SecurityManagerTest */
- String getPublicKeyFingerPrint() {
- return publicKeyFingerPrint;
- }
+ /** refreshToken - regenerate or fetch a new token */
+ abstract void refreshToken();
- /** Currently, it only shuts down the instance of ExecutorService. */
- @Override
- public void close() {
- keyRenewer.shutdown();
- }
+ public abstract void close();
}
diff --git a/src/main/java/net/snowflake/ingest/connection/ServiceResponseHandler.java b/src/main/java/net/snowflake/ingest/connection/ServiceResponseHandler.java
index 8c9590a22..bf20afda8 100644
--- a/src/main/java/net/snowflake/ingest/connection/ServiceResponseHandler.java
+++ b/src/main/java/net/snowflake/ingest/connection/ServiceResponseHandler.java
@@ -8,11 +8,16 @@
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.util.UUID;
+import net.snowflake.client.jdbc.internal.apache.http.HttpEntity;
+import net.snowflake.client.jdbc.internal.apache.http.HttpResponse;
+import net.snowflake.client.jdbc.internal.apache.http.HttpStatus;
+import net.snowflake.client.jdbc.internal.apache.http.StatusLine;
+import net.snowflake.client.jdbc.internal.apache.http.client.methods.HttpGet;
+import net.snowflake.client.jdbc.internal.apache.http.client.methods.HttpPost;
+import net.snowflake.client.jdbc.internal.apache.http.client.methods.HttpUriRequest;
+import net.snowflake.client.jdbc.internal.apache.http.impl.client.CloseableHttpClient;
+import net.snowflake.client.jdbc.internal.apache.http.util.EntityUtils;
import net.snowflake.ingest.utils.BackOffException;
-import org.apache.http.HttpResponse;
-import org.apache.http.HttpStatus;
-import org.apache.http.StatusLine;
-import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -33,9 +38,8 @@ public enum ApiName {
INSERT_FILES("POST"),
INSERT_REPORT("GET"),
LOAD_HISTORY_SCAN("GET"),
- CLIENT_CONFIGURE("POST"),
- CLIENT_STATUS("GET"),
STREAMING_OPEN_CHANNEL("POST"),
+ STREAMING_DROP_CHANNEL("POST"),
STREAMING_CHANNEL_STATUS("POST"),
STREAMING_REGISTER_BLOB("POST"),
STREAMING_CLIENT_CONFIGURE("POST");
@@ -75,12 +79,20 @@ private static boolean isStatusOK(StatusLine statusLine) {
*
* @param response the HTTPResponse we want to distill into an IngestResponse
* @param requestId
+ * @param httpClient HttpClient for retries
+ * @param httpPostForIngestFile HttpRequest for retries
+ * @param builder RequestBuilder for retries
* @return An IngestResponse with all of the parsed out information
* @throws IOException if our entity is somehow corrupt or we can't get it
* @throws IngestResponseException - if we have an uncategorized network issue
* @throws BackOffException - if we have a 503 issue
*/
- public static IngestResponse unmarshallIngestResponse(HttpResponse response, UUID requestId)
+ public static IngestResponse unmarshallIngestResponse(
+ HttpResponse response,
+ UUID requestId,
+ CloseableHttpClient httpClient,
+ HttpPost httpPostForIngestFile,
+ RequestBuilder builder)
throws IOException, IngestResponseException, BackOffException {
// we can't unmarshall a null response
if (response == null) {
@@ -89,10 +101,11 @@ public static IngestResponse unmarshallIngestResponse(HttpResponse response, UUI
}
// handle the exceptional status code
- handleExceptionalStatus(response, requestId, ApiName.INSERT_FILES);
+ response =
+ handleExceptionalStatus(
+ response, requestId, ApiName.INSERT_FILES, httpClient, httpPostForIngestFile, builder);
- // grab the response entity
- String blob = EntityUtils.toString(response.getEntity());
+ String blob = consumeAndReturnResponseEntityAsString(response.getEntity());
// Read out the blob entity into a class
return mapper.readValue(blob, IngestResponse.class);
@@ -104,12 +117,20 @@ public static IngestResponse unmarshallIngestResponse(HttpResponse response, UUI
*
* @param response the HttpResponse object we are trying to deserialize
* @param requestId
+ * @param httpClient HttpClient for retries
+ * @param httpGetHistory HttpRequest for retries
+ * @param builder RequestBuilder for retries
* @return a HistoryResponse with all the parsed out information
* @throws IOException if our entity is somehow corrupt or we can't get it
* @throws IngestResponseException - if we have an uncategorized network issue
* @throws BackOffException - if we have a 503 issue
*/
- public static HistoryResponse unmarshallHistoryResponse(HttpResponse response, UUID requestId)
+ public static HistoryResponse unmarshallHistoryResponse(
+ HttpResponse response,
+ UUID requestId,
+ CloseableHttpClient httpClient,
+ HttpGet httpGetHistory,
+ RequestBuilder builder)
throws IOException, IngestResponseException, BackOffException {
// we can't unmarshall a null response
if (response == null) {
@@ -118,10 +139,11 @@ public static HistoryResponse unmarshallHistoryResponse(HttpResponse response, U
}
// handle the exceptional status code
- handleExceptionalStatus(response, requestId, ApiName.INSERT_REPORT);
+ response =
+ handleExceptionalStatus(
+ response, requestId, ApiName.INSERT_REPORT, httpClient, httpGetHistory, builder);
- // grab the string version of the response entity
- String blob = EntityUtils.toString(response.getEntity());
+ String blob = consumeAndReturnResponseEntityAsString(response.getEntity());
// read out our blob into a pojo
return mapper.readValue(blob, HistoryResponse.class);
@@ -132,13 +154,20 @@ public static HistoryResponse unmarshallHistoryResponse(HttpResponse response, U
*
* @param response the HttpResponse object we are trying to deserialize
* @param requestId
+ * @param httpClient HttpClient for retries
+ * @param request HttpRequest for retries
+ * @param builder HttpBuilder for retries
* @return HistoryRangeResponse
* @throws IOException if our entity is somehow corrupt or we can't get it
* @throws IngestResponseException - if we have an uncategorized network issue
* @throws BackOffException - if we have a 503 issue
*/
public static HistoryRangeResponse unmarshallHistoryRangeResponse(
- HttpResponse response, UUID requestId)
+ HttpResponse response,
+ UUID requestId,
+ CloseableHttpClient httpClient,
+ HttpGet request,
+ RequestBuilder builder)
throws IOException, IngestResponseException, BackOffException {
// we can't unmarshall a null response
@@ -148,73 +177,15 @@ public static HistoryRangeResponse unmarshallHistoryRangeResponse(
}
// handle the exceptional status code
- handleExceptionalStatus(response, requestId, ApiName.LOAD_HISTORY_SCAN);
-
- // grab the string version of the response entity
- String blob = EntityUtils.toString(response.getEntity());
+ response =
+ handleExceptionalStatus(
+ response, requestId, ApiName.LOAD_HISTORY_SCAN, httpClient, request, builder);
+ String blob = consumeAndReturnResponseEntityAsString(response.getEntity());
// read out our blob into a pojo
return mapper.readValue(blob, HistoryRangeResponse.class);
}
- /**
- * unmarshallConfigureClientResponse - Given an HttpResponse object, attempts to deserialize it
- * into a ConfigureClientResponse
- *
- * @param response HttpResponse
- * @param requestId
- * @return ConfigureClientResponse
- * @throws IOException if our entity is somehow corrupt or we can't get it
- * @throws IngestResponseException - if we have an uncategorized network issue
- * @throws BackOffException - if we have a 503 issue
- */
- public static ConfigureClientResponse unmarshallConfigureClientResponse(
- HttpResponse response, UUID requestId)
- throws IOException, IngestResponseException, BackOffException {
- if (response == null) {
- LOGGER.warn("Null response passed to unmarshallConfigureClientResponse");
- throw new IllegalArgumentException();
- }
-
- // handle the exceptional status code
- handleExceptionalStatus(response, requestId, ApiName.CLIENT_CONFIGURE);
-
- // grab the string version of the response entity
- String blob = EntityUtils.toString(response.getEntity());
-
- // read out our blob into a pojo
- return mapper.readValue(blob, ConfigureClientResponse.class);
- }
-
- /**
- * unmarshallGetClientStatus - Given an HttpResponse object, attempts to deserialize it into a
- * ClientStatusResponse
- *
- * @param response HttpResponse
- * @param requestId
- * @return ClientStatusResponse
- * @throws IOException if our entity is somehow corrupt or we can't get it
- * @throws IngestResponseException - if we have an uncategorized network issue
- * @throws BackOffException - if we have a 503 issue
- */
- public static ClientStatusResponse unmarshallGetClientStatus(
- HttpResponse response, UUID requestId)
- throws IOException, IngestResponseException, BackOffException {
- if (response == null) {
- LOGGER.warn("Null response passed to unmarshallClientStatusResponse");
- throw new IllegalArgumentException();
- }
-
- // handle the exceptional status code
- handleExceptionalStatus(response, requestId, ApiName.CLIENT_STATUS);
-
- // grab the string version of the response entity
- String blob = EntityUtils.toString(response.getEntity());
-
- // read out our blob into a pojo
- return mapper.readValue(blob, ClientStatusResponse.class);
- }
-
/**
* unmarshallStreamingIngestResponse Given an HttpResponse object - attempts to deserialize it
* into an Object based on input type
@@ -227,7 +198,12 @@ public static ClientStatusResponse unmarshallGetClientStatus(
* @throws IngestResponseException if received an exceptional status code
*/
public static T unmarshallStreamingIngestResponse(
- HttpResponse response, Class valueType, ApiName apiName)
+ HttpResponse response,
+ Class valueType,
+ ApiName apiName,
+ CloseableHttpClient httpClient,
+ HttpUriRequest request,
+ RequestBuilder requestBuilder)
throws IOException, IngestResponseException {
// We can't unmarshall a null response
if (response == null) {
@@ -236,10 +212,11 @@ public static T unmarshallStreamingIngestResponse(
}
// Handle the exceptional status code
- handleExceptionalStatus(response, null, apiName);
+ response =
+ handleExceptionalStatus(response, null, apiName, httpClient, request, requestBuilder);
// Grab the string version of the response entity
- String blob = EntityUtils.toString(response.getEntity());
+ String blob = consumeAndReturnResponseEntityAsString(response.getEntity());
// Read out our blob into a pojo
return mapper.readValue(blob, valueType);
@@ -250,35 +227,73 @@ public static T unmarshallStreamingIngestResponse(
*
* @param response HttpResponse
* @param requestId
+ * @param apiName enum to represent the corresponding api name
+ * @param httpClient HttpClient for retries
+ * @param request HttpRequest for retries
+ * @param requestBuilder RequestBuilder for retries
+ * @return modified HttpResponse
* @throws IOException if our entity is somehow corrupt or we can't get it
* @throws IngestResponseException - for all other non OK status
* @throws BackOffException - if we have a 503 issue
*/
- private static void handleExceptionalStatus(
- HttpResponse response, UUID requestId, ApiName apiName)
+ private static HttpResponse handleExceptionalStatus(
+ HttpResponse response,
+ UUID requestId,
+ ApiName apiName,
+ CloseableHttpClient httpClient,
+ HttpUriRequest request,
+ RequestBuilder requestBuilder)
throws IOException, IngestResponseException, BackOffException {
- StatusLine statusLine = response.getStatusLine();
- if (!isStatusOK(statusLine)) {
+ if (!isStatusOK(response.getStatusLine())) {
+ StatusLine statusLine = response.getStatusLine();
+ LOGGER.warn(
+ "{} Status hit from {}, requestId:{}",
+ statusLine.getStatusCode(),
+ apiName,
+ requestId == null ? "" : requestId.toString());
+
// if we have a 503 exception throw a backoff
switch (statusLine.getStatusCode()) {
// If we have a 503, BACKOFF
case HttpStatus.SC_SERVICE_UNAVAILABLE:
- LOGGER.warn(
- "503 Status hit from {}, backoff, requestId:{}",
- apiName,
- requestId == null ? "" : requestId.toString());
throw new BackOffException();
+ case HttpStatus.SC_UNAUTHORIZED:
+ LOGGER.warn("Authorization failed, refreshing Token succeeded, retry");
+ requestBuilder.refreshToken();
+ requestBuilder.addToken(request);
+ response = httpClient.execute(request);
+ if (!isStatusOK(response.getStatusLine())) {
+ throw new SecurityException("Authorization failed after retry");
+ }
+ break;
default:
- LOGGER.error(
- "Exceptional Status Code from {}: {}, requestId:{}",
- apiName,
- statusLine.getStatusCode(),
- requestId == null ? "" : requestId.toString());
- String blob = EntityUtils.toString(response.getEntity());
+ String blob = consumeAndReturnResponseEntityAsString(response.getEntity());
throw new IngestResponseException(
statusLine.getStatusCode(),
IngestResponseException.IngestExceptionBody.parseBody(blob));
}
}
+ return response;
+ }
+
+ /**
+ * Consumes the HttpEntity as mentioned in HttpClient Docs
+ *
+ *
Also returns the string version of this entity.
+ *
+ * @param httpResponseEntity the response entity obtained after successfully calling associated
+ * Rest APIs
+ * @return String version of this http response which will be later used to deserialize into
+ * respective Response Object
+ * @throws IOException if parsing error
+ */
+ private static String consumeAndReturnResponseEntityAsString(HttpEntity httpResponseEntity)
+ throws IOException {
+ // grab the string version of the response entity
+ String responseEntityAsString = EntityUtils.toString(httpResponseEntity);
+
+ EntityUtils.consumeQuietly(httpResponseEntity);
+ return responseEntityAsString;
}
}
diff --git a/src/main/java/net/snowflake/ingest/connection/SnowflakeOAuthClient.java b/src/main/java/net/snowflake/ingest/connection/SnowflakeOAuthClient.java
new file mode 100644
index 000000000..a90540659
--- /dev/null
+++ b/src/main/java/net/snowflake/ingest/connection/SnowflakeOAuthClient.java
@@ -0,0 +1,138 @@
+/*
+ * Copyright (c) 2023 Snowflake Computing Inc. All rights reserved.
+ */
+
+package net.snowflake.ingest.connection;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import net.snowflake.client.jdbc.internal.apache.http.HttpHeaders;
+import net.snowflake.client.jdbc.internal.apache.http.client.methods.CloseableHttpResponse;
+import net.snowflake.client.jdbc.internal.apache.http.client.methods.HttpPost;
+import net.snowflake.client.jdbc.internal.apache.http.client.methods.HttpUriRequest;
+import net.snowflake.client.jdbc.internal.apache.http.client.utils.URIBuilder;
+import net.snowflake.client.jdbc.internal.apache.http.entity.ContentType;
+import net.snowflake.client.jdbc.internal.apache.http.entity.StringEntity;
+import net.snowflake.client.jdbc.internal.apache.http.impl.client.CloseableHttpClient;
+import net.snowflake.client.jdbc.internal.apache.http.util.EntityUtils;
+import net.snowflake.client.jdbc.internal.google.api.client.http.HttpStatusCodes;
+import net.snowflake.client.jdbc.internal.google.gson.JsonObject;
+import net.snowflake.client.jdbc.internal.google.gson.JsonParser;
+import net.snowflake.ingest.utils.ErrorCode;
+import net.snowflake.ingest.utils.HttpUtil;
+import net.snowflake.ingest.utils.SFException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/*
+ * Implementation of Snowflake OAuth Client, used for refreshing an OAuth access token.
+ */
+public class SnowflakeOAuthClient implements OAuthClient {
+
+ static final Logger LOGGER = LoggerFactory.getLogger(SnowflakeOAuthClient.class);
+ private static final String TOKEN_REQUEST_ENDPOINT = "/oauth/token-request";
+
+ // Content type header to specify the encoding
+ private static final String OAUTH_CONTENT_TYPE_HEADER = "application/x-www-form-urlencoded";
+ private static final String GRANT_TYPE_PARAM = "grant_type";
+ private static final String ACCESS_TOKEN = "access_token";
+ private static final String REFRESH_TOKEN = "refresh_token";
+ private static final String EXPIRES_IN = "expires_in";
+
+ // OAuth credential
+ private final AtomicReference oAuthCredential;
+
+ // exact uri for token request
+ private final URI tokenRequestURI;
+
+ // Http client for submitting token refresh request
+ private final CloseableHttpClient httpClient;
+
+ /**
+ * Creates an SnowflakeOAuthClient given account, credential and base uri
+ *
+ * @param accountName - the snowflake account name of this user
+ * @param oAuthCredential - the OAuth credential we're using to connect
+ * @param baseURIBuilder - the uri builder with common scheme, host and port
+ */
+ SnowflakeOAuthClient(
+ String accountName, OAuthCredential oAuthCredential, URIBuilder baseURIBuilder) {
+ this.oAuthCredential = new AtomicReference<>(oAuthCredential);
+
+ // build token request uri
+ baseURIBuilder.setPath(TOKEN_REQUEST_ENDPOINT);
+ try {
+ this.tokenRequestURI = baseURIBuilder.build();
+ } catch (URISyntaxException e) {
+ throw new SFException(e, ErrorCode.MAKE_URI_FAILURE, e.getMessage());
+ }
+
+ this.httpClient = HttpUtil.getHttpClient(accountName);
+ }
+
+ /** Get access token */
+ @Override
+ public AtomicReference getoAuthCredentialRef() {
+ return oAuthCredential;
+ }
+
+ /** Refresh access token using a valid refresh token */
+ @Override
+ public void refreshToken() {
+ String respBodyString = null;
+ try (CloseableHttpResponse httpResponse = httpClient.execute(makeRefreshTokenRequest())) {
+ respBodyString = EntityUtils.toString(httpResponse.getEntity());
+
+ if (httpResponse.getStatusLine().getStatusCode() == HttpStatusCodes.STATUS_CODE_OK) {
+ JsonObject respBody = JsonParser.parseString(respBodyString).getAsJsonObject();
+
+ if (respBody.has(ACCESS_TOKEN) && respBody.has(EXPIRES_IN)) {
+ // Trim surrounding quotation marks
+ String newAccessToken = respBody.get(ACCESS_TOKEN).toString().replaceAll("^\"|\"$", "");
+
+ oAuthCredential.get().setAccessToken(newAccessToken);
+ oAuthCredential.get().setExpiresIn(respBody.get(EXPIRES_IN).getAsInt());
+ return;
+ }
+ }
+ throw new SFException(
+ ErrorCode.OAUTH_REFRESH_TOKEN_ERROR,
+ "Refresh access token fail with response: " + respBodyString);
+ } catch (Exception e) {
+ throw new SFException(ErrorCode.OAUTH_REFRESH_TOKEN_ERROR, e.getMessage());
+ }
+ }
+
+ /** Helper method for making refresh request */
+ private HttpUriRequest makeRefreshTokenRequest() {
+ HttpPost post = new HttpPost(tokenRequestURI);
+ post.addHeader(HttpHeaders.CONTENT_TYPE, OAUTH_CONTENT_TYPE_HEADER);
+ post.addHeader(HttpHeaders.AUTHORIZATION, oAuthCredential.get().getAuthHeader());
+
+ Map