Repository: kubernetes-csi/external-attacher
Pull request #184 (2019-09-28 00:33:31)

Reconcile VolumeAttachment with ListVolumes Published Nodes
Main code-paths ready for review.

Tested with one dynamically provisioned volume with the native PD CSI Driver. After detaching the volume on the cloud provider -> external-attacher logs show correct identification of state change and update of VolumeAttachment object -> external-attacher succeeds with immediate reattachment of volume.

WIP:
- [x] Depends on CSI Spec `1.2`.
- [x] Depends on https://github.com/kubernetes/kubernetes/pull/83593
- [x] Only reconcile volumes if plugin supports `ListVolumePublishedNodes` capability
- [x] Handle migration cases for volume ID
- [x] Update logging to include verbosity
- [x] Write some unit tests

/kind feature
/assign @jsafrane @msau42 

```release-note
Reconciles VolumeAttachment attachment status with actual back-end volume attachment state if plugin supports LIST_VOLUMES_PUBLISHED_NODES capability.
```


k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-csi/external-attacher/pull/184#" title="Author self-approved">davidz627</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **saad-ali**
You can assign the PR to them by writing `/assign @saad-ali` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-csi%2Fexternal-attacher).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/kubernetes-csi/external-attacher/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["saad-ali"]} -->
k8s-ci-robot: @davidz627: The following test **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-kubernetes-csi-external-attacher-unit | 08716f83ccb33f7a4d8172e4ad6b8f4f54e6ba1b | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-unit/1177743226577096711) | `/test pull-kubernetes-csi-external-attacher-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-csi&repo=external-attacher&pr=184). [Your PR dashboard](https://gubernator.k8s.io/pr/davidz627). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-csi/external-attacher/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-csi/external-attacher/pull/184#" title="Author self-approved">davidz627</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **jsafrane**
You can assign the PR to them by writing `/assign @jsafrane` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-csi%2Fexternal-attacher).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/kubernetes-csi/external-attacher/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["jsafrane"]} -->
msau42: cc @misterikkit
Do we also want to handle the reverse scenario? Detach if it got attached out of band?
davidz627: > Do we also want to handle the reverse scenario? Detach if it got attached out of band?

I thought we didn't want to do this for data safety reasons (don't want to detach a disk if it may be in use)
msau42: I think to handle the dangling volumes case we may want to do something like that. Maybe for data safety can we do something like only detach if we're trying to attach it to a different node? cc @gnufied 
k8s-ci-robot: @davidz627: The following test **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-kubernetes-csi-external-attacher-unit | 363c2fb0d14c96dc3e914dc42d15564d10d3b1fe | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-unit/1180272029631254529) | `/test pull-kubernetes-csi-external-attacher-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-csi&repo=external-attacher&pr=184). [Your PR dashboard](https://gubernator.k8s.io/pr/davidz627). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-csi/external-attacher/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
msau42: /assign @misterikkit 
k8s-ci-robot: @davidz627: The following test **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-kubernetes-csi-external-attacher-unit | 00dba93777fdb848d3b34568025b5274e241e6a3 | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-unit/1181339438563725312) | `/test pull-kubernetes-csi-external-attacher-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-csi&repo=external-attacher&pr=184). [Your PR dashboard](https://gubernator.k8s.io/pr/davidz627). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-csi/external-attacher/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @davidz627: The following test **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-kubernetes-csi-external-attacher-unit | 74709374091d8281d12f582ee00717337d3ac22b | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-unit/1181353910258896898) | `/test pull-kubernetes-csi-external-attacher-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-csi&repo=external-attacher&pr=184). [Your PR dashboard](https://gubernator.k8s.io/pr/davidz627). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-csi/external-attacher/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
davidz627: @msau42 @misterikkit @jsafrane 
This is ready for review. Only reason it is still WIP is because of the dependencies that need to be updated and vendored in. The code is for the actual feature is complete :+1: 
k8s-ci-robot: @davidz627: The following tests **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-kubernetes-csi-external-attacher-unit | 74709374091d8281d12f582ee00717337d3ac22b | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-unit/1181353910258896898) | `/test pull-kubernetes-csi-external-attacher-unit`
pull-kubernetes-csi-external-attacher-1-15-on-kubernetes-1-15 | b936885ee12e2305968bffcf36a065be2ce50c4b | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-1-15-on-kubernetes-1-15/1182037547497295878) | `/test pull-kubernetes-csi-external-attacher-1-15-on-kubernetes-1-15`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-csi&repo=external-attacher&pr=184). [Your PR dashboard](https://gubernator.k8s.io/pr/davidz627). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-csi/external-attacher/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @davidz627: The following tests **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-kubernetes-csi-external-attacher-unit | 74709374091d8281d12f582ee00717337d3ac22b | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-unit/1181353910258896898) | `/test pull-kubernetes-csi-external-attacher-unit`
pull-kubernetes-csi-external-attacher-1-15-on-kubernetes-1-15 | b936885ee12e2305968bffcf36a065be2ce50c4b | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-1-15-on-kubernetes-1-15/1182037547497295878) | `/test pull-kubernetes-csi-external-attacher-1-15-on-kubernetes-1-15`
pull-kubernetes-csi-external-attacher-1-14-on-kubernetes-1-14 | b936885ee12e2305968bffcf36a065be2ce50c4b | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-1-14-on-kubernetes-1-14/1182037547497295876) | `/test pull-kubernetes-csi-external-attacher-1-14-on-kubernetes-1-14`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-csi&repo=external-attacher&pr=184). [Your PR dashboard](https://gubernator.k8s.io/pr/davidz627). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-csi/external-attacher/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @davidz627: The following tests **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-kubernetes-csi-external-attacher-1-15-on-kubernetes-1-15 | b936885ee12e2305968bffcf36a065be2ce50c4b | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-1-15-on-kubernetes-1-15/1182037547497295878) | `/test pull-kubernetes-csi-external-attacher-1-15-on-kubernetes-1-15`
pull-kubernetes-csi-external-attacher-1-14-on-kubernetes-1-14 | b936885ee12e2305968bffcf36a065be2ce50c4b | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-1-14-on-kubernetes-1-14/1182037547497295876) | `/test pull-kubernetes-csi-external-attacher-1-14-on-kubernetes-1-14`
pull-kubernetes-csi-external-attacher-unit | b936885ee12e2305968bffcf36a065be2ce50c4b | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-unit/1182037547497295877) | `/test pull-kubernetes-csi-external-attacher-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-csi&repo=external-attacher&pr=184). [Your PR dashboard](https://gubernator.k8s.io/pr/davidz627). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-csi/external-attacher/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
davidz627: @jsafrane @misterikkit @msau42 PTAL this is 100% ready for review
davidz627: > A volume was just attached by the handler, yet this reconciler is processing slightly older ListVolumes and marks the volume as detached.

There are two directions of race condition to think about:
1. We have old ListVolumes ("actual state" has changed) when we compare that state to VolumeAttachment

Lets say actual state was updated from `x` to `y` but we have gotten an "old" actual state `x` and the VA has state `y`. In this case we would "incorrectly" reconcile `VA` to state `x`. This would then get resolved in the next loop as we would get actual state `y` and reconcile `VA` to `y`. Though this is probably not ideal to wait an entire reconcile cycle and could cause errors.

To fix we can reverse the direction and always get the `VA` first before the actual state. This way we have two possibilities:
1a.  the `VA` changes between get and update and we catch it in the second case below
1b. The `VA` is up to date, we get actual state `x`, some other controller changes actual state to `y` before we update the `VA` then:
1bi. `VA` gets updated before we can change it and we're in case 2.
1bii. We set the "old" `VA` to state `x` and the controller that changed state to `y` can now go and change `VA` state to `y` still.

2. We have an up to date ListVolumes (actual state is accurate) when we compare that state to a old VolumeAttachment
  - This should be easy to catch by checking for `ResourceVersion` and making sure that hasn't changed.
msau42: If all of the controller paths are calling either Update() or Patch() with resource version, then I think that should handle the races?

The only issue is in case 1bii, this would probably cause the VA update in the attach handler to fail, and we would have to retry the attach operation, but I'm not sure how big of an issue that is since the driver should be idempotent, and I'm not sure how frequent this race will be.

We could potentially reduce the window of a race by checking if an operation is in progress, and skipping that volume for this loop or having some other per-volume locking mechanism.

k8s-ci-robot: @davidz627: The following test **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-kubernetes-csi-external-attacher-unit | 080839fd0fddfbba03a88ec8b3477a6fab75cac5 | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-unit/1205645782472986627) | `/test pull-kubernetes-csi-external-attacher-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-csi&repo=external-attacher&pr=184). [Your PR dashboard](https://gubernator.k8s.io/pr/davidz627). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-csi/external-attacher/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
jsafrane: What if `ReconcileVA` does not update `VolumeAttachment` that something is wrong, but puts the VA silently into the controller workqueue, marked as "force resync"*? The attacher then calls ControllerPublish / ControllerUnpublish first and updates VA only if it failed using already existing code.

*) `ReconcileVA` needs to mark the VA so VA handler knows it should call ControllerPublish  even when the volume looks attached here:
https://github.com/kubernetes-csi/external-attacher/blob/b0b09820457732ebf7588ad111ffd88ea89812ed/pkg/controller/csi_handler.go#L108-L113
(+similar check in detach?)

There won't be any race + volume attachments get fixed automatically, however, user (or A/D controller) won't get a short blip that volume got detached and back attached. 
davidz627: Thanks for the suggestion Jan. After some careful thought here are the reasons why I think we should continue with the current separate reconciler model:
1) From a conceptual standpoint it is more sound to me to have one controller that's job is to reconcile actual state to the desired state (attacher) and a separate loop that reconciles the known state to the actual state (reconcileVA). This is simple and results in eventual consistency even if there are race conditions. The model of injecting imperative forced re-syncs to the attacher work-queue breaks abstractions and increases complexity within the attacher (now it has to branch on this force resync)
2) I think a "short blip that volume got detached and back attached." is actually desirable from a debugging/logs/user perspective since we can tell that the actual state got out of sync with the `VolumeAttachment` state and we can see the timeline of when that was detected + remedied. This is better than trying to fix it silently IMO. If the fix goes wrong it is harder to figure out at what step things broke down.
  2a) In the case when we have a `VolumeAttachment.Status.Attached=true` but the actual state has the volume detached. If there is a long-running `Attach` error that happens every loop, we would be stuck in a state retrying `Attach` in the background while not updating the state for the user or any other controllers that may be watching this object.

I can definitely be convinced of the force resync model but the above are my major concerns. In summary, it feels like a smell to have a "force" something flag as well as to hide actual state (when we know it) from the user and attempt to silently fix it. Let me know what you think.
jsafrane: > From a conceptual standpoint it is more sound to me to have one controller that's job is to reconcile actual state to the desired state (attacher) and a separate loop that reconciles the known state to the actual state (reconcileVA). This is simple and results in eventual consistency even if there are race conditions.

Not really. I can agree that `va.Status.Attached` gets eventually fixed even with races. What about `va.Status.AttachmentMetadata`? You don't get them from CSI via `ListVolumes`. The first `Attached = true` must be saved by VA handler, together with the metadata. And I would argue that all of them should be saved by it, what if `AttachmentMetadata` changes? (Not that we have facility in A/D controller / kubelet to use that information, but that's a separate issue).

So, can the `ListVolume loop write only `Attached = false` and ignore ` = true`?
* During attach, if a volume is attached and VA is not yet marked so, we can be sure that the VA is in the attacher workqueue.
* During detach, if a volume is attached and VA is already marked as deched, that means that Detach succeeded and removed its finializer from VA and VA most likely already does not exists (it had no finalizer and set deletion timestamp).

> I think a "short blip that volume got detached and back attached." is actually desirable from a debugging/logs/user perspective 

I agree.

>  the case when we have a VolumeAttachment.Status.Attached=true but the actual state has the volume detached. If there is a long-running Attach error that happens every loop, we would be stuck in a state retrying Attach in the background.

That could be fixed if we wanted...

davidz627: Ack. I changed the functionality to add volumes back to the workqueue with a forced resync flag. This causes intended behavior for reattaching "accidentally detached" volumes with minimal API updates and no known races. For the detach workflow it's unlikely that the forced resync is necessary as it's very diffiult to get into a state where the `Status.Attached=false` and the finalizer still exists and the volume is still attached. However, I don't think its wrong to cover this case as well anyway.
davidz627: /assign @misterikkit 
k8s-ci-robot: @davidz627: The following test **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-kubernetes-csi-external-attacher-unit | 836557e8db36fdf583d513a455e8e12c4a7042fa | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-unit/1207388530599268354) | `/test pull-kubernetes-csi-external-attacher-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-csi&repo=external-attacher&pr=184). [Your PR dashboard](https://gubernator.k8s.io/pr/davidz627). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-csi/external-attacher/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @davidz627: The following test **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-kubernetes-csi-external-attacher-unit | fdbf3e18e6e05c5f80fb30bb217b0603bd5fda1f | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/kubernetes-csi_external-attacher/184/pull-kubernetes-csi-external-attacher-unit/1207785146011357184) | `/test pull-kubernetes-csi-external-attacher-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-csi&repo=external-attacher&pr=184). [Your PR dashboard](https://gubernator.k8s.io/pr/davidz627). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-csi/external-attacher/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/cmd/csi-attacher/main.go b/cmd/csi-attacher/main.go
index 87f3bd9b..a085c781 100644
--- a/cmd/csi-attacher/main.go
+++ b/cmd/csi-attacher/main.go
@@ -63,6 +63,8 @@ var (
 
 	enableLeaderElection    = flag.Bool("leader-election", false, "Enable leader election.")
 	leaderElectionNamespace = flag.String("leader-election-namespace", "", "Namespace where the leader election resource lives. Defaults to the pod namespace if not set.")
+
+	reconcileSync = flag.Duration("reconcile-sync", 1*time.Minute, "Resync interval of the VolumeAttachment reconciler.")
 )
 
 var (
@@ -148,8 +150,9 @@ func main() {
 			nodeLister := factory.Core().V1().Nodes().Lister()
 			vaLister := factory.Storage().V1beta1().VolumeAttachments().Lister()
 			csiNodeLister := factory.Storage().V1beta1().CSINodes().Lister()
-			attacher := attacher.NewAttacher(csiConn)
-			handler = controller.NewCSIHandler(clientset, csiAttacher, attacher, pvLister, nodeLister, csiNodeLister, vaLister, timeout, supportsReadOnly, csitrans.New())
+			volAttacher := attacher.NewAttacher(csiConn)
+			CSIVolumeLister := attacher.NewVolumeLister(csiConn)
+			handler = controller.NewCSIHandler(clientset, csiAttacher, volAttacher, CSIVolumeLister, pvLister, nodeLister, csiNodeLister, vaLister, timeout, supportsReadOnly, csitrans.New())
 			klog.V(2).Infof("CSI driver supports ControllerPublishUnpublish, using real CSI handler")
 		} else {
 			handler = controller.NewTrivialHandler(clientset)
@@ -157,6 +160,15 @@ func main() {
 		}
 	}
 
+	slvpn, err := supportsListVolumesPublishedNodes(ctx, csiConn)
+	if err != nil {
+		klog.Errorf("Failed to check if driver supports ListVolumesPublishedNodes, assuming it does not: %v", err)
+	}
+
+	if slvpn {
+		klog.V(2).Infof("CSI driver supports list volumes published nodes. Using capability to reconcile volume attachment objects with actual backend state")
+	}
+
 	ctrl := controller.NewCSIAttachController(
 		clientset,
 		csiAttacher,
@@ -165,6 +177,8 @@ func main() {
 		factory.Core().V1().PersistentVolumes(),
 		workqueue.NewItemExponentialFailureRateLimiter(*retryIntervalStart, *retryIntervalMax),
 		workqueue.NewItemExponentialFailureRateLimiter(*retryIntervalStart, *retryIntervalMax),
+		slvpn,
+		*reconcileSync,
 	)
 
 	run := func(ctx context.Context) {
@@ -208,6 +222,15 @@ func supportsControllerPublish(ctx context.Context, csiConn *grpc.ClientConn) (s
 	return supportsControllerPublish, supportsPublishReadOnly, nil
 }
 
+func supportsListVolumesPublishedNodes(ctx context.Context, csiConn *grpc.ClientConn) (bool, error) {
+	caps, err := rpc.GetControllerCapabilities(ctx, csiConn)
+	if err != nil {
+		return false, fmt.Errorf("failed to get controller capabilities: %v", err)
+	}
+
+	return caps[csi.ControllerServiceCapability_RPC_LIST_VOLUMES] && caps[csi.ControllerServiceCapability_RPC_LIST_VOLUMES_PUBLISHED_NODES], nil
+}
+
 func supportsPluginControllerService(ctx context.Context, csiConn *grpc.ClientConn) (bool, error) {
 	caps, err := rpc.GetPluginCapabilities(ctx, csiConn)
 	if err != nil {
diff --git a/pkg/attacher/lister.go b/pkg/attacher/lister.go
new file mode 100644
index 00000000..707c547d
--- /dev/null
+++ b/pkg/attacher/lister.go
@@ -0,0 +1,64 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+Licensed 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 attacher
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+
+	"google.golang.org/grpc"
+)
+
+type CSIVolumeLister struct {
+	conn *grpc.ClientConn
+}
+
+// NewVolumeLister provides a new VolumeLister object.
+func NewVolumeLister(conn *grpc.ClientConn) *CSIVolumeLister {
+	return &CSIVolumeLister{
+		conn: conn,
+	}
+}
+
+func (a *CSIVolumeLister) ListVolumes(ctx context.Context) (map[string]([]string), error) {
+	client := csi.NewControllerClient(a.conn)
+
+	p := map[string][]string{}
+
+	tok := ""
+	for {
+		rsp, err := client.ListVolumes(ctx, &csi.ListVolumesRequest{
+			StartingToken: tok,
+		})
+		if err != nil {
+			return nil, fmt.Errorf("failed to list volumes: %v", err)
+		}
+
+		for _, e := range rsp.Entries {
+			p[e.GetVolume().VolumeId] = e.Status.PublishedNodeIds
+		}
+		tok = rsp.NextToken
+
+		if len(tok) == 0 {
+			break
+		}
+	}
+
+	return p, nil
+}
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index 543bb677..a59a283c 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.go
@@ -18,6 +18,7 @@ package controller
 
 import (
 	"fmt"
+	"time"
 
 	"k8s.io/klog"
 
@@ -51,6 +52,9 @@ type CSIAttachController struct {
 	vaListerSynced cache.InformerSynced
 	pvLister       corelisters.PersistentVolumeLister
 	pvListerSynced cache.InformerSynced
+
+	shouldReconcileVolumeAttachment bool
+	reconcileSync                   time.Duration
 }
 
 // Handler is responsible for handling VolumeAttachment events from informer.
@@ -67,22 +71,26 @@ type Handler interface {
 	SyncNewOrUpdatedVolumeAttachment(va *storage.VolumeAttachment)
 
 	SyncNewOrUpdatedPersistentVolume(pv *v1.PersistentVolume)
+
+	ReconcileVA() error
 }
 
 // NewCSIAttachController returns a new *CSIAttachController
-func NewCSIAttachController(client kubernetes.Interface, attacherName string, handler Handler, volumeAttachmentInformer storageinformers.VolumeAttachmentInformer, pvInformer coreinformers.PersistentVolumeInformer, vaRateLimiter, paRateLimiter workqueue.RateLimiter) *CSIAttachController {
+func NewCSIAttachController(client kubernetes.Interface, attacherName string, handler Handler, volumeAttachmentInformer storageinformers.VolumeAttachmentInformer, pvInformer coreinformers.PersistentVolumeInformer, vaRateLimiter, paRateLimiter workqueue.RateLimiter, shouldReconcileVolumeAttachment bool, reconcileSync time.Duration) *CSIAttachController {
 	broadcaster := record.NewBroadcaster()
 	broadcaster.StartRecordingToSink(&corev1.EventSinkImpl{Interface: client.CoreV1().Events(v1.NamespaceAll)})
 	var eventRecorder record.EventRecorder
 	eventRecorder = broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: fmt.Sprintf("csi-attacher %s", attacherName)})
 
 	ctrl := &CSIAttachController{
-		client:        client,
-		attacherName:  attacherName,
-		handler:       handler,
-		eventRecorder: eventRecorder,
-		vaQueue:       workqueue.NewNamedRateLimitingQueue(vaRateLimiter, "csi-attacher-va"),
-		pvQueue:       workqueue.NewNamedRateLimitingQueue(paRateLimiter, "csi-attacher-pv"),
+		client:                          client,
+		attacherName:                    attacherName,
+		handler:                         handler,
+		eventRecorder:                   eventRecorder,
+		vaQueue:                         workqueue.NewNamedRateLimitingQueue(vaRateLimiter, "csi-attacher-va"),
+		pvQueue:                         workqueue.NewNamedRateLimitingQueue(paRateLimiter, "csi-attacher-pv"),
+		shouldReconcileVolumeAttachment: shouldReconcileVolumeAttachment,
+		reconcileSync:                   reconcileSync,
 	}
 
 	volumeAttachmentInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -122,6 +130,15 @@ func (ctrl *CSIAttachController) Run(workers int, stopCh <-chan struct{}) {
 		go wait.Until(ctrl.syncPV, 0, stopCh)
 	}
 
+	if ctrl.shouldReconcileVolumeAttachment {
+		go wait.Until(func() {
+			err := ctrl.handler.ReconcileVA()
+			if err != nil {
+				klog.Errorf("Failed to reconcile volume attachments: %v", err)
+			}
+		}, ctrl.reconcileSync, stopCh)
+	}
+
 	<-stopCh
 }
 
diff --git a/pkg/controller/csi_handler.go b/pkg/controller/csi_handler.go
index 74ed3857..6e93a195 100644
--- a/pkg/controller/csi_handler.go
+++ b/pkg/controller/csi_handler.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"k8s.io/klog"
@@ -42,6 +43,16 @@ type AttacherCSITranslator interface {
 	RepairVolumeHandle(pluginName, volumeHandle, nodeID string) (string, error)
 }
 
+// Lister implements list operations against a remote CSI driver.
+type VolumeLister interface {
+	// ListVolumes calls ListVolumes on the driver and returns a map with keys
+	// of VolumeID and values of the list of Node IDs that volume is published
+	// on
+	ListVolumes(ctx context.Context) (map[string][]string, error)
+}
+
+var _ VolumeLister = &attacher.CSIVolumeLister{}
+
 // csiHandler is a handler that calls CSI to attach/detach volume.
 // It adds finalizer to VolumeAttachment instance to make sure they're detached
 // before deletion.
@@ -49,11 +60,14 @@ type csiHandler struct {
 	client                  kubernetes.Interface
 	attacherName            string
 	attacher                attacher.Attacher
+	CSIVolumeLister         VolumeLister
 	pvLister                corelisters.PersistentVolumeLister
 	nodeLister              corelisters.NodeLister
 	csiNodeLister           storagelisters.CSINodeLister
 	vaLister                storagelisters.VolumeAttachmentLister
 	vaQueue, pvQueue        workqueue.RateLimitingInterface
+	forceSync               map[string]bool
+	forceSyncMux            sync.Mutex
 	timeout                 time.Duration
 	supportsPublishReadOnly bool
 	translator              AttacherCSITranslator
@@ -66,6 +80,7 @@ func NewCSIHandler(
 	client kubernetes.Interface,
 	attacherName string,
 	attacher attacher.Attacher,
+	CSIVolumeLister VolumeLister,
 	pvLister corelisters.PersistentVolumeLister,
 	nodeLister corelisters.NodeLister,
 	csiNodeLister storagelisters.CSINodeLister,
@@ -78,6 +93,7 @@ func NewCSIHandler(
 		client:                  client,
 		attacherName:            attacherName,
 		attacher:                attacher,
+		CSIVolumeLister:         CSIVolumeLister,
 		pvLister:                pvLister,
 		nodeLister:              nodeLister,
 		csiNodeLister:           csiNodeLister,
@@ -85,6 +101,8 @@ func NewCSIHandler(
 		timeout:                 *timeout,
 		supportsPublishReadOnly: supportsPublishReadOnly,
 		translator:              translator,
+		forceSync:               map[string]bool{},
+		forceSyncMux:            sync.Mutex{},
 	}
 }
 
@@ -93,6 +111,103 @@ func (h *csiHandler) Init(vaQueue workqueue.RateLimitingInterface, pvQueue workq
 	h.pvQueue = pvQueue
 }
 
+// ReconcileVA lists volumes from the CSI Driver and reconciles the attachment
+// status with the corresponding VolumeAttachment object. If the attachment
+// status of the volume is different from the state on the VolumeAttachment the
+// VolumeAttachment object is patched to the correct state.
+func (h *csiHandler) ReconcileVA() error {
+	klog.V(4).Info("Reconciling VolumeAttachments with driver backend state")
+
+	ctx, cancel := context.WithTimeout(context.Background(), h.timeout)
+	defer cancel()
+
+	// Loop over all volume attachment objects
+	vas, err := h.vaLister.List(labels.Everything())
+	if err != nil {
+		return errors.New("failed to list all VolumeAttachment objects")
+	}
+
+	published, err := h.CSIVolumeLister.ListVolumes(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to ListVolumes: %v", err)
+	}
+
+	for _, va := range vas {
+		nodeID, ok := va.Annotations[vaNodeIDAnnotation]
+		if !ok {
+			klog.Warningf("Failed to find node ID in VolumeAttachment %s annotation", va.Name)
+			continue
+		}
+		pvSpec, err := h.getProcessedPVSpec(va)
+		if err != nil {
+			klog.Warningf("Failed to get PV Spec: %v", err)
+			continue
+		}
+
+		source, err := getCSISource(pvSpec)
+		if err != nil {
+			klog.Warningf("Failed to get CSI Source: %v", err)
+			continue
+		}
+
+		volumeHandle, _, err := GetVolumeHandle(source)
+		if err != nil {
+			klog.Warningf("Failed to get volume handle: %v", err)
+			continue
+		}
+		attachedStatus := va.Status.Attached
+
+		volumeHandle, err = h.translator.RepairVolumeHandle(source.Driver, volumeHandle, nodeID)
+		if err != nil {
+			klog.Warningf("Failed to repair volume handle %s for driver %s: %v", volumeHandle, source.Driver, err)
+			continue
+		}
+
+		// Check whether the volume is published to this node
+		found := false
+		for _, gotNodeID := range published[volumeHandle] {
+			if gotNodeID == nodeID {
+				found = true
+				break
+			}
+		}
+
+		// If ListVolumes Attached Status is different, add to shared workQueue.
+		if attachedStatus != found {
+			klog.Warningf("VA %s for volume %s has attached status %v but actual state %v. Adding back to VA queue for forced reprocessing", va.Name, volumeHandle, attachedStatus, found)
+			// Add this item to the vaQueue with forceSync so that it is force
+			// processed again, we avoid UPDATE on the VA or forcing a direct
+			// attach/detach as to avoid race conditions with the main attacher
+			// queue
+			h.setForceSync(va.Name)
+			h.vaQueue.Add(va.Name)
+		}
+	}
+	return nil
+}
+
+// setForceSync sets the intention that next time the VolumeAttachment
+// referenced by vaName is processed on the VA queue that attach or detach will
+// proceed even when the VA.Status.Attached mayalready show the desired state
+func (h *csiHandler) setForceSync(vaName string) {
+	h.forceSyncMux.Lock()
+	defer h.forceSyncMux.Unlock()
+	h.forceSync[vaName] = true
+}
+
+// consumeForceSync is used to check whether forceSync was set for the VA
+// referenced by vaName. It will then remove the forceSync intention so that the
+// VA will only be forceSync-ed once per request
+func (h *csiHandler) consumeForceSync(vaName string) bool {
+	h.forceSyncMux.Lock()
+	defer h.forceSyncMux.Unlock()
+	s, ok := h.forceSync[vaName]
+	if ok {
+		delete(h.forceSync, vaName)
+	}
+	return s
+}
+
 func (h *csiHandler) SyncNewOrUpdatedVolumeAttachment(va *storage.VolumeAttachment) {
 	klog.V(4).Infof("CSIHandler: processing VA %q", va.Name)
 
@@ -114,8 +229,8 @@ func (h *csiHandler) SyncNewOrUpdatedVolumeAttachment(va *storage.VolumeAttachme
 }
 
 func (h *csiHandler) syncAttach(va *storage.VolumeAttachment) error {
-	if va.Status.Attached {
-		// Volume is attached, there is nothing to be done.
+	if !h.consumeForceSync(va.Name) && va.Status.Attached {
+		// Volume is attached and no force sync, there is nothing to be done.
 		klog.V(4).Infof("%q is already attached", va.Name)
 		return nil
 	}
@@ -146,7 +261,7 @@ func (h *csiHandler) syncAttach(va *storage.VolumeAttachment) error {
 
 func (h *csiHandler) syncDetach(va *storage.VolumeAttachment) error {
 	klog.V(4).Infof("Starting detach operation for %q", va.Name)
-	if !h.hasVAFinalizer(va) {
+	if !h.consumeForceSync(va.Name) && !h.hasVAFinalizer(va) {
 		klog.V(4).Infof("%q is already detached", va.Name)
 		return nil
 	}
@@ -243,14 +358,41 @@ func (h *csiHandler) hasVAFinalizer(va *storage.VolumeAttachment) bool {
 	return false
 }
 
-func getCSISource(pv *v1.PersistentVolume) (*v1.CSIPersistentVolumeSource, error) {
-	if pv == nil {
-		return nil, fmt.Errorf("could not get CSI source, pv was nil")
+func getCSISource(pvSpec *v1.PersistentVolumeSpec) (*v1.CSIPersistentVolumeSource, error) {
+	if pvSpec == nil {
+		return nil, errors.New("could not get CSI source, pv spec was nil")
+	}
+	if pvSpec.CSI != nil {
+		return pvSpec.CSI, nil
 	}
-	if pv.Spec.CSI != nil {
-		return pv.Spec.CSI, nil
+	return nil, errors.New("pv spec contained non-csi source that was not migrated")
+}
+
+func (h *csiHandler) getProcessedPVSpec(va *storage.VolumeAttachment) (*v1.PersistentVolumeSpec, error) {
+	if va.Spec.Source.PersistentVolumeName != nil {
+		if va.Spec.Source.InlineVolumeSpec != nil {
+			return nil, errors.New("both InlineCSIVolumeSource and PersistentVolumeName specified in VA source")
+		}
+		pv, err := h.pvLister.Get(*va.Spec.Source.PersistentVolumeName)
+		if err != nil {
+			return nil, err
+		}
+		if h.translator.IsPVMigratable(pv) {
+			pv, err = h.translator.TranslateInTreePVToCSI(pv)
+			if err != nil {
+				return nil, fmt.Errorf("failed to TranslateInTreePVToCSI(%v): %v", pv, err)
+			}
+		}
+		return &pv.Spec, nil
+	} else if va.Spec.Source.InlineVolumeSpec != nil {
+		if va.Spec.Source.InlineVolumeSpec.CSI == nil {
+			return nil, errors.New("inline volume spec contains nil CSI source")
+		}
+
+		return va.Spec.Source.InlineVolumeSpec, nil
+	} else {
+		return nil, errors.New("neither InlineCSIVolumeSource nor PersistentVolumeName specified in VA source")
 	}
-	return nil, fmt.Errorf("pv contained non-csi source that was not migrated")
 }
 
 func (h *csiHandler) csiAttach(va *storage.VolumeAttachment) (*storage.VolumeAttachment, map[string]string, error) {
@@ -286,7 +428,7 @@ func (h *csiHandler) csiAttach(va *storage.VolumeAttachment) (*storage.VolumeAtt
 
 		// Both csiSource and pvSpec could be translated here if the PV was
 		// migrated
-		csiSource, err = getCSISource(pv)
+		csiSource, err = getCSISource(&pv.Spec)
 		if err != nil {
 			return va, nil, err
 		}
@@ -371,7 +513,7 @@ func (h *csiHandler) csiDetach(va *storage.VolumeAttachment) (*storage.VolumeAtt
 				return va, fmt.Errorf("failed to translate in tree pv to CSI: %v", err)
 			}
 		}
-		csiSource, err = getCSISource(pv)
+		csiSource, err = getCSISource(&pv.Spec)
 		if err != nil {
 			return va, err
 		}
diff --git a/pkg/controller/trivial_handler.go b/pkg/controller/trivial_handler.go
index 931fdbf8..d5b4aba6 100644
--- a/pkg/controller/trivial_handler.go
+++ b/pkg/controller/trivial_handler.go
@@ -17,7 +17,7 @@ limitations under the License.
 package controller
 
 import (
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	storage "k8s.io/api/storage/v1beta1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/util/workqueue"
@@ -45,6 +45,10 @@ func (h *trivialHandler) Init(vaQueue workqueue.RateLimitingInterface, pvQueue w
 	h.pvQueue = pvQueue
 }
 
+func (h *trivialHandler) ReconcileVA() error {
+	return nil
+}
+
 func (h *trivialHandler) SyncNewOrUpdatedVolumeAttachment(va *storage.VolumeAttachment) {
 	klog.V(4).Infof("Trivial sync[%s] started", va.Name)
 	if !va.Status.Attached {


Test patch:
diff --git a/pkg/controller/csi_handler_test.go b/pkg/controller/csi_handler_test.go
index 20557904..00461f2c 100644
--- a/pkg/controller/csi_handler_test.go
+++ b/pkg/controller/csi_handler_test.go
@@ -34,7 +34,6 @@ import (
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	core "k8s.io/client-go/testing"
-	csitrans "k8s.io/csi-translation-lib"
 	"k8s.io/klog"
 )
 
@@ -51,33 +50,35 @@ var (
 
 var timeout = 10 * time.Millisecond
 
-func csiHandlerFactory(client kubernetes.Interface, informerFactory informers.SharedInformerFactory, csi attacher.Attacher) Handler {
+func csiHandlerFactory(client kubernetes.Interface, informerFactory informers.SharedInformerFactory, csi attacher.Attacher, lister VolumeLister) Handler {
 	return NewCSIHandler(
 		client,
 		testAttacherName,
 		csi,
+		lister,
 		informerFactory.Core().V1().PersistentVolumes().Lister(),
 		informerFactory.Core().V1().Nodes().Lister(),
 		informerFactory.Storage().V1beta1().CSINodes().Lister(),
 		informerFactory.Storage().V1beta1().VolumeAttachments().Lister(),
 		&timeout,
 		true, /* supports PUBLISH_READONLY */
-		csitrans.New(),
+		fakeInTreeToCSITranslator{},
 	)
 }
 
-func csiHandlerFactoryNoReadOnly(client kubernetes.Interface, informerFactory informers.SharedInformerFactory, csi attacher.Attacher) Handler {
+func csiHandlerFactoryNoReadOnly(client kubernetes.Interface, informerFactory informers.SharedInformerFactory, csi attacher.Attacher, lister VolumeLister) Handler {
 	return NewCSIHandler(
 		client,
 		testAttacherName,
 		csi,
+		lister,
 		informerFactory.Core().V1().PersistentVolumes().Lister(),
 		informerFactory.Core().V1().Nodes().Lister(),
 		informerFactory.Storage().V1beta1().CSINodes().Lister(),
 		informerFactory.Storage().V1beta1().VolumeAttachments().Lister(),
 		&timeout,
 		false, /* does not support PUBLISH_READONLY */
-		csitrans.New(),
+		fakeInTreeToCSITranslator{},
 	)
 }
 
@@ -1276,6 +1277,64 @@ func TestCSIHandler(t *testing.T) {
 	runTests(t, csiHandlerFactory, tests)
 }
 
+func TestCSIHandlerReconcileVA(t *testing.T) {
+	nID := map[string]string{
+		vaNodeIDAnnotation: testNodeName,
+	}
+
+	tests := []testCase{
+		{
+			name: "va attached actual state not attached",
+			initialObjects: []runtime.Object{
+				va(true /*attached*/, "" /*finalizer*/, nID /*annotations*/),
+				pvWithFinalizer(),
+			},
+			listerResponse: map[string][]string{
+				// Intentionally empty
+			},
+			expectedCSICalls: []csiCall{
+				{"attach", testVolumeHandle, testNodeID, nil, nil, false, nil, false, nil, 0},
+			},
+		},
+		{
+			name: "va attached actual state attached",
+			initialObjects: []runtime.Object{
+				va(true /*attached*/, "" /*finalizer*/, nID /*annotations*/),
+				pvWithFinalizer(),
+			},
+			listerResponse: map[string][]string{
+				testVolumeHandle: []string{testNodeName},
+			},
+			expectedActions: []core.Action{
+				// Intentionally empty
+			},
+		},
+		{
+			name: "va not attached actual state attached",
+			initialObjects: []runtime.Object{
+				va(false /*attached*/, "" /*finalizer*/, nID /*annotations*/),
+				pvWithFinalizer(),
+			},
+			listerResponse: map[string][]string{
+				testVolumeHandle: []string{testNodeName},
+			},
+			expectedActions: []core.Action{},
+			expectedCSICalls: []csiCall{
+				{"detach", testVolumeHandle, testNodeID, nil, nil, false, nil, true, nil, 0},
+			},
+		},
+		{
+			name:           "no volume attachments but existing lister response results in no action",
+			initialObjects: []runtime.Object{},
+			listerResponse: map[string][]string{
+				testVolumeHandle: []string{testNodeName},
+			},
+			expectedActions: []core.Action{},
+		},
+	}
+	runTests(t, csiHandlerFactory, tests)
+}
+
 func TestCSIHandlerReadOnly(t *testing.T) {
 	vaGroupResourceVersion := schema.GroupVersionResource{
 		Group:    storage.GroupName,
diff --git a/pkg/controller/framework_test.go b/pkg/controller/framework_test.go
index afdc0fb5..cc167327 100644
--- a/pkg/controller/framework_test.go
+++ b/pkg/controller/framework_test.go
@@ -18,6 +18,7 @@ package controller
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"reflect"
 	"strings"
@@ -27,10 +28,8 @@ import (
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/davecgh/go-spew/spew"
 	"github.com/kubernetes-csi/external-attacher/pkg/attacher"
-	"k8s.io/klog"
 
-	"encoding/json"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	storage "k8s.io/api/storage/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -39,6 +38,8 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 	core "k8s.io/client-go/testing"
 	"k8s.io/client-go/util/workqueue"
+	csitrans "k8s.io/csi-translation-lib"
+	"k8s.io/klog"
 )
 
 // This is an unit test framework. It is heavily inspired by serviceaccount
@@ -72,6 +73,8 @@ type testCase struct {
 	expectedActions []core.Action
 	// List of expected CSI calls
 	expectedCSICalls []csiCall
+	// Expected lister response
+	listerResponse map[string][]string
 	// Function to perform additional checks after the test finishes
 	additionalCheck func(t *testing.T, test testCase)
 }
@@ -100,7 +103,7 @@ type csiCall struct {
 	delay time.Duration
 }
 
-type handlerFactory func(client kubernetes.Interface, informerFactory informers.SharedInformerFactory, csi attacher.Attacher) Handler
+type handlerFactory func(client kubernetes.Interface, informerFactory informers.SharedInformerFactory, csi attacher.Attacher, lister VolumeLister) Handler
 
 func runTests(t *testing.T, handlerFactory handlerFactory, tests []testCase) {
 	for _, test := range tests {
@@ -176,8 +179,9 @@ func runTests(t *testing.T, handlerFactory handlerFactory, tests []testCase) {
 
 		// Construct controller
 		csiConnection := &fakeCSIConnection{t: t, calls: test.expectedCSICalls}
-		handler := handlerFactory(client, informers, csiConnection)
-		ctrl := NewCSIAttachController(client, testAttacherName, handler, vaInformer, pvInformer, workqueue.DefaultControllerRateLimiter(), workqueue.DefaultControllerRateLimiter())
+		lister := &fakeLister{t: t, publishedNodes: test.listerResponse}
+		handler := handlerFactory(client, informers, csiConnection, lister)
+		ctrl := NewCSIAttachController(client, testAttacherName, handler, vaInformer, pvInformer, workqueue.DefaultControllerRateLimiter(), workqueue.DefaultControllerRateLimiter(), test.listerResponse != nil, 1*time.Minute)
 
 		// Start the test by enqueueing the right event
 		if test.addedVA != nil {
@@ -213,6 +217,13 @@ func runTests(t *testing.T, handlerFactory handlerFactory, tests []testCase) {
 				// There is still some work in the queue, process it now
 				continue
 			}
+			if test.listerResponse != nil {
+				// Reconcile VA with the actual state
+				err := ctrl.handler.ReconcileVA()
+				if err != nil {
+					t.Errorf("Failed to reconcile Volume Attachment objects: %v", err)
+				}
+			}
 			currentActionCount := len(client.Actions())
 			if currentActionCount < len(test.expectedActions) {
 				if lastReportedActionCount < currentActionCount {
@@ -380,6 +391,15 @@ func vaWithDetachError(va *storage.VolumeAttachment, message string) *storage.Vo
 	return va
 }
 
+type fakeLister struct {
+	t              *testing.T
+	publishedNodes map[string][]string
+}
+
+func (l *fakeLister) ListVolumes(ctx context.Context) (map[string][]string, error) {
+	return l.publishedNodes, nil
+}
+
 // Fake CSIConnection implementation that check that Attach/Detach is called
 // with the right parameters and it returns proper error code and metadata.
 type fakeCSIConnection struct {
@@ -494,3 +514,17 @@ func (f *fakeCSIConnection) Close() error {
 func (f *fakeCSIConnection) Probe(timeout time.Duration) error {
 	return nil
 }
+
+// TODO: Remove hardcoding for GCE tests and make more general
+type fakeInTreeToCSITranslator struct{}
+
+func (f fakeInTreeToCSITranslator) TranslateInTreePVToCSI(pv *v1.PersistentVolume) (*v1.PersistentVolume, error) {
+	t := csitrans.New()
+	return t.TranslateInTreePVToCSI(pv)
+}
+func (f fakeInTreeToCSITranslator) IsPVMigratable(pv *v1.PersistentVolume) bool {
+	return pv.Spec.GCEPersistentDisk != nil
+}
+func (f fakeInTreeToCSITranslator) RepairVolumeHandle(pluginName, volumeHandle, nodeID string) (string, error) {
+	return volumeHandle, nil
+}
diff --git a/pkg/controller/trivial_handler_test.go b/pkg/controller/trivial_handler_test.go
index eae7314f..f21f2e05 100644
--- a/pkg/controller/trivial_handler_test.go
+++ b/pkg/controller/trivial_handler_test.go
@@ -33,7 +33,7 @@ import (
 	core "k8s.io/client-go/testing"
 )
 
-func trivialHandlerFactory(client kubernetes.Interface, informerFactory informers.SharedInformerFactory, csi attacher.Attacher) Handler {
+func trivialHandlerFactory(client kubernetes.Interface, informerFactory informers.SharedInformerFactory, csi attacher.Attacher, lister VolumeLister) Handler {
 	return NewTrivialHandler(client)
 }