Repository: odahu/odahu-flow (language: python)

Issue:
New Async Batch Inference API 
- [x] API Server CRUD, Validation, Persistence
- [x] Spec/Status Syncer API Server <-> Kubernetes
- [x] Kubernetes Operator of BatchInferenceJob (Convert to TaskRun)
- [x] Pipeline implemented as TaskRun
  - [x] Tool to convert ODAHU Connection to Rclone config
  - [x] Tool to send local jsons to model log storage
  - [x] Step 1: Generate Rclone Config
  - [x] Step 2: Sync request data and model
  - [x] Step 3: Validate and filter request data for next step (only .json files) in kubeflow format
  - [x] Step 4: Dump request data to model log (Enrich request id if required) 
  - [x] Step 5: Run user container with appropriate configuration
  - [x] Step 6: Upload results to model log
- [x] Python SDK
- [x] Airflow Operator

Out of scope:
-  UI
-  CLI
-  Tensor types and shape validation (e.g. using config.pbtxt)

![image](https://user-images.githubusercontent.com/30889080/109056825-013efb00-76f2-11eb-97b2-97ac61e80c6a.png)




Discussion / hints:

Fix patch:
diff --git a/Makefile b/Makefile
index 87b846995..05402c1d2 100644
--- a/Makefile
+++ b/Makefile
@@ -85,6 +85,10 @@ docker-build-api:
 docker-build-controller:
 	docker build --target controller -t odahu/odahu-flow-controller:${BUILD_TAG} -f containers/operator/Dockerfile .
 
+## docker-build-tools: Build tools docker image
+docker-build-tools:
+	docker build --target tools -t odahu/odahu-flow-tools:${BUILD_TAG} -f containers/operator/Dockerfile .
+
 ## docker-build-model-trainer: Build model builder docker image
 docker-build-model-trainer:
 	docker build --target model-trainer -t odahu/odahu-flow-model-trainer:${BUILD_TAG} -f containers/operator/Dockerfile .
@@ -127,6 +131,11 @@ docker-push-controller:  check-tag
 	docker tag odahu/odahu-flow-controller:${BUILD_TAG} ${DOCKER_REGISTRY}/odahu/odahu-flow-controller:${TAG}
 	docker push ${DOCKER_REGISTRY}/odahu/odahu-flow-controller:${TAG}
 
+## docker-push-tools: Push tools docker image
+docker-push-tools:  check-tag
+	docker tag odahu/odahu-flow-tools:${BUILD_TAG} ${DOCKER_REGISTRY}/odahu/odahu-flow-tools:${TAG}
+	docker push ${DOCKER_REGISTRY}/odahu/odahu-flow-tools:${TAG}
+
 ## docker-push-model-packager: Push model packager docker image
 docker-push-model-packager:  check-tag
 	docker tag odahu/odahu-flow-model-packager:${BUILD_TAG} ${DOCKER_REGISTRY}/odahu/odahu-flow-model-packager:${TAG}
diff --git a/containers/operator/Dockerfile b/containers/operator/Dockerfile
index fffcd8967..a792b0e9d 100644
--- a/containers/operator/Dockerfile
+++ b/containers/operator/Dockerfile
@@ -66,6 +66,24 @@ WORKDIR "${ODAHUFLOW_DIR}"
 CMD ["./webserver"]
 
 
+#########################################################
+#########################################################
+#########################################################
+
+FROM ubuntu:18.04 as tools
+
+ENV ODAHUFLOW_DIR="/opt/odahu-flow"
+
+RUN apt-get -yq update && \
+    apt-get -yqq install openssh-client ca-certificates && \
+    apt-get clean && rm -rf /var/lib/apt/lists/*
+
+COPY --from=builder /app/odahu-tools "${ODAHUFLOW_DIR}/"
+
+WORKDIR "${ODAHUFLOW_DIR}"
+CMD ["./odahu-tools"]
+
+
 #########################################################
 #########################################################
 #########################################################
diff --git a/packages/commons/feedback/feedback.go b/packages/commons/feedback/feedback.go
new file mode 100644
index 000000000..2562787bd
--- /dev/null
+++ b/packages/commons/feedback/feedback.go
@@ -0,0 +1,31 @@
+package feedback
+
+type RequestResponse struct {
+	RequestID           string            `msg:"request_id"`
+	RequestHttpHeaders  map[string]string `msg:"request_http_headers"`
+	RequestContent      string            `msg:"request_content"`
+	RequestUri          string            `msg:"request_uri"`
+	ResponseStatus      string            `msg:"response_status"`
+	ResponseHttpHeaders map[string]string `msg:"response_http_headers"`
+	RequestHost         string            `msg:"request_host"`
+	ModelEndpoint       string            `msg:"model_endpoint"`
+	ModelVersion        string            `msg:"model_version"`
+	ModelName           string            `msg:"model_name"`
+	RequestHttpMethod   string            `msg:"request_http_method"`
+}
+
+type ResponseBody struct {
+	RequestID       string `msg:"request_id"`
+	ModelEndpoint   string `msg:"model_endpoint"`
+	ModelVersion    string `msg:"model_version"`
+	ModelName       string `msg:"model_name"`
+	ResponseContent string `msg:"response_content"`
+}
+
+type ModelFeedback struct {
+	RequestID    string                 `msg:"request_id"`
+	ModelVersion string                 `msg:"model_version"`
+	ModelName    string                 `msg:"model_name"`
+	Payload      map[string]interface{} `msg:"payload"`
+}
+
diff --git a/packages/feedback/go.sum b/packages/feedback/go.sum
index ef9adc04e..432f7a606 100644
--- a/packages/feedback/go.sum
+++ b/packages/feedback/go.sum
@@ -2169,6 +2169,7 @@ gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20190709130402-674ba3eaed22/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776 h1:tQIYjPdBoyREyB9XMu+nnTclpTYkz2zFM+lzLJFO4gQ=
 gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/packages/feedback/pkg/collector/handlers.go b/packages/feedback/pkg/collector/handlers.go
index 06203c2f4..e4376b18b 100644
--- a/packages/feedback/pkg/collector/handlers.go
+++ b/packages/feedback/pkg/collector/handlers.go
@@ -21,6 +21,7 @@ import (
 	"github.com/odahu/odahu-flow/packages/feedback/pkg/feedback"
 	"net/http"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	commons_feedback "odahu-commons/feedback"
 
 	"github.com/gin-gonic/gin"
 )
@@ -35,7 +36,7 @@ func handleFeedbackEndpoint(c *gin.Context) {
 	modelName := c.GetHeader(feedback.ModelNameHeaderKey)
 	modelVersion := c.GetHeader(feedback.ModelVersionHeaderKey)
 
-	message := feedback.ModelFeedback{
+	message := commons_feedback.ModelFeedback{
 		ModelName:    modelName,
 		ModelVersion: modelVersion,
 	}
diff --git a/packages/feedback/pkg/collector/http.go b/packages/feedback/pkg/collector/http.go
index 76e7d08c3..f67ee55ff 100644
--- a/packages/feedback/pkg/collector/http.go
+++ b/packages/feedback/pkg/collector/http.go
@@ -19,6 +19,7 @@ package collector
 import (
 	"fmt"
 	"github.com/odahu/odahu-flow/packages/feedback/pkg/feedback"
+	commons_feedback "odahu-commons/feedback"
 	"github.com/spf13/viper"
 	"github.com/zsais/go-gin-prometheus"
 	"log"
@@ -36,7 +37,7 @@ const (
 )
 
 type FeedbackResponse struct {
-	Message feedback.ModelFeedback `json:"message"`
+	Message commons_feedback.ModelFeedback `json:"message"`
 }
 
 func attachRoutes(router *gin.Engine) {
diff --git a/packages/feedback/pkg/feedback/types.go b/packages/feedback/pkg/feedback/types.go
index 5a8015a51..2342e09b1 100644
--- a/packages/feedback/pkg/feedback/types.go
+++ b/packages/feedback/pkg/feedback/types.go
@@ -12,35 +12,6 @@ const (
 	ForwardedHostHeaderKey      = "x-forwarded-host"
 )
 
-type RequestResponse struct {
-	RequestID           string            `msg:"request_id"`
-	RequestHttpHeaders  map[string]string `msg:"request_http_headers"`
-	RequestContent      string            `msg:"request_content"`
-	RequestUri          string            `msg:"request_uri"`
-	ResponseStatus      string            `msg:"response_status"`
-	ResponseHttpHeaders map[string]string `msg:"response_http_headers"`
-	RequestHost         string            `msg:"request_host"`
-	ModelEndpoint       string            `msg:"model_endpoint"`
-	ModelVersion        string            `msg:"model_version"`
-	ModelName           string            `msg:"model_name"`
-	RequestHttpMethod   string            `msg:"request_http_method"`
-}
-
-type ResponseBody struct {
-	RequestID       string `msg:"request_id"`
-	ModelEndpoint   string `msg:"model_endpoint"`
-	ModelVersion    string `msg:"model_version"`
-	ModelName       string `msg:"model_name"`
-	ResponseContent string `msg:"response_content"`
-}
-
-type ModelFeedback struct {
-	RequestID    string                 `msg:"request_id"`
-	ModelVersion string                 `msg:"model_version"`
-	ModelName    string                 `msg:"model_name"`
-	Payload      map[string]interface{} `msg:"payload"`
-}
-
 type DataLogging interface {
 	Post(tag string, message interface{}) error
 	Close() error
diff --git a/packages/feedback/pkg/tapping/collector.go b/packages/feedback/pkg/tapping/collector.go
index c7b46a4a8..60fef146f 100644
--- a/packages/feedback/pkg/tapping/collector.go
+++ b/packages/feedback/pkg/tapping/collector.go
@@ -11,6 +11,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"net/url"
+	commons_feedback "odahu-commons/feedback"
 	"odahu-commons/predictors"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"time"
@@ -82,9 +83,9 @@ func NewRequestCollector(
 	}, nil
 }
 
-func (rc *RequestCollector) convertToFeedback(message *Message) (*feedback.RequestResponse, *feedback.ResponseBody, error) {
-	responseBody := &feedback.ResponseBody{}
-	requestResponse := &feedback.RequestResponse{}
+func (rc *RequestCollector) convertToFeedback(message *Message) (*commons_feedback.RequestResponse, *commons_feedback.ResponseBody, error) {
+	responseBody := &commons_feedback.ResponseBody{}
+	requestResponse := &commons_feedback.RequestResponse{}
 
 	requestHeaders := make(map[string]string, len(message.HttpBufferedTrace.Request.Headers))
 	for _, header := range message.HttpBufferedTrace.Request.Headers {
diff --git a/packages/operator/Makefile b/packages/operator/Makefile
index cd0639e91..33085ed7e 100644
--- a/packages/operator/Makefile
+++ b/packages/operator/Makefile
@@ -42,6 +42,10 @@ build-operator: generate
 build-webserver:
 	go build -o webserver ./cmd/api
 
+## build-odahu-tools: Build odahu-tools executable binary
+build-tools:
+	go build -o odahu-tools ./cmd/tools
+
 ## build-controller: Build odahu system controller
 build-controller:
 	go build -o controller ./cmd/controller
@@ -59,7 +63,7 @@ build-service-catalog:
 	go build -o service-catalog ./cmd/service_catalog
 
 ## build-all: Build all executable binyries
-build-all: build-webserver build-controller build-model-trainer build-model-packager build-operator build-service-catalog
+build-all: build-webserver build-controller build-model-trainer build-model-packager build-operator build-service-catalog build-tools
 
 ## run-operator: Start operator
 run-operator:
@@ -156,6 +160,10 @@ static-content:
 	statik -src static -dest pkg -p static
 	git add pkg/static
 
+## generate-predict-v2-types: Generate structures according to OpenAPI 3.0 Spec of Kubeflow predict v2 API
+generate-predict-v2-types:
+	oapi-codegen -generate types -package predict_v2 hack/kubeflow_prediction_v2/spec.yaml > pkg/apis/predict_v2/predict_v2.gen.go
+
 ## help: Show the help message
 help: Makefile
 	@echo "Choose a command run in "$(PROJECTNAME)":"
diff --git a/packages/operator/cmd/tools/cmd/auth.go b/packages/operator/cmd/tools/cmd/auth.go
new file mode 100644
index 000000000..63059bde9
--- /dev/null
+++ b/packages/operator/cmd/tools/cmd/auth.go
@@ -0,0 +1,60 @@
+package cmd
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/spf13/cobra"
+	connAPI "github.com/odahu/odahu-flow/packages/operator/pkg/apiclient/connection"
+	connTools "github.com/odahu/odahu-flow/packages/operator/pkg/utils/connections"
+)
+
+const (
+	configureRCloneCommandUsage = `
+List of ODAHU Connection IDs that must exist in ODAHU Cluster 
+Each connection ID can be suffixed by :<RClone Storage Name>. 
+If connection ID is not suffixed then RClone storage name is generated as "odahu-<Connection ID>".
+For example next flags:
+--conn model-output:output --conn input-data 
+will lead to adding two storages to Rclone config with names accordingly:
+[output]
+...
+[odahu-input-data]
+...`
+)
+
+var (
+	fileValues    []string
+	clusterValues []string
+)
+
+func init() {
+	rootCmd.AddCommand(authCommand)
+	authCommand.AddCommand(configureRCloneCommand)
+	configureRCloneCommand.Flags().StringArrayVar(&clusterValues, "conn", []string{}, configureRCloneCommandUsage)
+}
+
+var authCommand = &cobra.Command{
+	Use:  "auth",
+	Short: "Support tools to work with Connections",
+	Run: func(cmd *cobra.Command, args []string) {
+		if len(args) == 0 {
+			_ = cmd.Help()
+			os.Exit(0)
+		}
+	},
+}
+
+var configureRCloneCommand = &cobra.Command{
+	Use:  "configure-rclone",
+	Short: `Generate rclone configuration based on ODAHU object storage connection`,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		client := connAPI.NewClient(cfg.Auth.APIURL, "",
+			cfg.Auth.ClientID, cfg.Auth.ClientSecret, cfg.Auth.OAuthOIDCTokenEndpoint)
+		err := connTools.GenerateRClone(clusterValues, fileValues, client)
+		if err != nil {
+			return fmt.Errorf("unable to configure RClone: %s", err)
+		}
+		return nil
+	},
+}
diff --git a/packages/operator/cmd/tools/cmd/batch_log.go b/packages/operator/cmd/tools/cmd/batch_log.go
new file mode 100644
index 000000000..7813affe0
--- /dev/null
+++ b/packages/operator/cmd/tools/cmd/batch_log.go
@@ -0,0 +1,250 @@
+package cmd
+
+import (
+	"gopkg.in/yaml.v2"
+	"fmt"
+	"github.com/fluent/fluent-logger-golang/fluent"
+	"github.com/odahu/odahu-flow/packages/operator/pkg/apis/predict_v2"
+	"github.com/odahu/odahu-flow/packages/operator/pkg/utils/feedback"
+	"io/ioutil"
+	feedback_utils "odahu-commons/feedback"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+	"go.uber.org/zap"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strconv"
+)
+
+// Model log commands
+
+const (
+	defaultRequestTag = "request_response"
+	defaultFluentdHost = "localhost"
+	defaultFluentdPort = 24224
+	defaultResponseTag = "response_body"
+	maxRetryToDeliver = 100
+	maxRetryWait      = 1000
+)
+
+var (
+	requestTag string
+	responseTag string
+	model string
+	requestID string
+)
+
+func init () {
+	batchCommand.AddCommand(logCommand)
+	logCommand.AddCommand(logModelInputCommand)
+	logCommand.AddCommand(logModelOutputCommand)
+	logCommand.PersistentFlags().StringVarP(
+		&model, "model", "m", ".", "directory with ML model files",
+	)
+	_ = logCommand.MarkFlagRequired("model")
+	logCommand.PersistentFlags().StringVarP(
+		&requestID, "request-id", "r", "",
+		"request id for which this request/response data is logged",
+	)
+	_ = logCommand.MarkFlagRequired("request-id")
+
+	logCommand.PersistentFlags().StringVar(
+		&apiURL, "fluentd", "", "fluentd base URL (schema://host:port)",
+	)
+	_ = viper.BindPFlag("feedback.fluentd.baseurl", logCommand.PersistentFlags().Lookup("fluentd"))
+
+	logModelInputCommand.Flags().StringVar(&requestTag, "tag", defaultRequestTag, "tag model request")
+	logModelOutputCommand.Flags().StringVar(&responseTag, "tag", defaultResponseTag, "tag model response")
+
+}
+
+type ModelOutputLogger interface {
+	Log(requestID string, request predict_v2.InferenceResponse) error
+}
+
+var logCommand = &cobra.Command{
+	Use:  "log",
+	Short: "Catch model input or output from json files to fluentd service",
+	Run: func(cmd *cobra.Command, args []string) {
+		if len(args) == 0 {
+			_ = cmd.Help()
+			os.Exit(0)
+		}
+	},
+}
+
+func initFluentd() (*fluent.Fluent, error) {
+
+	var host string
+	var port int
+	rawBaseURL := cfg.Feedback.Fluentd.BaseURL
+	if rawBaseURL == "" {
+		host = defaultFluentdHost
+		port = defaultFluentdPort
+	} else {
+		baseURL, err := url.Parse(rawBaseURL)
+		if err != nil {
+			return nil, fmt.Errorf("unable to parse fluend base url: %s", rawBaseURL)
+		}
+		host = baseURL.Hostname()
+		portString := baseURL.Port()
+		if portString != "" {
+			port, err = strconv.Atoi(portString)
+			if err != nil {
+				return nil, fmt.Errorf("fluentd port must be integer %s", portString)
+			}
+		} else {
+			port = defaultFluentdPort
+		}
+	}
+
+	zap.S().Infof("Connecting to fluentd using host %s and port %d", host, port)
+	f, err := fluent.New(fluent.Config{
+		FluentPort:   port,
+		FluentHost:   host,
+		MaxRetry:     maxRetryToDeliver,
+		Async:        true,
+		MaxRetryWait: maxRetryWait,
+	})
+	return f, err
+
+
+}
+
+func getRequestWrapper(modelName string, modelVersion string) func(content string)interface{}{
+	return func(content string) interface{} {
+		return feedback_utils.RequestResponse{
+			RequestID:           requestID,
+			RequestContent:      content,
+			ModelVersion:        modelVersion,
+			ModelName:           modelName,
+		}
+	}
+}
+func getResponseWrapper(modelName string, modelVersion string) func(content string)interface{}{
+	return func(content string) interface{} {
+		return feedback_utils.ResponseBody{
+			RequestID:       requestID,
+			ModelVersion:    modelVersion,
+			ModelName:       modelName,
+			ResponseContent: content,
+		}
+	}
+}
+
+func getModelNameVersion() (string, string, error) {
+
+	type Model struct {
+		Name string `json:"name"`
+		Version string `json:"version"`
+	}
+
+	type ModelFile struct {
+		Model Model `json:"model"`
+	}
+
+	modelFiles := []string{"odahuflow.project.yaml", "odahuflow.project.yml"}
+
+	items, err := ioutil.ReadDir(model)
+	if err != nil {
+		return "", "", err
+
+	}
+	for _, item := range items {
+		file := item.Name()
+		for _, mFile := range modelFiles {
+			if mFile == file {
+
+				fp := filepath.Join(model, file)
+				zap.S().Infof("Model metadata file is found %s", fp)
+				data, err := ioutil.ReadFile(fp)
+				if err != nil {
+					return "", "", err
+				}
+				mf := ModelFile{}
+				if err := yaml.Unmarshal(data, &mf); err != nil {
+					return "", "", err
+				}
+
+				return mf.Model.Name, mf.Model.Version, nil
+			}
+		}
+	}
+	return "", "", fmt.Errorf("unable to find model metadata file")
+}
+
+var logModelInputCommand = &cobra.Command{
+	Use:   "input",
+	Short: "log model input to feedback storage",
+	Args:  cobra.ExactArgs(1),
+	Example: "odahu-tools batch log input <path-to-folder-with-json-files>",
+	RunE: func(cmd *cobra.Command, args []string) error {
+		logEngine, err := initFluentd()
+		if err != nil {
+			return err
+		}
+		defer func() {
+			if err := logEngine.Close(); err != nil {
+				zap.S().Errorw("Error closing fluentd", zap.Error(err))
+			} else {
+				zap.S().Info("Fluentd logs are flushed")
+			}
+		}()
+
+		dataLogger := feedback.NewLogger(logEngine)
+
+		modelName, modelVer, err := getModelNameVersion()
+		if err != nil {
+			return err
+		}
+
+		wrap := getRequestWrapper(modelName, modelVer)
+
+		for _, source := range args {
+			zap.S().Infof("Handle %s directory", source)
+			if err := dataLogger.LogDir(source, requestTag, wrap); err != nil {
+				zap.S().Errorw("Error during logging model input", zap.Error(err))
+				return err
+			}
+		}
+		return nil
+	},
+}
+
+var logModelOutputCommand = &cobra.Command{
+	Use:                        "output",
+	Short: "log model output to feedback storage",
+	Args: cobra.ExactArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		logEngine, err := initFluentd()
+		if err != nil {
+			return err
+		}
+		defer func() {
+			if err := logEngine.Close(); err != nil {
+				zap.S().Errorw("Error closing fluentd", zap.Error(err))
+			} else {
+				zap.S().Info("Fluentd logs are flushed")
+			}
+		}()
+
+		dataLogger := feedback.NewLogger(logEngine)
+
+		modelName, modelVer, err := getModelNameVersion()
+		if err != nil {
+			return err
+		}
+
+		wrap := getResponseWrapper(modelName, modelVer)
+
+		for _, source := range args {
+			zap.S().Infof("Handle %s directory", source)
+			if err := dataLogger.LogDir(source, responseTag, wrap); err != nil {
+				zap.S().Errorw("Error during logging model output", zap.Error(err))
+				return err
+			}
+		}
+		return nil
+	},
+}
\ No newline at end of file
diff --git a/packages/operator/cmd/tools/cmd/batch_root.go b/packages/operator/cmd/tools/cmd/batch_root.go
new file mode 100644
index 000000000..2b36574cd
--- /dev/null
+++ b/packages/operator/cmd/tools/cmd/batch_root.go
@@ -0,0 +1,22 @@
+package cmd
+
+import (
+	"github.com/spf13/cobra"
+	"os"
+)
+
+func init() {
+	rootCmd.AddCommand(batchCommand)
+}
+
+
+var batchCommand = &cobra.Command{
+	Use:  "batch",
+	Short: "Support tools to prepare environment for batch inference",
+	Run: func(cmd *cobra.Command, args []string) {
+		if len(args) == 0 {
+			_ = cmd.Help()
+			os.Exit(0)
+		}
+	},
+}
\ No newline at end of file
diff --git a/packages/operator/cmd/tools/cmd/batch_validate.go b/packages/operator/cmd/tools/cmd/batch_validate.go
new file mode 100644
index 000000000..a362ba878
--- /dev/null
+++ b/packages/operator/cmd/tools/cmd/batch_validate.go
@@ -0,0 +1,85 @@
+package cmd
+
+import (
+	predict_v2_tools "github.com/odahu/odahu-flow/packages/operator/pkg/utils/predictv2"
+	"github.com/spf13/cobra"
+	"go.uber.org/zap"
+	"os"
+	"runtime"
+)
+
+const inputCommandUsage = `
+Path to directory that contains set of files with .json extensions in "Kubeflow Predict V2 Inference Request" format
+`
+const outputCommandUsage = `
+Path to directory where json files that passed validation will be dumped
+`
+
+var (
+	source      string
+	destination string
+	cpuprofile string
+)
+
+func init() {
+	batchCommand.AddCommand(validateCommand)
+	validateCommand.AddCommand(validateInputCommand)
+	validateCommand.AddCommand(validateOutputCommand)
+	validateCommand.AddCommand(validateModelCommand)
+	validateCommand.PersistentFlags().StringVarP(
+		&source, "source", "s", ".", inputCommandUsage,
+	)
+	validateCommand.PersistentFlags().StringVarP(
+		&destination, "destination", "d", ".", outputCommandUsage,
+	)
+}
+
+
+var validateCommand = &cobra.Command{
+	Use:  "validate",
+	Short: "Validate that json files have expected format",
+	Run: func(cmd *cobra.Command, args []string) {
+		if len(args) == 0 {
+			_ = cmd.Help()
+			os.Exit(0)
+		}
+	},
+}
+
+var validateInputCommand = &cobra.Command{
+	Use:  "input",
+	Short: "validate input for user batch inference container",
+	Run: func(cmd *cobra.Command, args []string) {
+
+		numCPU := runtime.NumCPU()
+		zap.S().Infof("Number of CPUs: %d", numCPU)
+
+		_, err := predict_v2_tools.ValidateDir(source, predict_v2_tools.ValidateInput, &destination)
+		if err != nil {
+			zap.S().Errorw("There are errors during validation", zap.Error(err))
+			os.Exit(1)
+		}
+	},
+}
+
+var validateOutputCommand = &cobra.Command{
+	Use:  "output",
+	Short: "validate output of user batch inference container",
+	Run: func(cmd *cobra.Command, args []string) {
+
+		_, err := predict_v2_tools.ValidateDir(source, predict_v2_tools.ValidateOutput, &destination)
+		if err != nil {
+			zap.S().Errorw("There are errors during validation", zap.Error(err))
+			os.Exit(1)
+		}
+	},
+}
+
+var validateModelCommand = &cobra.Command{
+	Use:                        "validate-model",
+	Short: "validate model for batch inference",
+	Run: func(cmd *cobra.Command, args []string) {
+
+	},
+}
+
diff --git a/packages/operator/cmd/tools/cmd/root.go b/packages/operator/cmd/tools/cmd/root.go
new file mode 100644
index 000000000..84996ad31
--- /dev/null
+++ b/packages/operator/cmd/tools/cmd/root.go
@@ -0,0 +1,122 @@
+package cmd
+
+import (
+	"github.com/mitchellh/go-homedir"
+	"github.com/odahu/odahu-flow/packages/operator/pkg/config"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+	"go.uber.org/zap"
+	"log"
+	"os"
+	"runtime/pprof"
+)
+
+var (
+	cfgFile string
+	cfg config.ToolsConfig
+	profilingFile *os.File
+	apiURL string
+)
+
+var rootCmd = &cobra.Command{
+	Use:   "odahu-tools",
+	Short: "odahu-tools is a simple command line tool that provides API to the set of ODAHU platform features",
+	Long: `odahu-tools provides API to execute the same logic that is used by the ODAHU platform in the cluster`,
+	SilenceUsage: true,
+	Run: func(cmd *cobra.Command, args []string) {
+		if len(args) == 0 {
+			_ = cmd.Help()
+			os.Exit(0)
+		}
+	},
+	PersistentPreRun: func(cmd *cobra.Command, args []string) {
+		if cpuprofile != "" {
+			var err error
+			profilingFile, err = os.Create(cpuprofile)
+			if err != nil {
+				zap.S().Fatal("could not create CPU profile: ", err)
+			}
+			if err := pprof.StartCPUProfile(profilingFile); err != nil {
+				zap.S().Fatal("could not start CPU profile: ", err)
+			}
+		}
+	},
+	PersistentPostRun: func(cmd *cobra.Command, args []string) {
+		if cpuprofile != "" {
+			pprof.StopCPUProfile()
+			if profilingFile != nil {
+				if err := profilingFile.Close(); err != nil {
+					zap.S().Fatal(err)
+				}
+			}
+		}
+
+	},
+}
+
+func initLogger() {
+	logger, err := zap.NewDevelopment()
+	if err != nil {
+		log.Fatal("Unable to initialize logger")
+	}
+	defer func() {
+		_ = logger.Sync()
+	}()
+
+	zap.ReplaceGlobals(logger)
+}
+
+func initConfig() {
+	// Don't forget to read config either from cfgFile or from home directory!
+	if cfgFile != "" {
+		// Use config file from the flag.
+		viper.SetConfigFile(cfgFile)
+	} else {
+		// Find home directory.
+		home, err := homedir.Dir()
+		if err != nil {
+			zap.S().Errorw("Unable to get home directory", zap.Error(err))
+			panic("Unable to get home directory")
+		}
+
+		// Search config in home directory with name ".cobra" (without extension).
+		viper.AddConfigPath(home)
+		viper.SetConfigName(".odahu-tools")
+	}
+
+	if err := viper.ReadInConfig(); err != nil {
+		zap.S().Errorw("Can't read config", zap.Error(err))
+		panic("Can't read config")
+	}
+	if err := viper.Unmarshal(&cfg); err != nil {
+		zap.S().Errorw("Unable to unmarshall config", zap.Error(err))
+		panic(err)
+	}
+
+}
+
+func init() {
+	initLogger()
+	cobra.OnInitialize(initConfig)
+	rootCmd.PersistentFlags().StringVar(
+		&cfgFile, "config", "", "config file (default is $HOME/.odahu-tools.yaml)",
+	)
+	rootCmd.PersistentFlags().StringVar(
+		&cpuprofile, "cpuprofile", "",
+		"if specified as TARGET cpu profiling results will be saved to TARGET file",
+	)
+	rootCmd.PersistentFlags().StringVar(
+		&apiURL, "api", "", "API server base URL (schema://host:port)",
+	)
+	_ = viper.BindPFlag("auth.apiUrl", rootCmd.PersistentFlags().Lookup("api"))
+	// These env variables are compatible with odahu-flow-sdk package
+	_ = viper.BindEnv("auth.apiUrl", "API_URL")
+	_ = viper.BindEnv("auth.apiToken", "API_TOKEN")
+}
+
+func Execute() {
+
+	if err := rootCmd.Execute(); err != nil {
+		os.Exit(1)
+	}
+}
\ No newline at end of file
diff --git a/packages/operator/cmd/tools/main.go b/packages/operator/cmd/tools/main.go
new file mode 100644
index 000000000..7b40c89e0
--- /dev/null
+++ b/packages/operator/cmd/tools/main.go
@@ -0,0 +1,11 @@
+package main
+
+import (
+	"github.com/odahu/odahu-flow/packages/operator/cmd/tools/cmd"
+)
+
+
+func main() {
+
+	cmd.Execute()
+}
\ No newline at end of file
diff --git a/packages/operator/go.mod b/packages/operator/go.mod
index 5b4d7919b..9c05aec40 100644
--- a/packages/operator/go.mod
+++ b/packages/operator/go.mod
@@ -13,6 +13,8 @@ require (
 	github.com/banzaicloud/bank-vaults/pkg/sdk v0.3.1
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/emicklei/go-restful v2.9.5+incompatible
+	github.com/fluent/fluent-logger-golang v1.4.0
+	github.com/tinylib/msgp v1.1.0  // indirect
 	github.com/gin-gonic/gin v1.6.2
 	github.com/go-logr/logr v0.1.0
 	github.com/gogo/protobuf v1.3.1
@@ -25,8 +27,8 @@ require (
 	github.com/hashicorp/yamux v0.0.0-20190923154419-df201c70410d // indirect
 	github.com/keybase/go-crypto v0.0.0-20190828182435-a05457805304 // indirect
 	github.com/lib/pq v1.2.0
+	github.com/mitchellh/go-homedir v1.1.0
 	github.com/mitchellh/hashstructure v1.0.0
-	github.com/nu7hatch/gouuid v0.0.0-20131221200532-179d4d0c4d8d
 	github.com/onsi/gomega v1.10.1
 	github.com/ory/dockertest v3.3.5+incompatible
 	github.com/pborman/uuid v1.2.0
@@ -56,15 +58,15 @@ require (
 	k8s.io/client-go v11.0.1-0.20190805182717-6502b5e7b1b5+incompatible
 	k8s.io/kubernetes v1.14.7
 	knative.dev/serving v0.17.0
-	sigs.k8s.io/controller-runtime v0.6.1
 	odahu-commons v0.0.0
+	sigs.k8s.io/controller-runtime v0.6.1
 )
 
 replace (
 	k8s.io/apiextensions-apiserver => k8s.io/apiextensions-apiserver v0.17.6
 	k8s.io/apimachinery => k8s.io/apimachinery v0.17.6
 	k8s.io/client-go => k8s.io/client-go v0.17.6
-	sigs.k8s.io/controller-runtime => sigs.k8s.io/controller-runtime v0.5.9
 	// Use local commons package to release them at the same time
 	odahu-commons v0.0.0 => ../commons
+	sigs.k8s.io/controller-runtime => sigs.k8s.io/controller-runtime v0.5.9
 )
diff --git a/packages/operator/go.sum b/packages/operator/go.sum
index 6dd984632..81aaa54fd 100644
--- a/packages/operator/go.sum
+++ b/packages/operator/go.sum
@@ -171,8 +171,6 @@ github.com/GoogleCloudPlatform/testgrid v0.0.13/go.mod h1:UlC/MvnkKjiVGijIKOHxnV
 github.com/Jeffail/gabs v1.1.1/go.mod h1:6xMvQMK4k33lb7GUUpaAPh6nKMmemQeg5d4gn7/bOXc=
 github.com/Jeffail/gabs v1.4.0 h1://5fYRRTq1edjfIrQGvdkcd22pkYUrHZ5YC/H2GJVAo=
 github.com/Jeffail/gabs v1.4.0/go.mod h1:6xMvQMK4k33lb7GUUpaAPh6nKMmemQeg5d4gn7/bOXc=
-github.com/KyleBanks/depth v1.2.1 h1:5h8fQADFrWtarTdtDudMmGsC7GPbOAu6RVB3ffsVFHc=
-github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6XgeJcm8brE=
 github.com/MakeNowJust/heredoc v0.0.0-20170808103936-bb23615498cd/go.mod h1:64YHyfSL2R96J44Nlwm39UHepQbyR5q10x7iYa1ks2E=
 github.com/Masterminds/goutils v1.1.0/go.mod h1:8cTjp+g8YejhMuvIA5y2vz3BpJxksy863GQaJW2MFNU=
 github.com/Masterminds/semver v1.4.2/go.mod h1:MB6lktGJrhw8PrUyiEoblNEGEQ+RzHPF078ddwwvV3Y=
@@ -434,7 +432,6 @@ github.com/coreos/pkg v0.0.0-20180108230652-97fdf19511ea/go.mod h1:E3G3o1h8I7cfc
 github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
 github.com/cpuguy83/go-md2man v1.0.10 h1:BSKMNlYxDvnunlTymqtgONjNnaRV1sTpcovwwjF22jk=
 github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwcJI5acqYI6dE=
-github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
 github.com/cpuguy83/go-md2man/v2 v2.0.0 h1:EoUDS0afbrsXAZ9YQ9jdu/mZ2sXgT1/2yyNng4PGlyM=
 github.com/cpuguy83/go-md2man/v2 v2.0.0/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
 github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
@@ -524,6 +521,8 @@ github.com/fatih/color v1.9.0 h1:8xPHl4/q1VyqGIPif1F+1V3Y3lSmrq01EabUW3CoW5s=
 github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU=
 github.com/fatih/structs v1.1.0 h1:Q7juDM0QtcnhCpeyLGQKyg4TOIghuNXrkL32pHAUMxo=
 github.com/fatih/structs v1.1.0/go.mod h1:9NiDSp5zOcgEDl+j00MP/WkGVPOlPRLejGD8Ga6PJ7M=
+github.com/fluent/fluent-logger-golang v1.4.0 h1:uT1Lzz5yFV16YvDwWbjX6s3AYngnJz8byTCsMTIS0tU=
+github.com/fluent/fluent-logger-golang v1.4.0/go.mod h1:2/HCT/jTy78yGyeNGQLGQsjF3zzzAuy6Xlk6FCMV5eU=
 github.com/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568 h1:BHsljHzVlRcyQhjrss6TZTdY2VfCqZPbv5k3iBFa2ZQ=
 github.com/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568/go.mod h1:xEzjJPgXI435gkrCt3MPfRiAkVrwSbHsst4LCFVfpJc=
 github.com/fortytw2/leaktest v1.2.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
@@ -608,17 +607,12 @@ github.com/go-openapi/jsonpointer v0.18.0/go.mod h1:cOnomiV+CVVwFLk0A/MExoFMjwds
 github.com/go-openapi/jsonpointer v0.19.2/go.mod h1:3akKfEdA7DF1sugOqz1dVQHBcuDBPKZGEoHC/NkiQRg=
 github.com/go-openapi/jsonpointer v0.19.3 h1:gihV7YNZK1iK6Tgwwsxo2rJbD1GTbdm72325Bq8FI3w=
 github.com/go-openapi/jsonpointer v0.19.3/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
-github.com/go-openapi/jsonpointer v0.19.5 h1:gZr+CIYByUqjcgeLXnQu2gHYQC9o73G2XUeOFYEICuY=
-github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
 github.com/go-openapi/jsonreference v0.0.0-20160704190145-13c6e3589ad9/go.mod h1:W3Z9FmVs9qj+KR4zFKmDPGiLdk1D9Rlm7cyMvf57TTg=
 github.com/go-openapi/jsonreference v0.17.0/go.mod h1:g4xxGn04lDIRh0GJb5QlpE3HfopLOL6uZrK/VgnsK9I=
 github.com/go-openapi/jsonreference v0.18.0/go.mod h1:g4xxGn04lDIRh0GJb5QlpE3HfopLOL6uZrK/VgnsK9I=
 github.com/go-openapi/jsonreference v0.19.2/go.mod h1:jMjeRr2HHw6nAVajTXJ4eiUwohSTlpa0o73RUL1owJc=
 github.com/go-openapi/jsonreference v0.19.3 h1:5cxNfTy0UVC3X8JL5ymxzyoUZmo8iZb+jeTWn7tUa8o=
 github.com/go-openapi/jsonreference v0.19.3/go.mod h1:rjx6GuL8TTa9VaixXglHmQmIL98+wF9xc8zWvFonSJ8=
-github.com/go-openapi/jsonreference v0.19.4/go.mod h1:RdybgQwPxbL4UEjuAruzK1x3nE69AqPYEJeo/TWfEeg=
-github.com/go-openapi/jsonreference v0.19.5 h1:1WJP/wi4OjB4iV8KVbH73rQaoialJrqv8gitZLxGLtM=
-github.com/go-openapi/jsonreference v0.19.5/go.mod h1:RdybgQwPxbL4UEjuAruzK1x3nE69AqPYEJeo/TWfEeg=
 github.com/go-openapi/loads v0.17.0/go.mod h1:72tmFy5wsWx89uEVddd0RjRWPZm92WRLhf7AC+0+OOU=
 github.com/go-openapi/loads v0.18.0/go.mod h1:72tmFy5wsWx89uEVddd0RjRWPZm92WRLhf7AC+0+OOU=
 github.com/go-openapi/loads v0.19.0/go.mod h1:72tmFy5wsWx89uEVddd0RjRWPZm92WRLhf7AC+0+OOU=
@@ -635,9 +629,6 @@ github.com/go-openapi/spec v0.19.3/go.mod h1:FpwSN1ksY1eteniUU7X0N/BgJ7a4WvBFVA8
 github.com/go-openapi/spec v0.19.4/go.mod h1:FpwSN1ksY1eteniUU7X0N/BgJ7a4WvBFVA8Lj9mJglo=
 github.com/go-openapi/spec v0.19.6 h1:rMMMj8cV38KVXK7SFc+I2MWClbEfbK705+j+dyqun5g=
 github.com/go-openapi/spec v0.19.6/go.mod h1:Hm2Jr4jv8G1ciIAo+frC/Ft+rR2kQDh8JHKHb3gWUSk=
-github.com/go-openapi/spec v0.19.14/go.mod h1:gwrgJS15eCUgjLpMjBJmbZezCsw88LmgeEip0M63doA=
-github.com/go-openapi/spec v0.20.0 h1:HGLc8AJ7ynOxwv0Lq4TsnwLsWMawHAYiJIFzbcML86I=
-github.com/go-openapi/spec v0.20.0/go.mod h1:+81FIL1JwC5P3/Iuuozq3pPE9dXdIEGxFutcFKaVbmU=
 github.com/go-openapi/strfmt v0.17.0/go.mod h1:P82hnJI0CXkErkXi8IKjPbNBM6lV6+5pLP5l494TcyU=
 github.com/go-openapi/strfmt v0.18.0/go.mod h1:P82hnJI0CXkErkXi8IKjPbNBM6lV6+5pLP5l494TcyU=
 github.com/go-openapi/strfmt v0.19.0/go.mod h1:+uW+93UVvGGq2qGaZxdDeJqSAqBqBdl+ZPMF/cC8nDY=
@@ -649,9 +640,6 @@ github.com/go-openapi/swag v0.19.2/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh
 github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
 github.com/go-openapi/swag v0.19.7 h1:VRuXN2EnMSsZdauzdss6JBC29YotDqG59BZ+tdlIL1s=
 github.com/go-openapi/swag v0.19.7/go.mod h1:ao+8BpOPyKdpQz3AOJfbeEVpLmWAvlT1IfTe5McPyhY=
-github.com/go-openapi/swag v0.19.11/go.mod h1:Uc0gKkdR+ojzsEpjh39QChyu92vPgIr72POcgHMAgSY=
-github.com/go-openapi/swag v0.19.12 h1:Bc0bnY2c3AoF7Gc+IMIAQQsD8fLHjHpc19wXvYuayQI=
-github.com/go-openapi/swag v0.19.12/go.mod h1:eFdyEBkTdoAf/9RXBvj4cr1nH7GD8Kzo5HTt47gr72M=
 github.com/go-openapi/validate v0.18.0/go.mod h1:Uh4HdOzKt19xGIGm1qHf/ofbX1YQ4Y+MYsct2VUrAJ4=
 github.com/go-openapi/validate v0.19.2/go.mod h1:1tRCw7m3jtI8eNWEEliiAqUIcBztB2KDnRCRMUi7GTA=
 github.com/go-openapi/validate v0.19.5/go.mod h1:8DJv2CVJQ6kGNpFW6eV9N3JviE1C85nY1c2z52x1Gk4=
@@ -1129,8 +1117,6 @@ github.com/jmoiron/sqlx v1.2.1-0.20190826204134-d7d95172beb5/go.mod h1:1FEQNm3xl
 github.com/joefitzgerald/rainbow-reporter v0.1.0/go.mod h1:481CNgqmVHQZzdIbN52CupLJyoVwB10FQ/IQlF1pdL8=
 github.com/joho/godotenv v1.3.0/go.mod h1:7hK45KPybAkOC6peb+G5yklZfMxEjkZhHbwpqxOKXbg=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
-github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
-github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
 github.com/joyent/triton-go v0.0.0-20190112182421-51ffac552869/go.mod h1:U+RSyWxWd04xTqnuOQxnai7XGS2PrPY2cfGoDKtMHjA=
 github.com/jpillora/backoff v0.0.0-20180909062703-3050d21c67d7/go.mod h1:2iMrUgbbvHEiQClaW2NsSzMyGHqN+rDFqY705q49KG0=
 github.com/jrick/logrotate v1.0.0/go.mod h1:LNinyqDIJnpAur+b8yyulnQw/wDuN1+BYKlTRt3OuAQ=
@@ -1227,8 +1213,6 @@ github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN
 github.com/mailru/easyjson v0.7.0/go.mod h1:KAzv3t3aY1NaHWoQz1+4F1ccyAH66Jk7yos7ldAVICs=
 github.com/mailru/easyjson v0.7.1-0.20191009090205-6c0755d89d1e h1:jcoUdG1TzY/M/eM5BLFLP8DJeMximx5NQYSlLL9YeWc=
 github.com/mailru/easyjson v0.7.1-0.20191009090205-6c0755d89d1e/go.mod h1:KAzv3t3aY1NaHWoQz1+4F1ccyAH66Jk7yos7ldAVICs=
-github.com/mailru/easyjson v0.7.6 h1:8yTIVnZgCoiM1TgqoeTl+LfU5Jg6/xL3QhGQnimLYnA=
-github.com/mailru/easyjson v0.7.6/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
 github.com/maratori/testpackage v1.0.1/go.mod h1:ddKdw+XG0Phzhx8BFDTKgpWP4i7MpApTE5fXSKAqwDU=
 github.com/markbates/inflect v1.0.4/go.mod h1:1fR9+pO2KHEO9ZRtto13gDwwZaAKstQzferVeWqbgNs=
 github.com/marstr/guid v1.1.0/go.mod h1:74gB1z2wpxxInTG6yaqA7KrtM0NZ+RbrcqDvYHefzho=
@@ -1357,13 +1341,10 @@ github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646/go.mod h1:jpp1/29i3P1S
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/nsf/termbox-go v0.0.0-20200418040025-38ba6e5628f1/go.mod h1:IuKpRQcYE1Tfu+oAQqaLisqDeXgjyyltCfsaoYN18NQ=
-github.com/nu7hatch/gouuid v0.0.0-20131221200532-179d4d0c4d8d h1:VhgPp6v9qf9Agr/56bj7Y/xa04UccTW04VP0Qed4vnQ=
-github.com/nu7hatch/gouuid v0.0.0-20131221200532-179d4d0c4d8d/go.mod h1:YUTz3bUH2ZwIWBy3CJBeOBEugqcmXREj14T+iG/4k4U=
 github.com/nwaples/rardecode v1.0.0/go.mod h1:5DzqNKiOdpKKBH87u8VlvAnPZMXcGRhxWkRpHbbfGS0=
 github.com/nxadm/tail v1.4.4 h1:DQuhQpB1tVlglWS2hLQ5OV6B5r8aGxSrPc5Qo6uTN78=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/octago/sflags v0.2.0/go.mod h1:G0bjdxh4qPRycF74a2B8pU36iTp9QHGx0w0dFZXPt80=
-github.com/odahu/odahu-flow v0.0.0-20210128095453-c27c63bb6718 h1:OOjtT92qvQdqKKYT2eju1+eFRBIAAeH+uqUYHZ737BE=
 github.com/oklog/run v1.0.0 h1:Ru7dDtJNOyC66gQ5dQmaCa0qIsAUFY3sFpK1Xk8igrw=
 github.com/oklog/run v1.0.0/go.mod h1:dlhp/R75TPv97u0XWUtDeV/lRKWPKSdTuV0TZvrmrQA=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
@@ -1441,6 +1422,7 @@ github.com/pengsrc/go-shared v0.2.1-0.20190131101655-1999055a4a14/go.mod h1:jVbl
 github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR/tNboyR3/BZd58JJSHlUSCU=
 github.com/phayes/checkstyle v0.0.0-20170904204023-bfd46e6a821d/go.mod h1:3OzsM7FXDQlpCiw2j81fOmAwQLnZnLGXVKUzeKQXIAw=
 github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2/go.mod h1:iIss55rKnNBTvrwdmkUpLnDpZoAHvWaiq5+iMmen4AE=
+github.com/philhofer/fwd v1.0.0 h1:UbZqGr5Y38ApvM/V/jEljVxwocdweyH+vmYvRPBnbqQ=
 github.com/philhofer/fwd v1.0.0/go.mod h1:gk3iGcWd9+svBvR0sR+KPcfE+RNWozjowpeBVG3ZVNU=
 github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
 github.com/pierrec/lz4 v1.0.2-0.20190131084431-473cd7ce01a1/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
@@ -1553,8 +1535,6 @@ github.com/russross/blackfriday v1.5.2 h1:HyvC0ARfnZBqnXwABFeSZHpKvJHJJfPz81GNue
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
 github.com/russross/blackfriday/v2 v2.0.1 h1:lPqVAte+HuHNfhJ/0LC98ESWRz8afy9tM/0RK8m9o+Q=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
-github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
-github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/ryancurrah/gomodguard v1.0.4/go.mod h1:9T/Cfuxs5StfsocWr4WzDL36HqnX0fVb9d5fSEaLhoE=
 github.com/ryancurrah/gomodguard v1.1.0/go.mod h1:4O8tr7hBODaGE6VIhfJDHcwzh5GUccKSJBU0UMXJFVM=
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
@@ -1675,8 +1655,6 @@ github.com/swaggo/gin-swagger v1.1.0/go.mod h1:FQlm07YuT1glfN3hQiO11UQ2m39vOCZ/a
 github.com/swaggo/swag v1.4.0/go.mod h1:hog2WgeMOrQ/LvQ+o1YGTeT+vWVrbi0SiIslBtxKTyM=
 github.com/swaggo/swag v1.5.0 h1:haK8VG3hj+v/c8hQ4f3U+oYpkdI/26m9LAUTXHOv+2U=
 github.com/swaggo/swag v1.5.0/go.mod h1:+xZrnu5Ut3GcUkKAJm9spnOooIS1WB1cUOkLNPrvrE0=
-github.com/swaggo/swag v1.7.0 h1:5bCA/MTLQoIqDXXyHfOpMeDvL9j68OY/udlK4pQoo4E=
-github.com/swaggo/swag v1.7.0/go.mod h1:BdPIL73gvS9NBsdi7M1JOxLvlbfvNRaBP8m6WT6Aajo=
 github.com/syndtr/gocapability v0.0.0-20170704070218-db04d3cc01c8/go.mod h1:hkRG7XYTFWNJGYcbNJQlaLq0fg1yr4J4t/NcTQtrfww=
 github.com/t3rm1n4l/go-mega v0.0.0-20200416171014-ffad7fcb44b8 h1:IGJQmLBLYBdAknj21W3JsVof0yjEXfy1Q0K3YZebDOg=
 github.com/t3rm1n4l/go-mega v0.0.0-20200416171014-ffad7fcb44b8/go.mod h1:XWL4vDyd3JKmJx+hZWUVgCNmmhZ2dTBcaNDcxH465s0=
@@ -1697,7 +1675,10 @@ github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/timakin/bodyclose v0.0.0-20190930140734-f7f2e9bca95e/go.mod h1:Qimiffbc6q9tBWlVV6x0P9sat/ao1xEkREYPPj9hphk=
 github.com/timakin/bodyclose v0.0.0-20200424151742-cb6215831a94/go.mod h1:Qimiffbc6q9tBWlVV6x0P9sat/ao1xEkREYPPj9hphk=
+github.com/tinylib/msgp v1.0.2 h1:DfdQrzQa7Yh2es9SuLkixqxuXS2SxsdYn0KbdrOGWD8=
 github.com/tinylib/msgp v1.0.2/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDWklFE=
+github.com/tinylib/msgp v1.1.0 h1:9fQd+ICuRIu/ue4vxJZu6/LzxN0HwMds2nq/0cFvxHU=
+github.com/tinylib/msgp v1.1.0/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDWklFE=
 github.com/tj/assert v0.0.0-20171129193455-018094318fb0/go.mod h1:mZ9/Rh9oLWpLLDRpvE+3b7gP/C2YyLFYxNmcLnPTMe0=
 github.com/tj/go-elastic v0.0.0-20171221160941-36157cbbebc2/go.mod h1:WjeM0Oo1eNAjXGDx2yma7uG2XoyRZTq1uv3M/o7imD0=
 github.com/tj/go-kinesis v0.0.0-20171128231115-08b17f58cb1b/go.mod h1:/yhzCV0xPfx6jb1bBgRFjl5lytqVqZXEaeqWP8lTEao=
@@ -1729,10 +1710,6 @@ github.com/urfave/cli v0.0.0-20171014202726-7bc6a0acffa5/go.mod h1:70zkFmudgCuE/
 github.com/urfave/cli v1.18.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA=
 github.com/urfave/cli v1.20.0 h1:fDqGv3UG/4jbVl/QkFwEdddtEDjh/5Ov6X+0B/3bPaw=
 github.com/urfave/cli v1.20.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA=
-github.com/urfave/cli v1.22.5 h1:lNq9sAHXK2qfdI8W+GRItjCEkI+2oR4d+MEHy1CKXoU=
-github.com/urfave/cli v1.22.5/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
-github.com/urfave/cli/v2 v2.3.0 h1:qph92Y649prgesehzOrQjdWyxFOp/QVM+6imKHad91M=
-github.com/urfave/cli/v2 v2.3.0/go.mod h1:LJmUH05zAU44vOAcrfzZQKsZbVcdbOG8rtL3/XcUArI=
 github.com/uudashr/gocognit v1.0.1/go.mod h1:j44Ayx2KW4+oB6SWMv8KsmHzZrOInQav7D3cQMJ5JUM=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
@@ -1773,7 +1750,6 @@ github.com/youmark/pkcs8 v0.0.0-20200520070018-fad002e585ce/go.mod h1:ul22v+Nro/
 github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
-github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yunify/qingstor-sdk-go/v3 v3.2.0 h1:9sB2WZMgjwSUNZhrgvaNGazVltoFUUfuS9f0uCWtTr8=
 github.com/yunify/qingstor-sdk-go/v3 v3.2.0/go.mod h1:KciFNuMu6F4WLk9nGwwK69sCGKLCdd9f97ac/wfumS4=
 github.com/yvasiyarov/go-metrics v0.0.0-20140926110328-57bccd1ccd43/go.mod h1:aX5oPXxHm3bOH+xeAttToC8pqch2ScQN/JoXYupl6xs=
@@ -1973,10 +1949,6 @@ golang.org/x/net v0.0.0-20200520182314-0ba52f642ac2/go.mod h1:qpuaurCH72eLCgpAm/
 golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20200707034311-ab3426394381 h1:VXak5I6aEWmAXeQjA+QSZzlgNrpq9mjcfDemuexIKsU=
 golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
-golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.0.0-20201216054612-986b41b23924 h1:QsnDpLLOKwHBBDa8nDws4DYNc/ryVW2vCpxCs09d4PY=
-golang.org/x/net v0.0.0-20201216054612-986b41b23924/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/oauth2 v0.0.0-20180724155351-3d292e4d0cdc/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20181017192945-9dcd33a902f4/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -1999,8 +1971,6 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208 h1:qwRHBd0NqMbJxfbotnDhm2ByMI1Shq4Y6oRJo21SGJA=
 golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9 h1:SQFwaSi55rU7vdNs9Yr0Z324VNlrF+0wMqRXT4St8ck=
-golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20170830134202-bb24a47a89ea/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -2082,10 +2052,6 @@ golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200720211630-cb9d2d5c5666/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200802091954-4b90ce9b60b3 h1:qDJKu1y/1SjhWac4BQZjLljqvqiWUhjmDMnonmVGDAU=
 golang.org/x/sys v0.0.0-20200802091954-4b90ce9b60b3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20201119102817-f84b799fce68 h1:nxC68pudNYkKU6jWhgrqdreuFiOQWj1Fs7T3VrH4Pjw=
-golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.0.0-20160726164857-2910a502d2bf/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -2094,8 +2060,6 @@ golang.org/x/text v0.3.1-0.20181227161524-e6919f6577db/go.mod h1:bEr9sfX3Q8Zfm5f
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.4 h1:0YWbFKbhXG/wIiuHDSKpS0Iy7FSA+u45VtBMfQcFTTc=
-golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -2204,15 +2168,10 @@ golang.org/x/tools v0.0.0-20200731060945-b5fad4ed8dd6 h1:qKpj8TpV+LEhel7H/fR788J
 golang.org/x/tools v0.0.0-20200731060945-b5fad4ed8dd6/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
 golang.org/x/tools v0.0.0-20200820180210-c8f393745106 h1:42Zs/g7pjhSIE/wiAuKcp8zp20zv7W2diNU6arpshOA=
 golang.org/x/tools v0.0.0-20200820180210-c8f393745106/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
-golang.org/x/tools v0.0.0-20201120155355-20be4ac4bd6e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
-golang.org/x/tools v0.0.0-20201218024724-ae774e9781d2 h1:lHDhNNs7asPT3p01mm8EP3B+bNyyVfg0bcYjhJUYgxw=
-golang.org/x/tools v0.0.0-20201218024724-ae774e9781d2/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
-golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gomodules.xyz/jsonpatch/v2 v2.0.1/go.mod h1:IhYNNY4jnS53ZnfE4PAmpKtDpTCj1JFXc+3mwe7XcUU=
 gomodules.xyz/jsonpatch/v2 v2.1.0 h1:Phva6wqu+xR//Njw6iorylFFgn/z547tw5Ne3HZPQ+k=
 gomodules.xyz/jsonpatch/v2 v2.1.0/go.mod h1:IhYNNY4jnS53ZnfE4PAmpKtDpTCj1JFXc+3mwe7XcUU=
@@ -2425,8 +2384,6 @@ gopkg.in/yaml.v2 v2.2.7/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
-gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20190709130402-674ba3eaed22/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/packages/operator/hack/kubeflow_prediction_v2/spec.yaml b/packages/operator/hack/kubeflow_prediction_v2/spec.yaml
new file mode 100644
index 000000000..f3ab5a911
--- /dev/null
+++ b/packages/operator/hack/kubeflow_prediction_v2/spec.yaml
@@ -0,0 +1,322 @@
+openapi: 3.0.0
+info:
+  title: Data Plane
+  version: '2.0'
+  description: 'https://github.com/kubeflow/kfserving/blob/master/docs/predict-api/v2/required_api.md'
+servers:
+  - url: 'http://localhost:3000'
+paths:
+  /v2/health/live:
+    get:
+      summary: Server Live
+      responses:
+        '200':
+          description: OK
+      operationId: get-v2-health-live
+      description: The “server live” API indicates if the inference server is able to receive and respond to metadata and inference requests. The “server live” API can be used directly to implement the Kubernetes livenessProbe.
+  /v2/health/ready:
+    get:
+      summary: Server Ready
+      tags: []
+      responses:
+        '200':
+          description: OK
+      operationId: get-v2-health-ready
+      description: The “server ready” health API indicates if all the models are ready for inferencing. The “server ready” health API can be used directly to implement the Kubernetes readinessProbe.
+  /v2/:
+    get:
+      summary: Server Metadata
+      tags: []
+      responses:
+        '200':
+          description: OK
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/metadata_server_response'
+        '400':
+          description: Bad Request
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/metadata_server_error_response'
+      operationId: get-v2
+      description: 'The server metadata endpoint provides information about the server. A server metadata request is made with an HTTP GET to a server metadata endpoint. In the corresponding response the HTTP body contains the [Server Metadata Response JSON Object](#server-metadata-response-json-object) or the [Server Metadata Response JSON Error Object](#server-metadata-response-json-error-object).'
+  '/v2/models/${MODEL_NAME}/versions/${MODEL_VERSION}':
+    parameters:
+      - schema:
+          type: string
+        name: MODEL_NAME
+        in: path
+        required: true
+      - schema:
+          type: string
+        name: MODEL_VERSION
+        in: path
+        required: true
+    get:
+      summary: Model Metadata
+      tags: []
+      responses:
+        '200':
+          description: OK
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/metadata_model_response'
+      operationId: get-v2-models-$-modelName-versions-$-modelVersion
+      description: 'The per-model metadata endpoint provides information about a model. A model metadata request is made with an HTTP GET to a model metadata endpoint. In the corresponding response the HTTP body contains the [Model Metadata Response JSON Object](#model-metadata-response-json-object) or the [Model Metadata Response JSON Error Object](#model-metadata-response-json-error-object). The model name and (optionally) version must be available in the URL. If a version is not provided the server may choose a version based on its own policies or return an error.'
+  '/v2/models/${MODEL_NAME}/versions/${MODEL_VERSION}/infer':
+    parameters:
+      - schema:
+          type: string
+        name: MODEL_NAME
+        in: path
+        required: true
+      - schema:
+          type: string
+        name: MODEL_VERSION
+        in: path
+        required: true
+    post:
+      summary: Inference
+      operationId: post-v2-models-$-MODEL_NAME-versions-$-MODEL_VERSION-infer
+      responses:
+        '200':
+          description: OK
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/inference_response'
+        '400':
+          description: Bad Request
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/inference_error_response'
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/inference_request'
+      description: 'An inference request is made with an HTTP POST to an inference endpoint. In the request the HTTP body contains the [Inference Request JSON Object](#inference-request-json-object). In the corresponding response the HTTP body contains the [Inference Response JSON Object](#inference-response-json-object) or [Inference Response JSON Error Object](#inference-response-json-error-object). See [Inference Request Examples](#inference-request-examples) for some example HTTP/REST requests and responses.'
+components:
+  schemas:
+    metadata_server_response:
+      title: metadata_server_response
+      type: object
+      description: ''
+      x-examples: {}
+      properties:
+        name:
+          type: string
+        version:
+          type: string
+        extensions:
+          type: array
+          items:
+            type: string
+      required:
+        - name
+        - version
+        - extensions
+    metadata_server_error_response:
+      title: metadata_server_error_response
+      type: object
+      properties:
+        error:
+          type: string
+      required:
+        - error
+    metadata_model_response:
+      title: metadata_model_response
+      type: object
+      properties:
+        name:
+          type: string
+        versions:
+          type: array
+          items:
+            type: string
+        platform:
+          type: string
+        inputs:
+          type: array
+          items:
+            $ref: '#/components/schemas/metadata_tensor'
+        outputs:
+          type: array
+          items:
+            $ref: '#/components/schemas/metadata_tensor'
+      required:
+        - name
+        - platform
+    metadata_tensor:
+      title: metadata_tensor
+      type: object
+      properties:
+        name:
+          type: string
+        datatype:
+          type: string
+        shape:
+          type: array
+          items:
+            type: integer
+      required:
+        - name
+        - datatype
+        - shape
+    metadata_model_error_response:
+      title: metadata_model_error_response
+      type: object
+      properties:
+        error:
+          type: string
+      required:
+        - error
+    inference_request:
+      title: inference_request
+      type: object
+      x-examples:
+        Example 1:
+          id: '42'
+          inputs:
+            - name: input0
+              shape:
+                - 2
+                - 2
+              datatype: UINT32
+              data:
+                - 1
+                - 2
+                - 3
+                - 4
+            - name: input1
+              shape:
+                - 3
+              datatype: BOOL
+              data:
+                - true
+          outputs:
+            - name: output0
+        Example 2:
+          id: '42'
+          outputs:
+            - name: output0
+              shape:
+                - 3
+                - 2
+              datatype: FP32
+              data:
+                - 1
+                - 1.1
+                - 2
+                - 2.1
+                - 3
+                - 3.1
+      properties:
+        id:
+          type: string
+        parameters:
+          $ref: '#/components/schemas/parameters'
+        inputs:
+          type: array
+          items:
+            $ref: '#/components/schemas/request_input'
+        outputs:
+          type: array
+          items:
+            $ref: '#/components/schemas/request_output'
+      required:
+        - inputs
+    parameters:
+      title: parameters
+      x-examples: {}
+      type: object
+    request_input:
+      title: request_input
+      type: object
+      properties:
+        name:
+          type: string
+        shape:
+          type: array
+          items:
+            type: integer
+        datatype:
+          type: string
+        parameters:
+          $ref: '#/components/schemas/parameters'
+        data:
+          $ref: '#/components/schemas/tensor_data'
+      required:
+        - name
+        - shape
+        - datatype
+        - data
+    tensor_data:
+      title: tensor_data
+      type: array
+      items:
+        anyOf:
+          - $ref: '#/components/schemas/tensor_data'
+          - type: number
+          - type: string
+          - type: boolean
+    request_output:
+      title: request_output
+      type: object
+      properties:
+        name:
+          type: string
+        parameters:
+          $ref: '#/components/schemas/parameters'
+      required:
+        - name
+    response_output:
+      title: response_output
+      type: object
+      properties:
+        name:
+          type: string
+        shape:
+          type: array
+          items:
+            type: integer
+        datatype:
+          type: string
+        parameters:
+          $ref: '#/components/schemas/parameters'
+        data:
+          $ref: '#/components/schemas/tensor_data'
+      required:
+        - name
+        - shape
+        - datatype
+        - data
+    inference_response:
+      title: inference_response
+      type: object
+      properties:
+        model_name:
+          type: string
+        model_version:
+          type: string
+        id:
+          type: string
+        parameters:
+          $ref: '#/components/schemas/parameters'
+        outputs:
+          type: array
+          items:
+            $ref: '#/components/schemas/response_output'
+      required:
+        - model_name
+        - outputs
+    inference_error_response:
+      title: inference_error_response
+      type: object
+      properties:
+        error:
+          type: string
diff --git a/packages/operator/pkg/apis/predict_v2/predict_v2.gen.go b/packages/operator/pkg/apis/predict_v2/predict_v2.gen.go
new file mode 100644
index 000000000..4c22744bd
--- /dev/null
+++ b/packages/operator/pkg/apis/predict_v2/predict_v2.gen.go
@@ -0,0 +1,91 @@
+// Package predict_v2 provides primitives to interact the openapi HTTP API.
+//
+// Code generated by github.com/deepmap/oapi-codegen DO NOT EDIT.
+package predict_v2
+
+// InferenceErrorResponse defines model for inference_error_response.
+type InferenceErrorResponse struct {
+	Error *string `json:"error,omitempty"`
+}
+
+// InferenceRequest defines model for inference_request.
+type InferenceRequest struct {
+	Id         *string          `json:"id,omitempty"`
+	Inputs     []RequestInput   `json:"inputs"`
+	Outputs    *[]RequestOutput `json:"outputs,omitempty"`
+	Parameters *Parameters      `json:"parameters,omitempty"`
+}
+
+// InferenceResponse defines model for inference_response.
+type InferenceResponse struct {
+	Id           *string          `json:"id,omitempty"`
+	ModelName    string           `json:"model_name"`
+	ModelVersion *string          `json:"model_version,omitempty"`
+	Outputs      []ResponseOutput `json:"outputs"`
+	Parameters   *Parameters      `json:"parameters,omitempty"`
+}
+
+// MetadataModelResponse defines model for metadata_model_response.
+type MetadataModelResponse struct {
+	Inputs   *[]MetadataTensor `json:"inputs,omitempty"`
+	Name     string            `json:"name"`
+	Outputs  *[]MetadataTensor `json:"outputs,omitempty"`
+	Platform string            `json:"platform"`
+	Versions *[]string         `json:"versions,omitempty"`
+}
+
+// MetadataServerErrorResponse defines model for metadata_server_error_response.
+type MetadataServerErrorResponse struct {
+	Error string `json:"error"`
+}
+
+// MetadataServerResponse defines model for metadata_server_response.
+type MetadataServerResponse struct {
+	Extensions []string `json:"extensions"`
+	Name       string   `json:"name"`
+	Version    string   `json:"version"`
+}
+
+// MetadataTensor defines model for metadata_tensor.
+type MetadataTensor struct {
+	Datatype string `json:"datatype"`
+	Name     string `json:"name"`
+	Shape    []int  `json:"shape"`
+}
+
+// Parameters defines model for parameters.
+type Parameters map[string]interface{}
+
+// RequestInput defines model for request_input.
+type RequestInput struct {
+	Data       TensorData  `json:"data"`
+	Datatype   string      `json:"datatype"`
+	Name       string      `json:"name"`
+	Parameters *Parameters `json:"parameters,omitempty"`
+	Shape      []int       `json:"shape"`
+}
+
+// RequestOutput defines model for request_output.
+type RequestOutput struct {
+	Name       string      `json:"name"`
+	Parameters *Parameters `json:"parameters,omitempty"`
+}
+
+// ResponseOutput defines model for response_output.
+type ResponseOutput struct {
+	Data       TensorData  `json:"data"`
+	Datatype   string      `json:"datatype"`
+	Name       string      `json:"name"`
+	Parameters *Parameters `json:"parameters,omitempty"`
+	Shape      []int       `json:"shape"`
+}
+
+// TensorData defines model for tensor_data.
+type TensorData []interface{}
+
+// PostV2ModelsMODELNAMEVersionsMODELVERSIONInferJSONBody defines parameters for PostV2ModelsMODELNAMEVersionsMODELVERSIONInfer.
+type PostV2ModelsMODELNAMEVersionsMODELVERSIONInferJSONBody InferenceRequest
+
+// PostV2ModelsMODELNAMEVersionsMODELVERSIONInferJSONRequestBody defines body for PostV2ModelsMODELNAMEVersionsMODELVERSIONInfer for application/json ContentType.
+type PostV2ModelsMODELNAMEVersionsMODELVERSIONInferJSONRequestBody PostV2ModelsMODELNAMEVersionsMODELVERSIONInferJSONBody
+
diff --git a/packages/operator/pkg/config/tools.go b/packages/operator/pkg/config/tools.go
new file mode 100644
index 000000000..bc0c546fa
--- /dev/null
+++ b/packages/operator/pkg/config/tools.go
@@ -0,0 +1,14 @@
+package config
+
+type FluentdConfig struct {
+	BaseURL string `json:"baseURL"`
+}
+
+type FeedbackConfig struct {
+	Fluentd FluentdConfig `json:"fluentd"`
+}
+
+type ToolsConfig struct {
+	Auth AuthConfig `json:"auth"`
+	Feedback FeedbackConfig `json:"feedback"`
+}
\ No newline at end of file
diff --git a/packages/operator/pkg/rclone/gcs.go b/packages/operator/pkg/rclone/gcs.go
index c5d6cb7e9..6bff99357 100644
--- a/packages/operator/pkg/rclone/gcs.go
+++ b/packages/operator/pkg/rclone/gcs.go
@@ -26,8 +26,6 @@ import (
 	"net/url"
 )
 
-const serviceAccountJSONPath = "gcs_service_account.json"
-
 func createGcsConfig(configName string, conn *v1alpha1.ConnectionSpec) (*FileDescription, error) {
 	_, err := fs.Find("googlecloudstorage")
 	if err != nil {
@@ -41,6 +39,8 @@ func createGcsConfig(configName string, conn *v1alpha1.ConnectionSpec) (*FileDes
 		"bucket_acl": "private",
 	}
 
+	serviceAccountJSONPath := "gcs-key-" + configName + ".json"
+
 	if len(conn.KeySecret) != 0 {
 		if err = ioutil.WriteFile(serviceAccountJSONPath, []byte(conn.KeySecret), 0600); err != nil {
 			log.Error(err, "Failed to write service account JSON-file")
diff --git a/packages/operator/pkg/rclone/type.go b/packages/operator/pkg/rclone/type.go
index d151dbbb4..f4026b200 100644
--- a/packages/operator/pkg/rclone/type.go
+++ b/packages/operator/pkg/rclone/type.go
@@ -48,16 +48,21 @@ type FileDescription struct {
 }
 
 func NewObjectStorage(conn *odahuflowv1alpha1.ConnectionSpec) (obj *ObjectStorage, err error) {
-	var config *FileDescription
 	name := uuid.NewV4()
 
+	return NewObjectStorageWithName(name.String(), conn)
+}
+
+func NewObjectStorageWithName(name string, conn *odahuflowv1alpha1.ConnectionSpec) (obj *ObjectStorage, err error) {
+	var config *FileDescription
+
 	switch conn.Type {
 	case connection.S3Type:
-		config, err = createS3config(name.String(), conn)
+		config, err = createS3config(name, conn)
 	case connection.GcsType:
-		config, err = createGcsConfig(name.String(), conn)
+		config, err = createGcsConfig(name, conn)
 	case connection.AzureBlobType:
-		config, err = createAzureBlobConfig(name.String(), conn)
+		config, err = createAzureBlobConfig(name, conn)
 	default:
 		return nil, errors.New(fmt.Sprintf("Unexpected connection type: %s", conn.Type))
 	}
diff --git a/packages/operator/pkg/utils/connections/connections.go b/packages/operator/pkg/utils/connections/connections.go
new file mode 100644
index 000000000..6fe4a6332
--- /dev/null
+++ b/packages/operator/pkg/utils/connections/connections.go
@@ -0,0 +1,60 @@
+package connections
+
+import (
+	"github.com/odahu/odahu-flow/packages/operator/pkg/apis/connection"
+	"github.com/odahu/odahu-flow/packages/operator/pkg/rclone"
+	"go.uber.org/multierr"
+	"go.uber.org/zap"
+	"strings"
+)
+
+const (
+	sep = ":"
+	defaultStorageNamePrefix = "odahu-"
+)
+
+type ConnectionClient interface {
+	GetConnection(id string) (*connection.Connection, error)
+}
+
+func getConnNameAndRCloneStorageName (s string) (connName string, storageName string) {
+	res := strings.Split(s, sep)
+	connName = res[0]
+	if len(res) > 1 {
+		storageName = res[1]
+	} else {
+		storageName = defaultStorageNamePrefix + connName
+	}
+
+	return
+}
+
+func GenerateRClone(clusterValues []string, fileValues []string, connClient ConnectionClient ) (err error) {
+
+	for _, value := range clusterValues {
+		connName, storageName := getConnNameAndRCloneStorageName(value)
+		conn, fetchErr := connClient.GetConnection(connName)
+
+		if fetchErr != nil {
+			err = multierr.Append(err, fetchErr)
+			continue
+		}
+
+		if decodeErr := conn.DecodeBase64Fields(); decodeErr != nil {
+			zap.S().Warnw(
+				"There are some problems with decoding base64 fields. Maybe they are already decoded",
+				zap.Error(decodeErr),
+			)
+		}
+
+		_, genErr := rclone.NewObjectStorageWithName(storageName, &conn.Spec)
+		if genErr != nil {
+			err = multierr.Append(err, genErr)
+			continue
+		}
+	}
+
+
+	return err
+
+}
diff --git a/packages/operator/pkg/utils/feedback/feedback.go b/packages/operator/pkg/utils/feedback/feedback.go
new file mode 100644
index 000000000..02b352dbd
--- /dev/null
+++ b/packages/operator/pkg/utils/feedback/feedback.go
@@ -0,0 +1,62 @@
+package feedback
+
+import (
+	"go.uber.org/zap"
+	"io/ioutil"
+	"path/filepath"
+	"strings"
+)
+
+type LoggerEngine interface {
+	Post(tag string, message interface{}) error
+	Close() error
+}
+
+type Logger struct {
+	logEngine LoggerEngine
+}
+
+type Wrapper func(content string) (wrappedContent interface{})
+
+func NewLogger(engine LoggerEngine) Logger {
+	return Logger{
+		logEngine: engine,
+	}
+}
+
+
+// LogDir logs content of each .json file in `path` but wrap it content using
+// Wrapper before.
+// `tag` is used to add extra information to LoggerEngine
+func (l Logger) LogDir(path string, tag string, wrap Wrapper) error {
+	items, err := ioutil.ReadDir(path)
+	if err != nil {
+		return err
+	}
+
+	for _, item := range items {
+		file := item.Name()
+		if item.IsDir() {
+			zap.S().Infof("%s is directory (not file). Skip logging", file)
+			continue
+		}
+		if strings.HasSuffix(file, ".json") {
+			fp := filepath.Join(path, file)
+			data, err := ioutil.ReadFile(fp)
+			if err != nil {
+				return err
+			}
+
+			logContent := wrap(string(data))
+
+			if err := l.logEngine.Post(tag, logContent); err != nil {
+				zap.S().Errorw("Error during logging", zap.Error(err))
+			} else {
+				zap.S().Infof("%s successfully logged", file)
+			}
+		} else {
+			zap.S().Infof("%s has not .json extension. Skip logging", file)
+		}
+	}
+	return nil
+}
diff --git a/packages/operator/pkg/utils/predictv2/directory_validator.go b/packages/operator/pkg/utils/predictv2/directory_validator.go
new file mode 100644
index 000000000..8a1dc5fdc
--- /dev/null
+++ b/packages/operator/pkg/utils/predictv2/directory_validator.go
@@ -0,0 +1,85 @@
+package predictv2
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/odahu/odahu-flow/packages/operator/pkg/apis/predict_v2"
+	"go.uber.org/zap"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+
+
+type Validator func (content []byte) error
+
+func ValidateInput(content []byte) error {
+	var request predict_v2.InferenceRequest
+	return json.Unmarshal(content, &request)
+}
+func ValidateOutput(content []byte) error {
+	var response predict_v2.InferenceResponse
+	return json.Unmarshal(content, &response)
+}
+
+// ValidateDir go through `source` directory files (not recursively) and tries to validate each file
+// with .json extension using Validator
+// If destination is not nil then files that passed validation will be copied to destination
+func ValidateDir(source string, validate Validator, destination *string) ([]string, error) {
+
+	// Files that passed validation
+	var passed []string
+
+	if destination != nil {
+		if _, err := os.Stat(*destination); os.IsNotExist(err) {
+			zap.S().Infof("destination directory specified by -d %s does not exist", *destination)
+			if err := os.MkdirAll(*destination, os.ModePerm); err != nil {
+				return passed, fmt.Errorf("unable to create %s. error %+v", *destination, err)
+			}
+			zap.S().Infof("destination directory %s is created", *destination)
+		}
+	}
+
+	items, err := ioutil.ReadDir(source)
+	if err != nil {
+		return passed, err
+	}
+
+	for _, item := range items {
+		file := item.Name()
+		if item.IsDir() {
+			zap.S().Infof("%s is directory (not file). Skip validating", file)
+			continue
+		}
+		if strings.HasSuffix(file, ".json") {
+			fp := filepath.Join(source, file)
+			data, err := ioutil.ReadFile(fp)
+			if err != nil {
+				return passed, err
+			}
+			parseErr := validate(data)
+			if parseErr != nil {
+				zap.S().Warnw("Unable to Unmarshal file, validation is failed",
+					"filepath", fp, "unmarshal error", parseErr)
+				continue
+			}
+
+			if destination != nil {
+				if err := ioutil.WriteFile(filepath.Join(*destination, file), data, item.Mode()); err != nil {
+					return passed, err
+				}
+
+			}
+			zap.S().Infof("%s successfully passed validation", file)
+			passed = append(passed, file)
+
+
+		} else {
+			zap.S().Infof("%s has not .json extension. Skip validating", file)
+		}
+	}
+
+	return passed, nil
+}
\ No newline at end of file
diff --git a/scripts/Jenkinsfile b/scripts/Jenkinsfile
index f0b8644b4..4fdb95208 100644
--- a/scripts/Jenkinsfile
+++ b/scripts/Jenkinsfile
@@ -138,6 +138,7 @@ pipeline {
                                 cicdLibrary.buildDockerImage('odahu-flow-model-trainer', ".", "containers/operator/Dockerfile", "--target model-trainer --cache-from odahu/odahu-flow-operator-dependencies:${Globals.buildVersion}")
                                 cicdLibrary.buildDockerImage('odahu-flow-model-packager', ".", "containers/operator/Dockerfile", "--target model-packager --cache-from odahu/odahu-flow-operator-dependencies:${Globals.buildVersion}")
                                 cicdLibrary.buildDockerImage('odahu-flow-controller', ".", "containers/operator/Dockerfile", "--target controller --cache-from odahu/odahu-flow-operator-dependencies:${Globals.buildVersion}")
+                                cicdLibrary.buildDockerImage('odahu-flow-tools', ".", "containers/operator/Dockerfile", "--target tools --cache-from odahu/odahu-flow-operator-dependencies:${Globals.buildVersion}")
                             }
                         }
                     }
@@ -267,6 +268,15 @@ pipeline {
                         }
                     }
                 }
+                stage('Upload ODAHU Tools image') {
+                    steps {
+                        timeout(time: param_upload_timeout, unit: 'MINUTES') {
+                            script {
+                                cicdLibrary.uploadDockerImage('odahu-flow-tools', env.param_stable_release.toBoolean() && env.param_dockerhub_publishing_enabled.toBoolean())
+                            }
+                        }
+                    }
+                }
                 stage('Upload Operator dependencies') {
                     steps {
                         timeout(time: param_upload_timeout, unit: 'MINUTES') {


Test patch:
diff --git a/packages/feedback/pkg/collector/http_test.go b/packages/feedback/pkg/collector/http_test.go
index 64e832ff3..aa68bcd69 100644
--- a/packages/feedback/pkg/collector/http_test.go
+++ b/packages/feedback/pkg/collector/http_test.go
@@ -19,6 +19,7 @@ package collector
 import (
 	"encoding/json"
 	"fmt"
+	feedback_commons "odahu-commons/feedback"
 	"github.com/odahu/odahu-flow/packages/feedback/pkg/feedback"
 	"net/http"
 	"net/http/httptest"
@@ -71,8 +72,8 @@ func TestSendFeedbackWithoutHeader(t *testing.T) {
 	assert.Equal(t, "{\"error\":\"x-request-id header is missed\"}", w.Body.String())
 }
 
-func buildMessage(modelName, modelVersion, requestID string, payload map[string]interface{}) feedback.ModelFeedback {
-	return feedback.ModelFeedback{
+func buildMessage(modelName, modelVersion, requestID string, payload map[string]interface{}) feedback_commons.ModelFeedback {
+	return feedback_commons.ModelFeedback{
 		ModelName:    modelName,
 		ModelVersion: modelVersion,
 		Payload:      payload,