diff --git a/etc/staffeln/staffeln.conf b/etc/staffeln/staffeln.conf
index 19a214a..13cfacf 100644
--- a/etc/staffeln/staffeln.conf
+++ b/etc/staffeln/staffeln.conf
@@ -7,6 +7,7 @@
 backup_cycle_timout = 5min
 retention_time = 2w3d
 backup_metadata_key="__automated_backup"
+retention_metadata_key="__staffeln_retention"
 
 [database]
 backend = sqlalchemy
@@ -30,4 +31,4 @@
 [notification]
 # receiver = reciever@gmail.com
 # sender_email = sender@vexxhost.com
-# smtp_server_domain = localhost
\ No newline at end of file
+# smtp_server_domain = localhost
diff --git a/staffeln/common/openstack.py b/staffeln/common/openstack.py
index 71bafc9..b6329a2 100644
--- a/staffeln/common/openstack.py
+++ b/staffeln/common/openstack.py
@@ -38,10 +38,13 @@
     def get_projects(self):
         return self.conn.list_projects()
 
-    def get_servers(self, project_id, all_projects=True, details=True):
-        return self.conn.compute.servers(
-            details=details, all_projects=all_projects, project_id=project_id
-        )
+    def get_servers(self, project_id=None, all_projects=True, details=True):
+        if project_id is not None:
+            return self.conn.compute.servers(
+                details=details, all_projects=all_projects, project_id=project_id
+            )
+        else:
+            return self.conn.compute.servers(details=details, all_projects=all_projects)
 
     def get_volume(self, uuid, project_id):
         return self.conn.get_volume_by_id(uuid)
diff --git a/staffeln/conductor/backup.py b/staffeln/conductor/backup.py
index 930ae0b..21ef724 100755
--- a/staffeln/conductor/backup.py
+++ b/staffeln/conductor/backup.py
@@ -23,6 +23,7 @@
         "instance_id",
         "backup_completed",
         "incremental",
+        "created_at",
     ],
 )
 
@@ -363,6 +364,27 @@
                     )
         return queues_map
 
+    def collect_instance_retention_map(self):
+        """Retrieves instance backup retention map"""
+
+        retention_map = {}
+        # No customized retention.
+        if not CONF.conductor.retention_metadata_key:
+            return retention_map
+        self.refresh_openstacksdk()
+
+        try:
+            servers = self.openstacksdk.get_servers(all_projects=True)
+        except OpenstackHttpException as ex:
+            LOG.warn(_("Failed to list servers for all projects."))
+
+        for server in servers:
+            if CONF.conductor.retention_metadata_key in server.metadata:
+                retention_map[server.id] = server.metadata[
+                    CONF.conductor.retention_metadata_key
+                ].lower()
+        return retention_map
+
     def _volume_queue(self, task):
         """
         Commits one backup task to queue table
diff --git a/staffeln/conductor/manager.py b/staffeln/conductor/manager.py
index bbd5bcb..ed324a1 100755
--- a/staffeln/conductor/manager.py
+++ b/staffeln/conductor/manager.py
@@ -1,3 +1,4 @@
+from datetime import datetime, timezone
 import threading
 import time
 
@@ -164,6 +165,7 @@
         super(RotationManager, self).__init__(worker_id)
         self._shutdown = threading.Event()
         self.conf = conf
+        self.lock_mgt = lock.LockManager()
         self.controller = backup.Backup()
         LOG.info("%s init" % self.name)
 
@@ -178,18 +180,12 @@
     def reload(self):
         LOG.info("%s reload" % self.name)
 
-    def get_backup_list(self):
-        threshold_strtime = self.get_threshold_strtime()
-        if threshold_strtime is None:
-            return False
-        self.backup_list = self.controller.get_backups(
-            filters={"created_at__lt": threshold_strtime}
-        )
-        return True
+    def get_backup_list(self, filters=None):
+        return self.controller.get_backups(filters=filters)
 
-    def remove_backups(self):
-        print(self.backup_list)
-        for retention_backup in self.backup_list:
+    def remove_backups(self, retention_backups):
+        LOG.info(_("Backups to be removed: %s" % retention_backups))
+        for retention_backup in retention_backups:
             self.controller.hard_remove_volume_backup(retention_backup)
 
     def rotation_engine(self, retention_service_period):
@@ -197,14 +193,46 @@
 
         @periodics.periodic(spacing=retention_service_period, run_immediately=True)
         def rotation_tasks():
-            self.controller.refresh_openstacksdk()
-            # 1. get the list of backups to remove based on the retention time
-            if not self.get_backup_list():
-                return
-            # 2. get project list
-            self.controller.update_project_list()
-            # 3. remove the backups
-            self.remove_backups()
+            try:
+                # TODO(rlin): change to use decorator for this
+                # Make sure only one retention at a time
+                with self.lock_mgt.coordinator.get_lock("retention"):
+                    self.controller.refresh_openstacksdk()
+                    # get the threshold time
+                    threshold_strtime = self.get_time_from_str(CONF.conductor.retention_time)
+                    instance_retention_map = self.controller.collect_instance_retention_map()
+                    retention_backups = []
+                    # 1. get init list of backups to remove based on the retention time
+                    if not instance_retention_map:
+                        retention_backups = self.get_backup_list(
+                            filters={
+                                "created_at__lt": threshold_strtime.strftime(
+                                    xtime.DEFAULT_TIME_FORMAT
+                                )
+                            }
+                        )
+                    else:
+                        full_backup_list = self.get_backup_list()
+                        for backup in full_backup_list:
+                            if backup.instance_id in instance_retention_map:
+                                retention_time = self.get_time_from_str(
+                                    instance_retention_map[backup.instance_id]
+                                )
+                                backup_age = datetime.now(timezone.utc) - backup.created_at
+                                if backup_age > retention_time:
+                                    # Backup remain longer than retention, need to purge it.
+                                    retention_backups.append(backup)
+                            elif threshold_strtime < backup.created_at:
+                                retention_backups.append(backup)
+
+                    if not retention_backups:
+                        return
+                    # 2. get project list
+                    self.controller.update_project_list()
+                    # 3. remove the backups
+                    self.remove_backups(retention_backups)
+            except coordination.LockAcquireFailed:
+                LOG.debug("Failed to lock for retention")
 
         periodic_callables = [
             (rotation_tasks, (), {}),
@@ -216,9 +244,9 @@
         periodic_thread.daemon = True
         periodic_thread.start()
 
-    # get the threshold time str
-    def get_threshold_strtime(self):
-        time_delta_dict = xtime.parse_timedelta_string(CONF.conductor.retention_time)
+    # get time
+    def get_time_from_str(self, time_str, to_str=False):
+        time_delta_dict = xtime.parse_timedelta_string(time_str)
         if time_delta_dict is None:
             LOG.info(
                 _(
@@ -237,4 +265,4 @@
             minutes=time_delta_dict["minutes"],
             seconds=time_delta_dict["seconds"],
         )
-        return res.strftime(xtime.DEFAULT_TIME_FORMAT)
+        return res.strftime(xtime.DEFAULT_TIME_FORMAT) if to_str else res
diff --git a/staffeln/conf/conductor.py b/staffeln/conf/conductor.py
index 6e5811d..5ab8e4b 100755
--- a/staffeln/conf/conductor.py
+++ b/staffeln/conf/conductor.py
@@ -39,6 +39,14 @@
         "backup_metadata_key",
         help=_("The key string of metadata the VM, which requres back up, has"),
     ),
+    cfg.StrOpt(
+        "retention_metadata_key",
+        regex=(
+            r"((?P<years>\d+?)y)?((?P<months>\d+?)mon)?((?P<weeks>\d+?)w)?"
+            r"((?P<days>\d+?)d)?((?P<hours>\d+?)h)?((?P<minutes>\d+?)min)?((?P<seconds>\d+?)s)?"
+        ),
+        help=_("The key string of metadata the VM, which use as backup retention period."),
+    ),
     cfg.IntOpt(
         "full_backup_depth",
         default=2,
diff --git a/staffeln/objects/volume.py b/staffeln/objects/volume.py
index 1f0bdb6..6b0e676 100644
--- a/staffeln/objects/volume.py
+++ b/staffeln/objects/volume.py
@@ -1,3 +1,4 @@
+from oslo_versionedobjects import fields as ovoo_fields
 from staffeln.db import api as db_api
 from staffeln.objects import base
 from staffeln.objects import fields as sfeild
@@ -19,6 +20,7 @@
         "volume_id": sfeild.UUIDField(),
         "backup_completed": sfeild.IntegerField(),
         "incremental": sfeild.BooleanField(),
+        "created_at": ovoo_fields.DateTimeField(),
     }
 
     @base.remotable_classmethod