Implement tlsEnabled options for driver and launcher.

Signed-off-by: agoins <alyssacgoins@gmail.com>

Author: alyssacgoins

backend/src/apiserver/common/config.go

@@ -32,7 +32,7 @@ const (
 	KubeflowUserIDPrefix                    string = "KUBEFLOW_USERID_PREFIX"
 	UpdatePipelineVersionByDefault          string = "AUTO_UPDATE_PIPELINE_DEFAULT_VERSION"
 	TokenReviewAudience                     string = "TOKEN_REVIEW_AUDIENCE"
-	MLPipelineTLSEnabled                    string = "ML_PIPELINE_TLS_ENABLED"
+	MetadataTLSEnabled                      string = "METADATA_TLS_ENABLED"
 	CaBundleSecretName                      string = "CABUNDLE_SECRET_NAME"
 )
 
@@ -130,12 +130,8 @@ func GetTokenReviewAudience() string {
 	return GetStringConfigWithDefault(TokenReviewAudience, DefaultTokenReviewAudience)
 }
 
-func SetMLPipelineServiceTLSEnabled(enabled bool) {
-	viper.Set(MLPipelineTLSEnabled, enabled)
-}
-
-func GetMLPipelineServiceTLSEnabled() bool {
-	return GetBoolConfigWithDefault(MLPipelineTLSEnabled, DefaultMLPipelineTLSEnabled)
+func GetMetadataTLSEnabled() bool {
+	return GetBoolConfigWithDefault(MetadataTLSEnabled, DefaultMetadataTLSEnabled)
 }
 
 func GetCaBundleSecretName() string {

backend/src/apiserver/common/const.go

@@ -53,7 +53,7 @@ const (
 
 const DefaultTokenReviewAudience string = "pipelines.kubeflow.org"
 
-const DefaultMLPipelineTLSEnabled = false
+const DefaultMetadataTLSEnabled = false
 
 const (
 	DefaultPipelineRunnerServiceAccount = "pipeline-runner"

backend/src/apiserver/main.go

@@ -99,8 +99,6 @@ func initCerts() (*tls.Config, error) {
 	config := &tls.Config{
 		Certificates: []tls.Certificate{serverCert},
 	}
-	// Update MLPipelineServiceTLSEnabled global variable to true once tls Config is initialized.
-	common.SetMLPipelineServiceTLSEnabled(true)
 	glog.Info("TLS cert key/pair loaded.")
 	return config, err
 }

backend/src/v2/client_manager/client_manager.go

@@ -33,6 +33,7 @@ type Options struct {
 	MLMDServerPort    string
 	CacheDisabled     bool
 	CaCertPath        string
+	MLMDTLSEnabled    bool
 }
 
 // NewClientManager creates and Init a new instance of ClientManager.
@@ -59,9 +60,13 @@ func (cm *ClientManager) CacheClient() cacheutils.Client {
 }
 
 func (cm *ClientManager) init(opts *Options) error {
-	tlsCfg, err := util.GetTLSConfig(opts.CaCertPath)
-	if err != nil {
-		return err
+	var tlsCfg *tls.Config
+	var err error
+	if opts.MLMDTLSEnabled {
+		tlsCfg, err = util.GetTLSConfig(opts.CaCertPath)
+		if err != nil {
+			return err
+		}
 	}
 	k8sClient, err := initK8sClient()
 	if err != nil {

backend/src/v2/cmd/driver/main.go

@@ -16,6 +16,7 @@ package main
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
 	"encoding/json"
 	"flag"
 	"fmt"
@@ -86,8 +87,10 @@ var (
 	noProxy           = flag.String(noProxyArg, unsetProxyArgValue, "Addresses that should ignore the proxy.")
 	publishLogs       = flag.String("publish_logs", "true", "Whether to publish component logs to the object store")
 	cacheDisabledFlag = flag.Bool("cache_disabled", false, "Disable cache globally.")
-
-	caCertPath = flag.String("ca_cert_path", "", "The path to the CA certificate to trust on connections to the ML pipeline API server and metadata server.")
+	//todo: retrieve exact wording
+	mlPipelineServiceTLSEnabled = flag.Bool("ml_pipeline_tls_enabled", false, "add wording here")
+	metadataTLSEnabled          = flag.Bool("metadata_tls_enabled", false, " add wording")
+	caCertPath                  = flag.String("ca_cert_path", "", "The path to the CA certificate to trust on connections to the ML pipeline API server and metadata server.")
 )
 
 // func RootDAG(pipelineName string, runID string, component *pipelinespec.ComponentSpec, task *pipelinespec.PipelineTaskSpec, mlmd *metadata.Client) (*Execution, error) {
@@ -177,11 +180,14 @@ func drive() (err error) {
 	if err != nil {
 		return err
 	}
-	client, err := newMlmdClient()
-	if err != nil {
-		return err
+	var tlsCfg *tls.Config
+	if *metadataTLSEnabled {
+		tlsCfg, err = util.GetTLSConfig(*caCertPath)
+		if err != nil {
+			return err
+		}
 	}
-	tlsCfg, err := util.GetTLSConfig(*caCertPath)
+	client, err := newMlmdClient(tlsCfg)
 	if err != nil {
 		return err
 	}
@@ -190,23 +196,25 @@ func drive() (err error) {
 		return err
 	}
 	options := driver.Options{
-		PipelineName:      *pipelineName,
-		RunID:             *runID,
-		RunName:           *runName,
-		RunDisplayName:    *runDisplayName,
-		Namespace:         namespace,
-		Component:         componentSpec,
-		Task:              taskSpec,
-		DAGExecutionID:    *dagExecutionID,
-		IterationIndex:    *iterationIndex,
-		PipelineLogLevel:  *logLevel,
-		PublishLogs:       *publishLogs,
-		CacheDisabled:     *cacheDisabledFlag,
-		DriverType:        *driverType,
-		TaskName:          *taskName,
-		MLMDServerAddress: *mlmdServerAddress,
-		MLMDServerPort:    *mlmdServerPort,
-		CaCertPath:        *caCertPath,
+		PipelineName:         *pipelineName,
+		RunID:                *runID,
+		RunName:              *runName,
+		RunDisplayName:       *runDisplayName,
+		Namespace:            namespace,
+		Component:            componentSpec,
+		Task:                 taskSpec,
+		DAGExecutionID:       *dagExecutionID,
+		IterationIndex:       *iterationIndex,
+		PipelineLogLevel:     *logLevel,
+		PublishLogs:          *publishLogs,
+		CacheDisabled:        *cacheDisabledFlag,
+		DriverType:           *driverType,
+		TaskName:             *taskName,
+		MLMDServerAddress:    *mlmdServerAddress,
+		MLMDServerPort:       *mlmdServerPort,
+		MLPipelineTLSEnabled: *mlPipelineServiceTLSEnabled,
+		MLMDTLSEnabled:       *metadataTLSEnabled,
+		CaCertPath:           *caCertPath,
 	}
 	var execution *driver.Execution
 	var driverErr error
@@ -339,15 +347,11 @@ func writeFile(path string, data []byte) (err error) {
 	return os.WriteFile(path, data, 0o644)
 }
 
-func newMlmdClient() (*metadata.Client, error) {
+func newMlmdClient(tlsCfg *tls.Config) (*metadata.Client, error) {
 	mlmdConfig := metadata.DefaultConfig()
 	if *mlmdServerAddress != "" && *mlmdServerPort != "" {
 		mlmdConfig.Address = *mlmdServerAddress
 		mlmdConfig.Port = *mlmdServerPort
 	}
-	tlsConfig, err := util.GetTLSConfig(*caCertPath)
-	if err != nil {
-		return nil, err
-	}
-	return metadata.NewClient(mlmdConfig.Address, mlmdConfig.Port, tlsConfig)
+	return metadata.NewClient(mlmdConfig.Address, mlmdConfig.Port, tlsCfg)
 }

backend/src/v2/cmd/launcher-v2/main.go

@@ -46,6 +46,9 @@ var (
 	publishLogs       = flag.String("publish_logs", "true", "Whether to publish component logs to the object store")
 	cacheDisabledFlag = flag.Bool("cache_disabled", false, "Disable cache globally.")
 	caCertPath        = flag.String("ca_cert_path", "", "The path to the CA certificate to trust on connections to the ML pipeline API server and metadata server.")
+	//todo: retrieve exact wording
+	mlPipelineServiceTLSEnabled = flag.Bool("ml_pipeline_tls_enabled", false, "add wording here")
+	metadataTLSEnabled          = flag.Bool("metadata_tls_enabled", false, " add wording")
 )
 
 func main() {
@@ -77,16 +80,18 @@ func run() error {
 	}
 
 	launcherV2Opts := &component.LauncherV2Options{
-		Namespace:         namespace,
-		PodName:           *podName,
-		PodUID:            *podUID,
-		MLMDServerAddress: *mlmdServerAddress,
-		MLMDServerPort:    *mlmdServerPort,
-		PipelineName:      *pipelineName,
-		RunID:             *runID,
-		PublishLogs:       *publishLogs,
-		CacheDisabled:     *cacheDisabledFlag,
-		CaCertPath:        *caCertPath,
+		Namespace:            namespace,
+		PodName:              *podName,
+		PodUID:               *podUID,
+		MLMDServerAddress:    *mlmdServerAddress,
+		MLMDServerPort:       *mlmdServerPort,
+		PipelineName:         *pipelineName,
+		RunID:                *runID,
+		PublishLogs:          *publishLogs,
+		CacheDisabled:        *cacheDisabledFlag,
+		MLPipelineTLSEnabled: *mlPipelineServiceTLSEnabled,
+		MetadataTLSEnabled:   *metadataTLSEnabled,
+		CaCertPath:           *caCertPath,
 	}
 
 	switch *executorType {
@@ -109,6 +114,7 @@ func run() error {
 			MLMDServerAddress: launcherV2Opts.MLMDServerAddress,
 			MLMDServerPort:    launcherV2Opts.MLMDServerPort,
 			CacheDisabled:     launcherV2Opts.CacheDisabled,
+			MLMDTLSEnabled:    launcherV2Opts.MetadataTLSEnabled,
 			CaCertPath:        launcherV2Opts.CaCertPath,
 		}
 		clientManager, err := client_manager.NewClientManager(clientOptions)

backend/src/v2/compiler/argocompiler/argo.go

@@ -171,6 +171,7 @@ func Compile(jobArg *pipelinespec.PipelineJob, kubernetesSpecArg *pipelinespec.S
 	if opts != nil {
 		c.cacheDisabled = opts.CacheDisabled
 		c.defaultWorkspace = opts.DefaultWorkspace
+		c.mlPipelineTLSEnabled = opts.TLSEnabled
 		if opts.DriverImage != "" {
 			c.driverImage = opts.DriverImage
 		}
@@ -199,14 +200,15 @@ type workflowCompiler struct {
 	spec      *pipelinespec.PipelineSpec
 	executors map[string]*pipelinespec.PipelineDeploymentConfig_ExecutorSpec
 	// state
-	wf               *wfapi.Workflow
-	templates        map[string]*wfapi.Template
-	driverImage      string
-	driverCommand    []string
-	launcherImage    string
-	launcherCommand  []string
-	cacheDisabled    bool
-	defaultWorkspace *k8score.PersistentVolumeClaimSpec
+	wf                   *wfapi.Workflow
+	templates            map[string]*wfapi.Template
+	driverImage          string
+	driverCommand        []string
+	launcherImage        string
+	launcherCommand      []string
+	cacheDisabled        bool
+	defaultWorkspace     *k8score.PersistentVolumeClaimSpec
+	mlPipelineTLSEnabled bool
 }
 
 func (c *workflowCompiler) Resolver(name string, component *pipelinespec.ComponentSpec, resolver *pipelinespec.PipelineDeploymentConfig_ResolverSpec) error {

backend/src/v2/compiler/argocompiler/common.go

@@ -61,18 +61,6 @@ func ConfigureCustomCABundle(tmpl *wfapi.Template) {
 		glog.Error("Env var CABUNDLE_SECRET_NAME is blank or empty. Failed to configure custom CA bundle.")
 		return
 	}
-	// OpenSSL default cert file env variable.
-	// Similar to AWS_CA_BUNDLE, the SSL_CERT_DIR equivalent for paths had unyielding
-	// results, even after rehashing.
-	// https://www.openssl.org/docs/man1.1.1/man3/SSL_CTX_set_default_verify_paths.html
-	tmpl.Container.Env = append(tmpl.Container.Env, k8score.EnvVar{
-		Name:  "SSL_CERT_FILE",
-		Value: common.TLSCertCAPath,
-	})
-	tmpl.Container.Env = append(tmpl.Container.Env, k8score.EnvVar{
-		Name:  "SSL_CERT_DIR",
-		Value: common.TLSCertCAPath,
-	})
 	volume := k8score.Volume{
 		Name: "ca-secret",
 		VolumeSource: k8score.VolumeSource{

backend/src/v2/compiler/argocompiler/container.go

@@ -210,7 +210,13 @@ func (c *workflowCompiler) addContainerDriverTemplate() string {
 	if c.cacheDisabled {
 		args = append(args, "--cache_disabled")
 	}
-	if common.GetMLPipelineServiceTLSEnabled() {
+	if c.mlPipelineTLSEnabled {
+		args = append(args, "--ml_pipeline_tls_enabled")
+	}
+	if common.GetMetadataTLSEnabled() {
+		args = append(args, "--metadata_tls_enabled")
+	}
+	if c.mlPipelineTLSEnabled || common.GetMetadataTLSEnabled() {
 		args = append(args, "--ca_cert_path", common.TLSCertCAPath)
 	}
 	if value, ok := os.LookupEnv(PipelineLogLevelEnvVar); ok {
@@ -249,7 +255,7 @@ func (c *workflowCompiler) addContainerDriverTemplate() string {
 		},
 	}
 	// If the apiserver is TLS-enabled, add the custom CA bundle to the container driver template.
-	if common.GetMLPipelineServiceTLSEnabled() {
+	if c.mlPipelineTLSEnabled {
 		ConfigureCustomCABundle(t)
 	}
 	c.templates[name] = t
@@ -535,7 +541,7 @@ func (c *workflowCompiler) addContainerExecutorTemplate(task *pipelinespec.Pipel
 		},
 	}
 	// If the apiserver is TLS-enabled, add the custom CA bundle to the executor.
-	if common.GetMLPipelineServiceTLSEnabled() {
+	if c.mlPipelineTLSEnabled {
 		ConfigureCustomCABundle(executor)
 	}
 

backend/src/v2/compiler/argocompiler/dag.go

@@ -575,7 +575,13 @@ func (c *workflowCompiler) addDAGDriverTemplate() string {
 	if c.cacheDisabled {
 		args = append(args, "--cache_disabled")
 	}
-	if common.GetMLPipelineServiceTLSEnabled() {
+	if c.mlPipelineTLSEnabled {
+		args = append(args, "--ml_pipeline_tls_enabled")
+	}
+	if common.GetMetadataTLSEnabled() {
+		args = append(args, "--metadata_tls_enabled")
+	}
+	if c.mlPipelineTLSEnabled || common.GetMetadataTLSEnabled() {
 		args = append(args, "--ca_cert_path", common.TLSCertCAPath)
 	}
 	if value, ok := os.LookupEnv(PipelineLogLevelEnvVar); ok {
@@ -614,7 +620,7 @@ func (c *workflowCompiler) addDAGDriverTemplate() string {
 		},
 	}
 	// If the apiserver is TLS-enabled, add the custom CA bundle to the DAG driver template.
-	if common.GetMLPipelineServiceTLSEnabled() {
+	if c.mlPipelineTLSEnabled {
 		ConfigureCustomCABundle(t)
 	}
 	c.templates[name] = t