Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feature/waitallprocesses config #1394

Merged
merged 6 commits into from
Nov 24, 2020
Merged

Feature/waitallprocesses config #1394

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
c2fc532
Add waitAllProcesses to metricsCollector config
robbertvdg Nov 18, 2020
59e63f6
wait_all_processes config for python metricscollector main
robbertvdg Nov 18, 2020
3f5d813
Correct boolean check
robbertvdg Nov 18, 2020
76d9373
waitAllProcesses config as bool
robbertvdg Nov 20, 2020
d87fd6b
add omitempty to suggestion and earlystopping
robbertvdg Nov 23, 2020
0df1035
correct default config
robbertvdg Nov 23, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions cmd/metricscollector/v1beta1/file-metricscollector/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,7 @@ var (
metricFilters = flag.String("f", "", "Metric filters")
pollInterval = flag.Duration("p", common.DefaultPollInterval, "Poll interval between running processes check")
timeout = flag.Duration("timeout", common.DefaultTimeout, "Timeout before invoke error during running processes check")
waitAll = flag.Bool("w", common.DefaultWaitAll, "Whether wait for all other main process of container exiting")
waitAllProcesses = flag.String("w", common.DefaultWaitAllProcesses, "Whether wait for all other main process of container exiting")
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need to keep this flag String, or we can use Bool?
I assume this flag can be only true or false.

Copy link
Contributor Author

@robbertvdg robbertvdg Nov 18, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I had problems with parsing booleans when a user has not specified the field in the katib-config. Do you know of a good way to do it?
I could leave it string in the katib-config and parse it to bool in GetMetricsCollectorConfigData, would that be better?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@robbertvdg It seems that empty bool in Go is false. So when user doesn't specify any values in config we send "false" to the metrics collector.
Maybe we can define MetricsCollectorConfig like this:

type MetricsCollectorConfig struct {
	Image            string                      `json:"image"`
	ImagePullPolicy  corev1.PullPolicy           `json:"imagePullPolicy"`
	Resource         corev1.ResourceRequirements `json:"resources"`
	WaitAllProcesses *bool                      `json:"waitAllProcesses"`
}

In that case, you can set -w flag if WaitAllProcesses != nil. Otherwise the default value is set.

In Katib config I was able to define bool value like this:

 "StdOut": {
        "image": "docker.io/andreyvelichkevich/file-metrics-collector",
        "imagePullPolicy": "Always",
        "waitAllProcesses": false
      },

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The problem lies with the v1.Container.args which is string[]. Therefore every container argument has to be a string, which I think is not compatible with the flag.Bool() argument (I tested it, doesn't work).

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@robbertvdg Got it.
In that case can we keep WaitAllProcesses *bool in MetricsCollectorConfig, convert it to string here:

katib/pkg/webhook/v1beta1/pod/inject_webhook.go

Line 299 in 3f5d813

args = append(args, "-w", metricsCollectorConfigData.WaitAllProcesses)

Then, use the String type like you did in Metrics Collector and convert it to Bool after.

In that case, we can avoid situation when user specify wrong values in Katib config.

What do you think @robbertvdg @gaocegege ?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@andreyvelich those were my thoughts as well, will update PR soon.

stopRules stopRulesFlag
isEarlyStopped = false
)
Expand Down Expand Up @@ -353,10 +353,12 @@ func main() {
go printMetricsFile(*metricsFilePath)
}

waitAll, _ := strconv.ParseBool(*waitAllProcesses)

wopts := common.WaitPidsOpts{
PollInterval: *pollInterval,
Timeout: *timeout,
WaitAll: *waitAll,
WaitAll: waitAll,
CompletedMarkedDirPath: filepath.Dir(*metricsFilePath),
}
if err := common.WaitMainProcesses(wopts); err != nil {
Expand Down
5 changes: 3 additions & 2 deletions cmd/metricscollector/v1beta1/tfevent-metricscollector/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ def parse_options():
parser.add_argument("-f", "--metric_filters", type=str, default="")
parser.add_argument("-p", "--poll_interval", type=int, default=const.DEFAULT_POLL_INTERVAL)
parser.add_argument("-timeout", "--timeout", type=int, default=const.DEFAULT_TIMEOUT)
parser.add_argument("-w", "--wait_all", type=bool, default=const.DEFAULT_WAIT_ALL)
parser.add_argument("-w", "--wait_all_processes", type=str, default=const.DEFAULT_WAIT_ALL_PROCESSES)

opt = parser.parse_args()
return opt
Expand All @@ -38,6 +38,7 @@ def parse_options():
logger.addHandler(handler)
logger.propagate = False
opt = parse_options()
wait_all_processes = opt.wait_all_processes.lower() == "true"
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does this compare (opt.wait_all_processes.lower() == "true") necessary ?
I assume wait_all_processes can be only "true" or "false" since WaitAllProcesses *bool in Katib config?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is the python way of converting string to bool ;) should be fine since the string can only have two possibilities

db_manager_server = opt.db_manager_server_addr.split(':')
if len(db_manager_server) != 2:
raise Exception("Invalid Katib DB manager service address: %s" %
Expand All @@ -46,7 +47,7 @@ def parse_options():
WaitMainProcesses(
pool_interval=opt.poll_interval,
timout=opt.timeout,
wait_all=opt.wait_all,
wait_all=wait_all_processes,
completed_marked_dir=opt.metrics_file_dir)

mc = MetricsCollector(opt.metric_names.split(';'))
Expand Down
2 changes: 1 addition & 1 deletion pkg/metricscollector/v1beta1/common/const.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ const (
// To run without timeout set value to 0
DefaultTimeout = 0
// DefaultWaitAll is the default value whether wait for all other main process of container exiting
DefaultWaitAll = true
DefaultWaitAllProcesses = "true"
// TrainingCompleted is the job finished marker in $$$$.pid file when main training process is completed
TrainingCompleted = "completed"

Expand Down
2 changes: 1 addition & 1 deletion pkg/metricscollector/v1beta1/common/const.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
# Default value for timeout before invoke error during running processes check
DEFAULT_TIMEOUT = 0
# Default value whether wait for all other main process of container exiting
DEFAULT_WAIT_ALL = True
DEFAULT_WAIT_ALL_PROCESSES = "True"
# Default value for directory where TF event metrics are reported
DEFAULT_METRICS_FILE_DIR = "/log"
# Job finished marker in $$$$.pid file when main process is completed
Expand Down
21 changes: 11 additions & 10 deletions pkg/util/v1beta1/katibconfig/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,25 +18,26 @@ import (
// SuggestionConfig is the JSON suggestion structure in Katib config.
type SuggestionConfig struct {
Image string `json:"image"`
ImagePullPolicy corev1.PullPolicy `json:"imagePullPolicy"`
Resource corev1.ResourceRequirements `json:"resources"`
ServiceAccountName string `json:"serviceAccountName"`
VolumeMountPath string `json:"volumeMountPath"`
PersistentVolumeClaimSpec corev1.PersistentVolumeClaimSpec `json:"persistentVolumeClaimSpec"`
PersistentVolumeSpec corev1.PersistentVolumeSpec `json:"persistentVolumeSpec"`
ImagePullPolicy corev1.PullPolicy `json:"imagePullPolicy,omitempty"`
Resource corev1.ResourceRequirements `json:"resources,omitempty"`
ServiceAccountName string `json:"serviceAccountName,omitempty"`
VolumeMountPath string `json:"volumeMountPath,omitempty"`
PersistentVolumeClaimSpec corev1.PersistentVolumeClaimSpec `json:"persistentVolumeClaimSpec,omitempty"`
PersistentVolumeSpec corev1.PersistentVolumeSpec `json:"persistentVolumeSpec,omitempty"`
}

// MetricsCollectorConfig is the JSON metrics collector structure in Katib config.
type MetricsCollectorConfig struct {
Image string `json:"image"`
ImagePullPolicy corev1.PullPolicy `json:"imagePullPolicy"`
Resource corev1.ResourceRequirements `json:"resources"`
Image string `json:"image"`
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we add omitempty for the fieds?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we can add it on all parameters but Image.

ImagePullPolicy corev1.PullPolicy `json:"imagePullPolicy,omitempty"`
Resource corev1.ResourceRequirements `json:"resources,omitempty"`
WaitAllProcesses *bool `json:"waitAllProcesses,omitempty"`
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you add omitempty to the SuggestionConfig also please ? It should be in all parameters but Image.

}

// EarlyStoppingConfig is the JSON early stopping structure in Katib config.
type EarlyStoppingConfig struct {
Image string `json:"image"`
ImagePullPolicy corev1.PullPolicy `json:"imagePullPolicy"`
ImagePullPolicy corev1.PullPolicy `json:"imagePullPolicy,omitempty"`
}

// GetSuggestionConfigData gets the config data for the given suggestion algorithm name.
Expand Down
9 changes: 6 additions & 3 deletions pkg/webhook/v1beta1/pod/inject_webhook.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,15 +204,15 @@ func (s *sidecarInjector) getMetricsCollectorContainer(trial *trialsv1beta1.Tria
newRule := rule.Name + ";" + rule.Value + ";" + string(rule.Comparison) + ";" + strconv.Itoa(rule.StartStep)
earlyStoppingRules = append(earlyStoppingRules, newRule)
}
metricsCollectorConfigData, err := katibconfig.GetMetricsCollectorConfigData(mc.Collector.Kind, s.client)

args, err := s.getMetricsCollectorArgs(trial, metricNames, mc, earlyStoppingRules)
args, err := s.getMetricsCollectorArgs(trial, metricNames, mc, metricsCollectorConfigData, earlyStoppingRules)
if err != nil {
return nil, err
}

sidecarContainerName := getSidecarContainerName(trial.Spec.MetricsCollector.Collector.Kind)

metricsCollectorConfigData, err := katibconfig.GetMetricsCollectorConfigData(mc.Collector.Kind, s.client)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -287,14 +287,17 @@ func (s *sidecarInjector) getKatibJob(object *unstructured.Unstructured, namespa
return jobKind, jobName, nil
}

func (s *sidecarInjector) getMetricsCollectorArgs(trial *trialsv1beta1.Trial, metricNames string, mc common.MetricsCollectorSpec, esRules []string) ([]string, error) {
func (s *sidecarInjector) getMetricsCollectorArgs(trial *trialsv1beta1.Trial, metricNames string, mc common.MetricsCollectorSpec, metricsCollectorConfigData katibconfig.MetricsCollectorConfig, esRules []string) ([]string, error) {
args := []string{"-t", trial.Name, "-m", metricNames, "-o-type", string(trial.Spec.Objective.Type), "-s-db", katibmanagerv1beta1.GetDBManagerAddr()}
if mountPath, _ := getMountPath(mc); mountPath != "" {
args = append(args, "-path", mountPath)
}
if mc.Source != nil && mc.Source.Filter != nil && len(mc.Source.Filter.MetricsFormat) > 0 {
args = append(args, "-f", strings.Join(mc.Source.Filter.MetricsFormat, ";"))
}
if metricsCollectorConfigData.WaitAllProcesses != nil {
args = append(args, "-w", strconv.FormatBool(*metricsCollectorConfigData.WaitAllProcesses))
}
// Add stop rules and service endpoint for Early Stopping
if len(esRules) > 0 {
for _, rule := range esRules {
Expand Down
17 changes: 15 additions & 2 deletions pkg/webhook/v1beta1/pod/inject_webhook_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package pod
import (
"context"
"fmt"
"github.com/kubeflow/katib/pkg/util/v1beta1/katibconfig"
"path/filepath"
"reflect"
"sync"
Expand Down Expand Up @@ -253,7 +254,7 @@ func TestGetMetricsCollectorArgs(t *testing.T) {
testMetricName := "accuracy"
katibDBAddress := fmt.Sprintf("katib-db-manager.%v:%v", testNamespace, consts.DefaultSuggestionPort)
katibEarlyStopAddress := fmt.Sprintf("%v-%v.%v:%v", testSuggestionName, testAlgorithm, testNamespace, consts.DefaultEarlyStoppingPort)

waitAllProcessesValue := false
testPath := "/test/path"

earlyStoppingRules := []string{
Expand Down Expand Up @@ -293,6 +294,7 @@ func TestGetMetricsCollectorArgs(t *testing.T) {
MetricNames string
MCSpec common.MetricsCollectorSpec
EarlyStoppingRules []string
KatibConfig katibconfig.MetricsCollectorConfig
ExpectedArgs []string
Name string
Err bool
Expand All @@ -305,12 +307,16 @@ func TestGetMetricsCollectorArgs(t *testing.T) {
Kind: common.StdOutCollector,
},
},
KatibConfig: katibconfig.MetricsCollectorConfig{
WaitAllProcesses: &waitAllProcessesValue,
},
ExpectedArgs: []string{
"-t", testTrialName,
"-m", testMetricName,
"-o-type", string(testObjective),
"-s-db", katibDBAddress,
"-path", common.DefaultFilePath,
"-w", "false",
},
Name: "StdOut MC",
},
Expand All @@ -333,6 +339,7 @@ func TestGetMetricsCollectorArgs(t *testing.T) {
},
},
},
KatibConfig: katibconfig.MetricsCollectorConfig{},
ExpectedArgs: []string{
"-t", testTrialName,
"-m", testMetricName,
Expand All @@ -356,6 +363,7 @@ func TestGetMetricsCollectorArgs(t *testing.T) {
},
},
},
KatibConfig: katibconfig.MetricsCollectorConfig{},
ExpectedArgs: []string{
"-t", testTrialName,
"-m", testMetricName,
Expand All @@ -373,6 +381,7 @@ func TestGetMetricsCollectorArgs(t *testing.T) {
Kind: common.CustomCollector,
},
},
KatibConfig: katibconfig.MetricsCollectorConfig{},
ExpectedArgs: []string{
"-t", testTrialName,
"-m", testMetricName,
Expand All @@ -394,6 +403,7 @@ func TestGetMetricsCollectorArgs(t *testing.T) {
},
},
},
KatibConfig: katibconfig.MetricsCollectorConfig{},
ExpectedArgs: []string{
"-t", testTrialName,
"-m", testMetricName,
Expand All @@ -411,6 +421,7 @@ func TestGetMetricsCollectorArgs(t *testing.T) {
Kind: common.PrometheusMetricCollector,
},
},
KatibConfig: katibconfig.MetricsCollectorConfig{},
ExpectedArgs: []string{
"-t", testTrialName,
"-m", testMetricName,
Expand All @@ -428,6 +439,7 @@ func TestGetMetricsCollectorArgs(t *testing.T) {
},
},
EarlyStoppingRules: earlyStoppingRules,
KatibConfig: katibconfig.MetricsCollectorConfig{},
ExpectedArgs: []string{
"-t", testTrialName,
"-m", testMetricName,
Expand All @@ -452,6 +464,7 @@ func TestGetMetricsCollectorArgs(t *testing.T) {
},
},
EarlyStoppingRules: earlyStoppingRules,
KatibConfig: katibconfig.MetricsCollectorConfig{},
Name: "Trial with invalid Experiment label name. Suggestion is not created",
Err: true,
},
Expand All @@ -465,7 +478,7 @@ func TestGetMetricsCollectorArgs(t *testing.T) {
}, timeout).ShouldNot(gomega.HaveOccurred())

for _, tc := range testCases {
args, err := si.getMetricsCollectorArgs(tc.Trial, tc.MetricNames, tc.MCSpec, tc.EarlyStoppingRules)
args, err := si.getMetricsCollectorArgs(tc.Trial, tc.MetricNames, tc.MCSpec, tc.KatibConfig, tc.EarlyStoppingRules)

if !tc.Err && err != nil {
t.Errorf("Case: %v failed. Expected nil, got %v", tc.Name, err)
Expand Down