feat(sdk): support HTTP/S PROXY for SDK client (#4215)

* done

* done

* reserve https

* done

sdk/python/kfp/_client.py

@@ -90,7 +90,7 @@ class Client(object):
   LOCAL_KFP_CONTEXT = os.path.expanduser('~/.config/kfp/context.json')
 
   # TODO: Wrap the configurations for different authentication methods.
-  def __init__(self, host=None, client_id=None, namespace='kubeflow', other_client_id=None, other_client_secret=None, existing_token=None, cookies=None):
+  def __init__(self, host=None, client_id=None, namespace='kubeflow', other_client_id=None, other_client_secret=None, existing_token=None, cookies=None, proxy=None, ssl_ca_cert=None):
     """Create a new instance of kfp client.
 
     Args:
@@ -109,10 +109,12 @@ def __init__(self, host=None, client_id=None, namespace='kubeflow', other_client
       existing_token: pass in token directly, it's used for cases better get token outside of SDK, e.x. GCP Cloud Functions
           or caller already has a token
       cookies: CookieJar object containing cookies that will be passed to the pipelines API.
+      proxy: HTTP or HTTPS proxy server
+      ssl_ca_cert: cert for proxy
     """
     host = host or os.environ.get(KF_PIPELINES_ENDPOINT_ENV)
     self._uihost = os.environ.get(KF_PIPELINES_UI_ENDPOINT_ENV, host)
-    config = self._load_config(host, client_id, namespace, other_client_id, other_client_secret, existing_token)
+    config = self._load_config(host, client_id, namespace, other_client_id, other_client_secret, existing_token, proxy, ssl_ca_cert)
     # Save the loaded API client configuration, as a reference if update is
     # needed.
     self._existing_config = config
@@ -125,14 +127,24 @@ def __init__(self, host=None, client_id=None, namespace='kubeflow', other_client
     self._upload_api = kfp_server_api.api.PipelineUploadServiceApi(api_client)
     self._load_context_setting_or_default()
 
-  def _load_config(self, host, client_id, namespace, other_client_id, other_client_secret, existing_token):
+  def _load_config(self, host, client_id, namespace, other_client_id, other_client_secret, existing_token, proxy, ssl_ca_cert):
     config = kfp_server_api.configuration.Configuration()
 
+    if proxy:
+      # https://github.com/kubeflow/pipelines/blob/c6ac5e0b1fd991e19e96419f0f508ec0a4217c29/backend/api/python_http_client/kfp_server_api/rest.py#L100
+      config.proxy = proxy
+
+    if ssl_ca_cert:
+      config.ssl_ca_cert = ssl_ca_cert
+
     host = host or ''
     # Preprocess the host endpoint to prevent some common user mistakes.
     # This should only be done for non-IAP cases (when client_id is None). IAP requires preserving the protocol.
     if not client_id:
-      host = re.sub(r'^(http|https)://', '', host).rstrip('/')
+      # Per feedback in proxy env, http or https is still required
+      if not proxy:
+        host = re.sub(r'^(http|https)://', '', host)
+      host = host.rstrip('/')
 
     if host:
       config.host = host