Block EDP deregistration if other instances exists with the same suba…

…ccount (#270)

cmd/broker/deprovisioning.go

@@ -40,7 +40,7 @@ func NewDeprovisioningProcessingQueue(ctx context.Context, workersAmount int, de
 			step: deprovisioning.NewAvsEvaluationsRemovalStep(avsDel, db.Operations(), externalEvalAssistant, internalEvalAssistant),
 		},
 		{
-			step:     deprovisioning.NewEDPDeregistrationStep(db.Operations(), edpClient, cfg.EDP),
+			step:     deprovisioning.NewEDPDeregistrationStep(db.Operations(), db.Instances(), edpClient, cfg.EDP),
 			disabled: cfg.EDP.Disabled,
 		},
 		{

internal/edp/client_fake.go

@@ -103,7 +103,7 @@ func (f *FakeClient) DeleteMetadataTenant(name, env, key string) error {
 
 func checkDataTenantPayload(data DataTenantPayload) error {
 	if data.Name == "" || data.Environment == "" || data.Secret == "" {
-		return fmt.Errorf("one of the fields in DataTenantPayload is missing")
+		return fmt.Errorf("one of the fields in DataTenantPayload is missing: %v", data)
 	}
 	return nil
 }

internal/process/deprovisioning/edp_deregistration.go

@@ -24,13 +24,22 @@ type EDPDeregistrationStep struct {
 	operationManager *process.OperationManager
 	client           EDPClient
 	config           edp.Config
+	dbInstances      storage.Instances
+	dbOperations     storage.Operations
 }
 
-func NewEDPDeregistrationStep(os storage.Operations, client EDPClient, config edp.Config) *EDPDeregistrationStep {
+type InstanceOperationStorage interface {
+	storage.Operations
+	storage.Instances
+}
+
+func NewEDPDeregistrationStep(os storage.Operations, is storage.Instances, client EDPClient, config edp.Config) *EDPDeregistrationStep {
 	return &EDPDeregistrationStep{
 		operationManager: process.NewOperationManager(os),
 		client:           client,
 		config:           config,
+		dbOperations:     os,
+		dbInstances:      is,
 	}
 }
 
@@ -39,6 +48,30 @@ func (s *EDPDeregistrationStep) Name() string {
 }
 
 func (s *EDPDeregistrationStep) Run(operation internal.Operation, log logrus.FieldLogger) (internal.Operation, time.Duration, error) {
+	instances, err := s.dbInstances.FindAllInstancesForSubAccounts([]string{operation.SubAccountID})
+	if err != nil {
+		log.Errorf("Unable to get instances for given subaccount: %s", err.Error())
+		return operation, time.Second, nil
+	}
+	// check if there is any other instance for given subaccount and such instances are not being deprovisioned
+	numberOfInstancesWithEDP := 0
+	var edpInstanceIDs []string
+	for _, instance := range instances {
+		lastOperation, err := s.dbOperations.GetLastOperation(instance.InstanceID)
+		if err != nil {
+			log.Errorf("Unable to get last operation for given instance (Id=%s): %s", instance.InstanceID, err.Error())
+			return operation, time.Second, nil
+		}
+		if lastOperation.Type != internal.OperationTypeDeprovision {
+			numberOfInstancesWithEDP = numberOfInstancesWithEDP + 1
+			edpInstanceIDs = append(edpInstanceIDs, operation.InstanceID)
+		}
+	}
+	if numberOfInstancesWithEDP > 0 {
+		log.Infof("Skipping EDP deregistration due to existing other instances: %s", strings.Join(edpInstanceIDs, ", "))
+		return operation, 0, nil
+	}
+
 	log.Info("Delete DataTenant metadata")
 
 	subAccountID := strings.ToLower(operation.SubAccountID)
@@ -55,7 +88,7 @@ func (s *EDPDeregistrationStep) Run(operation internal.Operation, log logrus.Fie
 	}
 
 	log.Info("Delete DataTenant")
-	err := s.client.DeleteDataTenant(subAccountID, s.config.Environment)
+	err = s.client.DeleteDataTenant(subAccountID, s.config.Environment)
 	if err != nil {
 		return s.handleError(operation, err, log, "cannot remove DataTenant")
 	}

internal/process/deprovisioning/edp_deregistration_test.go

@@ -6,6 +6,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/google/uuid"
+	"github.com/kyma-project/kyma-environment-broker/internal/fixture"
+
 	"github.com/kyma-project/kyma-environment-broker/internal"
 	"github.com/kyma-project/kyma-environment-broker/internal/edp"
 	"github.com/kyma-project/kyma-environment-broker/internal/storage"
@@ -18,55 +21,165 @@ const (
 	edpEnvironment = "test"
 )
 
+var metadataTenantKeys = []string{
+	edp.MaasConsumerEnvironmentKey,
+	edp.MaasConsumerRegionKey,
+	edp.MaasConsumerSubAccountKey,
+	edp.MaasConsumerServicePlan,
+}
+
 func TestEDPDeregistration_Run(t *testing.T) {
 	// given
 	client := edp.NewFakeClient()
-	err := client.CreateDataTenant(edp.DataTenantPayload{
-		Name:        edpName,
+	prepareEDP(edpName, client)
+
+	memoryStorage := storage.NewMemoryStorage()
+	_, operation := prepareDeprovisioningInstanceWithSubaccount(edpName, memoryStorage.Instances(), memoryStorage.Operations())
+
+	step := NewEDPDeregistrationStep(memoryStorage.Operations(), memoryStorage.Instances(), client, edp.Config{
 		Environment: edpEnvironment,
-		Secret:      base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf("%s%s", edpName, edpEnvironment))),
 	})
+
+	// when
+	_, repeat, err := step.Run(operation, logrus.New())
+
+	// then
+	assert.Equal(t, 0*time.Second, repeat)
 	assert.NoError(t, err)
 
-	metadataTenantKeys := []string{
-		edp.MaasConsumerEnvironmentKey,
-		edp.MaasConsumerRegionKey,
-		edp.MaasConsumerSubAccountKey,
-		edp.MaasConsumerServicePlan,
+	for _, key := range metadataTenantKeys {
+		metadataTenant, metadataTenantExists := client.GetMetadataItem(edpName, edpEnvironment, key)
+		assert.False(t, metadataTenantExists)
+		assert.Equal(t, edp.MetadataItem{}, metadataTenant)
 	}
 
+	dataTenant, dataTenantExists := client.GetDataTenantItem(edpName, edpEnvironment)
+	assert.False(t, dataTenantExists)
+	assert.Equal(t, edp.DataTenantItem{}, dataTenant)
+}
+
+func TestEDPDeregistration_RunWithOtherInstances(t *testing.T) {
+	// given
+	client := edp.NewFakeClient()
+	prepareEDP(edpName, client)
+
+	memoryStorage := storage.NewMemoryStorage()
+	_, _ = prepareProvisionedInstanceWithSubaccount(edpName, memoryStorage.Instances(), memoryStorage.Operations())
+	_, operation := prepareDeprovisioningInstanceWithSubaccount(edpName, memoryStorage.Instances(), memoryStorage.Operations())
+
+	step := NewEDPDeregistrationStep(memoryStorage.Operations(), memoryStorage.Instances(), client, edp.Config{
+		Environment: edpEnvironment,
+	})
+
+	// when
+	_, repeat, err := step.Run(operation, logrus.New())
+
+	// then
+	assert.Equal(t, 0*time.Second, repeat)
+	assert.NoError(t, err)
+
 	for _, key := range metadataTenantKeys {
-		err = client.CreateMetadataTenant(edpName, edpEnvironment, edp.MetadataTenantPayload{
-			Key:   key,
-			Value: "-",
-		})
-		assert.NoError(t, err)
+		_, metadataTenantExists := client.GetMetadataItem(edpName, edpEnvironment, key)
+		assert.True(t, metadataTenantExists)
+	}
+
+	_, dataTenantExists := client.GetDataTenantItem(edpName, edpEnvironment)
+	assert.True(t, dataTenantExists)
+}
+
+func TestEDPDeregistration_RunWithOtherInstancesButDifferentSubaccount(t *testing.T) {
+	// given
+	client := edp.NewFakeClient()
+	prepareEDP(edpName, client)
+
+	memoryStorage := storage.NewMemoryStorage()
+	_, _ = prepareProvisionedInstanceWithSubaccount("subaccount-other", memoryStorage.Instances(), memoryStorage.Operations())
+	_, operation := prepareDeprovisioningInstanceWithSubaccount(edpName, memoryStorage.Instances(), memoryStorage.Operations())
+
+	step := NewEDPDeregistrationStep(memoryStorage.Operations(), memoryStorage.Instances(), client, edp.Config{
+		Environment: edpEnvironment,
+	})
+
+	// when
+	_, repeat, err := step.Run(operation, logrus.New())
+
+	// then
+	assert.Equal(t, 0*time.Second, repeat)
+	assert.NoError(t, err)
+
+	for _, key := range metadataTenantKeys {
+		_, metadataTenantExists := client.GetMetadataItem(edpName, edpEnvironment, key)
+		assert.False(t, metadataTenantExists)
 	}
 
+	_, dataTenantExists := client.GetDataTenantItem(edpName, edpEnvironment)
+	assert.False(t, dataTenantExists)
+}
+
+func TestEDPDeregistration_RunWithOtherInstancesInDeprovisioningState(t *testing.T) {
+	// given
+	client := edp.NewFakeClient()
+	prepareEDP(edpName, client)
+
 	memoryStorage := storage.NewMemoryStorage()
-	step := NewEDPDeregistrationStep(memoryStorage.Operations(), client, edp.Config{
+	_, _ = prepareDeprovisioningInstanceWithSubaccount(edpName, memoryStorage.Instances(), memoryStorage.Operations())
+	_, operation := prepareDeprovisioningInstanceWithSubaccount(edpName, memoryStorage.Instances(), memoryStorage.Operations())
+
+	step := NewEDPDeregistrationStep(memoryStorage.Operations(), memoryStorage.Instances(), client, edp.Config{
 		Environment: edpEnvironment,
 	})
 
 	// when
-	_, repeat, err := step.Run(
-		internal.Operation{
-			InstanceDetails: internal.InstanceDetails{
-				SubAccountID: edpName,
-			},
-		}, logrus.New())
+	_, repeat, err := step.Run(operation, logrus.New())
 
 	// then
 	assert.Equal(t, 0*time.Second, repeat)
 	assert.NoError(t, err)
 
 	for _, key := range metadataTenantKeys {
-		metadataTenant, metadataTenantExists := client.GetMetadataItem(edpName, edpEnvironment, key)
+		_, metadataTenantExists := client.GetMetadataItem(edpName, edpEnvironment, key)
 		assert.False(t, metadataTenantExists)
-		assert.Equal(t, edp.MetadataItem{}, metadataTenant)
 	}
 
-	dataTenant, dataTenantExists := client.GetDataTenantItem(edpName, edpEnvironment)
+	_, dataTenantExists := client.GetDataTenantItem(edpName, edpEnvironment)
 	assert.False(t, dataTenantExists)
-	assert.Equal(t, edp.DataTenantItem{}, dataTenant)
+}
+
+func prepareEDP(subaccountId string, client *edp.FakeClient) {
+	client.CreateDataTenant(edp.DataTenantPayload{
+		Name:        subaccountId,
+		Environment: edpEnvironment,
+		Secret:      base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf("%s%s", edpName, edpEnvironment))),
+	})
+
+	for _, key := range metadataTenantKeys {
+		client.CreateMetadataTenant(subaccountId, edpEnvironment, edp.MetadataTenantPayload{
+			Key:   key,
+			Value: "-",
+		})
+	}
+}
+
+func prepareProvisionedInstanceWithSubaccount(subaccountId string, instances storage.Instances, operations storage.Operations) (internal.Instance, internal.Operation) {
+	instanceID := uuid.New().String()
+	instance := fixture.FixInstance(instanceID)
+	instance.SubAccountID = subaccountId
+	operation := fixture.FixProvisioningOperation(uuid.New().String(), instanceID)
+	operation.SubAccountID = subaccountId
+	instances.Insert(instance)
+	operations.InsertOperation(operation)
+
+	return instance, operation
+}
+
+func prepareDeprovisioningInstanceWithSubaccount(subaccountId string, instances storage.Instances, operations storage.Operations) (internal.Instance, internal.Operation) {
+	instanceID := uuid.New().String()
+	instance := fixture.FixInstance(instanceID)
+	instance.SubAccountID = subaccountId
+	operation := fixture.FixDeprovisioningOperationAsOperation(uuid.New().String(), instanceID)
+	operation.SubAccountID = subaccountId
+	instances.Insert(instance)
+	operations.InsertOperation(operation)
+
+	return instance, operation
 }