Repository: crossplane/provider-aws (go)

Issue:
It is not possible to import securitygroup which is created previously by composition with same claim. 
This problem is best described with an example: 

1. Create a composition with a securitygroup with deletionPolicy: Orphan
2. Create a claim for that composition and deploy. 
3. Remove the claim
4. Now the securitygroup stays behind in AWS which is as expected. 
5. Create the same claim again. 
6. Now the security will have a reconcile error and cannot create a duplicate groupname. 

We expect that it can reconcile the existing securitygroup.



We used the following composition for the securitygroup and we set the following annotation
  annotations:
    crossplane.io/external-name: sg-0002 **# we don't know the id that is created by the composition so we try to use our own 
this part does not work, aws creates its own ID.** 

Import resources is described here: https://github.com/crossplane/crossplane/issues/1093

```
    - base:
        apiVersion: ec2.aws.crossplane.io/v1beta1
        kind: SecurityGroup
        spec:
          deletionPolicy: Orphan
          forProvider:
            groupName: crossplane-test-sg
            ingress:
              - fromPort: 5432
                ipProtocol: tcp
                ipRanges:
                  - cidrIp: 0.0.0.0/0
                    description: Everywhere
                toPort: 5432
            region: eu-central-1
            vpcId: vpc-xxxxxxxx
          providerConfigRef:
            name: aws-provider
      name: securitygroup-whitelist
      patches:
        - fromFieldPath: 'metadata.annotations[crossplane.io/external-name]'
          toFieldPath: 'metadata.annotations[crossplane.io/external-name]'
```

We want to import this created securitygroup but this is not possible this way.


Pull request description:
ec2.SecurityGroup: Observe existing SecurityGroup
This allows transferring ownership of an existing SecurityGroup.

Signed-off-by: Otto Nordander <otto.nordander@resurs.se>

<!--
Thank you for helping to improve Crossplane!

Please read through https://git.io/fj2m9 if this is your first time opening a
Crossplane pull request. Find us in https://slack.crossplane.io/messages/dev if
you need any help contributing.
-->

### Description of your changes
Check if a SecurityGroup exists during `controller.Observe` to allow transferring ownership of SGs.
<!--
Briefly describe what this pull request does. Be sure to direct your reviewers'
attention to anything that needs special consideration.

We love pull requests that resolve an open Crossplane issue. If yours does, you
can uncomment the below line to indicate which issue your PR fixes, for example
"Fixes #500":

-->
Fixes #1175 

I have:

- [X] Read and followed Crossplane's [contribution process].
- [X] Run `make reviewable test` to ensure this PR is ready for review.

### How has this code been tested

<!--
Before reviewers can be confident in the correctness of this pull request, it
needs to tested and shown to be correct. Briefly describe the testing that has
already been done or which is planned for this change.
-->
Tested with local Kind cluster.
```yaml
apiVersion: ec2.aws.crossplane.io/v1beta1
kind: SecurityGroup
metadata:
  name: test-security-group
spec:
  deletionPolicy: Orphan
  providerConfigRef:
    name: local
  forProvider:
    region: eu-central-1
    groupName: test-security-group
    description: Test security group
    vpcId: <vpc-id>
```
```bash
kubectl apply -f securitygroup.yaml
kubectl get securitygroup test-security-group # ID: sg-<ID>
kubectl delete -f securitygroup.yaml
kubectl get securitygroup # No resources found
kubectl apply -f securitygroup.yaml
kubectl get securitygroup test-security-group # ID: sg-<ID> (same ID as previously)


Patch:
diff --git a/pkg/controller/ec2/securitygroup/controller.go b/pkg/controller/ec2/securitygroup/controller.go
index 49bee9436..157b4d7fe 100644
--- a/pkg/controller/ec2/securitygroup/controller.go
+++ b/pkg/controller/ec2/securitygroup/controller.go
@@ -49,6 +49,7 @@ const (
 	errUnexpectedObject = "The managed resource is not an SecurityGroup resource"
 
 	errDescribe         = "failed to describe SecurityGroup"
+	errGetSecurityGroup = "failed to get SecurityGroup based on groupName"
 	errMultipleItems    = "retrieved multiple SecurityGroups for the given securityGroupId"
 	errCreate           = "failed to create the SecurityGroup resource"
 	errAuthorizeIngress = "failed to authorize ingress rules"
@@ -117,7 +118,13 @@ func (e *external) Observe(ctx context.Context, mgd resource.Managed) (managed.E
 	}
 
 	if meta.GetExternalName(cr) == "" {
-		return managed.ExternalObservation{}, nil
+		securityGroupArn, err := e.getSecurityGroupByName(ctx, cr.Spec.ForProvider.GroupName)
+		if securityGroupArn == nil || err != nil {
+			return managed.ExternalObservation{}, awsclient.Wrap(resource.Ignore(ec2.IsSecurityGroupNotFoundErr, err), errGetSecurityGroup)
+		}
+
+		meta.SetExternalName(cr, aws.ToString(securityGroupArn))
+		_ = e.kube.Update(ctx, cr)
 	}
 
 	response, err := e.sg.DescribeSecurityGroups(ctx, &awsec2.DescribeSecurityGroupsInput{
@@ -292,3 +299,17 @@ func (e *external) Delete(ctx context.Context, mgd resource.Managed) error {
 
 	return awsclient.Wrap(resource.Ignore(ec2.IsSecurityGroupNotFoundErr, err), errDelete)
 }
+
+func (e *external) getSecurityGroupByName(ctx context.Context, groupName string) (*string, error) {
+	groups, err := e.sg.DescribeSecurityGroups(ctx, &awsec2.DescribeSecurityGroupsInput{
+		Filters: []awsec2types.Filter{
+			{Name: aws.String("group-name"), Values: []string{groupName}},
+		},
+	})
+
+	if err != nil || len(groups.SecurityGroups) == 0 {
+		return nil, err
+	}
+
+	return groups.SecurityGroups[0].GroupId, nil
+}
diff --git a/pkg/controller/rds/dbinstance/setup.go b/pkg/controller/rds/dbinstance/setup.go
index 2541ea2d8..2f963540d 100644
--- a/pkg/controller/rds/dbinstance/setup.go
+++ b/pkg/controller/rds/dbinstance/setup.go
@@ -206,17 +206,18 @@ func lateInitialize(in *svcapitypes.DBInstanceParameters, out *svcsdk.DescribeDB
 	in.DBClusterIdentifier = aws.LateInitializeStringPtr(in.DBClusterIdentifier, db.DBClusterIdentifier)
 	if in.DBClusterIdentifier == nil {
 		in.AllocatedStorage = aws.LateInitializeInt64Ptr(in.AllocatedStorage, db.AllocatedStorage)
+		in.BackupRetentionPeriod = aws.LateInitializeInt64Ptr(in.BackupRetentionPeriod, db.BackupRetentionPeriod)
+		in.CopyTagsToSnapshot = aws.LateInitializeBoolPtr(in.CopyTagsToSnapshot, db.CopyTagsToSnapshot)
+		in.EnableIAMDatabaseAuthentication = aws.LateInitializeBoolPtr(in.EnableIAMDatabaseAuthentication, db.IAMDatabaseAuthenticationEnabled)
+		in.PreferredBackupWindow = aws.LateInitializeStringPtr(in.PreferredBackupWindow, db.PreferredBackupWindow)
 		in.StorageEncrypted = aws.LateInitializeBoolPtr(in.StorageEncrypted, db.StorageEncrypted)
 		in.StorageType = aws.LateInitializeStringPtr(in.StorageType, db.StorageType)
 	}
 	in.AutoMinorVersionUpgrade = aws.LateInitializeBoolPtr(in.AutoMinorVersionUpgrade, db.AutoMinorVersionUpgrade)
 	in.AvailabilityZone = aws.LateInitializeStringPtr(in.AvailabilityZone, db.AvailabilityZone)
-	in.BackupRetentionPeriod = aws.LateInitializeInt64Ptr(in.BackupRetentionPeriod, db.BackupRetentionPeriod)
 	in.CharacterSetName = aws.LateInitializeStringPtr(in.CharacterSetName, db.CharacterSetName)
-	in.CopyTagsToSnapshot = aws.LateInitializeBoolPtr(in.CopyTagsToSnapshot, db.CopyTagsToSnapshot)
 	in.DBName = aws.LateInitializeStringPtr(in.DBName, db.DBName)
 	in.DeletionProtection = aws.LateInitializeBoolPtr(in.DeletionProtection, db.DeletionProtection)
-	in.EnableIAMDatabaseAuthentication = aws.LateInitializeBoolPtr(in.EnableIAMDatabaseAuthentication, db.IAMDatabaseAuthenticationEnabled)
 	in.EnablePerformanceInsights = aws.LateInitializeBoolPtr(in.EnablePerformanceInsights, db.PerformanceInsightsEnabled)
 	in.IOPS = aws.LateInitializeInt64Ptr(in.IOPS, db.Iops)
 	in.KMSKeyID = aws.LateInitializeStringPtr(in.KMSKeyID, db.KmsKeyId)
@@ -231,7 +232,6 @@ func lateInitialize(in *svcapitypes.DBInstanceParameters, out *svcsdk.DescribeDB
 	in.MultiAZ = aws.LateInitializeBoolPtr(in.MultiAZ, db.MultiAZ)
 	in.PerformanceInsightsKMSKeyID = aws.LateInitializeStringPtr(in.PerformanceInsightsKMSKeyID, db.PerformanceInsightsKMSKeyId)
 	in.PerformanceInsightsRetentionPeriod = aws.LateInitializeInt64Ptr(in.PerformanceInsightsRetentionPeriod, db.PerformanceInsightsRetentionPeriod)
-	in.PreferredBackupWindow = aws.LateInitializeStringPtr(in.PreferredBackupWindow, db.PreferredBackupWindow)
 	in.PreferredMaintenanceWindow = aws.LateInitializeStringPtr(in.PreferredMaintenanceWindow, db.PreferredMaintenanceWindow)
 	in.PromotionTier = aws.LateInitializeInt64Ptr(in.PromotionTier, db.PromotionTier)
 	in.PubliclyAccessible = aws.LateInitializeBoolPtr(in.PubliclyAccessible, db.PubliclyAccessible)


Test patch:
diff --git a/pkg/controller/ec2/securitygroup/controller_test.go b/pkg/controller/ec2/securitygroup/controller_test.go
index 03347f728..dd5b813e0 100644
--- a/pkg/controller/ec2/securitygroup/controller_test.go
+++ b/pkg/controller/ec2/securitygroup/controller_test.go
@@ -150,6 +150,73 @@ func TestObserve(t *testing.T) {
 				},
 			},
 		},
+		"EmptyExternalNameExistingSG": {
+			args: args{
+				kube: &test.MockClient{
+					MockUpdate: test.NewMockUpdateFn(nil),
+				},
+				sg: &fake.MockSecurityGroupClient{
+					MockDescribe: func(ctx context.Context, input *awsec2.DescribeSecurityGroupsInput, opts []func(*awsec2.Options)) (*awsec2.DescribeSecurityGroupsOutput, error) {
+						return &awsec2.DescribeSecurityGroupsOutput{
+							SecurityGroups: []awsec2types.SecurityGroup{{GroupId: aws.String(sgID)}},
+						}, nil
+					},
+				},
+				cr: sg(withExternalName("")),
+			},
+			want: want{
+				cr: sg(withStatus(v1beta1.SecurityGroupObservation{
+					SecurityGroupID: sgID,
+				}),
+					withExternalName(sgID),
+					withConditions(xpv1.Available())),
+				result: managed.ExternalObservation{
+					ResourceExists:   true,
+					ResourceUpToDate: true,
+				},
+			},
+		},
+		"EmptyExternalNameNonExistingSG": {
+			args: args{
+				kube: &test.MockClient{
+					MockUpdate: test.NewMockUpdateFn(nil),
+				},
+				sg: &fake.MockSecurityGroupClient{
+					MockDescribe: func(ctx context.Context, input *awsec2.DescribeSecurityGroupsInput, opts []func(*awsec2.Options)) (*awsec2.DescribeSecurityGroupsOutput, error) {
+						return &awsec2.DescribeSecurityGroupsOutput{
+							SecurityGroups: []awsec2types.SecurityGroup{},
+						}, nil
+					},
+				},
+				cr: sg(),
+			},
+			want: want{
+				cr: sg(),
+				result: managed.ExternalObservation{
+					ResourceExists: false,
+				},
+			},
+		},
+		"EmptyExternalNameClientError": {
+			args: args{
+				kube: &test.MockClient{
+					MockUpdate: test.NewMockUpdateFn(nil),
+				},
+				sg: &fake.MockSecurityGroupClient{
+					MockDescribe: func(ctx context.Context, input *awsec2.DescribeSecurityGroupsInput, opts []func(*awsec2.Options)) (*awsec2.DescribeSecurityGroupsOutput, error) {
+						return nil, errBoom
+					},
+				},
+				cr: sg(),
+			},
+			want: want{
+				cr: sg(),
+				result: managed.ExternalObservation{
+					ResourceExists: false,
+				},
+				err: awsclient.Wrap(errBoom, errGetSecurityGroup),
+			},
+		},
 		"MultipleSGs": {
 			args: args{
 				sg: &fake.MockSecurityGroupClient{