Merge pull request #13846 from thalesmg/20240920-r58-snowflake-proxy

feat(snowflake): add http proxy support

apps/emqx_bridge_snowflake/src/emqx_bridge_snowflake_action_schema.erl

@@ -102,6 +102,11 @@ fields(aggreg_parameters) ->
                     importance => ?IMPORTANCE_HIDDEN,
                     required => true
                 }
+            )},
+        {proxy,
+            mk(
+                hoconsc:union([none, ref(proxy_config)]),
+                #{default => none, desc => ?DESC("proxy_config")}
             )}
     ];
 fields(direct_parameters) ->
@@ -129,6 +134,12 @@ fields(aggregation) ->
                 }
             )}
     ];
+fields(proxy_config) ->
+    [
+        {host, mk(binary(), #{required => true, desc => ?DESC("proxy_config_host")})},
+        {port,
+            mk(emqx_schema:port_number(), #{required => true, desc => ?DESC("proxy_config_port")})}
+    ];
 fields(action_resource_opts) ->
     %% NOTE: This action should benefit from generous batching defaults.
     emqx_bridge_v2_schema:action_resource_opts_fields([
@@ -140,7 +151,8 @@ desc(Name) when
     Name =:= ?ACTION_TYPE;
     Name =:= aggreg_parameters;
     Name =:= aggregation;
-    Name =:= parameters
+    Name =:= parameters;
+    Name =:= proxy_config
 ->
     ?DESC(Name);
 desc(action_resource_opts) ->

apps/emqx_bridge_snowflake/src/emqx_bridge_snowflake_connector.erl

@@ -75,7 +75,8 @@
     username := binary(),
     password := emqx_schema_secret:secret(),
     dsn := binary(),
-    pool_size := pos_integer()
+    pool_size := pos_integer(),
+    proxy := none | proxy_config()
 }.
 -type connector_state() :: #{
     account := account(),
@@ -106,6 +107,11 @@
 -type stage() :: binary().
 -type pipe() :: binary().
 
+-type proxy_config() :: #{
+    host := binary(),
+    port := emqx_schema:port_number()
+}.
+
 -type odbc_pool() :: connector_resource_id().
 -type http_pool() :: action_resource_id().
 -type http_client_config() :: #{
@@ -192,7 +198,8 @@ on_start(ConnResId, ConnConfig) ->
         username := Username,
         password := Password,
         dsn := DSN,
-        pool_size := PoolSize
+        pool_size := PoolSize,
+        proxy := ProxyConfig
     } = ConnConfig,
     #{hostname := Host, port := Port} = emqx_schema:parse_server(Server, ?SERVER_OPTS),
     PoolOpts = [
@@ -202,6 +209,7 @@ on_start(ConnResId, ConnConfig) ->
         {server, Server},
         {username, Username},
         {password, Password},
+        {proxy, ProxyConfig},
         {on_disconnect, {?MODULE, disconnect, []}}
     ],
     case emqx_resource_pool:start(ConnResId, ?MODULE, PoolOpts) of
@@ -643,7 +651,8 @@ start_http_pool(ActionResId, ActionConfig, ConnState) ->
             connect_timeout := ConnectTimeout,
             pipelining := Pipelining,
             pool_size := PoolSize,
-            max_retries := MaxRetries
+            max_retries := MaxRetries,
+            proxy := ProxyConfig0
         },
         resource_opts := #{request_ttl := RequestTTL}
     } = ActionConfig,
@@ -668,17 +677,31 @@ start_http_pool(ActionResId, ActionConfig, ConnState) ->
     ]),
     JWTConfig = jwt_config(ActionResId, ActionConfig, ConnState),
     TransportOpts = emqx_tls_lib:to_client_opts(#{enable => true, verify => verify_none}),
-    PoolOpts = [
-        {host, Host},
-        {port, Port},
-        {connect_timeout, ConnectTimeout},
-        {keepalive, 30_000},
-        {pool_type, random},
-        {pool_size, PoolSize},
-        {transport, tls},
-        {transport_opts, TransportOpts},
-        {enable_pipelining, Pipelining}
-    ],
+    ProxyConfig =
+        case ProxyConfig0 of
+            none ->
+                [];
+            #{host := ProxyHost, port := ProxyPort} ->
+                [
+                    {proxy, #{
+                        host => str(ProxyHost),
+                        port => ProxyPort
+                    }}
+                ]
+        end,
+    PoolOpts =
+        ProxyConfig ++
+            [
+                {host, Host},
+                {port, Port},
+                {connect_timeout, ConnectTimeout},
+                {keepalive, 30_000},
+                {pool_type, random},
+                {pool_size, PoolSize},
+                {transport, tls},
+                {transport_opts, TransportOpts},
+                {enable_pipelining, Pipelining}
+            ],
     case ehttpc_sup:start_pool(ActionResId, PoolOpts) of
         {ok, _} ->
             {ok, #{
@@ -691,6 +714,9 @@ start_http_pool(ActionResId, ActionConfig, ConnState) ->
                     request_ttl => RequestTTL
                 }
             }};
+        {error, {already_started, _}} ->
+            _ = ehttpc_sup:stop_pool(ActionResId),
+            start_http_pool(ActionResId, ActionConfig, ConnState);
         {error, Reason} ->
             {error, Reason}
     end.
@@ -803,6 +829,10 @@ conn_str([{username, Username} | Opts], Acc) ->
     conn_str(Opts, ["uid=" ++ str(Username) | Acc]);
 conn_str([{password, Password} | Opts], Acc) ->
     conn_str(Opts, ["pwd=" ++ str(emqx_secret:unwrap(Password)) | Acc]);
+conn_str([{proxy, none} | Opts], Acc) ->
+    conn_str(Opts, Acc);
+conn_str([{proxy, #{host := Host, port := Port}} | Opts], Acc) ->
+    conn_str(Opts, ["proxy=" ++ str(Host) ++ ":" ++ str(Port) | Acc]);
 conn_str([{_, _} | Opts], Acc) ->
     conn_str(Opts, Acc).
 

apps/emqx_bridge_snowflake/src/emqx_bridge_snowflake_connector_schema.erl

@@ -74,16 +74,29 @@ fields(connector_config) ->
                 desc => ?DESC("account"),
                 validator => fun account_id_validator/1
             })},
-        {dsn, mk(binary(), #{required => true, desc => ?DESC("dsn")})}
+        {dsn, mk(binary(), #{required => true, desc => ?DESC("dsn")})},
+        {proxy,
+            mk(
+                hoconsc:union([none, hoconsc:ref(?MODULE, proxy_config)]),
+                #{default => none, desc => ?DESC("proxy_config")}
+            )}
         | Fields
     ] ++
         emqx_connector_schema:resource_opts() ++
-        emqx_connector_schema_lib:ssl_fields().
+        emqx_connector_schema_lib:ssl_fields();
+fields(proxy_config) ->
+    [
+        {host, mk(binary(), #{required => true, desc => ?DESC("proxy_config_host")})},
+        {port,
+            mk(emqx_schema:port_number(), #{required => true, desc => ?DESC("proxy_config_port")})}
+    ].
 
 desc("config_connector") ->
     ?DESC("config_connector");
 desc(resource_opts) ->
     ?DESC(emqx_resource_schema, resource_opts);
+desc(proxy_config) ->
+    ?DESC("proxy_config");
 desc(_Name) ->
     undefined.
 

mix.exs

@@ -192,7 +192,10 @@ defmodule EMQXUmbrella.MixProject do
   def common_dep(:gun), do: {:gun, github: "emqx/gun", tag: "1.3.11", override: true}
   # in conflict by cowboy_swagger and cowboy
   def common_dep(:ranch), do: {:ranch, github: "emqx/ranch", tag: "1.8.1-emqx", override: true}
-  def common_dep(:ehttpc), do: {:ehttpc, github: "emqx/ehttpc", tag: "0.4.14", override: true}
+
+  def common_dep(:ehttpc),
+    do: {:ehttpc, github: "emqx/ehttpc", tag: "0.5.0", override: true}
+
   def common_dep(:jiffy), do: {:jiffy, github: "emqx/jiffy", tag: "1.0.6", override: true}
 
   def common_dep(:grpc),

rebar.config

@@ -77,7 +77,7 @@
     {gpb, "4.19.9"},
     {typerefl, {git, "https://github.com/ieQu1/typerefl", {tag, "0.9.1"}}},
     {gun, {git, "https://github.com/emqx/gun", {tag, "1.3.11"}}},
-    {ehttpc, {git, "https://github.com/emqx/ehttpc", {tag, "0.4.14"}}},
+    {ehttpc, {git, "https://github.com/emqx/ehttpc", {tag, "0.5.0"}}},
     {gproc, {git, "https://github.com/emqx/gproc", {tag, "0.9.0.1"}}},
     {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.6"}}},
     {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.9.2"}}},

rel/i18n/emqx_bridge_snowflake_action_schema.hocon

@@ -88,4 +88,19 @@ emqx_bridge_snowflake_action_schema {
   max_retries.desc:
   """Max retry attempts if there's an error when sending an HTTP request."""
 
+  proxy_config.label:
+  """Proxy"""
+  proxy_config.desc:
+  """Proxy configuration.  Only plain HTTP proxies are currently supported (no HTTPS)."""
+
+  proxy_config_host.label:
+  """Proxy Host"""
+  proxy_config_host.desc:
+  """Proxy host address to connect to."""
+
+  proxy_config_port.label:
+  """Proxy Port"""
+  proxy_config_port.desc:
+  """Proxy port to use when connecting."""
+
 }

rel/i18n/emqx_bridge_snowflake_connector_schema.hocon

@@ -19,4 +19,19 @@ emqx_bridge_snowflake_connector_schema {
   dsn.desc:
   """Data Source Name (DSN) associated with the installed Snowflake ODBC driver."""
 
+  proxy_config.label:
+  """Proxy"""
+  proxy_config.desc:
+  """Proxy configuration.  Only plain HTTP proxies are currently supported (no HTTPS)."""
+
+  proxy_config_host.label:
+  """Proxy Host"""
+  proxy_config_host.desc:
+  """Proxy host address to connect to."""
+
+  proxy_config_port.label:
+  """Proxy Port"""
+  proxy_config_port.desc:
+  """Proxy port to use when connecting."""
+
 }