diff --git a/deploy/kubernetes/controller.yaml b/deploy/kubernetes/controller.yaml
index 2f2cd6e389..8333090a9a 100644
--- a/deploy/kubernetes/controller.yaml
+++ b/deploy/kubernetes/controller.yaml
@@ -105,6 +105,56 @@ roleRef:
 
 ---
 
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: external-snapshotter-role
+rules:
+  - apiGroups: [""]
+    resources: ["persistentvolumes"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: [""]
+    resources: ["persistentvolumeclaims"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: ["storage.k8s.io"]
+    resources: ["storageclasses"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: [""]
+    resources: ["events"]
+    verbs: ["list", "watch", "create", "update", "patch"]
+  - apiGroups: [""]
+    resources: ["secrets"]
+    verbs: ["get", "list"]
+  - apiGroups: ["snapshot.storage.k8s.io"]
+    resources: ["volumesnapshotclasses"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: ["snapshot.storage.k8s.io"]
+    resources: ["volumesnapshotcontents"]
+    verbs: ["create", "get", "list", "watch", "update", "delete"]
+  - apiGroups: ["snapshot.storage.k8s.io"]
+    resources: ["volumesnapshots"]
+    verbs: ["get", "list", "watch", "update"]
+  - apiGroups: ["apiextensions.k8s.io"]
+    resources: ["customresourcedefinitions"]
+    verbs: ["create", "list", "watch", "delete"]
+
+---
+
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: csi-snapshotter-binding
+subjects:
+  - kind: ServiceAccount
+    name: csi-controller-sa
+    namespace: kube-system
+roleRef:
+  kind: ClusterRole
+  name: external-snapshotter-role
+  apiGroup: rbac.authorization.k8s.io
+
+---
+
 kind: StatefulSet
 apiVersion: apps/v1beta1
 metadata:
@@ -186,6 +236,18 @@ spec:
           volumeMounts:
             - name: socket-dir
               mountPath: /var/lib/csi/sockets/pluginproxy/
+        - name: csi-snapshotter
+          image: quay.io/k8scsi/csi-snapshotter:v1.0.1
+          args:
+            - --csi-address=$(ADDRESS)
+            - --connection-timeout=15s
+          env:
+            - name: ADDRESS
+              value: /var/lib/csi/sockets/pluginproxy/csi.sock
+          imagePullPolicy: Always
+          volumeMounts:
+            - name: socket-dir
+              mountPath: /var/lib/csi/sockets/pluginproxy/
       volumes:
         - name: socket-dir
           emptyDir: {}
diff --git a/docs/README.md b/docs/README.md
index 3fd2b0505f..0d85c7195b 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -61,7 +61,7 @@ There are several optional parameters that could be passed into `CreateVolumeReq
 
 2. Enable the flag `--allow-privileged=true` in the manifest entries of kubelet and kube-apiserver.
 
-3. Add `--feature-gates=CSINodeInfo=true,CSIDriverRegistry=true` in the manifest entries of kubelet and kube-apiserver. This is required to enable topology support of EBS volumes in Kubernetes.
+3. Add `--feature-gates=CSINodeInfo=true,CSIDriverRegistry=true,VolumeSnapshotDataSource=true` in the manifest entries of kubelet and kube-apiserver. This is required to enable topology support of EBS volumes in Kubernetes and restoring volumes from snapshots.
 
 4. Install the `CSINodeInfo` CRD on the cluster using the instructions provided here: [Enabling CSINodeInfo](https://kubernetes-csi.github.io/docs/csi-node-info-object.html#enabling-csinodeinfo).
 
diff --git a/examples/kubernetes/snapshot/README.md b/examples/kubernetes/snapshot/README.md
new file mode 100644
index 0000000000..2a607999d3
--- /dev/null
+++ b/examples/kubernetes/snapshot/README.md
@@ -0,0 +1,42 @@
+# Volume Snapshots with AWS EBS CSI Driver
+
+## Overview
+
+This driver implements basic volume snapshotting functionality, i.e. it is possible to use it along with the [external
+snapshotter](https://github.com/kubernetes-csi/external-snapshotter) sidecar and create snapshots of EBS volumes using
+the `VolumeSnapshot` custom resources.
+
+## Prerequisites
+
+1. Kubernetes 1.13+ (CSI 1.0) is required
+
+2. The `VolumeSnapshotDataSource` feature gate of Kubernetes API server and controller manager must be turned on.
+
+## Usage
+
+This directory contains example YAML files to test the feature. First, see the [deployment example](../../../deploy/kubernetes) and [volume scheduling example](../volume_scheduling)
+to set up the external provisioner:
+
+### Set up
+
+1. Create the RBAC rules
+
+2. Start the contoller `StatefulSet`
+
+3. Start the node `DaemonSet`
+
+4. Create a `StorageClass` for dynamic provisioning of the AWS CSI volumes
+
+5. Create a `SnapshotClass` to create `VolumeSnapshot`s using the AWS CSI external controller
+
+6. Create a `PersistentVolumeClaim` and a pod using it
+
+### Taking and restoring volume snapshot
+
+7. Create a `VolumeSnapshot` referencing the `PersistentVolumeClaim`; the snapshot creation may take time to finish:
+   check the `ReadyToUse` attribute of the `VolumeSnapshot` object to find out when a new `PersistentVolume` can be
+   created from the snapshot
+
+8. To restore a volume from a snapshot use a `PersistentVolumeClaim` referencing the `VolumeSnapshot` in its `dataSource`; see the
+   [Kubernetes Persistent Volumes documentation](https://kubernetes.io/docs/concepts/storage/persistent-volumes/#volume-snapshot-and-restore-volume-from-snapshot-support)
+   and the example [restore claim](./restore-claim.yaml)
diff --git a/examples/kubernetes/snapshot/claim.yaml b/examples/kubernetes/snapshot/claim.yaml
new file mode 100644
index 0000000000..a883baa530
--- /dev/null
+++ b/examples/kubernetes/snapshot/claim.yaml
@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: ebs-claim
+spec:
+  accessModes:
+    - ReadWriteOnce
+  storageClassName: ebs-sc
+  resources:
+    requests:
+      storage: 4Gi
diff --git a/examples/kubernetes/snapshot/pod.yaml b/examples/kubernetes/snapshot/pod.yaml
new file mode 100644
index 0000000000..f274015916
--- /dev/null
+++ b/examples/kubernetes/snapshot/pod.yaml
@@ -0,0 +1,17 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: app
+spec:
+  containers:
+  - name: app
+    image: centos
+    command: ["/bin/sh"]
+    args: ["-c", "while true; do echo $(date -u) >> /data/out.txt; sleep 5; done"]
+    volumeMounts:
+    - name: persistent-storage
+      mountPath: /data
+  volumes:
+  - name: persistent-storage
+    persistentVolumeClaim:
+      claimName: ebs-claim
diff --git a/examples/kubernetes/snapshot/restore-claim.yaml b/examples/kubernetes/snapshot/restore-claim.yaml
new file mode 100644
index 0000000000..eba2c0646f
--- /dev/null
+++ b/examples/kubernetes/snapshot/restore-claim.yaml
@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: ebs-restore-claim
+spec:
+  accessModes:
+    - ReadWriteOnce
+  storageClassName: ebs-sc
+  resources:
+    requests:
+      storage: 2Gi
+  dataSource:
+    name: ebs-volume-snapshot
+    kind: VolumeSnapshot
+    apiGroup: snapshot.storage.k8s.io
diff --git a/examples/kubernetes/snapshot/snapshot.yaml b/examples/kubernetes/snapshot/snapshot.yaml
new file mode 100644
index 0000000000..69bf3ef2f4
--- /dev/null
+++ b/examples/kubernetes/snapshot/snapshot.yaml
@@ -0,0 +1,9 @@
+apiVersion: snapshot.storage.k8s.io/v1alpha1
+kind: VolumeSnapshot
+metadata:
+  name: ebs-volume-snapshot
+spec:
+  snapshotClassName: csi-aws-snapclass
+  source:
+    name: ebs-claim
+    kind: PersistentVolumeClaim
diff --git a/examples/kubernetes/snapshot/snapshotclass.yaml b/examples/kubernetes/snapshot/snapshotclass.yaml
new file mode 100644
index 0000000000..bc1220078f
--- /dev/null
+++ b/examples/kubernetes/snapshot/snapshotclass.yaml
@@ -0,0 +1,5 @@
+apiVersion: snapshot.storage.k8s.io/v1alpha1
+kind: VolumeSnapshotClass
+metadata:
+  name: csi-aws-snapclass
+snapshotter: ebs.csi.aws.com
diff --git a/examples/kubernetes/snapshot/storageclass.yaml b/examples/kubernetes/snapshot/storageclass.yaml
new file mode 100644
index 0000000000..d6e168e1ec
--- /dev/null
+++ b/examples/kubernetes/snapshot/storageclass.yaml
@@ -0,0 +1,6 @@
+kind: StorageClass
+apiVersion: storage.k8s.io/v1
+metadata:
+  name: ebs-sc
+provisioner: ebs.csi.aws.com
+volumeBindingMode: WaitForFirstConsumer
diff --git a/go.mod b/go.mod
index 870de9cf1c..f4c46e07b6 100644
--- a/go.mod
+++ b/go.mod
@@ -31,6 +31,7 @@ require (
 	github.com/gogo/protobuf v1.1.1 // indirect
 	github.com/golang/groupcache v0.0.0-20181024230925-c65c006176ff // indirect
 	github.com/golang/mock v1.2.0
+	github.com/golang/protobuf v1.2.0
 	github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c // indirect
 	github.com/google/go-cmp v0.2.0 // indirect
 	github.com/google/gofuzz v0.0.0-20170612174753-24818f796faf // indirect
diff --git a/go.sum b/go.sum
index f7b3d335af..100b9c9486 100644
--- a/go.sum
+++ b/go.sum
@@ -3,7 +3,6 @@ bitbucket.org/ww/goautoneg v0.0.0-20120707110453-75cd24fc2f2c/go.mod h1:1vhO7Mn/
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78 h1:w+iIsaOQNcT7OZ575w+acHgRric5iCyQh+xv+KJ4HB8=
 github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=
-github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/NYTimes/gziphandler v1.0.1 h1:iLrQrdwjDd52kHDA5op2UBJFjmOb9g+7scBan4RN8F0=
 github.com/NYTimes/gziphandler v1.0.1/go.mod h1:3wb06e3pkSAbeQ52E9H9iFoQsEEwGN64994WTCIhntQ=
@@ -17,24 +16,19 @@ github.com/aws/aws-sdk-go v1.16.3 h1:esEQzoR8SVXtwg42nRoR/YLftI4ktsZg6Qwr7jnDXy8
 github.com/aws/aws-sdk-go v1.16.3/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 h1:xJ4a3vCFaGF/jqvzLMYoU8P317H5OQ+Via4RmuPwCS0=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
-github.com/boltdb/bolt v1.3.1 h1:JQmyP4ZBrce+ZQu0dY660FMfatumYDLun9hBCUVIkF4=
 github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/container-storage-interface/spec v1.0.0 h1:3DyXuJgf9MU6kyULESegQUmozsSxhpyrrv9u5bfwA3E=
 github.com/container-storage-interface/spec v1.0.0/go.mod h1:6URME8mwIBbpVyZV93Ce5St17xBiQJQY67NDsuohiy4=
-github.com/coreos/bbolt v1.3.0 h1:HIgH5xUWXT914HCI671AxuTTqjj64UOFr7pHn48LUTI=
 github.com/coreos/bbolt v1.3.0/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.10+incompatible h1:jFneRYjIvLMLhDLCzuTuU4rSJUjRplcJQ7pD7MnhC04=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
-github.com/coreos/go-semver v0.2.0 h1:3Jm3tLmsgAYcjC+4Up7hJrFBPr+n7rAqYeSw/SZazuY=
 github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
 github.com/coreos/go-systemd v0.0.0-20181031085051-9002847aa142 h1:3jFq2xL4ZajGK4aZY8jz+DAF0FHjI51BXjjSwCzS1Dk=
 github.com/coreos/go-systemd v0.0.0-20181031085051-9002847aa142/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
-github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f h1:lBNOc5arjvs8E5mO2tbpBpLoyyu8B6e44T7hJy6potg=
 github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/docker/distribution v2.7.0+incompatible h1:neUDAlf3wX6Ml4HdqTrbcOHXtfRN0TFIwt6YFL7N9RU=
 github.com/docker/distribution v2.7.0+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
@@ -44,7 +38,6 @@ github.com/docker/spdystream v0.0.0-20181023171402-6480d4af844c h1:ZfSZ3P3BedhKG
 github.com/docker/spdystream v0.0.0-20181023171402-6480d4af844c/go.mod h1:Qh8CwZgvJUkLughtfhJv5dyTYa91l1fOUCrgjqmcifM=
 github.com/elazarl/go-bindata-assetfs v1.0.0 h1:G/bYguwHIzWq9ZoyUQqrjTmJbbYn3j3CKKpKinvZLFk=
 github.com/elazarl/go-bindata-assetfs v1.0.0/go.mod h1:v+YaWX3bdea5J/mo8dSETolEo7R71Vk1u8bnjau5yw4=
-github.com/elazarl/goproxy v0.0.0-20181111060418-2ce16c963a8a h1:A4wNiqeKqU56ZhtnzJCTyPZ1+cyu8jKtIchQ3TtxHgw=
 github.com/elazarl/goproxy v0.0.0-20181111060418-2ce16c963a8a/go.mod h1:/Zj4wYkgs4iZTTu3o/KG3Itv/qCCa8VVMlb3i9OVuzc=
 github.com/emicklei/go-restful v2.8.0+incompatible h1:wN8GCRDPGHguIynsnBartv5GUgGUg1LAU7+xnSn1j7Q=
 github.com/emicklei/go-restful v2.8.0+incompatible/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs=
@@ -54,9 +47,7 @@ github.com/evanphx/json-patch v4.1.0+incompatible h1:K1MDoo4AZ4wU0GIU/fPmtZg7Vpz
 github.com/evanphx/json-patch v4.1.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
 github.com/fatih/camelcase v1.0.0 h1:hxNvNX/xYBp0ovncs8WyWZrOrpBNub/JfaMvbURyft8=
 github.com/fatih/camelcase v1.0.0/go.mod h1:yN2Sb0lFhZJUdVvtELVWefmrXpuZESvPmqwoZc+/fpc=
-github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
-github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/globalsign/mgo v0.0.0-20180905125535-1ca0a4f7cbcb h1:D4uzjWwKYQ5XnAvUbuvHW93esHg7F8N/OYeBBcJoTr0=
 github.com/globalsign/mgo v0.0.0-20180905125535-1ca0a4f7cbcb/go.mod h1:xkRDCp4j0OGD1HRkm4kmhM+pmpv3AKq5SU7GMg4oO/Q=
@@ -85,7 +76,6 @@ github.com/go-openapi/validate v0.17.2 h1:lwFfiS4sv5DvOrsYDsYq4N7UU8ghXiYtPJ+VcQ
 github.com/go-openapi/validate v0.17.2/go.mod h1:Uh4HdOzKt19xGIGm1qHf/ofbX1YQ4Y+MYsct2VUrAJ4=
 github.com/gogo/protobuf v1.1.1 h1:72R+M5VuhED/KujmZVcIquuo8mBgX4oVda//DQb3PXo=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
-github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/groupcache v0.0.0-20181024230925-c65c006176ff h1:kOkM9whyQYodu09SJ6W3NCsHG7crFaJILQ22Gozp3lg=
 github.com/golang/groupcache v0.0.0-20181024230925-c65c006176ff/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
@@ -97,7 +87,6 @@ github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c h1:964Od4U6p2jUkFxvCydnIczKteheJEzHRToSGK3Bnlw=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
-github.com/google/go-cmp v0.2.0 h1:+dTQ8DZQJz0Mb/HjFlkptS1FeQ4cWSnN941F8aEG4SQ=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/gofuzz v0.0.0-20170612174753-24818f796faf h1:+RRA9JqSOZFfKrOeqr2z77+8R2RKyh8PG66dcu1V0ck=
 github.com/google/gofuzz v0.0.0-20170612174753-24818f796faf/go.mod h1:HP5RmnzzSNb993RKQDq4+1A4ia9nllfqcQFTQJedwGI=
@@ -105,17 +94,13 @@ github.com/google/uuid v1.0.0 h1:b4Gk+7WdP/d3HZH8EJsZpvV7EtDOgaZLtnaNGIu1adA=
 github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gnostic v0.2.0 h1:l6N3VoaVzTncYYW+9yOz2LJJammFZGBO13sqgEhpy9g=
 github.com/googleapis/gnostic v0.2.0/go.mod h1:sJBsCZ4ayReDTBIg8b9dl28c5xFWyhBTVRp3pOg5EKY=
-github.com/gorilla/websocket v1.4.0 h1:WDFjx/TMzVgy9VdMMQi2K2Emtwi2QcUQsztZ/zLaH/Q=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
-github.com/gotestyourself/gotestyourself v2.2.0+incompatible h1:AQwinXlbQR2HvPjQZOmDhRqsv5mZf+Jb1RnSLxcqZcI=
 github.com/gotestyourself/gotestyourself v2.2.0+incompatible/go.mod h1:zZKM6oeNM8k+FRljX1mnzVYeS8wiGgQyvST1/GafPbY=
 github.com/gregjones/httpcache v0.0.0-20181110185634-c63ab54fda8f h1:ShTPMJQes6tubcjzGMODIVG5hlrCeImaBnZzKF2N8SM=
 github.com/gregjones/httpcache v0.0.0-20181110185634-c63ab54fda8f/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
-github.com/grpc-ecosystem/go-grpc-middleware v1.0.0 h1:Iju5GlWwrvL6UBg4zJJt3btmonfrMlCDdsejg4CZE7c=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 h1:Ovs26xHkKqVztRpIrF/92BcuyuQ/YW4NSIpoGtfXNho=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
-github.com/grpc-ecosystem/grpc-gateway v1.5.1 h1:3scN4iuXkNOyP98jF55Lv8a9j1o/IwvnDIZ0LHJK1nk=
 github.com/grpc-ecosystem/grpc-gateway v1.5.1/go.mod h1:RSKVYQBd5MCa4OVpNdGskqpgL2+G+NZTnrVHpWWfpdw=
 github.com/hashicorp/golang-lru v0.5.0 h1:CL2msUPvZTLb5O648aiLNJw3hnBxN2+1Jq8rCOH9wdo=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
@@ -127,7 +112,6 @@ github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NH
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af h1:pmfjZENx5imkbgOkpRUYLnmbU7UEFbjtDA2hxJ1ichM=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
-github.com/jonboulle/clockwork v0.1.0 h1:VKV+ZcuP6l3yW9doeqz6ziZGgcynBVQO+obU0+0hcPo=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
 github.com/json-iterator/go v1.1.5 h1:gL2yXlmiIo4+t+y32d4WGwOjKGYcGOuyrg46vadswDE=
 github.com/json-iterator/go v1.1.5/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
@@ -159,7 +143,6 @@ github.com/peterbourgon/diskv v2.0.1+incompatible h1:UBdAOUP5p4RWqPBg048CAvpKN+v
 github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR/tNboyR3/BZd58JJSHlUSCU=
 github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
-github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_golang v0.9.1 h1:K47Rk0v/fkEfwfQet2KWhscE0cJzjgCCDBG2KHZoVno=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
@@ -171,29 +154,19 @@ github.com/prometheus/procfs v0.0.0-20181204211112-1dc9a6cbc91a h1:9a8MnZMP0X2nL
 github.com/prometheus/procfs v0.0.0-20181204211112-1dc9a6cbc91a/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/sirupsen/logrus v1.2.0 h1:juTguoYk5qI21pwyTXY3B3Y5cOTH3ZUyZCg1v/mihuo=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
-github.com/soheilhy/cmux v0.1.4 h1:0HKaf1o97UwFjHH9o5XsHUOF+tqmdA7KEzXLpiyaw0E=
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
-github.com/spf13/afero v1.1.2 h1:m8/z1t7/fwjysjQRYbP0RD+bUIF/8tJwPdEZsI83ACI=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
 github.com/spf13/cobra v0.0.3 h1:ZlrZ4XsMRm04Fr5pSFxBgfND2EBVa1nLpiy1stUsX/8=
 github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ=
 github.com/spf13/pflag v1.0.3 h1:zPAT6CGy6wXeQ7NtTnaTerfKOsV6V6F8agHXFiazDkg=
 github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
-github.com/stretchr/objx v0.1.1 h1:2vfRuCMp5sSVIDSqO8oNnWJq7mPa6KVP3iPIwFBuy8A=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
-github.com/tmc/grpc-websocket-proxy v0.0.0-20171017195756-830351dc03c6 h1:lYIiVDtZnyTWlNwiAxLj0bbpTcx1BWCFhXjfsvmPdNc=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20171017195756-830351dc03c6/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
-github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8 h1:3SVOIvH7Ae1KRYyQWRjXWJEA9sS/c/pjvH++55Gr648=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
-github.com/xiang90/probing v0.0.0-20160813154853-07dd2e8dfe18 h1:MPPkRncZLN9Kh4MEFmbnK4h3BD7AUmskWv2+EeZJCCs=
 github.com/xiang90/probing v0.0.0-20160813154853-07dd2e8dfe18/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
-go.uber.org/atomic v1.3.2 h1:2Oa65PReHzfn29GpvgsYwloV9AVFHPDk8tYxt2c2tr4=
 go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
-go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
-go.uber.org/zap v1.9.1 h1:XCJQEf3W6eZaVwhRBof6ImoYGJSITeKWsyeh3HFu/5o=
 go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9 h1:mKdxBk7AujPs8kU4m80U72y/zjbZ3UcXC7dClwKbUI0=
@@ -210,7 +183,6 @@ golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAG
 golang.org/x/oauth2 v0.0.0-20181203162652-d668ce993890 h1:uESlIz09WIHT2I+pasSXcpLYqYK8wHcdCetU3VuMBJE=
 golang.org/x/oauth2 v0.0.0-20181203162652-d668ce993890/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20181108010431-42b317875d0f h1:Bl/8QSvNqXvPGPGXa2z5xUTmV7VDcZyvRZ+QQXkXTZQ=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -231,7 +203,6 @@ google.golang.org/genproto v0.0.0-20181202183823-bd91e49a0898/go.mod h1:7Ep/1NZk
 google.golang.org/grpc v1.16.0/go.mod h1:0JHn/cJsOMiMfNA9+DeHDlAU7KAAB5GDlYFpa9MZMio=
 google.golang.org/grpc v1.17.0 h1:TRJYBgMclJvGYn2rIMjj+h9KtMt5r1Ij7ODVRIZkwhk=
 google.golang.org/grpc v1.17.0/go.mod h1:6QZJwpn2B+Zp71q/5VxRsJ6NXXVCE5NRUHRo+f3cWCs=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
@@ -246,7 +217,6 @@ gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWD
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo=
 gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=
 honnef.co/go/tools v0.0.0-20180728063816-88497007e858/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 k8s.io/api v0.0.0-20181204000039-89a74a8d264d h1:HQoGWsWUe/FmRcX9BU440AAMnzBFEf+DBo4nbkQlNzs=
diff --git a/pkg/cloud/cloud.go b/pkg/cloud/cloud.go
index e4022447e7..b308e14426 100644
--- a/pkg/cloud/cloud.go
+++ b/pkg/cloud/cloud.go
@@ -73,6 +73,8 @@ const (
 const (
 	// VolumeNameTagKey is the key value that refers to the volume's name.
 	VolumeNameTagKey = "CSIVolumeName"
+	// SnapshotNameTagKey is the key value that refers to the snapshot's name.
+	SnapshotNameTagKey = "CSIVolumeSnapshotName"
 )
 
 var (
@@ -109,7 +111,22 @@ type DiskOptions struct {
 	Encrypted        bool
 	// KmsKeyID represents a fully qualified resource name to the key to use for encryption.
 	// example: arn:aws:kms:us-east-1:012345678910:key/abcd1234-a123-456a-a12b-a123b4cd56ef
-	KmsKeyID string
+	KmsKeyID   string
+	SnapshotID string
+}
+
+// Snapshot represents an EBS volume snapshot
+type Snapshot struct {
+	SnapshotID     string
+	SourceVolumeID string
+	Size           int64
+	CreationTime   time.Time
+	ReadyToUse     bool
+}
+
+// SnapshotOptions represents parameters to create an EBS volume
+type SnapshotOptions struct {
+	Tags map[string]string
 }
 
 // EC2 abstracts aws.EC2 to facilitate its mocking.
@@ -121,6 +138,9 @@ type EC2 interface {
 	DetachVolumeWithContext(ctx aws.Context, input *ec2.DetachVolumeInput, opts ...request.Option) (*ec2.VolumeAttachment, error)
 	AttachVolumeWithContext(ctx aws.Context, input *ec2.AttachVolumeInput, opts ...request.Option) (*ec2.VolumeAttachment, error)
 	DescribeInstancesWithContext(ctx aws.Context, input *ec2.DescribeInstancesInput, opts ...request.Option) (*ec2.DescribeInstancesOutput, error)
+	CreateSnapshotWithContext(ctx aws.Context, input *ec2.CreateSnapshotInput, opts ...request.Option) (*ec2.Snapshot, error)
+	DeleteSnapshotWithContext(ctx aws.Context, input *ec2.DeleteSnapshotInput, opts ...request.Option) (*ec2.DeleteSnapshotOutput, error)
+	DescribeSnapshotsWithContext(ctx aws.Context, input *ec2.DescribeSnapshotsInput, opts ...request.Option) (*ec2.DescribeSnapshotsOutput, error)
 }
 
 type Cloud interface {
@@ -133,6 +153,9 @@ type Cloud interface {
 	GetDiskByName(ctx context.Context, name string, capacityBytes int64) (disk *Disk, err error)
 	GetDiskByID(ctx context.Context, volumeID string) (disk *Disk, err error)
 	IsExistInstance(ctx context.Context, nodeID string) (success bool)
+	CreateSnapshot(ctx context.Context, volumeID string, snapshotOptions *SnapshotOptions) (snapshot *Snapshot, err error)
+	DeleteSnapshot(ctx context.Context, snapshotID string) (success bool, err error)
+	GetSnapshotByName(ctx context.Context, name string) (snapshot *Snapshot, err error)
 }
 
 type cloud struct {
@@ -245,6 +268,10 @@ func (c *cloud) CreateDisk(ctx context.Context, volumeName string, diskOptions *
 	if iops > 0 {
 		request.Iops = aws.Int64(iops)
 	}
+	snapshotID := diskOptions.SnapshotID
+	if len(snapshotID) > 0 {
+		request.SnapshotId = aws.String(snapshotID)
+	}
 
 	response, err := c.ec2.CreateVolumeWithContext(ctx, request)
 	if err != nil {
@@ -457,6 +484,87 @@ func (c *cloud) IsExistInstance(ctx context.Context, nodeID string) bool {
 	return true
 }
 
+func (c *cloud) CreateSnapshot(ctx context.Context, volumeID string, snapshotOptions *SnapshotOptions) (snapshot *Snapshot, err error) {
+	descriptions := "Created by AWS EBS CSI driver for volume " + volumeID
+
+	var tags []*ec2.Tag
+	for key, value := range snapshotOptions.Tags {
+		tags = append(tags, &ec2.Tag{Key: &key, Value: &value})
+	}
+	tagSpec := ec2.TagSpecification{
+		ResourceType: aws.String("snapshot"),
+		Tags:         tags,
+	}
+	request := &ec2.CreateSnapshotInput{
+		VolumeId:          aws.String(volumeID),
+		DryRun:            aws.Bool(false),
+		TagSpecifications: []*ec2.TagSpecification{&tagSpec},
+		Description:       aws.String(descriptions),
+	}
+
+	res, err := c.ec2.CreateSnapshotWithContext(ctx, request)
+	if err != nil {
+		return nil, fmt.Errorf("error creating snapshot of volume %s: %v", volumeID, err)
+	}
+	if res == nil {
+		return nil, fmt.Errorf("nil CreateSnapshotResponse")
+	}
+
+	return c.ec2SnapshotResponseToStruct(res), nil
+}
+
+func (c *cloud) DeleteSnapshot(ctx context.Context, snapshotID string) (success bool, err error) {
+	request := &ec2.DeleteSnapshotInput{}
+	request.SnapshotId = aws.String(snapshotID)
+	request.DryRun = aws.Bool(false)
+	if _, err := c.ec2.DeleteSnapshotWithContext(ctx, request); err != nil {
+		if isAWSErrorSnapshotNotFound(err) {
+			return false, ErrNotFound
+		}
+		return false, fmt.Errorf("DeleteSnapshot could not delete volume: %v", err)
+	}
+	return true, nil
+}
+
+func (c *cloud) GetSnapshotByName(ctx context.Context, name string) (snapshot *Snapshot, err error) {
+	request := &ec2.DescribeSnapshotsInput{
+		Filters: []*ec2.Filter{
+			{
+				Name:   aws.String("tag:" + SnapshotNameTagKey),
+				Values: []*string{aws.String(name)},
+			},
+		},
+	}
+
+	ec2snapshot, err := c.getSnapshot(ctx, request)
+	if err != nil {
+		return nil, err
+	}
+
+	return c.ec2SnapshotResponseToStruct(ec2snapshot), nil
+}
+
+// Helper method converting EC2 snapshot type to the internal struct
+func (c *cloud) ec2SnapshotResponseToStruct(ec2Snapshot *ec2.Snapshot) *Snapshot {
+	if ec2Snapshot == nil {
+		return nil
+	}
+	snapshotSize := util.GiBToBytes(aws.Int64Value(ec2Snapshot.VolumeSize))
+	snapshot := &Snapshot{
+		SnapshotID:     aws.StringValue(ec2Snapshot.SnapshotId),
+		SourceVolumeID: aws.StringValue(ec2Snapshot.VolumeId),
+		Size:           snapshotSize,
+		CreationTime:   aws.TimeValue(ec2Snapshot.StartTime),
+	}
+	if aws.StringValue(ec2Snapshot.State) == "completed" {
+		snapshot.ReadyToUse = true
+	} else {
+		snapshot.ReadyToUse = false
+	}
+
+	return snapshot
+}
+
 func (c *cloud) getVolume(ctx context.Context, request *ec2.DescribeVolumesInput) (*ec2.Volume, error) {
 	var volumes []*ec2.Volume
 	var nextToken *string
@@ -516,6 +624,32 @@ func (c *cloud) getInstance(ctx context.Context, nodeID string) (*ec2.Instance,
 	return instances[0], nil
 }
 
+func (c *cloud) getSnapshot(ctx context.Context, request *ec2.DescribeSnapshotsInput) (*ec2.Snapshot, error) {
+	var snapshots []*ec2.Snapshot
+	var nextToken *string
+
+	for {
+		response, err := c.ec2.DescribeSnapshotsWithContext(ctx, request)
+		if err != nil {
+			return nil, err
+		}
+		snapshots = append(snapshots, response.Snapshots...)
+		nextToken = response.NextToken
+		if aws.StringValue(nextToken) == "" {
+			break
+		}
+		request.NextToken = nextToken
+	}
+
+	if l := len(snapshots); l > 1 {
+		return nil, errors.New("Multiple snapshots with the same name found")
+	} else if l < 1 {
+		return nil, ErrNotFound
+	}
+
+	return snapshots[0], nil
+}
+
 // waitForVolume waits for volume to be in the "available" state.
 // On a random AWS account (shared among several developers) it took 4s on average.
 func (c *cloud) waitForVolume(ctx context.Context, volumeID string) error {
@@ -538,14 +672,7 @@ func (c *cloud) waitForVolume(ctx context.Context, volumeID string) error {
 			return true, err
 		}
 		if vol.State != nil {
-			switch *vol.State {
-			case "available":
-				return true, nil
-			case "creating":
-				return false, nil
-			default:
-				return true, fmt.Errorf("unexpected state for volume %s: %q", volumeID, *vol.State)
-			}
+			return *vol.State == "available", nil
 		}
 		return false, nil
 	})
@@ -564,3 +691,16 @@ func isAWSErrorVolumeNotFound(err error) bool {
 	}
 	return false
 }
+
+// Helper function for describeSnapshot callers. Tries to retype given error to AWS error
+// and returns true in case the AWS error is "InvalidSnapshot.NotFound", false otherwise
+func isAWSErrorSnapshotNotFound(err error) bool {
+	if awsError, ok := err.(awserr.Error); ok {
+		// https://docs.aws.amazon.com/AWSEC2/latest/APIReference/errors-overview.html
+		if awsError.Code() == "InvalidSnapshot.NotFound" {
+			return true
+		}
+	}
+
+	return false
+}
diff --git a/pkg/cloud/cloud_test.go b/pkg/cloud/cloud_test.go
index d03cb3a698..66bb979008 100644
--- a/pkg/cloud/cloud_test.go
+++ b/pkg/cloud/cloud_test.go
@@ -127,6 +127,22 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: fmt.Errorf("failed to get an available volume in EC2: timed out waiting for the condition"),
 		},
+		{
+			name:       "success: normal from snapshot",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test"},
+				AvailabilityZone: expZone,
+				SnapshotID:       "snapshot-test",
+			},
+			expDisk: &Disk{
+				VolumeID:         "vol-test",
+				CapacityGiB:      1,
+				AvailabilityZone: expZone,
+			},
+			expErr: nil,
+		},
 	}
 
 	for _, tc := range testCases {
@@ -146,10 +162,17 @@ func TestCreateDisk(t *testing.T) {
 				State:            aws.String(volState),
 				AvailabilityZone: aws.String(tc.diskOptions.AvailabilityZone),
 			}
-
+			snapshot := &ec2.Snapshot{
+				SnapshotId: aws.String(tc.diskOptions.SnapshotID),
+				VolumeId:   aws.String("snap-test-volume"),
+				State:      aws.String("completed"),
+			}
 			ctx := context.Background()
 			mockEC2.EXPECT().CreateVolumeWithContext(gomock.Eq(ctx), gomock.Any()).Return(vol, tc.expCreateVolumeErr)
 			mockEC2.EXPECT().DescribeVolumesWithContext(gomock.Eq(ctx), gomock.Any()).Return(&ec2.DescribeVolumesOutput{Volumes: []*ec2.Volume{vol}}, tc.expDescVolumeErr).AnyTimes()
+			if len(tc.diskOptions.SnapshotID) > 0 {
+				mockEC2.EXPECT().DescribeSnapshotsWithContext(gomock.Eq(ctx), gomock.Any()).Return(&ec2.DescribeSnapshotsOutput{Snapshots: []*ec2.Snapshot{snapshot}}, nil).AnyTimes()
+			}
 
 			disk, err := c.CreateDisk(ctx, tc.volumeName, tc.diskOptions)
 			if err != nil {
@@ -463,6 +486,167 @@ func TestGetDiskByID(t *testing.T) {
 	}
 }
 
+func TestCreateSnapshot(t *testing.T) {
+	testCases := []struct {
+		name            string
+		snapshotName    string
+		snapshotOptions *SnapshotOptions
+		expSnapshot     *Snapshot
+		expErr          error
+	}{
+		{
+			name:         "success: normal",
+			snapshotName: "snap-test-name",
+			snapshotOptions: &SnapshotOptions{
+				Tags: map[string]string{
+					SnapshotNameTagKey: "snap-test-name",
+				},
+			},
+			expSnapshot: &Snapshot{
+				SourceVolumeID: "snap-test-volume",
+			},
+			expErr: nil,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			mockCtrl := gomock.NewController(t)
+			mockEC2 := mocks.NewMockEC2(mockCtrl)
+			c := newCloud(mockEC2)
+
+			ec2snapshot := &ec2.Snapshot{
+				SnapshotId: aws.String(tc.snapshotOptions.Tags[SnapshotNameTagKey]),
+				VolumeId:   aws.String("snap-test-volume"),
+				State:      aws.String("completed"),
+			}
+
+			ctx := context.Background()
+			mockEC2.EXPECT().CreateSnapshotWithContext(gomock.Eq(ctx), gomock.Any()).Return(ec2snapshot, tc.expErr)
+			mockEC2.EXPECT().DescribeSnapshotsWithContext(gomock.Eq(ctx), gomock.Any()).Return(&ec2.DescribeSnapshotsOutput{Snapshots: []*ec2.Snapshot{ec2snapshot}}, nil).AnyTimes()
+
+			snapshot, err := c.CreateSnapshot(ctx, tc.expSnapshot.SourceVolumeID, tc.snapshotOptions)
+			if err != nil {
+				if tc.expErr == nil {
+					t.Fatalf("CreateSnapshot() failed: expected no error, got: %v", err)
+				}
+			} else {
+				if tc.expErr != nil {
+					t.Fatal("CreateSnapshot() failed: expected error, got nothing")
+				} else {
+					if snapshot.SourceVolumeID != tc.expSnapshot.SourceVolumeID {
+						t.Fatalf("CreateSnapshot() failed: expected source volume ID %s, got %v", tc.expSnapshot.SourceVolumeID, snapshot.SourceVolumeID)
+					}
+				}
+			}
+
+			mockCtrl.Finish()
+		})
+	}
+}
+
+func TestDeleteSnapshot(t *testing.T) {
+	testCases := []struct {
+		name         string
+		snapshotName string
+		expErr       error
+	}{
+		{
+			name:         "success: normal",
+			snapshotName: "snap-test-name",
+			expErr:       nil,
+		},
+		{
+			name:         "fail: delete snapshot return generic error",
+			snapshotName: "snap-test-name",
+			expErr:       fmt.Errorf("DeleteSnapshot generic error"),
+		},
+		{
+			name:         "fail: delete snapshot return not found error",
+			snapshotName: "snap-test-name",
+			expErr:       awserr.New("InvalidSnapshot.NotFound", "", nil),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			mockCtrl := gomock.NewController(t)
+			mockEC2 := mocks.NewMockEC2(mockCtrl)
+			c := newCloud(mockEC2)
+
+			ctx := context.Background()
+			mockEC2.EXPECT().DeleteSnapshotWithContext(gomock.Eq(ctx), gomock.Any()).Return(&ec2.DeleteSnapshotOutput{}, tc.expErr)
+
+			_, err := c.DeleteSnapshot(ctx, tc.snapshotName)
+			if err != nil {
+				if tc.expErr == nil {
+					t.Fatalf("DeleteSnapshot() failed: expected no error, got: %v", err)
+				}
+			} else {
+				if tc.expErr != nil {
+					t.Fatal("DeleteSnapshot() failed: expected error, got nothing")
+				}
+			}
+
+			mockCtrl.Finish()
+		})
+	}
+}
+
+func TestGetSnapshotByName(t *testing.T) {
+	testCases := []struct {
+		name            string
+		snapshotName    string
+		snapshotOptions *SnapshotOptions
+		expSnapshot     *Snapshot
+		expErr          error
+	}{
+		{
+			name:         "success: normal",
+			snapshotName: "snap-test-name",
+			snapshotOptions: &SnapshotOptions{
+				Tags: map[string]string{
+					SnapshotNameTagKey: "snap-test-name",
+				},
+			},
+			expSnapshot: &Snapshot{
+				SourceVolumeID: "snap-test-volume",
+			},
+			expErr: nil,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			mockCtrl := gomock.NewController(t)
+			mockEC2 := mocks.NewMockEC2(mockCtrl)
+			c := newCloud(mockEC2)
+
+			ec2snapshot := &ec2.Snapshot{
+				SnapshotId: aws.String(tc.snapshotOptions.Tags[SnapshotNameTagKey]),
+				VolumeId:   aws.String("snap-test-volume"),
+				State:      aws.String("completed"),
+			}
+
+			ctx := context.Background()
+			mockEC2.EXPECT().DescribeSnapshotsWithContext(gomock.Eq(ctx), gomock.Any()).Return(&ec2.DescribeSnapshotsOutput{Snapshots: []*ec2.Snapshot{ec2snapshot}}, nil)
+
+			_, err := c.GetSnapshotByName(ctx, tc.snapshotOptions.Tags[SnapshotNameTagKey])
+			if err != nil {
+				if tc.expErr == nil {
+					t.Fatalf("GetSnapshotByName() failed: expected no error, got: %v", err)
+				}
+			} else {
+				if tc.expErr != nil {
+					t.Fatal("GetSnapshotByName() failed: expected error, got nothing")
+				}
+			}
+
+			mockCtrl.Finish()
+		})
+	}
+}
+
 func newCloud(mockEC2 EC2) Cloud {
 	return &cloud{
 		metadata: &metadata{
diff --git a/pkg/cloud/fakes.go b/pkg/cloud/fakes.go
index 8bc085fe9b..e6941fac43 100644
--- a/pkg/cloud/fakes.go
+++ b/pkg/cloud/fakes.go
@@ -26,9 +26,10 @@ import (
 )
 
 type FakeCloudProvider struct {
-	disks map[string]*fakeDisk
-	m     *metadata
-	pub   map[string]string
+	disks     map[string]*fakeDisk
+	snapshots map[string]*fakeSnapshot
+	m         *metadata
+	pub       map[string]string
 }
 
 type fakeDisk struct {
@@ -36,11 +37,17 @@ type fakeDisk struct {
 	tags map[string]string
 }
 
+type fakeSnapshot struct {
+	*Snapshot
+	tags map[string]string
+}
+
 func NewFakeCloudProvider() *FakeCloudProvider {
 	return &FakeCloudProvider{
-		disks: make(map[string]*fakeDisk),
-		pub:   make(map[string]string),
-		m:     &metadata{"instanceID", "region", "az"},
+		disks:     make(map[string]*fakeDisk),
+		snapshots: make(map[string]*fakeSnapshot),
+		pub:       make(map[string]string),
+		m:         &metadata{"instanceID", "region", "az"},
 	}
 }
 
@@ -119,3 +126,45 @@ func (c *FakeCloudProvider) GetDiskByID(ctx context.Context, volumeID string) (*
 func (c *FakeCloudProvider) IsExistInstance(ctx context.Context, nodeID string) bool {
 	return nodeID == c.m.GetInstanceID()
 }
+
+func (c *FakeCloudProvider) CreateSnapshot(ctx context.Context, volumeID string, snapshotOptions *SnapshotOptions) (snapshot *Snapshot, err error) {
+	r1 := rand.New(rand.NewSource(time.Now().UnixNano()))
+	snapshotID := fmt.Sprintf("snapshot-%d", r1.Uint64())
+	if len(snapshotOptions.Tags[SnapshotNameTagKey]) == 0 {
+		// for simplicity: let's have the Name and ID identical
+		snapshotOptions.Tags[SnapshotNameTagKey] = snapshotID
+	}
+	s := &fakeSnapshot{
+		Snapshot: &Snapshot{
+			SnapshotID:     snapshotID,
+			SourceVolumeID: volumeID,
+			Size:           1,
+			CreationTime:   time.Now(),
+		},
+		tags: snapshotOptions.Tags,
+	}
+	c.snapshots[snapshotID] = s
+	return s.Snapshot, nil
+
+}
+
+func (c *FakeCloudProvider) DeleteSnapshot(ctx context.Context, snapshotID string) (success bool, err error) {
+	delete(c.snapshots, snapshotID)
+	return true, nil
+
+}
+
+func (c *FakeCloudProvider) GetSnapshotByName(ctx context.Context, name string) (snapshot *Snapshot, err error) {
+	var snapshots []*fakeSnapshot
+	for _, s := range c.snapshots {
+		for key, value := range s.tags {
+			if key == SnapshotNameTagKey && value == name {
+				snapshots = append(snapshots, s)
+			}
+		}
+	}
+	if len(snapshots) == 0 {
+		return nil, nil
+	}
+	return snapshots[0].Snapshot, nil
+}
diff --git a/pkg/cloud/mocks/mock_ec2.go b/pkg/cloud/mocks/mock_ec2.go
index 53d4ac1ba2..d893d813e5 100644
--- a/pkg/cloud/mocks/mock_ec2.go
+++ b/pkg/cloud/mocks/mock_ec2.go
@@ -126,3 +126,51 @@ func (_mr *_MockEC2Recorder) DetachVolumeWithContext(arg0, arg1 interface{}, arg
 	_s := append([]interface{}{arg0, arg1}, arg2...)
 	return _mr.mock.ctrl.RecordCall(_mr.mock, "DetachVolumeWithContext", _s...)
 }
+
+func (_m *MockEC2) CreateSnapshotWithContext(arg0 aws.Context, arg1 *ec2.CreateSnapshotInput, arg2 ...request.Option) (*ec2.Snapshot, error) {
+	_s := []interface{}{arg0, arg1}
+	for _, a := range arg2 {
+		_s = append(_s, a)
+	}
+	ret := _m.ctrl.Call(_m, "CreateSnapshotWithContext", _s...)
+	ret0, _ := ret[0].(*ec2.Snapshot)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+func (_mr *_MockEC2Recorder) CreateSnapshotWithContext(arg0, arg1 interface{}, arg2 ...interface{}) *gomock.Call {
+	_s := append([]interface{}{arg0, arg1}, arg2...)
+	return _mr.mock.ctrl.RecordCall(_mr.mock, "CreateSnapshotWithContext", _s...)
+}
+
+func (_m *MockEC2) DeleteSnapshotWithContext(arg0 aws.Context, arg1 *ec2.DeleteSnapshotInput, arg2 ...request.Option) (*ec2.DeleteSnapshotOutput, error) {
+	_s := []interface{}{arg0, arg1}
+	for _, a := range arg2 {
+		_s = append(_s, a)
+	}
+	ret := _m.ctrl.Call(_m, "DeleteSnapshotWithContext", _s...)
+	ret0, _ := ret[0].(*ec2.DeleteSnapshotOutput)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+func (_mr *_MockEC2Recorder) DeleteSnapshotWithContext(arg0, arg1 interface{}) *gomock.Call {
+	_s := []interface{}{arg0, arg1}
+	return _mr.mock.ctrl.RecordCall(_mr.mock, "DeleteSnapshotWithContext", _s...)
+}
+
+func (_m *MockEC2) DescribeSnapshotsWithContext(arg0 aws.Context, arg1 *ec2.DescribeSnapshotsInput, arg2 ...request.Option) (*ec2.DescribeSnapshotsOutput, error) {
+	_s := []interface{}{arg0, arg1}
+	for _, a := range arg2 {
+		_s = append(_s, a)
+	}
+	ret := _m.ctrl.Call(_m, "DescribeSnapshotsWithContext", _s...)
+	ret0, _ := ret[0].(*ec2.DescribeSnapshotsOutput)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+func (_mr *_MockEC2Recorder) DescribeSnapshotsWithContext(arg0, arg1 interface{}) *gomock.Call {
+	_s := []interface{}{arg0, arg1}
+	return _mr.mock.ctrl.RecordCall(_mr.mock, "DescribeSnapshotsWithContext", _s...)
+}
diff --git a/pkg/driver/controller.go b/pkg/driver/controller.go
index d9436c2873..d33fe40193 100644
--- a/pkg/driver/controller.go
+++ b/pkg/driver/controller.go
@@ -21,6 +21,7 @@ import (
 	"strconv"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/golang/protobuf/ptypes"
 	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/cloud"
 	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/util"
 	"google.golang.org/grpc/codes"
@@ -42,6 +43,7 @@ var (
 	controllerCaps = []csi.ControllerServiceCapability_RPC_Type{
 		csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
 		csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME,
+		csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
 	}
 )
 
@@ -124,6 +126,19 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 		Encrypted:        isEncrypted,
 		KmsKeyID:         kmsKeyId,
 	}
+
+	volumeSource := req.GetVolumeContentSource()
+	if volumeSource != nil {
+		if _, ok := volumeSource.GetType().(*csi.VolumeContentSource_Snapshot); !ok {
+			return nil, status.Error(codes.InvalidArgument, "Unsupported volumeContentSource type")
+		}
+		sourceSnapshot := volumeSource.GetSnapshot()
+		if sourceSnapshot == nil {
+			return nil, status.Error(codes.InvalidArgument, "Error retrieving snapshot from the volumeContentSource")
+		}
+		opts.SnapshotID = sourceSnapshot.GetSnapshotId()
+	}
+
 	disk, err = d.cloud.CreateDisk(ctx, volName, opts)
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "Could not create volume %q: %v", volName, err)
@@ -290,11 +305,56 @@ func (d *Driver) isValidVolumeCapabilities(volCaps []*csi.VolumeCapability) bool
 }
 
 func (d *Driver) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "")
+	klog.V(4).Infof("CreateSnapshot: called with args %+v", req)
+	snapshotName := req.GetName()
+	if len(snapshotName) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Snapshot name not provided")
+	}
+
+	volumeID := req.GetSourceVolumeId()
+	if len(volumeID) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Snapshot volume source ID not provided")
+	}
+	snapshot, err := d.cloud.GetSnapshotByName(ctx, snapshotName)
+	if err != nil && err != cloud.ErrNotFound {
+		klog.Errorf("Error looking for the snapshot %s: %v", snapshotName, err)
+		return nil, err
+	}
+	if snapshot != nil {
+		if snapshot.SourceVolumeID != volumeID {
+			return nil, status.Errorf(codes.AlreadyExists, "Snapshot %s already exists for different volume (%s)", snapshotName, snapshot.SourceVolumeID)
+		} else {
+			klog.Infof("Snapshot %s of volume %s already exists; nothing to do", snapshotName, volumeID)
+			return newCreateSnapshotResponse(snapshot)
+		}
+	}
+	opts := &cloud.SnapshotOptions{
+		Tags: map[string]string{cloud.SnapshotNameTagKey: snapshotName},
+	}
+	snapshot, err = d.cloud.CreateSnapshot(ctx, volumeID, opts)
+
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "Could not create snapshot %q: %v", snapshotName, err)
+	}
+	return newCreateSnapshotResponse(snapshot)
 }
 
 func (d *Driver) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "")
+	klog.V(4).Infof("DeleteSnapshot: called with args %+v", req)
+	snapshotID := req.GetSnapshotId()
+	if len(snapshotID) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Snapshot ID not provided")
+	}
+
+	if _, err := d.cloud.DeleteSnapshot(ctx, snapshotID); err != nil {
+		if err == cloud.ErrNotFound {
+			klog.V(4).Info("DeleteSnapshot: snapshot not found, returning with success")
+			return &csi.DeleteSnapshotResponse{}, nil
+		}
+		return nil, status.Errorf(codes.Internal, "Could not delete snapshot ID %q: %v", snapshotID, err)
+	}
+
+	return &csi.DeleteSnapshotResponse{}, nil
 }
 
 func (d *Driver) ListSnapshots(ctx context.Context, req *csi.ListSnapshotsRequest) (*csi.ListSnapshotsResponse, error) {
@@ -338,3 +398,19 @@ func newCreateVolumeResponse(disk *cloud.Disk) *csi.CreateVolumeResponse {
 		},
 	}
 }
+
+func newCreateSnapshotResponse(snapshot *cloud.Snapshot) (*csi.CreateSnapshotResponse, error) {
+	ts, err := ptypes.TimestampProto(snapshot.CreationTime)
+	if err != nil {
+		return nil, err
+	}
+	return &csi.CreateSnapshotResponse{
+		Snapshot: &csi.Snapshot{
+			SnapshotId:     snapshot.SnapshotID,
+			SourceVolumeId: snapshot.SourceVolumeID,
+			SizeBytes:      snapshot.Size,
+			CreationTime:   ts,
+			ReadyToUse:     snapshot.ReadyToUse,
+		},
+	}, nil
+}
diff --git a/pkg/driver/controller_test.go b/pkg/driver/controller_test.go
index 57b9f2ff18..0c1603bf3b 100644
--- a/pkg/driver/controller_test.go
+++ b/pkg/driver/controller_test.go
@@ -429,3 +429,163 @@ func TestPickAvailabilityZone(t *testing.T) {
 		})
 	}
 }
+
+func TestCreateSnapshot(t *testing.T) {
+	testCases := []struct {
+		name            string
+		req             *csi.CreateSnapshotRequest
+		extraReq        *csi.CreateSnapshotRequest
+		expSnapshot     *csi.Snapshot
+		expErrCode      codes.Code
+		extraExpErrCode codes.Code
+	}{
+		{
+			name: "success normal",
+			req: &csi.CreateSnapshotRequest{
+				Name:           "test-snapshot",
+				Parameters:     nil,
+				SourceVolumeId: "vol-test",
+			},
+			expSnapshot: &csi.Snapshot{
+				ReadyToUse: true,
+			},
+			expErrCode: codes.OK,
+		},
+		{
+			name: "fail no name",
+			req: &csi.CreateSnapshotRequest{
+				Parameters:     nil,
+				SourceVolumeId: "vol-test",
+			},
+			expSnapshot: nil,
+			expErrCode:  codes.InvalidArgument,
+		},
+		{
+			name: "fail same name different volume ID",
+			req: &csi.CreateSnapshotRequest{
+				Name:           "test-snapshot",
+				Parameters:     nil,
+				SourceVolumeId: "vol-test",
+			},
+			extraReq: &csi.CreateSnapshotRequest{
+				Name:           "test-snapshot",
+				Parameters:     nil,
+				SourceVolumeId: "vol-xxx",
+			},
+			expSnapshot: &csi.Snapshot{
+				ReadyToUse: true,
+			},
+			expErrCode:      codes.OK,
+			extraExpErrCode: codes.AlreadyExists,
+		},
+		{
+			name: "success same name same volume ID",
+			req: &csi.CreateSnapshotRequest{
+				Name:           "test-snapshot",
+				Parameters:     nil,
+				SourceVolumeId: "vol-test",
+			},
+			extraReq: &csi.CreateSnapshotRequest{
+				Name:           "test-snapshot",
+				Parameters:     nil,
+				SourceVolumeId: "vol-test",
+			},
+			expSnapshot: &csi.Snapshot{
+				ReadyToUse: true,
+			},
+			expErrCode:      codes.OK,
+			extraExpErrCode: codes.OK,
+		},
+	}
+	for _, tc := range testCases {
+		t.Logf("Test case: %s", tc.name)
+		awsDriver := NewFakeDriver("", NewFakeMounter())
+		resp, err := awsDriver.CreateSnapshot(context.TODO(), tc.req)
+		if err != nil {
+			srvErr, ok := status.FromError(err)
+			if !ok {
+				t.Fatalf("Could not get error status code from error: %v", srvErr)
+			}
+			if srvErr.Code() != tc.expErrCode {
+				t.Fatalf("Expected error code %d, got %d message %s", tc.expErrCode, srvErr.Code(), srvErr.Message())
+			}
+			continue
+		}
+		if tc.expErrCode != codes.OK {
+			t.Fatalf("Expected error %v, got no error", tc.expErrCode)
+		}
+		snap := resp.GetSnapshot()
+		if snap == nil && tc.expSnapshot != nil {
+			t.Fatalf("Expected snapshot %v, got nil", tc.expSnapshot)
+		}
+		if tc.extraReq != nil {
+			// extraReq is never used in a situation when a new snapshot
+			// should be really created: checking the return code is enough
+			_, err = awsDriver.CreateSnapshot(context.TODO(), tc.extraReq)
+			if err != nil {
+				srvErr, ok := status.FromError(err)
+				if !ok {
+					t.Fatalf("Could not get error status code from error: %v", srvErr)
+				}
+				if srvErr.Code() != tc.extraExpErrCode {
+					t.Fatalf("Expected error code %d, got %d message %s", tc.expErrCode, srvErr.Code(), srvErr.Message())
+				}
+				continue
+			}
+			if tc.extraExpErrCode != codes.OK {
+				t.Fatalf("Expected error %v, got no error", tc.extraExpErrCode)
+			}
+		}
+	}
+}
+
+func TestDeleteSnapshot(t *testing.T) {
+	snapReq := &csi.CreateSnapshotRequest{
+		Name:           "test-snapshot",
+		Parameters:     nil,
+		SourceVolumeId: "vol-test",
+	}
+	testCases := []struct {
+		name       string
+		req        *csi.DeleteSnapshotRequest
+		expErrCode codes.Code
+	}{
+		{
+			name:       "success normal",
+			req:        &csi.DeleteSnapshotRequest{},
+			expErrCode: codes.OK,
+		},
+		{
+			name: "success not found",
+			req: &csi.DeleteSnapshotRequest{
+				SnapshotId: "xxx",
+			},
+			expErrCode: codes.OK,
+		},
+	}
+	for _, tc := range testCases {
+		t.Logf("Test case: %s", tc.name)
+		awsDriver := NewFakeDriver("", NewFakeMounter())
+		snapResp, err := awsDriver.CreateSnapshot(context.TODO(), snapReq)
+		if err != nil {
+			t.Fatalf("Error creating testing snapshot: %v", err)
+		}
+		if len(tc.req.SnapshotId) == 0 {
+			tc.req.SnapshotId = snapResp.Snapshot.SnapshotId
+		}
+		_, err = awsDriver.DeleteSnapshot(context.TODO(), tc.req)
+		if err != nil {
+			srvErr, ok := status.FromError(err)
+			if !ok {
+				t.Fatalf("Could not get error status code from error: %v", srvErr)
+			}
+			if srvErr.Code() != tc.expErrCode {
+				t.Fatalf("Expected error code %d, got %d message %s", tc.expErrCode, srvErr.Code(), srvErr.Message())
+			}
+			continue
+		}
+		if tc.expErrCode != codes.OK {
+			t.Fatalf("Expected error %v, got no error", tc.expErrCode)
+		}
+	}
+}