diff --git a/pkg/cloud/cloud.go b/pkg/cloud/cloud.go
index 705734c093..783ba2ff29 100644
--- a/pkg/cloud/cloud.go
+++ b/pkg/cloud/cloud.go
@@ -102,6 +102,10 @@ type DiskOptions struct {
 	VolumeType       string
 	IOPSPerGB        int
 	AvailabilityZone string
+	Encrypted        bool
+	// 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
 }
 
 // EC2 abstracts aws.EC2 to facilitate its mocking.
@@ -215,6 +219,11 @@ func (c *cloud) CreateDisk(ctx context.Context, volumeName string, diskOptions *
 		Size:              aws.Int64(capacityGiB),
 		VolumeType:        aws.String(createType),
 		TagSpecifications: []*ec2.TagSpecification{&tagSpec},
+		Encrypted:         aws.Bool(diskOptions.Encrypted),
+	}
+	if len(diskOptions.KmsKeyID) > 0 {
+		request.KmsKeyId = aws.String(diskOptions.KmsKeyID)
+		request.Encrypted = aws.Bool(true)
 	}
 	if iops > 0 {
 		request.Iops = aws.Int64(iops)
@@ -235,16 +244,23 @@ func (c *cloud) CreateDisk(ctx context.Context, volumeName string, diskOptions *
 		return nil, fmt.Errorf("disk size was not returned by CreateVolume")
 	}
 
+	if len(diskOptions.KmsKeyID) > 0 {
+		err := c.waitForCreate(ctx, volumeID)
+		if err != nil {
+			if isAWSErrorVolumeNotFound(err) {
+				return nil, fmt.Errorf("failed to create encrypted volume: the volume disappeared after creation, most likely due to inaccessible KMS encryption key")
+			}
+		}
+	}
+
 	return &Disk{CapacityGiB: size, VolumeID: volumeID, AvailabilityZone: zone}, nil
 }
 
 func (c *cloud) DeleteDisk(ctx context.Context, volumeID string) (bool, error) {
 	request := &ec2.DeleteVolumeInput{VolumeId: &volumeID}
 	if _, err := c.ec2.DeleteVolumeWithContext(ctx, request); err != nil {
-		if awsErr, ok := err.(awserr.Error); ok {
-			if awsErr.Code() == "InvalidVolume.NotFound" {
-				return false, ErrNotFound
-			}
+		if isAWSErrorVolumeNotFound(err) {
+			return false, ErrNotFound
 		}
 		return false, fmt.Errorf("DeleteDisk could not delete volume: %v", err)
 	}
@@ -487,3 +503,56 @@ func (c *cloud) waitForAttachmentState(ctx context.Context, volumeID, state stri
 
 	return wait.ExponentialBackoff(backoff, verifyVolumeFunc)
 }
+
+// waitForCreate waits for volume to be created for encrypted volume only
+// it polls for created volume to check it has not been silently removed by AWS.
+// On a random AWS account (shared among several developers) it took 4s on average.
+func (c *cloud) waitForCreate(ctx context.Context, volumeID string) error {
+	var (
+		checkInterval = 1 * time.Second
+		checkTimeout  = 30 * time.Second
+	)
+
+	request := &ec2.DescribeVolumesInput{
+		VolumeIds: []*string{
+			aws.String(volumeID),
+		},
+	}
+
+	err := wait.Poll(checkInterval, checkTimeout, func() (done bool, err error) {
+		vol, err := c.getVolume(ctx, request)
+		if err != nil {
+			return true, err
+		}
+		if vol.State != nil {
+			switch *vol.State {
+			case "available":
+				// The volume is Available, it won't be deleted now.
+				return true, nil
+			case "creating":
+				return false, nil
+			default:
+				return true, fmt.Errorf("unexpected State of newly created AWS EBS volume %s: %q", volumeID, *vol.State)
+			}
+		}
+		return false, nil
+	})
+
+	return err
+}
+
+// Helper function for describeVolume callers. Tries to retype given error to AWS error
+// and returns true in case the AWS error is "InvalidVolume.NotFound", false otherwise
+func isAWSErrorVolumeNotFound(err error) bool {
+	if err == nil {
+		return false
+	}
+	if awsError, ok := err.(awserr.Error); ok {
+		// https://docs.aws.amazon.com/AWSEC2/latest/APIReference/errors-overview.html
+		if awsError.Code() == "InvalidVolume.NotFound" {
+			return true
+		}
+	}
+
+	return false
+}
diff --git a/pkg/cloud/cloud_test.go b/pkg/cloud/cloud_test.go
index 741691fe09..f9179d80d1 100644
--- a/pkg/cloud/cloud_test.go
+++ b/pkg/cloud/cloud_test.go
@@ -77,6 +77,22 @@ func TestCreateDisk(t *testing.T) {
 			},
 			expErr: fmt.Errorf("CreateVolume generic error"),
 		},
+		{
+			name:       "success: normal with encrypted volume",
+			volumeName: "vol-test-name",
+			diskOptions: &DiskOptions{
+				CapacityBytes:    util.GiBToBytes(1),
+				Tags:             map[string]string{VolumeNameTagKey: "vol-test"},
+				AvailabilityZone: "us-west-2",
+				Encrypted:        true,
+				KmsKeyID:         "arn:aws:kms:us-east-1:012345678910:key/abcd1234-a123-456a-a12b-a123b4cd56ef",
+			},
+			expDisk: &Disk{
+				VolumeID:    "vol-test",
+				CapacityGiB: 1,
+			},
+			expErr: nil,
+		},
 	}
 
 	for _, tc := range testCases {
@@ -90,12 +106,17 @@ func TestCreateDisk(t *testing.T) {
 			vol = &ec2.Volume{
 				VolumeId: aws.String(tc.diskOptions.Tags[VolumeNameTagKey]),
 				Size:     aws.Int64(util.BytesToGiB(tc.diskOptions.CapacityBytes)),
+				State:    aws.String("available"),
 			}
 		}
 
 		ctx := context.Background()
 		mockEC2.EXPECT().CreateVolumeWithContext(gomock.Eq(ctx), gomock.Any()).Return(vol, tc.expErr)
 
+		if tc.diskOptions.Encrypted {
+			mockEC2.EXPECT().DescribeVolumesWithContext(gomock.Eq(ctx), gomock.Any()).Return(&ec2.DescribeVolumesOutput{Volumes: []*ec2.Volume{vol}}, nil)
+		}
+
 		disk, err := c.CreateDisk(ctx, tc.volumeName, tc.diskOptions)
 		if err != nil {
 			if tc.expErr == nil {
diff --git a/pkg/driver/controller.go b/pkg/driver/controller.go
index 1e7a745dbe..cc19d03fe0 100644
--- a/pkg/driver/controller.go
+++ b/pkg/driver/controller.go
@@ -86,12 +86,23 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 		}
 	}
 
+	var (
+		isEncrypted bool
+		kmsKeyId    string
+	)
+	if volumeParams["encrypted"] == "true" {
+		isEncrypted = true
+		kmsKeyId = volumeParams["kmsKeyId"]
+	}
+
 	opts := &cloud.DiskOptions{
 		CapacityBytes:    volSizeBytes,
 		Tags:             map[string]string{cloud.VolumeNameTagKey: volName},
 		VolumeType:       volumeType,
 		IOPSPerGB:        iopsPerGB,
 		AvailabilityZone: zone,
+		Encrypted:        isEncrypted,
+		KmsKeyID:         kmsKeyId,
 	}
 	disk, err = d.cloud.CreateDisk(ctx, volName, opts)
 	if err != nil {
diff --git a/pkg/driver/controller_test.go b/pkg/driver/controller_test.go
index 850e309b01..d565fc5fec 100644
--- a/pkg/driver/controller_test.go
+++ b/pkg/driver/controller_test.go
@@ -182,6 +182,39 @@ func TestCreateVolume(t *testing.T) {
 				Attributes:    map[string]string{"fsType": ""},
 			},
 		},
+		{
+			name: "success with volume encrpytion",
+			req: &csi.CreateVolumeRequest{
+				Name:               "vol-test",
+				CapacityRange:      stdCapRange,
+				VolumeCapabilities: stdVolCap,
+				Parameters: map[string]string{
+					"encrypted": "true",
+				},
+			},
+			expVol: &csi.Volume{
+				CapacityBytes: stdVolSize,
+				Id:            "vol-test",
+				Attributes:    map[string]string{"fsType": ""},
+			},
+		},
+		{
+			name: "success with volume encrpytion with KMS key",
+			req: &csi.CreateVolumeRequest{
+				Name:               "vol-test",
+				CapacityRange:      stdCapRange,
+				VolumeCapabilities: stdVolCap,
+				Parameters: map[string]string{
+					"encrypted": "true",
+					"kmsKeyId":  "arn:aws:kms:us-east-1:012345678910:key/abcd1234-a123-456a-a12b-a123b4cd56ef",
+				},
+			},
+			expVol: &csi.Volume{
+				CapacityBytes: stdVolSize,
+				Id:            "vol-test",
+				Attributes:    map[string]string{"fsType": ""},
+			},
+		},
 	}
 
 	for _, tc := range testCases {