diff --git a/api/v1alpha4/awscluster_webhook.go b/api/v1alpha4/awscluster_webhook.go
index e9b8a1ab8c..a6b6bc5603 100644
--- a/api/v1alpha4/awscluster_webhook.go
+++ b/api/v1alpha4/awscluster_webhook.go
@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1alpha4"
+	"sigs.k8s.io/cluster-api/util/annotations"
 	ctrl "sigs.k8s.io/controller-runtime"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
@@ -104,6 +105,13 @@ func (r *AWSCluster) ValidateUpdate(old runtime.Object) error {
 		)
 	}
 
+	if annotations.IsExternallyManaged(oldC) && !annotations.IsExternallyManaged(r) {
+		allErrs = append(allErrs,
+			field.Invalid(field.NewPath("metadata", "annotations"),
+				r.Annotations, "removal of externally managed annotation is not allowed"),
+		)
+	}
+
 	allErrs = append(allErrs, r.Spec.Bastion.Validate()...)
 
 	return aggregateObjErrors(r.GroupVersionKind().GroupKind(), r.Name, allErrs)
diff --git a/api/v1alpha4/awscluster_webhook_test.go b/api/v1alpha4/awscluster_webhook_test.go
index a899a8e266..da0802d26d 100644
--- a/api/v1alpha4/awscluster_webhook_test.go
+++ b/api/v1alpha4/awscluster_webhook_test.go
@@ -152,18 +152,38 @@ func TestAWSCluster_ValidateUpdate(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "removal of externally managed annotation is not allowed",
+			oldCluster: &AWSCluster{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{clusterv1.ManagedByAnnotation: ""},
+				},
+			},
+			newCluster: &AWSCluster{},
+			wantErr:    true,
+		},
+		{
+			name:       "adding externally managed annotation is allowed",
+			oldCluster: &AWSCluster{},
+			newCluster: &AWSCluster{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{clusterv1.ManagedByAnnotation: ""},
+				},
+			},
+			wantErr: false,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			ctx := context.TODO()
 			cluster := tt.oldCluster.DeepCopy()
-			cluster.ObjectMeta = metav1.ObjectMeta{
-				GenerateName: "cluster-",
-				Namespace:    "default",
-			}
+			cluster.ObjectMeta.GenerateName = "cluster-"
+			cluster.ObjectMeta.Namespace = "default"
+
 			if err := testEnv.Create(ctx, cluster); err != nil {
 				t.Errorf("failed to create cluster: %v", err)
 			}
+			cluster.ObjectMeta.Annotations = tt.newCluster.Annotations
 			cluster.Spec = tt.newCluster.Spec
 			if err := testEnv.Update(ctx, cluster); (err != nil) != tt.wantErr {
 				t.Errorf("ValidateUpdate() error = %v, wantErr %v", err, tt.wantErr)
diff --git a/api/v1alpha4/suite_test.go b/api/v1alpha4/suite_test.go
index 0980f4ba59..39f5cfd3cb 100644
--- a/api/v1alpha4/suite_test.go
+++ b/api/v1alpha4/suite_test.go
@@ -27,16 +27,16 @@ import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
-	ctrl "sigs.k8s.io/controller-runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/cluster-api-provider-aws/test/helpers"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/envtest/printer"
 )
 
 var (
 	testEnv *helpers.TestEnvironment
-	ctx = ctrl.SetupSignalHandler()
+	ctx     = ctrl.SetupSignalHandler()
 )
 
 func TestAPIs(t *testing.T) {
diff --git a/controllers/awscluster_controller.go b/controllers/awscluster_controller.go
index 7f6a337fff..33c2b520ca 100644
--- a/controllers/awscluster_controller.go
+++ b/controllers/awscluster_controller.go
@@ -26,6 +26,7 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 	infrav1 "sigs.k8s.io/cluster-api-provider-aws/api/v1alpha4"
 	"sigs.k8s.io/cluster-api-provider-aws/feature"
@@ -304,6 +305,7 @@ func (r *AWSClusterReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Ma
 				},
 			},
 		).
+		WithEventFilter(predicates.ResourceIsNotExternallyManaged(log)).
 		Build(r)
 	if err != nil {
 		return errors.Wrap(err, "error creating controller")
@@ -311,7 +313,7 @@ func (r *AWSClusterReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Ma
 
 	return controller.Watch(
 		&source.Kind{Type: &clusterv1.Cluster{}},
-		handler.EnqueueRequestsFromMapFunc(r.requeueAWSClusterForUnpausedCluster(log)),
+		handler.EnqueueRequestsFromMapFunc(r.requeueAWSClusterForUnpausedCluster(ctx, log)),
 		predicate.Funcs{
 			UpdateFunc: func(e event.UpdateEvent) bool {
 				oldCluster := e.ObjectOld.(*clusterv1.Cluster)
@@ -354,7 +356,7 @@ func (r *AWSClusterReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Ma
 	)
 }
 
-func (r *AWSClusterReconciler) requeueAWSClusterForUnpausedCluster(log logr.Logger) handler.MapFunc {
+func (r *AWSClusterReconciler) requeueAWSClusterForUnpausedCluster(ctx context.Context, log logr.Logger) handler.MapFunc {
 	return func(o client.Object) []ctrl.Request {
 		c, ok := o.(*clusterv1.Cluster)
 		if !ok {
@@ -380,6 +382,19 @@ func (r *AWSClusterReconciler) requeueAWSClusterForUnpausedCluster(log logr.Logg
 			return nil
 		}
 
+		awsCluster := &infrav1.AWSCluster{}
+		key := types.NamespacedName{Namespace: c.Spec.InfrastructureRef.Namespace, Name: c.Spec.InfrastructureRef.Name}
+
+		if err := r.Get(ctx, key, awsCluster); err != nil {
+			log.V(4).Error(err, "Failed to get AWS cluster")
+			return nil
+		}
+
+		if annotations.IsExternallyManaged(awsCluster) {
+			log.V(4).Info("AWSCluster is externally managed, skipping mapping.")
+			return nil
+		}
+
 		log.V(4).Info("Adding request.", "awsCluster", c.Spec.InfrastructureRef.Name)
 		return []ctrl.Request{
 			{
diff --git a/pkg/cloud/scope/machine.go b/pkg/cloud/scope/machine.go
index 0f05257048..f6f3653336 100644
--- a/pkg/cloud/scope/machine.go
+++ b/pkg/cloud/scope/machine.go
@@ -32,6 +32,7 @@ import (
 	"sigs.k8s.io/cluster-api/controllers/noderefutil"
 	capierrors "sigs.k8s.io/cluster-api/errors"
 	"sigs.k8s.io/cluster-api/util"
+	"sigs.k8s.io/cluster-api/util/annotations"
 	"sigs.k8s.io/cluster-api/util/conditions"
 	"sigs.k8s.io/cluster-api/util/patch"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -343,6 +344,10 @@ func (m *MachineScope) IsEKSManaged() bool {
 	return m.InfraCluster.InfraCluster().GetObjectKind().GroupVersionKind().Kind == "AWSManagedControlPlane"
 }
 
+func (m *MachineScope) IsExternallyManaged() bool {
+	return annotations.IsExternallyManaged(m.InfraCluster.InfraCluster())
+}
+
 // SetInterruptible sets the AWSMachine status Interruptible
 func (m *MachineScope) SetInterruptible() {
 	if m.AWSMachine.Spec.SpotMarketOptions != nil {
diff --git a/pkg/cloud/services/ec2/instances.go b/pkg/cloud/services/ec2/instances.go
index 040a2b4b1a..4d8e8aaac3 100644
--- a/pkg/cloud/services/ec2/instances.go
+++ b/pkg/cloud/services/ec2/instances.go
@@ -173,7 +173,7 @@ func (s *Service) CreateInstance(scope *scope.MachineScope, userData []byte) (*i
 	}
 	input.SubnetID = subnetID
 
-	if !scope.IsEKSManaged() && s.scope.Network().APIServerELB.DNSName == "" {
+	if !scope.IsExternallyManaged() && !scope.IsEKSManaged() && s.scope.Network().APIServerELB.DNSName == "" {
 		record.Eventf(s.scope.InfraCluster(), "FailedCreateInstance", "Failed to run controlplane, APIServer ELB not available")
 
 		return nil, awserrors.NewFailedDependency("failed to run controlplane, APIServer ELB not available")
@@ -210,7 +210,9 @@ func (s *Service) CreateInstance(scope *scope.MachineScope, userData []byte) (*i
 		// fallback to AWSCluster.Spec.SSHKeyName if it is defined
 		prioritizedSSHKeyName = *scope.InfraCluster.SSHKeyName()
 	default:
-		prioritizedSSHKeyName = defaultSSHKeyName
+		if !scope.IsExternallyManaged() {
+			prioritizedSSHKeyName = defaultSSHKeyName
+		}
 	}
 
 	// Only set input.SSHKeyName if the user did not explicitly request no ssh key be set (explicitly setting "" on either the Machine or related Cluster)
@@ -291,7 +293,9 @@ func (s *Service) findSubnet(scope *scope.MachineScope) (string, error) {
 	case scope.AWSMachine.Spec.Subnet != nil && scope.AWSMachine.Spec.Subnet.Filters != nil:
 		criteria := []*ec2.Filter{
 			filter.EC2.SubnetStates(ec2.SubnetStatePending, ec2.SubnetStateAvailable),
-			filter.EC2.VPC(s.scope.VPC().ID),
+		}
+		if !scope.IsExternallyManaged() {
+			criteria = append(criteria, filter.EC2.VPC(s.scope.VPC().ID))
 		}
 		if failureDomain != nil {
 			criteria = append(criteria, filter.EC2.AvailabilityZone(*failureDomain))
@@ -355,6 +359,10 @@ func (s *Service) getFilteredSubnets(criteria ...*ec2.Filter) ([]*ec2.Subnet, er
 // GetCoreSecurityGroups looks up the security group IDs managed by this actuator
 // They are considered "core" to its proper functioning
 func (s *Service) GetCoreSecurityGroups(scope *scope.MachineScope) ([]string, error) {
+	if scope.IsExternallyManaged() {
+		return nil, nil
+	}
+
 	// These are common across both controlplane and node machines
 	sgRoles := []infrav1.SecurityGroupRole{
 		infrav1.SecurityGroupNode,