ci(tests): add sweeper to destroy resources after test runs

.github/workflows/tests.yml

@@ -36,3 +36,20 @@ jobs:
           AIVEN_TOKEN: ${{ secrets.AIVEN_TOKEN }}
           AIVEN_PROJECT_NAME: >-
             ${{ secrets.AIVEN_PROJECT_NAME_PREFIX }}${{ needs.setup_aiven_project_suffix.outputs.project_name_suffix }}
+
+  sweep:
+    if: always()
+    needs: [test, setup_aiven_project_suffix]
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+      - uses: actions/setup-go@v5
+        with:
+          go-version-file: go.mod
+      - run: make sweep
+        env:
+          AIVEN_TOKEN: ${{ secrets.AIVEN_TOKEN }}
+          AIVEN_PROJECT_NAME: >-
+            ${{ secrets.AIVEN_PROJECT_NAME_PREFIX }}${{ needs.setup_aiven_project_suffix.outputs.project_name_suffix }}

Makefile

@@ -301,3 +301,7 @@ define go-install-tool
 		mv "$$(echo "$(1)" | sed "s/-$(3)$$//")" $(1) ;\
 	}
 endef
+
+PHONY: sweep
+sweep: ## Run sweep to remove all resources created by e2e tests.
+	go run ./sweeper/...

sweeper/main.go

@@ -0,0 +1,51 @@
+package main
+
+import (
+	"context"
+	"log"
+
+	avngen "github.com/aiven/go-client-codegen"
+	"github.com/kelseyhightower/envconfig"
+)
+
+type sweepConfig struct {
+	Token         string   `envconfig:"AIVEN_TOKEN" required:"true"`
+	Project       string   `envconfig:"AIVEN_PROJECT_NAME" required:"true"`
+	SweepPrefixes []string `envconfig:"AIVEN_SWEEP_PREFIX" default:"test"`
+	DebugLogging  bool     `envconfig:"ENABLE_DEBUG_LOGGING"`
+}
+
+type sweeper interface {
+	Name() string
+	Sweep(ctx context.Context, projectName string) error
+}
+
+func main() {
+	envVars := new(sweepConfig)
+	ctx := context.Background()
+	err := envconfig.Process("", envVars)
+	if err != nil {
+		log.Fatalf("error processing environment variables: %v\n", err)
+	}
+
+	// generate a new client
+	client, err := avngen.NewClient(avngen.TokenOpt(envVars.Token), avngen.DebugOpt(envVars.DebugLogging))
+	if err != nil {
+		log.Fatalf("error creating aiven client: %v\n", err)
+	}
+
+	sweepers := []sweeper{
+		&servicesSweeper{client: client, sweepPrefixes: envVars.SweepPrefixes},
+		&vpcsSweeper{client},
+		&serviceIntegrationEndpointsSweeper{client},
+	}
+
+	for _, sweeper := range sweepers {
+		log.Printf("Sweeping %s\n", sweeper.Name())
+
+		err := sweeper.Sweep(ctx, envVars.Project)
+		if err != nil {
+			log.Fatalf("error sweeping %s: %v\n", sweeper.Name(), err)
+		}
+	}
+}

sweeper/service_integrations.go

@@ -0,0 +1,32 @@
+package main
+
+import (
+	"context"
+	"fmt"
+
+	avngen "github.com/aiven/go-client-codegen"
+)
+
+type serviceIntegrationEndpointsSweeper struct {
+	client avngen.Client
+}
+
+func (sweeper *serviceIntegrationEndpointsSweeper) Name() string {
+	return "Service integration endpoints"
+}
+
+// Sweep deletes all service integration endpoints in a project
+func (sweeper *serviceIntegrationEndpointsSweeper) Sweep(ctx context.Context, projectName string) error {
+	endpoints, err := sweeper.client.ServiceIntegrationEndpointList(ctx, projectName)
+	if err != nil {
+		return fmt.Errorf("error retrieving a list of integration endpoints: %w", err)
+	}
+
+	for _, s := range endpoints {
+		if err := sweeper.client.ServiceIntegrationEndpointDelete(ctx, projectName, s.EndpointId); avngen.OmitNotFound(err) != nil {
+			return fmt.Errorf("error deleting service integration endpoint %q: %w", s.EndpointName, err)
+		}
+	}
+
+	return nil
+}

sweeper/services.go

@@ -0,0 +1,61 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	avngen "github.com/aiven/go-client-codegen"
+	"github.com/aiven/go-client-codegen/handler/service"
+)
+
+type servicesSweeper struct {
+	client        avngen.Client
+	sweepPrefixes []string
+}
+
+func (sweeper *servicesSweeper) Name() string {
+	return "services"
+}
+
+// Sweep deletes services in a project
+func (sweeper *servicesSweeper) Sweep(ctx context.Context, projectName string) error {
+	services, err := sweeper.client.ServiceList(ctx, projectName)
+	if err != nil {
+		return fmt.Errorf("error retrieving a list of services : %w", err)
+	}
+
+	for _, s := range services {
+		// only delete services that have a specified prefix in their name
+		if !hasPrefixAny(s.ServiceName, sweeper.sweepPrefixes) {
+			continue
+		}
+
+		// if service termination_protection is on service cannot be deleted
+		// update service and turn termination_protection off
+		if s.TerminationProtection {
+			terminationProtection := false
+			_, err := sweeper.client.ServiceUpdate(ctx, projectName, s.ServiceName, &service.ServiceUpdateIn{
+				TerminationProtection: &terminationProtection,
+			})
+			if err != nil {
+				return fmt.Errorf("error disabling `termination_protection` for service %q: %w", s.ServiceName, err)
+			}
+		}
+
+		if err := sweeper.client.ServiceDelete(ctx, projectName, s.ServiceName); avngen.OmitNotFound(err) != nil {
+			return fmt.Errorf("error deleting service %s: %w", s.ServiceName, err)
+		}
+	}
+
+	return nil
+}
+
+func hasPrefixAny(s string, prefixes []string) bool {
+	for _, prefix := range prefixes {
+		if strings.HasPrefix(s, prefix) {
+			return true
+		}
+	}
+	return false
+}

sweeper/vpcs.go

@@ -0,0 +1,86 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"net/http"
+	"time"
+
+	avngen "github.com/aiven/go-client-codegen"
+)
+
+type vpcsSweeper struct {
+	client avngen.Client
+}
+
+func (sweeper *vpcsSweeper) Name() string {
+	return "VPCs"
+}
+
+// Sweep deletes VPCs within a project
+func (sweeper *vpcsSweeper) Sweep(ctx context.Context, projectName string) error {
+	vpcs, err := sweeper.client.VpcList(ctx, projectName)
+	if err != nil {
+		return fmt.Errorf("error retrieving a list of VPCs: %w", err)
+	}
+
+	for _, v := range vpcs {
+		// If VPC is being deleted, skip it
+		if v.State == "DELETING" {
+			continue
+		}
+		// VPCs cannot be deleted if there is a service in it, or if it is moving out of it
+		// (e.g. service was deleted from the VPC). Thus, we need to use a retry mechanism to delete the VPC
+		err := waitForTaskToComplete(ctx, func() (bool, error) {
+			if _, vpcDeleteErr := sweeper.client.VpcDelete(ctx, projectName, v.ProjectVpcId); vpcDeleteErr != nil {
+				if isCriticalVpcDeleteError(vpcDeleteErr) {
+					return false, fmt.Errorf("error fetching VPC %s: %q", v.ProjectVpcId, vpcDeleteErr)
+				}
+				log.Printf("VPC in cloud %q (ID: %s) is not ready for deletion yet", v.CloudName, v.ProjectVpcId)
+				return true, nil
+			}
+
+			return false, nil
+		})
+		if err != nil {
+			return fmt.Errorf("error deleting VPC in cloud %q (ID: %s): %w", v.CloudName, v.ProjectVpcId, err)
+		}
+	}
+
+	return nil
+}
+
+// isCriticalVpcDeleteError returns true if the given error has any status code other than 409
+func isCriticalVpcDeleteError(err error) bool {
+	var e avngen.Error
+
+	return errors.As(err, &e) && e.Status != http.StatusConflict
+}
+
+// waitForTaskToCompleteInterval is the interval to wait before running a task again
+const waitForTaskToCompleteInterval = time.Second * 10
+
+// waitForTaskToComplete waits for a task to complete
+func waitForTaskToComplete(ctx context.Context, f func() (bool, error)) (err error) {
+	retry := false
+
+outer:
+	for {
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("context timeout while retrying operation, error=%q", ctx.Err().Error())
+		case <-time.After(waitForTaskToCompleteInterval):
+			retry, err = f()
+			if err != nil {
+				return err
+			}
+			if !retry {
+				break outer
+			}
+		}
+	}
+
+	return nil
+}