diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index a275f540c..d71b046b2 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -89,6 +89,9 @@ jobs:
           # SSH bastion
           - ssh-bastion
 
+          # img-export for persistent VM instances snapshot
+          - img-exporter
+
           # Laboratory environments
           - novnc
           - tigervnc
@@ -156,6 +159,10 @@ jobs:
           - component: ssh-bastion
             context: ./operators/build/ssh-bastion
 
+          # img-exporter image for InstanceSnapshot
+          - component: img-exporter
+            context: ./operators/build/img-exporter
+
     steps:
       - name: Checkout
         uses: actions/checkout@v2
diff --git a/deploy/crownlabs/templates/clusterroles.yaml b/deploy/crownlabs/templates/clusterroles.yaml
index 00a299378..70414ae00 100644
--- a/deploy/crownlabs/templates/clusterroles.yaml
+++ b/deploy/crownlabs/templates/clusterroles.yaml
@@ -38,6 +38,47 @@ rules:
       - patch
       - delete
       - deletecollection
+      
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: crownlabs-view-instance-snapshots
+  labels:
+    {{- include "crownlabs.labels" . | nindent 4 }}
+rules:
+  - apiGroups:
+      - crownlabs.polito.it
+    resources:
+      - instancesnapshots
+      - instancesnapshots/status
+    verbs:
+      - get
+      - list
+      - watch
+
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: crownlabs-manage-instance-snapshots
+  labels:
+    {{- include "crownlabs.labels" . | nindent 4 }}
+rules:
+  - apiGroups:
+      - crownlabs.polito.it
+    resources:
+      - instancesnapshots
+      - instancesnapshots/status
+    verbs:
+      - get
+      - list
+      - watch
+      - create
+      - update
+      - patch
+      - delete
+      - deletecollection
 
 ---
 apiVersion: rbac.authorization.k8s.io/v1
diff --git a/deploy/crownlabs/values.yaml b/deploy/crownlabs/values.yaml
index 00d7d7c5c..fe9b7ea74 100644
--- a/deploy/crownlabs/values.yaml
+++ b/deploy/crownlabs/values.yaml
@@ -72,6 +72,13 @@ instance-operator:
       novncImage: crownlabs/novnc
       filebrowserImage: filebrowser/filebrowser
       filebrowserImageTag: latest
+    containerVmSnapshots:
+      kanikoImage: gcr.io/kaniko-project/executor
+      exportImage: "crownlabs/img-exporter"
+      exportImageTag: ""
+    privateContainerRegistry:
+      url: registry.crownlabs.example.com
+      secretName: registry-credentials      
 
 tenant-operator:
   replicaCount: 1
diff --git a/operators/api/v1alpha2/instance_types.go b/operators/api/v1alpha2/instance_types.go
index 315935863..192650a43 100644
--- a/operators/api/v1alpha2/instance_types.go
+++ b/operators/api/v1alpha2/instance_types.go
@@ -31,6 +31,7 @@ type InstanceSpec struct {
 	Tenant GenericRef `json:"tenant.crownlabs.polito.it/TenantRef"`
 
 	// +kubebuilder:default=true
+	// +kubebuilder:validation:Optional
 
 	// Whether the current instance is running or not. This field is meaningful
 	// only in case the Instance refers to persistent environments, and it allows
@@ -39,7 +40,7 @@ type InstanceSpec struct {
 	// attaching it to the same disk used previously. The flag, on the other hand,
 	// is silently ignored in case of non-persistent environments, as the state
 	// cannot be preserved among reboots.
-	Running bool `json:"running,omitempty"`
+	Running bool `json:"running"`
 }
 
 // InstanceStatus reflects the most recently observed status of the Instance.
diff --git a/operators/api/v1alpha2/instancesnapshot_types.go b/operators/api/v1alpha2/instancesnapshot_types.go
new file mode 100644
index 000000000..8fb2352aa
--- /dev/null
+++ b/operators/api/v1alpha2/instancesnapshot_types.go
@@ -0,0 +1,94 @@
+/*
+
+
+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 v1alpha2
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// 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.
+
+// SnapshotStatus is an enumeration representing the current state of the InstanceSnapshot.
+type SnapshotStatus string
+
+const (
+	// Pending -> The snapshot resource has been observed and the
+	// process is waiting to be started.
+	Pending SnapshotStatus = "Pending"
+	// Processing -> The process of creation of the snapshot started.
+	Processing SnapshotStatus = "Processing"
+	// Completed -> The snapshot of the instance has been created.
+	Completed SnapshotStatus = "Completed"
+	// Failed -> The process of creation of the snapshot failed.
+	Failed SnapshotStatus = "Failed"
+)
+
+// InstanceSnapshotSpec defines the desired state of InstanceSnapshot.
+type InstanceSnapshotSpec struct {
+	// INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
+	// Important: Run "make" to regenerate code after modifying this file
+
+	// Instance is the reference to the persistent VM instance to be snapshotted.
+	// The instance should not be running, otherwise it won't be possible to
+	// steal the volume and extract its content.
+	Instance GenericRef `json:"instanceRef"`
+
+	// Environment represents the reference to the environment to be snapshotted, in case more are
+	// associated with the same Instance. If not specified, the first available environment is considered.
+	Environment GenericRef `json:"environmentRef,omitempty"`
+
+	// +kubebuilder:validation:MinLength=1
+
+	// ImageName is the name of the image to pushed in the docker registry.
+	ImageName string `json:"imageName"`
+}
+
+// InstanceSnapshotStatus defines the observed state of InstanceSnapshot.
+type InstanceSnapshotStatus struct {
+	// Phase represents the current state of the Instance Snapshot.
+	Phase SnapshotStatus `json:"phase"`
+}
+
+// +kubebuilder:object:root=true
+// +kubebuilder:subresource:status
+// +kubebuilder:resource:shortName="isnap"
+// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase`
+// +kubebuilder:printcolumn:name="ImageName",type=string,JSONPath=`.spec.imageName`
+// +kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`
+
+// InstanceSnapshot is the Schema for the instancesnapshots API.
+type InstanceSnapshot struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec   InstanceSnapshotSpec   `json:"spec,omitempty"`
+	Status InstanceSnapshotStatus `json:"status,omitempty"`
+}
+
+// +kubebuilder:object:root=true
+
+// InstanceSnapshotList contains a list of InstanceSnapshot.
+type InstanceSnapshotList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []InstanceSnapshot `json:"items"`
+}
+
+func init() {
+	SchemeBuilder.Register(&InstanceSnapshot{}, &InstanceSnapshotList{})
+}
diff --git a/operators/api/v1alpha2/zz_generated.deepcopy.go b/operators/api/v1alpha2/zz_generated.deepcopy.go
index 4bd6b0060..7d698c129 100644
--- a/operators/api/v1alpha2/zz_generated.deepcopy.go
+++ b/operators/api/v1alpha2/zz_generated.deepcopy.go
@@ -130,6 +130,97 @@ func (in *InstanceList) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *InstanceSnapshot) DeepCopyInto(out *InstanceSnapshot) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	out.Spec = in.Spec
+	out.Status = in.Status
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InstanceSnapshot.
+func (in *InstanceSnapshot) DeepCopy() *InstanceSnapshot {
+	if in == nil {
+		return nil
+	}
+	out := new(InstanceSnapshot)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *InstanceSnapshot) 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 *InstanceSnapshotList) DeepCopyInto(out *InstanceSnapshotList) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ListMeta.DeepCopyInto(&out.ListMeta)
+	if in.Items != nil {
+		in, out := &in.Items, &out.Items
+		*out = make([]InstanceSnapshot, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InstanceSnapshotList.
+func (in *InstanceSnapshotList) DeepCopy() *InstanceSnapshotList {
+	if in == nil {
+		return nil
+	}
+	out := new(InstanceSnapshotList)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *InstanceSnapshotList) 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 *InstanceSnapshotSpec) DeepCopyInto(out *InstanceSnapshotSpec) {
+	*out = *in
+	out.Instance = in.Instance
+	out.Environment = in.Environment
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InstanceSnapshotSpec.
+func (in *InstanceSnapshotSpec) DeepCopy() *InstanceSnapshotSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(InstanceSnapshotSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *InstanceSnapshotStatus) DeepCopyInto(out *InstanceSnapshotStatus) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InstanceSnapshotStatus.
+func (in *InstanceSnapshotStatus) DeepCopy() *InstanceSnapshotStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(InstanceSnapshotStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *InstanceSpec) DeepCopyInto(out *InstanceSpec) {
 	*out = *in
diff --git a/operators/build/img-exporter/Dockerfile b/operators/build/img-exporter/Dockerfile
new file mode 100644
index 000000000..899d098a4
--- /dev/null
+++ b/operators/build/img-exporter/Dockerfile
@@ -0,0 +1,10 @@
+FROM alpine:3.13.4
+
+# Install the qemu-img useful to convert the image
+RUN apk add --update --no-cache qemu-img
+
+# Copy the entrypoint script
+COPY exporter.sh /
+
+# Run the entrypoint which converts the image and creates the dockerfile
+CMD ["/exporter.sh"]
diff --git a/operators/build/img-exporter/exporter.sh b/operators/build/img-exporter/exporter.sh
new file mode 100755
index 000000000..010bd450b
--- /dev/null
+++ b/operators/build/img-exporter/exporter.sh
@@ -0,0 +1,63 @@
+#!/bin/sh
+IMG_DIR=/data
+OUT_DIR=/img-tmp
+IMG_NAME=disk.img
+OUT_IMAGE=vm-snapshot.qcow2
+PROG_NAME=$0
+
+usage(){
+	echo "Usage: $PROG_NAME [-options]"
+	echo "  -d, --img-dir        Specify the working directory [DEFAULT=$IMG_DIR]"
+	echo "  -o, --out-dir        Specify the output directory  [DEFAULT=$OUT_DIR]"
+	echo "  -n, --img-name       Specify the name of the image [DEFAULT=$OUT_IMAGE]"
+	exit 1
+}
+
+parse_args(){
+	while [ "${1:-}" != "" ]; do
+		case "$1" in
+			"-d" | "--img-dir")
+				shift
+				IMG_DIR=$1
+				;;
+			"-o" | "--out-dir")
+				shift
+				OUT_DIR=$1
+				;;
+			"-n" | "--img-name")
+				shift
+				IMG_NAME=$1
+				;;
+			*)
+				usage
+				;;
+		esac
+		shift
+	done
+}
+
+export_img(){
+	echo "Converting the image..."
+
+	# Check if output directory exists, if not create it
+	# and try with the conversion of the image.
+	mkdir -p "$OUT_DIR"
+	qemu-img convert -c -f raw -O qcow2  "${IMG_DIR}/${IMG_NAME}" "${OUT_DIR}/${OUT_IMAGE}"
+
+	echo "Creating Dockerfile..."
+	# Create the Dockerfile.
+	cat <<EOF > "${OUT_DIR}/Dockerfile"
+FROM scratch
+ADD ${OUT_IMAGE} /disk/
+EOF
+}
+
+parse_args "$@"
+
+if export_img;
+then
+	echo "${IMG_DIR}/${IMG_NAME} successully converted"
+else
+	echo "Conversion unsuccessfully completed"
+	exit 1
+fi
\ No newline at end of file
diff --git a/operators/cmd/instance-operator/main.go b/operators/cmd/instance-operator/main.go
index c88c32254..ed7f79792 100644
--- a/operators/cmd/instance-operator/main.go
+++ b/operators/cmd/instance-operator/main.go
@@ -34,6 +34,7 @@ import (
 	crownlabsv1alpha1 "github.com/netgroup-polito/CrownLabs/operators/api/v1alpha1"
 	crownlabsv1alpha2 "github.com/netgroup-polito/CrownLabs/operators/api/v1alpha2"
 	instance_controller "github.com/netgroup-polito/CrownLabs/operators/pkg/instance-controller"
+	instancesnapshot_controller "github.com/netgroup-polito/CrownLabs/operators/pkg/instancesnapshot-controller"
 	// +kubebuilder:scaffold:imports
 )
 
@@ -68,6 +69,10 @@ func main() {
 	var containerEnvVncImg string
 	var containerEnvWebsockifyImg string
 	var containerEnvNovncImg string
+	var vmRegistry string
+	var vmRegistrySecret string
+	var containerImgExport string
+	var containerKaniko string
 	var containerEnvFileBrowserImg string
 	var containerEnvFileBrowserImgTag string
 
@@ -87,6 +92,10 @@ func main() {
 	flag.StringVar(&containerEnvVncImg, "container-env-vnc-img", "crownlabs/tigervnc", "The image name for the vnc image (sidecar for graphical container environment)")
 	flag.StringVar(&containerEnvWebsockifyImg, "container-env-websockify-img", "crownlabs/websockify", "The image name for the websockify image (sidecar for graphical container environment)")
 	flag.StringVar(&containerEnvNovncImg, "container-env-novnc-img", "crownlabs/novnc", "The image name for the novnc image (sidecar for graphical container environment)")
+	flag.StringVar(&vmRegistry, "vm-registry", "", "The registry where VMs should be uploaded")
+	flag.StringVar(&vmRegistrySecret, "vm-registry-secret", "", "The name of the secret for the VM registry")
+	flag.StringVar(&containerImgExport, "container-export-img", "crownlabs/img-exporter", "The image for the img-exporter (container in charge of exporting the disk of a persistent vm)")
+	flag.StringVar(&containerKaniko, "container-kaniko-img", "gcr.io/kaniko-project/executor", "The image for the Kaniko container to be deployed")
 	flag.StringVar(&containerEnvFileBrowserImg, "container-env-filebrowser-img", "filebrowser/filebrowser", "The image name for the filebrowser image (sidecar for gui-based file manager)")
 	flag.StringVar(&containerEnvFileBrowserImgTag, "container-env-filebrowser-img-tag", "latest", "The tag for the FileBrowser container (the gui-based file manager)")
 	klog.InitFlags(nil)
@@ -129,6 +138,21 @@ func main() {
 		klog.Fatal(err, "unable to create controller", "controller", "Instance")
 	}
 
+	if err = (&instancesnapshot_controller.InstanceSnapshotReconciler{
+		Client:             mgr.GetClient(),
+		Scheme:             mgr.GetScheme(),
+		EventsRecorder:     mgr.GetEventRecorderFor("instance-snapshot"),
+		NamespaceWhitelist: metav1.LabelSelector{MatchLabels: whiteListMap, MatchExpressions: []metav1.LabelSelectorRequirement{}},
+		VMRegistry:         vmRegistry,
+		RegistrySecretName: vmRegistrySecret,
+		ContainersSnapshot: instancesnapshot_controller.ContainersSnapshotOpts{
+			ContainerKaniko:    containerKaniko,
+			ContainerImgExport: containerImgExport,
+		},
+	}).SetupWithManager(mgr); err != nil {
+		klog.Fatal(err, "unable to create controller", "controller", "InstanceSnapshot")
+	}
+
 	// +kubebuilder:scaffold:builder
 	// Add readiness probe
 	err = mgr.AddReadyzCheck("ready-ping", healthz.Ping)
diff --git a/operators/deploy/crds/crownlabs.polito.it_instancesnapshots.yaml b/operators/deploy/crds/crownlabs.polito.it_instancesnapshots.yaml
new file mode 100644
index 000000000..0e025f240
--- /dev/null
+++ b/operators/deploy/crds/crownlabs.polito.it_instancesnapshots.yaml
@@ -0,0 +1,109 @@
+
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  annotations:
+    controller-gen.kubebuilder.io/version: v0.4.1
+  creationTimestamp: null
+  name: instancesnapshots.crownlabs.polito.it
+spec:
+  group: crownlabs.polito.it
+  names:
+    kind: InstanceSnapshot
+    listKind: InstanceSnapshotList
+    plural: instancesnapshots
+    shortNames:
+    - isnap
+    singular: instancesnapshot
+  scope: Namespaced
+  versions:
+  - additionalPrinterColumns:
+    - jsonPath: .status.phase
+      name: Phase
+      type: string
+    - jsonPath: .spec.imageName
+      name: ImageName
+      type: string
+    - jsonPath: .metadata.creationTimestamp
+      name: Age
+      type: date
+    name: v1alpha2
+    schema:
+      openAPIV3Schema:
+        description: InstanceSnapshot is the Schema for the instancesnapshots 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: InstanceSnapshotSpec defines the desired state of InstanceSnapshot.
+            properties:
+              environmentRef:
+                description: Environment represents the reference to the environment
+                  to be snapshotted, in case more are associated with the same Instance.
+                  If not specified, the first available environment is considered.
+                properties:
+                  name:
+                    description: The name of the resource to be referenced.
+                    type: string
+                  namespace:
+                    description: The namespace containing the resource to be referenced.
+                      It should be left empty in case of cluster-wide resources.
+                    type: string
+                required:
+                - name
+                type: object
+              imageName:
+                description: ImageName is the name of the image to pushed in the docker
+                  registry.
+                minLength: 1
+                type: string
+              instanceRef:
+                description: Instance is the reference to the persistent VM instance
+                  to be snapshotted. The instance should not be running, otherwise
+                  it won't be possible to steal the volume and extract its content.
+                properties:
+                  name:
+                    description: The name of the resource to be referenced.
+                    type: string
+                  namespace:
+                    description: The namespace containing the resource to be referenced.
+                      It should be left empty in case of cluster-wide resources.
+                    type: string
+                required:
+                - name
+                type: object
+            required:
+            - imageName
+            - instanceRef
+            type: object
+          status:
+            description: InstanceSnapshotStatus defines the observed state of InstanceSnapshot.
+            properties:
+              phase:
+                description: Phase represents the current state of the Instance Snapshot.
+                type: string
+            required:
+            - phase
+            type: object
+        type: object
+    served: true
+    storage: true
+    subresources:
+      status: {}
+status:
+  acceptedNames:
+    kind: ""
+    plural: ""
+  conditions: []
+  storedVersions: []
diff --git a/operators/deploy/instance-operator/templates/_helpers.tpl b/operators/deploy/instance-operator/templates/_helpers.tpl
index bb782e3cc..6b2ac8f2d 100644
--- a/operators/deploy/instance-operator/templates/_helpers.tpl
+++ b/operators/deploy/instance-operator/templates/_helpers.tpl
@@ -73,3 +73,10 @@ The tag to be used for sidecar containers images
 {{- define "instance-operator.containerEnvironmentSidecarsTag" -}}
 {{- .Values.configurations.containerEnvironmentOptions.tag | default ( include "instance-operator.version" . ) }}
 {{- end }}
+
+{{/*
+The tag to be used for image exporter container for VM snapshots
+*/}}
+{{- define "instance-operator.containerExportImageTag" -}}
+{{- .Values.configurations.containerVmSnapshots.exportImageTag | default ( include "instance-operator.version" . ) }}
+{{- end }}
\ No newline at end of file
diff --git a/operators/deploy/instance-operator/templates/clusterrole.yaml b/operators/deploy/instance-operator/templates/clusterrole.yaml
index ec03dd50a..514273971 100644
--- a/operators/deploy/instance-operator/templates/clusterrole.yaml
+++ b/operators/deploy/instance-operator/templates/clusterrole.yaml
@@ -9,6 +9,10 @@ rules:
   resources: ["instances", "instances/status"]
   verbs: ["get","list","watch","create","update","patch","delete", "deleteCollection"]
 
+- apiGroups: ["crownlabs.polito.it"]
+  resources: ["instancesnapshots", "instancesnapshots/status"]
+  verbs: ["get","list","watch","create","update","patch"]
+
 - apiGroups: ["crownlabs.polito.it"]
   resources: ["templates", "tenants"]
   verbs: ["get","list","watch"]
@@ -25,6 +29,10 @@ rules:
   resources: ["deployments"]
   verbs: ["get","list","watch","create","patch","update"]
 
+- apiGroups: ["batch"]
+  resources: ["jobs", "jobs/status"]
+  verbs: ["get", "list", "watch", "create", "update", "patch"]
+
 - apiGroups: ["networking.k8s.io"]
   resources: ["ingresses"]
   verbs: ["get","list","watch","create","patch","update"]
diff --git a/operators/deploy/instance-operator/templates/deployment.yaml b/operators/deploy/instance-operator/templates/deployment.yaml
index f3c4afed9..ada3cffbe 100644
--- a/operators/deploy/instance-operator/templates/deployment.yaml
+++ b/operators/deploy/instance-operator/templates/deployment.yaml
@@ -47,6 +47,10 @@ spec:
             - "--container-env-vnc-img={{ .Values.configurations.containerEnvironmentOptions.vncImage }}"
             - "--container-env-websockify-img={{ .Values.configurations.containerEnvironmentOptions.websockifyImage }}"
             - "--container-env-novnc-img={{ .Values.configurations.containerEnvironmentOptions.novncImage }}"
+            - "--vm-registry={{ .Values.configurations.privateContainerRegistry.url }}"
+            - "--vm-registry-secret={{ .Values.configurations.privateContainerRegistry.secretName }}"
+            - "--container-export-img={{ .Values.configurations.containerVmSnapshots.exportImage }}:{{ include "instance-operator.containerExportImageTag" . }}"
+            - "--container-kaniko-img={{ .Values.configurations.containerVmSnapshots.kanikoImage }}"
             - "--container-env-filebrowser-img={{ .Values.configurations.containerEnvironmentOptions.filebrowserImage }}"
             - "--container-env-filebrowser-img-tag={{ .Values.configurations.containerEnvironmentOptions.filebrowserImageTag }}"
           ports:
diff --git a/operators/deploy/instance-operator/values.yaml b/operators/deploy/instance-operator/values.yaml
index ab53ef6ce..161ebedb2 100644
--- a/operators/deploy/instance-operator/values.yaml
+++ b/operators/deploy/instance-operator/values.yaml
@@ -23,6 +23,13 @@ configurations:
     novncImage: crownlabs/novnc
     filebrowserImage: filebrowser/filebrowser
     filebrowserImageTag: latest
+  containerVmSnapshots:
+    kanikoImage: gcr.io/kaniko-project/executor:latest
+    exportImage: "crownlabs/img-exporter"
+    exportImageTag: ""
+  privateContainerRegistry:
+    url: registry.crownlabs.example.com
+    secretName: registry-credentials
 
 image:
   repository: crownlabs/instance-operator
diff --git a/operators/pkg/instance-controller/controller.go b/operators/pkg/instance-controller/controller.go
index a058888be..a47109e49 100644
--- a/operators/pkg/instance-controller/controller.go
+++ b/operators/pkg/instance-controller/controller.go
@@ -22,7 +22,6 @@ import (
 	"time"
 
 	appsv1 "k8s.io/api/apps/v1"
-	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -35,7 +34,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
 	crownlabsv1alpha2 "github.com/netgroup-polito/CrownLabs/operators/api/v1alpha2"
-	instance_creation "github.com/netgroup-polito/CrownLabs/operators/pkg/instance-creation"
+	"github.com/netgroup-polito/CrownLabs/operators/pkg/utils"
 )
 
 // ContainerEnvOpts contains images name and tag for container environment.
@@ -82,24 +81,16 @@ func (r *InstanceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		// reconcile was triggered by a delete request
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
-	ns := v1.Namespace{}
-	namespaceName := types.NamespacedName{
-		Name:      instance.Namespace,
-		Namespace: "",
-	}
 
-	// It performs reconciliation only if the Instance belongs to whitelisted namespaces
-	// by checking the existence of keys in the instance namespace
-	if err := r.Get(ctx, namespaceName, &ns); err == nil {
-		if !instance_creation.CheckLabels(&ns, r.NamespaceWhitelist.MatchLabels) {
-			klog.Info("Namespace " + req.Namespace + " does not meet the selector labels")
-			return ctrl.Result{}, nil
+	// Check the selector label, in order to know whether to perform or not reconciliation.
+	if proceed, err := utils.CheckSelectorLabel(ctx, r.Client, instance.Namespace, r.NamespaceWhitelist.MatchLabels); !proceed {
+		// If there was an error while checking, show the error and try again.
+		if err != nil {
+			klog.Error(err)
+			return ctrl.Result{}, err
 		}
-	} else {
-		klog.Error("Unable to get Instance namespace")
-		klog.Error(err)
+		return ctrl.Result{}, nil
 	}
-	klog.Info("Namespace " + req.Namespace + " met the selector labels")
 
 	// check if the Template exists
 	templateName := types.NamespacedName{
diff --git a/operators/pkg/instance-creation/creation.go b/operators/pkg/instance-creation/creation.go
index 9e82a9d01..4be25a11d 100644
--- a/operators/pkg/instance-creation/creation.go
+++ b/operators/pkg/instance-creation/creation.go
@@ -292,14 +292,3 @@ func CreateOrUpdate(ctx context.Context, c client.Client, object interface{}) er
 
 	return nil
 }
-
-// CheckLabels verifies whether a namespace is characterized by a set of
-// required labels.
-func CheckLabels(ns *corev1.Namespace, matchLabels map[string]string) bool {
-	for key, value := range matchLabels {
-		if v1, ok := ns.Labels[key]; !ok || v1 != value {
-			return false
-		}
-	}
-	return true
-}
diff --git a/operators/pkg/instance-creation/creation_test.go b/operators/pkg/instance-creation/creation_test.go
index a29f779d5..a0305eddf 100644
--- a/operators/pkg/instance-creation/creation_test.go
+++ b/operators/pkg/instance-creation/creation_test.go
@@ -4,14 +4,16 @@ import (
 	"strconv"
 	"testing"
 
-	"github.com/netgroup-polito/CrownLabs/operators/api/v1alpha2"
-
-	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	virtv1 "kubevirt.io/client-go/api/v1"
 	cdiv1 "kubevirt.io/containerized-data-importer/pkg/apis/core/v1alpha1"
+
+	"github.com/netgroup-polito/CrownLabs/operators/api/v1alpha2"
+	"github.com/netgroup-polito/CrownLabs/operators/pkg/utils"
+
+	"github.com/stretchr/testify/assert"
 )
 
 var ns1 = v1.Namespace{
@@ -43,8 +45,8 @@ var labels = map[string]string{
 }
 
 func TestWhitelist(t *testing.T) {
-	c1 := CheckLabels(&ns1, labels)
-	c2 := CheckLabels(&ns2, labels)
+	c1 := utils.CheckLabels(&ns1, labels)
+	c2 := utils.CheckLabels(&ns2, labels)
 	assert.Equal(t, c1, true, "The two label set should be identical and return true.")
 	assert.Equal(t, c2, false, "The two labels set should be different and return false.")
 }
@@ -168,10 +170,10 @@ func TestCheckLabels(t *testing.T) {
 			Labels: map[string]string{},
 		},
 	}
-	assert.Equal(t, CheckLabels(&ns, labels), true)
-	assert.Equal(t, CheckLabels(&ns1, labels), false)
-	assert.Equal(t, CheckLabels(&ns2, labels), false)
-	assert.Equal(t, CheckLabels(&ns3, labels), false)
+	assert.Equal(t, utils.CheckLabels(&ns, labels), true)
+	assert.Equal(t, utils.CheckLabels(&ns1, labels), false)
+	assert.Equal(t, utils.CheckLabels(&ns2, labels), false)
+	assert.Equal(t, utils.CheckLabels(&ns3, labels), false)
 }
 
 func TestComputeCPULimits(t *testing.T) {
diff --git a/operators/pkg/instancesnapshot-controller/instancesnapshot_controller.go b/operators/pkg/instancesnapshot-controller/instancesnapshot_controller.go
new file mode 100644
index 000000000..c77fd9252
--- /dev/null
+++ b/operators/pkg/instancesnapshot-controller/instancesnapshot_controller.go
@@ -0,0 +1,152 @@
+/*
+
+
+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 instancesnapshot_controller groups the functionalities related to the creation of a persistent VM snapshot.
+package instancesnapshot_controller
+
+import (
+	"context"
+	"fmt"
+
+	batch "k8s.io/api/batch/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/klog/v2"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+
+	crownlabsv1alpha2 "github.com/netgroup-polito/CrownLabs/operators/api/v1alpha2"
+	"github.com/netgroup-polito/CrownLabs/operators/pkg/utils"
+)
+
+// ContainersSnapshotOpts contains image names and tags of the containers needed for the VM snapshot.
+type ContainersSnapshotOpts struct {
+	ContainerKaniko    string
+	ContainerImgExport string
+}
+
+// InstanceSnapshotReconciler reconciles a InstanceSnapshot object.
+type InstanceSnapshotReconciler struct {
+	client.Client
+	EventsRecorder     record.EventRecorder
+	Scheme             *runtime.Scheme
+	NamespaceWhitelist metav1.LabelSelector
+	VMRegistry         string
+	RegistrySecretName string
+	ContainersSnapshot ContainersSnapshotOpts
+
+	// This function, if configured, is deferred at the beginning of the Reconcile.
+	// Specifically, it is meant to be set to GinkgoRecover during the tests,
+	// in order to lead to a controlled failure in case the Reconcile panics.
+	ReconcileDeferHook func()
+}
+
+// Reconcile reconciles the status of the InstanceSnapshot resource.
+func (r *InstanceSnapshotReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	if r.ReconcileDeferHook != nil {
+		defer r.ReconcileDeferHook()
+	}
+
+	isnap := &crownlabsv1alpha2.InstanceSnapshot{}
+
+	if err := r.Get(ctx, req.NamespacedName, isnap); client.IgnoreNotFound(err) != nil {
+		klog.Errorf("Error when getting InstanceSnapshot %s before starting reconcile -> %s", isnap.Name, err)
+		return ctrl.Result{}, err
+	} else if err != nil {
+		klog.Infof("InstanceSnapshot %s already deleted", req.NamespacedName.Name)
+		return ctrl.Result{}, nil
+	}
+
+	// Check the selector label, in order to know whether to perform or not reconciliation.
+	if proceed, err := utils.CheckSelectorLabel(ctx, r.Client, isnap.Namespace, r.NamespaceWhitelist.MatchLabels); !proceed {
+		// If there was an error while checking, show the error and try again.
+		if err != nil {
+			klog.Error(err)
+			return ctrl.Result{}, err
+		}
+		return ctrl.Result{}, nil
+	}
+
+	klog.Infof("Start InstanceSnapshot reconciliation of %s in %s namespace", isnap.Name, isnap.Namespace)
+
+	// Check the current status of the InstanceSnapshot by checking
+	// the state of its assigned job.
+	jobName := types.NamespacedName{
+		Namespace: isnap.Namespace,
+		Name:      isnap.Name,
+	}
+	found := &batch.Job{}
+	err := r.Get(ctx, jobName, found)
+
+	switch {
+	case err != nil && errors.IsNotFound(err):
+		if retry, err1 := r.CreateSnapshottingJob(ctx, isnap); err1 != nil {
+			klog.Error(err1)
+			// Check if we need to try again with the creation of the job
+			if retry {
+				return ctrl.Result{}, err1
+			}
+			// Since we don't have to retry, validation failed
+			// Add the event and stop reconciliation since the request is not valid.
+			r.EventsRecorder.Event(isnap, "Warning", "ValidationError", fmt.Sprintf("%s", err1))
+			return ctrl.Result{}, nil
+		}
+		// Job successfully created
+		r.EventsRecorder.Event(isnap, "Normal", "Creation", fmt.Sprintf("Job %s for snapshot creation started", isnap.Name))
+	case err != nil:
+		klog.Errorf("Unable to retrieve the job of InstanceSnapshot %s -> %s", isnap.Name, err)
+		return ctrl.Result{}, err
+	default:
+		// Check the current state of the job and log according to its state
+		jstatus, err1 := r.HandleExistingJob(ctx, isnap, found)
+		switch {
+		case err1 != nil:
+
+			klog.Error(err1)
+			return ctrl.Result{}, err1
+		case jstatus == batch.JobComplete:
+
+			successMessage := fmt.Sprintf("Image %s created and uploaded", isnap.Spec.ImageName)
+			// If we are able to retrieve the execution time, report it
+			if found.Status.StartTime != nil && found.Status.CompletionTime != nil {
+				extime := found.Status.CompletionTime.Sub(found.Status.StartTime.Time)
+				successMessage = fmt.Sprintf("%s in %s", successMessage, extime)
+			}
+			klog.Info(successMessage)
+			r.EventsRecorder.Event(isnap, "Normal", "Created", successMessage)
+		case jstatus == batch.JobFailed:
+
+			klog.Infof("Image %s could not be created", isnap.Spec.ImageName)
+			r.EventsRecorder.Event(isnap, "Warning", "CreationFailed", "The creation job failed")
+		}
+	}
+
+	return ctrl.Result{}, nil
+}
+
+// SetupWithManager registers a new controller for InstanceSnapshot resources.
+func (r *InstanceSnapshotReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		// The generation changed predicate allow to avoid updates on the status changes of the InstanceSnapshot
+		For(&crownlabsv1alpha2.InstanceSnapshot{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
+		Owns(&batch.Job{}).
+		Complete(r)
+}
diff --git a/operators/pkg/instancesnapshot-controller/instancesnapshot_controller_suite_test.go b/operators/pkg/instancesnapshot-controller/instancesnapshot_controller_suite_test.go
new file mode 100644
index 000000000..31b73da21
--- /dev/null
+++ b/operators/pkg/instancesnapshot-controller/instancesnapshot_controller_suite_test.go
@@ -0,0 +1,100 @@
+package instancesnapshot_controller_test
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+	"sigs.k8s.io/controller-runtime/pkg/envtest/printer"
+
+	crownlabsv1alpha2 "github.com/netgroup-polito/CrownLabs/operators/api/v1alpha2"
+	instancesnapshot_controller "github.com/netgroup-polito/CrownLabs/operators/pkg/instancesnapshot-controller"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	gomegaTypes "github.com/onsi/gomega/types"
+)
+
+var cfg *rest.Config
+var k8sClient client.Client
+var testEnv *envtest.Environment
+
+func TestAPIs(t *testing.T) {
+	RegisterFailHandler(Fail)
+
+	RunSpecsWithDefaultAndCustomReporters(t,
+		"Controller Suite",
+		[]Reporter{printer.NewlineReporter{}})
+}
+
+var _ = BeforeSuite(func(done Done) {
+	By("bootstrapping test environment")
+	testEnv = &envtest.Environment{
+		CRDDirectoryPaths: []string{filepath.Join("..", "..", "deploy", "crds"),
+			filepath.Join("..", "..", "tests", "crds")},
+	}
+	var err error
+	cfg, err = testEnv.Start()
+	Expect(err).ToNot(HaveOccurred())
+	Expect(cfg).ToNot(BeNil())
+
+	err = crownlabsv1alpha2.AddToScheme(scheme.Scheme)
+	Expect(err).NotTo(HaveOccurred())
+
+	// +kubebuilder:scaffold:scheme
+	k8sManager, err := ctrl.NewManager(cfg, ctrl.Options{
+		Scheme:             scheme.Scheme,
+		MetricsBindAddress: "0",
+	})
+	Expect(err).ToNot(HaveOccurred())
+
+	// Generate whitelist map for InstanceSnapshot controller reconciliation
+	whiteListMap := map[string]string{
+		"test-suite": "true",
+	}
+
+	err = (&instancesnapshot_controller.InstanceSnapshotReconciler{
+		Client:             k8sManager.GetClient(),
+		Scheme:             k8sManager.GetScheme(),
+		EventsRecorder:     k8sManager.GetEventRecorderFor("instance-snapshot"),
+		NamespaceWhitelist: metav1.LabelSelector{MatchLabels: whiteListMap, MatchExpressions: []metav1.LabelSelectorRequirement{}},
+		VMRegistry:         "my-registry",
+		RegistrySecretName: "kaniko-secret",
+		ContainersSnapshot: instancesnapshot_controller.ContainersSnapshotOpts{
+			ContainerKaniko:    "kaniko",
+			ContainerImgExport: "crownlabs/img-export",
+		},
+	}).SetupWithManager(k8sManager)
+	Expect(err).ToNot(HaveOccurred())
+
+	go func() {
+		err = k8sManager.Start(ctrl.SetupSignalHandler())
+		Expect(err).ToNot(HaveOccurred())
+	}()
+
+	k8sClient = k8sManager.GetClient()
+	Expect(k8sClient).ToNot(BeNil())
+
+	close(done)
+}, 60)
+
+var _ = AfterSuite(func() {
+	By("tearing down the test environment")
+	err := testEnv.Stop()
+	Expect(err).ToNot(HaveOccurred())
+})
+
+func doesEventuallyExists(ctx context.Context, objLookupKey types.NamespacedName, targetObj client.Object, expectedStatus gomegaTypes.GomegaMatcher, timeout, interval time.Duration) {
+	Eventually(func() bool {
+		err := k8sClient.Get(ctx, objLookupKey, targetObj)
+		return err == nil
+	}, timeout, interval).Should(expectedStatus)
+}
diff --git a/operators/pkg/instancesnapshot-controller/instancesnapshot_controller_test.go b/operators/pkg/instancesnapshot-controller/instancesnapshot_controller_test.go
new file mode 100644
index 000000000..d9952e29f
--- /dev/null
+++ b/operators/pkg/instancesnapshot-controller/instancesnapshot_controller_test.go
@@ -0,0 +1,334 @@
+package instancesnapshot_controller_test
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	batch "k8s.io/api/batch/v1"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/rand"
+
+	crownlabsv1alpha2 "github.com/netgroup-polito/CrownLabs/operators/api/v1alpha2"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	. "github.com/onsi/gomega/gstruct"
+)
+
+var _ = Describe("InstancesnapshotController", func() {
+	// Define utility constants for object names and testing timeouts/durations and intervals.
+	const (
+		InstanceName         = "test-instance"
+		WorkingNamespace     = "working-namespace"
+		TemplateName         = "test-template"
+		TenantName           = "test-tenant"
+		InstanceSnapshotName = "isnap-sample"
+
+		timeout  = time.Second * 20
+		interval = time.Millisecond * 500
+	)
+
+	var (
+		workingNs = v1.Namespace{
+			TypeMeta: metav1.TypeMeta{},
+			ObjectMeta: metav1.ObjectMeta{
+				Name: WorkingNamespace,
+				Labels: map[string]string{
+					"test-suite": "true",
+				},
+			},
+			Spec:   v1.NamespaceSpec{},
+			Status: v1.NamespaceStatus{},
+		}
+		templateEnvironment = crownlabsv1alpha2.TemplateSpec{
+			WorkspaceRef: crownlabsv1alpha2.GenericRef{},
+			PrettyName:   "My Template",
+			Description:  "Description of my template",
+			EnvironmentList: []crownlabsv1alpha2.Environment{
+				{
+					Name:       "Env-1",
+					GuiEnabled: true,
+					Resources: crownlabsv1alpha2.EnvironmentResources{
+						CPU:                   1,
+						ReservedCPUPercentage: 1,
+						Memory:                resource.MustParse("1024M"),
+					},
+					EnvironmentType: crownlabsv1alpha2.ClassVM,
+					Persistent:      true,
+					Image:           "crownlabs/vm",
+				},
+			},
+			DeleteAfter: "",
+		}
+		template = crownlabsv1alpha2.Template{
+			TypeMeta: metav1.TypeMeta{},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      TemplateName,
+				Namespace: WorkingNamespace,
+			},
+			Spec:   templateEnvironment,
+			Status: crownlabsv1alpha2.TemplateStatus{},
+		}
+		instance = crownlabsv1alpha2.Instance{
+			TypeMeta: metav1.TypeMeta{},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      InstanceName,
+				Namespace: WorkingNamespace,
+			},
+			Spec: crownlabsv1alpha2.InstanceSpec{
+				Running: false,
+				Template: crownlabsv1alpha2.GenericRef{
+					Name:      TemplateName,
+					Namespace: WorkingNamespace,
+				},
+				Tenant: crownlabsv1alpha2.GenericRef{
+					Name: TenantName,
+				},
+			},
+			Status: crownlabsv1alpha2.InstanceStatus{},
+		}
+
+		instanceSnapshot = crownlabsv1alpha2.InstanceSnapshot{
+			TypeMeta: metav1.TypeMeta{},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      InstanceSnapshotName,
+				Namespace: WorkingNamespace,
+			},
+			Spec: crownlabsv1alpha2.InstanceSnapshotSpec{
+				Instance: crownlabsv1alpha2.GenericRef{
+					Name:      InstanceName,
+					Namespace: WorkingNamespace,
+				},
+				ImageName: "test-image",
+			},
+		}
+	)
+
+	ctx := context.Background()
+
+	BeforeEach(func() {
+		By("Preparing the environment for the new test")
+		newNs := workingNs.DeepCopy()
+		newTemplate := template.DeepCopy()
+		newInstance := instance.DeepCopy()
+		By("Creating the namespace where to create instance and template")
+		err := k8sClient.Create(ctx, newNs)
+		if err != nil && errors.IsAlreadyExists(err) {
+			By("Cleaning up the environment")
+			By("Deleting template")
+			Expect(k8sClient.Delete(ctx, &template)).Should(Succeed())
+			By("Deleting instance")
+			Expect(k8sClient.Delete(ctx, &instance)).Should(Succeed())
+		} else if err != nil {
+			Fail(fmt.Sprintf("Unable to create namespace -> %s", err))
+		}
+
+		By("By checking that the namespace has been created")
+		createdNs := &v1.Namespace{}
+
+		nsLookupKey := types.NamespacedName{Name: WorkingNamespace}
+		doesEventuallyExists(ctx, nsLookupKey, createdNs, BeTrue(), timeout, interval)
+
+		By("Creating the template")
+		Expect(k8sClient.Create(ctx, newTemplate)).Should(Succeed())
+
+		By("By checking that the template has been created")
+		templateLookupKey := types.NamespacedName{Name: TemplateName, Namespace: WorkingNamespace}
+		createdTemplate := &crownlabsv1alpha2.Template{}
+
+		doesEventuallyExists(ctx, templateLookupKey, createdTemplate, BeTrue(), timeout, interval)
+
+		By("Creating the instance")
+		Expect(k8sClient.Create(ctx, newInstance)).Should(Succeed())
+
+		By("Checking that the instance has been created")
+		instanceLookupKey := types.NamespacedName{Name: InstanceName, Namespace: WorkingNamespace}
+		createdInstance := &crownlabsv1alpha2.Instance{}
+
+		doesEventuallyExists(ctx, instanceLookupKey, createdInstance, BeTrue(), timeout, interval)
+	})
+
+	Context("Creating a snapshot of a persistent VM", func() {
+		It("Should start snapshot creation", func() {
+			newInstanceSnapshot := instanceSnapshot.DeepCopy()
+			newInstanceSnapshot.Name = fmt.Sprintf("isnap-sample-%v", rand.Int())
+			checkIsnapSuccessfulCreation(ctx, newInstanceSnapshot, WorkingNamespace, timeout, interval)
+
+			By("Changing the job status to completed and set Job start and end time")
+			jobLookupKey := types.NamespacedName{Name: newInstanceSnapshot.Name, Namespace: WorkingNamespace}
+			snapjob := &batch.Job{}
+			Expect(k8sClient.Get(ctx, jobLookupKey, snapjob)).Should(Succeed())
+			snapjob.Status.Conditions = []batch.JobCondition{
+				{Type: batch.JobComplete, Status: v1.ConditionTrue},
+			}
+			snapjob.Status.CompletionTime = &metav1.Time{Time: time.Now()}
+			snapjob.Status.StartTime = &metav1.Time{Time: time.Now().Add(-4 * time.Minute)}
+			Expect(k8sClient.Status().Update(ctx, snapjob)).Should(Succeed())
+
+			By("Checking if the InstanceSnapshot status is Completed")
+			checkIsnapStatus(ctx, newInstanceSnapshot.Name, WorkingNamespace, crownlabsv1alpha2.Completed, timeout, interval)
+		})
+
+		It("Should start snapshot creation given an environment name", func() {
+			newInstanceSnapshot := instanceSnapshot.DeepCopy()
+			newInstanceSnapshot.Name = fmt.Sprintf("isnap-sample-%v", rand.Int())
+			newInstanceSnapshot.Spec.Environment.Name = templateEnvironment.EnvironmentList[0].Name
+			checkIsnapSuccessfulCreation(ctx, newInstanceSnapshot, WorkingNamespace, timeout, interval)
+
+			By("Changing the job status to completed without setting Job start and end time")
+			jobLookupKey := types.NamespacedName{Name: newInstanceSnapshot.Name, Namespace: WorkingNamespace}
+			snapjob := &batch.Job{}
+			Expect(k8sClient.Get(ctx, jobLookupKey, snapjob)).Should(Succeed())
+			snapjob.Status.Conditions = []batch.JobCondition{
+				{Type: batch.JobComplete, Status: v1.ConditionTrue},
+			}
+			Expect(k8sClient.Status().Update(ctx, snapjob)).Should(Succeed())
+
+			By("Checking if the InstanceSnapshot status is Completed")
+			checkIsnapStatus(ctx, newInstanceSnapshot.Name, WorkingNamespace, crownlabsv1alpha2.Completed, timeout, interval)
+		})
+	})
+
+	Context("Testing incorrect environment configurations", func() {
+		It("Should fail: the VM is running", func() {
+			By("Getting current instance")
+			currentInstance := &crownlabsv1alpha2.Instance{}
+			instanceLookupKey := types.NamespacedName{Name: InstanceName, Namespace: WorkingNamespace}
+			Expect(k8sClient.Get(ctx, instanceLookupKey, currentInstance)).Should(Succeed())
+
+			By("Setting instance as powered on")
+			currentInstance.Spec.Running = true
+			Expect(k8sClient.Update(ctx, currentInstance)).Should(Succeed())
+
+			newInstanceSnapshot := instanceSnapshot.DeepCopy()
+			newInstanceSnapshot.Name = fmt.Sprintf("isnap-sample-%v", rand.Int())
+			checkIsnapCreationFailure(ctx, newInstanceSnapshot, WorkingNamespace, timeout, interval)
+		})
+
+		It("Should fail: vm is not persistent", func() {
+			By("Getting current Template")
+			currentTemplate := &crownlabsv1alpha2.Template{}
+			templateLookupKey := types.NamespacedName{Name: TemplateName, Namespace: WorkingNamespace}
+			Expect(k8sClient.Get(ctx, templateLookupKey, currentTemplate)).Should(Succeed())
+
+			By("Setting environment VM as not persistent")
+			currentTemplate.Spec.EnvironmentList[0].Persistent = false
+			Expect(k8sClient.Update(ctx, currentTemplate)).Should(Succeed())
+
+			newInstanceSnapshot := instanceSnapshot.DeepCopy()
+			newInstanceSnapshot.Name = fmt.Sprintf("isnap-sample-%v", rand.Int())
+			checkIsnapCreationFailure(ctx, newInstanceSnapshot, WorkingNamespace, timeout, interval)
+		})
+
+		It("Should fail: environment is a container", func() {
+			By("Getting current Template")
+			currentTemplate := &crownlabsv1alpha2.Template{}
+			templateLookupKey := types.NamespacedName{Name: TemplateName, Namespace: WorkingNamespace}
+			Expect(k8sClient.Get(ctx, templateLookupKey, currentTemplate)).Should(Succeed())
+
+			By("Setting environment as Container")
+			currentTemplate.Spec.EnvironmentList[0].EnvironmentType = crownlabsv1alpha2.ClassContainer
+			Expect(k8sClient.Update(ctx, currentTemplate)).Should(Succeed())
+
+			newInstanceSnapshot := instanceSnapshot.DeepCopy()
+			newInstanceSnapshot.Name = fmt.Sprintf("isnap-sample-%v", rand.Int())
+			checkIsnapCreationFailure(ctx, newInstanceSnapshot, WorkingNamespace, timeout, interval)
+		})
+
+		It("Should fail: template does not exist", func() {
+			By("Getting current instance")
+			currentInstance := &crownlabsv1alpha2.Instance{}
+			instanceLookupKey := types.NamespacedName{Name: InstanceName, Namespace: WorkingNamespace}
+			Expect(k8sClient.Get(ctx, instanceLookupKey, currentInstance)).Should(Succeed())
+
+			By("Changing template with a non-existing one")
+			currentInstance.Spec.Template.Name = "invalid"
+			Expect(k8sClient.Update(ctx, currentInstance)).Should(Succeed())
+
+			newInstanceSnapshot := instanceSnapshot.DeepCopy()
+			newInstanceSnapshot.Name = fmt.Sprintf("isnap-sample-%v", rand.Int())
+			checkIsnapCreationFailure(ctx, newInstanceSnapshot, WorkingNamespace, timeout, interval)
+		})
+
+		It("Should fail: instance does not exist", func() {
+			By("Setting not existing instance name")
+			newInstanceSnapshot := instanceSnapshot.DeepCopy()
+			newInstanceSnapshot.Name = fmt.Sprintf("isnap-sample-%v", rand.Int())
+			newInstanceSnapshot.Spec.Instance.Name = "invalid"
+			checkIsnapCreationFailure(ctx, newInstanceSnapshot, WorkingNamespace, timeout, interval)
+		})
+	})
+
+	Context("Testing snapshotting job failures", func() {
+		It("Should fail: job failed", func() {
+			newInstanceSnapshot := instanceSnapshot.DeepCopy()
+			newInstanceSnapshot.Name = fmt.Sprintf("isnap-sample-%v", rand.Int())
+			checkIsnapSuccessfulCreation(ctx, newInstanceSnapshot, WorkingNamespace, timeout, interval)
+
+			By("Changing the job status to failed")
+			jobLookupKey := types.NamespacedName{Name: newInstanceSnapshot.Name, Namespace: WorkingNamespace}
+			snapjob := &batch.Job{}
+			Expect(k8sClient.Get(ctx, jobLookupKey, snapjob)).Should(Succeed())
+			snapjob.Status.Conditions = []batch.JobCondition{
+				{Type: batch.JobFailed, Status: v1.ConditionTrue},
+			}
+			Expect(k8sClient.Status().Update(ctx, snapjob)).Should(Succeed())
+
+			By("Checking if the InstanceSnapshot status is Failed")
+			checkIsnapStatus(ctx, newInstanceSnapshot.Name, WorkingNamespace, crownlabsv1alpha2.Failed, timeout, interval)
+		})
+	})
+})
+
+func checkIsnapStatus(ctx context.Context, isnapName, workingNamespace string, desiredStatus crownlabsv1alpha2.SnapshotStatus, timeout, interval time.Duration) {
+	isnapLookupKey := types.NamespacedName{Name: isnapName, Namespace: workingNamespace}
+	retrievedIsnap := &crownlabsv1alpha2.InstanceSnapshot{}
+	Eventually(func() crownlabsv1alpha2.SnapshotStatus {
+		err := k8sClient.Get(ctx, isnapLookupKey, retrievedIsnap)
+		if err != nil {
+			return ""
+		}
+		return retrievedIsnap.Status.Phase
+	}, timeout, interval).Should(Equal(desiredStatus))
+}
+
+func checkIsnapSuccessfulCreation(ctx context.Context, isnap *crownlabsv1alpha2.InstanceSnapshot, workingNamespace string, timeout, interval time.Duration) {
+	By("Creating the InstanceSnapshot resource")
+	Expect(k8sClient.Create(ctx, isnap)).Should(Succeed())
+
+	By("Checking that the instance snapshot has been created")
+	instanceSnapshotLookupKey := types.NamespacedName{Name: isnap.Name, Namespace: workingNamespace}
+	createdInstanceSnapshot := &crownlabsv1alpha2.InstanceSnapshot{}
+
+	doesEventuallyExists(ctx, instanceSnapshotLookupKey, createdInstanceSnapshot, BeTrue(), timeout, interval)
+
+	By("Checking if the job for the creation of the snapshot has been created")
+	jobLookupKey := types.NamespacedName{Name: isnap.Name, Namespace: workingNamespace}
+	createdJob := &batch.Job{}
+
+	doesEventuallyExists(ctx, jobLookupKey, createdJob, BeTrue(), timeout, interval)
+
+	By("Checking the owner reference of the job")
+	Expect(createdJob.ObjectMeta.OwnerReferences).To(ContainElement(MatchFields(IgnoreExtras, Fields{
+		"UID": Equal(createdInstanceSnapshot.UID),
+	})))
+}
+
+func checkIsnapCreationFailure(ctx context.Context, isnap *crownlabsv1alpha2.InstanceSnapshot, workingNamespace string, timeout, interval time.Duration) {
+	By("Creating the InstanceSnapshot resource")
+	Expect(k8sClient.Create(ctx, isnap)).Should(Succeed())
+
+	By("Checking that the instance has been created")
+	instanceSnapshotLookupKey := types.NamespacedName{Name: isnap.Name, Namespace: workingNamespace}
+	createdInstanceSnapshot := &crownlabsv1alpha2.InstanceSnapshot{}
+
+	doesEventuallyExists(ctx, instanceSnapshotLookupKey, createdInstanceSnapshot, BeTrue(), timeout, interval)
+
+	By("Checking that the InstanceSnapshot failed")
+	checkIsnapStatus(ctx, isnap.Name, workingNamespace, crownlabsv1alpha2.Failed, timeout, interval)
+}
diff --git a/operators/pkg/instancesnapshot-controller/instancesnapshot_helpers.go b/operators/pkg/instancesnapshot-controller/instancesnapshot_helpers.go
new file mode 100644
index 000000000..b0d722fd4
--- /dev/null
+++ b/operators/pkg/instancesnapshot-controller/instancesnapshot_helpers.go
@@ -0,0 +1,260 @@
+/*
+
+
+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 instancesnapshot_controller
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	batch "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+
+	crownlabsv1alpha2 "github.com/netgroup-polito/CrownLabs/operators/api/v1alpha2"
+	"github.com/netgroup-polito/CrownLabs/operators/pkg/utils"
+)
+
+// ValidateRequest validates the InstanceSnapshot request, returns an error and if there's the need to try again.
+func (r *InstanceSnapshotReconciler) ValidateRequest(ctx context.Context, isnap *crownlabsv1alpha2.InstanceSnapshot) (bool, error) {
+	// First it is needed to check if the instance actually exists.
+	instanceName := types.NamespacedName{
+		Namespace: isnap.Spec.Instance.Namespace,
+		Name:      isnap.Spec.Instance.Name,
+	}
+	instance := &crownlabsv1alpha2.Instance{}
+
+	if err := r.Get(ctx, instanceName, instance); err != nil && errors.IsNotFound(err) {
+		// The declared instance does not exist so don't try again.
+		return false, fmt.Errorf("instance %s not found in namespace %s. It is not possible to complete the InstanceSnapshot %s",
+			instanceName.Name, instanceName.Namespace, isnap.Name)
+	} else if err != nil {
+		return true, fmt.Errorf("error in retrieving the instance for InstanceSnapshot %s -> %w", isnap.Name, err)
+	}
+
+	// Get the template of the instance in order to check if it has the requirements to be snapshotted.
+	// In order to create a snapshot of the vm, we need first to check that:
+	// - the vm is powered off, since it is not possible to steal the DataVolume if it is still running;
+	// - the environment is a persistent vm and not a container.
+
+	templateName := types.NamespacedName{
+		Namespace: instance.Spec.Template.Namespace,
+		Name:      instance.Spec.Template.Name,
+	}
+	template := &crownlabsv1alpha2.Template{}
+
+	if err := r.Get(ctx, templateName, template); err != nil && errors.IsNotFound(err) {
+		// The declared template does not exist set the phase as failed and don't try again.
+		return false, fmt.Errorf("template %s not found in namespace %s. It is not possible to complete the InstanceSnapshot %s",
+			templateName.Name, templateName.Namespace, isnap.Name)
+	} else if err != nil {
+		return true, fmt.Errorf("error in retrieving the template for InstanceSnapshot %s -> %w", isnap.Name, err)
+	}
+
+	// Retrieve the environment from the template.
+	var env *crownlabsv1alpha2.Environment = nil
+
+	if isnap.Spec.Environment.Name != "" {
+		for i := range template.Spec.EnvironmentList {
+			if template.Spec.EnvironmentList[i].Name == isnap.Spec.Environment.Name {
+				env = &template.Spec.EnvironmentList[i]
+				break
+			}
+		}
+
+		// Check if the specified environment was found.
+		if env == nil {
+			return false, fmt.Errorf("environment %s not found in template %s. It is not possible to complete the InstanceSnapshot %s",
+				isnap.Spec.Environment.Name, template.Name, isnap.Name)
+		}
+	} else {
+		// If the environment is not explicitly declared, take the first one.
+		env = &template.Spec.EnvironmentList[0]
+	}
+
+	// Check if the environment is a persistent VM.
+	if env.EnvironmentType != crownlabsv1alpha2.ClassVM || !env.Persistent {
+		return false, fmt.Errorf("environment %s is not a persistent VM. It is not possible to complete the InstanceSnapshot %s",
+			env.Name, isnap.Name)
+	}
+
+	// Check if the VM is running.
+	if instance.Spec.Running {
+		return false, fmt.Errorf("the vm is running. It is not possible to complete the InstanceSnapshot %s", isnap.Name)
+	}
+
+	return false, nil
+}
+
+// GetJobStatus sets a Job and returns its status.
+func (r *InstanceSnapshotReconciler) GetJobStatus(job *batch.Job) (bool, batch.JobConditionType) {
+	for _, c := range job.Status.Conditions {
+		// If the status corresponding to Success or failed is true, it means that the job completed.
+		if c.Status == corev1.ConditionTrue && (c.Type == batch.JobFailed || c.Type == batch.JobComplete) {
+			return true, c.Type
+		}
+	}
+
+	// Job did not complete.
+	return false, ""
+}
+
+// CreateSnapshottingJobDefinition generates the job to be created.
+func (r *InstanceSnapshotReconciler) CreateSnapshottingJobDefinition(ctx context.Context, isnap *crownlabsv1alpha2.InstanceSnapshot) (batch.Job, error) {
+	// Get the tenant name in order to set it as directory of the image
+	instanceName := types.NamespacedName{
+		Namespace: isnap.Spec.Instance.Namespace,
+		Name:      isnap.Spec.Instance.Name,
+	}
+	instance := &crownlabsv1alpha2.Instance{}
+
+	if err := r.Get(ctx, instanceName, instance); err != nil {
+		return batch.Job{}, fmt.Errorf("error in retrieving the instance for InstanceSnapshot %s -> %w", isnap.Name, err)
+	}
+
+	var backoff int32 = 2
+	imagetag := fmt.Sprint(time.Now().Format("20060102t150405"))
+	// Volume name does not accept dots, replace them with dashes
+	volumename := strings.ReplaceAll(isnap.Spec.Instance.Name, ".", "-")
+	imagedir := utils.ParseDockerDirectory(instance.Spec.Tenant.Name)
+
+	// Define volumes.
+
+	// Define VM VolumeSource.
+	vmvolume := corev1.VolumeSource{
+		PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
+			ClaimName: volumename,
+		},
+	}
+
+	// Define temp VolumeSource.
+	tmpvol := corev1.VolumeSource{
+		EmptyDir: &corev1.EmptyDirVolumeSource{},
+	}
+
+	// Define secret VolumeSource.
+	secretvol := corev1.VolumeSource{
+		Secret: &corev1.SecretVolumeSource{
+			SecretName: r.RegistrySecretName,
+			Items: []corev1.KeyToPath{
+				{
+					Key:  ".dockerconfigjson",
+					Path: "config.json",
+				},
+			},
+		},
+	}
+
+	volumes := []corev1.Volume{
+		{
+			Name:         volumename,
+			VolumeSource: vmvolume,
+		},
+		{
+			Name:         "tmp-vol",
+			VolumeSource: tmpvol,
+		},
+		{
+			Name:         "kaniko-secret",
+			VolumeSource: secretvol,
+		},
+	}
+
+	// Define containers.
+
+	// Define Docker pusher container.
+	pushcontainer := corev1.Container{
+		Name:  "docker-pusher",
+		Image: r.ContainersSnapshot.ContainerKaniko,
+		Args: []string{"--dockerfile=/workspace/Dockerfile",
+			fmt.Sprintf("--destination=%s/%s/%s:%s", r.VMRegistry, imagedir, isnap.Spec.ImageName, imagetag)},
+		VolumeMounts: []corev1.VolumeMount{
+			{
+				Name:      "tmp-vol",
+				MountPath: "/workspace",
+			},
+			{
+				Name:      "kaniko-secret",
+				MountPath: "/kaniko/.docker/",
+			},
+		},
+		Resources: corev1.ResourceRequirements{
+			Requests: corev1.ResourceList{
+				"cpu":    resource.MustParse("1"),
+				"memory": resource.MustParse("8Gi"),
+			},
+			Limits: corev1.ResourceList{
+				"cpu":    resource.MustParse("1"),
+				"memory": resource.MustParse("32Gi"),
+			},
+		},
+	}
+
+	// Define image exporter container.
+	exportcontainer := corev1.Container{
+		Name:  "img-generator",
+		Image: r.ContainersSnapshot.ContainerImgExport,
+		VolumeMounts: []corev1.VolumeMount{
+			{
+				Name:      volumename,
+				MountPath: "/data",
+			},
+			{
+				Name:      "tmp-vol",
+				MountPath: "/img-tmp",
+			},
+		},
+		Resources: corev1.ResourceRequirements{
+			Requests: corev1.ResourceList{
+				"cpu":    resource.MustParse("1"),
+				"memory": resource.MustParse("128Mi"),
+			},
+			Limits: corev1.ResourceList{
+				"cpu":    resource.MustParse("1"),
+				"memory": resource.MustParse("256Mi"),
+			},
+		},
+	}
+
+	snapjob := batch.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      isnap.Name,
+			Namespace: isnap.Namespace,
+		},
+		Spec: batch.JobSpec{
+			BackoffLimit: &backoff,
+			Template: corev1.PodTemplateSpec{
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						pushcontainer,
+					},
+					InitContainers: []corev1.Container{
+						exportcontainer,
+					},
+					Volumes:       volumes,
+					RestartPolicy: corev1.RestartPolicyOnFailure,
+				},
+			},
+		},
+	}
+
+	return snapjob, nil
+}
diff --git a/operators/pkg/instancesnapshot-controller/instancesnapshot_logic.go b/operators/pkg/instancesnapshot-controller/instancesnapshot_logic.go
new file mode 100644
index 000000000..4ec0cf9ea
--- /dev/null
+++ b/operators/pkg/instancesnapshot-controller/instancesnapshot_logic.go
@@ -0,0 +1,81 @@
+package instancesnapshot_controller
+
+import (
+	"context"
+	"fmt"
+
+	batch "k8s.io/api/batch/v1"
+	ctrl "sigs.k8s.io/controller-runtime"
+
+	crownlabsv1alpha2 "github.com/netgroup-polito/CrownLabs/operators/api/v1alpha2"
+)
+
+// CreateSnapshottingJob creates the job in charge of creating the snapshot.
+func (r *InstanceSnapshotReconciler) CreateSnapshottingJob(ctx context.Context, isnap *crownlabsv1alpha2.InstanceSnapshot) (bool, error) {
+	r.EventsRecorder.Event(isnap, "Normal", "Validating", "Start validation of the request")
+
+	isnap.Status.Phase = crownlabsv1alpha2.Pending
+	if err := r.Status().Update(ctx, isnap); err != nil {
+		return true, fmt.Errorf("error when updating status of InstanceSnapshot %s -> %w", isnap.Name, err)
+	}
+
+	if retry, err := r.ValidateRequest(ctx, isnap); err != nil {
+		// Print the validation error in the log and check if there is the need to
+		// set the operation as failed, or to try again.
+		if retry {
+			return true, err
+		}
+
+		// Set the status as failed
+		isnap.Status.Phase = crownlabsv1alpha2.Failed
+		if uerr := r.Status().Update(ctx, isnap); uerr != nil {
+			return true, fmt.Errorf("error when updating status of InstanceSnapshot %s -> %w", isnap.Name, uerr)
+		}
+
+		return false, err
+	}
+
+	// Get the job to be created
+	snapjob, err1 := r.CreateSnapshottingJobDefinition(ctx, isnap)
+	if err1 != nil {
+		return true, err1
+	}
+
+	// Set the owner reference in order to delete the job when the InstanceSnapshot is deleted.
+	if err := ctrl.SetControllerReference(isnap, &snapjob, r.Scheme); err != nil {
+		return true, err
+	}
+
+	if err := r.Create(ctx, &snapjob); err != nil {
+		// It was not possible to create the job
+		return true, fmt.Errorf("error when creating the job for %s -> %w", isnap.Name, err)
+	}
+
+	isnap.Status.Phase = crownlabsv1alpha2.Processing
+	if err := r.Status().Update(ctx, isnap); err != nil {
+		return true, fmt.Errorf("error when updating status of InstanceSnapshot %s -> %w", isnap.Name, err)
+	}
+
+	return false, nil
+}
+
+// HandleExistingJob checks the status of the existing job and updates the status of the InstanceSnapshot accordingly.
+func (r *InstanceSnapshotReconciler) HandleExistingJob(ctx context.Context, isnap *crownlabsv1alpha2.InstanceSnapshot, snapjob *batch.Job) (batch.JobConditionType, error) {
+	completed, jstatus := r.GetJobStatus(snapjob)
+	if completed {
+		if jstatus == batch.JobComplete {
+			// The job is completed and the image has been uploaded to the registry
+			isnap.Status.Phase = crownlabsv1alpha2.Completed
+			if err := r.Status().Update(ctx, isnap); err != nil {
+				return "", fmt.Errorf("error when updating status of InstanceSnapshot %s -> %w", isnap.Name, err)
+			}
+		} else {
+			// The creation of the snapshot failed since the job failed
+			isnap.Status.Phase = crownlabsv1alpha2.Failed
+			if err := r.Status().Update(ctx, isnap); err != nil {
+				return "", fmt.Errorf("error when updating status of InstanceSnapshot %s -> %w", isnap.Name, err)
+			}
+		}
+	}
+	return jstatus, nil
+}
diff --git a/operators/pkg/utils/common.go b/operators/pkg/utils/common.go
new file mode 100644
index 000000000..e172fb230
--- /dev/null
+++ b/operators/pkg/utils/common.go
@@ -0,0 +1,69 @@
+/*
+
+
+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 utils collects all the logic shared between different controllers
+package utils
+
+import (
+	"context"
+	"fmt"
+	"regexp"
+	"strings"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// ParseDockerDirectory returns a valid Docker image directory.
+func ParseDockerDirectory(name string) string {
+	reg := regexp.MustCompile("[^a-zA-Z0-9]+")
+	return strings.ToLower(reg.ReplaceAllString(name, ""))
+}
+
+// CheckLabels verifies whether a namespace is characterized by a set of required labels.
+func CheckLabels(ns *corev1.Namespace, matchLabels map[string]string) bool {
+	for key, value := range matchLabels {
+		if v1, ok := ns.Labels[key]; !ok || v1 != value {
+			return false
+		}
+	}
+	return true
+}
+
+// CheckSelectorLabel checks if the given namespace belongs to the whitelisted namespaces where to perform reconciliation.
+func CheckSelectorLabel(ctx context.Context, k8sClient client.Client, namespaceName string, matchLabels map[string]string) (bool, error) {
+	ns := corev1.Namespace{}
+	namespaceLookupKey := types.NamespacedName{
+		Name:      namespaceName,
+		Namespace: "",
+	}
+
+	// It performs reconciliation only if the InstanceSnapshot belongs to whitelisted namespaces
+	// by checking the existence of keys in the namespace of the InstanceSnapshot.
+	if err := k8sClient.Get(ctx, namespaceLookupKey, &ns); err == nil {
+		if !CheckLabels(&ns, matchLabels) {
+			klog.Infof("Namespace %s does not meet the selector labels", namespaceName)
+			return false, nil
+		}
+	} else {
+		return false, fmt.Errorf("error when retrieving the InstanceSnapshot namespace -> %w", err)
+	}
+
+	klog.Info("Namespace " + namespaceName + " met the selector labels")
+	return true, nil
+}
diff --git a/operators/samples/instance-snapshot.yaml b/operators/samples/instance-snapshot.yaml
new file mode 100644
index 000000000..91e0ff176
--- /dev/null
+++ b/operators/samples/instance-snapshot.yaml
@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: tenant-john-doe
+---
+apiVersion: crownlabs.polito.it/v1alpha2
+kind: InstanceSnapshot
+metadata:
+  name: green-tea-6831-snapshot
+  namespace: tenant-john-doe
+spec:
+  instanceRef:
+    name: green-tea-6831
+    namespace: workspace-tea
+  imageName: new-green-tea-image
\ No newline at end of file