diff --git a/maro/cli/grass/executors/grass_azure_executor.py b/maro/cli/grass/executors/grass_azure_executor.py
index 46c7c26b0..02a43c1f1 100644
--- a/maro/cli/grass/executors/grass_azure_executor.py
+++ b/maro/cli/grass/executors/grass_azure_executor.py
@@ -14,12 +14,13 @@
 import yaml
 
 from maro.cli.grass.executors.grass_executor import GrassExecutor
-from maro.cli.grass.utils.copy import copy_files_to_node, copy_files_from_node, sync_mkdir
+from maro.cli.grass.utils.copy import copy_files_to_node, copy_files_from_node, sync_mkdir, copy_and_rename
 from maro.cli.grass.utils.hash import get_checksum
 from maro.cli.utils.details import load_cluster_details, save_cluster_details, load_job_details, save_job_details, \
     load_schedule_details, save_schedule_details
 from maro.cli.utils.executors.azure_executor import AzureExecutor
-from maro.cli.utils.naming import generate_cluster_id, generate_job_id, generate_component_id, generate_node_name
+from maro.cli.utils.naming import generate_cluster_id, generate_job_id, generate_component_id, generate_node_name, \
+    get_valid_file_name
 from maro.cli.utils.params import GlobalParams, GlobalPaths
 from maro.cli.utils.subprocess import SubProcess
 from maro.cli.utils.validation import validate_and_fill_dict
@@ -708,16 +709,17 @@ def push_image(self, image_name: str, image_path: str, remote_context_path: str,
 
         # Push image
         if image_name:
-            image_path = f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/images/{image_name}"
-            GrassAzureExecutor._save_image(
+            new_file_name = get_valid_file_name(image_name)
+            image_path = f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/images/{new_file_name}"
+            self._save_image(
                 image_name=image_name,
-                export_path=image_path
+                export_path=os.path.expanduser(image_path)
             )
             if self._check_checksum_validity(
                     local_file_path=os.path.expanduser(image_path),
                     remote_file_path=os.path.join(images_dir, image_name)
             ):
-                logger.info_green(f"The image '{image_name}' already exists")
+                logger.info_green(f"The image file '{new_file_name}' already exists")
                 return
             copy_files_to_node(
                 local_path=image_path,
@@ -729,15 +731,21 @@ def push_image(self, image_name: str, image_path: str, remote_context_path: str,
             logger.info_green(f"Image {image_name} is loaded")
         elif image_path:
             file_name = os.path.basename(image_path)
+            new_file_name = get_valid_file_name(file_name)
+            image_path = f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/images/{new_file_name}"
+            copy_and_rename(
+                source_path=os.path.expanduser(image_path),
+                target_dir=image_path
+            )
             if self._check_checksum_validity(
                     local_file_path=os.path.expanduser(image_path),
-                    remote_file_path=os.path.join(images_dir, file_name)
+                    remote_file_path=os.path.join(images_dir, new_file_name)
             ):
-                logger.info_green(f"The image file '{file_name}' already exists")
+                logger.info_green(f"The image file '{new_file_name}' already exists")
                 return
             copy_files_to_node(
                 local_path=image_path,
-                remote_dir=f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/images/",
+                remote_dir=images_dir,
                 admin_username=admin_username, node_ip_address=master_public_ip_address
             )
             self.grass_executor.remote_update_image_files_details()
@@ -754,7 +762,7 @@ def push_image(self, image_name: str, image_path: str, remote_context_path: str,
     @staticmethod
     def _save_image(image_name: str, export_path: str):
         # Save image to specific folder
-        command = f"docker save {image_name} > {export_path}"
+        command = f"docker save '{image_name}' --output '{export_path}'"
         _ = SubProcess.run(command)
 
     def _batch_load_images(self):
@@ -1028,6 +1036,24 @@ def clean(self):
         # Remote clean
         self.grass_executor.remote_clean(parallels=GlobalParams.PARALLELS)
 
+    # maro grass status
+    
+    def status(self, resource_name: str):
+        if resource_name == "master":
+            return_status = self.grass_executor.remote_get_master_details()
+        elif resource_name == "nodes":
+            return_status = self.grass_executor.remote_get_nodes_details()
+        else:
+            raise CliException(f"Resource {resource_name} is unsupported")
+
+        # Print status
+        logger.info(
+            json.dumps(
+                return_status,
+                indent=4, sort_keys=True
+            )
+        )
+
     # maro grass template
 
     @staticmethod
diff --git a/maro/cli/grass/executors/grass_executor.py b/maro/cli/grass/executors/grass_executor.py
index f0226df0c..96acb4852 100644
--- a/maro/cli/grass/executors/grass_executor.py
+++ b/maro/cli/grass/executors/grass_executor.py
@@ -49,6 +49,13 @@ def remote_get_jobs_details(self):
         return_str = SubProcess.run(command)
         return json.loads(return_str)
 
+    def remote_get_master_details(self):
+        command = f"ssh -o StrictHostKeyChecking=no " \
+                  f"{self.admin_username}@{self.cluster_details['master']['public_ip_address']} " \
+                  f"'python3 {GlobalPaths.MARO_GRASS_LIB}/scripts/get_master_details.py {self.cluster_name}'"
+        return_str = SubProcess.run(command)
+        return json.loads(return_str)
+
     def remote_get_node_details(self, node_name: str):
         command = f"ssh -o StrictHostKeyChecking=no " \
                   f"{self.admin_username}@{self.cluster_details['master']['public_ip_address']} " \
diff --git a/maro/cli/grass/lib/scripts/load_images.py b/maro/cli/grass/lib/scripts/load_images.py
index 34677a41f..24273be7c 100644
--- a/maro/cli/grass/lib/scripts/load_images.py
+++ b/maro/cli/grass/lib/scripts/load_images.py
@@ -3,6 +3,7 @@
 
 
 import argparse
+import os
 import subprocess
 import sys
 from multiprocessing.pool import ThreadPool
@@ -12,12 +13,12 @@
 from utils import load_cluster_details, get_node_details, set_node_details, get_master_details
 
 LOAD_IMAGE_COMMAND = '''\
-docker load -q -i {image_path}
+docker load -q -i "{image_path}"
 '''
 
 
-def load_image(unloaded_image: str):
-    command = LOAD_IMAGE_COMMAND.format(image_path=unloaded_image)
+def load_image(image_path: str):
+    command = LOAD_IMAGE_COMMAND.format(image_path=image_path)
     completed_process = subprocess.run(command,
                                        shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf8')
     if completed_process.returncode != 0:
@@ -69,7 +70,8 @@ def load_image(unloaded_image: str):
 
     # Parallel load
     with ThreadPool(args.parallels) as pool:
-        params = [[unloaded_image] for unloaded_image in unloaded_images]
+        params = [[os.path.expanduser(f"~/.maro/clusters/{args.cluster_name}/images/{unloaded_image}")]
+                  for unloaded_image in unloaded_images]
         pool.starmap(
             load_image,
             params
diff --git a/maro/cli/grass/lib/scripts/update_image_files_details.py b/maro/cli/grass/lib/scripts/update_image_files_details.py
index d85999e75..05a898c36 100644
--- a/maro/cli/grass/lib/scripts/update_image_files_details.py
+++ b/maro/cli/grass/lib/scripts/update_image_files_details.py
@@ -11,14 +11,15 @@
 from utils import load_cluster_details, get_master_details, set_master_details
 
 
-def get_current_image_files_details() -> dict:
-    current_images = glob.glob(os.path.expanduser(f"~/.maro/clusters/maro_grass_test/images/*"))
+def get_current_image_files_details(cluster_name: str) -> dict:
+    image_paths = glob.glob(os.path.expanduser(f"~/.maro/clusters/{cluster_name}/images/*"))
     image_files_details = {}
 
-    for current_image in current_images:
-        image_files_details[current_image] = {
-            'modify_time': os.path.getmtime(current_image),
-            'size': os.path.getsize(current_image)
+    for image_path in image_paths:
+        file_name = os.path.basename(image_path)
+        image_files_details[file_name] = {
+            'modify_time': os.path.getmtime(image_path),
+            'size': os.path.getsize(image_path)
         }
 
     return image_files_details
@@ -39,7 +40,7 @@ def get_current_image_files_details() -> dict:
                   charset="utf-8", decode_responses=True)
 
     # Get details
-    curr_image_files_details = get_current_image_files_details()
+    curr_image_files_details = get_current_image_files_details(cluster_name=args.cluster_name)
     with redis.lock("lock:master"):
         master_details = get_master_details(
             redis=redis,
diff --git a/maro/cli/grass/status.py b/maro/cli/grass/status.py
new file mode 100644
index 000000000..dc3a45d1d
--- /dev/null
+++ b/maro/cli/grass/status.py
@@ -0,0 +1,18 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+
+from maro.cli.grass.executors.grass_azure_executor import GrassAzureExecutor
+from maro.cli.utils.checkers import check_details_validity
+from maro.cli.utils.details import load_cluster_details
+from maro.cli.utils.lock import lock
+
+
+@check_details_validity(mode='grass')
+@lock
+def status(cluster_name: str, resource_name: str, **kwargs):
+    cluster_details = load_cluster_details(cluster_name=cluster_name)
+
+    if cluster_details['cloud']['infra'] == 'azure':
+        executor = GrassAzureExecutor(cluster_name=cluster_name)
+        executor.status(resource_name=resource_name)
diff --git a/maro/cli/maro.py b/maro/cli/maro.py
index 8bfc184bd..7c3f4368f 100644
--- a/maro/cli/maro.py
+++ b/maro/cli/maro.py
@@ -388,6 +388,18 @@ def load_parser_grass(prev_parser: ArgumentParser, global_parser: ArgumentParser
         'cluster_name', help='Name of the cluster')
     parser_clean.set_defaults(func=clean)
 
+    # maro grass status
+    from maro.cli.grass.status import status
+    parser_status = subparsers.add_parser(
+        'status',
+        help='Get status of the cluster',
+        examples=CliExamples.MARO_GRASS_STATUS,
+        parents=[global_parser]
+    )
+    parser_status.add_argument('cluster_name', help='Name of the cluster')
+    parser_status.add_argument('resource_name', help='Name of the resource')
+    parser_status.set_defaults(func=status)
+
     # maro grass template
     from maro.cli.grass.template import template
     parser_clean = subparsers.add_parser(
diff --git a/maro/cli/utils/examples.py b/maro/cli/utils/examples.py
index 7193201a1..1d2f1863c 100644
--- a/maro/cli/utils/examples.py
+++ b/maro/cli/utils/examples.py
@@ -109,6 +109,13 @@
     maro grass clean MyClusterName
 """
 
+MARO_GRASS_STATUS = """
+Examples:
+  Get status of the resource in the cluster
+    maro grass status MyClusterName master
+    maro grass status MyClusterName nodes
+"""
+
 MARO_GRASS_TEMPLATES = """
 Examples:
   Get deployment templates to target directory
diff --git a/maro/cli/utils/naming.py b/maro/cli/utils/naming.py
index ec7d78768..f4c3ae8f1 100644
--- a/maro/cli/utils/naming.py
+++ b/maro/cli/utils/naming.py
@@ -3,9 +3,14 @@
 
 
 import hashlib
+import re
 import uuid
 
 
+def get_valid_file_name(file_name: str):
+    return re.sub(r'[\\/*?:"<>|]', "_", file_name)
+
+
 def generate_name_with_uuid(prefix: str, uuid_len: int = 16) -> str:
     postfix = uuid.uuid4().hex[:uuid_len]
     return f"{prefix}{postfix}"