feat: update proto with search flags, lineage options, update_only, is_deleted, and group assets

- Update PROTON_COMMIT to include new proto fields
- Wire SearchFlags (disable_fuzzy, enable_highlight, is_column_search) from proto to handler
- Wire with_attributes (from proto optional bool) and include_deleted in lineage handler
- Add update_only support to UpsertAsset and UpsertPatchAsset handlers
- Add is_deleted filter to GetAllAssets and is_deleted field to Asset proto
- Add IsDeleted to asset.Filter with builder method and postgres filter query
- Add GroupAssets RPC definition (handler implementation pending)
- Remove remaining StatsD references from handler layer

Author: ravisuhag

Makefile

@@ -3,7 +3,7 @@ COMMIT := $(shell git rev-parse --short HEAD)
 TAG := "$(shell git rev-list --tags --max-count=1)"
 VERSION := "$(shell git describe --tags ${TAG})-next"
 BUILD_DIR=dist
-PROTON_COMMIT := "ccbf219312db35a934361ebad895cb40145ca235"
+PROTON_COMMIT := "755609a0cc9e55bf1933ae661558a328e2eadd5f"
 
 .PHONY:  all build clean test tidy vet proto setup format generat
 

buf.gen.yaml

@@ -15,7 +15,6 @@ plugins:
     out: proto
     opt:
       - paths=source_relative
-      - lang=go
   - plugin: buf.build/grpc-ecosystem/openapiv2:v2.16.0
     out: proto
     opt:

core/asset/filter.go

@@ -16,6 +16,7 @@ type Filter struct {
 	QueryFields   []string
 	Query         string
 	Data          map[string][]string
+	IsDeleted     *bool
 }
 
 func (f *Filter) Validate() error {
@@ -32,6 +33,7 @@ type filterBuilder struct {
 	offset        int
 	sortBy        string
 	sortDirection string
+	isDeleted     *bool
 }
 
 func NewFilterBuilder() *filterBuilder {
@@ -83,13 +85,19 @@ func (fb *filterBuilder) SortDirection(sortDirection string) *filterBuilder {
 	return fb
 }
 
+func (fb *filterBuilder) IsDeleted(isDeleted bool) *filterBuilder {
+	fb.isDeleted = &isDeleted
+	return fb
+}
+
 func (fb *filterBuilder) Build() (Filter, error) {
 	flt := Filter{
 		Size:          fb.size,
 		Offset:        fb.offset,
 		SortBy:        fb.sortBy,
 		SortDirection: fb.sortDirection,
 		Query:         fb.q,
+		IsDeleted:     fb.isDeleted,
 	}
 
 	if len(fb.data) != 0 {

core/asset/lineage.go

@@ -26,6 +26,7 @@ type LineageQuery struct {
 	Level          int
 	Direction      LineageDirection
 	WithAttributes bool
+	IncludeDeleted bool
 }
 
 //go:generate mockery --name=LineageRepository -r --case underscore --with-expecter --structname=LineageRepository --filename=lineage_repository.go --output=./mocks

internal/server/v1beta1/asset.go

@@ -12,7 +12,6 @@ import (
 	"github.com/raystack/compass/core/asset"
 	"github.com/raystack/compass/core/star"
 	"github.com/raystack/compass/core/user"
-	"github.com/raystack/compass/pkg/statsd"
 	compassv1beta1 "github.com/raystack/compass/proto/raystack/compass/v1beta1"
 	"github.com/r3labs/diff/v2"
 	"google.golang.org/grpc/codes"
@@ -21,11 +20,6 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
-//go:generate mockery --name=StatsDClient -r --case underscore --with-expecter --structname StatsDClient --filename statsd_monitor.go --output=./mocks
-type StatsDClient interface {
-	Incr(name string) *statsd.Metric
-}
-
 type AssetService interface {
 	GetAllAssets(ctx context.Context, flt asset.Filter, withTotal bool) ([]asset.Asset, uint32, error)
 	GetAssetByID(ctx context.Context, id string) (asset.Asset, error)
@@ -54,7 +48,7 @@ func (server *APIServer) GetAllAssets(ctx context.Context, req *compassv1beta1.G
 		return nil, err
 	}
 
-	flt, err := asset.NewFilterBuilder().
+	fb := asset.NewFilterBuilder().
 		Types(req.GetTypes()).
 		Services(req.GetServices()).
 		Q(req.GetQ()).
@@ -63,8 +57,11 @@ func (server *APIServer) GetAllAssets(ctx context.Context, req *compassv1beta1.G
 		Offset(int(req.GetOffset())).
 		SortBy(req.GetSort()).
 		SortDirection(req.GetDirection()).
-		Data(req.GetData()).
-		Build()
+		Data(req.GetData())
+	if req.GetIsDeleted() {
+		fb = fb.IsDeleted(true)
+	}
+	flt, err := fb.Build()
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, bodyParserErrorMsg(err))
 	}
@@ -260,6 +257,9 @@ func (server *APIServer) UpsertAsset(ctx context.Context, req *compassv1beta1.Up
 		req.GetDownstreams(),
 	)
 	if err != nil {
+		if req.GetUpdateOnly() && errors.As(err, new(asset.NotFoundError)) {
+			return &compassv1beta1.UpsertAssetResponse{Id: ""}, nil
+		}
 		return nil, err
 	}
 
@@ -290,8 +290,14 @@ func (server *APIServer) UpsertPatchAsset(ctx context.Context, req *compassv1bet
 	}
 
 	ast, err := server.assetService.GetAssetByID(ctx, urn)
-	if err != nil && !errors.As(err, &asset.NotFoundError{}) {
-		return nil, internalServerError(server.logger, err.Error())
+	if err != nil {
+		if errors.As(err, &asset.NotFoundError{}) {
+			if req.GetUpdateOnly() {
+				return &compassv1beta1.UpsertPatchAssetResponse{Id: ""}, nil
+			}
+		} else {
+			return nil, internalServerError(server.logger, err.Error())
+		}
 	}
 
 	patchAssetMap := decodePatchAssetToMap(baseAsset)
@@ -332,12 +338,6 @@ func (server *APIServer) DeleteAsset(ctx context.Context, req *compassv1beta1.De
 		if errors.As(err, new(asset.NotFoundError)) {
 			return nil, status.Error(codes.NotFound, err.Error())
 		}
-		if errors.As(err, new(asset.DiscoveryError)) {
-			server.sendStatsDCounterMetric("discovery_error",
-				map[string]string{
-					"method": "delete",
-				})
-		}
 		return nil, internalServerError(server.logger, err.Error())
 	}
 
@@ -404,21 +404,9 @@ func (server *APIServer) upsertAsset(
 	if errors.As(err, new(asset.InvalidError)) {
 		return "", status.Error(codes.InvalidArgument, err.Error())
 	} else if err != nil {
-		if errors.As(err, new(asset.DiscoveryError)) {
-			server.sendStatsDCounterMetric("discovery_error",
-				map[string]string{
-					"method": mode,
-				})
-		}
 		return "", internalServerError(server.logger, err.Error())
 	}
 
-	server.sendStatsDCounterMetric(mode,
-		map[string]string{
-			"type":    ast.Type.String(),
-			"service": ast.Service,
-		})
-
 	return
 }
 
@@ -431,24 +419,12 @@ func (server *APIServer) upsertAssetWithoutLineage(ctx context.Context, ns *name
 
 	assetID, err := server.assetService.UpsertAssetWithoutLineage(ctx, ns, &ast)
 	if err != nil {
-		switch {
-		case errors.As(err, new(asset.InvalidError)):
+		if errors.As(err, new(asset.InvalidError)) {
 			return "", status.Error(codes.InvalidArgument, err.Error())
-
-		case errors.As(err, new(asset.DiscoveryError)):
-			server.sendStatsDCounterMetric("discovery_error",
-				map[string]string{
-					"method": mode,
-				})
 		}
-
 		return "", internalServerError(server.logger, err.Error())
 	}
 
-	server.sendStatsDCounterMetric(mode, map[string]string{
-		"type":    ast.Type.String(),
-		"service": ast.Service,
-	})
 	return assetID, nil
 }
 
@@ -629,6 +605,7 @@ func assetToProto(a asset.Asset, withChangelog bool) (assetPB *compassv1beta1.As
 		CreatedAt:   createdAtPB,
 		UpdatedAt:   updatedAtPB,
 		Probes:      probes,
+		IsDeleted:   a.IsDeleted,
 	}
 	return
 }

internal/server/v1beta1/lineage.go

@@ -28,11 +28,15 @@ func (server *APIServer) GetGraph(ctx context.Context, req *compassv1beta1.GetGr
 
 	// Default to true for backward compatibility
 	withAttributes := true
+	if req.WithAttributes != nil {
+		withAttributes = *req.WithAttributes
+	}
 
 	lineage, err := server.assetService.GetLineage(ctx, req.GetUrn(), asset.LineageQuery{
 		Level:          int(req.GetLevel()),
 		Direction:      direction,
 		WithAttributes: withAttributes,
+		IncludeDeleted: req.GetIncludeDeleted(),
 	})
 	if err != nil {
 		return nil, internalServerError(server.logger, err.Error())

internal/server/v1beta1/mocks/statsd_monitor.go

@@ -1,9 +1,3 @@
 package handlersv1beta1
 
 type Option func(*APIServer)
-
-func WithStatsD(st StatsDClient) Option {
-	return func(s *APIServer) {
-		s.statsDReporter = st
-	}
-}

internal/server/v1beta1/option.go

@@ -29,6 +29,7 @@ func (server *APIServer) SearchAssets(ctx context.Context, req *compassv1beta1.S
 		RankBy:        req.GetRankby(),
 		Queries:       req.GetQuery(),
 		IncludeFields: req.GetIncludeFields(),
+		Flags:         getSearchFlagsFromProto(req.GetFlags()),
 		Namespace:     ns,
 	}
 
@@ -79,6 +80,17 @@ func (server *APIServer) SuggestAssets(ctx context.Context, req *compassv1beta1.
 	}, nil
 }
 
+func getSearchFlagsFromProto(flags *compassv1beta1.SearchFlags) asset.SearchFlags {
+	if flags == nil {
+		return asset.SearchFlags{}
+	}
+	return asset.SearchFlags{
+		DisableFuzzy:    flags.GetDisableFuzzy(),
+		EnableHighlight: flags.GetEnableHighlight(),
+		IsColumnSearch:  flags.GetIsColumnSearch(),
+	}
+}
+
 func filterConfigFromValues(fltMap map[string]string) map[string][]string {
 	var filter = make(map[string][]string)
 	for key, value := range fltMap {

internal/server/v1beta1/search.go

@@ -25,7 +25,6 @@ type APIServer struct {
 	tagTemplateService TagTemplateService
 	userService        UserService
 	logger             log.Logger
-	statsDReporter     StatsDClient
 }
 
 var (
@@ -72,16 +71,6 @@ func (server *APIServer) validateUserInCtx(ctx context.Context, ns *namespace.Na
 	return userID, nil
 }
 
-func (server *APIServer) sendStatsDCounterMetric(metricName string, kvTags map[string]string) {
-	if server.statsDReporter != nil {
-		metric := server.statsDReporter.Incr(metricName)
-		for k, v := range kvTags {
-			metric.Tag(k, v)
-		}
-		metric.Publish()
-	}
-}
-
 func internalServerError(logger log.Logger, msg string) error {
 	ref := time.Now().Unix()