This repository has been archived by the owner on Oct 9, 2023. It is now read-only.
K8 array resource manager #120
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,10 +5,16 @@ | |
| package k8s | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "io/ioutil" | ||
|
|
||
| "github.com/pkg/errors" | ||
| v1 "k8s.io/api/core/v1" | ||
| restclient "k8s.io/client-go/rest" | ||
| "sigs.k8s.io/controller-runtime/pkg/client" | ||
|
|
||
| pluginsConfig "github.com/lyft/flyteplugins/go/tasks/config" | ||
| "github.com/lyft/flyteplugins/go/tasks/pluginmachinery/workqueue" | ||
| ) | ||
|
|
||
| //go:generate pflags Config --default-var=defaultConfig | ||
|
|
@@ -31,14 +37,80 @@ var ( | |
| }, | ||
| } | ||
|
|
||
| configSection = pluginsConfig.MustRegisterSubSection(configSectionKey, defaultConfig) | ||
| ) | ||
|
|
||
| type ResourceConfig struct { | ||
| PrimaryLabel string `json:"primaryLabel" pflag:",PrimaryLabel of a given service cluster"` | ||
| Limit int `json:"limit" pflag:",Resource quota (in the number of outstanding requests) for the cluster"` | ||
| } | ||
|
|
||
| type ClusterConfig struct { | ||
|
There was a problem hiding this comment. can we put some comments here, explaining why we have a cluster config and why we need to do this? This seems a little complicated IMO |
||
| Name string `json:"name" pflag:",Friendly name of the remote cluster"` | ||
| Endpoint string `json:"endpoint" pflag:", Remote K8s cluster endpoint"` | ||
| Auth Auth `json:"auth" pflag:"-, Auth setting for the cluster"` | ||
| Enabled bool `json:"enabled" pflag:", Boolean flag to enable or disable"` | ||
| } | ||
|
|
||
| type Auth struct { | ||
| Type string `json:"type" pflag:", Authentication type"` | ||
| TokenPath string `json:"tokenPath" pflag:", Token path"` | ||
| CertPath string `json:"certPath" pflag:", Certificate path"` | ||
| } | ||
|
|
||
| func (auth Auth) GetCA() ([]byte, error) { | ||
| cert, err := ioutil.ReadFile(auth.CertPath) | ||
| if err != nil { | ||
| return nil, errors.Wrap(err, "failed to read k8s CA cert from configured path") | ||
| } | ||
| return cert, nil | ||
| } | ||
|
|
||
| func (auth Auth) GetToken() (string, error) { | ||
| token, err := ioutil.ReadFile(auth.TokenPath) | ||
| if err != nil { | ||
| return "", errors.Wrap(err, "failed to read k8s bearer token from configured path") | ||
| } | ||
| return string(token), nil | ||
| } | ||
|
|
||
| // TODO: Move logic to flytestdlib | ||
| // Reads secret values from paths specified in the config to initialize a Kubernetes rest client Config. | ||
|
There was a problem hiding this comment. we should move this to flytestdlib since it also exists in flyteadmin There was a problem hiding this comment. Yup. Will add TODO. There was a problem hiding this comment. @katrogan / @anandswaminathan I prefer we dont do this at all, I know you want to do this to test, but keep it for shorter time. My preference is that we run the entire workflow in a separate cluster if need be |
||
| func RemoteClusterConfig(host string, auth Auth) (*restclient.Config, error) { | ||
|
There was a problem hiding this comment. What is the purpose of remoteCluster? There was a problem hiding this comment. This is our V0 milestone. We do not have a way to prioritize regular tasks over array tasks in the same namespace. Array task will end up utilizing all the resource quotas, that will prevent all the other tasks from progressing. |
||
| tokenString, err := auth.GetToken() | ||
| if err != nil { | ||
| return nil, errors.New(fmt.Sprintf("Failed to get auth token: %+v", err)) | ||
| } | ||
|
|
||
| caCert, err := auth.GetCA() | ||
| if err != nil { | ||
| return nil, errors.New(fmt.Sprintf("Failed to get auth CA: %+v", err)) | ||
| } | ||
|
|
||
| tlsClientConfig := restclient.TLSClientConfig{} | ||
| tlsClientConfig.CAData = caCert | ||
| return &restclient.Config{ | ||
| Host: host, | ||
| TLSClientConfig: tlsClientConfig, | ||
| BearerToken: tokenString, | ||
| }, nil | ||
| } | ||
|
|
||
| func GetK8sClient(config ClusterConfig) (client.Client, error) { | ||
| kubeConf, err := RemoteClusterConfig(config.Endpoint, config.Auth) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return client.New(kubeConf, client.Options{}) | ||
| } | ||
|
|
||
| // Defines custom config for K8s Array plugin | ||
| type Config struct { | ||
| DefaultScheduler string `json:"scheduler" pflag:",Decides the scheduler to use when launching array-pods."` | ||
| MaxErrorStringLength int `json:"maxErrorLength" pflag:",Determines the maximum length of the error string returned for the array."` | ||
| MaxArrayJobSize int64 `json:"maxArrayJobSize" pflag:",Maximum size of array job."` | ||
| ResourceConfig ResourceConfig `json:"resourceConfig" pflag:"-,ResourceConfiguration to limit number of resources used by k8s-array."` | ||
| RemoteClusterConfig ClusterConfig `json:"remoteClusterConfig" pflag:"-,Configuration of remote K8s cluster for array jobs"` | ||
| NodeSelector map[string]string `json:"node-selector" pflag:"-,Defines a set of node selector labels to add to the pod."` | ||
| Tolerations []v1.Toleration `json:"tolerations" pflag:"-,Tolerations to be applied for k8s-array pods"` | ||
| OutputAssembler workqueue.Config | ||
|
|
@@ -48,3 +120,8 @@ type Config struct { | |
| func GetConfig() *Config { | ||
| return configSection.GetConfig().(*Config) | ||
| } | ||
|
|
||
| func IsResourceConfigSet(resourceConfig ResourceConfig) bool { | ||
| emptyResouceConfig := ResourceConfig{} | ||
| return resourceConfig != emptyResouceConfig | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
wow how was this working correctly before?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Because we never have
totalWaitingForResources? Is that your question?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
oh I didn't realize that phase was introduced here. yes, thanks for explaining