V1alpha2 Metrics collector (part 1)

cmd/metricscollector/Dockerfile

@@ -5,9 +5,13 @@ FROM golang:alpine AS build-env
 ADD . /go/src/github.com/kubeflow/katib
 
 WORKDIR /go/src/github.com/kubeflow/katib/cmd/metricscollector
+
 # Build
 RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -a -o metricscollector ./v1alpha1
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -a -o metricscollector.v1alpha2 ./v1alpha2
+
 # Copy the controller-manager into a thin image
 FROM alpine:3.7
 WORKDIR /app
 COPY --from=build-env /go/src/github.com/kubeflow/katib/cmd/metricscollector/metricscollector .
+COPY --from=build-env /go/src/github.com/kubeflow/katib/cmd/metricscollector/metricscollector.v1alpha2 .

cmd/metricscollector/v1alpha2/main.go

@@ -0,0 +1,87 @@
+/*
+Copyright 2018 The Kubeflow Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+/*
+MetricsCollector is a default metricscollector for worker.
+It will collect metrics from pod log.
+You should print metrics in {{MetricsName}}={{MetricsValue}} format.
+For example, the objective value name is F1 and the metrics are loss, your training code should print like below.
+     ---
+     epoch 1:
+     batch1 loss=0.8
+     batch2 loss=0.6
+
+     F1=0.4
+
+     epoch 2:
+     batch1 loss=0.4
+     batch2 loss=0.2
+
+     F1=0.7
+     ---
+The metrics collector will collect all logs of metrics.
+*/
+
+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+	"strings"
+
+	api "github.com/kubeflow/katib/pkg/api/v1alpha2"
+	"github.com/kubeflow/katib/pkg/util/v1alpha2/metricscollector"
+
+	"google.golang.org/grpc"
+)
+
+var experimentName = flag.String("e", "", "Experiment Name")
+var trialName = flag.String("t", "", "Trial Name")
+var jobKind = flag.String("k", "", "Job Kind")
+var namespace = flag.String("n", "", "NameSpace")
+var managerService = flag.String("m", "", "Katib Manager service")
+var metricNames = flag.String("mn", "", "Metric names")
+
+func main() {
+	flag.Parse()
+	log.Printf("Experiment Name: %s, Trial Name: %s, Job Kind: %s", *experimentName, *trialName, *jobKind)
+	conn, err := grpc.Dial(*managerService, grpc.WithInsecure())
+	if err != nil {
+		log.Fatalf("could not connect: %v", err)
+	}
+	defer conn.Close()
+	c := api.NewManagerClient(conn)
+	mc, err := metricscollector.NewMetricsCollector()
+	if err != nil {
+		log.Fatalf("Failed to create MetricsCollector: %v", err)
+	}
+	ctx := context.Background()
+	olog, err := mc.CollectObservationLog(*trialName, *jobKind, strings.Split(*metricNames, ";"), *namespace)
+	if err != nil {
+		log.Fatalf("Failed to collect logs: %v", err)
+	}
+	reportreq := &api.ReportObservationLogRequest{
+		TrialName:      *trialName,
+		ObservationLog: olog,
+	}
+	_, err = c.ReportObservationLog(ctx, reportreq)
+	if err != nil {
+		log.Fatalf("Failed to Report logs: %v", err)
+	}
+	log.Printf("Metrics reported. :\n%v", olog)
+	return
+}

manifests/v1alpha2/katib-controller/mcrbac.yaml

@@ -0,0 +1,38 @@
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: metrics-collector
+rules:
+- apiGroups:
+  - ""
+  resources:
+  - pods
+  - pods/log
+  - pods/status
+  verbs:
+  - "*"
+- apiGroups:
+  - batch
+  resources:
+  - jobs
+  verbs:
+  - "*"
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: metrics-collector
+  namespace: kubeflow
+---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: metrics-collector
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: metrics-collector
+subjects:
+- kind: ServiceAccount
+  name: metrics-collector
+  namespace: kubeflow

manifests/v1alpha2/katib-controller/metricsControllerConfigMap.yaml

@@ -0,0 +1,40 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: metrics-collector-template
+  namespace: kubeflow
+data:
+  defaultMetricsCollectorTemplate.yaml : |-
+    apiVersion: batch/v1beta1
+    kind: CronJob
+    metadata:
+      name: {{.Trial}}
+      namespace: {{.NameSpace}}
+    spec:
+      schedule: "*/1 * * * *"
+      successfulJobsHistoryLimit: 0
+      failedJobsHistoryLimit: 1
+      jobTemplate:
+        spec:
+          backoffLimit: 0
+          template:
+            spec:
+              serviceAccountName: metrics-collector
+              containers:
+              - name: {{.Trial}}
+                image: katib/metrics-collector
+                args:
+                - "./metricscollector.v1alpha2"
+                - "-e"
+                - "{{.Experiment}}"
+                - "-t"
+                - "{{.Trial}}"
+                - "-k"
+                - "{{.JobKind}}"
+                - "-n"
+                - "{{.NameSpace}}"
+                - "-m"
+                - "{{.ManagerService}}"
+                - "-mn"
+                - "{{.MetricNames}}"
+              restartPolicy: Never

pkg/controller/v1alpha2/experiment/util/manifest_util.go

@@ -0,0 +1,71 @@
+// Copyright 2018 The Kubeflow Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package util
+
+import (
+	"bytes"
+	"fmt"
+	"text/template"
+	"strings"
+
+	"github.com/kubeflow/katib/pkg"
+	katibv1alpha2 "github.com/kubeflow/katib/pkg/api/operators/apis/experiment/v1alpha2"
+	"github.com/kubeflow/katib/pkg/util/v1alpha2/katibclient"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func getMetricsCollectorManifest(experimentName string, trialName string, jobKind string, namespace string, metricNames []string, mcs *katibv1alpha2.MetricsCollectorSpec) (*bytes.Buffer, error) {
+	var mtp *template.Template = nil
+	var err error
+	tmpValues := map[string]string{
+		"Experiment":     experimentName,
+		"Trial":          trialName,
+		"JobKind":        jobKind,
+		"NameSpace":      namespace,
+		"ManagerService": pkg.GetManagerAddr(),
+		"MetricNames":    strings.Join(metricNames, ";"),
+	}
+	if mcs != nil && mcs.GoTemplate.RawTemplate != "" {
+		mtp, err = template.New("MetricsCollector").Parse(mcs.GoTemplate.RawTemplate)
+	} else {
+		mctp := "defaultMetricsCollectorTemplate.yaml"
+		if mcs != nil && mcs.GoTemplate.TemplateSpec != nil {
+			mctp = mcs.GoTemplate.TemplateSpec.TemplatePath
+		}
+		kc, err := katibclient.NewClient(client.Options{})
+		if err != nil {
+			return nil, err
+		}
+		mtl, err := kc.GetMetricsCollectorTemplates()
+		if err != nil {
+			return nil, err
+		}
+		if mt, ok := mtl[mctp]; !ok {
+			return nil, fmt.Errorf("No MetricsCollector template name %s", mctp)
+		} else {
+			mtp, err = template.New("MetricsCollector").Parse(mt)
+		}
+	}
+	if err != nil {
+		return nil, err
+	}
+	var b bytes.Buffer
+	err = mtp.Execute(&b, tmpValues)
+	if err != nil {
+		return nil, err
+	}
+	return &b, nil
+}

pkg/controller/v1alpha2/experiment/util/runjob_util.go

@@ -19,11 +19,10 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
-
-func GetSupportdJobList() []schema.GroupVersionKind {
-	// TODO: append other supported jobs, such as tfjob, pytorch and so on 
-	supportedJobList := []schema.GroupVersionKind {
-		schema.GroupVersionKind {
+func GetSupportedJobList() []schema.GroupVersionKind {
+	// TODO: append other supported jobs, such as tfjob, pytorch and so on
+	supportedJobList := []schema.GroupVersionKind{
+		schema.GroupVersionKind{
 			Group:   "batch",
 			Version: "v1",
 			Kind:    "Job",

pkg/controller/v1alpha2/experiment/util/webhook_util.go

@@ -20,8 +20,10 @@ import (
 	"bytes"
 	"database/sql"
 	"fmt"
+	logger "log"
 
 	ep_v1alpha2 "github.com/kubeflow/katib/pkg/api/operators/apis/experiment/v1alpha2"
+	batchv1beta "k8s.io/api/batch/v1beta1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	k8syaml "k8s.io/apimachinery/pkg/util/yaml"
 )
@@ -54,6 +56,11 @@ func ValidateExperiment(instance *ep_v1alpha2.Experiment) error {
 	if err := validateAlgorithmSettings(instance); err != nil {
 		return err
 	}
+
+	if err := validateMetricsCollector(instance); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -121,7 +128,7 @@ func validateTrialTemplate(instance *ep_v1alpha2.Experiment) error {
 
 func validateSupportedJob(job *unstructured.Unstructured) error {
 	gvk := job.GroupVersionKind()
-	supportedJobs := GetSupportdJobList()
+	supportedJobs := GetSupportedJobList()
 	for _, sJob := range supportedJobs {
 		if gvk == sJob {
 			return nil
@@ -140,3 +147,49 @@ func validateForCreate(inst *ep_v1alpha2.Experiment) error {
 		return fmt.Errorf("Record for the experiment has existed in DB; Please try to rename the experiment")
 	}
 }
+
+func validateMetricsCollector(inst *ep_v1alpha2.Experiment) error {
+	BUFSIZE := 1024
+	experimentName := inst.GetName()
+	trialName := fmt.Sprintf("%s-trial", inst.GetName())
+	namespace := inst.GetNamespace()
+	trialParams := TrialTemplateParams{
+		Experiment: experimentName,
+		Trial:      trialName,
+		NameSpace:  namespace,
+	}
+	var metricNames []string
+	metricNames = append(metricNames, inst.Spec.Objective.ObjectiveMetricName)
+	for _, mn := range inst.Spec.Objective.AdditionalMetricsNames {
+		metricNames = append(metricNames, mn)
+	}
+
+	runSpec, err := GetRunSpec(inst, trialParams)
+	if err != nil {
+		return fmt.Errorf("Invalid spec.trialTemplate: %v.", err)
+	}
+
+	buf := bytes.NewBufferString(runSpec)
+
+	job := &unstructured.Unstructured{}
+	if err := k8syaml.NewYAMLOrJSONDecoder(buf, BUFSIZE).Decode(job); err != nil {
+		return fmt.Errorf("Invalid spec.trialTemplate: %v.", err)
+	}
+
+	var mcjob batchv1beta.CronJob
+	mcm, err := getMetricsCollectorManifest(experimentName, trialName, job.GetKind(), namespace, metricNames, inst.Spec.MetricsCollectorSpec)
+	if err != nil {
+		logger.Printf("getMetricsCollectorManifest error %v", err)
+		return err
+	}
+
+	if err := k8syaml.NewYAMLOrJSONDecoder(mcm, BUFSIZE).Decode(&mcjob); err != nil {
+		logger.Printf("MetricsCollector Yaml decode error %v", err)
+		return err
+	}
+
+	if mcjob.GetNamespace() != namespace || mcjob.GetName() != trialName {
+		return fmt.Errorf("Invalid metricsCollector template.")
+	}
+	return nil
+}