Merge pull request #160 from appuio/feat/cleanup-broken-records

Add a command to clean up stale inflight records

Makefile

@@ -74,6 +74,10 @@ run-controller: build ## Starts control api controller against the current Kuber
 	$(localenv_make) webhook-certs/tls.key
 	$(BIN_FILENAME) controller --username-prefix "appuio#" --webhook-cert-dir=./local-env/webhook-certs --webhook-port=9444 --zap-log-level debug --billingentity-email-cron-interval "@every 1m"
 
+.PHONY: run-cleanup
+run-cleanup: build ## Starts cleanup command
+	$(BIN_FILENAME) cleanup --billing-entity-odoo8-url $(BE_ODOO_URL)
+
 .PHONY: local-env
 local-env-setup: ## Setup local kind-based dev environment
 	$(localenv_make) setup

apiserver/billing/odoostorage/odoo/odoo8/client/model/partner.go

@@ -142,3 +142,7 @@ func (o Odoo) CreatePartner(ctx context.Context, p Partner) (id int, err error)
 func (o Odoo) UpdateRawPartner(ctx context.Context, ids []int, raw any) error {
 	return o.querier.UpdateGenericModel(ctx, PartnerModel, ids, raw)
 }
+
+func (o Odoo) DeletePartner(ctx context.Context, ids []int) error {
+	return o.querier.DeleteGenericModel(ctx, PartnerModel, ids)
+}

apiserver/billing/odoostorage/odoo/odoo8/odoo8.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/google/uuid"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -31,6 +32,7 @@ var metaUIDNamespace = uuid.MustParse("51887759-C769-4829-9910-BB9D5F92767D")
 var roleAccountFilter = []any{"category_id", "in", []int{roleAccountCategory}}
 var activeFilter = []any{"active", "in", []bool{true}}
 var notInflightFilter = []any{"x_control_api_inflight", "in", []any{false}}
+var mustInflightFilter = []any{"x_control_api_inflight", "not in", []any{false}}
 
 var (
 	// There's a ton of fields we don't want to override in Odoo.
@@ -63,22 +65,36 @@ type Config struct {
 	PaymentTermID                int
 }
 
-func NewOdoo8Storage(odooURL string, debugTransport bool, conf Config) odoo.OdooStorage {
-	return &oodo8Storage{
+var _ odoo.OdooStorage = &Odoo8Storage{}
+
+func NewOdoo8Storage(odooURL string, debugTransport bool, conf Config) *Odoo8Storage {
+	return &Odoo8Storage{
 		config: conf,
 		sessionCreator: func(ctx context.Context) (client.QueryExecutor, error) {
 			return client.Open(ctx, odooURL, client.ClientOptions{UseDebugLogger: debugTransport})
 		},
 	}
 }
 
-type oodo8Storage struct {
+func NewFailedRecordScrubber(odooURL string, debugTransport bool) *FailedRecordScrubber {
+	return &FailedRecordScrubber{
+		sessionCreator: func(ctx context.Context) (client.QueryExecutor, error) {
+			return client.Open(ctx, odooURL, client.ClientOptions{UseDebugLogger: debugTransport})
+		},
+	}
+}
+
+type Odoo8Storage struct {
 	config Config
 
 	sessionCreator func(ctx context.Context) (client.QueryExecutor, error)
 }
 
-func (s *oodo8Storage) Get(ctx context.Context, name string) (*billingv1.BillingEntity, error) {
+type FailedRecordScrubber struct {
+	sessionCreator func(ctx context.Context) (client.QueryExecutor, error)
+}
+
+func (s *Odoo8Storage) Get(ctx context.Context, name string) (*billingv1.BillingEntity, error) {
 	company, accountingContact, err := s.get(ctx, name)
 	if err != nil {
 		return nil, err
@@ -88,7 +104,7 @@ func (s *oodo8Storage) Get(ctx context.Context, name string) (*billingv1.Billing
 	return &be, nil
 }
 
-func (s *oodo8Storage) get(ctx context.Context, name string) (company model.Partner, accountingContact model.Partner, err error) {
+func (s *Odoo8Storage) get(ctx context.Context, name string) (company model.Partner, accountingContact model.Partner, err error) {
 	id, err := k8sIDToOdooID(name)
 	if err != nil {
 		return model.Partner{}, model.Partner{}, err
@@ -117,7 +133,7 @@ func (s *oodo8Storage) get(ctx context.Context, name string) (company model.Part
 	return company, accountingContact, nil
 }
 
-func (s *oodo8Storage) List(ctx context.Context) ([]billingv1.BillingEntity, error) {
+func (s *Odoo8Storage) List(ctx context.Context) ([]billingv1.BillingEntity, error) {
 	l := klog.FromContext(ctx)
 
 	session, err := s.sessionCreator(ctx)
@@ -173,7 +189,7 @@ func (s *oodo8Storage) List(ctx context.Context) ([]billingv1.BillingEntity, err
 	return bes, nil
 }
 
-func (s *oodo8Storage) Create(ctx context.Context, be *billingv1.BillingEntity) error {
+func (s *Odoo8Storage) Create(ctx context.Context, be *billingv1.BillingEntity) error {
 	l := klog.FromContext(ctx)
 
 	if be == nil {
@@ -225,7 +241,7 @@ func (s *oodo8Storage) Create(ctx context.Context, be *billingv1.BillingEntity)
 	return nil
 }
 
-func (s *oodo8Storage) Update(ctx context.Context, be *billingv1.BillingEntity) error {
+func (s *Odoo8Storage) Update(ctx context.Context, be *billingv1.BillingEntity) error {
 	l := klog.FromContext(ctx)
 
 	if be == nil {
@@ -274,6 +290,42 @@ func (s *oodo8Storage) Update(ctx context.Context, be *billingv1.BillingEntity)
 	return nil
 }
 
+// CleanupIncompleteRecords looks for partner records in Odoo that still have the "inflight" flag set despite being older than `minAge`. Those records are then deleted.
+// Such records might come into existence due to a partially failed creation request.
+func (s *FailedRecordScrubber) CleanupIncompleteRecords(ctx context.Context, minAge time.Duration) error {
+	l := klog.FromContext(ctx)
+	l.Info("Looking for stale inflight partner records...")
+
+	session, err := s.sessionCreator(ctx)
+	if err != nil {
+		return err
+	}
+	o := model.NewOdoo(session)
+
+	inflightRecords, err := o.SearchPartners(ctx, []client.Filter{
+		mustInflightFilter,
+	})
+	if err != nil {
+		return err
+	}
+
+	ids := []int{}
+
+	for _, record := range inflightRecords {
+		createdTime := record.CreationTimestamp.ToTime()
+
+		if createdTime.Before(time.Now().Add(-1 * minAge)) {
+			ids = append(ids, record.ID)
+			l.Info("Preparing to delete inflight partner record", "name", record.Name, "id", record.ID)
+		}
+	}
+
+	if len(ids) != 0 {
+		return o.DeletePartner(ctx, ids)
+	}
+	return nil
+}
+
 func k8sIDToOdooID(id string) (int, error) {
 	if !strings.HasPrefix(id, "be-") {
 		return 0, fmt.Errorf("invalid ID, missing prefix: %s", id)

apiserver/billing/odoostorage/odoo/odoo8/odoo8_test.go

@@ -390,11 +390,11 @@ func Test_CreateUpdate_UnknownCountry(t *testing.T) {
 	require.ErrorContains(t, subject.Update(context.Background(), s), "unknown country")
 }
 
-func createStorage(t *testing.T) (*gomock.Controller, *clientmock.MockQueryExecutor, *oodo8Storage) {
+func createStorage(t *testing.T) (*gomock.Controller, *clientmock.MockQueryExecutor, *Odoo8Storage) {
 	ctrl := gomock.NewController(t)
 	mock := clientmock.NewMockQueryExecutor(ctrl)
 
-	return ctrl, mock, &oodo8Storage{
+	return ctrl, mock, &Odoo8Storage{
 		config: Config{
 			CountryIDs: map[string]int{
 				"":            0,
@@ -410,3 +410,51 @@ func createStorage(t *testing.T) (*gomock.Controller, *clientmock.MockQueryExecu
 		},
 	}
 }
+
+func createFailedRecordScrubber(t *testing.T) (*gomock.Controller, *clientmock.MockQueryExecutor, *FailedRecordScrubber) {
+	ctrl := gomock.NewController(t)
+	mock := clientmock.NewMockQueryExecutor(ctrl)
+
+	return ctrl, mock, &FailedRecordScrubber{
+		sessionCreator: func(ctx context.Context) (client.QueryExecutor, error) {
+			return mock, nil
+		},
+	}
+}
+
+func TestCleanup(t *testing.T) {
+	ctrl, mock, subject := createFailedRecordScrubber(t)
+	defer ctrl.Finish()
+
+	tn := time.Now()
+	to := tn.Add(time.Hour * -1)
+
+	gomock.InOrder(
+		// Fetch stale records
+		mock.EXPECT().SearchGenericModel(gomock.Any(), gomock.Any(), gomock.Any()).SetArg(2, model.PartnerList{
+			Items: []model.Partner{
+				{
+					ID:                702,
+					Name:              "Accounting",
+					CreationTimestamp: client.Date(tn),
+					Parent:            model.OdooCompositeID{ID: 700, Valid: true},
+					EmailRaw:          model.NewNullable("accounting@test.com, notifications@test.com"),
+					Inflight:          model.NewNullable("fooo"),
+				},
+				{
+					ID:                703,
+					Name:              "Accounting",
+					CreationTimestamp: client.Date(to),
+					Parent:            model.OdooCompositeID{ID: 700, Valid: true},
+					EmailRaw:          model.NewNullable("accounting@test.com, notifications@test.com"),
+					Inflight:          model.NewNullable("fooo"),
+				},
+			},
+		}),
+		mock.EXPECT().DeleteGenericModel(gomock.Any(), gomock.Any(), gomock.Eq([]int{703})).Return(nil),
+	)
+
+	err := subject.CleanupIncompleteRecords(context.Background(), time.Minute)
+	require.NoError(t, err)
+
+}

cleanup.go

@@ -0,0 +1,41 @@
+package main
+
+import (
+	"os"
+	"time"
+
+	"github.com/spf13/cobra"
+	"k8s.io/klog/v2"
+	ctrl "sigs.k8s.io/controller-runtime"
+
+	"github.com/appuio/control-api/apiserver/billing/odoostorage/odoo/odoo8"
+)
+
+// APICommand creates a new command allowing to start the API server
+func CleanupCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use: "cleanup",
+	}
+
+	odooUrl := cmd.Flags().String("billing-entity-odoo8-url", "http://localhost:8069", "URL of the Odoo instance to use for billing entities")
+	debugTransport := cmd.Flags().Bool("billing-entity-odoo8-debug-transport", false, "Enable debug logging for the Odoo transport")
+	minAge := cmd.Flags().Duration("billing-entity-odoo8-cleanup-after", time.Hour, "Clean up only records older than this")
+
+	cmd.Run = func(cmd *cobra.Command, args []string) {
+		ctx := ctrl.SetupSignalHandler()
+		l := klog.FromContext(ctx)
+		scrubber := odoo8.NewFailedRecordScrubber(
+			*odooUrl,
+			*debugTransport,
+		)
+
+		err := scrubber.CleanupIncompleteRecords(ctx, *minAge)
+		if err != nil {
+			l.Error(err, "Unable to clean up incomplete records")
+			os.Exit(1)
+		}
+		l.Info("Cleanup complete!")
+	}
+
+	return cmd
+}

main.go

@@ -20,7 +20,7 @@ var (
 )
 
 func main() {
-	rootCommand.AddCommand(ControllerCommand(), APICommand())
+	rootCommand.AddCommand(ControllerCommand(), APICommand(), CleanupCommand())
 
 	if err := rootCommand.Execute(); err != nil {
 		fmt.Fprintln(os.Stderr, err)