@@ -22,8 +22,6 @@ import (
22
22
"strings"
23
23
"time"
24
24
25
- "github.com/kubeflow/pipelines/backend/src/v2/objectstore"
26
-
27
25
"github.com/golang/glog"
28
26
"github.com/golang/protobuf/ptypes/timestamp"
29
27
"github.com/google/uuid"
@@ -34,6 +32,7 @@ import (
34
32
"github.com/kubeflow/pipelines/backend/src/v2/config"
35
33
"github.com/kubeflow/pipelines/backend/src/v2/expression"
36
34
"github.com/kubeflow/pipelines/backend/src/v2/metadata"
35
+ "github.com/kubeflow/pipelines/backend/src/v2/objectstore"
37
36
"github.com/kubeflow/pipelines/kubernetes_platform/go/kubernetesplatform"
38
37
pb "github.com/kubeflow/pipelines/third_party/ml-metadata/go/ml_metadata"
39
38
"google.golang.org/protobuf/encoding/protojson"
@@ -134,28 +133,30 @@ func RootDAG(ctx context.Context, opts Options, mlmd *metadata.Client) (executio
134
133
}
135
134
// TODO(v2): in pipeline spec, rename GCS output directory to pipeline root.
136
135
pipelineRoot := opts .RuntimeConfig .GetGcsOutputDirectory ()
136
+
137
+ restConfig , err := rest .InClusterConfig ()
138
+ if err != nil {
139
+ return nil , fmt .Errorf ("failed to initialize kubernetes client: %w" , err )
140
+ }
141
+ k8sClient , err := kubernetes .NewForConfig (restConfig )
142
+ if err != nil {
143
+ return nil , fmt .Errorf ("failed to initialize kubernetes client set: %w" , err )
144
+ }
145
+ cfg , err := config .FromConfigMap (ctx , k8sClient , opts .Namespace )
146
+ if err != nil {
147
+ return nil , err
148
+ }
149
+
137
150
pipelineBucketSessionInfo := objectstore.SessionInfo {}
138
151
if pipelineRoot != "" {
139
152
glog .Infof ("PipelineRoot=%q" , pipelineRoot )
140
153
} else {
141
- restConfig , err := rest .InClusterConfig ()
142
- if err != nil {
143
- return nil , fmt .Errorf ("failed to initialize kubernetes client: %w" , err )
144
- }
145
- k8sClient , err := kubernetes .NewForConfig (restConfig )
146
- if err != nil {
147
- return nil , fmt .Errorf ("failed to initialize kubernetes client set: %w" , err )
148
- }
149
- cfg , err := config .FromConfigMap (ctx , k8sClient , opts .Namespace )
150
- if err != nil {
151
- return nil , err
152
- }
153
154
pipelineRoot = cfg .DefaultPipelineRoot ()
154
155
glog .Infof ("PipelineRoot=%q from default config" , pipelineRoot )
155
- pipelineBucketSessionInfo , err = cfg . GetBucketSessionInfo ()
156
- if err != nil {
157
- return nil , err
158
- }
156
+ }
157
+ pipelineBucketSessionInfo , err = cfg . GetBucketSessionInfo ( pipelineRoot )
158
+ if err != nil {
159
+ return nil , err
159
160
}
160
161
bucketSessionInfo , err := json .Marshal (pipelineBucketSessionInfo )
161
162
if err != nil {
0 commit comments