Repository: kubernetes-sigs/cloud-provider-azure
Pull request #2288 (2022-09-08 23:52:11)

VMSS Flex Support: Embed FlexScaleSet logic into existing CCM execution flow
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide/first-contribution.md#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Please label this pull request according to what type of issue you are addressing, especially if this is a release targeted pull request. For reference on required PR/issue labels, read here:
https://git.k8s.io/community/contributors/devel/sig-release/release.md#issuepr-kind-label
3. Ensure you have added or ran the appropriate tests for your PR: https://git.k8s.io/community/contributors/devel/sig-testing/testing.md
4. If you want *faster* PR reviews, read how: https://git.k8s.io/community/contributors/guide/pull-requests.md#best-practices-for-faster-reviews
5. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

#### What type of PR is this?
/kind feature
<!--
Add one of the following kinds:
/kind bug
/kind cleanup
/kind documentation
/kind feature
/kind design

Optionally add one or more of the following kinds if applicable:
/kind api-change
/kind deprecation
/kind failing-test
/kind flake
/kind regression
-->

#### What this PR does / why we need it:
This PR embeds FlexScaleSet logic into existing CCM execution flow. Key points of the change:

1. Change AvailabilitySetNodesCache to NonUniformNodesCache
2. Change isNodeManagedByAvailabilitySet to getVMManagementType
3. Integrate FlexScaleSet logic into existing CCM execution flow
- vmType == standard: only AvSet VMs or standalone VMs (no change)
- vmType == vmss && DisableAvailabilitySetNodes == true && EnableVmssFlexNodes == false: only VMSS Uniform VMs (for current AKS with Uniform Scale Set)
- vmType == vmss && DisableAvailabilitySetNodes == false && EnableVmssFlexNodes == true: mix mode: AvSet + Uniform + Flex


#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Refer to #639 

#### Special notes for your reviewer:

#### Does this PR introduce a user-facing change?
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".

For more information on release notes see: https://git.k8s.io/community/contributors/guide/release-notes.md
-->
```release-note
NONE
```

#### Additional documentation e.g., KEPs (Kubernetes Enhancement Proposals), usage docs, etc.:

<!--
This section can be blank if this pull request does not require a release note.

When adding links which point to resources within git repositories, like
KEPs or supporting documentation, please reference a specific commit and avoid
linking directly to the master branch. This ensures that links reference a
specific point in time, rather than a document that may change over time.

See here for guidance on getting permanent links to files: https://help.github.com/en/articles/getting-permanent-links-to-files

Please use the following format for linking documentation:
- [KEP]: <link>
- [Usage]: <link>
- [Other doc]: <link>
-->
```docs

```


netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *kubernetes-sigs-cloud-provide-azure* processing.


|  Name | Link |
|---------------------------------|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | 791c98785c4b5b4330b6e42bfe9f2ad32cab8ac0 |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/kubernetes-sigs-cloud-provide-azure/deploys/631a802dc4f3890008b89e19 |
k8s-ci-robot: Hi @zmyzheng. Thanks for your PR.

I'm waiting for a [kubernetes-sigs](https://github.com/orgs/kubernetes-sigs/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://git.k8s.io/community/community-membership.md#member) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fcloud-provider-azure).

<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.
</details>

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

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/cloud-provider-azure/pull/2288#" title="Author self-approved">zmyzheng</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign nilo19 for approval by writing `/assign @nilo19` in a comment. For more information see:[The Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

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

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

- **[OWNERS](https://github.com/kubernetes-sigs/cloud-provider-azure/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":["nilo19"]} -->
coveralls: 
[![Coverage Status](https://coveralls.io/builds/52328713/badge)](https://coveralls.io/builds/52328713)

Coverage decreased (-0.5%) to 79.806% when pulling **54ba2de30d44e4139b95e7710a5e8982e535bbb5 on zmyzheng:vmssflex-integration** into **abbc43b63f6703fb8bf440352a593eb88b057a61 on kubernetes-sigs:master**.

MartinForReal: /ok-to-test
k8s-ci-robot: @zmyzheng: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz | 54ba2de30d44e4139b95e7710a5e8982e535bbb5 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz/1568132776144670720) | false | `/test pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=2288). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Azmyzheng). 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-sigs/cloud-provider-azure/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: @zmyzheng: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz | 54ba2de30d44e4139b95e7710a5e8982e535bbb5 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz/1568132776144670720) | false | `/test pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz`
pull-cloud-provider-azure-e2e-ccm-vmss-capz | 54ba2de30d44e4139b95e7710a5e8982e535bbb5 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-e2e-ccm-vmss-capz/1568132776002064384) | true | `/test pull-cloud-provider-azure-e2e-ccm-vmss-capz`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=2288). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Azmyzheng). 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-sigs/cloud-provider-azure/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: @zmyzheng: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm-vmss-capz | 54ba2de30d44e4139b95e7710a5e8982e535bbb5 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-e2e-ccm-vmss-capz/1568132776002064384) | true | `/test pull-cloud-provider-azure-e2e-ccm-vmss-capz`
pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz | edcecef075a7c91b30797f5be29e5e12982cf4ba | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz/1569386850114605056) | false | `/test pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=2288). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Azmyzheng). 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-sigs/cloud-provider-azure/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: @zmyzheng: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz | edcecef075a7c91b30797f5be29e5e12982cf4ba | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz/1569386850114605056) | false | `/test pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz`
pull-cloud-provider-azure-e2e-ccm-vmss-capz | edcecef075a7c91b30797f5be29e5e12982cf4ba | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-e2e-ccm-vmss-capz/1569386849963610112) | true | `/test pull-cloud-provider-azure-e2e-ccm-vmss-capz`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=2288). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Azmyzheng). 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-sigs/cloud-provider-azure/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 -->
andyzhangx: in our failed csi driver test cases, `DisableAvailabilitySetNodes` is set as `false`, so the following case is broken?
```
For pure vmss uniform cluster, vm type should be set to vmss and DisableAvailabilitySetNodes should be set to true. (no change)
```

actually `DisableAvailabilitySetNodes` just means we don't want to list avset since we could make sure there are no avset in the cluster, it does not mean current cluster is vmss uniform or vmss flex.

```
I0914 06:21:45.179932       1 azuredisk.go:201] cloud: AzurePublicCloud, location: uksouth, rg: kubetest-zshepzqk, VMType: vmss, PrimaryScaleSetName: 2679k8s00, PrimaryAvailabilitySetName: , DisableAvailabilitySetNodes: false
```
k8s-ci-robot: @zmyzheng: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz | edcecef075a7c91b30797f5be29e5e12982cf4ba | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz/1569386850114605056) | false | `/test pull-cloud-provider-azure-e2e-ccm-vmss-ip-lb-capz`
pull-cloud-provider-azure-e2e-ccm-vmss-capz | edcecef075a7c91b30797f5be29e5e12982cf4ba | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-e2e-ccm-vmss-capz/1569386849963610112) | true | `/test pull-cloud-provider-azure-e2e-ccm-vmss-capz`
pull-cloud-provider-azure-e2e-capz | 57b639ccacd900d55986beafd208c3b1b64b13bb | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-e2e-capz/1570137095316443136) | true | `/test pull-cloud-provider-azure-e2e-capz`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=2288). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Azmyzheng). 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-sigs/cloud-provider-azure/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 -->
zmyzheng: > in our failed csi driver test cases, `DisableAvailabilitySetNodes` is set as `false`, so the following case is broken?
> 
> ```
> For pure vmss uniform cluster, vm type should be set to vmss and DisableAvailabilitySetNodes should be set to true. (no change)
> ```
> 
> actually `DisableAvailabilitySetNodes` just means we don't want to list avset since we could make sure there are no avset in the cluster, it does not mean current cluster is vmss uniform or vmss flex.
> 
> ```
> I0914 06:21:45.179932       1 azuredisk.go:201] cloud: AzurePublicCloud, location: uksouth, rg: kubetest-zshepzqk, VMType: vmss, PrimaryScaleSetName: 2679k8s00, PrimaryAvailabilitySetName: , DisableAvailabilitySetNodes: false
> ```

Per discussion, the root cause is the value of OrchestrationMode is empty. The lastest code has fixed this issue
zmyzheng: > ```
> The name DisableAvailabilitySetNodes is kind of misleading: it actually means disable both AvailabilitySet nodes and VMSS Flex nodes
> ```
> 
> I think there is problem for this since `DisableAvailabilitySetNodes` only means disable avset VM list now, while in the future, it would also disable vmss flex list, that may break in the future.
> 
> The motivation for adding `DisableAvailabilitySetNodes` flag is on AKS, avset and vmss cannot be enabled on one cluster, while for vmss uniform and vmss flex, both could be enabled on one cluster?

I will leave this question to @feiskyer for confirmation.

My main consideration here is backward compatibility. Before my code change, there is a way (set DisableAvailabilitySetNodes to true) to tell CCM all nodes are Uniform nodes and node type check can be bypassed to reduce the API call. After adding my code change, CCM should also have the ability to do so. This will make sure the latest CCM code can continue to be used in the old all Uniform node AKS cluster and the behavior and performance are exactly the same as before. I agree continuing to use the flag DisableAvailabilitySetNodes is a little misleading (it would be nice if it was named as something like "EnableAllUniformNodesCluster"), however, since it is already named in such as way, looks like we don't have better option to keep both backward compatibility and naming clarity.
andyzhangx: > > ```
> > The name DisableAvailabilitySetNodes is kind of misleading: it actually means disable both AvailabilitySet nodes and VMSS Flex nodes
> > ```
> > 
> > 
> >     
> >       
> >     
> > 
> >       
> >     
> > 
> >     
> >   
> > I think there is problem for this since `DisableAvailabilitySetNodes` only means disable avset VM list now, while in the future, it would also disable vmss flex list, that may break in the future.
> > The motivation for adding `DisableAvailabilitySetNodes` flag is on AKS, avset and vmss cannot be enabled on one cluster, while for vmss uniform and vmss flex, both could be enabled on one cluster?
> 
> I will leave this question to @feiskyer for confirmation.
> 
> My main consideration here is backward compatibility. Before my code change, there is a way (set DisableAvailabilitySetNodes to true) to tell CCM all nodes are Uniform nodes and node type check can be bypassed to reduce the API call. After adding my code change, CCM should also have the ability to do so. This will make sure the latest CCM code can continue to be used in the old all Uniform node AKS cluster and the behavior and performance are exactly the same as before. I agree continuing to use the flag DisableAvailabilitySetNodes is a little misleading (it would be nice if it was named as something like "EnableAllUniformNodesCluster"), however, since it is already named in such as way, looks like we don't have better option to keep both backward compatibility and naming clarity.

@zmyzheng I think the main concern is whether it's supported to enable vmss uniform and vmss flex on one AKS cluster in the future? If yes, we should not use `DisableAvailabilitySetNodes` to disallow vmss flex list all. We could add new flags, e.g. `DisableVMSSUniformNodes`, `DisableVMSSFlexNodes` flags, that would be compatible, and more straigntforward.
zmyzheng: > > > ```
> > > The name DisableAvailabilitySetNodes is kind of misleading: it actually means disable both AvailabilitySet nodes and VMSS Flex nodes
> > > ```
> > > 
> > > 
> > >     
> > >       
> > >     
> > > 
> > >       
> > >     
> > > 
> > >     
> > >   
> > > I think there is problem for this since `DisableAvailabilitySetNodes` only means disable avset VM list now, while in the future, it would also disable vmss flex list, that may break in the future.
> > > The motivation for adding `DisableAvailabilitySetNodes` flag is on AKS, avset and vmss cannot be enabled on one cluster, while for vmss uniform and vmss flex, both could be enabled on one cluster?
> > 
> > 
> > I will leave this question to @feiskyer for confirmation.
> > My main consideration here is backward compatibility. Before my code change, there is a way (set DisableAvailabilitySetNodes to true) to tell CCM all nodes are Uniform nodes and node type check can be bypassed to reduce the API call. After adding my code change, CCM should also have the ability to do so. This will make sure the latest CCM code can continue to be used in the old all Uniform node AKS cluster and the behavior and performance are exactly the same as before. I agree continuing to use the flag DisableAvailabilitySetNodes is a little misleading (it would be nice if it was named as something like "EnableAllUniformNodesCluster"), however, since it is already named in such as way, looks like we don't have better option to keep both backward compatibility and naming clarity.
> 
> @zmyzheng I think the main concern is whether it's supported to enable vmss uniform and vmss flex on one AKS cluster in the future? If yes, we should not use `DisableAvailabilitySetNodes` to disallow vmss flex list all. We could add new flags, e.g. `DisableVMSSUniformNodes`, `DisableVMSSFlexNodes` flags, that would be compatible, and more straigntforward.

I agree adding new flags is more straightforward. However, one conner case I can think of is, how do we handle the existing AKS clusters? The existing clusters only knows the "DisableAvailabilitySetNodes "flag and asumes all nodes are Uniform and skip node type check when "DisableAvailabilitySetNodes" is set to true. If we choose to add new flags, after existing AKS cluster upgrades to the new CCM, the behavior will change: it will start to check node type, which is unnecessary for those existing all Uniform node clusters. If this node type check sounds OK to you, then I'm OK to add new flags.
k8s-ci-robot: @zmyzheng: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-e2e-capz | 71a0ad20999b7768f25adf39c7811d99927693a4 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-e2e-capz/1570649914331893760) | true | `/test pull-cloud-provider-azure-e2e-capz`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=2288). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Azmyzheng). 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-sigs/cloud-provider-azure/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 -->
feiskyer: > I think there is problem for this since DisableAvailabilitySetNodes only means disable avset VM list now, while in the future, it would also disable vmss flex list, that may break in the future.

Correct. DisableAvailabilitySetNodes is only for VMAS nodes, it should not be used for VMSS flex nodes.

> Add a new vm type "vmssflex" for the K8s cluster with all nodes from VMSS Flex.

Could you make the vmss flex supported via `vmss` vmType? Adding another mode would make it not be able to support different kinds of nodes together. A new `DisableVMSSFlexNodes` could be added to disable vmss flex in case flex nodes are not desired for the cluster (and vmss flex should be disabled by default on the alpha/beta stage).

> However, one conner case I can think of is, how do we handle the existing AKS clusters? The existing clusters only knows the "DisableAvailabilitySetNodes "flag and asumes all nodes are Uniform and skip node type check when "DisableAvailabilitySetNodes" is set to true.

Checking node type is ok. When the VMAS or VMSS Flex are disabled, our caches won't need to invoke the related APIs, hence reduce the chances of throttling issues (this is why DisableAvailabilitySetNodes=true when the cluster only want to use VMSS Uniform).

zmyzheng: Per discussion with @feiskyer offline, below are the latest configurations:
<img width="622" alt="image" src="https://user-images.githubusercontent.com/24629574/191586195-be79df81-6ede-4815-86f5-1d42c451028c.png">

k8s-ci-robot: @zmyzheng: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-e2e-capz | d15fad9c12df479526604e297ae8041be1fe6dab | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-e2e-capz/1572654499707752448) | true | `/test pull-cloud-provider-azure-e2e-capz`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=2288). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Azmyzheng). 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-sigs/cloud-provider-azure/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: @zmyzheng: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-e2e-capz | d137c3c8e6747aa87051b9ebb91bdb4a84f7e8a0 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-e2e-capz/1572703639217115136) | true | `/test pull-cloud-provider-azure-e2e-capz`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=2288). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Azmyzheng). 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-sigs/cloud-provider-azure/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 -->
zmyzheng: /retest
k8s-ci-robot: @zmyzheng: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-check | 1d2f456a65942cad74e8955960af42c0d4a9971e | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2288/pull-cloud-provider-azure-check/1574540663746203648) | true | `/test pull-cloud-provider-azure-check`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=2288). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Azmyzheng). 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-sigs/cloud-provider-azure/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 -->
zmyzheng: > Have you run the tests on CSI repo again?

I updated the code in CSI driver and triggered the test again: https://github.com/kubernetes-sigs/azuredisk-csi-driver/pull/1512

k8s-ci-robot: @zmyzheng: PR needs rebase.

<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.
</details>

Patch:
diff --git a/pkg/consts/consts.go b/pkg/consts/consts.go
index 153b2dc03..548d9547c 100644
--- a/pkg/consts/consts.go
+++ b/pkg/consts/consts.go
@@ -135,6 +135,8 @@ const (
 	VMSSKey = "k8svmssKey"
 	// VMASKey is the key when querying vmss cache
 	VMASKey = "k8svmasKey"
+	// NonVmssUniformNodesKey is the key when querying nonVmssUniformNodes cache
+	NonVmssUniformNodesKey = "k8sNonVmssUniformNodesKey"
 	// AvailabilitySetNodesKey is the availability set nodes key
 	AvailabilitySetNodesKey = "k8sAvailabilitySetNodesKey"
 
@@ -143,7 +145,11 @@ const (
 
 	// GetNodeVmssFlexIDLockKey is the key for getting the lock for getNodeVmssFlexID function
 	GetNodeVmssFlexIDLockKey = "k8sGetNodeVmssFlexIDLockKey"
+	// VMManagementTypeLockKey is the key for getting the lock for getVMManagementType function
+	VMManagementTypeLockKey = "VMManagementType"
 
+	// NonVmssUniformNodesCacheTTLDefaultInSeconds is the TTL of the non vmss uniform node cache
+	NonVmssUniformNodesCacheTTLDefaultInSeconds = 900
 	// AvailabilitySetNodesCacheTTLDefaultInSeconds is the TTL of the availabilitySet node cache
 	AvailabilitySetNodesCacheTTLDefaultInSeconds = 900
 	// VMSSCacheTTLDefaultInSeconds is the TTL of the vmss cache
@@ -155,10 +161,8 @@ const (
 
 	// VmssFlexCacheTTLDefaultInSeconds is the TTL of the vmss flex cache
 	VmssFlexCacheTTLDefaultInSeconds = 600
-	// VmssFlexVMCacheTTLInSeconds is the TTL of the vmss flex vm cache
-	VmssFlexVMCacheTTLInSeconds = 600
-	// VmssFlexVMStatusCacheTTLInSeconds is the TTL of the vmss flex vm status cache
-	VmssFlexVMStatusCacheTTLInSeconds = 600
+	// VmssFlexVMCacheTTLDefaultInSeconds is the TTL of the vmss flex vm cache
+	VmssFlexVMCacheTTLDefaultInSeconds = 600
 
 	// ZoneFetchingInterval defines the interval of performing zoneClient.GetZones
 	ZoneFetchingInterval = 30 * time.Minute
diff --git a/pkg/provider/azure.go b/pkg/provider/azure.go
index bca068322..682d29e96 100644
--- a/pkg/provider/azure.go
+++ b/pkg/provider/azure.go
@@ -178,6 +178,8 @@ type Config struct {
 
 	// DisableAvailabilitySetNodes disables VMAS nodes support when "VMType" is set to "vmss".
 	DisableAvailabilitySetNodes bool `json:"disableAvailabilitySetNodes,omitempty" yaml:"disableAvailabilitySetNodes,omitempty"`
+	// EnableVmssFlexNodes enables vmss flex nodes support when "VMType" is set to "vmss".
+	EnableVmssFlexNodes bool `json:"enableVmssFlexNodes,omitempty" yaml:"enableVmssFlexNodes,omitempty"`
 	// DisableAzureStackCloud disables AzureStackCloud support. It should be used
 	// when setting AzureAuthConfig.Cloud with "AZURESTACKCLOUD" to customize ARM endpoints
 	// while the cluster is not running on AzureStack.
@@ -215,6 +217,9 @@ type Config struct {
 	CloudProviderBackoffRetries int `json:"cloudProviderBackoffRetries,omitempty" yaml:"cloudProviderBackoffRetries,omitempty"`
 	// Backoff duration
 	CloudProviderBackoffDuration int `json:"cloudProviderBackoffDuration,omitempty" yaml:"cloudProviderBackoffDuration,omitempty"`
+	// NonVmssUniformNodesCacheTTLInSeconds sets the Cache TTL for NonVmssUniformNodesCacheTTLInSeconds
+	// if not set, will use default value
+	NonVmssUniformNodesCacheTTLInSeconds int `json:"nonVmssUniformNodesCacheTTLInSeconds,omitempty" yaml:"nonVmssUniformNodesCacheTTLInSeconds,omitempty"`
 	// AvailabilitySetNodesCacheTTLInSeconds sets the Cache TTL for availabilitySetNodesCache
 	// if not set, will use default value
 	AvailabilitySetNodesCacheTTLInSeconds int `json:"availabilitySetNodesCacheTTLInSeconds,omitempty" yaml:"availabilitySetNodesCacheTTLInSeconds,omitempty"`
diff --git a/pkg/provider/azure_controller_common.go b/pkg/provider/azure_controller_common.go
index 34baafc9e..18f3b6df7 100644
--- a/pkg/provider/azure_controller_common.go
+++ b/pkg/provider/azure_controller_common.go
@@ -124,18 +124,24 @@ func (c *controllerCommon) getNodeVMSet(nodeName types.NodeName, crt azcache.Azu
 		return nil, fmt.Errorf("error of converting vmSet (%q) to ScaleSet with vmType %q", c.cloud.VMSet, c.cloud.VMType)
 	}
 
-	// 3. If the node is managed by availability set, then return ss.availabilitySet.
-	managedByAS, err := ss.isNodeManagedByAvailabilitySet(mapNodeNameToVMName(nodeName), crt)
+	vmManagementType, err := ss.getVMManagementTypeByNodeName(mapNodeNameToVMName(nodeName), crt)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("getNodeVMSet: failed to check the node %s management type: %w", mapNodeNameToVMName(nodeName), err)
 	}
-	if managedByAS {
+	// 3. If the node is managed by availability set, then return ss.availabilitySet.
+	if vmManagementType == ManagedByAvSet {
 		// vm is managed by availability set.
 		return ss.availabilitySet, nil
 	}
+	if vmManagementType == ManagedByVmssFlex {
+		// 4. If the node is managed by vmss flex, then return ss.flexScaleSet.
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet, nil
+	}
 
-	// 4. Node is managed by vmss
+	// 5. Node is managed by vmss
 	return ss, nil
+
 }
 
 // AttachDisk attaches a disk to vm
diff --git a/pkg/provider/azure_vmss.go b/pkg/provider/azure_vmss.go
index 929cf4149..61ce4394a 100644
--- a/pkg/provider/azure_vmss.go
+++ b/pkg/provider/azure_vmss.go
@@ -74,9 +74,21 @@ type ScaleSet struct {
 	// this also allows for clusters with both VM and VMSS nodes.
 	availabilitySet VMSet
 
-	vmssCache                 *azcache.TimedCache
-	vmssVMCache               *sync.Map // [resourcegroup/vmssname]*azcache.TimedCache
-	availabilitySetNodesCache *azcache.TimedCache
+	// flexScaleSet is required for self hosted K8s cluster (for example, capz)
+	// It is also used when there are vmssflex node and other types of node in
+	// the same cluster.
+	flexScaleSet VMSet
+
+	vmssCache   *azcache.TimedCache
+	vmssVMCache *sync.Map // [resourcegroup/vmssname]*azcache.TimedCache
+
+	// nonVmssUniformNodesCache is used to store node names from non uniform vm.
+	// Currently, the nodes can from avset or vmss flex or individual vm.
+	// This cache contains an entry called nonVmssUniformNodesEntry.
+	// nonVmssUniformNodesEntry contains avSetVMNodeNames list, clusterNodeNames list
+	// and current clusterNodeNames.
+	nonVmssUniformNodesCache *azcache.TimedCache
+
 	// lockMap in cache refresh
 	lockMap *lockMap
 }
@@ -92,15 +104,21 @@ func newScaleSet(az *Cloud) (VMSet, error) {
 	if err != nil {
 		return nil, err
 	}
+	fs, err := newFlexScaleSet(az)
+	if err != nil {
+		return nil, err
+	}
+
 	ss := &ScaleSet{
 		Cloud:           az,
 		availabilitySet: as,
+		flexScaleSet:    fs,
 		vmssVMCache:     &sync.Map{},
 		lockMap:         newLockMap(),
 	}
 
-	if !ss.DisableAvailabilitySetNodes {
-		ss.availabilitySetNodesCache, err = ss.newAvailabilitySetNodesCache()
+	if !ss.DisableAvailabilitySetNodes || ss.EnableVmssFlexNodes {
+		ss.nonVmssUniformNodesCache, err = ss.newNonVmssUniformNodesCache()
 		if err != nil {
 			return nil, err
 		}
@@ -232,16 +250,21 @@ func (ss *ScaleSet) getVmssVM(nodeName string, crt azcache.AzureCacheReadType) (
 
 // GetPowerStatusByNodeName returns the power state of the specified node.
 func (ss *ScaleSet) GetPowerStatusByNodeName(name string) (powerState string, err error) {
-	managedByAS, err := ss.isNodeManagedByAvailabilitySet(name, azcache.CacheReadTypeUnsafe)
+	vmManagementType, err := ss.getVMManagementTypeByNodeName(name, azcache.CacheReadTypeUnsafe)
 	if err != nil {
-		klog.Errorf("Failed to check isNodeManagedByAvailabilitySet: %v", err)
+		klog.Errorf("Failed to check VM management type: %v", err)
 		return "", err
 	}
-	if managedByAS {
+
+	if vmManagementType == ManagedByAvSet {
 		// vm is managed by availability set.
 		return ss.availabilitySet.GetPowerStatusByNodeName(name)
 	}
-
+	if vmManagementType == ManagedByVmssFlex {
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet.GetPowerStatusByNodeName(name)
+	}
+	// VM is managed by vmss
 	vm, err := ss.getVmssVM(name, azcache.CacheReadTypeDefault)
 	if err != nil {
 		return powerState, err
@@ -267,15 +290,20 @@ func (ss *ScaleSet) GetPowerStatusByNodeName(name string) (powerState string, er
 
 // GetProvisioningStateByNodeName returns the provisioningState for the specified node.
 func (ss *ScaleSet) GetProvisioningStateByNodeName(name string) (provisioningState string, err error) {
-	managedByAS, err := ss.isNodeManagedByAvailabilitySet(name, azcache.CacheReadTypeUnsafe)
+	vmManagementType, err := ss.getVMManagementTypeByNodeName(name, azcache.CacheReadTypeUnsafe)
 	if err != nil {
-		klog.Errorf("Failed to check isNodeManagedByAvailabilitySet: %v", err)
+		klog.Errorf("Failed to check VM management type: %v", err)
 		return "", err
 	}
-	if managedByAS {
+
+	if vmManagementType == ManagedByAvSet {
 		// vm is managed by availability set.
 		return ss.availabilitySet.GetProvisioningStateByNodeName(name)
 	}
+	if vmManagementType == ManagedByVmssFlex {
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet.GetProvisioningStateByNodeName(name)
+	}
 
 	vm, err := ss.getVmssVM(name, azcache.CacheReadTypeDefault)
 	if err != nil {
@@ -352,18 +380,29 @@ func (ss *ScaleSet) getVmssVMByInstanceID(resourceGroup, scaleSetName, instanceI
 // It must return ("", cloudprovider.InstanceNotFound) if the instance does
 // not exist or is no longer running.
 func (ss *ScaleSet) GetInstanceIDByNodeName(name string) (string, error) {
-	managedByAS, err := ss.isNodeManagedByAvailabilitySet(name, azcache.CacheReadTypeUnsafe)
+	vmManagementType, err := ss.getVMManagementTypeByNodeName(name, azcache.CacheReadTypeUnsafe)
 	if err != nil {
-		klog.Errorf("Failed to check isNodeManagedByAvailabilitySet: %v", err)
+		klog.Errorf("Failed to check VM management type: %v", err)
 		return "", err
 	}
-	if managedByAS {
+
+	if vmManagementType == ManagedByAvSet {
 		// vm is managed by availability set.
 		return ss.availabilitySet.GetInstanceIDByNodeName(name)
 	}
+	if vmManagementType == ManagedByVmssFlex {
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet.GetInstanceIDByNodeName(name)
+	}
 
 	vm, err := ss.getVmssVM(name, azcache.CacheReadTypeUnsafe)
 	if err != nil {
+		// special case: during scaling in, if the vm is deleted and nonVmssUniformNodesCache is refreshed,
+		// then getVMManagementTypeByNodeName will return ManagedByVmssUniform no matter what the actual managementType is.
+		// In this case, if it is actually a non vmss uniform node, return InstanceNotFound
+		if errors.Is(err, ErrorNotVmssInstance) {
+			return "", cloudprovider.InstanceNotFound
+		}
 		klog.Errorf("Unable to find node %s: %v", name, err)
 		return "", err
 	}
@@ -384,13 +423,27 @@ func (ss *ScaleSet) GetInstanceIDByNodeName(name string) (string, error) {
 // azure:///subscriptions/subsid/resourceGroups/rg/providers/Microsoft.Compute/virtualMachineScaleSets/aks-agentpool-22126781-vmss/virtualMachines/1
 // /subscriptions/subsid/resourceGroups/rg/providers/Microsoft.Compute/virtualMachineScaleSets/aks-agentpool-22126781-vmss/virtualMachines/k8s-agentpool-36841236-vmss_1
 func (ss *ScaleSet) GetNodeNameByProviderID(providerID string) (types.NodeName, error) {
-	// NodeName is not part of providerID for vmss instances.
-	scaleSetName, err := extractScaleSetNameByProviderID(providerID)
+
+	vmManagementType, err := ss.getVMManagementTypeByProviderID(providerID, azcache.CacheReadTypeUnsafe)
 	if err != nil {
-		klog.V(4).Infof("Can not extract scale set name from providerID (%s), assuming it is managed by availability set: %v", providerID, err)
+		klog.Errorf("Failed to check VM management type: %v", err)
+		return "", err
+	}
+
+	if vmManagementType == ManagedByAvSet {
+		// vm is managed by availability set.
 		return ss.availabilitySet.GetNodeNameByProviderID(providerID)
 	}
+	if vmManagementType == ManagedByVmssFlex {
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet.GetNodeNameByProviderID(providerID)
+	}
 
+	// NodeName is not part of providerID for vmss instances.
+	scaleSetName, err := extractScaleSetNameByProviderID(providerID)
+	if err != nil {
+		return "", fmt.Errorf("error of extracting vmss name for node %q", providerID)
+	}
 	resourceGroup, err := extractResourceGroupByProviderID(providerID)
 	if err != nil {
 		return "", fmt.Errorf("error of extracting resource group for node %q", providerID)
@@ -426,15 +479,20 @@ func (ss *ScaleSet) GetNodeNameByProviderID(providerID string) (types.NodeName,
 
 // GetInstanceTypeByNodeName gets the instance type by node name.
 func (ss *ScaleSet) GetInstanceTypeByNodeName(name string) (string, error) {
-	managedByAS, err := ss.isNodeManagedByAvailabilitySet(name, azcache.CacheReadTypeUnsafe)
+	vmManagementType, err := ss.getVMManagementTypeByNodeName(name, azcache.CacheReadTypeUnsafe)
 	if err != nil {
-		klog.Errorf("Failed to check isNodeManagedByAvailabilitySet: %v", err)
+		klog.Errorf("Failed to check VM management type: %v", err)
 		return "", err
 	}
-	if managedByAS {
+
+	if vmManagementType == ManagedByAvSet {
 		// vm is managed by availability set.
 		return ss.availabilitySet.GetInstanceTypeByNodeName(name)
 	}
+	if vmManagementType == ManagedByVmssFlex {
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet.GetInstanceTypeByNodeName(name)
+	}
 
 	vm, err := ss.getVmssVM(name, azcache.CacheReadTypeUnsafe)
 	if err != nil {
@@ -454,15 +512,20 @@ func (ss *ScaleSet) GetInstanceTypeByNodeName(name string) (string, error) {
 // GetZoneByNodeName gets availability zone for the specified node. If the node is not running
 // with availability zone, then it returns fault domain.
 func (ss *ScaleSet) GetZoneByNodeName(name string) (cloudprovider.Zone, error) {
-	managedByAS, err := ss.isNodeManagedByAvailabilitySet(name, azcache.CacheReadTypeUnsafe)
+	vmManagementType, err := ss.getVMManagementTypeByNodeName(name, azcache.CacheReadTypeUnsafe)
 	if err != nil {
-		klog.Errorf("Failed to check isNodeManagedByAvailabilitySet: %v", err)
+		klog.Errorf("Failed to check VM management type: %v", err)
 		return cloudprovider.Zone{}, err
 	}
-	if managedByAS {
+
+	if vmManagementType == ManagedByAvSet {
 		// vm is managed by availability set.
 		return ss.availabilitySet.GetZoneByNodeName(name)
 	}
+	if vmManagementType == ManagedByVmssFlex {
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet.GetZoneByNodeName(name)
+	}
 
 	vm, err := ss.getVmssVM(name, azcache.CacheReadTypeUnsafe)
 	if err != nil {
@@ -505,6 +568,21 @@ func (ss *ScaleSet) GetPrimaryVMSetName() string {
 
 // GetIPByNodeName gets machine private IP and public IP by node name.
 func (ss *ScaleSet) GetIPByNodeName(nodeName string) (string, string, error) {
+	vmManagementType, err := ss.getVMManagementTypeByNodeName(nodeName, azcache.CacheReadTypeUnsafe)
+	if err != nil {
+		klog.Errorf("Failed to check VM management type: %v", err)
+		return "", "", err
+	}
+
+	if vmManagementType == ManagedByAvSet {
+		// vm is managed by availability set.
+		return ss.availabilitySet.GetIPByNodeName(nodeName)
+	}
+	if vmManagementType == ManagedByVmssFlex {
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet.GetIPByNodeName(nodeName)
+	}
+
 	nic, err := ss.GetPrimaryInterface(nodeName)
 	if err != nil {
 		klog.Errorf("error: ss.GetIPByNodeName(%s), GetPrimaryInterface(%q), err=%v", nodeName, nodeName, err)
@@ -568,6 +646,21 @@ func (ss *ScaleSet) getVMSSPublicIPAddress(resourceGroupName string, virtualMach
 // allowing users to split ipv4/v6 on multiple nics
 func (ss *ScaleSet) GetPrivateIPsByNodeName(nodeName string) ([]string, error) {
 	ips := make([]string, 0)
+	vmManagementType, err := ss.getVMManagementTypeByNodeName(nodeName, azcache.CacheReadTypeUnsafe)
+	if err != nil {
+		klog.Errorf("Failed to check VM management type: %v", err)
+		return ips, err
+	}
+
+	if vmManagementType == ManagedByAvSet {
+		// vm is managed by availability set.
+		return ss.availabilitySet.GetPrivateIPsByNodeName(nodeName)
+	}
+	if vmManagementType == ManagedByVmssFlex {
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet.GetPrivateIPsByNodeName(nodeName)
+	}
+
 	nic, err := ss.GetPrimaryInterface(nodeName)
 	if err != nil {
 		klog.Errorf("error: ss.GetIPByNodeName(%s), GetPrimaryInterface(%q), err=%v", nodeName, nodeName, err)
@@ -811,9 +904,9 @@ func (ss *ScaleSet) GetVMSetNames(service *v1.Service, nodes []*v1.Node) (*[]str
 		return scaleSetNames, nil
 	}
 
-	scaleSetNames, err := ss.getAgentPoolScaleSets(nodes)
+	scaleSetNames, err := ss.GetAgentPoolVMSetNames(nodes)
 	if err != nil {
-		klog.Errorf("ss.GetVMSetNames - getAgentPoolScaleSets failed err=(%v)", err)
+		klog.Errorf("ss.GetVMSetNames - GetAgentPoolVMSetNames failed err=(%v)", err)
 		return nil, err
 	}
 	if len(*scaleSetNames) == 0 {
@@ -852,15 +945,20 @@ func extractResourceGroupByVMSSNicID(nicID string) (string, error) {
 
 // GetPrimaryInterface gets machine primary network interface by node name and vmSet.
 func (ss *ScaleSet) GetPrimaryInterface(nodeName string) (network.Interface, error) {
-	managedByAS, err := ss.isNodeManagedByAvailabilitySet(nodeName, azcache.CacheReadTypeDefault)
+	vmManagementType, err := ss.getVMManagementTypeByNodeName(nodeName, azcache.CacheReadTypeUnsafe)
 	if err != nil {
-		klog.Errorf("Failed to check isNodeManagedByAvailabilitySet: %v", err)
+		klog.Errorf("Failed to check VM management type: %v", err)
 		return network.Interface{}, err
 	}
-	if managedByAS {
+
+	if vmManagementType == ManagedByAvSet {
 		// vm is managed by availability set.
 		return ss.availabilitySet.GetPrimaryInterface(nodeName)
 	}
+	if vmManagementType == ManagedByVmssFlex {
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet.GetPrimaryInterface(nodeName)
+	}
 
 	vm, err := ss.getVmssVM(nodeName, azcache.CacheReadTypeDefault)
 	if err != nil {
@@ -1272,9 +1370,7 @@ func (ss *ScaleSet) ensureVMSSInPool(service *v1.Service, nodes []*v1.Node, back
 	return nil
 }
 
-// EnsureHostsInPool ensures the given Node's primary IP configurations are
-// participating in the specified LoadBalancer Backend Pool.
-func (ss *ScaleSet) EnsureHostsInPool(service *v1.Service, nodes []*v1.Node, backendPoolID string, vmSetNameOfLB string) error {
+func (ss *ScaleSet) ensureHostsInPool(service *v1.Service, nodes []*v1.Node, backendPoolID string, vmSetNameOfLB string) error {
 	mc := metrics.NewMetricContext("services", "vmss_ensure_hosts_in_pool", ss.ResourceGroup, ss.SubscriptionID, getServiceName(service))
 	isOperationSucceeded := false
 	defer func() {
@@ -1302,28 +1398,6 @@ func (ss *ScaleSet) EnsureHostsInPool(service *v1.Service, nodes []*v1.Node, bac
 			continue
 		}
 
-		// Check whether the node is VMAS virtual machine.
-		managedByAS, err := ss.isNodeManagedByAvailabilitySet(localNodeName, azcache.CacheReadTypeDefault)
-		if err != nil {
-			klog.Errorf("Failed to check isNodeManagedByAvailabilitySet(%s): %v", localNodeName, err)
-			errors = append(errors, err)
-			continue
-		}
-
-		if managedByAS {
-			// VMAS nodes should also be added to the SLB backends.
-			if ss.useStandardLoadBalancer() {
-				hostUpdates = append(hostUpdates, func() error {
-					_, _, _, _, err := ss.availabilitySet.EnsureHostInPool(service, types.NodeName(localNodeName), backendPoolID, vmSetNameOfLB)
-					return err
-				})
-				continue
-			}
-
-			klog.V(3).Infof("EnsureHostsInPool skips node %s because VMAS nodes couldn't be added to basic LB with VMSS backends", localNodeName)
-			continue
-		}
-
 		nodeResourceGroup, nodeVMSS, nodeInstanceID, nodeVMSSVM, err := ss.EnsureHostInPool(service, types.NodeName(localNodeName), backendPoolID, vmSetNameOfLB)
 		if err != nil {
 			klog.Errorf("EnsureHostInPool(%s): backendPoolID(%s) - failed to ensure host in pool: %q", getServiceName(service), backendPoolID, err)
@@ -1404,6 +1478,83 @@ func (ss *ScaleSet) EnsureHostsInPool(service *v1.Service, nodes []*v1.Node, bac
 	return nil
 }
 
+// EnsureHostsInPool ensures the given Node's primary IP configurations are
+// participating in the specified LoadBalancer Backend Pool.
+func (ss *ScaleSet) EnsureHostsInPool(service *v1.Service, nodes []*v1.Node, backendPoolID string, vmSetNameOfLB string) error {
+	if ss.DisableAvailabilitySetNodes && !ss.EnableVmssFlexNodes {
+		return ss.ensureHostsInPool(service, nodes, backendPoolID, vmSetNameOfLB)
+	}
+	vmssUniformNodes := make([]*v1.Node, 0)
+	vmssFlexNodes := make([]*v1.Node, 0)
+	vmasNodes := make([]*v1.Node, 0)
+	errors := make([]error, 0)
+	for _, node := range nodes {
+		localNodeName := node.Name
+
+		if ss.useStandardLoadBalancer() && ss.excludeMasterNodesFromStandardLB() && isControlPlaneNode(node) {
+			klog.V(4).Infof("Excluding master node %q from load balancer backendpool %q", localNodeName, backendPoolID)
+			continue
+		}
+
+		shouldExcludeLoadBalancer, err := ss.ShouldNodeExcludedFromLoadBalancer(localNodeName)
+		if err != nil {
+			klog.Errorf("ShouldNodeExcludedFromLoadBalancer(%s) failed with error: %v", localNodeName, err)
+			return err
+		}
+		if shouldExcludeLoadBalancer {
+			klog.V(4).Infof("Excluding unmanaged/external-resource-group node %q", localNodeName)
+			continue
+		}
+
+		vmManagementType, err := ss.getVMManagementTypeByNodeName(localNodeName, azcache.CacheReadTypeDefault)
+		if err != nil {
+			klog.Errorf("Failed to check vmManagementType(%s): %v", localNodeName, err)
+			errors = append(errors, err)
+			continue
+		}
+
+		if vmManagementType == ManagedByAvSet {
+			// vm is managed by availability set.
+			// VMAS nodes should also be added to the SLB backends.
+			if ss.useStandardLoadBalancer() {
+				vmasNodes = append(vmasNodes, node)
+				continue
+			}
+			klog.V(3).Infof("EnsureHostsInPool skips node %s because VMAS nodes couldn't be added to basic LB with VMSS backends", localNodeName)
+			continue
+		}
+		if vmManagementType == ManagedByVmssFlex {
+			// vm is managed by vmss flex.
+			if ss.useStandardLoadBalancer() {
+				vmssFlexNodes = append(vmssFlexNodes, node)
+				continue
+			}
+			klog.V(3).Infof("EnsureHostsInPool skips node %s because VMSS Flex nodes deos not support Basic Load Balancer", localNodeName)
+			continue
+		}
+		vmssUniformNodes = append(vmssUniformNodes, node)
+	}
+
+	if len(vmssFlexNodes) > 0 {
+		vmssFlexError := ss.flexScaleSet.EnsureHostsInPool(service, vmssFlexNodes, backendPoolID, vmSetNameOfLB)
+		errors = append(errors, vmssFlexError)
+	}
+
+	if len(vmasNodes) > 0 {
+		vmasError := ss.availabilitySet.EnsureHostsInPool(service, vmasNodes, backendPoolID, vmSetNameOfLB)
+		errors = append(errors, vmasError)
+	}
+
+	if len(vmssUniformNodes) > 0 {
+		vmssUniformError := ss.ensureHostsInPool(service, vmssUniformNodes, backendPoolID, vmSetNameOfLB)
+		errors = append(errors, vmssUniformError)
+	}
+
+	allErrors := utilerrors.Flatten(utilerrors.NewAggregate(errors))
+
+	return allErrors
+}
+
 // ensureBackendPoolDeletedFromNode ensures the loadBalancer backendAddressPools deleted
 // from the specified node, which returns (resourceGroup, vmasName, instanceID, vmssVM, error).
 func (ss *ScaleSet) ensureBackendPoolDeletedFromNode(nodeName, backendPoolID string) (string, string, string, *compute.VirtualMachineScaleSetVM, error) {
@@ -1479,15 +1630,24 @@ func (ss *ScaleSet) ensureBackendPoolDeletedFromNode(nodeName, backendPoolID str
 
 // GetNodeNameByIPConfigurationID gets the node name and the VMSS name by IP configuration ID.
 func (ss *ScaleSet) GetNodeNameByIPConfigurationID(ipConfigurationID string) (string, string, error) {
+	vmManagementType, err := ss.getVMManagementTypeByIPConfigurationID(ipConfigurationID, azcache.CacheReadTypeUnsafe)
+	if err != nil {
+		klog.Errorf("Failed to check VM management type: %v", err)
+		return "", "", err
+	}
+
+	if vmManagementType == ManagedByAvSet {
+		// vm is managed by availability set.
+		return ss.availabilitySet.GetNodeNameByIPConfigurationID(ipConfigurationID)
+	}
+	if vmManagementType == ManagedByVmssFlex {
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet.GetNodeNameByIPConfigurationID(ipConfigurationID)
+	}
+
 	matches := vmssIPConfigurationRE.FindStringSubmatch(ipConfigurationID)
 	if len(matches) != 4 {
-		klog.V(4).Infof("Can not extract scale set name from ipConfigurationID (%s), assuming it is managed by availability set", ipConfigurationID)
-		name, rg, err := ss.availabilitySet.GetNodeNameByIPConfigurationID(ipConfigurationID)
-		if err != nil && !errors.Is(err, cloudprovider.InstanceNotFound) {
-			klog.Errorf("Unable to find node by IPConfigurationID %s: %v", ipConfigurationID, err)
-			return "", "", ErrorNotVmssInstance
-		}
-		return name, rg, nil
+		return "", "", fmt.Errorf("can not extract scale set name from ipConfigurationID (%s)", ipConfigurationID)
 	}
 
 	resourceGroup := matches[1]
@@ -1509,7 +1669,7 @@ func (ss *ScaleSet) GetNodeNameByIPConfigurationID(ipConfigurationID string) (st
 func getScaleSetAndResourceGroupNameByIPConfigurationID(ipConfigurationID string) (string, string, error) {
 	matches := vmssIPConfigurationRE.FindStringSubmatch(ipConfigurationID)
 	if len(matches) != 4 {
-		klog.V(4).Infof("Can not extract scale set name from ipConfigurationID (%s), assuming it is managed by availability set", ipConfigurationID)
+		klog.V(4).Infof("Can not extract scale set name from ipConfigurationID (%s), assuming it is managed by availability set or vmss flex", ipConfigurationID)
 		return "", "", ErrorNotVmssInstance
 	}
 
@@ -1542,8 +1702,8 @@ func (ss *ScaleSet) ensureBackendPoolDeletedFromVMSS(service *v1.Service, backen
 	return ss.EnsureBackendPoolDeletedFromVMSets(vmssNamesMap, backendPoolID)
 }
 
-// EnsureBackendPoolDeleted ensures the loadBalancer backendAddressPools deleted from the specified nodes.
-func (ss *ScaleSet) EnsureBackendPoolDeleted(service *v1.Service, backendPoolID, vmSetName string, backendAddressPools *[]network.BackendAddressPool, deleteFromVMSet bool) error {
+// ensureBackendPoolDeleted ensures the loadBalancer backendAddressPools deleted from the specified nodes.
+func (ss *ScaleSet) ensureBackendPoolDeleted(service *v1.Service, backendPoolID, vmSetName string, backendAddressPools *[]network.BackendAddressPool, deleteFromVMSet bool) error {
 	// Returns nil if backend address pools already deleted.
 	if backendAddressPools == nil {
 		return nil
@@ -1599,7 +1759,7 @@ func (ss *ScaleSet) EnsureBackendPoolDeleted(service *v1.Service, backendPoolID,
 			}
 
 			if errors.Is(err, cloudprovider.InstanceNotFound) {
-				klog.Infof("EnsureBackendPoolDeleted(%s): skipping ip config %s because the corresponding vmss vm is not"+
+				klog.Infof("ensureBackendPoolDeleted(%s): skipping ip config %s because the corresponding vmss vm is not"+
 					" found", getServiceName(service), ipConfigurationID)
 				continue
 			}
@@ -1612,7 +1772,7 @@ func (ss *ScaleSet) EnsureBackendPoolDeleted(service *v1.Service, backendPoolID,
 		nodeResourceGroup, nodeVMSS, nodeInstanceID, nodeVMSSVM, err := ss.ensureBackendPoolDeletedFromNode(nodeName, backendPoolID)
 		if err != nil {
 			if !errors.Is(err, ErrorNotVmssInstance) { // Do nothing for the VMAS nodes.
-				klog.Errorf("EnsureBackendPoolDeleted(%s): backendPoolID(%s) - failed with error %v", getServiceName(service), backendPoolID, err)
+				klog.Errorf("ensureBackendPoolDeleted(%s): backendPoolID(%s) - failed with error %v", getServiceName(service), backendPoolID, err)
 				allErrs = append(allErrs, err)
 			}
 			continue
@@ -1684,8 +1844,109 @@ func (ss *ScaleSet) EnsureBackendPoolDeleted(service *v1.Service, backendPoolID,
 	return nil
 }
 
+// EnsureBackendPoolDeleted ensures the loadBalancer backendAddressPools deleted from the specified nodes.
+func (ss *ScaleSet) EnsureBackendPoolDeleted(service *v1.Service, backendPoolID, vmSetName string, backendAddressPools *[]network.BackendAddressPool, deleteFromVMSet bool) error {
+	if backendAddressPools == nil {
+		return nil
+	}
+	vmssUniformBackendIPConfigurations := []network.InterfaceIPConfiguration{}
+	vmssFlexBackendIPConfigurations := []network.InterfaceIPConfiguration{}
+	avSetBackendIPConfigurations := []network.InterfaceIPConfiguration{}
+
+	for _, backendPool := range *backendAddressPools {
+		if strings.EqualFold(*backendPool.ID, backendPoolID) && backendPool.BackendIPConfigurations != nil {
+			for _, ipConf := range *backendPool.BackendIPConfigurations {
+				if ipConf.ID == nil {
+					continue
+				}
+
+				vmManagementType, err := ss.getVMManagementTypeByIPConfigurationID(*ipConf.ID, azcache.CacheReadTypeUnsafe)
+				if err != nil {
+					klog.Warningf("Failed to check VM management type by ipConfigurationID %s: %v, skip it", *ipConf.ID, err)
+				}
+
+				if vmManagementType == ManagedByAvSet {
+					// vm is managed by availability set.
+					avSetBackendIPConfigurations = append(avSetBackendIPConfigurations, ipConf)
+				}
+				if vmManagementType == ManagedByVmssFlex {
+					// vm is managed by vmss flex.
+					vmssFlexBackendIPConfigurations = append(vmssFlexBackendIPConfigurations, ipConf)
+				}
+				if vmManagementType == ManagedByVmssUniform {
+					// vm is managed by vmss flex.
+					vmssUniformBackendIPConfigurations = append(vmssUniformBackendIPConfigurations, ipConf)
+				}
+			}
+		}
+	}
+
+	if len(vmssUniformBackendIPConfigurations) > 0 {
+		vmssUniformBackendPools := &[]network.BackendAddressPool{
+			{
+				ID: to.StringPtr(backendPoolID),
+				BackendAddressPoolPropertiesFormat: &network.BackendAddressPoolPropertiesFormat{
+					BackendIPConfigurations: &vmssUniformBackendIPConfigurations,
+				},
+			},
+		}
+		err := ss.ensureBackendPoolDeleted(service, backendPoolID, vmSetName, vmssUniformBackendPools, deleteFromVMSet)
+		if err != nil {
+			return err
+		}
+	}
+
+	if len(vmssFlexBackendIPConfigurations) > 0 {
+		vmssFlexBackendPools := &[]network.BackendAddressPool{
+			{
+				ID: to.StringPtr(backendPoolID),
+				BackendAddressPoolPropertiesFormat: &network.BackendAddressPoolPropertiesFormat{
+					BackendIPConfigurations: &vmssFlexBackendIPConfigurations,
+				},
+			},
+		}
+		err := ss.flexScaleSet.EnsureBackendPoolDeleted(service, backendPoolID, vmSetName, vmssFlexBackendPools, deleteFromVMSet)
+		if err != nil {
+			return err
+		}
+	}
+
+	if len(avSetBackendIPConfigurations) > 0 {
+		avSetBackendPools := &[]network.BackendAddressPool{
+			{
+				ID: to.StringPtr(backendPoolID),
+				BackendAddressPoolPropertiesFormat: &network.BackendAddressPoolPropertiesFormat{
+					BackendIPConfigurations: &avSetBackendIPConfigurations,
+				},
+			},
+		}
+		err := ss.availabilitySet.EnsureBackendPoolDeleted(service, backendPoolID, vmSetName, avSetBackendPools, deleteFromVMSet)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+
+}
+
 // GetNodeCIDRMaskByProviderID returns the node CIDR subnet mask by provider ID.
 func (ss *ScaleSet) GetNodeCIDRMasksByProviderID(providerID string) (int, int, error) {
+	vmManagementType, err := ss.getVMManagementTypeByProviderID(providerID, azcache.CacheReadTypeUnsafe)
+	if err != nil {
+		klog.Errorf("Failed to check VM management type: %v", err)
+		return 0, 0, err
+	}
+
+	if vmManagementType == ManagedByAvSet {
+		// vm is managed by availability set.
+		return ss.availabilitySet.GetNodeCIDRMasksByProviderID(providerID)
+	}
+	if vmManagementType == ManagedByVmssFlex {
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet.GetNodeCIDRMasksByProviderID(providerID)
+	}
+
 	_, vmssName, err := getVmssAndResourceGroupNameByVMProviderID(providerID)
 	if err != nil {
 		return 0, 0, err
@@ -1810,44 +2071,76 @@ func (ss *ScaleSet) EnsureBackendPoolDeletedFromVMSets(vmssNamesMap map[string]b
 // like capz allows mixed instance type.
 func (ss *ScaleSet) GetAgentPoolVMSetNames(nodes []*v1.Node) (*[]string, error) {
 	vmSetNames := make([]string, 0)
-	as := ss.availabilitySet.(*availabilitySet)
+
+	vmssFlexVMNodes := make([]*v1.Node, 0)
+	avSetVMNodes := make([]*v1.Node, 0)
 
 	for _, node := range nodes {
 		var names *[]string
-		managedByAS, err := ss.isNodeManagedByAvailabilitySet(node.Name, azcache.CacheReadTypeDefault)
+
+		vmManagementType, err := ss.getVMManagementTypeByNodeName(node.Name, azcache.CacheReadTypeDefault)
 		if err != nil {
-			return nil, fmt.Errorf("GetAgentPoolVMSetNames: failed to check if the node %s is managed by VMAS: %w", node.Name, err)
+			return nil, fmt.Errorf("GetAgentPoolVMSetNames: failed to check the node %s management type: %w", node.Name, err)
 		}
-		if managedByAS {
-			cached, err := ss.availabilitySetNodesCache.Get(consts.AvailabilitySetNodesKey, azcache.CacheReadTypeDefault)
-			if err != nil {
-				return nil, fmt.Errorf("GetAgentPoolVMSetNames: failed to get availabilitySetNodesCache")
-			}
-			vms := cached.(*availabilitySetNodeEntry).vms
-			names, err = as.getAgentPoolAvailabilitySets(vms, []*v1.Node{node})
-			if err != nil {
-				return nil, fmt.Errorf("GetAgentPoolVMSetNames: failed to execute getAgentPoolAvailabilitySets: %w", err)
-			}
-			vmSetNames = append(vmSetNames, *names...)
+
+		if vmManagementType == ManagedByAvSet {
+			// vm is managed by vmss flex.
+			avSetVMNodes = append(avSetVMNodes, node)
+			continue
+		}
+		if vmManagementType == ManagedByVmssFlex {
+			// vm is managed by vmss flex.
+			vmssFlexVMNodes = append(vmssFlexVMNodes, node)
 			continue
 		}
 
 		names, err = ss.getAgentPoolScaleSets([]*v1.Node{node})
 		if err != nil {
-			return nil, fmt.Errorf("GetAgentPoolVMSetNames: failed to execute getAgentPoolScaleSets: %w", err)
+			return nil, fmt.Errorf("GetAgentPoolVMSetNames: failed to execute getAgentPoolScaleSets: %v", err)
 		}
 		vmSetNames = append(vmSetNames, *names...)
 	}
 
+	if len(vmssFlexVMNodes) > 0 {
+		vmssFlexVMnames, err := ss.flexScaleSet.GetAgentPoolVMSetNames(vmssFlexVMNodes)
+		if err != nil {
+			return nil, fmt.Errorf("ss.flexScaleSet.GetAgentPoolVMSetNames: failed to execute : %w", err)
+		}
+		vmSetNames = append(vmSetNames, *vmssFlexVMnames...)
+	}
+
+	if len(avSetVMNodes) > 0 {
+		avSetVMnames, err := ss.availabilitySet.GetAgentPoolVMSetNames(avSetVMNodes)
+		if err != nil {
+			return nil, fmt.Errorf("ss.availabilitySet.GetAgentPoolVMSetNames: failed to execute : %w", err)
+		}
+		vmSetNames = append(vmSetNames, *avSetVMnames...)
+	}
+
 	return &vmSetNames, nil
 }
 
 func (ss *ScaleSet) GetNodeVMSetName(node *v1.Node) (string, error) {
+	vmManagementType, err := ss.getVMManagementTypeByNodeName(node.Name, azcache.CacheReadTypeUnsafe)
+	if err != nil {
+		klog.Errorf("Failed to check VM management type: %v", err)
+		return "", err
+	}
+
+	if vmManagementType == ManagedByAvSet {
+		// vm is managed by availability set.
+		return ss.availabilitySet.GetNodeVMSetName(node)
+	}
+	if vmManagementType == ManagedByVmssFlex {
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet.GetNodeVMSetName(node)
+	}
+
 	providerID := node.Spec.ProviderID
 	_, vmssName, err := getVmssAndResourceGroupNameByVMProviderID(providerID)
 	if err != nil {
-		klog.Warningf("ss.GetNodeVMSetName: the provider ID %s of node %s does not match the format of a VMSS instance, assuming it is managed by an availability set", providerID, node.Name)
-		return ss.availabilitySet.GetNodeVMSetName(node)
+		klog.Errorf("getVmssAndResourceGroupNameByVMProviderID failed: %v", err)
+		return "", err
 	}
 
 	klog.V(4).Infof("ss.GetNodeVMSetName: found vmss name %s from node name %s", vmssName, node.Name)
diff --git a/pkg/provider/azure_vmss_cache.go b/pkg/provider/azure_vmss_cache.go
index 62e4ad831..07e1a51cb 100644
--- a/pkg/provider/azure_vmss_cache.go
+++ b/pkg/provider/azure_vmss_cache.go
@@ -47,12 +47,23 @@ type vmssEntry struct {
 	lastUpdate    time.Time
 }
 
-type availabilitySetNodeEntry struct {
-	vmNames   sets.String
-	nodeNames sets.String
-	vms       []compute.VirtualMachine
+type nonVmssUniformNodesEntry struct {
+	vmssFlexVMNodeNames   sets.String
+	vmssFlexVMProviderIDs sets.String
+	avSetVMNodeNames      sets.String
+	avSetVMProviderIDs    sets.String
+	clusterNodeNames      sets.String
 }
 
+type VMManagementType string
+
+const (
+	ManagedByVmssUniform  VMManagementType = "ManagedByVmssUniform"
+	ManagedByVmssFlex     VMManagementType = "ManagedByVmssFlex"
+	ManagedByAvSet        VMManagementType = "ManagedByAvSet"
+	ManagedByUnknownVMSet VMManagementType = "ManagedByUnknownVMSet"
+)
+
 func (ss *ScaleSet) newVMSSCache() (*azcache.TimedCache, error) {
 	getter := func(key string) (interface{}, error) {
 		localCache := &sync.Map{} // [vmasName]*vmssEntry
@@ -81,11 +92,13 @@ func (ss *ScaleSet) newVMSSCache() (*azcache.TimedCache, error) {
 					klog.Warning("failed to get the name of VMSS")
 					continue
 				}
-				localCache.Store(*scaleSet.Name, &vmssEntry{
-					vmss:          &scaleSet,
-					resourceGroup: resourceGroup,
-					lastUpdate:    time.Now().UTC(),
-				})
+				if scaleSet.OrchestrationMode == "" || scaleSet.OrchestrationMode == compute.OrchestrationModeUniform {
+					localCache.Store(*scaleSet.Name, &vmssEntry{
+						vmss:          &scaleSet,
+						resourceGroup: resourceGroup,
+						lastUpdate:    time.Now().UTC(),
+					})
+				}
 			}
 		}
 
@@ -267,34 +280,19 @@ func (ss *ScaleSet) newVMSSVirtualMachinesCache(resourceGroupName, vmssName, cac
 }
 
 func (ss *ScaleSet) DeleteCacheForNode(nodeName string) error {
-	managedByAS, err := ss.isNodeManagedByAvailabilitySet(nodeName, azcache.CacheReadTypeUnsafe)
+	vmManagementType, err := ss.getVMManagementTypeByNodeName(nodeName, azcache.CacheReadTypeUnsafe)
 	if err != nil {
-		klog.Warningf("DeleteCacheForNode(%s): failed to check if the node is managed by AvailabilitySet", nodeName)
-		return nil
+		klog.Errorf("Failed to check VM management type: %v", err)
+		return err
 	}
-	if managedByAS {
-		cached, err := ss.availabilitySetNodesCache.Get(consts.AvailabilitySetNodesKey, azcache.CacheReadTypeDefault)
-		if err != nil {
-			klog.Warningf("DeleteCacheForNode(%s): failed to get availabilitySetNodesCache: %s", err.Error())
-			return nil
-		}
-
-		ss.lockMap.LockEntry(consts.AvailabilitySetNodesKey)
-		defer ss.lockMap.UnlockEntry(consts.AvailabilitySetNodesKey)
-
-		entry := cached.(*availabilitySetNodeEntry)
-		entry.nodeNames.Delete(nodeName)
-		entry.vmNames.Delete(nodeName)
-		for i := len(entry.vms) - 1; i >= 0; i-- {
-			if strings.EqualFold(to.String(entry.vms[i].Name), nodeName) {
-				entry.vms = append(entry.vms[:i], entry.vms[i+1:]...)
-				break
-			}
-		}
 
-		_ = ss.vmCache.Delete(nodeName)
-
-		return nil
+	if vmManagementType == ManagedByAvSet {
+		// vm is managed by availability set.
+		return ss.availabilitySet.DeleteCacheForNode(nodeName)
+	}
+	if vmManagementType == ManagedByVmssFlex {
+		// vm is managed by vmss flex.
+		return ss.flexScaleSet.DeleteCacheForNode(nodeName)
 	}
 
 	node, err := ss.getNodeIdentityByNodeName(nodeName, azcache.CacheReadTypeUnsafe)
@@ -324,24 +322,36 @@ func (ss *ScaleSet) DeleteCacheForNode(nodeName string) error {
 	return nil
 }
 
-func (ss *ScaleSet) newAvailabilitySetNodesCache() (*azcache.TimedCache, error) {
+func (ss *ScaleSet) newNonVmssUniformNodesCache() (*azcache.TimedCache, error) {
 	getter := func(key string) (interface{}, error) {
-		vmNames := sets.NewString()
+		klog.V(6).Infof("refresh the cache of NonVmssUniformNodesCache")
+		vmssFlexVMNodeNames := sets.NewString()
+		vmssFlexVMProviderIDs := sets.NewString()
+		avSetVMNodeNames := sets.NewString()
+		avSetVMProviderIDs := sets.NewString()
 		resourceGroups, err := ss.GetResourceGroups()
 		if err != nil {
 			return nil, err
 		}
 
-		vmList := make([]compute.VirtualMachine, 0)
 		for _, resourceGroup := range resourceGroups.List() {
 			vms, err := ss.Cloud.ListVirtualMachines(resourceGroup)
 			if err != nil {
-				return nil, fmt.Errorf("newAvailabilitySetNodesCache: failed to list vms in the resource group %s: %w", resourceGroup, err)
+				return nil, fmt.Errorf("getter function of nonVmssUniformNodesCache: failed to list vms in the resource group %s: %w", resourceGroup, err)
 			}
 			for _, vm := range vms {
-				if vm.Name != nil {
-					vmNames.Insert(to.String(vm.Name))
-					vmList = append(vmList, vm)
+				if vm.OsProfile != nil && vm.OsProfile.ComputerName != nil {
+					if vm.VirtualMachineScaleSet != nil {
+						vmssFlexVMNodeNames.Insert(strings.ToLower(to.String(vm.OsProfile.ComputerName)))
+						if vm.ID != nil {
+							vmssFlexVMProviderIDs.Insert(ss.ProviderName() + "://" + to.String(vm.ID))
+						}
+					} else {
+						avSetVMNodeNames.Insert(strings.ToLower(to.String(vm.OsProfile.ComputerName)))
+						if vm.ID != nil {
+							avSetVMProviderIDs.Insert(ss.ProviderName() + "://" + to.String(vm.ID))
+						}
+					}
 				}
 			}
 		}
@@ -352,43 +362,119 @@ func (ss *ScaleSet) newAvailabilitySetNodesCache() (*azcache.TimedCache, error)
 			return nil, err
 		}
 
-		localCache := &availabilitySetNodeEntry{
-			vmNames:   vmNames,
-			nodeNames: nodeNames,
-			vms:       vmList,
+		localCache := nonVmssUniformNodesEntry{
+			vmssFlexVMNodeNames:   vmssFlexVMNodeNames,
+			vmssFlexVMProviderIDs: vmssFlexVMProviderIDs,
+			avSetVMNodeNames:      avSetVMNodeNames,
+			avSetVMProviderIDs:    avSetVMProviderIDs,
+			clusterNodeNames:      nodeNames,
 		}
 
 		return localCache, nil
 	}
 
-	if ss.Config.AvailabilitySetNodesCacheTTLInSeconds == 0 {
-		ss.Config.AvailabilitySetNodesCacheTTLInSeconds = consts.AvailabilitySetNodesCacheTTLDefaultInSeconds
+	if ss.Config.NonVmssUniformNodesCacheTTLInSeconds == 0 {
+		ss.Config.NonVmssUniformNodesCacheTTLInSeconds = consts.NonVmssUniformNodesCacheTTLDefaultInSeconds
 	}
-	return azcache.NewTimedcache(time.Duration(ss.Config.AvailabilitySetNodesCacheTTLInSeconds)*time.Second, getter)
+	return azcache.NewTimedcache(time.Duration(ss.Config.NonVmssUniformNodesCacheTTLInSeconds)*time.Second, getter)
 }
 
-func (ss *ScaleSet) isNodeManagedByAvailabilitySet(nodeName string, crt azcache.AzureCacheReadType) (bool, error) {
-	// Assume all nodes are managed by VMSS when DisableAvailabilitySetNodes is enabled.
-	if ss.DisableAvailabilitySetNodes {
-		klog.V(6).Infof("Assuming node %q is managed by VMSS since DisableAvailabilitySetNodes is set to true", nodeName)
-		return false, nil
+func (ss *ScaleSet) getVMManagementTypeByNodeName(nodeName string, crt azcache.AzureCacheReadType) (VMManagementType, error) {
+	if ss.DisableAvailabilitySetNodes && !ss.EnableVmssFlexNodes {
+		return ManagedByVmssUniform, nil
 	}
-
-	cached, err := ss.availabilitySetNodesCache.Get(consts.AvailabilitySetNodesKey, crt)
+	ss.lockMap.LockEntry(consts.VMManagementTypeLockKey)
+	defer ss.lockMap.UnlockEntry(consts.VMManagementTypeLockKey)
+	cached, err := ss.nonVmssUniformNodesCache.Get(consts.NonVmssUniformNodesKey, crt)
 	if err != nil {
-		return false, err
+		return ManagedByUnknownVMSet, err
 	}
 
-	cachedNodes := cached.(*availabilitySetNodeEntry).nodeNames
+	cachedNodes := cached.(nonVmssUniformNodesEntry).clusterNodeNames
+
 	// if the node is not in the cache, assume the node has joined after the last cache refresh and attempt to refresh the cache.
 	if !cachedNodes.Has(nodeName) {
-		klog.V(2).Infof("Node %s has joined the cluster since the last VM cache refresh, refreshing the cache", nodeName)
-		cached, err = ss.availabilitySetNodesCache.Get(consts.AvailabilitySetNodesKey, azcache.CacheReadTypeForceRefresh)
+		klog.V(2).Infof("Node %s has joined the cluster since the last VM cache refresh in nonVmssUniformNodesEntry, refreshing the cache", nodeName)
+		cached, err = ss.nonVmssUniformNodesCache.Get(consts.NonVmssUniformNodesKey, azcache.CacheReadTypeForceRefresh)
 		if err != nil {
-			return false, err
+			return ManagedByUnknownVMSet, err
 		}
 	}
 
-	cachedVMs := cached.(*availabilitySetNodeEntry).vmNames
-	return cachedVMs.Has(nodeName), nil
+	cachedAvSetVMs := cached.(nonVmssUniformNodesEntry).avSetVMNodeNames
+	cachedVmssFlexVMs := cached.(nonVmssUniformNodesEntry).vmssFlexVMNodeNames
+
+	if cachedAvSetVMs.Has(nodeName) {
+		return ManagedByAvSet, nil
+	}
+	if cachedVmssFlexVMs.Has(nodeName) {
+		return ManagedByVmssFlex, nil
+	}
+	return ManagedByVmssUniform, nil
+}
+
+func (ss *ScaleSet) getVMManagementTypeByProviderID(providerID string, crt azcache.AzureCacheReadType) (VMManagementType, error) {
+	if ss.DisableAvailabilitySetNodes && !ss.EnableVmssFlexNodes {
+		return ManagedByVmssUniform, nil
+	}
+	_, err := extractScaleSetNameByProviderID(providerID)
+	if err == nil {
+		return ManagedByVmssUniform, nil
+	}
+
+	ss.lockMap.LockEntry(consts.VMManagementTypeLockKey)
+	defer ss.lockMap.UnlockEntry(consts.VMManagementTypeLockKey)
+	cached, err := ss.nonVmssUniformNodesCache.Get(consts.NonVmssUniformNodesKey, crt)
+	if err != nil {
+		return ManagedByUnknownVMSet, err
+	}
+
+	cachedVmssFlexVMProviderIDs := cached.(nonVmssUniformNodesEntry).vmssFlexVMProviderIDs
+	cachedAvSetVMProviderIDs := cached.(nonVmssUniformNodesEntry).avSetVMProviderIDs
+
+	if cachedAvSetVMProviderIDs.Has(providerID) {
+		return ManagedByAvSet, nil
+	}
+	if cachedVmssFlexVMProviderIDs.Has(providerID) {
+		return ManagedByVmssFlex, nil
+	}
+	return ManagedByUnknownVMSet, fmt.Errorf("getVMManagementTypeByProviderID : failed to check the providerID %s management type", providerID)
+
+}
+
+func (ss *ScaleSet) getVMManagementTypeByIPConfigurationID(ipConfigurationID string, crt azcache.AzureCacheReadType) (VMManagementType, error) {
+	if ss.DisableAvailabilitySetNodes && !ss.EnableVmssFlexNodes {
+		return ManagedByVmssUniform, nil
+	}
+
+	_, _, err := getScaleSetAndResourceGroupNameByIPConfigurationID(ipConfigurationID)
+	if err == nil {
+		return ManagedByVmssUniform, nil
+	}
+
+	ss.lockMap.LockEntry(consts.VMManagementTypeLockKey)
+	defer ss.lockMap.UnlockEntry(consts.VMManagementTypeLockKey)
+	cached, err := ss.nonVmssUniformNodesCache.Get(consts.NonVmssUniformNodesKey, crt)
+	if err != nil {
+		return ManagedByUnknownVMSet, err
+	}
+
+	matches := nicIDRE.FindStringSubmatch(ipConfigurationID)
+	if len(matches) != 3 {
+		return ManagedByUnknownVMSet, fmt.Errorf("can not extract nic name from ipConfigurationID (%s)", ipConfigurationID)
+	}
+
+	nicResourceGroup, nicName := matches[1], matches[2]
+	if nicResourceGroup == "" || nicName == "" {
+		return ManagedByUnknownVMSet, fmt.Errorf("invalid ip config ID %s", ipConfigurationID)
+	}
+
+	vmName := strings.Replace(nicName, "-nic", "", 1)
+
+	cachedAvSetVMs := cached.(nonVmssUniformNodesEntry).avSetVMNodeNames
+
+	if cachedAvSetVMs.Has(vmName) {
+		return ManagedByAvSet, nil
+	}
+	return ManagedByVmssFlex, nil
 }
diff --git a/pkg/provider/azure_vmssflex_cache.go b/pkg/provider/azure_vmssflex_cache.go
index 1b86b50cb..766058863 100644
--- a/pkg/provider/azure_vmssflex_cache.go
+++ b/pkg/provider/azure_vmssflex_cache.go
@@ -123,7 +123,7 @@ func (fs *FlexScaleSet) newVmssFlexVMCache() (*azcache.TimedCache, error) {
 	}
 
 	if fs.Config.VmssFlexVMCacheTTLInSeconds == 0 {
-		fs.Config.VmssFlexVMCacheTTLInSeconds = consts.VmssFlexVMCacheTTLInSeconds
+		fs.Config.VmssFlexVMCacheTTLInSeconds = consts.VmssFlexVMCacheTTLDefaultInSeconds
 	}
 	return azcache.NewTimedcache(time.Duration(fs.Config.VmssFlexVMCacheTTLInSeconds)*time.Second, getter)
 }


Test patch:
diff --git a/pkg/nodeipam/ipam/cloud_cidr_allocator_test.go b/pkg/nodeipam/ipam/cloud_cidr_allocator_test.go
index 7c9f3f810..f969d497a 100644
--- a/pkg/nodeipam/ipam/cloud_cidr_allocator_test.go
+++ b/pkg/nodeipam/ipam/cloud_cidr_allocator_test.go
@@ -32,6 +32,7 @@ import (
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes/fake"
 
+	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmclient/mockvmclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmssclient/mockvmssclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
 	azureprovider "sigs.k8s.io/cloud-provider-azure/pkg/provider"
@@ -233,10 +234,15 @@ func TestUpdateNodeSubnetMaskSizes(t *testing.T) {
 			expectedVMSS := compute.VirtualMachineScaleSet{
 				Name: to.StringPtr("vmss"),
 				Tags: tc.tags,
+				VirtualMachineScaleSetProperties: &compute.VirtualMachineScaleSetProperties{
+					OrchestrationMode: compute.OrchestrationModeUniform,
+				},
 			}
 			mockVMSSClient := ss.VirtualMachineScaleSetsClient.(*mockvmssclient.MockInterface)
 			mockVMSSClient.EXPECT().List(gomock.Any(), cloud.ResourceGroup).Return([]compute.VirtualMachineScaleSet{expectedVMSS}, nil).MaxTimes(1)
 			cloud.VMSet = ss
+			mockVMsClient := ss.VirtualMachinesClient.(*mockvmclient.MockInterface)
+			mockVMsClient.EXPECT().List(gomock.Any(), gomock.Any()).Return([]compute.VirtualMachine{}, nil).AnyTimes()
 
 			clusterCIDRs := func() []*net.IPNet {
 				_, cidrIPV4, _ := net.ParseCIDR("10.240.0.0/16")
diff --git a/pkg/provider/azure_ratelimit_test.go b/pkg/provider/azure_ratelimit_test.go
index 3ba565254..af0910151 100644
--- a/pkg/provider/azure_ratelimit_test.go
+++ b/pkg/provider/azure_ratelimit_test.go
@@ -56,6 +56,7 @@ var (
 		"networkResourceTenantId": "networkResourceTenantId",
 		"networkResourceSubscriptionId": "networkResourceSubscriptionId",
 		"availabilitySetNodesCacheTTLInSeconds": 100,
+		"nonVmssUniformNodesCacheTTLInSeconds": 100,
 		"vmssCacheTTLInSeconds": 100,
 		"vmssVirtualMachinesCacheTTLInSeconds": 100,
 		"vmCacheTTLInSeconds": 100,
@@ -132,6 +133,7 @@ func TestParseConfig(t *testing.T) {
 			},
 		},
 		AvailabilitySetNodesCacheTTLInSeconds: 100,
+		NonVmssUniformNodesCacheTTLInSeconds:  100,
 		VmssCacheTTLInSeconds:                 100,
 		VmssVirtualMachinesCacheTTLInSeconds:  100,
 		VMCacheTTLInSeconds:                   100,
diff --git a/pkg/provider/azure_test.go b/pkg/provider/azure_test.go
index b821aede7..35dd799f8 100644
--- a/pkg/provider/azure_test.go
+++ b/pkg/provider/azure_test.go
@@ -2123,6 +2123,7 @@ func TestNewCloudFromJSON(t *testing.T) {
 		"cloudProviderRateLimitQPS": 0.5,
 		"cloudProviderRateLimitBucket": 5,
 		"availabilitySetNodesCacheTTLInSeconds": 100,
+		"nonVmssUniformNodesCacheTTLInSeconds": 100,
 		"vmssCacheTTLInSeconds": 100,
 		"vmssVirtualMachinesCacheTTLInSeconds": 100,
 		"vmCacheTTLInSeconds": 100,
@@ -2185,6 +2186,7 @@ cloudProviderRatelimit: true
 cloudProviderRateLimitQPS: 0.5
 cloudProviderRateLimitBucket: 5
 availabilitySetNodesCacheTTLInSeconds: 100
+nonVmssUniformNodesCacheTTLInSeconds: 100
 vmssCacheTTLInSeconds: 100
 vmssVirtualMachinesCacheTTLInSeconds: 100
 vmCacheTTLInSeconds: 100
@@ -2277,6 +2279,9 @@ func validateConfig(t *testing.T, config string) { //nolint
 	if azureCloud.AvailabilitySetNodesCacheTTLInSeconds != 100 {
 		t.Errorf("got incorrect value for availabilitySetNodesCacheTTLInSeconds")
 	}
+	if azureCloud.NonVmssUniformNodesCacheTTLInSeconds != 100 {
+		t.Errorf("got incorrect value for nonVmssUniformNodesCacheTTLInSeconds")
+	}
 	if azureCloud.VmssCacheTTLInSeconds != 100 {
 		t.Errorf("got incorrect value for vmssCacheTTLInSeconds")
 	}
diff --git a/pkg/provider/azure_vmss_cache_test.go b/pkg/provider/azure_vmss_cache_test.go
index d0a82df9e..ae5d67e6c 100644
--- a/pkg/provider/azure_vmss_cache_test.go
+++ b/pkg/provider/azure_vmss_cache_test.go
@@ -28,14 +28,12 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 
-	"k8s.io/apimachinery/pkg/util/sets"
 	cloudprovider "k8s.io/cloud-provider"
 
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmclient/mockvmclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmssclient/mockvmssclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmssvmclient/mockvmssvmclient"
 	azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
-	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
 	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
 )
 
@@ -139,32 +137,6 @@ func TestVMSSVMCache(t *testing.T) {
 	assert.Equal(t, &vm, realVM.AsVirtualMachineScaleSetVM())
 }
 
-func TestDeleteCacheForAvailabilitySetNodeInVMSS(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-
-	ss, err := NewTestScaleSet(ctrl)
-	assert.NoError(t, err)
-
-	mockVMClient := mockvmclient.NewMockInterface(ctrl)
-	ss.cloud.VirtualMachinesClient = mockVMClient
-
-	mockVMClient.EXPECT().List(gomock.Any(), gomock.Any()).Return([]compute.VirtualMachine{
-		{Name: to.StringPtr("vm1")},
-	}, nil).AnyTimes()
-	ss.cloud.nodeNames = sets.NewString("vm1")
-
-	err = ss.DeleteCacheForNode("vm1")
-	assert.NoError(t, err)
-
-	cached, err := ss.availabilitySetNodesCache.Get(consts.AvailabilitySetNodesKey, azcache.CacheReadTypeUnsafe)
-	assert.NoError(t, err)
-	entry := cached.(*availabilitySetNodeEntry)
-	assert.Equal(t, 0, len(entry.nodeNames))
-	assert.Equal(t, 0, len(entry.vmNames))
-	assert.Equal(t, 0, len(entry.vms))
-}
-
 func TestVMSSVMCacheWithDeletingNodes(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -241,3 +213,244 @@ func TestVMSSVMCacheClearedWhenRGDeleted(t *testing.T) {
 	_, ok = ss.vmssVMCache.Load(cacheKey)
 	assert.Equal(t, false, ok)
 }
+
+func TestGetVMManagementTypeByNodeName(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	testVM1 := generateVmssFlexTestVMWithoutInstanceView(testVM1Spec)
+	testVM2 := generateVmssFlexTestVMWithoutInstanceView(testVM2Spec)
+	testVM2.VirtualMachineScaleSet = nil
+
+	testVMList := []compute.VirtualMachine{
+		testVM1,
+		testVM2,
+	}
+
+	testCases := []struct {
+		description                 string
+		nodeName                    string
+		DisableAvailabilitySetNodes bool
+		EnableVmssFlexNodes         bool
+		vmListErr                   *retry.Error
+		expectedVMManagementType    VMManagementType
+		expectedErr                 error
+	}{
+		{
+			description:              "getVMManagementTypeByNodeName should return ManagedByVmssFlex for vmss flex node",
+			nodeName:                 "vmssflex1000001",
+			vmListErr:                nil,
+			expectedVMManagementType: ManagedByVmssFlex,
+			expectedErr:              nil,
+		},
+		{
+			description:              "getVMManagementTypeByNodeName should return ManagedByAvSet for availabilityset node",
+			nodeName:                 "vmssflex1000002",
+			vmListErr:                nil,
+			expectedVMManagementType: ManagedByAvSet,
+			expectedErr:              nil,
+		},
+		{
+			description:              "getVMManagementTypeByNodeName should return ManagedByVmssUniform for vmss uniform node",
+			nodeName:                 "vmssflex1000003",
+			vmListErr:                nil,
+			expectedVMManagementType: ManagedByVmssUniform,
+			expectedErr:              nil,
+		},
+		{
+			description:                 "getVMManagementTypeByNodeName should return ManagedByVmssUniform if DisableAvailabilitySetNodes is set to true and EnableVmssFlexNodes is set to false",
+			nodeName:                    "anyName",
+			DisableAvailabilitySetNodes: true,
+			EnableVmssFlexNodes:         false,
+			vmListErr:                   nil,
+			expectedVMManagementType:    ManagedByVmssUniform,
+			expectedErr:                 nil,
+		},
+		{
+			description:              "getVMManagementTypeByNodeName should return ManagedByUnknownVMSet if error happens",
+			nodeName:                 "fakeName",
+			vmListErr:                &retry.Error{RawError: fmt.Errorf("failed to list VMs")},
+			expectedVMManagementType: ManagedByUnknownVMSet,
+			expectedErr:              fmt.Errorf("getter function of nonVmssUniformNodesCache: failed to list vms in the resource group rg: Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: failed to list VMs"),
+		},
+	}
+
+	for _, tc := range testCases {
+		ss, err := NewTestScaleSet(ctrl)
+		assert.NoError(t, err, tc.description)
+
+		ss.DisableAvailabilitySetNodes = tc.DisableAvailabilitySetNodes
+		ss.EnableVmssFlexNodes = tc.EnableVmssFlexNodes
+
+		mockVMClient := ss.cloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
+		mockVMClient.EXPECT().List(gomock.Any(), gomock.Any()).Return(testVMList, tc.vmListErr).AnyTimes()
+
+		vmManagementType, err := ss.getVMManagementTypeByNodeName(tc.nodeName, azcache.CacheReadTypeDefault)
+		assert.Equal(t, tc.expectedVMManagementType, vmManagementType, tc.description)
+		if tc.expectedErr != nil {
+			assert.EqualError(t, err, tc.expectedErr.Error(), tc.description)
+		}
+
+	}
+}
+
+func TestGetVMManagementTypeByProviderID(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	testVM1 := generateVmssFlexTestVMWithoutInstanceView(testVM1Spec)
+	testVM2 := generateVmssFlexTestVMWithoutInstanceView(testVM2Spec)
+	testVM2.VirtualMachineScaleSet = nil
+
+	testVMList := []compute.VirtualMachine{
+		testVM1,
+		testVM2,
+	}
+
+	testCases := []struct {
+		description                 string
+		providerID                  string
+		DisableAvailabilitySetNodes bool
+		EnableVmssFlexNodes         bool
+		vmListErr                   *retry.Error
+		expectedVMManagementType    VMManagementType
+		expectedErr                 error
+	}{
+		{
+			description:              "getVMManagementTypeByProviderID should return ManagedByVmssFlex for vmss flex node",
+			providerID:               "azure:///subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/testvm1",
+			vmListErr:                nil,
+			expectedVMManagementType: ManagedByVmssFlex,
+			expectedErr:              nil,
+		},
+		{
+			description:              "getVMManagementTypeByProviderID should return ManagedByAvSet for availabilityset node",
+			providerID:               "azure:///subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/testvm2",
+			vmListErr:                nil,
+			expectedVMManagementType: ManagedByAvSet,
+			expectedErr:              nil,
+		},
+		{
+			description:              "getVMManagementTypeByProviderID should return ManagedByVmssUniform for vmss uniform node",
+			providerID:               "azure:///subscriptions/script/resourceGroups/rg/providers/Microsoft.Compute/virtualMachineScaleSets/vmss/virtualMachines/1",
+			vmListErr:                nil,
+			expectedVMManagementType: ManagedByVmssUniform,
+			expectedErr:              nil,
+		},
+		{
+			description:                 "getVMManagementTypeByProviderID should return ManagedByVmssUniform if DisableAvailabilitySetNodes is set to true and EnableVmssFlexNodes is set to false",
+			providerID:                  "anyName",
+			DisableAvailabilitySetNodes: true,
+			EnableVmssFlexNodes:         false,
+			vmListErr:                   nil,
+			expectedVMManagementType:    ManagedByVmssUniform,
+			expectedErr:                 nil,
+		},
+		{
+			description:              "getVMManagementTypeByProviderID should return ManagedByUnknownVMSet if error happens",
+			providerID:               "fakeName",
+			vmListErr:                &retry.Error{RawError: fmt.Errorf("failed to list VMs")},
+			expectedVMManagementType: ManagedByUnknownVMSet,
+			expectedErr:              fmt.Errorf("getter function of nonVmssUniformNodesCache: failed to list vms in the resource group rg: Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: failed to list VMs"),
+		},
+	}
+
+	for _, tc := range testCases {
+		ss, err := NewTestScaleSet(ctrl)
+		assert.NoError(t, err, tc.description)
+
+		ss.DisableAvailabilitySetNodes = tc.DisableAvailabilitySetNodes
+		ss.EnableVmssFlexNodes = tc.EnableVmssFlexNodes
+
+		mockVMClient := ss.cloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
+		mockVMClient.EXPECT().List(gomock.Any(), gomock.Any()).Return(testVMList, tc.vmListErr).AnyTimes()
+
+		vmManagementType, err := ss.getVMManagementTypeByProviderID(tc.providerID, azcache.CacheReadTypeDefault)
+		assert.Equal(t, tc.expectedVMManagementType, vmManagementType, tc.description)
+		if tc.expectedErr != nil {
+			assert.EqualError(t, err, tc.expectedErr.Error(), tc.description)
+		}
+
+	}
+}
+
+func TestGetVMManagementTypeByIPConfigurationID(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	testVM1 := generateVmssFlexTestVMWithoutInstanceView(testVM1Spec)
+	testVM2 := generateVmssFlexTestVMWithoutInstanceView(testVM2Spec)
+	testVM2.VirtualMachineScaleSet = nil
+	testVM2.VirtualMachineProperties.OsProfile.ComputerName = to.StringPtr("testvm2")
+
+	testVMList := []compute.VirtualMachine{
+		testVM1,
+		testVM2,
+	}
+
+	testCases := []struct {
+		description                 string
+		ipConfigurationID           string
+		DisableAvailabilitySetNodes bool
+		EnableVmssFlexNodes         bool
+		vmListErr                   *retry.Error
+		expectedVMManagementType    VMManagementType
+		expectedErr                 error
+	}{
+		{
+			description:              "getVMManagementTypeByIPConfigurationID should return ManagedByVmssFlex for vmss flex node",
+			ipConfigurationID:        "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/networkInterfaces/testvm1-nic/ipConfigurations/pipConfig",
+			vmListErr:                nil,
+			expectedVMManagementType: ManagedByVmssFlex,
+			expectedErr:              nil,
+		},
+		{
+			description:              "getVMManagementTypeByIPConfigurationID should return ManagedByAvSet for availabilityset node",
+			ipConfigurationID:        "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/networkInterfaces/testvm2-nic/ipConfigurations/pipConfig",
+			vmListErr:                nil,
+			expectedVMManagementType: ManagedByAvSet,
+			expectedErr:              nil,
+		},
+		{
+			description:              "getVMManagementTypeByIPConfigurationID should return ManagedByVmssUniform for vmss uniform node",
+			ipConfigurationID:        "/subscriptions/script/resourceGroups/rg/providers/Microsoft.Compute/virtualMachineScaleSets/vmss/virtualMachines/1/networkInterfaces/nic1",
+			vmListErr:                nil,
+			expectedVMManagementType: ManagedByVmssUniform,
+			expectedErr:              nil,
+		},
+		{
+			description:                 "getVMManagementTypeByIPConfigurationID should return ManagedByVmssUniform if DisableAvailabilitySetNodes is set to true and EnableVmssFlexNodes is set to false",
+			ipConfigurationID:           "anyID",
+			DisableAvailabilitySetNodes: true,
+			EnableVmssFlexNodes:         false,
+			vmListErr:                   nil,
+			expectedVMManagementType:    ManagedByVmssUniform,
+			expectedErr:                 nil,
+		},
+		{
+			description:              "getVMManagementTypeByIPConfigurationID should return ManagedByUnknownVMSet if error happens",
+			ipConfigurationID:        "fakeID",
+			vmListErr:                &retry.Error{RawError: fmt.Errorf("failed to list VMs")},
+			expectedVMManagementType: ManagedByUnknownVMSet,
+			expectedErr:              fmt.Errorf("getter function of nonVmssUniformNodesCache: failed to list vms in the resource group rg: Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: failed to list VMs"),
+		},
+	}
+
+	for _, tc := range testCases {
+		ss, err := NewTestScaleSet(ctrl)
+		assert.NoError(t, err, tc.description)
+
+		ss.DisableAvailabilitySetNodes = tc.DisableAvailabilitySetNodes
+		ss.EnableVmssFlexNodes = tc.EnableVmssFlexNodes
+
+		mockVMClient := ss.cloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
+		mockVMClient.EXPECT().List(gomock.Any(), gomock.Any()).Return(testVMList, tc.vmListErr).AnyTimes()
+
+		vmManagementType, err := ss.getVMManagementTypeByIPConfigurationID(tc.ipConfigurationID, azcache.CacheReadTypeDefault)
+		assert.Equal(t, tc.expectedVMManagementType, vmManagementType, tc.description)
+		if tc.expectedErr != nil {
+			assert.EqualError(t, err, tc.expectedErr.Error(), tc.description)
+		}
+
+	}
+}
diff --git a/pkg/provider/azure_vmss_test.go b/pkg/provider/azure_vmss_test.go
index ea2752273..0a03d0030 100644
--- a/pkg/provider/azure_vmss_test.go
+++ b/pkg/provider/azure_vmss_test.go
@@ -80,6 +80,7 @@ func buildTestVMSSWithLB(name, namePrefix string, lbBackendpoolIDs []string, ipv
 	expectedVMSS := compute.VirtualMachineScaleSet{
 		Name: &name,
 		VirtualMachineScaleSetProperties: &compute.VirtualMachineScaleSetProperties{
+			OrchestrationMode: compute.OrchestrationModeUniform,
 			ProvisioningState: to.StringPtr("Running"),
 			VirtualMachineProfile: &compute.VirtualMachineScaleSetVMProfile{
 				OsProfile: &compute.VirtualMachineScaleSetOSProfile{
@@ -106,6 +107,7 @@ func buildTestVMSS(name, computerNamePrefix string) compute.VirtualMachineScaleS
 	return compute.VirtualMachineScaleSet{
 		Name: &name,
 		VirtualMachineScaleSetProperties: &compute.VirtualMachineScaleSetProperties{
+			OrchestrationMode: compute.OrchestrationModeUniform,
 			VirtualMachineProfile: &compute.VirtualMachineScaleSetVMProfile{
 				OsProfile: &compute.VirtualMachineScaleSetOSProfile{
 					ComputerNamePrefix: &computerNamePrefix,
@@ -609,6 +611,9 @@ func TestGetNodeNameByIPConfigurationID(t *testing.T) {
 		expectedVMs, _, _ := buildTestVirtualMachineEnv(ss.cloud, test.scaleSet, "", 0, test.vmList, "", false)
 		mockVMSSVMClient.EXPECT().List(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(expectedVMs, nil).AnyTimes()
 
+		mockVMsClient := ss.cloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
+		mockVMsClient.EXPECT().List(gomock.Any(), gomock.Any()).Return([]compute.VirtualMachine{}, nil).AnyTimes()
+
 		nodeName, scalesetName, err := ss.GetNodeNameByIPConfigurationID(test.ipConfigurationID)
 		if test.expectError {
 			assert.Error(t, err, test.description)
@@ -1010,7 +1015,7 @@ func TestGetInstanceTypeByNodeName(t *testing.T) {
 			vmList:       []string{"vmss-vm-000000"},
 			vmClientErr:  &retry.Error{RawError: fmt.Errorf("error")},
 			expectedType: "",
-			expectedErr:  fmt.Errorf("newAvailabilitySetNodesCache: failed to list vms in the resource group rg: Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: error"),
+			expectedErr:  fmt.Errorf("getter function of nonVmssUniformNodesCache: failed to list vms in the resource group rg: Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: error"),
 		},
 	}
 
@@ -1134,7 +1139,7 @@ func TestGetPrimaryInterface(t *testing.T) {
 			vmList:              []string{"vmss-vm-000000"},
 			hasPrimaryInterface: true,
 			vmClientErr:         &retry.Error{RawError: fmt.Errorf("error")},
-			expectedErr:         fmt.Errorf("newAvailabilitySetNodesCache: failed to list vms in the resource group rg: Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: error"),
+			expectedErr:         fmt.Errorf("getter function of nonVmssUniformNodesCache: failed to list vms in the resource group rg: Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: error"),
 		},
 		{
 			description:         "GetPrimaryInterface should report the error if vmss client returns retry error",
@@ -1300,7 +1305,7 @@ func TestGetPrivateIPsByNodeName(t *testing.T) {
 			vmList:             []string{"vmss-vm-000000"},
 			vmClientErr:        &retry.Error{RawError: fmt.Errorf("error")},
 			expectedPrivateIPs: []string{},
-			expectedErr:        fmt.Errorf("newAvailabilitySetNodesCache: failed to list vms in the resource group rg: Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: error"),
+			expectedErr:        fmt.Errorf("getter function of nonVmssUniformNodesCache: failed to list vms in the resource group rg: Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: error"),
 		},
 	}
 
@@ -1641,7 +1646,7 @@ func TestGetVMSetNames(t *testing.T) {
 					},
 				},
 			},
-			expectedErr: fmt.Errorf("instance not found"),
+			expectedErr: fmt.Errorf("GetAgentPoolVMSetNames: failed to execute getAgentPoolScaleSets: instance not found"),
 		},
 		{
 			description: "GetVMSetNames should return the correct vmss names",
@@ -2585,8 +2590,8 @@ func TestEnsureBackendPoolDeleted(t *testing.T) {
 		mockVMSSVMClient.EXPECT().List(gomock.Any(), ss.ResourceGroup, testVMSSName, gomock.Any()).Return(expectedVMSSVMs, nil).AnyTimes()
 		mockVMSSVMClient.EXPECT().UpdateVMs(gomock.Any(), ss.ResourceGroup, testVMSSName, gomock.Any(), gomock.Any(), gomock.Any()).Return(test.vmClientErr).Times(test.expectedVMSSVMPutTimes)
 
-		mockVMClient := ss.cloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
-		mockVMClient.EXPECT().List(gomock.Any(), ss.ResourceGroup).Return([]compute.VirtualMachine{}, nil).AnyTimes()
+		mockVMsClient := ss.cloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
+		mockVMsClient.EXPECT().List(gomock.Any(), gomock.Any()).Return([]compute.VirtualMachine{}, nil).AnyTimes()
 
 		err = ss.EnsureBackendPoolDeleted(&v1.Service{}, test.backendpoolID, testVMSSName, test.backendAddressPools, true)
 		assert.Equal(t, test.expectedErr, err != nil, test.description+", but an error occurs")
@@ -2692,7 +2697,7 @@ func TestGetNodeCIDRMasksByProviderID(t *testing.T) {
 		{
 			description: "GetNodeCIDRMasksByProviderID should report an error if the providerID is not valid",
 			providerID:  "invalid",
-			expectedErr: ErrorNotVmssInstance,
+			expectedErr: fmt.Errorf("getVMManagementTypeByProviderID : failed to check the providerID invalid management type"),
 		},
 		{
 			description: "GetNodeCIDRMaksByProviderID should return the correct mask sizes",
@@ -2729,14 +2734,20 @@ func TestGetNodeCIDRMasksByProviderID(t *testing.T) {
 			expectedVMSS := compute.VirtualMachineScaleSet{
 				Name: to.StringPtr("vmss"),
 				Tags: tc.tags,
+				VirtualMachineScaleSetProperties: &compute.VirtualMachineScaleSetProperties{
+					OrchestrationMode: compute.OrchestrationModeUniform,
+				},
 			}
 			mockVMSSClient := ss.cloud.VirtualMachineScaleSetsClient.(*mockvmssclient.MockInterface)
 			mockVMSSClient.EXPECT().List(gomock.Any(), ss.ResourceGroup).Return([]compute.VirtualMachineScaleSet{expectedVMSS}, nil).MaxTimes(1)
 
+			mockVMsClient := ss.cloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
+			mockVMsClient.EXPECT().List(gomock.Any(), gomock.Any()).Return([]compute.VirtualMachine{}, nil).AnyTimes()
+
 			ipv4MaskSize, ipv6MaskSize, err := ss.GetNodeCIDRMasksByProviderID(tc.providerID)
-			assert.Equal(t, tc.expectedErr, err)
-			assert.Equal(t, tc.expectedIPV4MaskSize, ipv4MaskSize)
-			assert.Equal(t, tc.expectedIPV6MaskSize, ipv6MaskSize)
+			assert.Equal(t, tc.expectedErr, err, tc.description)
+			assert.Equal(t, tc.expectedIPV4MaskSize, ipv4MaskSize, tc.description)
+			assert.Equal(t, tc.expectedIPV6MaskSize, ipv6MaskSize, tc.description)
 		})
 	}
 }
@@ -2752,6 +2763,9 @@ func TestGetAgentPoolVMSetNamesMixedInstances(t *testing.T) {
 		{
 			Name: to.StringPtr("vm-0"),
 			VirtualMachineProperties: &compute.VirtualMachineProperties{
+				OsProfile: &compute.OSProfile{
+					ComputerName: to.StringPtr("vm-0"),
+				},
 				AvailabilitySet: &compute.SubResource{
 					ID: to.StringPtr("vmas-0"),
 				},
@@ -2760,6 +2774,9 @@ func TestGetAgentPoolVMSetNamesMixedInstances(t *testing.T) {
 		{
 			Name: to.StringPtr("vm-1"),
 			VirtualMachineProperties: &compute.VirtualMachineProperties{
+				OsProfile: &compute.OSProfile{
+					ComputerName: to.StringPtr("vm-1"),
+				},
 				AvailabilitySet: &compute.SubResource{
 					ID: to.StringPtr("vmas-1"),
 				},
@@ -2809,13 +2826,13 @@ func TestGetNodeVMSetNameVMSS(t *testing.T) {
 
 	ss, err := NewTestScaleSet(ctrl)
 	assert.NoError(t, err)
-	mockVMSet := NewMockVMSet(ctrl)
-	mockVMSet.EXPECT().GetNodeVMSetName(gomock.Any()).Return("as", nil)
-	ss.availabilitySet = mockVMSet
+
+	mockVMsClient := ss.cloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
+	mockVMsClient.EXPECT().List(gomock.Any(), gomock.Any()).Return([]compute.VirtualMachine{}, nil).AnyTimes()
 
 	vmSetName, err := ss.GetNodeVMSetName(node)
-	assert.NoError(t, err)
-	assert.Equal(t, "as", vmSetName)
+	assert.Equal(t, ErrorNotVmssInstance, err)
+	assert.Equal(t, "", vmSetName)
 
 	node = &v1.Node{
 		Spec: v1.NodeSpec{
@@ -2864,6 +2881,9 @@ func TestScaleSet_VMSSBatchSize(t *testing.T) {
 			Tags: map[string]*string{
 				consts.VMSSTagForBatchOperation: to.StringPtr(""),
 			},
+			VirtualMachineScaleSetProperties: &compute.VirtualMachineScaleSetProperties{
+				OrchestrationMode: compute.OrchestrationModeUniform,
+			},
 		}
 		mockVMSSClient := ss.Cloud.VirtualMachineScaleSetsClient.(*mockvmssclient.MockInterface)
 		mockVMSSClient.EXPECT().List(gomock.Any(), gomock.Any()).
@@ -2883,6 +2903,9 @@ func TestScaleSet_VMSSBatchSize(t *testing.T) {
 
 		scaleSet := compute.VirtualMachineScaleSet{
 			Name: to.StringPtr("bar"),
+			VirtualMachineScaleSetProperties: &compute.VirtualMachineScaleSetProperties{
+				OrchestrationMode: compute.OrchestrationModeUniform,
+			},
 		}
 		mockVMSSClient := ss.Cloud.VirtualMachineScaleSetsClient.(*mockvmssclient.MockInterface)
 		mockVMSSClient.EXPECT().List(gomock.Any(), gomock.Any()).