feat: make time out in delete assets to be configurable

cli/config.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"time"
 
 	"github.com/MakeNowJust/heredoc"
 	"github.com/goto/compass/core/asset"
@@ -21,6 +22,8 @@ import (
 
 const configFlag = "config"
 
+var errDeleteAssetsTimeoutIsZero = errors.New("delete assets timeout must greater than 0 second")
+
 func configCommand(cfg *Config) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "config <command>",
@@ -108,7 +111,16 @@ type Config struct {
 }
 
 type Asset struct {
-	AdditionalTypes []string `mapstructure:"additional_types"`
+	AdditionalTypes     []string      `mapstructure:"additional_types"`
+	DeleteAssetsTimeout time.Duration `mapstructure:"delete_assets_timeout" default:"5m"`
+}
+
+func (a *Asset) Validate() error {
+	if a.DeleteAssetsTimeout == 0 {
+		return errDeleteAssetsTimeoutIsZero
+	}
+
+	return nil
 }
 
 func LoadConfig() (*Config, error) {

cli/server.go

@@ -83,6 +83,10 @@ func serverMigrateCommand(cfg *Config) *cobra.Command {
 }
 
 func runServer(ctx context.Context, cfg *Config) error {
+	if err := cfg.Asset.Validate(); err != nil {
+		return err
+	}
+
 	logger := initLogger(cfg.LogLevel)
 	logger.Info("compass starting", "version", Version)
 
@@ -149,11 +153,12 @@ func runServer(ctx context.Context, cfg *Config) error {
 	}()
 
 	assetService, cancel := asset.NewService(asset.ServiceDeps{
-		AssetRepo:     assetRepository,
-		DiscoveryRepo: discoveryRepository,
-		LineageRepo:   lineageRepository,
-		Worker:        wrkr,
-		Logger:        logger,
+		AssetRepo:           assetRepository,
+		DiscoveryRepo:       discoveryRepository,
+		LineageRepo:         lineageRepository,
+		Worker:              wrkr,
+		Logger:              logger,
+		DeleteAssetsTimeout: cfg.Asset.DeleteAssetsTimeout,
 	})
 	defer cancel()
 

compass.yaml.example

@@ -78,3 +78,4 @@ client:
 asset:
     additional_types:
         - fact_source
+    delete_assets_timeout: 5m

core/asset/service.go

@@ -19,6 +19,7 @@ type Service struct {
 	lineageRepository   LineageRepository
 	worker              Worker
 	logger              log.Logger
+	deleteAssetsTimeout time.Duration
 	cancelFnList        []func()
 
 	assetOpCounter metric.Int64Counter
@@ -35,11 +36,12 @@ type Worker interface {
 }
 
 type ServiceDeps struct {
-	AssetRepo     Repository
-	DiscoveryRepo DiscoveryRepository
-	LineageRepo   LineageRepository
-	Worker        Worker
-	Logger        log.Logger
+	AssetRepo           Repository
+	DiscoveryRepo       DiscoveryRepository
+	LineageRepo         LineageRepository
+	Worker              Worker
+	Logger              log.Logger
+	DeleteAssetsTimeout time.Duration
 }
 
 func NewService(deps ServiceDeps) (service *Service, cancel func()) {
@@ -55,6 +57,7 @@ func NewService(deps ServiceDeps) (service *Service, cancel func()) {
 		lineageRepository:   deps.LineageRepo,
 		worker:              deps.Worker,
 		logger:              deps.Logger,
+		deleteAssetsTimeout: deps.DeleteAssetsTimeout,
 		cancelFnList:        make([]func(), 0),
 
 		assetOpCounter: assetOpCounter,
@@ -149,8 +152,8 @@ func (s *Service) DeleteAssets(ctx context.Context, request DeleteAssetsRequest)
 		return 0, err
 	}
 
-	if !request.DryRun {
-		newCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	if !request.DryRun && total > 0 {
+		newCtx, cancel := context.WithTimeout(context.Background(), s.deleteAssetsTimeout)
 		s.cancelFnList = append(s.cancelFnList, cancel)
 		go s.executeDeleteAssets(newCtx, deleteSQLExpr)
 	}
@@ -161,16 +164,16 @@ func (s *Service) DeleteAssets(ctx context.Context, request DeleteAssetsRequest)
 func (s *Service) executeDeleteAssets(ctx context.Context, deleteSQLExpr queryexpr.ExprStr) {
 	deletedURNs, err := s.assetRepository.DeleteByQueryExpr(ctx, deleteSQLExpr)
 	if err != nil {
-		s.logger.Error("Asset deletion failed, skipping Elasticsearch and Lineage deletions. Err:", err)
+		s.logger.Error("asset deletion failed, skipping elasticsearch and lineage deletions", "err:", err)
 		return
 	}
 
 	if err := s.lineageRepository.DeleteByURNs(ctx, deletedURNs); err != nil {
-		s.logger.Error("Error occurred during Lineage deletion:", err)
+		s.logger.Error("error occurred during lineage deletion", "err:", err)
 	}
 
 	if err := s.worker.EnqueueDeleteAssetsByQueryExprJob(ctx, deleteSQLExpr.String()); err != nil {
-		s.logger.Error("Error occurred during Elasticsearch deletion:", err)
+		s.logger.Error("error occurred during elasticsearch deletion", "err:", err)
 	}
 }