[WIP] CC-33579 - use connector level proxy configs and make HttpClient keyed on proxy

pom.xml

@@ -3,9 +3,9 @@
   <modelVersion>4.0.0</modelVersion>
 
   <!-- Arifact name and version information -->
-  <groupId>io.confluent</groupId>
+  <groupId>net.snowflake</groupId>
   <artifactId>snowflake-ingest-sdk</artifactId>
-  <version>2.1.0-hotfix1</version>
+  <version>2.1.0-hotfix1-local-snapshot</version>
   <packaging>jar</packaging>
   <name>Snowflake Ingest SDK</name>
   <description>Snowflake Ingest SDK</description>

src/main/java/net/snowflake/ingest/SimpleIngestManager.java

@@ -15,6 +15,7 @@
 import java.security.spec.InvalidKeySpecException;
 import java.util.Collections;
 import java.util.List;
+import java.util.Properties;
 import java.util.Set;
 import java.util.UUID;
 import java.util.stream.Collectors;
@@ -384,6 +385,30 @@ 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 =========
 
   /**
@@ -396,14 +421,28 @@ 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(account);
+    // make our client for sending requests with proxy properties support
+    httpClient = HttpUtil.getHttpClient(account, proxyProperties);
     // make the request builder we'll use to build messages to the service
   }
 

src/main/java/net/snowflake/ingest/streaming/internal/FlushService.java

@@ -167,7 +167,8 @@ List<List<ChannelData<T>>> getData() {
               client.getHttpClient(),
               client.getRequestBuilder(),
               client.getName(),
-              DEFAULT_MAX_UPLOAD_RETRIES);
+              DEFAULT_MAX_UPLOAD_RETRIES,
+              client.getProxyProperties());
     } catch (SnowflakeSQLException | IOException err) {
       throw new SFException(err, ErrorCode.UNABLE_TO_CONNECT_TO_STAGE);
     }

src/main/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestClientInternal.java

@@ -142,6 +142,9 @@ public class SnowflakeStreamingIngestClientInternal<T> implements SnowflakeStrea
   // Background thread that uploads telemetry data periodically
   private ScheduledExecutorService telemetryWorker;
 
+  // Store original properties for proxy configuration
+  private final Properties originalProperties;
+
   /**
    * Constructor
    *
@@ -162,11 +165,12 @@ public class SnowflakeStreamingIngestClientInternal<T> implements SnowflakeStrea
       RequestBuilder requestBuilder,
       Map<String, Object> parameterOverrides) {
     this.parameterProvider = new ParameterProvider(parameterOverrides, prop);
+    this.originalProperties = prop;
 
     this.name = name;
     String accountName = accountURL == null ? null : accountURL.getAccount();
     this.isTestMode = isTestMode;
-    this.httpClient = httpClient == null ? HttpUtil.getHttpClient(accountName) : httpClient;
+    this.httpClient = HttpUtil.getHttpClient(accountName, prop);
     this.channelCache = new ChannelCache<>();
     this.isClosed = false;
     this.requestBuilder = requestBuilder;
@@ -1075,4 +1079,35 @@ private void cleanUpResources() {
       HttpUtil.shutdownHttpConnectionManagerDaemonThread();
     }
   }
+
+  /**
+   * Extract proxy properties from the original Properties object
+   *
+   * @return Properties containing proxy configuration
+   */
+  Properties getProxyProperties() {
+    Properties proxyProperties = new Properties();
+
+
+    if (this.originalProperties != null) {
+
+      for (String key : this.originalProperties.stringPropertyNames()) {
+        if (key.equals(SFSessionProperty.USE_PROXY.getPropertyKey()) ||
+            key.equals(SFSessionProperty.PROXY_HOST.getPropertyKey()) ||
+            key.equals(SFSessionProperty.PROXY_PORT.getPropertyKey()) ||
+            key.equals(SFSessionProperty.NON_PROXY_HOSTS.getPropertyKey()) ||
+            key.equals(SFSessionProperty.PROXY_USER.getPropertyKey()) ||
+            key.equals(SFSessionProperty.PROXY_PASSWORD.getPropertyKey())) {
+          proxyProperties.put(key, this.originalProperties.getProperty(key));
+        }
+      }
+    }
+
+    // If no proxy properties found in original properties, fall back to system properties
+    if (proxyProperties.isEmpty()) {
+      return HttpUtil.generateProxyPropertiesForJDBC();
+    }
+
+    return proxyProperties;
+  }
 }

src/main/java/net/snowflake/ingest/streaming/internal/StreamingIngestStage.java

@@ -98,11 +98,34 @@ state to record unknown age.
       String clientName,
       int maxUploadRetries)
       throws SnowflakeSQLException, IOException {
+    this(isTestMode, role, httpClient, requestBuilder, clientName, maxUploadRetries, null);
+  }
+
+  /**
+   * Constructor with proxy properties support
+   *
+   * @param isTestMode whether we're under test mode
+   * @param role Snowflake role used by the Client
+   * @param httpClient http client reference
+   * @param requestBuilder request builder to build the HTTP request
+   * @param clientName the client name
+   * @param maxUploadRetries maximum number of upload retries
+   * @param proxyProperties proxy properties for HTTP client configuration
+   */
+  StreamingIngestStage(
+      boolean isTestMode,
+      String role,
+      CloseableHttpClient httpClient,
+      RequestBuilder requestBuilder,
+      String clientName,
+      int maxUploadRetries,
+      Properties proxyProperties)
+      throws SnowflakeSQLException, IOException {
     this.httpClient = httpClient;
     this.role = role;
     this.requestBuilder = requestBuilder;
     this.clientName = clientName;
-    this.proxyProperties = generateProxyPropertiesForJDBC();
+    this.proxyProperties = proxyProperties != null ? proxyProperties : generateProxyPropertiesForJDBC();
     this.maxUploadRetries = maxUploadRetries;
 
     if (!isTestMode) {
@@ -129,7 +152,7 @@ state to record unknown age.
       SnowflakeFileTransferMetadataWithAge testMetadata,
       int maxRetryCount)
       throws SnowflakeSQLException, IOException {
-    this(isTestMode, role, httpClient, requestBuilder, clientName, maxRetryCount);
+    this(isTestMode, role, httpClient, requestBuilder, clientName, maxRetryCount, null);
     if (!isTestMode) {
       throw new SFException(ErrorCode.INTERNAL_ERROR);
     }

src/main/java/net/snowflake/ingest/utils/HttpClientSettingsKey.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright (c) 2012-2017 Snowflake Computing Inc. All rights reserved.
+ */
+
+package net.snowflake.ingest.utils;
+
+import static net.snowflake.ingest.utils.Utils.isNullOrEmpty;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * This class defines all parameters needed to create an HttpClient object for the ingest service.
+ * It is used as the key for the static hashmap of reusable http clients.
+ */
+public class HttpClientSettingsKey implements Serializable {
+
+  private boolean useProxy;
+  private String proxyHost = "";
+  private int proxyPort = 0;
+  private String nonProxyHosts = "";
+  private String proxyUser = "";
+  private String proxyPassword = "";
+  private String accountName = "";
+
+  /**
+   * Constructor for proxy configuration
+   */
+  public HttpClientSettingsKey(
+      String accountName,
+      String proxyHost,
+      int proxyPort,
+      String nonProxyHosts,
+      String proxyUser,
+      String proxyPassword) {
+    this.useProxy = true;
+    this.accountName = !isNullOrEmpty(accountName) ? accountName.trim() : "";
+    this.proxyHost = !isNullOrEmpty(proxyHost) ? proxyHost.trim() : "";
+    this.proxyPort = proxyPort;
+    this.nonProxyHosts = !isNullOrEmpty(nonProxyHosts) ? nonProxyHosts.trim() : "";
+    this.proxyUser = !isNullOrEmpty(proxyUser) ? proxyUser.trim() : "";
+    this.proxyPassword = !isNullOrEmpty(proxyPassword) ? proxyPassword.trim() : "";
+  }
+
+  /**
+   * Constructor for non-proxy configuration
+   */
+  public HttpClientSettingsKey(String accountName) {
+    this.useProxy = false;
+    this.accountName = !isNullOrEmpty(accountName) ? accountName.trim() : "";
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj) return true;
+    if (obj == null || getClass() != obj.getClass()) return false;
+
+    HttpClientSettingsKey that = (HttpClientSettingsKey) obj;
+
+    return useProxy == that.useProxy &&
+           proxyPort == that.proxyPort &&
+           Objects.equals(accountName, that.accountName) &&
+           Objects.equals(proxyHost, that.proxyHost) &&
+           Objects.equals(nonProxyHosts, that.nonProxyHosts) &&
+           Objects.equals(proxyUser, that.proxyUser) &&
+           Objects.equals(proxyPassword, that.proxyPassword);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(useProxy, accountName, proxyHost, proxyPort, nonProxyHosts, proxyUser, proxyPassword);
+  }
+
+  public boolean usesProxy() {
+    return this.useProxy;
+  }
+
+  public String getAccountName() {
+    return this.accountName;
+  }
+
+  public String getProxyHost() {
+    return this.proxyHost;
+  }
+
+  public int getProxyPort() {
+    return this.proxyPort;
+  }
+
+  public String getProxyUser() {
+    return this.proxyUser;
+  }
+
+  public String getProxyPassword() {
+    return this.proxyPassword;
+  }
+
+  public String getNonProxyHosts() {
+    return this.nonProxyHosts;
+  }
+
+  @Override
+  public String toString() {
+    return "HttpClientSettingsKey[" +
+           "accountName='" + accountName + '\'' +
+           ", useProxy=" + useProxy +
+           ", proxyHost='" + proxyHost + '\'' +
+           ", proxyPort=" + proxyPort +
+           ", nonProxyHosts='" + nonProxyHosts + '\'' +
+           ", proxyUser='" + proxyUser + '\'' +
+           ", proxyPassword=" + (proxyPassword.isEmpty() ? "not set" : "set") +
+           ']';
+  }
+}