-
Notifications
You must be signed in to change notification settings - Fork 108
/
loaders.go
246 lines (224 loc) · 7.2 KB
/
loaders.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
package loaders
import (
"fmt"
"os"
"path/filepath"
"io/ioutil"
netUrl "net/url"
"path"
"strings"
"github.com/ghodss/yaml"
gogetter "github.com/hashicorp/go-getter"
kfapis "github.com/opendatahub-io/opendatahub-operator/apis"
"github.com/opendatahub-io/opendatahub-operator/pkg/kfconfig"
"github.com/opendatahub-io/opendatahub-operator/pkg/utils"
log "github.com/sirupsen/logrus"
)
type Loader interface {
LoadKfConfig(kfdef interface{}) (*kfconfig.KfConfig, error)
LoadKfDef(config kfconfig.KfConfig, out interface{}) error
}
const (
Api = "kfdef.apps.kubeflow.org"
)
func isValidUrl(toTest string) bool {
_, err := netUrl.ParseRequestURI(toTest)
if err != nil {
return false
} else {
return true
}
}
// LoadConfigFromURI reads the kfdef from a remote URI or local file,
// and returns the kfconfig.
// It will set the AppDir and ConfigFilename in kfconfig:
// AppDir = cwd if configFile is remote, or it will be the dir of configFile.
// ConfigFilename = the file name of configFile.
func LoadConfigFromURI(configFile string) (*kfconfig.KfConfig, error) {
if configFile == "" {
return nil, fmt.Errorf("config file must be the URI of a KfDef spec")
}
isRemoteFile, err := utils.IsRemoteFile(configFile)
if err != nil {
return nil, err
}
// appFile is configFile if configFile is local.
// Otherwise (configFile is remote), appFile points to a downloaded copy of configFile in tmp.
appFile := configFile
// If config is remote, download it to a temp dir.
if isRemoteFile {
// TODO(jlewi): We should check if configFile doesn't specify a protocol or the protocol
// is file:// then we can just read it rather than fetching with go-getter.
appDir, err := ioutil.TempDir("", "")
if err != nil {
return nil, fmt.Errorf("Create a temporary directory to copy the file to.")
}
// Open config file
//
// TODO(jlewi): Should we use hashicorp go-getter.GetAny here? We use that to download
// the tarballs for the repos. Maybe we should use that here as well to be consistent.
appFile = path.Join(appDir, "tmp_app.yaml")
log.Infof("Downloading %v to %v", configFile, appFile)
configFileUri, err := netUrl.Parse(configFile)
if err != nil {
log.Errorf("Could not parse configFile url")
}
if isValidUrl(configFile) {
errGet := gogetter.GetFile(appFile, configFile)
if errGet != nil {
return nil, &kfapis.KfError{
Code: int(kfapis.INVALID_ARGUMENT),
Message: fmt.Sprintf("could not fetch specified config %s: %v", configFile, errGet),
}
}
} else {
g := new(gogetter.FileGetter)
g.Copy = true
errGet := g.GetFile(appFile, configFileUri)
if errGet != nil {
return nil, &kfapis.KfError{
Code: int(kfapis.INVALID_ARGUMENT),
Message: fmt.Sprintf("could not fetch specified config %s: %v", configFile, err),
}
}
}
}
// Read contents
configFileBytes, err := ioutil.ReadFile(appFile)
if err != nil {
return nil, &kfapis.KfError{
Code: int(kfapis.INTERNAL_ERROR),
Message: fmt.Sprintf("could not read from config file %s: %v", configFile, err),
}
}
// Check API version.
var obj map[string]interface{}
if err = yaml.Unmarshal(configFileBytes, &obj); err != nil {
return nil, &kfapis.KfError{
Code: int(kfapis.INVALID_ARGUMENT),
Message: fmt.Sprintf("invalid config file format: %v", err),
}
}
apiVersion, ok := obj["apiVersion"]
if !ok {
return nil, &kfapis.KfError{
Code: int(kfapis.INVALID_ARGUMENT),
Message: "invalid config: apiVersion is not found.",
}
}
apiVersionSeparated := strings.Split(apiVersion.(string), "/")
if len(apiVersionSeparated) < 2 || apiVersionSeparated[0] != Api {
return nil, &kfapis.KfError{
Code: int(kfapis.INVALID_ARGUMENT),
Message: fmt.Sprintf("invalid config: apiVersion must be in the format of %v/<version>, got %v", Api, apiVersion),
}
}
// Add this check because kfctl binary can not properly install Kubeflow using v1alpha1 configuration.
// See https://github.com/kubeflow/kubeflow/issues/4371.
if apiVersionSeparated[1] == "v1alpha1" {
return nil, &kfapis.KfError{
Code: int(kfapis.INVALID_ARGUMENT),
Message: fmt.Sprintf(
"KfDef version v1alpha1 is not supported by this binary. Please use configs at %s for to deploy Kubeflow 0.7 or use old kfctl at %s to deploy Kubeflow 0.6",
"https://github.com/kubeflow/manifests/tree/v0.7-branch/kfdef",
"https://github.com/kubeflow/kubeflow/releases/tag/v0.6.2",
)}
}
converters := map[string]Loader{
"v1": V1{},
}
converter, ok := converters[apiVersionSeparated[1]]
if !ok {
versions := []string{}
for key := range converters {
versions = append(versions, key)
}
return nil, &kfapis.KfError{
Code: int(kfapis.INVALID_ARGUMENT),
Message: fmt.Sprintf("invalid config: version not supported; supported versions: %v, got %v",
strings.Join(versions, ", "), apiVersionSeparated[1]),
}
}
kfconfig, err := converter.LoadKfConfig(obj)
if err != nil {
log.Errorf("Failed to convert kfdef to kfconfig: %v", err)
return nil, err
}
// Set the AppDir and ConfigFileName for kfconfig
if isRemoteFile {
cwd, err := os.Getwd()
if err != nil {
return nil, &kfapis.KfError{
Code: int(kfapis.INTERNAL_ERROR),
Message: fmt.Sprintf("could not get current directory for KfDef %v", err),
}
}
kfconfig.Spec.AppDir = cwd
} else {
kfconfig.Spec.AppDir = filepath.Dir(configFile)
}
kfconfig.Spec.ConfigFileName = filepath.Base(configFile)
return kfconfig, nil
}
func isCwdEmpty() string {
cwd, _ := os.Getwd()
files, _ := ioutil.ReadDir(cwd)
if len(files) > 1 {
return ""
}
return cwd
}
func WriteConfigToFile(config kfconfig.KfConfig) error {
if config.Spec.AppDir == "" {
return &kfapis.KfError{
Code: int(kfapis.INVALID_ARGUMENT),
Message: "No AppDir, cannot write to file.",
}
}
if config.Spec.ConfigFileName == "" {
return &kfapis.KfError{
Code: int(kfapis.INVALID_ARGUMENT),
Message: "No ConfigFileName, cannot write to file.",
}
}
filename := filepath.Join(config.Spec.AppDir, config.Spec.ConfigFileName)
converters := map[string]Loader{
"v1": V1{},
}
apiVersionSeparated := strings.Split(config.APIVersion, "/")
if len(apiVersionSeparated) < 2 || apiVersionSeparated[0] != Api {
return &kfapis.KfError{
Code: int(kfapis.INVALID_ARGUMENT),
Message: fmt.Sprintf("invalid config: apiVersion must be in the format of %v/<version>, got %v", Api, config.APIVersion),
}
}
converter, ok := converters[apiVersionSeparated[1]]
if !ok {
return &kfapis.KfError{
Code: int(kfapis.INVALID_ARGUMENT),
Message: fmt.Sprintf("invalid config: unable to find converter for version %v", config.APIVersion),
}
}
var kfdef interface{}
if err := converter.LoadKfDef(config, &kfdef); err != nil {
return &kfapis.KfError{
Code: int(kfapis.INVALID_ARGUMENT),
Message: fmt.Sprintf("error when loading KfDef: %v", err),
}
}
kfdefBytes, err := yaml.Marshal(kfdef)
if err != nil {
return &kfapis.KfError{
Code: int(kfapis.INVALID_ARGUMENT),
Message: fmt.Sprintf("error when marshaling KfDef: %v", err),
}
}
err = ioutil.WriteFile(filename, kfdefBytes, 0644)
if err != nil {
return &kfapis.KfError{
Code: int(kfapis.INTERNAL_ERROR),
Message: fmt.Sprintf("error when writing KfDef: %v", err),
}
}
return nil
}