diff --git a/agent/src/com/cloud/agent/resource/computing/LibvirtComputingResource.java b/agent/src/com/cloud/agent/resource/computing/LibvirtComputingResource.java
index 112ab61e50b..347ff87ac9d 100755
--- a/agent/src/com/cloud/agent/resource/computing/LibvirtComputingResource.java
+++ b/agent/src/com/cloud/agent/resource/computing/LibvirtComputingResource.java
@@ -30,12 +30,15 @@ import java.net.URI;
 import java.net.URISyntaxException;
 import java.net.URL;
 import java.net.URLConnection;
+import java.text.DateFormat;
 import java.text.MessageFormat;
+import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Calendar;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -154,6 +157,7 @@ import com.cloud.agent.resource.computing.KVMHABase.NfsStoragePool;
 import com.cloud.agent.resource.computing.LibvirtVMDef.ConsoleDef;
 import com.cloud.agent.resource.computing.LibvirtVMDef.DevicesDef;
 import com.cloud.agent.resource.computing.LibvirtVMDef.DiskDef;
+import com.cloud.agent.resource.computing.LibvirtVMDef.DiskDef.diskProtocol;
 import com.cloud.agent.resource.computing.LibvirtVMDef.FeaturesDef;
 import com.cloud.agent.resource.computing.LibvirtVMDef.GraphicDef;
 import com.cloud.agent.resource.computing.LibvirtVMDef.GuestDef;
@@ -1298,6 +1302,13 @@ public class LibvirtComputingResource extends ServerResourceBase implements
 
             KVMStoragePool primaryPool = _storagePoolMgr.getStoragePool(cmd
                     .getPool().getUuid());
+
+            if (primaryPool.getType() == StoragePoolType.RBD) {
+                s_logger.debug("Snapshots are not supported on RBD volumes");
+                return new ManageSnapshotAnswer(cmd, false,
+                    "Snapshots are not supported on RBD volumes");
+            }
+
             KVMPhysicalDisk disk = primaryPool.getPhysicalDisk(cmd
                     .getVolumePath());
             if (state == DomainInfo.DomainState.VIR_DOMAIN_RUNNING
@@ -1644,16 +1655,43 @@ public class LibvirtComputingResource extends ServerResourceBase implements
                     + templateInstallFolder;
             _storage.mkdirs(tmpltPath);
 
-            Script command = new Script(_createTmplPath, _cmdsTimeout, s_logger);
-            command.add("-f", disk.getPath());
-            command.add("-t", tmpltPath);
-            command.add("-n", cmd.getUniqueName() + ".qcow2");
+            if (primary.getType() != StoragePoolType.RBD) {
+                Script command = new Script(_createTmplPath, _cmdsTimeout, s_logger);
+                command.add("-f", disk.getPath());
+                command.add("-t", tmpltPath);
+                command.add("-n", cmd.getUniqueName() + ".qcow2");
 
-            String result = command.execute();
+                String result = command.execute();
 
-            if (result != null) {
-                s_logger.debug("failed to create template: " + result);
-                return new CreatePrivateTemplateAnswer(cmd, false, result);
+                if (result != null) {
+                    s_logger.debug("failed to create template: " + result);
+                    return new CreatePrivateTemplateAnswer(cmd, false, result);
+                }
+            } else {
+                s_logger.debug("Converting RBD disk " + disk.getPath() + " into template " + cmd.getUniqueName());
+                Script.runSimpleBashScript("qemu-img convert"
+                                + " -f raw -O qcow2 "
+                                + KVMPhysicalDisk.RBDStringBuilder(primary.getSourceHost(),
+                                                primary.getSourcePort(),
+                                                primary.getAuthUserName(),
+                                                primary.getAuthSecret(),
+                                                disk.getPath())
+                                + " " + tmpltPath + "/" + cmd.getUniqueName() + ".qcow2");
+                File templateProp = new File(tmpltPath + "/template.properties");
+                if (!templateProp.exists()) {
+                    templateProp.createNewFile();
+                }
+
+                String templateContent = "filename=" + cmd.getUniqueName() + ".qcow2" + System.getProperty("line.separator");
+
+                DateFormat dateFormat = new SimpleDateFormat("MM_dd_yyyy");
+                Date date = new Date();
+                templateContent += "snapshot.name=" + dateFormat.format(date) + System.getProperty("line.separator");
+
+                FileOutputStream templFo = new FileOutputStream(templateProp);
+                templFo.write(templateContent.getBytes());
+                templFo.flush();
+                templFo.close();
             }
 
             Map<String, Object> params = new HashMap<String, Object>();
@@ -1756,8 +1794,8 @@ public class LibvirtComputingResource extends ServerResourceBase implements
 
     protected Answer execute(ModifyStoragePoolCommand cmd) {
         KVMStoragePool storagepool = _storagePoolMgr.createStoragePool(cmd
-                .getPool().getUuid(), cmd.getPool().getHost(), cmd.getPool()
-                .getPath(), cmd.getPool().getType());
+                .getPool().getUuid(), cmd.getPool().getHost(), cmd.getPool().getPort(),
+                cmd.getPool().getPath(), cmd.getPool().getUserInfo(), cmd.getPool().getType());
         if (storagepool == null) {
             return new Answer(cmd, false, " Failed to create storage pool");
         }
@@ -2624,10 +2662,19 @@ public class LibvirtComputingResource extends ServerResourceBase implements
             } else {
                 int devId = (int) volume.getDeviceId();
 
-                if (volume.getType() == Volume.Type.DATADISK) {
-                    disk.defFileBasedDisk(physicalDisk.getPath(), devId,
-                            DiskDef.diskBus.VIRTIO,
-                            DiskDef.diskFmtType.QCOW2);
+                if (pool.getType() == StoragePoolType.RBD) {
+                                        /*
+                                                For RBD pools we use the secret mechanism in libvirt.
+                                                We store the secret under the UUID of the pool, that's why
+                                                we pass the pool's UUID as the authSecret
+                                        */
+                                        disk.defNetworkBasedDisk(physicalDisk.getPath().replace("rbd:", ""), pool.getSourceHost(), pool.getSourcePort(),
+                                                                 pool.getAuthUserName(), pool.getUuid(),
+                                                                 devId, diskBusType, diskProtocol.RBD);
+                                } else if (volume.getType() == Volume.Type.DATADISK) {
+                                        disk.defFileBasedDisk(physicalDisk.getPath(), devId,
+                                                        DiskDef.diskBus.VIRTIO,
+                                                        DiskDef.diskFmtType.QCOW2);
                 } else {
                     disk.defFileBasedDisk(physicalDisk.getPath(), devId,
                             diskBusType, DiskDef.diskFmtType.QCOW2);
@@ -2982,8 +3029,8 @@ public class LibvirtComputingResource extends ServerResourceBase implements
         try {
 
             KVMStoragePool localStoragePool = _storagePoolMgr
-                    .createStoragePool(_localStorageUUID, "localhost",
-                            _localStoragePath, StoragePoolType.Filesystem);
+                    .createStoragePool(_localStorageUUID, "localhost", -1,
+                            _localStoragePath, "", StoragePoolType.Filesystem);
             com.cloud.agent.api.StoragePoolInfo pi = new com.cloud.agent.api.StoragePoolInfo(
                     localStoragePool.getUuid(), cmd.getPrivateIpAddress(),
                     _localStoragePath, _localStoragePath,
@@ -4083,5 +4130,4 @@ public class LibvirtComputingResource extends ServerResourceBase implements
 
         return new Answer(cmd, success, "");
     }
-
 }
diff --git a/agent/src/com/cloud/agent/resource/computing/LibvirtSecretDef.java b/agent/src/com/cloud/agent/resource/computing/LibvirtSecretDef.java
new file mode 100644
index 00000000000..f7e10c38ddf
--- /dev/null
+++ b/agent/src/com/cloud/agent/resource/computing/LibvirtSecretDef.java
@@ -0,0 +1,106 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 com.cloud.agent.resource.computing;
+
+public class LibvirtSecretDef {
+
+	public enum usage {
+		VOLUME("volume"), CEPH("ceph");
+		String _usage;
+
+		usage(String usage) {
+			_usage = usage;
+		}
+
+		@Override
+		public String toString() {
+			return _usage;
+		}
+	}
+
+	private usage _usage;
+	private boolean _ephemeral;
+	private boolean _private;
+	private String _uuid;
+	private String _description;
+	private String _cephName;
+	private String _volumeVolume;
+
+	public LibvirtSecretDef (usage usage, String uuid) {
+		_usage = usage;
+		_uuid = uuid;
+	}
+
+	public LibvirtSecretDef (usage usage, String uuid, String description) {
+		_usage = usage;
+		_uuid = uuid;
+		_description = description;
+	}
+
+	public boolean getEphemeral() {
+		return _ephemeral;
+	}
+
+	public boolean getPrivate() {
+		return _private;
+	}
+
+	public String getUuid() {
+		return _uuid;
+	}
+
+	public String getDescription() {
+		return _description;
+	}
+
+	public String getVolumeVolume() {
+		return _volumeVolume;
+	}
+
+	public String getCephName() {
+		return _cephName;
+	}
+
+	public void setVolumeVolume(String volume) {
+		_volumeVolume = volume;
+	}
+
+	public void setCephName(String name) {
+		_cephName = name;
+	}
+
+	@Override
+	public String toString() {
+		StringBuilder secretBuilder = new StringBuilder();
+		secretBuilder.append("<secret ephemeral='" + (_ephemeral ? "yes" : "no") + "' private='" + (_private ? "yes" : "no") + "'>\n");
+		secretBuilder.append("<uuid>" + _uuid + "</uuid>\n");
+		if (_description != null) {
+			secretBuilder.append("<description>" + _description + "</description>\n");
+		}
+		secretBuilder.append("<usage type='" + _usage + "'>\n");
+		if (_usage == _usage.VOLUME) {
+			secretBuilder.append("<volume>" + _volumeVolume + "</volume>\n");
+		}
+		if (_usage == _usage.CEPH) {
+			secretBuilder.append("<name>" + _cephName + "</name>\n");
+		}
+		secretBuilder.append("</usage>\n");
+		secretBuilder.append("</secret>\n");
+		return secretBuilder.toString();
+	}
+	
+}
diff --git a/agent/src/com/cloud/agent/resource/computing/LibvirtStoragePoolDef.java b/agent/src/com/cloud/agent/resource/computing/LibvirtStoragePoolDef.java
index 582cd2e2e6d..9c285284563 100644
--- a/agent/src/com/cloud/agent/resource/computing/LibvirtStoragePoolDef.java
+++ b/agent/src/com/cloud/agent/resource/computing/LibvirtStoragePoolDef.java
@@ -18,7 +18,7 @@ package com.cloud.agent.resource.computing;
 
 public class LibvirtStoragePoolDef {
     public enum poolType {
-        ISCSI("iscsi"), NETFS("netfs"), LOGICAL("logical"), DIR("dir");
+        ISCSI("iscsi"), NETFS("netfs"), LOGICAL("logical"), DIR("dir"), RBD("rbd");
         String _poolType;
 
         poolType(String poolType) {
@@ -31,12 +31,41 @@ public class LibvirtStoragePoolDef {
         }
     }
 
+    public enum authType {
+        CHAP("chap"), CEPH("ceph");
+        String _authType;
+
+        authType(String authType) {
+            _authType = authType;
+        }
+
+        @Override
+        public String toString() {
+            return _authType;
+        }
+    }
+
     private poolType _poolType;
     private String _poolName;
     private String _uuid;
     private String _sourceHost;
+    private int _sourcePort;
     private String _sourceDir;
     private String _targetPath;
+    private String _authUsername;
+    private authType _authType;
+    private String _secretUuid;
+
+    public LibvirtStoragePoolDef(poolType type, String poolName, String uuid,
+            String host, int port, String dir, String targetPath) {
+        _poolType = type;
+        _poolName = poolName;
+        _uuid = uuid;
+        _sourceHost = host;
+        _sourcePort = port;
+        _sourceDir = dir;
+        _targetPath = targetPath;
+    }
 
     public LibvirtStoragePoolDef(poolType type, String poolName, String uuid,
             String host, String dir, String targetPath) {
@@ -48,6 +77,20 @@ public class LibvirtStoragePoolDef {
         _targetPath = targetPath;
     }
 
+    public LibvirtStoragePoolDef(poolType type, String poolName, String uuid,
+            String sourceHost, int sourcePort, String dir, String authUsername,
+            authType authType, String secretUuid) {
+        _poolType = type;
+        _poolName = poolName;
+        _uuid = uuid;
+        _sourceHost = sourceHost;
+        _sourcePort = sourcePort;
+        _sourceDir = dir;
+        _authUsername = authUsername;
+        _authType = authType;
+        _secretUuid = secretUuid;
+    }
+
     public String getPoolName() {
         return _poolName;
     }
@@ -60,6 +103,10 @@ public class LibvirtStoragePoolDef {
         return _sourceHost;
     }
 
+    public int getSourcePort() {
+        return _sourcePort;
+    }
+
     public String getSourceDir() {
         return _sourceDir;
     }
@@ -68,6 +115,18 @@ public class LibvirtStoragePoolDef {
         return _targetPath;
     }
 
+    public String getAuthUserName() {
+        return _authUsername;
+    }
+
+    public String getSecretUUID() {
+        return _secretUuid;
+    }
+
+    public authType getAuthType() {
+        return _authType;
+    }
+
     @Override
     public String toString() {
         StringBuilder storagePoolBuilder = new StringBuilder();
@@ -81,9 +140,22 @@ public class LibvirtStoragePoolDef {
             storagePoolBuilder.append("<dir path='" + _sourceDir + "'/>\n");
             storagePoolBuilder.append("</source>\n");
         }
-        storagePoolBuilder.append("<target>\n");
-        storagePoolBuilder.append("<path>" + _targetPath + "</path>\n");
-        storagePoolBuilder.append("</target>\n");
+        if (_poolType == poolType.RBD) {
+            storagePoolBuilder.append("<source>\n");
+            storagePoolBuilder.append("<host name='" + _sourceHost + "' port='" + _sourcePort + "'/>\n");
+            storagePoolBuilder.append("<name>" + _sourceDir + "</name>\n");
+            if (_authUsername != null) {
+                storagePoolBuilder.append("<auth username='" + _authUsername + "' type='" + _authType + "'>\n");
+                storagePoolBuilder.append("<secret uuid='" + _secretUuid + "'/>\n");
+                storagePoolBuilder.append("</auth>\n");
+            }
+            storagePoolBuilder.append("</source>\n");
+        }
+        if (_poolType != poolType.RBD) {
+            storagePoolBuilder.append("<target>\n");
+            storagePoolBuilder.append("<path>" + _targetPath + "</path>\n");
+            storagePoolBuilder.append("</target>\n");
+        }
         storagePoolBuilder.append("</pool>\n");
         return storagePoolBuilder.toString();
     }
diff --git a/agent/src/com/cloud/agent/resource/computing/LibvirtStoragePoolXMLParser.java b/agent/src/com/cloud/agent/resource/computing/LibvirtStoragePoolXMLParser.java
index 5c45d76e82d..cff4c2b74aa 100644
--- a/agent/src/com/cloud/agent/resource/computing/LibvirtStoragePoolXMLParser.java
+++ b/agent/src/com/cloud/agent/resource/computing/LibvirtStoragePoolXMLParser.java
@@ -51,15 +51,34 @@ public class LibvirtStoragePoolXMLParser {
             Element source = (Element) rootElement.getElementsByTagName(
                     "source").item(0);
             String host = getAttrValue("host", "name", source);
-            String path = getAttrValue("dir", "path", source);
 
-            Element target = (Element) rootElement.getElementsByTagName(
-                    "target").item(0);
-            String targetPath = getTagValue("path", target);
+            if (type.equalsIgnoreCase("rbd")) {
+                int port = Integer.parseInt(getAttrValue("host", "port", source));
+                String pool = getTagValue("name", source);
 
-            return new LibvirtStoragePoolDef(
-                    LibvirtStoragePoolDef.poolType.valueOf(type.toUpperCase()),
-                    poolName, uuid, host, path, targetPath);
+                Element auth = (Element) source.getElementsByTagName(
+                    "auth").item(0);
+
+                if (auth != null) {
+                    String authUsername = auth.getAttribute("username");
+                    String authType = auth.getAttribute("type");
+                    return new LibvirtStoragePoolDef(LibvirtStoragePoolDef.poolType.valueOf(type.toUpperCase()),
+                        poolName, uuid, host, port, pool, authUsername, LibvirtStoragePoolDef.authType.valueOf(authType.toUpperCase()), uuid);
+                } else {
+                    return new LibvirtStoragePoolDef(LibvirtStoragePoolDef.poolType.valueOf(type.toUpperCase()),
+                        poolName, uuid, host, port, pool, "");
+                }
+            } else {
+                String path = getAttrValue("dir", "path", source);
+
+                Element target = (Element) rootElement.getElementsByTagName(
+                        "target").item(0);
+                String targetPath = getTagValue("path", target);
+
+                return new LibvirtStoragePoolDef(
+                        LibvirtStoragePoolDef.poolType.valueOf(type.toUpperCase()),
+                        poolName, uuid, host, path, targetPath);
+            }
         } catch (ParserConfigurationException e) {
             s_logger.debug(e.toString());
         } catch (SAXException e) {
diff --git a/agent/src/com/cloud/agent/resource/computing/LibvirtVMDef.java b/agent/src/com/cloud/agent/resource/computing/LibvirtVMDef.java
index 8fd7815bf69..3b07bcd16b7 100644
--- a/agent/src/com/cloud/agent/resource/computing/LibvirtVMDef.java
+++ b/agent/src/com/cloud/agent/resource/computing/LibvirtVMDef.java
@@ -338,7 +338,7 @@ public class LibvirtVMDef {
         }
 
         enum diskType {
-            FILE("file"), BLOCK("block"), DIRECTROY("dir");
+            FILE("file"), BLOCK("block"), DIRECTROY("dir"), NETWORK("network");
             String _diskType;
 
             diskType(String type) {
@@ -351,6 +351,20 @@ public class LibvirtVMDef {
             }
         }
 
+        enum diskProtocol {
+            RBD("rbd"), SHEEPDOG("sheepdog");
+            String _diskProtocol;
+
+            diskProtocol(String protocol) {
+                _diskProtocol = protocol;
+            }
+
+            @Override
+            public String toString() {
+                return _diskProtocol;
+            }
+        }
+
         enum diskBus {
             IDE("ide"), SCSI("scsi"), VIRTIO("virtio"), XEN("xen"), USB("usb"), UML(
                     "uml"), FDC("fdc");
@@ -382,7 +396,12 @@ public class LibvirtVMDef {
 
         private deviceType _deviceType; /* floppy, disk, cdrom */
         private diskType _diskType;
+        private diskProtocol _diskProtocol;
         private String _sourcePath;
+        private String _sourceHost;
+        private int _sourcePort;
+        private String _authUserName;
+        private String _authSecretUUID;
         private String _diskLabel;
         private diskBus _bus;
         private diskFmtType _diskFmtType; /* qcow2, raw etc. */
@@ -461,6 +480,38 @@ public class LibvirtVMDef {
             _bus = bus;
         }
 
+        public void defNetworkBasedDisk(String diskName, String sourceHost, int sourcePort,
+                                        String authUserName, String authSecretUUID,
+                                        int devId, diskBus bus, diskProtocol protocol) {
+            _diskType = diskType.NETWORK;
+            _deviceType = deviceType.DISK;
+            _diskFmtType = diskFmtType.RAW;
+            _sourcePath = diskName;
+            _sourceHost = sourceHost;
+            _sourcePort = sourcePort;
+            _authUserName = authUserName;
+            _authSecretUUID = authSecretUUID;
+            _diskLabel = getDevLabel(devId, bus);
+            _bus = bus;
+            _diskProtocol = protocol;
+        }
+
+        public void defNetworkBasedDisk(String diskName, String sourceHost, int sourcePort,
+                                        String authUserName, String authSecretUUID,
+                                        String diskLabel, diskBus bus, diskProtocol protocol) {
+            _diskType = diskType.NETWORK;
+            _deviceType = deviceType.DISK;
+            _diskFmtType = diskFmtType.RAW;
+            _sourcePath = diskName;
+            _sourceHost = sourceHost;
+            _sourcePort = sourcePort;
+            _authUserName = authUserName;
+            _authSecretUUID = authSecretUUID;
+            _diskLabel = diskLabel;
+            _bus = bus;
+            _diskProtocol = protocol;
+        }
+
         public void setReadonly() {
             _readonly = true;
         }
@@ -527,6 +578,18 @@ public class LibvirtVMDef {
                     diskBuilder.append(" dev='" + _sourcePath + "'");
                 }
                 diskBuilder.append("/>\n");
+            } else if (_diskType == diskType.NETWORK) {
+                diskBuilder.append("<source ");
+                diskBuilder.append(" protocol='" + _diskProtocol + "'");
+                diskBuilder.append(" name='" + _sourcePath + "'");
+                diskBuilder.append(">\n");
+                diskBuilder.append("<host name='" + _sourceHost + "' port='" + _sourcePort + "'/>\n");
+                diskBuilder.append("</source>\n");
+                if (_authUserName != null) {
+                    diskBuilder.append("<auth username='" + _authUserName + "'>\n");
+                    diskBuilder.append("<secret type='ceph' uuid='" + _authSecretUUID + "'/>\n");
+                    diskBuilder.append("</auth>\n");
+                }
             }
             diskBuilder.append("<target dev='" + _diskLabel + "'");
             if (_bus != null) {
@@ -898,5 +961,4 @@ public class LibvirtVMDef {
 
         System.out.println(vm.toString());
     }
-
 }
diff --git a/agent/src/com/cloud/agent/storage/KVMPhysicalDisk.java b/agent/src/com/cloud/agent/storage/KVMPhysicalDisk.java
index d0e43df9e61..cb790d94620 100644
--- a/agent/src/com/cloud/agent/storage/KVMPhysicalDisk.java
+++ b/agent/src/com/cloud/agent/storage/KVMPhysicalDisk.java
@@ -34,6 +34,22 @@ public class KVMPhysicalDisk {
         }
     }
 
+    public static String RBDStringBuilder(String monHost, int monPort,
+                            String authUserName, String authSecret, String image) {
+        String rbdOpts;
+
+        rbdOpts = "rbd:" + image;
+        rbdOpts += ":mon_host=" + monHost + "\\\\:" + monPort;
+        if (authUserName == null) {
+            rbdOpts += ":auth_supported=none";
+        } else {
+            rbdOpts += ":auth_supported=cephx";
+            rbdOpts += ":id=" + authUserName;
+            rbdOpts += ":key=" + authSecret;
+        }
+        return rbdOpts;
+    }
+
     private PhysicalDiskFormat format;
     private long size;
     private long virtualSize;
diff --git a/agent/src/com/cloud/agent/storage/KVMStoragePool.java b/agent/src/com/cloud/agent/storage/KVMStoragePool.java
index 9961088bc02..f1192ede940 100644
--- a/agent/src/com/cloud/agent/storage/KVMStoragePool.java
+++ b/agent/src/com/cloud/agent/storage/KVMStoragePool.java
@@ -45,6 +45,16 @@ public interface KVMStoragePool {
 
     public String getLocalPath();
 
+    public String getSourceHost();
+
+    public String getSourceDir();
+
+    public int getSourcePort();
+
+    public String getAuthUserName();
+
+    public String getAuthSecret();
+
     public StoragePoolType getType();
 
     public boolean delete();
diff --git a/agent/src/com/cloud/agent/storage/KVMStoragePoolManager.java b/agent/src/com/cloud/agent/storage/KVMStoragePoolManager.java
index 53373e50b45..a82318d2c2d 100644
--- a/agent/src/com/cloud/agent/storage/KVMStoragePoolManager.java
+++ b/agent/src/com/cloud/agent/storage/KVMStoragePoolManager.java
@@ -52,10 +52,10 @@ public class KVMStoragePoolManager {
         return this._storageAdaptor.getStoragePoolByUri(uri);
     }
 
-    public KVMStoragePool createStoragePool(String name, String host,
-            String path, StoragePoolType type) {
+    public KVMStoragePool createStoragePool(String name, String host, int port, String path,
+                                            String userInfo, StoragePoolType type) {
         KVMStoragePool pool = this._storageAdaptor.createStoragePool(name,
-                host, path, type);
+                                host, port, path, userInfo, type);
         if (type == StoragePoolType.NetworkFilesystem) {
             KVMHABase.NfsStoragePool nfspool = new KVMHABase.NfsStoragePool(
                     pool.getUuid(), host, path, pool.getLocalPath(),
@@ -73,11 +73,16 @@ public class KVMStoragePoolManager {
         return true;
     }
 
-    public KVMPhysicalDisk createDiskFromTemplate(KVMPhysicalDisk template,
-            String name, KVMStoragePool destPool) {
-        return this._storageAdaptor.createDiskFromTemplate(template, name,
-                KVMPhysicalDisk.PhysicalDiskFormat.QCOW2,
-                template.getSize(), destPool);
+    public KVMPhysicalDisk createDiskFromTemplate(KVMPhysicalDisk template, String name,
+                                                    KVMStoragePool destPool) {
+        if (destPool.getType() == StoragePoolType.RBD) {
+            return this._storageAdaptor.createDiskFromTemplate(template, name,
+                    KVMPhysicalDisk.PhysicalDiskFormat.RAW, template.getSize(), destPool);
+        } else {
+            return this._storageAdaptor.createDiskFromTemplate(template, name,
+                    KVMPhysicalDisk.PhysicalDiskFormat.QCOW2,
+            template.getSize(), destPool);
+        }
     }
 
     public KVMPhysicalDisk createTemplateFromDisk(KVMPhysicalDisk disk,
diff --git a/agent/src/com/cloud/agent/storage/LibvirtStorageAdaptor.java b/agent/src/com/cloud/agent/storage/LibvirtStorageAdaptor.java
index e1e050f75d5..da000aaff19 100644
--- a/agent/src/com/cloud/agent/storage/LibvirtStorageAdaptor.java
+++ b/agent/src/com/cloud/agent/storage/LibvirtStorageAdaptor.java
@@ -23,8 +23,10 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
 import org.apache.log4j.Logger;
+import org.apache.commons.codec.binary.Base64;
 import org.libvirt.Connect;
 import org.libvirt.LibvirtException;
+import org.libvirt.Secret;
 import org.libvirt.StoragePool;
 import org.libvirt.StoragePoolInfo;
 import org.libvirt.StorageVol;
@@ -32,10 +34,13 @@ import org.libvirt.StoragePoolInfo.StoragePoolState;
 
 import com.cloud.agent.api.ManageSnapshotCommand;
 import com.cloud.agent.resource.computing.LibvirtConnection;
+import com.cloud.agent.resource.computing.LibvirtSecretDef;
+import com.cloud.agent.resource.computing.LibvirtSecretDef.usage;
 import com.cloud.agent.resource.computing.LibvirtStoragePoolDef;
 import com.cloud.agent.resource.computing.LibvirtStoragePoolXMLParser;
 import com.cloud.agent.resource.computing.LibvirtStorageVolumeDef;
 import com.cloud.agent.resource.computing.LibvirtStoragePoolDef.poolType;
+import com.cloud.agent.resource.computing.LibvirtStoragePoolDef.authType;
 import com.cloud.agent.resource.computing.LibvirtStorageVolumeDef.volFormat;
 import com.cloud.agent.resource.computing.LibvirtStorageVolumeXMLParser;
 import com.cloud.agent.storage.KVMPhysicalDisk.PhysicalDiskFormat;
@@ -143,7 +148,6 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
 
                 synchronized (getStoragePool(uuid)) {
                     sp = conn.storagePoolDefineXML(spd.toString(), 0);
-
                     if (sp == null) {
                         s_logger.debug("Failed to define storage pool");
                         return null;
@@ -270,6 +274,60 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
 
     }
 
+    private StoragePool createRBDStoragePool(Connect conn, String uuid,
+        String host, int port, String userInfo, String path) {
+
+        LibvirtStoragePoolDef spd;
+        StoragePool sp = null;
+
+        String[] userInfoTemp = userInfo.split(":");
+        if (userInfoTemp.length == 2) {
+            s_logger.debug("libvirt secret information found. id: " + userInfoTemp[0] + " secret: " + userInfoTemp[1]);
+            LibvirtSecretDef sd = new LibvirtSecretDef(usage.CEPH, uuid);
+
+            Secret s = null;
+
+            sd.setCephName(userInfoTemp[0]);
+
+            try {
+                s_logger.debug(sd.toString());
+                s = conn.secretDefineXML(sd.toString());
+                s.setValue(Base64.decodeBase64(userInfoTemp[1]));
+            } catch (LibvirtException e) {
+                s_logger.debug(e.toString());
+                if (s != null) {
+                    try {
+                        s.undefine();
+                        s.free();
+                    } catch (LibvirtException l) {
+                        s_logger.debug("Failed to define secret with: " + l.toString());
+                        }
+                    }
+            }
+            spd = new LibvirtStoragePoolDef(poolType.RBD, uuid, uuid, host, port, path, userInfoTemp[0], authType.CEPH, uuid);
+        } else {
+            spd = new LibvirtStoragePoolDef(poolType.RBD, uuid, uuid, host, port, path, "");
+        }
+
+        try {
+            s_logger.debug(spd.toString());
+            sp = conn.storagePoolDefineXML(spd.toString(), 0);
+            sp.create(0);
+            return sp;
+        } catch (LibvirtException e) {
+            s_logger.debug(e.toString());
+            if (sp != null) {
+                try {
+                    sp.undefine();
+                    sp.free();
+                } catch (LibvirtException l) {
+                    s_logger.debug("Failed to define RBD storage pool with: " + l.toString());
+                }
+            }
+            return null;
+        }
+    }
+
     public StorageVol copyVolume(StoragePool destPool,
             LibvirtStorageVolumeDef destVol, StorageVol srcVol, int timeout)
             throws LibvirtException {
@@ -422,11 +480,36 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
             if (spd.getPoolType() == LibvirtStoragePoolDef.poolType.NETFS
                     || spd.getPoolType() == LibvirtStoragePoolDef.poolType.DIR) {
                 type = StoragePoolType.Filesystem;
+            } else if (spd.getPoolType() == LibvirtStoragePoolDef.poolType.RBD) {
+                type = StoragePoolType.RBD;
             }
-            LibvirtStoragePool pool = new LibvirtStoragePool(uuid,
-                    storage.getName(), type, this, storage);
-            pool.setLocalPath(spd.getTargetPath());
-            getStats(pool);
+
+            LibvirtStoragePool pool = new LibvirtStoragePool(uuid, storage.getName(),
+                                                            type, this, storage);
+
+            if (pool.getType() != StoragePoolType.RBD) {
+                pool.setLocalPath(spd.getTargetPath());
+            } else {
+                pool.setLocalPath("");
+                pool.setSourceHost(spd.getSourceHost());
+                pool.setSourcePort(spd.getSourcePort());
+                pool.setSourceDir(spd.getSourceDir());
+                String authUsername = spd.getAuthUserName();
+                if (authUsername != null) {
+                    Secret secret = conn.secretLookupByUUIDString(spd.getSecretUUID());
+                    String secretValue = new String(Base64.encodeBase64(secret.getByteValue()));
+                    pool.setAuthUsername(authUsername);
+                    pool.setAuthSecret(secretValue);
+                }
+            }
+
+            if (pool.getType() == StoragePoolType.RBD) {
+                pool.setCapacity(storage.getInfo().capacity);
+                pool.setUsed(storage.getInfo().allocation);
+            } else {
+                getStats(pool);
+            }
+
             return pool;
         } catch (LibvirtException e) {
             throw new CloudRuntimeException(e.toString());
@@ -448,6 +531,8 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
             disk.setVirtualSize(vol.getInfo().capacity);
             if (voldef.getFormat() == null) {
                 disk.setFormat(pool.getDefaultFormat());
+            } else if (pool.getType() == StoragePoolType.RBD) {
+                disk.setFormat(KVMPhysicalDisk.PhysicalDiskFormat.RAW);
             } else if (voldef.getFormat() == LibvirtStorageVolumeDef.volFormat.QCOW2) {
                 disk.setFormat(KVMPhysicalDisk.PhysicalDiskFormat.QCOW2);
             } else if (voldef.getFormat() == LibvirtStorageVolumeDef.volFormat.RAW) {
@@ -461,8 +546,8 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
     }
 
     @Override
-    public KVMStoragePool createStoragePool(String name, String host,
-            String path, StoragePoolType type) {
+    public KVMStoragePool createStoragePool(String name, String host, int port,
+                                            String path, String userInfo, StoragePoolType type) {
         StoragePool sp = null;
         Connect conn = null;
         try {
@@ -487,6 +572,8 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
             } else if (type == StoragePoolType.SharedMountPoint
                     || type == StoragePoolType.Filesystem) {
                 sp = CreateSharedStoragePool(conn, name, host, path);
+            } else if (type == StoragePoolType.RBD) {
+                sp = createRBDStoragePool(conn, name, host, port, userInfo, path);
             }
         }
 
@@ -499,15 +586,23 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
             LibvirtStoragePoolDef spd = getStoragePoolDef(conn, sp);
             LibvirtStoragePool pool = new LibvirtStoragePool(name,
                     sp.getName(), type, this, sp);
-            pool.setLocalPath(spd.getTargetPath());
 
-            getStats(pool);
-            
+            if (pool.getType() != StoragePoolType.RBD) {
+                pool.setLocalPath(spd.getTargetPath());
+            } else {
+                pool.setLocalPath("");
+            }
+
+            if (pool.getType() == StoragePoolType.RBD) {
+                pool.setCapacity(sp.getInfo().capacity);
+                pool.setUsed(sp.getInfo().allocation);
+            } else {
+                getStats(pool);
+            }
             return pool;
         } catch (LibvirtException e) {
             throw new CloudRuntimeException(e.toString());
         }
-
     }
 
     @Override
@@ -520,6 +615,7 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
         }
 
         StoragePool sp = null;
+        Secret s = null;
 
         try {
             sp = conn.storagePoolLookupByUUIDString(uuid);
@@ -527,10 +623,23 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
             return true;
         }
 
+        /*
+         * Some storage pools, like RBD also have 'secret' information stored in libvirt
+         * Destroy them if they exist
+        */
+        try {
+            s = conn.secretLookupByUUIDString(uuid);
+        } catch (LibvirtException e) {
+        }
+
         try {
             sp.destroy();
             sp.undefine();
             sp.free();
+            if (s != null) {
+                s.undefine();
+                s.free();
+            }
             return true;
         } catch (LibvirtException e) {
             throw new CloudRuntimeException(e.toString());
@@ -543,6 +652,11 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
         LibvirtStoragePool libvirtPool = (LibvirtStoragePool) pool;
         StoragePool virtPool = libvirtPool.getPool();
         LibvirtStorageVolumeDef.volFormat libvirtformat = null;
+
+        if (pool.getType() == StoragePoolType.RBD) {
+            format = PhysicalDiskFormat.RAW;
+        }
+
         if (format == PhysicalDiskFormat.QCOW2) {
             libvirtformat = LibvirtStorageVolumeDef.volFormat.QCOW2;
         } else if (format == PhysicalDiskFormat.RAW) {
@@ -580,19 +694,58 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
 
     @Override
     public KVMPhysicalDisk createDiskFromTemplate(KVMPhysicalDisk template,
-            String name, PhysicalDiskFormat format, long size,
-            KVMStoragePool destPool) {
-        KVMPhysicalDisk disk = destPool.createPhysicalDisk(UUID.randomUUID()
-                .toString(), format, template.getVirtualSize());
+            String name, PhysicalDiskFormat format, long size, KVMStoragePool destPool) {
+
+        String newUuid = UUID.randomUUID().toString();
+        KVMStoragePool srcPool = template.getPool();
+        KVMPhysicalDisk disk = null;
+
+        /*
+            With RBD you can't run qemu-img convert with an existing RBD image as destination
+            qemu-img will exit with the error that the destination already exists.
+            So for RBD we don't create the image, but let qemu-img do that for us.
+
+            We then create a KVMPhysicalDisk object that we can return
+        */
+
+        if (destPool.getType() != StoragePoolType.RBD) {
+            disk = destPool.createPhysicalDisk(newUuid, format, template.getVirtualSize());
 
-        if (format == PhysicalDiskFormat.QCOW2) {
             Script.runSimpleBashScript("qemu-img create -f "
                     + template.getFormat() + " -b  " + template.getPath() + " "
                     + disk.getPath());
-        } else if (format == PhysicalDiskFormat.RAW) {
-            Script.runSimpleBashScript("qemu-img convert -f "
-                    + template.getFormat() + " -O raw " + template.getPath()
-                    + " " + disk.getPath());
+        } else {
+            disk = new KVMPhysicalDisk(destPool.getSourceDir() + "/" + newUuid, newUuid, destPool);
+            disk.setFormat(format);
+            disk.setSize(template.getVirtualSize());
+            disk.setVirtualSize(disk.getSize());
+
+            if (srcPool.getType() != StoragePoolType.RBD) {
+                Script.runSimpleBashScript("qemu-img convert"
+                        + " -f " + template.getFormat()
+                        + " -O " + format
+                        + " " + template.getPath()
+                        + " " + KVMPhysicalDisk.RBDStringBuilder(destPool.getSourceHost(),
+                                                destPool.getSourcePort(),
+                                                destPool.getAuthUserName(),
+                                                destPool.getAuthSecret(),
+                                                disk.getPath()));
+            } else {
+                template.setFormat(PhysicalDiskFormat.RAW);
+                Script.runSimpleBashScript("qemu-img convert"
+                        + " -f " + template.getFormat()
+                        + " -O " + format
+                        + " " + KVMPhysicalDisk.RBDStringBuilder(srcPool.getSourceHost(),
+                                                srcPool.getSourcePort(),
+                                                srcPool.getAuthUserName(),
+                                                srcPool.getAuthSecret(),
+                                                template.getPath())
+                        + " " + KVMPhysicalDisk.RBDStringBuilder(destPool.getSourceHost(),
+                                                destPool.getSourcePort(),
+                                                destPool.getAuthUserName(),
+                                                destPool.getAuthSecret(),
+                                                disk.getPath()));
+            }
         }
         return disk;
     }
@@ -625,14 +778,60 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
     @Override
     public KVMPhysicalDisk copyPhysicalDisk(KVMPhysicalDisk disk, String name,
             KVMStoragePool destPool) {
-        KVMPhysicalDisk newDisk = destPool.createPhysicalDisk(name,
-                disk.getVirtualSize());
-        String sourcePath = disk.getPath();
-        String destPath = newDisk.getPath();
 
-        Script.runSimpleBashScript("qemu-img convert -f " + disk.getFormat()
-                + " -O " + newDisk.getFormat() + " " + sourcePath + " "
-                + destPath);
+        /*
+            With RBD you can't run qemu-img convert with an existing RBD image as destination
+            qemu-img will exit with the error that the destination already exists.
+            So for RBD we don't create the image, but let qemu-img do that for us.
+
+            We then create a KVMPhysicalDisk object that we can return
+        */
+
+        KVMPhysicalDisk newDisk;
+        if (destPool.getType() != StoragePoolType.RBD) {
+            newDisk = destPool.createPhysicalDisk(name, disk.getVirtualSize());
+        } else {
+            newDisk = new KVMPhysicalDisk(destPool.getSourceDir() + "/" + name, name, destPool);
+            newDisk.setFormat(PhysicalDiskFormat.RAW);
+            newDisk.setSize(disk.getVirtualSize());
+            newDisk.setVirtualSize(disk.getSize());
+        }
+
+        KVMStoragePool srcPool = disk.getPool();
+        String destPath = newDisk.getPath();
+        String sourcePath = disk.getPath();
+        PhysicalDiskFormat sourceFormat = disk.getFormat();
+        PhysicalDiskFormat destFormat = newDisk.getFormat();
+
+        if ((srcPool.getType() != StoragePoolType.RBD) && (destPool.getType() != StoragePoolType.RBD)) {
+            Script.runSimpleBashScript("qemu-img convert -f " + sourceFormat
+                + " -O " + destFormat
+                + " " + sourcePath
+                + " " + destPath);
+        } else if ((srcPool.getType() != StoragePoolType.RBD) && (destPool.getType() == StoragePoolType.RBD))  {
+            Script.runSimpleBashScript("qemu-img convert -f " + sourceFormat
+                    + " -O " + destFormat
+                    + " " + sourcePath
+                    + " " + KVMPhysicalDisk.RBDStringBuilder(destPool.getSourceHost(),
+                                                destPool.getSourcePort(),
+                                                destPool.getAuthUserName(),
+                                                destPool.getAuthSecret(),
+                                                destPath));
+        } else {
+            Script.runSimpleBashScript("qemu-img convert -f " + sourceFormat
+                    + " -O " + destFormat
+                    + " " + KVMPhysicalDisk.RBDStringBuilder(srcPool.getSourceHost(),
+                                                srcPool.getSourcePort(),
+                                                srcPool.getAuthUserName(),
+                                                srcPool.getAuthSecret(),
+                                                sourcePath)
+                    + " " + KVMPhysicalDisk.RBDStringBuilder(destPool.getSourceHost(),
+                                                destPool.getSourcePort(),
+                                                destPool.getAuthUserName(),
+                                                destPool.getAuthSecret(),
+                                                destPath));
+        }
+
         return newDisk;
     }
 
@@ -658,7 +857,7 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
             protocal = StoragePoolType.NetworkFilesystem;
         }
 
-        return createStoragePool(uuid, sourceHost, sourcePath, protocal);
+            return createStoragePool(uuid, sourceHost, 0, sourcePath, "", protocal);
     }
 
     @Override
@@ -699,5 +898,4 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
 
         return true;
     }
-
 }
diff --git a/agent/src/com/cloud/agent/storage/LibvirtStoragePool.java b/agent/src/com/cloud/agent/storage/LibvirtStoragePool.java
index d638bb99c59..943b8b268fd 100644
--- a/agent/src/com/cloud/agent/storage/LibvirtStoragePool.java
+++ b/agent/src/com/cloud/agent/storage/LibvirtStoragePool.java
@@ -34,6 +34,11 @@ public class LibvirtStoragePool implements KVMStoragePool {
     protected StoragePoolType type;
     protected StorageAdaptor _storageAdaptor;
     protected StoragePool _pool;
+    protected String authUsername;
+    protected String authSecret;
+    protected String sourceHost;
+    protected int sourcePort;
+    protected String sourceDir;
 
     public LibvirtStoragePool(String uuid, String name, StoragePoolType type,
             StorageAdaptor adaptor, StoragePool pool) {
@@ -137,6 +142,51 @@ public class LibvirtStoragePool implements KVMStoragePool {
         this.localPath = localPath;
     }
 
+    @Override
+    public String getAuthUserName() {
+        return this.authUsername;
+    }
+
+    public void setAuthUsername(String authUsername) {
+        this.authUsername = authUsername;
+    }
+
+    @Override
+    public String getAuthSecret() {
+        return this.authSecret;
+    }
+
+    public void setAuthSecret(String authSecret) {
+        this.authSecret = authSecret;
+    }
+
+    @Override
+    public String getSourceHost() {
+        return this.sourceHost;
+    }
+
+    public void setSourceHost(String host) {
+        this.sourceHost = host;
+    }
+
+    @Override
+    public int getSourcePort() {
+        return this.sourcePort;
+    }
+
+    public void setSourcePort(int port) {
+        this.sourcePort = port;
+    }
+
+    @Override
+    public String getSourceDir() {
+        return this.sourceDir;
+    }
+
+    public void setSourceDir(String dir) {
+        this.sourceDir = dir;
+    }
+
     @Override
     public StoragePoolType getType() {
         return this.type;
diff --git a/agent/src/com/cloud/agent/storage/StorageAdaptor.java b/agent/src/com/cloud/agent/storage/StorageAdaptor.java
index 9cc5d14d875..82386772ba4 100644
--- a/agent/src/com/cloud/agent/storage/StorageAdaptor.java
+++ b/agent/src/com/cloud/agent/storage/StorageAdaptor.java
@@ -30,8 +30,8 @@ public interface StorageAdaptor {
     public KVMPhysicalDisk getPhysicalDisk(String volumeUuid,
             KVMStoragePool pool);
 
-    public KVMStoragePool createStoragePool(String name, String host,
-            String path, StoragePoolType type);
+    public KVMStoragePool createStoragePool(String name, String host, int port,
+            String path, String userInfo, StoragePoolType type);
 
     public boolean deleteStoragePool(String uuid);
 
diff --git a/api/src/com/cloud/agent/api/to/StorageFilerTO.java b/api/src/com/cloud/agent/api/to/StorageFilerTO.java
index c0be269b11f..26344daaeb6 100644
--- a/api/src/com/cloud/agent/api/to/StorageFilerTO.java
+++ b/api/src/com/cloud/agent/api/to/StorageFilerTO.java
@@ -24,6 +24,7 @@ public class StorageFilerTO {
     String uuid;
     String host;
     String path;
+    String userInfo;
     int port;
     StoragePoolType type;
     
@@ -34,6 +35,7 @@ public class StorageFilerTO {
         this.path = pool.getPath();
         this.type = pool.getPoolType();
         this.uuid = pool.getUuid();
+        this.userInfo = pool.getUserInfo();
     }
 
     public long getId() {
@@ -52,6 +54,10 @@ public class StorageFilerTO {
         return path;
     }
 
+    public String getUserInfo() {
+        return userInfo;
+    }
+
     public int getPort() {
         return port;
     }
diff --git a/api/src/com/cloud/storage/Storage.java b/api/src/com/cloud/storage/Storage.java
index 856f79dd390..d6bd04ea61e 100755
--- a/api/src/com/cloud/storage/Storage.java
+++ b/api/src/com/cloud/storage/Storage.java
@@ -96,6 +96,7 @@ public class Storage {
         Iscsi(true), // for e.g., ZFS Comstar
         ISO(false), // for iso image
         LVM(false), // XenServer local LVM SR
+        RBD(true),
         SharedMountPoint(true),
         VMFS(true), // VMware VMFS storage
         PreSetup(true), // for XenServer, Storage Pool is set up by customers.
diff --git a/api/src/com/cloud/storage/StoragePool.java b/api/src/com/cloud/storage/StoragePool.java
index 582898608e0..ff38c54da05 100644
--- a/api/src/com/cloud/storage/StoragePool.java
+++ b/api/src/com/cloud/storage/StoragePool.java
@@ -85,6 +85,11 @@ public interface StoragePool {
      */
     String getPath();
 
+    /**
+     * @return the user information / credentials for the storage host
+     */
+    String getUserInfo();
+
     /**
      * @return the storage pool represents a shared storage resource
      */
diff --git a/core/src/com/cloud/storage/StoragePoolVO.java b/core/src/com/cloud/storage/StoragePoolVO.java
index b1f02019af1..3ca4e6c1434 100644
--- a/core/src/com/cloud/storage/StoragePoolVO.java
+++ b/core/src/com/cloud/storage/StoragePoolVO.java
@@ -157,6 +157,9 @@ public class StoragePoolVO implements StoragePool, Identity {
     @Column(name="port")
     private int port;
 
+    @Column(name="user_info")
+    private String userInfo;
+
     @Column(name="cluster_id")
     private Long clusterId;
     
@@ -179,6 +182,11 @@ public class StoragePoolVO implements StoragePool, Identity {
     public String getPath() {
         return path;
     }
+
+    @Override
+    public String getUserInfo() {
+        return userInfo;
+    }
     
     public StoragePoolVO(long poolId, String name, String uuid, StoragePoolType type,
             long dataCenterId, Long podId, long availableBytes, long capacityBytes, String hostAddress, int port, String hostPath) {
@@ -208,6 +216,16 @@ public class StoragePoolVO implements StoragePool, Identity {
         this.setStatus(StoragePoolStatus.Up);
         this.uuid = UUID.randomUUID().toString();
     }
+
+    public StoragePoolVO(StoragePoolType type, String hostAddress, int port, String path, String userInfo) {
+        this.poolType = type;
+        this.hostAddress = hostAddress;
+        this.port = port;
+        this.path = path;
+        this.userInfo = userInfo;
+        this.setStatus(StoragePoolStatus.Up);
+        this.uuid = UUID.randomUUID().toString();
+    }
     
     public void setStatus(StoragePoolStatus status)
     {
@@ -233,6 +251,10 @@ public class StoragePoolVO implements StoragePool, Identity {
     public void setPath(String path) {
     	this.path = path;
     }
+
+    public void setUserInfo(String userInfo) {
+        this.userInfo = userInfo;
+    }
     
     @Override
     public int getPort() {
diff --git a/server/src/com/cloud/storage/StorageManagerImpl.java b/server/src/com/cloud/storage/StorageManagerImpl.java
index ed9df3069ff..a2ea98d872b 100755
--- a/server/src/com/cloud/storage/StorageManagerImpl.java
+++ b/server/src/com/cloud/storage/StorageManagerImpl.java
@@ -1277,10 +1277,10 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
                 if (uriPath == null) {
                     throw new InvalidParameterValueException("host or path is null, should be sharedmountpoint://localhost/path");
                 }
-            } else if (uri.getScheme().equalsIgnoreCase("clvm")) {
+            }  else if (uri.getScheme().equalsIgnoreCase("rbd")) {
                 String uriPath = uri.getPath();
                 if (uriPath == null) {
-                    throw new InvalidParameterValueException("host or path is null, should be clvm://localhost/path");
+                    throw new InvalidParameterValueException("host or path is null, should be rbd://hostname/pool");
                 }
             }
         } catch (URISyntaxException e) {
@@ -1303,6 +1303,7 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
         String scheme = uri.getScheme();
         String storageHost = uri.getHost();
         String hostPath = uri.getPath();
+        String userInfo = uri.getUserInfo();
         int port = uri.getPort();
         StoragePoolVO pool = null;
         if (s_logger.isDebugEnabled()) {
@@ -1323,6 +1324,11 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
             pool = new StoragePoolVO(StoragePoolType.Filesystem, "localhost", 0, hostPath);
         } else if (scheme.equalsIgnoreCase("sharedMountPoint")) {
             pool = new StoragePoolVO(StoragePoolType.SharedMountPoint, storageHost, 0, hostPath);
+        } else if (scheme.equalsIgnoreCase("rbd")) {
+            if (port == -1) {
+                port = 6789;
+            }
+            pool = new StoragePoolVO(StoragePoolType.RBD, storageHost, port, hostPath.replaceFirst("/", ""), userInfo);
         } else if (scheme.equalsIgnoreCase("PreSetup")) {
             pool = new StoragePoolVO(StoragePoolType.PreSetup, storageHost, 0, hostPath);
         } else if (scheme.equalsIgnoreCase("iscsi")) {
@@ -1621,7 +1627,7 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
         s_logger.debug("creating pool " + pool.getName() + " on  host " + hostId);
         if (pool.getPoolType() != StoragePoolType.NetworkFilesystem && pool.getPoolType() != StoragePoolType.Filesystem && pool.getPoolType() != StoragePoolType.IscsiLUN
                 && pool.getPoolType() != StoragePoolType.Iscsi && pool.getPoolType() != StoragePoolType.VMFS && pool.getPoolType() != StoragePoolType.SharedMountPoint
-                && pool.getPoolType() != StoragePoolType.PreSetup && pool.getPoolType() != StoragePoolType.OCFS2) {
+                && pool.getPoolType() != StoragePoolType.PreSetup && pool.getPoolType() != StoragePoolType.OCFS2 && pool.getPoolType() != StoragePoolType.RBD) {
             s_logger.warn(" Doesn't support storage pool type " + pool.getPoolType());
             return false;
         }
diff --git a/server/src/com/cloud/storage/allocator/FirstFitStoragePoolAllocator.java b/server/src/com/cloud/storage/allocator/FirstFitStoragePoolAllocator.java
index 18c8cdeb466..006931d4814 100644
--- a/server/src/com/cloud/storage/allocator/FirstFitStoragePoolAllocator.java
+++ b/server/src/com/cloud/storage/allocator/FirstFitStoragePoolAllocator.java
@@ -31,18 +31,25 @@ import org.apache.log4j.Logger;
 import com.cloud.deploy.DeploymentPlan;
 import com.cloud.deploy.DeploymentPlanner.ExcludeList;
 import com.cloud.server.StatsCollector;
+import com.cloud.storage.DiskOfferingVO;
+import com.cloud.storage.dao.DiskOfferingDao;
 import com.cloud.storage.StoragePool;
 import com.cloud.storage.StoragePoolVO;
 import com.cloud.storage.VMTemplateVO;
+import com.cloud.storage.Storage.StoragePoolType;
 import com.cloud.user.Account;
 import com.cloud.vm.DiskProfile;
 import com.cloud.vm.VirtualMachine;
 import com.cloud.vm.VirtualMachineProfile;
+import com.cloud.utils.component.Inject;
 
 @Local(value=StoragePoolAllocator.class)
 public class FirstFitStoragePoolAllocator extends AbstractStoragePoolAllocator {
     private static final Logger s_logger = Logger.getLogger(FirstFitStoragePoolAllocator.class);
     protected String _allocationAlgorithm = "random";
+
+    @Inject
+    DiskOfferingDao _diskOfferingDao;
     
     @Override
     public boolean allocatorIsCorrectType(DiskProfile dskCh) {
@@ -97,10 +104,16 @@ public class FirstFitStoragePoolAllocator extends AbstractStoragePoolAllocator {
             s_logger.debug("FirstFitStoragePoolAllocator has " + pools.size() + " pools to check for allocation");
         }
     	
+        DiskOfferingVO diskOffering = _diskOfferingDao.findById(dskCh.getDiskOfferingId());
         for (StoragePoolVO pool: pools) {
         	if(suitablePools.size() == returnUpTo){
         		break;
         	}
+            if (diskOffering.getSystemUse() && pool.getPoolType() == StoragePoolType.RBD) {
+                s_logger.debug("Skipping RBD pool " + pool.getName() + " as a suitable pool. RBD is not supported for System VM's");
+                continue;
+            }
+
         	if (checkPool(avoid, pool, dskCh, template, null, sc, plan)) {
         		suitablePools.add(pool);
         	}
diff --git a/server/src/com/cloud/storage/snapshot/SnapshotManagerImpl.java b/server/src/com/cloud/storage/snapshot/SnapshotManagerImpl.java
index 50dcf384205..6e3f9c14116 100755
--- a/server/src/com/cloud/storage/snapshot/SnapshotManagerImpl.java
+++ b/server/src/com/cloud/storage/snapshot/SnapshotManagerImpl.java
@@ -79,6 +79,7 @@ import com.cloud.storage.SnapshotPolicyVO;
 import com.cloud.storage.SnapshotScheduleVO;
 import com.cloud.storage.SnapshotVO;
 import com.cloud.storage.Storage;
+import com.cloud.storage.Storage.StoragePoolType;
 import com.cloud.storage.StorageManager;
 import com.cloud.storage.StoragePool;
 import com.cloud.storage.StoragePoolVO;
@@ -298,6 +299,13 @@ public class SnapshotManagerImpl implements SnapshotManager, SnapshotService, Ma
             }
         }
         StoragePoolVO srcPool = _storagePoolDao.findById(volume.getPoolId());
+
+        // RBD volumes do not support snapshotting in the way CloudStack does it.
+        // For now we leave the snapshot feature disabled for RBD volumes
+        if (srcPool.getPoolType() == StoragePoolType.RBD) {
+            throw new CloudRuntimeException("RBD volumes do not support snapshotting");
+        }
+
         ManageSnapshotCommand cmd = new ManageSnapshotCommand(snapshotId, volume.getPath(), srcPool, preSnapshotPath, snapshot.getName(), vmName);
       
         ManageSnapshotAnswer answer = (ManageSnapshotAnswer) sendToPool(volume, cmd);
diff --git a/setup/db/create-schema.sql b/setup/db/create-schema.sql
index a15222aa203..ae5f58160a4 100755
--- a/setup/db/create-schema.sql
+++ b/setup/db/create-schema.sql
@@ -1477,6 +1477,7 @@ CREATE TABLE  `cloud`.`storage_pool` (
   `available_bytes` bigint unsigned,
   `capacity_bytes` bigint unsigned,
   `host_address` varchar(255) NOT NULL COMMENT 'FQDN or IP of storage server',
+  `user_info` varchar(255) NULL COMMENT 'Authorization information for the storage pool. Used by network filesystems',
   `path` varchar(255) NOT NULL COMMENT 'Filesystem path that is shared',
   `created` datetime COMMENT 'date the pool created',
   `removed` datetime COMMENT 'date removed if not null',
diff --git a/ui/index.jsp b/ui/index.jsp
index 1ee8f0bc1b9..94fbdde88bd 100644
--- a/ui/index.jsp
+++ b/ui/index.jsp
@@ -2046,6 +2046,7 @@ dictionary = {
 'label.SR.name ': '<fmt:message key="label.SR.name " />',
 'label.SharedMountPoint': '<fmt:message key="label.SharedMountPoint" />',
 'label.clvm': '<fmt:message key="label.clvm" />',
+'label.rbd': '<fmt:message key="label.rbd" />',
 'label.volgroup': '<fmt:message key="label.volgroup" />',
 'label.VMFS.datastore': '<fmt:message key="label.VMFS.datastore" />',
 'label.network.device': '<fmt:message key="label.network.device" />',
diff --git a/ui/scripts/sharedFunctions.js b/ui/scripts/sharedFunctions.js
index 12d3498350c..c4cf58c86e5 100644
--- a/ui/scripts/sharedFunctions.js
+++ b/ui/scripts/sharedFunctions.js
@@ -475,6 +475,31 @@ function SharedMountPointURL(server, path) {
 	return url;
 }
 
+function rbdURL(monitor, pool, id, secret) {
+	var url;
+
+	/*
+	Replace the + and / symbols by - and _ to have URL-safe base64 going to the API
+	It's hacky, but otherwise we'll confuse java.net.URI which splits the incoming URI
+	*/
+	secret = str.replace("+", "-");
+	secret = str.replace("/", "_");
+
+	if (id != null && secret != null) {
+		monitor = id + ":" + secret + "@" + monitor;
+	}
+
+	if(pool.substring(0,1) != "/")
+		pool = "/" + pool;
+
+	if(monitor.indexOf("://")==-1)
+		url = "rbd://" + monitor + pool;
+	else
+		url = monitor + pool;
+
+	return url;
+}
+
 function clvmURL(vgname) {
 	var url;
 	if(vgname.indexOf("://")==-1)
diff --git a/ui/scripts/system.js b/ui/scripts/system.js
index 712c122f68a..59d2d8703d9 100644
--- a/ui/scripts/system.js
+++ b/ui/scripts/system.js
@@ -7494,6 +7494,7 @@
                         var items = [];
                         items.push({id: "nfs", description: "nfs"});
                         items.push({id: "SharedMountPoint", description: "SharedMountPoint"});
+                        items.push({id: "rbd", description: "RBD"});
                         args.response.success({data: items});
                       }
                       else if(selectedClusterObj.hypervisortype == "XenServer") {
@@ -7677,6 +7678,27 @@
                           $form.find('.form-item[rel=vCenterDataCenter]').hide();
                           $form.find('.form-item[rel=vCenterDataStore]').hide();
                         }
+                        else if(protocol == "rbd") {
+                          $form.find('.form-item[rel=rbdmonitor]').css('display', 'inline-block');
+                          $form.find('.form-item[rel=rbdmonitor]').find(".name").find("label").text("RADOS Monitor:");
+
+                          $form.find('.form-item[rel=rbdpool]').css('display', 'inline-block');
+                          $form.find('.form-item[rel=rbdpool]').find(".name").find("label").text("RADOS Pool:");
+
+                          $form.find('.form-item[rel=rbdid]').css('display', 'inline-block');
+                          $form.find('.form-item[rel=rbdid]').find(".name").find("label").text("RADOS User:");
+
+                          $form.find('.form-item[rel=rbdsecret]').css('display', 'inline-block');
+                          $form.find('.form-item[rel=rbdsecret]').find(".name").find("label").text("RADOS Secret:");
+
+                          $form.find('.form-item[rel=server]').hide();
+                          $form.find('.form-item[rel=iqn]').hide();
+                          $form.find('.form-item[rel=lun]').hide();
+                          $form.find('.form-item[rel=volumegroup]').hide();
+                          $form.find('.form-item[rel=path]').hide();
+                          $form.find('.form-item[rel=vCenterDataCenter]').hide();
+                          $form.find('.form-item[rel=vCenterDataStore]').hide();
+                        }
                         else {
                           //$dialogAddPool.find("#add_pool_server_container").show();
                           $form.find('.form-item[rel=server]').css('display', 'inline-block');
@@ -7744,6 +7766,28 @@
                     validation: { required: true },
                     isHidden: true
                   },
+ 
+                  // RBD
+                  rbdmonitor: {
+                    label: 'label.rbd.monitor',
+                    validation: { required: true },
+                    isHidden: true
+                  },
+                  rbdpool: {
+                    label: 'label.rbd.pool',
+                    validation: { required: true },
+                    isHidden: true
+                  },
+                   rbdid: {
+                    label: 'label.rbd.id',
+                    validation: { required: false },
+                    isHidden: true
+                  },
+                   rbdsecret: {
+                    label: 'label.rbd.secret',
+                    validation: { required: false },
+                    isHidden: true
+                  },
 
                   //always appear (begin)
                   storageTags: {
@@ -7803,6 +7847,14 @@
                     vg = "/" + vg;
 									url = clvmURL(vg);
 								}
+                else if (args.data.protocol == "rbd") {
+                  var rbdmonitor = args.data.rbdmonitor;
+                  var rbdpool = args.data.rbdpool;
+                  var rbdid = args.data.rbdid;
+                  var rbdsecret = args.data.rbdsecret;
+
+                  url = rbdURL(rbdmonitor, rbdpool, rbdid, rbdsecret);
+                }
                 else if (args.data.protocol == "vmfs") {
                   //var path = trim($thisDialog.find("#add_pool_vmfs_dc").val());
                   var path = args.data.vCenterDataCenter;