diff --git a/cluster-autoscaler/cloudprovider/alicloud/alicloud_cloud_provider.go b/cluster-autoscaler/cloudprovider/alicloud/alicloud_cloud_provider.go
index c3aa703cf6b..17c5fdc131e 100644
--- a/cluster-autoscaler/cloudprovider/alicloud/alicloud_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/alicloud/alicloud_cloud_provider.go
@@ -127,6 +127,11 @@ func (ali *aliCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovider.N
 	return ali.manager.GetAsgForInstance(instanceId)
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (ali *aliCloudProvider) HasInstance(*apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available.
 func (ali *aliCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
 	return nil, cloudprovider.ErrNotImplemented
diff --git a/cluster-autoscaler/cloudprovider/aws/aws_cloud_provider.go b/cluster-autoscaler/cloudprovider/aws/aws_cloud_provider.go
index dd4dc3fc7d9..23b00fb4668 100644
--- a/cluster-autoscaler/cloudprovider/aws/aws_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/aws/aws_cloud_provider.go
@@ -120,6 +120,11 @@ func (aws *awsCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovider.N
 	}, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (aws *awsCloudProvider) HasInstance(*apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available.
 func (aws *awsCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
 	return nil, cloudprovider.ErrNotImplemented
diff --git a/cluster-autoscaler/cloudprovider/azure/azure_cloud_provider.go b/cluster-autoscaler/cloudprovider/azure/azure_cloud_provider.go
index 491c4202344..40a0205922a 100644
--- a/cluster-autoscaler/cloudprovider/azure/azure_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/azure/azure_cloud_provider.go
@@ -106,6 +106,11 @@ func (azure *AzureCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovid
 	return azure.azureManager.GetNodeGroupForInstance(ref)
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (azure *AzureCloudProvider) HasInstance(*apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available.
 func (azure *AzureCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
 	return nil, cloudprovider.ErrNotImplemented
diff --git a/cluster-autoscaler/cloudprovider/baiducloud/baiducloud_cloud_provider.go b/cluster-autoscaler/cloudprovider/baiducloud/baiducloud_cloud_provider.go
index 1dc695f46e6..8849d962661 100644
--- a/cluster-autoscaler/cloudprovider/baiducloud/baiducloud_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/baiducloud/baiducloud_cloud_provider.go
@@ -180,6 +180,11 @@ func (baiducloud *baiducloudCloudProvider) NodeGroupForNode(node *apiv1.Node) (c
 	return asg, err
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (baiducloud *baiducloudCloudProvider) HasInstance(*apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available.
 // Implementation optional.
 func (baiducloud *baiducloudCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
diff --git a/cluster-autoscaler/cloudprovider/bizflycloud/bizflycloud_cloud_provider.go b/cluster-autoscaler/cloudprovider/bizflycloud/bizflycloud_cloud_provider.go
index 1a52f18295e..59f2149227e 100644
--- a/cluster-autoscaler/cloudprovider/bizflycloud/bizflycloud_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/bizflycloud/bizflycloud_cloud_provider.go
@@ -104,6 +104,11 @@ func (d *bizflycloudCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprov
 	return nil, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (d *bizflycloudCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not
 // available. Implementation optional.
 func (d *bizflycloudCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
diff --git a/cluster-autoscaler/cloudprovider/brightbox/brightbox_cloud_provider.go b/cluster-autoscaler/cloudprovider/brightbox/brightbox_cloud_provider.go
index f069f901926..322a02b2fca 100644
--- a/cluster-autoscaler/cloudprovider/brightbox/brightbox_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/brightbox/brightbox_cloud_provider.go
@@ -81,6 +81,11 @@ func (b *brightboxCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovid
 	return nil, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (b *brightboxCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Refresh is before every main loop and can be used to dynamically
 // update cloud provider state.
 // In particular the list of node groups returned by NodeGroups can
diff --git a/cluster-autoscaler/cloudprovider/cherryservers/cherry_cloud_provider.go b/cluster-autoscaler/cloudprovider/cherryservers/cherry_cloud_provider.go
index b3f6b8888ce..95fabbcb30a 100644
--- a/cluster-autoscaler/cloudprovider/cherryservers/cherry_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/cherryservers/cherry_cloud_provider.go
@@ -122,6 +122,11 @@ func (ccp *cherryCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovide
 	return nil, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (ccp *cherryCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available.
 func (ccp *cherryCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
 	return nil, cloudprovider.ErrNotImplemented
diff --git a/cluster-autoscaler/cloudprovider/civo/civo_cloud_provider.go b/cluster-autoscaler/cloudprovider/civo/civo_cloud_provider.go
index 0c724574139..4b8e6606143 100644
--- a/cluster-autoscaler/cloudprovider/civo/civo_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/civo/civo_cloud_provider.go
@@ -99,6 +99,11 @@ func (d *civoCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovider.No
 	return nil, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (d *civoCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not
 // available. Implementation optional.
 func (d *civoCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
diff --git a/cluster-autoscaler/cloudprovider/cloud_provider.go b/cluster-autoscaler/cloudprovider/cloud_provider.go
index b35a7d5a8ed..566f63b7b26 100644
--- a/cluster-autoscaler/cloudprovider/cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/cloud_provider.go
@@ -100,6 +100,10 @@ type CloudProvider interface {
 	// occurred. Must be implemented.
 	NodeGroupForNode(*apiv1.Node) (NodeGroup, error)
 
+	// HasInstance returns whether the node has corresponding instance in cloud provider,
+	// true if the node has an instance, false if it no longer exists
+	HasInstance(*apiv1.Node) (bool, error)
+
 	// Pricing returns pricing model for this cloud provider or error if not available.
 	// Implementation optional.
 	Pricing() (PricingModel, errors.AutoscalerError)
diff --git a/cluster-autoscaler/cloudprovider/cloudstack/cloudstack_cloud_provider.go b/cluster-autoscaler/cloudprovider/cloudstack/cloudstack_cloud_provider.go
index 63a228d8e36..e4537132125 100644
--- a/cluster-autoscaler/cloudprovider/cloudstack/cloudstack_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/cloudstack/cloudstack_cloud_provider.go
@@ -68,6 +68,11 @@ func (provider *cloudStackCloudProvider) NodeGroupForNode(node *v1.Node) (cloudp
 	return provider.manager.clusterForNode(node)
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (provider *cloudStackCloudProvider) HasInstance(node *v1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Cleanup cleans up open resources before the cloud provider is destroyed, i.e. go routines etc.
 func (provider *cloudStackCloudProvider) Cleanup() error {
 	return provider.manager.cleanup()
diff --git a/cluster-autoscaler/cloudprovider/clusterapi/clusterapi_provider.go b/cluster-autoscaler/cloudprovider/clusterapi/clusterapi_provider.go
index 290be55f538..e2928f9c945 100644
--- a/cluster-autoscaler/cloudprovider/clusterapi/clusterapi_provider.go
+++ b/cluster-autoscaler/cloudprovider/clusterapi/clusterapi_provider.go
@@ -81,6 +81,11 @@ func (p *provider) NodeGroupForNode(node *corev1.Node) (cloudprovider.NodeGroup,
 	return ng, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (p *provider) HasInstance(node *corev1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 func (*provider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
 	return nil, cloudprovider.ErrNotImplemented
 }
diff --git a/cluster-autoscaler/cloudprovider/digitalocean/digitalocean_cloud_provider.go b/cluster-autoscaler/cloudprovider/digitalocean/digitalocean_cloud_provider.go
index 36fb1e5e2e8..1bbf51c3e98 100644
--- a/cluster-autoscaler/cloudprovider/digitalocean/digitalocean_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/digitalocean/digitalocean_cloud_provider.go
@@ -101,6 +101,11 @@ func (d *digitaloceanCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudpro
 	return nil, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (d *digitaloceanCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not
 // available. Implementation optional.
 func (d *digitaloceanCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
diff --git a/cluster-autoscaler/cloudprovider/exoscale/exoscale_cloud_provider.go b/cluster-autoscaler/cloudprovider/exoscale/exoscale_cloud_provider.go
index 82ed546896e..021ef6316e4 100644
--- a/cluster-autoscaler/cloudprovider/exoscale/exoscale_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/exoscale/exoscale_cloud_provider.go
@@ -131,6 +131,11 @@ func (e *exoscaleCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovide
 	return nodeGroup, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (e *exoscaleCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available.
 // Implementation optional.
 func (e *exoscaleCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
diff --git a/cluster-autoscaler/cloudprovider/externalgrpc/externalgrpc_cloud_provider.go b/cluster-autoscaler/cloudprovider/externalgrpc/externalgrpc_cloud_provider.go
index d8743e918ba..0fb7917f447 100644
--- a/cluster-autoscaler/cloudprovider/externalgrpc/externalgrpc_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/externalgrpc/externalgrpc_cloud_provider.go
@@ -134,6 +134,11 @@ func (e *externalGrpcCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudpro
 	return ng, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (e *externalGrpcCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // pricingModel implements cloudprovider.PricingModel interface.
 type pricingModel struct {
 	client protos.CloudProviderClient
diff --git a/cluster-autoscaler/cloudprovider/gce/gce_cloud_provider.go b/cluster-autoscaler/cloudprovider/gce/gce_cloud_provider.go
index 39b62c50cab..83a275c25ce 100644
--- a/cluster-autoscaler/cloudprovider/gce/gce_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/gce/gce_cloud_provider.go
@@ -101,6 +101,11 @@ func (gce *GceCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovider.N
 	return mig, err
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (gce *GceCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available.
 func (gce *GceCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
 	return gce.pricingModel, nil
diff --git a/cluster-autoscaler/cloudprovider/hetzner/hetzner_cloud_provider.go b/cluster-autoscaler/cloudprovider/hetzner/hetzner_cloud_provider.go
index 7e7f35bf521..fd8c8d1121c 100644
--- a/cluster-autoscaler/cloudprovider/hetzner/hetzner_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/hetzner/hetzner_cloud_provider.go
@@ -99,6 +99,11 @@ func (d *HetznerCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovider
 	return group, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (d *HetznerCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not
 // available. Implementation optional.
 func (d *HetznerCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
diff --git a/cluster-autoscaler/cloudprovider/huaweicloud/huaweicloud_cloud_provider.go b/cluster-autoscaler/cloudprovider/huaweicloud/huaweicloud_cloud_provider.go
index feff955034b..7a66aec2bac 100644
--- a/cluster-autoscaler/cloudprovider/huaweicloud/huaweicloud_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/huaweicloud/huaweicloud_cloud_provider.go
@@ -123,6 +123,11 @@ func (hcp *huaweicloudCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudpr
 	return hcp.cloudServiceManager.GetAsgForInstance(instanceID)
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (hcp *huaweicloudCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available. Not implemented.
 func (hcp *huaweicloudCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
 	return nil, cloudprovider.ErrNotImplemented
diff --git a/cluster-autoscaler/cloudprovider/ionoscloud/ionoscloud_cloud_provider.go b/cluster-autoscaler/cloudprovider/ionoscloud/ionoscloud_cloud_provider.go
index c6cd1617362..21d758c5d9c 100644
--- a/cluster-autoscaler/cloudprovider/ionoscloud/ionoscloud_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/ionoscloud/ionoscloud_cloud_provider.go
@@ -232,6 +232,11 @@ func (ic *IonosCloudCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprov
 	return nil, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (ic *IonosCloudCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not
 // available. Implementation optional.
 func (ic *IonosCloudCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
diff --git a/cluster-autoscaler/cloudprovider/kamatera/kamatera_cloud_provider.go b/cluster-autoscaler/cloudprovider/kamatera/kamatera_cloud_provider.go
index a77d7376a01..2c5187ad1d3 100644
--- a/cluster-autoscaler/cloudprovider/kamatera/kamatera_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/kamatera/kamatera_cloud_provider.go
@@ -70,6 +70,11 @@ func (k *kamateraCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovide
 	return nil, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (k *kamateraCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available.
 // Implementation optional.
 func (k *kamateraCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
diff --git a/cluster-autoscaler/cloudprovider/kubemark/kubemark_linux.go b/cluster-autoscaler/cloudprovider/kubemark/kubemark_linux.go
index be092583743..915109b5f96 100644
--- a/cluster-autoscaler/cloudprovider/kubemark/kubemark_linux.go
+++ b/cluster-autoscaler/cloudprovider/kubemark/kubemark_linux.go
@@ -139,6 +139,11 @@ func (kubemark *KubemarkCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloud
 	return nil, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (kubemark *KubemarkCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // GetAvailableMachineTypes get all machine types that can be requested from the cloud provider.
 // Implementation optional.
 func (kubemark *KubemarkCloudProvider) GetAvailableMachineTypes() ([]string, error) {
diff --git a/cluster-autoscaler/cloudprovider/kubemark/kubemark_other.go b/cluster-autoscaler/cloudprovider/kubemark/kubemark_other.go
index dc857e68c74..895ced6d889 100644
--- a/cluster-autoscaler/cloudprovider/kubemark/kubemark_other.go
+++ b/cluster-autoscaler/cloudprovider/kubemark/kubemark_other.go
@@ -80,6 +80,11 @@ func (kubemark *KubemarkCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloud
 	return nil, cloudprovider.ErrNotImplemented
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (kubemark *KubemarkCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // GetAvailableMachineTypes get all machine types that can be requested from the cloud provider.
 // Implementation optional.
 func (kubemark *KubemarkCloudProvider) GetAvailableMachineTypes() ([]string, error) {
diff --git a/cluster-autoscaler/cloudprovider/linode/linode_cloud_provider.go b/cluster-autoscaler/cloudprovider/linode/linode_cloud_provider.go
index b14e97f151e..a5ade015cb5 100644
--- a/cluster-autoscaler/cloudprovider/linode/linode_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/linode/linode_cloud_provider.go
@@ -67,6 +67,11 @@ func (l *linodeCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovider.
 	return nil, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (l *linodeCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available.
 // Implementation optional.
 func (l *linodeCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
diff --git a/cluster-autoscaler/cloudprovider/magnum/magnum_cloud_provider.go b/cluster-autoscaler/cloudprovider/magnum/magnum_cloud_provider.go
index c48f54560c8..ea72859e39d 100644
--- a/cluster-autoscaler/cloudprovider/magnum/magnum_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/magnum/magnum_cloud_provider.go
@@ -135,6 +135,11 @@ func (mcp *magnumCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovide
 	return nil, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (mcp *magnumCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing is not implemented.
 func (mcp *magnumCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
 	return nil, cloudprovider.ErrNotImplemented
diff --git a/cluster-autoscaler/cloudprovider/mocks/CloudProvider.go b/cluster-autoscaler/cloudprovider/mocks/CloudProvider.go
index 6363f5d9d20..f07960f742d 100644
--- a/cluster-autoscaler/cloudprovider/mocks/CloudProvider.go
+++ b/cluster-autoscaler/cloudprovider/mocks/CloudProvider.go
@@ -177,6 +177,29 @@ func (_m *CloudProvider) NodeGroupForNode(_a0 *v1.Node) (cloudprovider.NodeGroup
 	return r0, r1
 }
 
+// HasInstance provides a mock function with given fields:
+func (_m *CloudProvider) HasInstance(_a0 *v1.Node) (bool, error) {
+	ret := _m.Called(_a0)
+
+	var r0 bool
+	if rf, ok := ret.Get(0).(func(*v1.Node) bool); ok {
+		r0 = rf(_a0)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(bool)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(*v1.Node) error); ok {
+		r1 = rf(_a0)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // NodeGroups provides a mock function with given fields:
 func (_m *CloudProvider) NodeGroups() []cloudprovider.NodeGroup {
 	ret := _m.Called()
diff --git a/cluster-autoscaler/cloudprovider/oci/oci_cloud_provider.go b/cluster-autoscaler/cloudprovider/oci/oci_cloud_provider.go
index 7eb9fbe704d..7aa83a1dacd 100644
--- a/cluster-autoscaler/cloudprovider/oci/oci_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/oci/oci_cloud_provider.go
@@ -96,6 +96,11 @@ func (ocp *OciCloudProvider) NodeGroupForNode(n *apiv1.Node) (cloudprovider.Node
 	return ng, err
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (ocp *OciCloudProvider) HasInstance(n *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available.
 // Implementation optional.
 func (ocp *OciCloudProvider) Pricing() (cloudprovider.PricingModel, caerrors.AutoscalerError) {
diff --git a/cluster-autoscaler/cloudprovider/ovhcloud/ovh_cloud_provider.go b/cluster-autoscaler/cloudprovider/ovhcloud/ovh_cloud_provider.go
index 075c6cb78a1..c3e4fe03f08 100644
--- a/cluster-autoscaler/cloudprovider/ovhcloud/ovh_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/ovhcloud/ovh_cloud_provider.go
@@ -151,6 +151,11 @@ func (provider *OVHCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovi
 	return ng, err
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (provider *OVHCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // findNodeGroupFromCache tries to retrieve the associated node group from an already built mapping in cache
 func (provider *OVHCloudProvider) findNodeGroupFromCache(providerID string) cloudprovider.NodeGroup {
 	if ng, ok := provider.manager.NodeGroupPerProviderID[providerID]; ok {
diff --git a/cluster-autoscaler/cloudprovider/packet/packet_cloud_provider.go b/cluster-autoscaler/cloudprovider/packet/packet_cloud_provider.go
index ca59a84f3d2..4ac539b53f2 100644
--- a/cluster-autoscaler/cloudprovider/packet/packet_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/packet/packet_cloud_provider.go
@@ -120,6 +120,11 @@ func (pcp *packetCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovide
 	return nil, fmt.Errorf("Could not find group for node: %s", node.Spec.ProviderID)
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (pcp *packetCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available.
 func (pcp *packetCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
 	return &PacketPriceModel{}, nil
diff --git a/cluster-autoscaler/cloudprovider/rancher/rancher_provider.go b/cluster-autoscaler/cloudprovider/rancher/rancher_provider.go
index 337234790e2..ee6f0579ece 100644
--- a/cluster-autoscaler/cloudprovider/rancher/rancher_provider.go
+++ b/cluster-autoscaler/cloudprovider/rancher/rancher_provider.go
@@ -172,6 +172,11 @@ func (provider *RancherCloudProvider) NodeGroupForNode(node *corev1.Node) (cloud
 	return nil, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (provider *RancherCloudProvider) HasInstance(node *corev1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // GetAvailableMachineTypes get all machine types that can be requested from the cloud provider.
 // Implementation optional.
 func (provider *RancherCloudProvider) GetAvailableMachineTypes() ([]string, error) {
diff --git a/cluster-autoscaler/cloudprovider/scaleway/scaleway_cloud_provider.go b/cluster-autoscaler/cloudprovider/scaleway/scaleway_cloud_provider.go
index 9d1608c5712..6bd612db321 100644
--- a/cluster-autoscaler/cloudprovider/scaleway/scaleway_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/scaleway/scaleway_cloud_provider.go
@@ -162,6 +162,11 @@ func (scw *scalewayCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovi
 	return scw.nodeGroupForNode(node)
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (scw *scalewayCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 func (scw *scalewayCloudProvider) NodePrice(node *apiv1.Node, startTime time.Time, endTime time.Time) (float64, error) {
 	ng, err := scw.nodeGroupForNode(node)
 	if err != nil {
diff --git a/cluster-autoscaler/cloudprovider/tencentcloud/tencentcloud_cloud_provider.go b/cluster-autoscaler/cloudprovider/tencentcloud/tencentcloud_cloud_provider.go
index 52768376dc4..4dc47eea3a2 100644
--- a/cluster-autoscaler/cloudprovider/tencentcloud/tencentcloud_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/tencentcloud/tencentcloud_cloud_provider.go
@@ -109,6 +109,11 @@ func (tencentcloud *tencentCloudProvider) NodeGroupForNode(node *apiv1.Node) (cl
 	return asg, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (tencentcloud *tencentCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // GPULabel returns the label added to nodes with GPU resource.
 func (tencentcloud *tencentCloudProvider) GPULabel() string {
 	return GPULabel
diff --git a/cluster-autoscaler/cloudprovider/test/test_cloud_provider.go b/cluster-autoscaler/cloudprovider/test/test_cloud_provider.go
index 234d6853e55..d3e0d68474a 100644
--- a/cluster-autoscaler/cloudprovider/test/test_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/test/test_cloud_provider.go
@@ -41,6 +41,9 @@ type OnNodeGroupCreateFunc func(string) error
 // OnNodeGroupDeleteFunc is a function called when a node group is deleted.
 type OnNodeGroupDeleteFunc func(string) error
 
+// HasInstance is a function called to determine if a node has been removed from the cloud provider.
+type HasInstance func(string) (bool, error)
+
 // TestCloudProvider is a dummy cloud provider to be used in tests.
 type TestCloudProvider struct {
 	sync.Mutex
@@ -50,6 +53,7 @@ type TestCloudProvider struct {
 	onScaleDown       func(string, string) error
 	onNodeGroupCreate func(string) error
 	onNodeGroupDelete func(string) error
+	hasInstance       func(string) (bool, error)
 	machineTypes      []string
 	machineTemplates  map[string]*schedulerframework.NodeInfo
 	priceModel        cloudprovider.PricingModel
@@ -84,6 +88,19 @@ func NewTestAutoprovisioningCloudProvider(onScaleUp OnScaleUpFunc, onScaleDown O
 	}
 }
 
+// NewTestNodeDeletionDetectionCloudProvider builds new TestCloudProvider with deletion detection support
+func NewTestNodeDeletionDetectionCloudProvider(onScaleUp OnScaleUpFunc, onScaleDown OnScaleDownFunc,
+	hasInstance HasInstance) *TestCloudProvider {
+	return &TestCloudProvider{
+		nodes:           make(map[string]string),
+		groups:          make(map[string]cloudprovider.NodeGroup),
+		onScaleUp:       onScaleUp,
+		onScaleDown:     onScaleDown,
+		hasInstance:     hasInstance,
+		resourceLimiter: cloudprovider.NewResourceLimiter(make(map[string]int64), make(map[string]int64)),
+	}
+}
+
 // Name returns name of the cloud provider.
 func (tcp *TestCloudProvider) Name() string {
 	return "TestCloudProvider"
@@ -140,6 +157,19 @@ func (tcp *TestCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovider.
 	return group, nil
 }
 
+// HasInstance returns true if the node has corresponding instance in cloud provider,
+// or ErrNotImplemented to fall back to taint-based node deletion in clusterstate
+// readiness calculation.
+func (tcp *TestCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	tcp.Lock()
+	defer tcp.Unlock()
+	if tcp.hasInstance != nil {
+		return tcp.hasInstance(node.Name)
+	}
+	_, found := tcp.nodes[node.Name]
+	return found, nil
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available.
 func (tcp *TestCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
 	if tcp.priceModel == nil {
@@ -253,6 +283,14 @@ func (tcp *TestCloudProvider) AddNode(nodeGroupId string, node *apiv1.Node) {
 	tcp.nodes[node.Name] = nodeGroupId
 }
 
+// DeleteNode delete the given node from the provider.
+func (tcp *TestCloudProvider) DeleteNode(node *apiv1.Node) {
+	tcp.Lock()
+	defer tcp.Unlock()
+
+	delete(tcp.nodes, node.Name)
+}
+
 // GetResourceLimiter returns struct containing limits (max, min) for resources (cores, memory etc.).
 func (tcp *TestCloudProvider) GetResourceLimiter() (*cloudprovider.ResourceLimiter, error) {
 	return tcp.resourceLimiter, nil
diff --git a/cluster-autoscaler/cloudprovider/vultr/vultr_cloud_provider.go b/cluster-autoscaler/cloudprovider/vultr/vultr_cloud_provider.go
index de8bfaedbd4..b0272788556 100644
--- a/cluster-autoscaler/cloudprovider/vultr/vultr_cloud_provider.go
+++ b/cluster-autoscaler/cloudprovider/vultr/vultr_cloud_provider.go
@@ -81,6 +81,11 @@ func (v *vultrCloudProvider) NodeGroupForNode(node *apiv1.Node) (cloudprovider.N
 	return nil, nil
 }
 
+// HasInstance returns whether a given node has a corresponding instance in this cloud provider
+func (v *vultrCloudProvider) HasInstance(node *apiv1.Node) (bool, error) {
+	return true, cloudprovider.ErrNotImplemented
+}
+
 // Pricing returns pricing model for this cloud provider or error if not available.
 // Implementation optional.
 func (v *vultrCloudProvider) Pricing() (cloudprovider.PricingModel, errors.AutoscalerError) {
diff --git a/cluster-autoscaler/clusterstate/clusterstate.go b/cluster-autoscaler/clusterstate/clusterstate.go
index 3c4a52bdce2..2141188eb03 100644
--- a/cluster-autoscaler/clusterstate/clusterstate.go
+++ b/cluster-autoscaler/clusterstate/clusterstate.go
@@ -17,6 +17,7 @@ limitations under the License.
 package clusterstate
 
 import (
+	"errors"
 	"fmt"
 	"reflect"
 	"strings"
@@ -121,6 +122,7 @@ type ClusterStateRegistry struct {
 	acceptableRanges                   map[string]AcceptableRange
 	incorrectNodeGroupSizes            map[string]IncorrectNodeGroupSize
 	unregisteredNodes                  map[string]UnregisteredNode
+	deletedNodes                       map[string]struct{}
 	candidatesForScaleDown             map[string][]string
 	backoff                            backoff.Backoff
 	lastStatus                         *api.ClusterAutoscalerStatus
@@ -153,6 +155,7 @@ func NewClusterStateRegistry(cloudProvider cloudprovider.CloudProvider, config C
 		acceptableRanges:                make(map[string]AcceptableRange),
 		incorrectNodeGroupSizes:         make(map[string]IncorrectNodeGroupSize),
 		unregisteredNodes:               make(map[string]UnregisteredNode),
+		deletedNodes:                    make(map[string]struct{}),
 		candidatesForScaleDown:          make(map[string][]string),
 		backoff:                         backoff,
 		lastStatus:                      emptyStatus,
@@ -295,6 +298,7 @@ func (csr *ClusterStateRegistry) UpdateNodes(nodes []*apiv1.Node, nodeInfosForGr
 	if err != nil {
 		return err
 	}
+	cloudProviderNodesRemoved := csr.getCloudProviderDeletedNodes(nodes)
 	notRegistered := getNotRegisteredNodes(nodes, cloudProviderNodeInstances, currentTime)
 
 	csr.Lock()
@@ -306,6 +310,7 @@ func (csr *ClusterStateRegistry) UpdateNodes(nodes []*apiv1.Node, nodeInfosForGr
 	csr.cloudProviderNodeInstances = cloudProviderNodeInstances
 
 	csr.updateUnregisteredNodes(notRegistered)
+	csr.updateCloudProviderDeletedNodes(cloudProviderNodesRemoved)
 	csr.updateReadinessStats(currentTime)
 
 	// update acceptable ranges based on requests from last loop and targetSizes
@@ -541,7 +546,7 @@ func (csr *ClusterStateRegistry) updateReadinessStats(currentTime time.Time) {
 
 	update := func(current Readiness, node *apiv1.Node, nr kube_util.NodeReadiness) Readiness {
 		current.Registered++
-		if deletetaint.HasToBeDeletedTaint(node) {
+		if _, isDeleted := csr.deletedNodes[node.Name]; isDeleted {
 			current.Deleted++
 		} else if nr.Ready {
 			current.Ready++
@@ -669,6 +674,14 @@ func (csr *ClusterStateRegistry) GetUnregisteredNodes() []UnregisteredNode {
 	return result
 }
 
+func (csr *ClusterStateRegistry) updateCloudProviderDeletedNodes(deletedNodes []*apiv1.Node) {
+	result := make(map[string]struct{}, len(deletedNodes))
+	for _, deleted := range deletedNodes {
+		result[deleted.Name] = struct{}{}
+	}
+	csr.deletedNodes = result
+}
+
 // UpdateScaleDownCandidates updates scale down candidates
 func (csr *ClusterStateRegistry) UpdateScaleDownCandidates(nodes []*apiv1.Node, now time.Time) {
 	result := make(map[string][]string)
@@ -958,6 +971,28 @@ func getNotRegisteredNodes(allNodes []*apiv1.Node, cloudProviderNodeInstances ma
 	return notRegistered
 }
 
+// Calculates which of the registered nodes in Kubernetes that do not exist in cloud provider.
+func (csr *ClusterStateRegistry) getCloudProviderDeletedNodes(allNodes []*apiv1.Node) []*apiv1.Node {
+	nodesRemoved := make([]*apiv1.Node, 0)
+	for _, node := range allNodes {
+		if !csr.hasCloudProviderInstance(node) {
+			nodesRemoved = append(nodesRemoved, node)
+		}
+	}
+	return nodesRemoved
+}
+
+func (csr *ClusterStateRegistry) hasCloudProviderInstance(node *apiv1.Node) bool {
+	exists, err := csr.cloudProvider.HasInstance(node)
+	if err == nil {
+		return exists
+	}
+	if !errors.Is(err, cloudprovider.ErrNotImplemented) {
+		klog.Warningf("Failed to check cloud provider has instance for %s: %v", node.Name, err)
+	}
+	return !deletetaint.HasToBeDeletedTaint(node)
+}
+
 // GetAutoscaledNodesCount calculates and returns the actual and the target number of nodes
 // belonging to autoscaled node groups in the cluster.
 func (csr *ClusterStateRegistry) GetAutoscaledNodesCount() (currentSize, targetSize int) {
diff --git a/cluster-autoscaler/clusterstate/clusterstate_test.go b/cluster-autoscaler/clusterstate/clusterstate_test.go
index 65d3bc43c5f..8587c32b6dc 100644
--- a/cluster-autoscaler/clusterstate/clusterstate_test.go
+++ b/cluster-autoscaler/clusterstate/clusterstate_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package clusterstate
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
@@ -24,9 +25,11 @@ import (
 
 	apiv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 	testprovider "k8s.io/autoscaler/cluster-autoscaler/cloudprovider/test"
 	"k8s.io/autoscaler/cluster-autoscaler/clusterstate/api"
 	"k8s.io/autoscaler/cluster-autoscaler/clusterstate/utils"
+	"k8s.io/autoscaler/cluster-autoscaler/utils/deletetaint"
 	. "k8s.io/autoscaler/cluster-autoscaler/utils/test"
 	"k8s.io/client-go/kubernetes/fake"
 	kube_record "k8s.io/client-go/tools/record"
@@ -35,6 +38,19 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/utils/backoff"
 )
 
+// GetCloudProviderDeletedNodeNames returns a list of the names of nodes removed
+// from cloud provider but registered in Kubernetes.
+func GetCloudProviderDeletedNodeNames(csr *ClusterStateRegistry) []string {
+	csr.Lock()
+	defer csr.Unlock()
+
+	result := make([]string, 0, len(csr.deletedNodes))
+	for nodeName := range csr.deletedNodes {
+		result = append(result, nodeName)
+	}
+	return result
+}
+
 func TestOKWithScaleUp(t *testing.T) {
 	now := time.Now()
 
@@ -481,6 +497,22 @@ func TestUpcomingNodes(t *testing.T) {
 	provider.AddNodeGroup("ng4", 1, 10, 1)
 	provider.AddNode("ng4", ng4_1)
 
+	// One node is already there, for a second nde deletion / draining was already started.
+	ng5_1 := BuildTestNode("ng5-1", 1000, 1000)
+	SetNodeReadyState(ng5_1, true, now.Add(-time.Minute))
+	ng5_2 := BuildTestNode("ng5-2", 1000, 1000)
+	SetNodeReadyState(ng5_2, true, now.Add(-time.Minute))
+	ng5_2.Spec.Taints = []apiv1.Taint{
+		{
+			Key:    deletetaint.ToBeDeletedTaint,
+			Value:  fmt.Sprint(time.Now().Unix()),
+			Effect: apiv1.TaintEffectNoSchedule,
+		},
+	}
+	provider.AddNodeGroup("ng5", 1, 10, 2)
+	provider.AddNode("ng5", ng5_1)
+	provider.AddNode("ng5", ng5_2)
+
 	assert.NotNil(t, provider)
 	fakeClient := &fake.Clientset{}
 	fakeLogRecorder, _ := utils.NewStatusMapRecorder(fakeClient, "kube-system", kube_record.NewFakeRecorder(5), false, "my-cool-configmap")
@@ -488,7 +520,7 @@ func TestUpcomingNodes(t *testing.T) {
 		MaxTotalUnreadyPercentage: 10,
 		OkTotalUnreadyCount:       1,
 	}, fakeLogRecorder, newBackoff())
-	err := clusterstate.UpdateNodes([]*apiv1.Node{ng1_1, ng2_1, ng3_1, ng4_1}, nil, now)
+	err := clusterstate.UpdateNodes([]*apiv1.Node{ng1_1, ng2_1, ng3_1, ng4_1, ng5_1, ng5_2}, nil, now)
 	assert.NoError(t, err)
 	assert.Empty(t, clusterstate.GetScaleUpFailures())
 
@@ -497,6 +529,45 @@ func TestUpcomingNodes(t *testing.T) {
 	assert.Equal(t, 1, upcomingNodes["ng2"])
 	assert.Equal(t, 2, upcomingNodes["ng3"])
 	assert.NotContains(t, upcomingNodes, "ng4")
+	assert.Equal(t, 0, upcomingNodes["ng5"])
+}
+
+func TestTaintBasedNodeDeletion(t *testing.T) {
+	// Create a new Cloud Provider that does not implement the HasInstance check
+	// it will return the ErrNotImplemented error instead.
+	provider := testprovider.NewTestNodeDeletionDetectionCloudProvider(nil, nil,
+		func(string) (bool, error) { return false, cloudprovider.ErrNotImplemented })
+	now := time.Now()
+
+	// One node is already there, for a second nde deletion / draining was already started.
+	ng1_1 := BuildTestNode("ng1-1", 1000, 1000)
+	SetNodeReadyState(ng1_1, true, now.Add(-time.Minute))
+	ng1_2 := BuildTestNode("ng1-2", 1000, 1000)
+	SetNodeReadyState(ng1_2, true, now.Add(-time.Minute))
+	ng1_2.Spec.Taints = []apiv1.Taint{
+		{
+			Key:    deletetaint.ToBeDeletedTaint,
+			Value:  fmt.Sprint(time.Now().Unix()),
+			Effect: apiv1.TaintEffectNoSchedule,
+		},
+	}
+	provider.AddNodeGroup("ng1", 1, 10, 2)
+	provider.AddNode("ng1", ng1_1)
+	provider.AddNode("ng1", ng1_2)
+
+	assert.NotNil(t, provider)
+	fakeClient := &fake.Clientset{}
+	fakeLogRecorder, _ := utils.NewStatusMapRecorder(fakeClient, "kube-system", kube_record.NewFakeRecorder(5), false, "my-cool-configmap")
+	clusterstate := NewClusterStateRegistry(provider, ClusterStateRegistryConfig{
+		MaxTotalUnreadyPercentage: 10,
+		OkTotalUnreadyCount:       1,
+	}, fakeLogRecorder, newBackoff())
+	err := clusterstate.UpdateNodes([]*apiv1.Node{ng1_1, ng1_2}, nil, now)
+	assert.NoError(t, err)
+	assert.Empty(t, clusterstate.GetScaleUpFailures())
+
+	upcomingNodes := clusterstate.GetUpcomingNodes()
+	assert.Equal(t, 1, upcomingNodes["ng1"])
 }
 
 func TestIncorrectSize(t *testing.T) {
@@ -570,6 +641,104 @@ func TestUnregisteredNodes(t *testing.T) {
 	assert.Equal(t, 0, len(clusterstate.GetUnregisteredNodes()))
 }
 
+func TestCloudProviderDeletedNodes(t *testing.T) {
+	now := time.Now()
+	ng1_1 := BuildTestNode("ng1-1", 1000, 1000)
+	SetNodeReadyState(ng1_1, true, now.Add(-time.Minute))
+	ng1_1.Spec.ProviderID = "ng1-1"
+	ng1_2 := BuildTestNode("ng1-2", 1000, 1000)
+	SetNodeReadyState(ng1_2, true, now.Add(-time.Minute))
+	ng1_2.Spec.ProviderID = "ng1-2"
+	// No Node Group - Not Autoscaled Node
+	noNgNode := BuildTestNode("no-ng", 1000, 1000)
+	SetNodeReadyState(noNgNode, true, now.Add(-time.Minute))
+
+	noNgNode.Spec.ProviderID = "no-ng"
+	provider := testprovider.NewTestCloudProvider(nil, nil)
+	provider.AddNodeGroup("ng1", 1, 10, 2)
+	provider.AddNode("ng1", ng1_1)
+	provider.AddNode("ng1", ng1_2)
+	provider.AddNode("no_ng", noNgNode)
+
+	fakeClient := &fake.Clientset{}
+	fakeLogRecorder, _ := utils.NewStatusMapRecorder(fakeClient, "kube-system", kube_record.NewFakeRecorder(5), false, "my-cool-configmap")
+	clusterstate := NewClusterStateRegistry(provider, ClusterStateRegistryConfig{
+		MaxTotalUnreadyPercentage: 10,
+		OkTotalUnreadyCount:       1,
+		MaxNodeProvisionTime:      10 * time.Second,
+	}, fakeLogRecorder, newBackoff())
+	now.Add(time.Minute)
+	err := clusterstate.UpdateNodes([]*apiv1.Node{ng1_1, ng1_2, noNgNode}, nil, now)
+
+	// Nodes are registered correctly between Kubernetes and cloud provider.
+	assert.NoError(t, err)
+	assert.Equal(t, 0, len(GetCloudProviderDeletedNodeNames(clusterstate)))
+
+	// The node was removed from Cloud Provider
+	// should be counted as Deleted by cluster state
+	nodeGroup, err := provider.NodeGroupForNode(ng1_2)
+	assert.NoError(t, err)
+	provider.DeleteNode(ng1_2)
+	clusterstate.InvalidateNodeInstancesCacheEntry(nodeGroup)
+	now.Add(time.Minute)
+
+	err = clusterstate.UpdateNodes([]*apiv1.Node{ng1_1, ng1_2, noNgNode}, nil, now)
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(GetCloudProviderDeletedNodeNames(clusterstate)))
+	assert.Equal(t, "ng1-2", GetCloudProviderDeletedNodeNames(clusterstate)[0])
+	assert.Equal(t, 1, clusterstate.GetClusterReadiness().Deleted)
+
+	// The node is removed from Kubernetes
+	now.Add(time.Minute)
+
+	err = clusterstate.UpdateNodes([]*apiv1.Node{ng1_1, noNgNode}, nil, now)
+	assert.NoError(t, err)
+	assert.Equal(t, 0, len(GetCloudProviderDeletedNodeNames(clusterstate)))
+
+	// New Node is added afterwards
+	ng1_3 := BuildTestNode("ng1-3", 1000, 1000)
+	SetNodeReadyState(ng1_3, true, now.Add(-time.Minute))
+	ng1_3.Spec.ProviderID = "ng1-3"
+	provider.AddNode("ng1", ng1_3)
+	clusterstate.InvalidateNodeInstancesCacheEntry(nodeGroup)
+	now.Add(time.Minute)
+
+	err = clusterstate.UpdateNodes([]*apiv1.Node{ng1_1, ng1_3, noNgNode}, nil, now)
+	assert.NoError(t, err)
+	assert.Equal(t, 0, len(GetCloudProviderDeletedNodeNames(clusterstate)))
+
+	// Newly added node is removed from Cloud Provider
+	// should be counted as Deleted by cluster state
+	nodeGroup, err = provider.NodeGroupForNode(ng1_3)
+	assert.NoError(t, err)
+	provider.DeleteNode(ng1_3)
+	clusterstate.InvalidateNodeInstancesCacheEntry(nodeGroup)
+	now.Add(time.Minute)
+
+	err = clusterstate.UpdateNodes([]*apiv1.Node{ng1_1, noNgNode, ng1_3}, nil, now)
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(GetCloudProviderDeletedNodeNames(clusterstate)))
+	assert.Equal(t, "ng1-3", GetCloudProviderDeletedNodeNames(clusterstate)[0])
+	assert.Equal(t, 1, clusterstate.GetClusterReadiness().Deleted)
+
+	// Confirm that previously identified deleted Cloud Provider nodes are still included
+	// until it is removed from Kubernetes
+	now.Add(time.Minute)
+
+	err = clusterstate.UpdateNodes([]*apiv1.Node{ng1_1, noNgNode, ng1_3}, nil, now)
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(GetCloudProviderDeletedNodeNames(clusterstate)))
+	assert.Equal(t, "ng1-3", GetCloudProviderDeletedNodeNames(clusterstate)[0])
+	assert.Equal(t, 1, clusterstate.GetClusterReadiness().Deleted)
+
+	// The node is removed from Kubernetes
+	now.Add(time.Minute)
+
+	err = clusterstate.UpdateNodes([]*apiv1.Node{ng1_1, noNgNode}, nil, now)
+	assert.NoError(t, err)
+	assert.Equal(t, 0, len(GetCloudProviderDeletedNodeNames(clusterstate)))
+}
+
 func TestUpdateLastTransitionTimes(t *testing.T) {
 	now := metav1.Time{Time: time.Now()}
 	later := metav1.Time{Time: now.Time.Add(10 * time.Second)}
diff --git a/cluster-autoscaler/core/static_autoscaler_test.go b/cluster-autoscaler/core/static_autoscaler_test.go
index 3bb56b4c0c5..9b0c3a55ce1 100644
--- a/cluster-autoscaler/core/static_autoscaler_test.go
+++ b/cluster-autoscaler/core/static_autoscaler_test.go
@@ -1122,6 +1122,10 @@ func TestStaticAutoscalerInstanceCreationErrors(t *testing.T) {
 			}
 			return nil
 		}, nil)
+	provider.On("HasInstance", mock.Anything).Return(
+		func(node *apiv1.Node) bool {
+			return false
+		}, nil)
 
 	now := time.Now()
 
@@ -1250,6 +1254,10 @@ func TestStaticAutoscalerInstanceCreationErrors(t *testing.T) {
 	provider = &mockprovider.CloudProvider{}
 	provider.On("NodeGroups").Return([]cloudprovider.NodeGroup{nodeGroupC})
 	provider.On("NodeGroupForNode", mock.Anything).Return(nil, nil)
+	provider.On("HasInstance", mock.Anything).Return(
+		func(node *apiv1.Node) bool {
+			return false
+		}, nil)
 
 	clusterState = clusterstate.NewClusterStateRegistry(provider, clusterStateConfig, context.LogRecorder, NewBackoff())
 	clusterState.RefreshCloudProviderNodeInstancesCache()