feat: add sweeper to destroy resources after test runs

Adds `sweep` target to Makefile, which will run a sweeper to delete
resources in a project. This would include:

- services that have "test-" as a prefix for their name
- service integration endpoints
- VPCs

.github/workflows/tests.yml

@@ -36,3 +36,19 @@ jobs:
           AIVEN_TOKEN: ${{ secrets.AIVEN_TOKEN }}
           AIVEN_PROJECT_NAME: >-
             ${{ secrets.AIVEN_PROJECT_NAME_PREFIX }}${{ needs.setup_aiven_project_suffix.outputs.project_name_suffix }}
+
+  sweep:
+    needs: test
+    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,59 @@
+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"`
+	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 := newAvnGenClient(envVars.Token, envVars.DebugLogging)
+	if err != nil {
+		log.Fatalf("error creating aiven client: %v\n", err)
+	}
+
+	sweepers := []sweeper{
+		&servicesSweeper{client},
+		&vpcsSweeper{client},
+		&serviceIntegrationEndpointsSweeper{client},
+	}
+
+	for _, sweeper := range sweepers {
+		err := sweeper.Sweep(ctx, envVars.Project)
+		if err != nil {
+			log.Fatalf("error sweeping %s: %v\n", sweeper.Name(), err)
+		}
+	}
+}
+
+// newAvnGenClient returns a common Aiven Gen Client setup needed for the sweeper
+func newAvnGenClient(token string, debug bool) (avngen.Client, error) {
+	// configures a default client, using the above env var
+	sharedClient, err := avngen.NewClient(avngen.TokenOpt(token), avngen.DebugOpt(debug))
+	if err != nil {
+		return nil, err
+	}
+
+	return sharedClient, nil
+}

sweeper/service_integrations.go

@@ -0,0 +1,42 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	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 {
+	log.Println("Sweeping service integration endpoints")
+
+	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); err != nil {
+			if isCriticalServiceIntegrationEndpointDeleteError(err) {
+				return fmt.Errorf("error deleting service integration endpoint '%s' during sweep: %w", s.EndpointName, err)
+			}
+		}
+	}
+
+	return nil
+}
+
+// isCriticalServiceIntegrationEndpointDeleteError returns true if the given error's status code is not 404
+func isCriticalServiceIntegrationEndpointDeleteError(err error) bool {
+	return err != nil && !avngen.IsNotFound(err)
+}

sweeper/services.go

@@ -0,0 +1,62 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"strings"
+
+	avngen "github.com/aiven/go-client-codegen"
+	"github.com/aiven/go-client-codegen/handler/service"
+)
+
+type servicesSweeper struct {
+	client avngen.Client
+}
+
+func (sweeper *servicesSweeper) Name() string {
+	return "services"
+}
+
+// Sweep deletes services that have "test-" prefix in their name
+func (sweeper *servicesSweeper) Sweep(ctx context.Context, projectName string) error {
+	log.Println("Sweeping services")
+
+	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 "test-" prefix in their name
+		if !strings.HasPrefix(s.ServiceName, "test-") {
+			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 '%s' during sweep: %w", s.ServiceName, err)
+			}
+		}
+
+		if err := sweeper.client.ServiceDelete(ctx, projectName, s.ServiceName); err != nil {
+			if isCriticalServiceDeleteError(err) {
+				return fmt.Errorf("error deleting service %s during sweep: %w", s.ServiceName, err)
+			}
+		}
+	}
+
+	return nil
+}
+
+// isCriticalServiceDeleteError returns true if the given error's status code is not 404
+func isCriticalServiceDeleteError(err error) bool {
+	return err != nil && !avngen.IsNotFound(err)
+}

sweeper/vpcs.go

@@ -0,0 +1,89 @@
+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 {
+	log.Println("Sweeping VPCs")
+
+	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 '%s' (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 '%s' (ID: %s) during sweep: %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
+}