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/helms/odahu-flow-core/templates/operator/decrypt_token.yaml b/helms/odahu-flow-core/templates/operator/decrypt_token.yaml
index 40c067424..bec10308d 100644
--- a/helms/odahu-flow-core/templates/operator/decrypt_token.yaml
+++ b/helms/odahu-flow-core/templates/operator/decrypt_token.yaml
@@ -20,3 +20,14 @@ metadata:
 data:
   "config.yaml": |
     {{ toYaml .Values.config | b64enc }}
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: "{{ .Release.Name }}-tools-config"
+  namespace: "{{ .Values.config.batch.namespace }}"
+  labels:
+    {{- include "odahuflow.helm-labels" (dict "component" "api" "root" .) | nindent 4 }}
+data:
+  ".odahu-tools.yaml": |
+    {{ toYaml .Values.toolsConfig | b64enc }}
diff --git a/helms/odahu-flow-core/templates/operator/generated/odahuflow.odahu.org_batchinferencejobs.yaml b/helms/odahu-flow-core/templates/operator/generated/odahuflow.odahu.org_batchinferencejobs.yaml
new file mode 100644
index 000000000..ce4db913f
--- /dev/null
+++ b/helms/odahu-flow-core/templates/operator/generated/odahuflow.odahu.org_batchinferencejobs.yaml
@@ -0,0 +1,191 @@
+
+---
+apiVersion: apiextensions.k8s.io/v1beta1
+kind: CustomResourceDefinition
+metadata:
+  annotations:
+    controller-gen.kubebuilder.io/version: v0.2.5
+  creationTimestamp: null
+  name: batchinferencejobs.odahuflow.odahu.org
+spec:
+  additionalPrinterColumns:
+  - JSONPath: .status.state
+    name: State
+    type: string
+  - JSONPath: .status.message
+    name: Message
+    type: string
+  - JSONPath: .status.reason
+    name: Reason
+    type: string
+  group: odahuflow.odahu.org
+  names:
+    kind: BatchInferenceJob
+    listKind: BatchInferenceJobList
+    plural: batchinferencejobs
+    shortNames:
+    - bij
+    singular: batchinferencejob
+  scope: Namespaced
+  subresources: {}
+  validation:
+    openAPIV3Schema:
+      description: BatchInferenceJob is the Schema for the batchinferencejobs API
+      properties:
+        apiVersion:
+          description: 'APIVersion defines the versioned schema of this representation
+            of an object. Servers should convert recognized schemas to the latest
+            internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources'
+          type: string
+        kind:
+          description: 'Kind is a string value representing the REST resource this
+            object represents. Servers may infer this from the endpoint the client
+            submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+          type: string
+        metadata:
+          type: object
+        spec:
+          description: BatchInferenceJobSpec defines the desired state of BatchInferenceJob
+          properties:
+            args:
+              description: 'Arguments to the entrypoint. The docker image''s CMD is
+                used if this is not provided. Variable references $(VAR_NAME) are
+                expanded using the container''s environment. If a variable cannot
+                be resolved, the reference in the input string will be unchanged.
+                The $(VAR_NAME) syntax can be escaped with a double $$, ie: $$(VAR_NAME).
+                Escaped references will never be expanded, regardless of whether the
+                variable exists or not. Cannot be updated. More info: https://kubernetes.io/docs/tasks/inject-data-application/define-command-argument-container/#running-a-command-in-a-shell'
+              items:
+                type: string
+              type: array
+            command:
+              description: 'Entrypoint array. Not executed within a shell. The docker
+                image''s ENTRYPOINT is used if this is not provided. Variable references
+                $(VAR_NAME) are expanded using the container''s environment. If a
+                variable cannot be resolved, the reference in the input string will
+                be unchanged. The $(VAR_NAME) syntax can be escaped with a double
+                $$, ie: $$(VAR_NAME). Escaped references will never be expanded, regardless
+                of whether the variable exists or not. Cannot be updated. More info:
+                https://kubernetes.io/docs/tasks/inject-data-application/define-command-argument-container/#running-a-command-in-a-shell'
+              items:
+                type: string
+              type: array
+            image:
+              description: 'INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
+                Important: Run "make" to regenerate code after modifying this file
+                Docker image'
+              type: string
+            inputConnection:
+              description: InputConnection is name of connection to object storage
+                bucket where input data are expected
+              type: string
+            inputPath:
+              description: InputPath is a source directory for BatchInferenceJob input
+                data relative to bucket root of InputConnection
+              type: string
+            modelConnection:
+              description: ModelConnection is name of connection to object storage
+                bucket where ML model files are expected
+              type: string
+            modelPath:
+              description: ModelPath is a directory inside ModelConnection where ML
+                model files are located
+              type: string
+            nodeSelector:
+              additionalProperties:
+                type: string
+              description: Node selector for specifying a node pool
+              type: object
+            outputConnection:
+              description: OutputConnection is name of connection to object storage
+                bucket where results should be saved
+              type: string
+            outputPath:
+              description: OutputPath is a destination directory for BatchInferenceJob
+                results relative to bucket root of OutputConnection
+              type: string
+            requestId:
+              description: BatchRequestID is unique identifier for BatchInferenceJob
+                that helps to correlate between Model input, model output and feedback
+                Take into account that it is not the same as kubeflow InferenceRequest
+                id Each BatchInferenceJob can process more than one InferenceRequest
+                (delivered in separate input file) So each BatchRequestID has set
+                of corresponding InferenceRequest and their IDs
+              type: string
+            resources:
+              description: Resources for model container The same format like k8s
+                uses for pod resources.
+              properties:
+                limits:
+                  description: Limits describes the maximum amount of compute resources
+                    allowed.
+                  properties:
+                    cpu:
+                      description: Read more about CPU resource here https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/#meaning-of-cpu
+                      type: string
+                    gpu:
+                      description: Read more about GPU resource here https://kubernetes.io/docs/tasks/manage-gpus/scheduling-gpus/#using-device-plugins
+                      type: string
+                    memory:
+                      description: Read more about memory resource here https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/#meaning-of-memory
+                      type: string
+                  type: object
+                requests:
+                  description: Requests describes the minimum amount of compute resources
+                    required.
+                  properties:
+                    cpu:
+                      description: Read more about CPU resource here https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/#meaning-of-cpu
+                      type: string
+                    gpu:
+                      description: Read more about GPU resource here https://kubernetes.io/docs/tasks/manage-gpus/scheduling-gpus/#using-device-plugins
+                      type: string
+                    memory:
+                      description: Read more about memory resource here https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/#meaning-of-memory
+                      type: string
+                  type: object
+              type: object
+          required:
+          - args
+          - command
+          - image
+          - inputConnection
+          - inputPath
+          - modelConnection
+          - modelPath
+          - outputConnection
+          - outputPath
+          - requestId
+          type: object
+        status:
+          description: BatchInferenceJobStatus defines the observed state of BatchInferenceJob
+          properties:
+            message:
+              type: string
+            podName:
+              type: string
+            reason:
+              type: string
+            state:
+              description: 'INSERT ADDITIONAL STATUS FIELD - define observed state
+                of cluster Important: Run "make" to regenerate code after modifying
+                this file'
+              type: string
+          required:
+          - message
+          - podName
+          - reason
+          - state
+          type: object
+      type: object
+  version: v1alpha1
+  versions:
+  - name: v1alpha1
+    served: true
+    storage: true
+status:
+  acceptedNames:
+    kind: ""
+    plural: ""
+  conditions: []
+  storedVersions: []
diff --git a/helms/odahu-flow-core/templates/operator/generated/operator-rbac-role.yaml b/helms/odahu-flow-core/templates/operator/generated/operator-rbac-role.yaml
index 43e282a11..6dc12fbf5 100644
--- a/helms/odahu-flow-core/templates/operator/generated/operator-rbac-role.yaml
+++ b/helms/odahu-flow-core/templates/operator/generated/operator-rbac-role.yaml
@@ -282,6 +282,26 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - odahuflow.odahu.org
+  resources:
+  - batchinferencejobs
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
+- apiGroups:
+  - odahuflow.odahu.org
+  resources:
+  - batchinferencejobs/status
+  verbs:
+  - get
+  - patch
+  - update
 - apiGroups:
   - odahuflow.odahu.org
   resources:
diff --git a/helms/odahu-flow-core/templates/operator/server.yaml b/helms/odahu-flow-core/templates/operator/server.yaml
index fb248c748..9ff6f908b 100644
--- a/helms/odahu-flow-core/templates/operator/server.yaml
+++ b/helms/odahu-flow-core/templates/operator/server.yaml
@@ -34,6 +34,8 @@ spec:
             value: "{{ include "odahuflow.image-name" (dict "root" . "service" .Values.config.packaging.modelPackager "tpl" "%sodahu-flow-model-packager:%s") }}"
           - name: TRAINING_MODEL_TRAINER_IMAGE
             value: "{{ include "odahuflow.image-name" (dict "root" . "service" .Values.config.training.modelTrainer "tpl" "%sodahu-flow-model-trainer:%s") }}"
+          - name: ODAHU_FLOW_TOOLS_IMAGE
+            value: "{{ include "odahuflow.image-name" (dict "root" . "service" .Values.config.batch.toolsImage "tpl" "%sodahu-flow-tools:%s") }}"
         resources:
           {{- toYaml .Values.operator.server.resources | nindent 10 }}
         command:
diff --git a/helms/odahu-flow-core/values.yaml b/helms/odahu-flow-core/values.yaml
index bd231265e..112ca1220 100644
--- a/helms/odahu-flow-core/values.yaml
+++ b/helms/odahu-flow-core/values.yaml
@@ -558,6 +558,42 @@ config:
     # Type: bool
     debug:
 
+  # Batch configuration
+  batch:
+    # Enable batch API/operator
+    # Type: boolean
+    enabled: true
+    # Batch namespace
+    # Required value
+    # Type: string
+    namespace: odahu-flow-batch
+    # tolerations for batch inference pod
+    tolerations: []
+    # Node pools to run batch jobs
+    nodePools: []
+    # RClone image that will be used to sync data with object storage
+    rcloneImage: rclone/rclone
+    # Tools image
+    toolsImage:
+
+toolsConfig:
+  # Auth configures connection parameters to ODAHU API Server
+  auth:
+    # ODAHU API URL
+    # Required
+    apiUrl: http://odahu-flow-api.odahu-flow.svc.cluster.local
+    # It is a mock for the future. Currently, it is always empty.
+    apiToken: ""
+    # OpenID client_id credential for service account
+    clientId: ""
+    # OpenID client_secret credential for service account
+    clientSecret: ""
+    # OpenID token url. Example: https://keycloak.cicd.odahu.org/auth/realms/legion/protocol/openid-connect/token
+    oauthOidcTokenEndpoint: ""
+  feedback:
+    # fluentd service to log model request/response
+    fluentd:
+      baseURL: http://fluentd.fluentd.svc.cluster.local
 
 # Service account used to upload odahu resources via odahuflowctl
 resource_uploader_sa: {}
diff --git a/packages/operator/.golangci.yml b/packages/operator/.golangci.yml
index ea1f83877..758ffd47f 100644
--- a/packages/operator/.golangci.yml
+++ b/packages/operator/.golangci.yml
@@ -78,3 +78,6 @@ issues:
         - golint
         - stylecheck
       text: "ExternalUrl"
+    - linters:
+        - gocritic
+      text: "ifElseChain"
diff --git a/packages/operator/PROJECT b/packages/operator/PROJECT
index 38753baa5..98e3e6a16 100644
--- a/packages/operator/PROJECT
+++ b/packages/operator/PROJECT
@@ -22,4 +22,10 @@ resources:
 - group: odahuflow
   kind: ModelRoute
   version: v1alpha1
+- group: odahuflow
+  kind: BatchInferenceService
+  version: v1alpha1
+- group: odahuflow
+  kind: BatchInferenceJob
+  version: v1alpha1
 version: "2"
diff --git a/packages/operator/api/v1alpha1/batchinferencejob_types.go b/packages/operator/api/v1alpha1/batchinferencejob_types.go
new file mode 100644
index 000000000..3a14e3c24
--- /dev/null
+++ b/packages/operator/api/v1alpha1/batchinferencejob_types.go
@@ -0,0 +1,125 @@
+/*
+
+
+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 v1alpha1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// BatchJobState defines current state
+type BatchJobState string
+
+const (
+	BatchScheduling BatchJobState = "scheduling"
+	BatchRunning    BatchJobState = "running"
+	BatchSucceeded  BatchJobState = "succeeded"
+	BatchFailed     BatchJobState = "failed"
+	BatchUnknown    BatchJobState = "unknown"
+)
+
+// EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
+// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.
+
+// BatchInferenceJobSpec defines the desired state of BatchInferenceJob
+type BatchInferenceJobSpec struct {
+	// INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
+	// Important: Run "make" to regenerate code after modifying this file
+	// Docker image
+	Image string `json:"image"`
+	// Entrypoint array. Not executed within a shell.
+	// The docker image's ENTRYPOINT is used if this is not provided.
+	// Variable references $(VAR_NAME) are expanded using the container's environment. If a variable
+	// cannot be resolved, the reference in the input string will be unchanged. The $(VAR_NAME) syntax
+	// can be escaped with a double $$, ie: $$(VAR_NAME). Escaped references will never be expanded,
+	// regardless of whether the variable exists or not.
+	// Cannot be updated.
+	// More info: https://kubernetes.io/docs/tasks/inject-data-application/define-command-argument-container/#running-a-command-in-a-shell
+	Command []string `json:"command"`
+	// Arguments to the entrypoint.
+	// The docker image's CMD is used if this is not provided.
+	// Variable references $(VAR_NAME) are expanded using the container's environment. If a variable
+	// cannot be resolved, the reference in the input string will be unchanged. The $(VAR_NAME) syntax
+	// can be escaped with a double $$, ie: $$(VAR_NAME). Escaped references will never be expanded,
+	// regardless of whether the variable exists or not.
+	// Cannot be updated.
+	// More info: https://kubernetes.io/docs/tasks/inject-data-application/define-command-argument-container/#running-a-command-in-a-shell
+	Args []string `json:"args"`
+	// InputConnection is name of connection to object storage bucket where input data are expected
+	InputConnection string `json:"inputConnection"`
+	// InputPath is a source directory for BatchInferenceJob input data
+	// relative to bucket root of InputConnection
+	InputPath string `json:"inputPath"`
+	// OutputConnection is name of connection to object storage bucket where results should be
+	// saved
+	OutputConnection string `json:"outputConnection"`
+	// OutputPath is a destination directory for BatchInferenceJob results
+	// relative to bucket root of OutputConnection
+	OutputPath string `json:"outputPath"`
+	// ModelConnection is name of connection to object storage bucket where ML model files are expected
+	ModelConnection string `json:"modelConnection"`
+	// ModelPath is a directory inside ModelConnection where ML model files are located
+	ModelPath string `json:"modelPath"`
+	// Node selector for specifying a node pool
+	NodeSelector map[string]string `json:"nodeSelector,omitempty"`
+	// Resources for model container
+	// The same format like k8s uses for pod resources.
+	Resources *ResourceRequirements `json:"resources,omitempty"`
+	// requestId is unique identifier for InferenceJob that helps to correlate between
+	// Model input, model output and feedback.
+	// Take into account that it is not the same as kubeflow InferenceRequest id.
+	// Each BatchInferenceJob can process more than one InferenceRequest (delivered in separate input file).
+	// So each RequestID has set of corresponding InferenceRequest and their IDs.
+	BatchRequestID string `json:"requestId"`
+}
+
+// BatchInferenceJobStatus defines the observed state of BatchInferenceJob
+type BatchInferenceJobStatus struct {
+	// INSERT ADDITIONAL STATUS FIELD - define observed state of cluster
+	// Important: Run "make" to regenerate code after modifying this file
+	State BatchJobState `json:"state"`
+	Message string `json:"message"`
+	Reason string `json:"reason"`
+	PodName string `json:"podName"`
+}
+
+// +kubebuilder:object:root=true
+
+// BatchInferenceJob is the Schema for the batchinferencejobs API
+// +kubebuilder:printcolumn:name="State",type="string",JSONPath=".status.state"
+// +kubebuilder:printcolumn:name="Message",type="string",JSONPath=".status.message"
+// +kubebuilder:printcolumn:name="Reason",type="string",JSONPath=".status.reason"
+// +kubebuilder:resource:shortName=bij
+type BatchInferenceJob struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec   BatchInferenceJobSpec   `json:"spec,omitempty"`
+	Status BatchInferenceJobStatus `json:"status,omitempty"`
+}
+
+// +kubebuilder:object:root=true
+
+// BatchInferenceJobList contains a list of BatchInferenceJob
+type BatchInferenceJobList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []BatchInferenceJob `json:"items"`
+}
+
+func init() {
+	SchemeBuilder.Register(&BatchInferenceJob{}, &BatchInferenceJobList{})
+}
diff --git a/packages/operator/api/v1alpha1/zz_generated.deepcopy.go b/packages/operator/api/v1alpha1/zz_generated.deepcopy.go
index cde329777..5c6c96408 100644
--- a/packages/operator/api/v1alpha1/zz_generated.deepcopy.go
+++ b/packages/operator/api/v1alpha1/zz_generated.deepcopy.go
@@ -24,6 +24,117 @@ import (
 	runtime "k8s.io/apimachinery/pkg/runtime"
 )
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *BatchInferenceJob) DeepCopyInto(out *BatchInferenceJob) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	in.Spec.DeepCopyInto(&out.Spec)
+	out.Status = in.Status
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BatchInferenceJob.
+func (in *BatchInferenceJob) DeepCopy() *BatchInferenceJob {
+	if in == nil {
+		return nil
+	}
+	out := new(BatchInferenceJob)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *BatchInferenceJob) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *BatchInferenceJobList) DeepCopyInto(out *BatchInferenceJobList) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ListMeta.DeepCopyInto(&out.ListMeta)
+	if in.Items != nil {
+		in, out := &in.Items, &out.Items
+		*out = make([]BatchInferenceJob, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BatchInferenceJobList.
+func (in *BatchInferenceJobList) DeepCopy() *BatchInferenceJobList {
+	if in == nil {
+		return nil
+	}
+	out := new(BatchInferenceJobList)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *BatchInferenceJobList) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *BatchInferenceJobSpec) DeepCopyInto(out *BatchInferenceJobSpec) {
+	*out = *in
+	if in.Command != nil {
+		in, out := &in.Command, &out.Command
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	if in.Args != nil {
+		in, out := &in.Args, &out.Args
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	if in.NodeSelector != nil {
+		in, out := &in.NodeSelector, &out.NodeSelector
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+	if in.Resources != nil {
+		in, out := &in.Resources, &out.Resources
+		*out = new(ResourceRequirements)
+		(*in).DeepCopyInto(*out)
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BatchInferenceJobSpec.
+func (in *BatchInferenceJobSpec) DeepCopy() *BatchInferenceJobSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(BatchInferenceJobSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *BatchInferenceJobStatus) DeepCopyInto(out *BatchInferenceJobStatus) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BatchInferenceJobStatus.
+func (in *BatchInferenceJobStatus) DeepCopy() *BatchInferenceJobStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(BatchInferenceJobStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Connection) DeepCopyInto(out *Connection) {
 	*out = *in
diff --git a/packages/operator/cmd/tools/cmd/root.go b/packages/operator/cmd/tools/cmd/root.go
index 84996ad31..5f6d1d9f1 100644
--- a/packages/operator/cmd/tools/cmd/root.go
+++ b/packages/operator/cmd/tools/cmd/root.go
@@ -68,11 +68,16 @@ func initLogger() {
 
 func initConfig() {
 	// Don't forget to read config either from cfgFile or from home directory!
+	if cfgFile == "" {
+		cfgFile = os.Getenv("ODAHU_TOOLS_CONFIG")
+	}
 	if cfgFile != "" {
 		// Use config file from the flag.
+		zap.S().Infof("Config path: %s", cfgFile)
 		viper.SetConfigFile(cfgFile)
 	} else {
 		// Find home directory.
+		zap.S().Info("Config path is not configured explicitly. Looking for a config in home directory")
 		home, err := homedir.Dir()
 		if err != nil {
 			zap.S().Errorw("Unable to get home directory", zap.Error(err))
diff --git a/packages/operator/controllers/batchinferencejob_controller.go b/packages/operator/controllers/batchinferencejob_controller.go
new file mode 100644
index 000000000..19dbbe93a
--- /dev/null
+++ b/packages/operator/controllers/batchinferencejob_controller.go
@@ -0,0 +1,337 @@
+//
+//    Copyright 2021 EPAM Systems
+//
+//    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.
+//
+
+
+// TODO:
+//	 1. If modelPath is .zip, .gz, .tar then use rclone cp but not sync
+//		and then unzip model in validate-model step
+//	 2. Pass on helm chart level env variable with tools image name
+package controllers
+
+import (
+	"context"
+	"github.com/go-logr/logr"
+	"github.com/google/uuid"
+	"github.com/odahu/odahu-flow/packages/operator/pkg/apis/connection"
+	"github.com/odahu/odahu-flow/packages/operator/pkg/config"
+	"github.com/odahu/odahu-flow/packages/operator/pkg/utils"
+	tektonv1beta1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
+	"gopkg.in/yaml.v2"
+	corev1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	odahuflowv1alpha1 "github.com/odahu/odahu-flow/packages/operator/api/v1alpha1"
+)
+
+const (
+	batchIDLabel = "odahu.org/batchID"
+)
+
+type ConnGetter interface {
+	GetConnection(id string) (*connection.Connection, error)
+}
+
+type PodGetter interface {
+	GetPod(ctx context.Context, name string, namespace string) (corev1.Pod, error)
+}
+
+
+// BatchInferenceJobReconciler reconciles a BatchInferenceJob object
+type BatchInferenceJobReconciler struct {
+	client.Client
+	Log         logr.Logger
+	Scheme      *runtime.Scheme
+	connAPI     ConnGetter
+	cfg         config.BatchConfig
+	gpuResName  string
+	podGetter   PodGetter
+}
+
+type defaultPodGetter struct {
+	client.Client
+}
+
+func (p defaultPodGetter) GetPod(ctx context.Context, name string, namespace string) (corev1.Pod, error) {
+	trainerPod := &corev1.Pod{}
+	err := p.Get(
+		ctx,
+		types.NamespacedName{
+			Name:      name,
+			Namespace: namespace,
+		},
+		trainerPod,
+	)
+	return *trainerPod, err
+}
+
+
+func setDefaultOptions(options *BatchInferenceJobReconcilerOptions) {
+	if options.PodGetter == nil {
+		options.PodGetter = defaultPodGetter{options.Client}
+	}
+
+}
+
+type BatchInferenceJobReconcilerOptions struct {
+	Client client.Client
+	Schema *runtime.Scheme
+	ConnGetter        ConnGetter
+	PodGetter         PodGetter
+	Cfg               config.BatchConfig
+	ResourceGPUName	  string
+}
+
+
+func NewBatchInferenceJobReconciler(opts BatchInferenceJobReconcilerOptions) *BatchInferenceJobReconciler {
+
+	setDefaultOptions(&opts)
+
+	return &BatchInferenceJobReconciler{
+		Client: opts.Client,
+		Scheme: opts.Schema,
+		podGetter: opts.PodGetter,
+		connAPI: opts.ConnGetter,
+		cfg: opts.Cfg,
+		gpuResName: opts.ResourceGPUName,
+		Log: logf.Log.WithName("batch-inference-controller"),
+	}
+}
+
+// +kubebuilder:rbac:groups=odahuflow.odahu.org,resources=batchinferencejobs,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=odahuflow.odahu.org,resources=batchinferencejobs/status,verbs=get;update;patch
+// +kubebuilder:rbac:groups=tekton.dev,resources=taskruns,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=tekton.dev,resources=taskruns/status,verbs=get;update;patch
+
+func (r *BatchInferenceJobReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
+	ctx := context.Background()
+	log = r.Log.WithValues("BatchInferenceJobName", req.NamespacedName)
+	reconcileLoopUID := uuid.New().String()
+	log = log.WithValues("ReconcileLoopUID", reconcileLoopUID)
+
+	batchJob := &odahuflowv1alpha1.BatchInferenceJob{}
+
+	log.Info("Getting BatchInferenceJob")
+	if err := r.Get(ctx, req.NamespacedName, batchJob); err != nil {
+		if k8serrors.IsNotFound(err) {
+			log.Info("BatchInferenceJob is not found")
+			return reconcile.Result{}, nil
+		}
+		log.Error(err, "Unable to fetch BatchInferenceJob from Kube API")
+		return reconcile.Result{}, err
+	}
+
+	log.Info("Reconciling TaskRun")
+	tr, err := r.reconcileTaskRun(batchJob,  log)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+
+	log.Info("Inferring BatchInferenceJob status")
+	if len(tr.Status.Conditions) > 0 {
+		if err := r.syncStatusFromTaskRun(batchJob, tr); err != nil {
+			log.Error(err,"Unable to infer BatchInferenceJob status from TaskRun")
+			return ctrl.Result{}, err
+		}
+	} else {
+		batchJob.Status.State = odahuflowv1alpha1.BatchScheduling
+	}
+
+	batchJob.Status.PodName = tr.Status.PodName
+
+	log.Info(
+		"Updating BatchInferenceJob status",
+		"Status", batchJob.Status,
+	)
+	if err := r.Update(ctx, batchJob); err != nil {
+		log.Error(err, "Unable to update BatchInferenceJob status")
+		return ctrl.Result{}, err
+	}
+
+	return ctrl.Result{}, nil
+
+}
+
+func (r *BatchInferenceJobReconciler) generateTaskSpec(
+	batchJob *odahuflowv1alpha1.BatchInferenceJob,
+	) (*tektonv1beta1.TaskSpec, error) {
+	return BatchJobToTaskSpec(
+		batchJob, r.connAPI, r.gpuResName, r.cfg.RCloneImage, r.cfg.ToolsSecret, r.cfg.ToolsImage)
+}
+
+func (r *BatchInferenceJobReconciler) calculateStateByPod(
+	podName string, job *odahuflowv1alpha1.BatchInferenceJob) error {
+
+	pod, err := r.podGetter.GetPod(context.TODO(), podName, job.Namespace)
+	if err != nil {
+		return err
+	}
+	job.Status.PodName = podName
+	job.Status.Message = pod.Status.Message
+	job.Status.Reason = pod.Status.Reason
+
+	if pod.Status.Reason == evictedPodReason {
+		job.Status.State = odahuflowv1alpha1.BatchFailed
+		return nil
+	}
+
+	switch pod.Status.Phase {
+	case corev1.PodPending:
+		job.Status.State = odahuflowv1alpha1.BatchScheduling
+	case corev1.PodUnknown:
+		job.Status.State = odahuflowv1alpha1.BatchScheduling
+	case corev1.PodRunning:
+		job.Status.State = odahuflowv1alpha1.BatchRunning
+	}
+
+	return nil
+}
+
+func (r *BatchInferenceJobReconciler) syncStatusFromTaskRun(
+	batchJob *odahuflowv1alpha1.BatchInferenceJob, taskRun *tektonv1beta1.TaskRun) error {
+	lastCondition := taskRun.Status.Conditions[len(taskRun.Status.Conditions)-1]
+
+	switch lastCondition.Status {
+	case corev1.ConditionUnknown:
+		if len(taskRun.Status.PodName) != 0 {
+			if err := r.calculateStateByPod(taskRun.Status.PodName, batchJob); err != nil {
+				return err
+			}
+		} else {
+			batchJob.Status.State = odahuflowv1alpha1.BatchScheduling
+			batchJob.Status.Message = lastCondition.Message
+			batchJob.Status.Reason = lastCondition.Reason
+		}
+	case corev1.ConditionTrue:
+		batchJob.Status.State = odahuflowv1alpha1.BatchSucceeded
+		batchJob.Status.Message = lastCondition.Message
+		batchJob.Status.Reason = lastCondition.Reason
+
+	case corev1.ConditionFalse:
+		batchJob.Status.State = odahuflowv1alpha1.BatchFailed
+		batchJob.Status.Message = lastCondition.Message
+		batchJob.Status.Reason = lastCondition.Reason
+	default:
+		batchJob.Status.State = odahuflowv1alpha1.BatchScheduling
+	}
+	return nil
+}
+
+func logYAML(prefix string, obj interface{}, log logr.Logger) {
+	b, err := yaml.Marshal(obj)
+	if err != nil {
+		log.Error(err, "Unable to serialize to json for logging")
+	}
+	log.Info(prefix + ":\"" + string(b) + "\"")
+}
+
+func (r *BatchInferenceJobReconciler) reconcileTaskRun(
+	job *odahuflowv1alpha1.BatchInferenceJob, log logr.Logger,
+) (*tektonv1beta1.TaskRun, error) {
+
+	if job.Status.State != "" && job.Status.State != odahuflowv1alpha1.BatchUnknown {
+		taskRun := &tektonv1beta1.TaskRun{}
+		log.Info(
+			"Getting TaskRun that should exist because of BatchInferenceJob has not unknown status",
+			"Status", job.Status)
+		if err := r.Get(
+			context.TODO(), types.NamespacedName{Name: job.Name, Namespace: r.cfg.Namespace}, taskRun); err != nil {
+			log.Error(err, "Unable to get TaskRun that should exist")
+			return nil, err
+		}
+		return taskRun, nil
+	}
+
+
+	taskSpec, err := r.generateTaskSpec(job)
+	if err != nil {
+		return nil, err
+	}
+
+	var affinity *corev1.Affinity
+	if len(job.Spec.NodeSelector) == 0 {
+		affinity = utils.BuildNodeAffinity(r.cfg.NodePools)
+	}
+
+	taskRun := &tektonv1beta1.TaskRun{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      job.Name,
+			Namespace: job.Namespace,
+			Labels: map[string]string{
+				batchIDLabel: job.Name,
+			},
+		},
+		Spec: tektonv1beta1.TaskRunSpec{
+			TaskSpec: taskSpec,
+			Timeout:  &metav1.Duration{Duration: r.cfg.Timeout},
+			PodTemplate: &tektonv1beta1.PodTemplate{
+				Tolerations:  r.cfg.Tolerations,
+				NodeSelector: job.Spec.NodeSelector,
+				Affinity:     affinity,
+			},
+		},
+	}
+
+	if err := controllerutil.SetControllerReference(job, taskRun, r.Scheme); err != nil {
+		return nil, err
+	}
+
+	found := &tektonv1beta1.TaskRun{}
+	err = r.Get(context.TODO(), types.NamespacedName{
+		Name: taskRun.Name, Namespace: r.cfg.Namespace,
+	}, found)
+	if err != nil && k8serrors.IsNotFound(err) {
+		log.Info("TaskRun does not exist. Creating")
+		if err := r.Create(context.TODO(), taskRun); err != nil {
+			logYAML("Unable to create TaskRun", taskRun, log)
+			return nil, err
+		}
+		return taskRun, nil
+	} else if err != nil {
+		return nil, err
+	}
+
+	log.Info("TaskRun exists. Re-Creating")
+
+	log.Info("Deleting TaskRun")
+	if err := r.Delete(context.TODO(), found); err != nil {
+		return nil, err
+	}
+
+	log.Info("Creating TaskRun")
+	if  err := r.Create(context.TODO(), taskRun); err != nil {
+		logYAML("Unable to create TaskRun", taskRun, log)
+		return nil, err
+	}
+	log.Info("TaskRun is created")
+	return taskRun, nil
+}
+
+
+func (r *BatchInferenceJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&odahuflowv1alpha1.BatchInferenceJob{}).
+		Owns(&tektonv1beta1.TaskRun{}).
+		Complete(r)
+}
diff --git a/packages/operator/controllers/batchinferencejob_steps.go b/packages/operator/controllers/batchinferencejob_steps.go
new file mode 100644
index 000000000..2b20857a0
--- /dev/null
+++ b/packages/operator/controllers/batchinferencejob_steps.go
@@ -0,0 +1,271 @@
+package controllers
+
+import (
+	"fmt"
+	tektonv1beta1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
+	corev1 "k8s.io/api/core/v1"
+	"path"
+)
+
+const (
+	pathToOdahuToolsBin = "/opt/odahu-flow/odahu-tools"
+	toolsConfigVolume = "config"
+	inputRCloneCfgName = "odahu-data-input"
+	outputRCloneCfgName = "odahu-data-output"
+	modelRCloneCfgName = "odahu-data-model"
+)
+
+
+var (
+	toolsConfigVM = corev1.VolumeMount{
+		Name: toolsConfigVolume,
+		ReadOnly:         true,
+		MountPath:        path.Join(XDGConfigHome, "odahu", ".odahu-tools.yaml"),
+		SubPath:          ".odahu-tools.yaml",
+	}
+)
+
+// paths
+var (
+	XDGConfigHome   = path.Join(workspacePath, "config")
+	// Dir where raw data should be synced
+	rawInputPath    = path.Join(workspacePath, "odahu-ws-raw-input")
+	// Dir where raw model should be copied after validation and unzip
+	rawModelPath    = path.Join(workspacePath, "odahu-ws-raw-model")
+	// Dir where raw data should be copied after validation
+	odahuInputPath       = path.Join(workspacePath, "odahu-ws-input")
+	// Dir where raw model should be copied after validation and unzip
+	odahuModelPath  = path.Join(workspacePath, "odahu-ws-model")
+	// Dir where raw user output is expected
+	odahuRawOutputPath = path.Join(workspacePath, "odahu-ws-raw-output")
+	// Dir where raw user output is copied after validation
+	outputPath = path.Join(workspacePath, "odahu-ws-output")
+)
+
+// ENVs
+var (
+	XDGConfigHomeEnv = corev1.EnvVar{
+		Name:  "XDG_CONFIG_HOME",
+		Value: XDGConfigHome,
+	}
+	ToolsConfigPathEnv = corev1.EnvVar{
+		Name:  "ODAHU_TOOLS_CONFIG",
+		Value: toolsConfigVM.MountPath,
+	}
+	OdahuModelPathEnv = corev1.EnvVar{
+		Name:  "ODAHU_MODEL_PATH",
+		Value: odahuModelPath,
+	}
+	OdahuInputPathEnv = corev1.EnvVar{
+		Name:  "ODAHU_INPUT_PATH",
+		Value: odahuInputPath,
+	}
+	OdahuOutputPathEnv = corev1.EnvVar{
+		Name:  "ODAHU_OUTPUT_PATH",
+		Value: odahuRawOutputPath,
+	}
+)
+
+// GetConfigureRCloneStep return step that
+// configures environment (rclone config) for syncing data and model
+// using ODAHU connections
+func GetConfigureRCloneStep(image string, inpConn string,
+	outConn string, modelConn string, res corev1.ResourceRequirements) tektonv1beta1.Step {
+
+	var args = []string{"auth", "configure-rclone"}
+	args = append(args, "--conn", fmt.Sprintf("%s:%s", inpConn, inputRCloneCfgName))
+	args = append(args, "--conn", fmt.Sprintf("%s:%s", outConn, outputRCloneCfgName))
+	args = append(args, "--conn", fmt.Sprintf("%s:%s", modelConn, modelRCloneCfgName))
+	return tektonv1beta1.Step{
+		Container: corev1.Container{
+			Name:         "configure-rclone",
+			Image:        image,
+			Command:      []string{pathToOdahuToolsBin},
+			Args:         args,
+			Env:          []corev1.EnvVar{XDGConfigHomeEnv, ToolsConfigPathEnv},
+			VolumeMounts: []corev1.VolumeMount{toolsConfigVM},
+			Resources: res,
+		},
+	}
+}
+
+// GetSyncDataStep return step that
+// syncs input data to pre-stage directory inside Pod
+// where input will be validated and copied to user container's input directory
+func GetSyncDataStep(
+	rcloneImage string,
+	bucketName string,
+	inputPath string,
+	res corev1.ResourceRequirements,
+	) tektonv1beta1.Step {
+	sourcePrefix := fmt.Sprintf("%s:%s", inputRCloneCfgName, bucketName)
+	source := path.Join(sourcePrefix, inputPath)
+	return tektonv1beta1.Step{
+		Container: corev1.Container{
+			Name:         "sync-data",
+			Image:        rcloneImage,
+			Command:      []string{"rclone"},
+			Args:         []string{"sync", "-P", source, rawInputPath},
+			Env:          []corev1.EnvVar{XDGConfigHomeEnv},
+			Resources: res,
+		},
+	}
+}
+
+// GetCopyUnzipModelStep return step that
+// copy model zipped file to pre-stage directory inside Pod, unzip it and pass to user container
+func GetCopyUnzipModelStep(
+	rcloneImage string,
+	bucketName string,
+	modelPath string,
+	res corev1.ResourceRequirements,
+	) tektonv1beta1.Step {
+	sourcePrefix := fmt.Sprintf("%s:%s", modelRCloneCfgName, bucketName)
+	source := path.Join(sourcePrefix, modelPath)
+
+	baseName := path.Base(modelPath)
+	localZippedPath := path.Join(rawModelPath, baseName)
+
+	cmdPipeline := fmt.Sprintf("rclone copy %s %s && mkdir -p %s && tar -xzvf %s -C %s",
+		source, rawModelPath, odahuModelPath, localZippedPath, odahuModelPath,
+	)
+
+	return tektonv1beta1.Step{
+		Container: corev1.Container{
+			Name:         "copy-unzip-model",
+			Image:        rcloneImage,
+			Command:      []string{"sh"},
+			Args:         []string{"-c", cmdPipeline},
+			Env:          []corev1.EnvVar{XDGConfigHomeEnv},
+			Resources: res,
+		},
+	}
+}
+// GetSyncModelStep return step that
+// syncs model to pre-stage directory inside Pod
+// where model will be validated and copied to user container's input directory
+func GetSyncModelStep(
+	rcloneImage string,
+	bucketName string,
+	modelPath string,
+	res corev1.ResourceRequirements,
+	) tektonv1beta1.Step {
+	sourcePrefix := fmt.Sprintf("%s:%s", modelRCloneCfgName, bucketName)
+	source := path.Join(sourcePrefix, modelPath)
+	return tektonv1beta1.Step{
+		Container: corev1.Container{
+			Name:         "sync-model",
+			Image:        rcloneImage,
+			Command:      []string{"rclone"},
+			Args:         []string{"sync", "-P", source, odahuModelPath},
+			Env:          []corev1.EnvVar{XDGConfigHomeEnv},
+			Resources: res,
+		},
+	}
+}
+
+// GetValidateInputStep return step that
+// validates raw input according kubeflow prediction api (version 2) InferenceRequest object.
+// Only validated files are copied to user container for inference.
+func GetValidateInputStep(image string, res corev1.ResourceRequirements) tektonv1beta1.Step {
+	return tektonv1beta1.Step{
+		Container: corev1.Container{
+			Name:         "validate-input",
+			Image:        image,
+			Command:      []string{pathToOdahuToolsBin},
+			Args:         []string{"batch", "validate", "input", "-s", rawInputPath, "-d", odahuInputPath},
+			VolumeMounts: []corev1.VolumeMount{toolsConfigVM},
+			Env:          []corev1.EnvVar{ToolsConfigPathEnv},
+			Resources: res,
+		},
+	}
+}
+
+// GetLogInputStep return step that
+// logs model input to feedback storage (catch requests)
+func GetLogInputStep(image string, requestID string, res corev1.ResourceRequirements) tektonv1beta1.Step {
+	return tektonv1beta1.Step{
+		Container: corev1.Container{
+			Name:         "log-input",
+			Image:        image,
+			Command:      []string{pathToOdahuToolsBin},
+			Args:         []string{"batch", "log", "input", odahuInputPath, "-m", odahuModelPath, "-r", requestID},
+			VolumeMounts: []corev1.VolumeMount{toolsConfigVM},
+			Env:          []corev1.EnvVar{ToolsConfigPathEnv},
+			Resources: res,
+		},
+	}
+}
+
+// GetUserContainer return step that
+// execute user defined container for inference
+func GetUserContainer(
+	image string, command []string, args []string, res corev1.ResourceRequirements) tektonv1beta1.Step {
+	return tektonv1beta1.Step{
+		Container: corev1.Container{
+			Name:         "user-container",
+			Image:        image,
+			Command:      command,
+			Args:         args,
+			VolumeMounts: []corev1.VolumeMount{toolsConfigVM},
+			Env: []corev1.EnvVar{OdahuInputPathEnv, OdahuOutputPathEnv, OdahuModelPathEnv},
+			Resources: res,
+		},
+	}
+}
+
+
+// GetValidateOutputStep return step that
+// validates raw output according kubeflow prediction api (version 2) InferenceResponse object.
+// Only validated files are copied from user container results to destination
+func GetValidateOutputStep(image string, res corev1.ResourceRequirements) tektonv1beta1.Step {
+	return tektonv1beta1.Step{
+		Container: corev1.Container{
+			Name:         "validate-output",
+			Image:        image,
+			Command:      []string{pathToOdahuToolsBin},
+			Args:         []string{"batch", "validate", "output", "-s", odahuRawOutputPath, "-d", outputPath},
+			VolumeMounts: []corev1.VolumeMount{toolsConfigVM},
+			Env:          []corev1.EnvVar{ToolsConfigPathEnv},
+			Resources: res,
+		},
+	}
+}
+
+// GetLogOutputStep return step that
+// logs model output to feedback storage (catch responses)
+func GetLogOutputStep(image string, requestID string, res corev1.ResourceRequirements) tektonv1beta1.Step {
+	return tektonv1beta1.Step{
+		Container: corev1.Container{
+			Name:         "log-output",
+			Image:        image,
+			Command:      []string{pathToOdahuToolsBin},
+			Args:         []string{"batch", "log", "output", outputPath, "-m", odahuModelPath, "-r", requestID},
+			VolumeMounts: []corev1.VolumeMount{toolsConfigVM},
+			Env:          []corev1.EnvVar{ToolsConfigPathEnv},
+			Resources: res,
+		},
+	}
+}
+
+// GetSyncOutputStep return step that
+// syncs output data to bucket
+func GetSyncOutputStep(
+	rcloneImage string,
+	bucketName string,
+	remoteOutputPath string,
+	res corev1.ResourceRequirements,
+) tektonv1beta1.Step {
+	prefix := fmt.Sprintf("%s:%s", outputRCloneCfgName, bucketName)
+	dest := path.Join(prefix, remoteOutputPath)
+	return tektonv1beta1.Step{
+		Container: corev1.Container{
+			Name:         "sync-output",
+			Image:        rcloneImage,
+			Command:      []string{"rclone"},
+			Args:         []string{"sync", "-P", outputPath, dest},
+			Env:          []corev1.EnvVar{XDGConfigHomeEnv},
+			Resources: res,
+		},
+	}
+}
\ No newline at end of file
diff --git a/packages/operator/controllers/batchinferencejob_utils.go b/packages/operator/controllers/batchinferencejob_utils.go
new file mode 100644
index 000000000..1c5ec9c6e
--- /dev/null
+++ b/packages/operator/controllers/batchinferencejob_utils.go
@@ -0,0 +1,107 @@
+package controllers
+
+import (
+	"fmt"
+	"github.com/odahu/odahu-flow/packages/operator/api/v1alpha1"
+	"github.com/odahu/odahu-flow/packages/operator/pkg/rclone"
+	"github.com/odahu/odahu-flow/packages/operator/pkg/repository/util/kubernetes"
+	"github.com/odahu/odahu-flow/packages/operator/pkg/utils"
+	tektonv1beta1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
+	"go.uber.org/multierr"
+	corev1 "k8s.io/api/core/v1"
+)
+
+// getBucketNames return bucket names for apitypes.InferenceService in next order:
+// inputConnection bucket, outputConnection bucket, modelConnection bucket
+func getBucketNames(job *v1alpha1.BatchInferenceJob, connAPI ConnGetter) (
+	iBucket string, oBucket string, mBucket string, errs error,
+	) {
+
+
+	conn, err := connAPI.GetConnection(job.Spec.InputConnection)
+	if err != nil {
+		errs = multierr.Append(errs, err)
+	} else {
+		iBucket, _, err = rclone.GetBucketAndPath(&conn.Spec)
+		if err != nil {
+			errs = multierr.Append(errs, err)
+		}
+	}
+	conn, err = connAPI.GetConnection(job.Spec.OutputConnection)
+	if err != nil {
+		errs = multierr.Append(errs, err)
+	} else {
+		oBucket, _, err = rclone.GetBucketAndPath(&conn.Spec)
+		if err != nil {
+			errs = multierr.Append(errs, err)
+		}
+	}
+	conn, err = connAPI.GetConnection(job.Spec.ModelConnection)
+	if err != nil {
+		errs = multierr.Append(errs, err)
+	} else {
+		mBucket, _, err = rclone.GetBucketAndPath(&conn.Spec)
+		if err != nil {
+			errs = multierr.Append(errs, err)
+		}
+	}
+	return iBucket, oBucket, mBucket, errs
+}
+
+
+// BatchJobToTaskSpec generate tektoncd TaskSpec based on v1alpha1.BatchInferenceJob
+func BatchJobToTaskSpec(job *v1alpha1.BatchInferenceJob,
+	connAPI ConnGetter,
+	gpuResourceName string, rcloneImage string,
+	toolsSecret string,
+	toolsImage string,
+) (ts *tektonv1beta1.TaskSpec, err error) {
+
+	jobRes, err := kubernetes.ConvertOdahuflowResourcesToK8s(job.Spec.Resources, gpuResourceName)
+	if err != nil {
+		return ts, fmt.Errorf("unable to convert odahu resources to kubernetes resources: %s", err)
+	}
+
+	helpContainerRes := utils.CalculateHelperContainerResources(jobRes, gpuResourceName)
+
+	iBucket, oBucket, mBucket, err := getBucketNames(job, connAPI)
+	if err != nil {
+		return ts, err
+	}
+
+	// TODO: Add model validation step
+	steps := []tektonv1beta1.Step{
+		GetConfigureRCloneStep(
+			toolsImage, job.Spec.InputConnection, job.Spec.OutputConnection, job.Spec.ModelConnection, helpContainerRes),
+		GetSyncDataStep(rcloneImage, iBucket, job.Spec.InputPath, helpContainerRes),
+	}
+
+
+	// TODO: Conditionally handle case when model is zipped file using GetCopyUnzipModelStep
+	steps = append(steps, GetSyncModelStep(rcloneImage, mBucket, job.Spec.ModelPath, helpContainerRes))
+
+	steps = append(steps, []tektonv1beta1.Step{
+		GetValidateInputStep(toolsImage, helpContainerRes),
+		GetLogInputStep(toolsImage, job.Spec.BatchRequestID, helpContainerRes),
+		GetUserContainer(job.Spec.Image, job.Spec.Command, job.Spec.Args, jobRes),
+		GetValidateOutputStep(toolsImage, helpContainerRes),
+		GetLogOutputStep(toolsImage, job.Spec.BatchRequestID, helpContainerRes),
+		GetSyncOutputStep(rcloneImage, oBucket, job.Spec.OutputPath, helpContainerRes),
+	}...)
+
+	ts = &tektonv1beta1.TaskSpec{
+		Steps:   steps,
+		Volumes: []corev1.Volume{
+			{
+				Name: toolsConfigVolume,
+				VolumeSource: corev1.VolumeSource{
+					Secret: &corev1.SecretVolumeSource{
+						SecretName: toolsSecret,
+					},
+				},
+			},
+		},
+	}
+
+	return ts, err
+}
\ No newline at end of file
diff --git a/packages/operator/controllers/mocks/ConnGetter.go b/packages/operator/controllers/mocks/ConnGetter.go
new file mode 100644
index 000000000..8faed86a0
--- /dev/null
+++ b/packages/operator/controllers/mocks/ConnGetter.go
@@ -0,0 +1,37 @@
+// Code generated by mockery v2.2.1. DO NOT EDIT.
+
+package mocks
+
+import (
+	connection "github.com/odahu/odahu-flow/packages/operator/pkg/apis/connection"
+
+	mock "github.com/stretchr/testify/mock"
+)
+
+// ConnGetter is an autogenerated mock type for the ConnGetter type
+type ConnGetter struct {
+	mock.Mock
+}
+
+// GetConnection provides a mock function with given fields: id
+func (_m *ConnGetter) GetConnection(id string) (*connection.Connection, error) {
+	ret := _m.Called(id)
+
+	var r0 *connection.Connection
+	if rf, ok := ret.Get(0).(func(string) *connection.Connection); ok {
+		r0 = rf(id)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*connection.Connection)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(string) error); ok {
+		r1 = rf(id)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
diff --git a/packages/operator/controllers/mocks/PodGetter.go b/packages/operator/controllers/mocks/PodGetter.go
new file mode 100644
index 000000000..3671e871b
--- /dev/null
+++ b/packages/operator/controllers/mocks/PodGetter.go
@@ -0,0 +1,37 @@
+// Code generated by mockery v2.2.1. DO NOT EDIT.
+
+package mocks
+
+import (
+	context "context"
+
+	mock "github.com/stretchr/testify/mock"
+
+	v1 "k8s.io/api/core/v1"
+)
+
+// PodGetter is an autogenerated mock type for the PodGetter type
+type PodGetter struct {
+	mock.Mock
+}
+
+// GetPod provides a mock function with given fields: ctx, name, namespace
+func (_m *PodGetter) GetPod(ctx context.Context, name string, namespace string) (v1.Pod, error) {
+	ret := _m.Called(ctx, name, namespace)
+
+	var r0 v1.Pod
+	if rf, ok := ret.Get(0).(func(context.Context, string, string) v1.Pod); ok {
+		r0 = rf(ctx, name, namespace)
+	} else {
+		r0 = ret.Get(0).(v1.Pod)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string, string) error); ok {
+		r1 = rf(ctx, name, namespace)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
diff --git a/packages/operator/docs/docs.go b/packages/operator/docs/docs.go
index 77766bd67..4c2556410 100644
--- a/packages/operator/docs/docs.go
+++ b/packages/operator/docs/docs.go
@@ -2076,6 +2076,46 @@ var doc = `{
                 }
             }
         },
+        "BatchConfig": {
+            "type": "object",
+            "properties": {
+                "enabled": {
+                    "description": "Enable batch API/operator",
+                    "type": "boolean"
+                },
+                "namespace": {
+                    "description": "Kubernetes namespace, where BatchInferenceService and BatchInferenceJob will be created",
+                    "type": "string"
+                },
+                "nodePools": {
+                    "description": "Node pools to run batch jobs",
+                    "type": "array",
+                    "items": {
+                        "$ref": "#/definitions/NodePool"
+                    }
+                },
+                "rcloneImage": {
+                    "description": "RClone image that will be used to sync data with object storage",
+                    "type": "string"
+                },
+                "timeout": {
+                    "description": "Timeout for full batch process",
+                    "type": "string"
+                },
+                "tolerations": {
+                    "description": "Kubernetes tolerations for batch jobs",
+                    "type": "string"
+                },
+                "toolsImage": {
+                    "description": "ODAHU tools image",
+                    "type": "string"
+                },
+                "toolsSecret": {
+                    "description": "ODAHU tools image",
+                    "type": "string"
+                }
+            }
+        },
         "Claims": {
             "type": "object",
             "properties": {
@@ -2130,6 +2170,10 @@ var doc = `{
                     "type": "object",
                     "$ref": "#/definitions/APIConfig"
                 },
+                "batch": {
+                    "type": "object",
+                    "$ref": "#/definitions/BatchConfig"
+                },
                 "common": {
                     "type": "object",
                     "$ref": "#/definitions/CommonConfig"
diff --git a/packages/operator/docs/swagger.json b/packages/operator/docs/swagger.json
index e483589b8..89378b60c 100644
--- a/packages/operator/docs/swagger.json
+++ b/packages/operator/docs/swagger.json
@@ -2060,6 +2060,46 @@
                 }
             }
         },
+        "BatchConfig": {
+            "type": "object",
+            "properties": {
+                "enabled": {
+                    "description": "Enable batch API/operator",
+                    "type": "boolean"
+                },
+                "namespace": {
+                    "description": "Kubernetes namespace, where BatchInferenceService and BatchInferenceJob will be created",
+                    "type": "string"
+                },
+                "nodePools": {
+                    "description": "Node pools to run batch jobs",
+                    "type": "array",
+                    "items": {
+                        "$ref": "#/definitions/NodePool"
+                    }
+                },
+                "rcloneImage": {
+                    "description": "RClone image that will be used to sync data with object storage",
+                    "type": "string"
+                },
+                "timeout": {
+                    "description": "Timeout for full batch process",
+                    "type": "string"
+                },
+                "tolerations": {
+                    "description": "Kubernetes tolerations for batch jobs",
+                    "type": "string"
+                },
+                "toolsImage": {
+                    "description": "ODAHU tools image",
+                    "type": "string"
+                },
+                "toolsSecret": {
+                    "description": "ODAHU tools image",
+                    "type": "string"
+                }
+            }
+        },
         "Claims": {
             "type": "object",
             "properties": {
@@ -2114,6 +2154,10 @@
                     "type": "object",
                     "$ref": "#/definitions/APIConfig"
                 },
+                "batch": {
+                    "type": "object",
+                    "$ref": "#/definitions/BatchConfig"
+                },
                 "common": {
                     "type": "object",
                     "$ref": "#/definitions/CommonConfig"
diff --git a/packages/operator/docs/swagger.yaml b/packages/operator/docs/swagger.yaml
index ef7b4e792..45804dad5 100644
--- a/packages/operator/docs/swagger.yaml
+++ b/packages/operator/docs/swagger.yaml
@@ -45,6 +45,35 @@ definitions:
         description: OpenID token url
         type: string
     type: object
+  BatchConfig:
+    properties:
+      enabled:
+        description: Enable batch API/operator
+        type: boolean
+      namespace:
+        description: Kubernetes namespace, where BatchInferenceService and BatchInferenceJob will be created
+        type: string
+      nodePools:
+        description: Node pools to run batch jobs
+        items:
+          $ref: '#/definitions/NodePool'
+        type: array
+      rcloneImage:
+        description: RClone image that will be used to sync data with object storage
+        type: string
+      timeout:
+        description: Timeout for full batch process
+        type: string
+      tolerations:
+        description: Kubernetes tolerations for batch jobs
+        type: string
+      toolsImage:
+        description: ODAHU tools image
+        type: string
+      toolsSecret:
+        description: ODAHU tools image
+        type: string
+    type: object
   Claims:
     properties:
       email:
@@ -85,6 +114,9 @@ definitions:
       api:
         $ref: '#/definitions/APIConfig'
         type: object
+      batch:
+        $ref: '#/definitions/BatchConfig'
+        type: object
       common:
         $ref: '#/definitions/CommonConfig'
         type: object
diff --git a/packages/operator/go.mod b/packages/operator/go.mod
index 9c05aec40..76686f73b 100644
--- a/packages/operator/go.mod
+++ b/packages/operator/go.mod
@@ -14,7 +14,6 @@ require (
 	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
@@ -43,6 +42,7 @@ require (
 	github.com/stretchr/testify v1.6.1
 	github.com/swaggo/swag v1.5.0
 	github.com/tektoncd/pipeline v0.13.1-0.20200625065359-44f22a067b75
+	github.com/tinylib/msgp v1.1.0 // indirect
 	github.com/xeipuuv/gojsonpointer v0.0.0-20190809123943-df4f5c81cb3b // indirect
 	github.com/xeipuuv/gojsonschema v1.1.0
 	github.com/zsais/go-gin-prometheus v0.1.0
@@ -57,6 +57,7 @@ require (
 	k8s.io/apimachinery v0.18.7-rc.0
 	k8s.io/client-go v11.0.1-0.20190805182717-6502b5e7b1b5+incompatible
 	k8s.io/kubernetes v1.14.7
+	knative.dev/pkg v0.0.0-20200812224206-44c860147a87
 	knative.dev/serving v0.17.0
 	odahu-commons v0.0.0
 	sigs.k8s.io/controller-runtime v0.6.1
diff --git a/packages/operator/main.go b/packages/operator/main.go
index 0159120c0..0f5c077ee 100644
--- a/packages/operator/main.go
+++ b/packages/operator/main.go
@@ -17,6 +17,7 @@ limitations under the License.
 package main
 
 import (
+	conn_api_client "github.com/odahu/odahu-flow/packages/operator/pkg/apiclient/connection"
 	"os"
 
 	istioschema "github.com/aspenmesh/istio-client-go/pkg/client/clientset/versioned/scheme"
@@ -123,6 +124,14 @@ func start(cmd *cobra.Command, args []string) {
 	// Setup the training toolchain repository
 	authCfg := odahuConfig.Operator.Auth
 
+	connAPI := conn_api_client.NewClient(
+		authCfg.APIURL,
+		authCfg.APIToken,
+		authCfg.ClientID,
+		authCfg.ClientSecret,
+		authCfg.OAuthOIDCTokenEndpoint,
+	)
+
 	if odahuConfig.Training.Enabled {
 		trainAPIClient := train_api_client.NewClient(
 			authCfg.APIURL, authCfg.APIToken, authCfg.ClientID, authCfg.ClientSecret, authCfg.OAuthOIDCTokenEndpoint,
@@ -164,6 +173,22 @@ func start(cmd *cobra.Command, args []string) {
 		}
 	}
 
+	if odahuConfig.Batch.Enabled {
+
+		batchOpts := controllers.BatchInferenceJobReconcilerOptions{
+			Client: mgr.GetClient(),
+			Schema: mgr.GetScheme(),
+			ConnGetter:      connAPI,
+			Cfg:             odahuConfig.Batch,
+			ResourceGPUName: odahuConfig.Common.ResourceGPUName,
+		}
+
+		if err = controllers.NewBatchInferenceJobReconciler(batchOpts).SetupWithManager(mgr); err != nil {
+			setupLog.Error(err, "unable to create controller", "controller", "Batch")
+			os.Exit(1)
+		}
+	}
+
 	// +kubebuilder:scaffold:builder
 
 	setupLog.Info("starting manager")
diff --git a/packages/operator/pkg/config/batch.go b/packages/operator/pkg/config/batch.go
new file mode 100644
index 000000000..bd2c6d6a0
--- /dev/null
+++ b/packages/operator/pkg/config/batch.go
@@ -0,0 +1,42 @@
+package config
+
+import (
+	"github.com/spf13/viper"
+	corev1 "k8s.io/api/core/v1"
+	"time"
+)
+
+func init() {
+	// workaround https://github.com/spf13/viper/issues/761
+	_ = viper.BindEnv("batch.toolsImage", "ODAHU_FLOW_TOOLS_IMAGE")
+}
+
+type BatchConfig struct {
+	// Kubernetes namespace, where BatchInferenceService and BatchInferenceJob will be created
+	Namespace string `json:"namespace"`
+	// Enable batch API/operator
+	Enabled  bool                          `json:"enabled"`
+	// Node pools to run batch jobs
+	NodePools []NodePool `json:"nodePools"`
+	// Kubernetes tolerations for batch jobs
+	Tolerations []corev1.Toleration        `json:"tolerations,omitempty"`
+	// Timeout for full batch process
+	Timeout time.Duration `json:"timeout"`
+	// RClone image that will be used to sync data with object storage
+	RCloneImage string  `json:"rcloneImage"`
+	// ODAHU tools image
+	ToolsImage string  `json:"toolsImage"`
+	// ODAHU tools image
+	ToolsSecret string  `json:"toolsSecret"`
+}
+
+
+func NewDefaultBatchConfig() BatchConfig {
+	return BatchConfig{
+		Namespace: "odahu-flow-batch",
+		Enabled:   true,
+		Timeout: 4 * time.Hour,
+		RCloneImage: "rclone/rclone",
+		ToolsSecret: "odahu-flow-tools-config",
+	}
+}
\ No newline at end of file
diff --git a/packages/operator/pkg/config/config.go b/packages/operator/pkg/config/config.go
index b848f7ee4..3b9b71f52 100644
--- a/packages/operator/pkg/config/config.go
+++ b/packages/operator/pkg/config/config.go
@@ -62,15 +62,25 @@ type Config struct {
 	Training       ModelTrainingConfig   `json:"training"`
 	Packaging      ModelPackagingConfig  `json:"packaging"`
 	Operator       OperatorConfig        `json:"operator"`
+	Batch          BatchConfig           `json:"batch"`
 }
 
 func LoadConfig() (*Config, error) {
 	viper.AutomaticEnv()
 	viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
 
+	if CfgFile == "" {
+		CfgFile = os.Getenv("ODAHU_OPERATOR_CONFIG")
+	}
+
 	if CfgFile != "" {
+		logC.Info(fmt.Sprintf("Config path: %s", CfgFile))
 		viper.SetConfigFile(CfgFile)
 	} else {
+		logC.Info(
+			fmt.Sprintf(
+				"Config path is not configured explicitly. Looking for a config in %s", defaultConfigPathForDev,
+			))
 		viper.AddConfigPath(defaultConfigPathForDev)
 	}
 
@@ -90,6 +100,7 @@ func LoadConfig() (*Config, error) {
 		Training:       NewDefaultModelTrainingConfig(),
 		Packaging:      NewDefaultModelPackagingConfig(),
 		Operator:       NewDefaultOperatorConfig(),
+		Batch:          NewDefaultBatchConfig(),
 	}
 
 	err := viper.Unmarshal(config)
diff --git a/packages/operator/pkg/rclone/azureblob.go b/packages/operator/pkg/rclone/azureblob.go
index e8de3d99c..bd3894281 100644
--- a/packages/operator/pkg/rclone/azureblob.go
+++ b/packages/operator/pkg/rclone/azureblob.go
@@ -17,15 +17,12 @@
 package rclone
 
 import (
-	"errors"
 	"fmt"
 	"github.com/odahu/odahu-flow/packages/operator/api/v1alpha1"
 	_ "github.com/rclone/rclone/backend/azureblob" // s3 specific handlers
 	"github.com/rclone/rclone/fs"
 	"github.com/rclone/rclone/fs/config"
 	"github.com/rclone/rclone/fs/rc"
-	"net/url"
-	"strings"
 )
 
 
@@ -42,23 +39,11 @@ func createAzureBlobConfig(configName string, conn *v1alpha1.ConnectionSpec) (*F
 		return nil, err
 	}
 
-	parsedURI, err := url.Parse(conn.URI)
+	bucketName, pathInsideBucket, err := GetBucketAndPath(conn)
 	if err != nil {
 		log.Error(err, "Parsing data binding URI", "connection uri", conn.URI)
 		return nil, err
 	}
-
-	uriPath := parsedURI.Path
-	uriPath= strings.TrimPrefix(uriPath, "/")
-
-	pathParts := strings.Split(uriPath, "/")
-	if len(pathParts) == 0 {
-		err = errors.New("azure URI must contain at least a bucket name")
-		log.Error(err, "azure URI is empty", "uri", conn.URI, "splitPath", pathParts)
-		return nil, err
-	}
-	bucketName := pathParts[0]
-	pathInsideBucket := "/" + strings.Join(pathParts[1:], "/")
 	return &FileDescription{
 		FsName: fmt.Sprintf("%s:%s", configName, bucketName),
 		Path:   pathInsideBucket,
diff --git a/packages/operator/pkg/rclone/gcs.go b/packages/operator/pkg/rclone/gcs.go
index efd336801..48ac62a03 100644
--- a/packages/operator/pkg/rclone/gcs.go
+++ b/packages/operator/pkg/rclone/gcs.go
@@ -23,7 +23,6 @@ import (
 	"github.com/rclone/rclone/fs"
 	"github.com/rclone/rclone/fs/config"
 	"io/ioutil"
-	"net/url"
 )
 
 func createGcsConfig(configName string, conn *v1alpha1.ConnectionSpec) (*FileDescription, error) {
@@ -55,15 +54,14 @@ func createGcsConfig(configName string, conn *v1alpha1.ConnectionSpec) (*FileDes
 		return nil, err
 	}
 
-	parsedURI, err := url.Parse(conn.URI)
+	bucketName, pathInsideBucket, err := GetBucketAndPath(conn)
 	if err != nil {
 		log.Error(err, "Parsing data binding URI", "connection uri", conn.URI)
-
 		return nil, err
 	}
 
 	return &FileDescription{
-		FsName: fmt.Sprintf("%s:%s", configName, parsedURI.Host),
-		Path:   parsedURI.Path,
+		FsName: fmt.Sprintf("%s:%s", configName, bucketName),
+		Path:   pathInsideBucket,
 	}, nil
 }
diff --git a/packages/operator/pkg/rclone/s3.go b/packages/operator/pkg/rclone/s3.go
index 5d3c50d34..42f281aed 100644
--- a/packages/operator/pkg/rclone/s3.go
+++ b/packages/operator/pkg/rclone/s3.go
@@ -22,7 +22,6 @@ import (
 	_ "github.com/rclone/rclone/backend/s3" // s3 specific handlers
 	"github.com/rclone/rclone/fs"
 	"github.com/rclone/rclone/fs/config"
-	"net/url"
 )
 
 func createS3config(configName string, conn *v1alpha1.ConnectionSpec) (*FileDescription, error) {
@@ -47,15 +46,14 @@ func createS3config(configName string, conn *v1alpha1.ConnectionSpec) (*FileDesc
 		return nil, err
 	}
 
-	parsedURI, err := url.Parse(conn.URI)
+	bucketName, pathInsideBucket, err := GetBucketAndPath(conn)
 	if err != nil {
 		log.Error(err, "Parsing data binding URI", "connection uri", conn.URI)
-
 		return nil, err
 	}
 
 	return &FileDescription{
-		FsName: fmt.Sprintf("%s:%s", configName, parsedURI.Host),
-		Path:   parsedURI.Path,
+		FsName: fmt.Sprintf("%s:%s", configName, bucketName),
+		Path:   pathInsideBucket,
 	}, nil
 }
diff --git a/packages/operator/pkg/rclone/utils.go b/packages/operator/pkg/rclone/utils.go
new file mode 100644
index 000000000..b1dcd1ec4
--- /dev/null
+++ b/packages/operator/pkg/rclone/utils.go
@@ -0,0 +1,39 @@
+package rclone
+
+import (
+	"errors"
+	"fmt"
+	"net/url"
+	"strings"
+	"github.com/odahu/odahu-flow/packages/operator/pkg/apis/connection"
+	"github.com/odahu/odahu-flow/packages/operator/api/v1alpha1"
+)
+
+// GetBucketAndPath return Connection bucket name and path
+func GetBucketAndPath(c *v1alpha1.ConnectionSpec) (string, string, error) {
+	if c.Type == connection.S3Type || c.Type == connection.GcsType {
+		parsedURI, err := url.Parse(c.URI)
+		if err != nil {
+			return "", "", fmt.Errorf("unable to parse conn URI: %s", err)
+		}
+		return parsedURI.Host, parsedURI.Path, nil
+	} else if c.Type == connection.AzureBlobType {
+		parsedURI, err := url.Parse(c.URI)
+		if err != nil {
+			return "", "", fmt.Errorf("unable to parse conn URI: %s", err)
+		}
+
+		uriPath := parsedURI.Path
+		uriPath= strings.TrimPrefix(uriPath, "/")
+
+		pathParts := strings.Split(uriPath, "/")
+		if len(pathParts) == 0 {
+			return "", "", errors.New("azure URI must contain at least a bucket name")
+		}
+		bucketName := pathParts[0]
+		pathInsideBucket := "/" + strings.Join(pathParts[1:], "/")
+		return bucketName, pathInsideBucket, nil
+	} else {
+		return "", "", fmt.Errorf("not available for connection type: %s", c.Type)
+	}
+}


Test patch:
diff --git a/packages/operator/controllers/batchinferencejob_controller_test.go b/packages/operator/controllers/batchinferencejob_controller_test.go
new file mode 100644
index 000000000..bfb10fe72
--- /dev/null
+++ b/packages/operator/controllers/batchinferencejob_controller_test.go
@@ -0,0 +1,322 @@
+//
+//    Copyright 2021 EPAM Systems
+//
+//    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 controllers_test
+
+import (
+	"context"
+	"fmt"
+	odahuflowv1alpha1 "github.com/odahu/odahu-flow/packages/operator/api/v1alpha1"
+	. "github.com/odahu/odahu-flow/packages/operator/controllers"
+	"github.com/odahu/odahu-flow/packages/operator/controllers/mocks"
+	connapitypes "github.com/odahu/odahu-flow/packages/operator/pkg/apis/connection"
+	"github.com/odahu/odahu-flow/packages/operator/pkg/config"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	tektonv1beta1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/pkg/apis/duck/v1beta1"
+	"log"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sync"
+	"testing"
+	"time"
+)
+
+const (
+	requestID = "requestID"
+)
+
+var conn = connapitypes.Connection{
+	ID:     "connection",
+	Spec:   odahuflowv1alpha1.ConnectionSpec{
+		Type:        connapitypes.GcsType,
+		URI:         "gs://bucket/output",
+		Username:    "username",
+		Password:    "pswd",
+		KeyID:       "key",
+		KeySecret:   "secret",
+	},
+	Status: odahuflowv1alpha1.ConnectionStatus{},
+}
+
+func getJob(id string) odahuflowv1alpha1.BatchInferenceJob{
+	return odahuflowv1alpha1.BatchInferenceJob{
+		ObjectMeta: metav1.ObjectMeta{Name: id, Namespace: testNamespace},
+		Spec:       odahuflowv1alpha1.BatchInferenceJobSpec{
+			Command:                 []string{"python"},
+			Args:                    []string{"/opt/app/src.py", "--forecast"},
+			InputPath:               "/input",
+			OutputPath:              "/output",
+			BatchRequestID:          requestID,
+			Image:            "odahu-tools:latest",
+			InputConnection:  "connection",
+			OutputConnection: "connection",
+			ModelConnection:  "connection",
+		},
+		Status:     odahuflowv1alpha1.BatchInferenceJobStatus{},
+	}
+}
+
+// job from getJob(ID string) will be created for each subtest
+var testCases = []struct{
+	// Appearing tektoncd TaskRun with trStatus will be emulated inside subtest to check
+	// reaction of BatchInferenceJob controller
+	trStatus tektonv1beta1.TaskRunStatus
+	// You can configure PodStatus of Pod that correspond to created TaskRun
+	podStatus corev1.PodStatus
+
+	// expectJobStatus is the status that we expect BatchInferenceJob controller set
+	// as reaction to TaskRun with trStatus
+	expectJobStatus odahuflowv1alpha1.BatchInferenceJobStatus
+} {
+	{
+		trStatus: tektonv1beta1.TaskRunStatus{
+			Status: v1beta1.Status{
+				Conditions: v1beta1.Conditions{{
+					Status: corev1.ConditionTrue,
+					Reason: "exit code 0",
+					Message: "Emulated success",
+				}},
+			},
+		},
+		podStatus:        corev1.PodStatus{},
+		expectJobStatus:        odahuflowv1alpha1.BatchInferenceJobStatus{
+			State:   odahuflowv1alpha1.BatchSucceeded,
+			Reason: "exit code 0",
+			Message: "Emulated success",
+		},
+	},
+	{
+		trStatus: tektonv1beta1.TaskRunStatus{
+			Status: v1beta1.Status{
+				Conditions: v1beta1.Conditions{{
+					Status: corev1.ConditionFalse,
+					Reason: "exit code 1",
+					Message: "Emulated fail",
+				}},
+			},
+		},
+		podStatus:        corev1.PodStatus{},
+		expectJobStatus:        odahuflowv1alpha1.BatchInferenceJobStatus{
+			State:   odahuflowv1alpha1.BatchFailed,
+			Reason: "exit code 1",
+			Message: "Emulated fail",
+		},
+	},
+	{
+		trStatus: tektonv1beta1.TaskRunStatus{
+			Status: v1beta1.Status{
+				Conditions: v1beta1.Conditions{{
+					Status: corev1.ConditionUnknown,
+				}},
+			},
+			TaskRunStatusFields: tektonv1beta1.TaskRunStatusFields{
+				PodName: "",
+			},
+		},
+		podStatus:        corev1.PodStatus{},
+		expectJobStatus:        odahuflowv1alpha1.BatchInferenceJobStatus{
+			State:   odahuflowv1alpha1.BatchScheduling,
+		},
+	},
+	{
+		trStatus:         tektonv1beta1.TaskRunStatus{
+			Status: v1beta1.Status{
+				Conditions: v1beta1.Conditions{{
+					Status: corev1.ConditionUnknown,
+				}},
+			},
+			TaskRunStatusFields: tektonv1beta1.TaskRunStatusFields{
+				PodName: "pod",
+			},
+		},
+		podStatus:        corev1.PodStatus{
+			Reason: "Evicted", Message: "Pod was evicted",
+		},
+		expectJobStatus:        odahuflowv1alpha1.BatchInferenceJobStatus{
+			State:   odahuflowv1alpha1.BatchFailed,
+			Reason: "Evicted",
+			Message: "Pod was evicted",
+			PodName: "pod",
+		},
+	},
+	{
+		trStatus:         tektonv1beta1.TaskRunStatus{
+			Status: v1beta1.Status{
+				Conditions: v1beta1.Conditions{{
+					Status: corev1.ConditionUnknown,
+				}},
+			},
+			TaskRunStatusFields: tektonv1beta1.TaskRunStatusFields{
+				PodName: "pod",
+			},
+		},
+		podStatus:        corev1.PodStatus{
+			Phase: corev1.PodPending,
+		},
+		expectJobStatus:        odahuflowv1alpha1.BatchInferenceJobStatus{
+			State:   odahuflowv1alpha1.BatchScheduling,
+			PodName: "pod",
+		},
+	},
+	{
+		trStatus:         tektonv1beta1.TaskRunStatus{
+			Status: v1beta1.Status{
+				Conditions: v1beta1.Conditions{{
+					Status: corev1.ConditionUnknown,
+				}},
+			},
+			TaskRunStatusFields: tektonv1beta1.TaskRunStatusFields{
+				PodName: "pod",
+			},
+		},
+		podStatus:        corev1.PodStatus{
+			Phase: corev1.PodUnknown,
+		},
+		expectJobStatus:        odahuflowv1alpha1.BatchInferenceJobStatus{
+			State:   odahuflowv1alpha1.BatchScheduling,
+			PodName: "pod",
+		},
+	},
+	{
+		trStatus:         tektonv1beta1.TaskRunStatus{
+			Status: v1beta1.Status{
+				Conditions: v1beta1.Conditions{{
+					Status: corev1.ConditionUnknown,
+				}},
+			},
+			TaskRunStatusFields: tektonv1beta1.TaskRunStatusFields{
+				PodName: "pod",
+			},
+		},
+		podStatus:        corev1.PodStatus{
+			Phase: corev1.PodRunning,
+		},
+		expectJobStatus:        odahuflowv1alpha1.BatchInferenceJobStatus{
+			State:   odahuflowv1alpha1.BatchRunning,
+			PodName: "pod",
+		},
+	},
+}
+
+// Tests what job status will be generated depending on the TaskRun and Pod statuses
+func TestJobStatus(t *testing.T) {
+
+	for i, test := range testCases {
+		i := i
+		test := test
+		t.Run(fmt.Sprintf("jobStatus test#%d", i), func(t *testing.T) {
+			as := require.New(t)
+
+			connGetter := mocks.ConnGetter{}
+			connGetter.On("GetConnection", "connection").Return(&conn, nil)
+			podGetter := mocks.PodGetter{}
+			podGetter.On(
+				"GetPod", mock.Anything, "pod", testNamespace,
+				).Return(corev1.Pod{Status: test.podStatus}, nil)
+
+			// Create isolated controller to avoid conflicts with different mocks
+			// in parallel tests
+			mgr, closeController := runBatchController(t, &connGetter, &podGetter)
+			defer closeController()
+			kubeClient := mgr.GetClient()
+
+			// First of all let's create InferenceJob and wait until tekton TaskRun will be
+			// created
+			jobID := fmt.Sprintf("test-job-status-job-%d", i)
+			job := getJob(jobID)
+
+			as.NoError(kubeClient.Create(context.TODO(), &job))
+			tr := &tektonv1beta1.TaskRun{}
+			trKey := types.NamespacedName{Name: job.Name, Namespace: job.Namespace}
+			as.Eventually(
+				func() bool {
+					if err := kubeClient.Get(context.TODO(), trKey, tr); err != nil {
+						return false
+					}
+					log.Print("TaskRun is detected")
+					return true
+				},
+				time.Minute,
+				2*time.Second)
+
+			// Then modify status of TaskRun to emulate TektonCD controller
+			tr.Status = test.trStatus
+			as.NoError(kubeClient.Status().Update(context.TODO(), tr))
+			// And check that we have expected status of BatchInferenceJob depending on this test case
+			//updatedJob := &odahuflowv1alpha1.BatchInferenceJob{}
+			jobKey := types.NamespacedName{Name: job.Name, Namespace: job.Namespace}
+			as.Eventually(
+				func() bool {
+					updatedJob := &odahuflowv1alpha1.BatchInferenceJob{}
+					if err := kubeClient.Get(context.TODO(), jobKey, updatedJob); err != nil {
+						log.Printf("BatchInferenceJob getting error %+v", err)
+						return false
+					}
+					if updatedJob.Status == test.expectJobStatus{
+						log.Print("BatchInferenceJob expected status is detected")
+						return true
+					}
+					log.Printf("BatchInferenceJob actual status %+v, but expected %+v",
+						updatedJob.Status, test.expectJobStatus)
+					return false
+				},
+				time.Minute,
+				2*time.Second)
+
+		})
+
+	}
+}
+
+func runBatchController(t *testing.T, connGetter ConnGetter, podGetter PodGetter) (manager.Manager, func()) {
+	mgr, err := manager.New(cfg, manager.Options{MetricsBindAddress: "0"})
+	req := require.New(t)
+	req.NoError(err)
+
+	stopChan := make(chan struct{})
+	mgrStopped := &sync.WaitGroup{}
+	mgrStopped.Add(1)
+
+	batchConfig := config.NewDefaultConfig().Batch
+	batchConfig.Namespace = testNamespace
+
+	opts := BatchInferenceJobReconcilerOptions{
+		Client: mgr.GetClient(),
+		Schema: mgr.GetScheme(),
+		ConnGetter:        connGetter,
+		PodGetter:         podGetter,
+		Cfg: batchConfig,
+	}
+
+	req.NoError(NewBatchInferenceJobReconciler(opts).SetupWithManager(mgr))
+
+	go func() {
+		defer mgrStopped.Done()
+		req.NoError(mgr.Start(stopChan))
+		log.Printf("Controller %p is started", mgr)
+	}()
+
+	return mgr, func() {
+		close(stopChan)
+		mgrStopped.Wait()
+		log.Printf("Controller %p is stopped", mgr)
+	}
+}
+