feat: port search, deletion, lineage, and observability features

.github/workflows/lint.yml

@@ -6,14 +6,14 @@ jobs:
   golangci:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v6
         with:
           fetch-depth: 0
-      - uses: actions/setup-go@v4
+      - uses: actions/setup-go@v6
         with:
           go-version: "1.20"
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v3
+        uses: golangci/golangci-lint-action@v9
         with:
-          version: v1.53
+          version: v2.1
           args: --timeout=5m

.github/workflows/main.yml

@@ -9,22 +9,22 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout code
-        uses: actions/checkout@v3
+        uses: actions/checkout@v6
         with:
           fetch-depth: 0
       - name: Set up Go
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v6
         with:
           go-version: "1.20"
       - name: Login to DockerHub
-        uses: docker/login-action@v1
+        uses: docker/login-action@v4
         with:
           registry: docker.io
           username: ${{ secrets.DOCKERHUB_USERNAME }}
           password: ${{ secrets.DOCKERHUB_TOKEN }}
       - name: Publish dev image
         id: docker_dev_build
-        uses: docker/build-push-action@v2
+        uses: docker/build-push-action@v7
         with:
           push: true
           file: "./Dockerfile.dev"

.github/workflows/release.yml

@@ -10,21 +10,21 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v6
         with:
           fetch-depth: 0
       - name: Set up Go
-        uses: actions/setup-go@v2
+        uses: actions/setup-go@v6
         with:
           go-version: "1.20"
       - name: Login to DockerHub
-        uses: docker/login-action@v1
+        uses: docker/login-action@v4
         with:
           registry: docker.io
           username: ${{ secrets.DOCKERHUB_USERNAME }}
           password: ${{ secrets.DOCKERHUB_TOKEN }}
       - name: Run GoReleaser
-        uses: goreleaser/goreleaser-action@v4
+        uses: goreleaser/goreleaser-action@v7
         with:
           distribution: goreleaser
           version: latest

.github/workflows/test.yml

@@ -12,8 +12,8 @@ jobs:
         env:
           ES_TEST_SERVER_URL: "http://elasticsearch:9200"
     steps:
-      - uses: actions/checkout@v3
-      - uses: actions/setup-go@v4
+      - uses: actions/checkout@v6
+      - uses: actions/setup-go@v6
         with:
           go-version: "1.20"
       - name: Install dependencies

.golangci.yml

@@ -0,0 +1,22 @@
+version: "2"
+
+linters:
+  enable:
+    - errcheck
+    - staticcheck
+  settings:
+    errcheck:
+      exclude-functions:
+        - (io.Closer).Close
+        - (*database/sql.Conn).Close
+        - (*github.com/jmoiron/sqlx.Conn).Close
+        - (*google.golang.org/grpc.ClientConn).Close
+        - (github.com/ory/dockertest/v3.Resource).Close
+  exclusions:
+    rules:
+      - linters:
+          - errcheck
+        text: '\.Close'
+      - linters:
+          - staticcheck
+        text: 'QF1008'

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 := "409f146"
 
 .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:

cli/config.go

@@ -11,7 +11,7 @@ import (
 	esStore "github.com/raystack/compass/internal/store/elasticsearch"
 	"github.com/raystack/compass/internal/store/postgres"
 	"github.com/raystack/compass/pkg/metrics"
-	"github.com/raystack/compass/pkg/statsd"
+	"github.com/raystack/compass/pkg/telemetry"
 	"github.com/raystack/salt/cmdx"
 	"github.com/raystack/salt/config"
 	"github.com/spf13/cobra"
@@ -79,12 +79,12 @@ type Config struct {
 	// Log
 	LogLevel string `yaml:"log_level" mapstructure:"log_level" default:"info"`
 
-	// StatsD
-	StatsD statsd.Config `mapstructure:"statsd"`
-
 	// NewRelic
 	NewRelic metrics.NewRelicConfig `mapstructure:"newrelic"`
 
+	// Telemetry
+	Telemetry telemetry.Config `mapstructure:"telemetry"`
+
 	// Elasticsearch
 	Elasticsearch esStore.Config `mapstructure:"elasticsearch"`
 

cli/server.go

@@ -18,7 +18,7 @@ import (
 	compassserver "github.com/raystack/compass/internal/server"
 	esStore "github.com/raystack/compass/internal/store/elasticsearch"
 	"github.com/raystack/compass/internal/store/postgres"
-	"github.com/raystack/compass/pkg/statsd"
+	"github.com/raystack/compass/pkg/telemetry"
 	"github.com/raystack/salt/log"
 	"github.com/spf13/cobra"
 )
@@ -100,10 +100,12 @@ func runServer(config *Config) error {
 	if err != nil {
 		return err
 	}
-	statsdReporter, err := statsd.Init(logger, config.StatsD)
+
+	otelCleanup, err := telemetry.Init(ctx, config.Telemetry, logger)
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to initialize telemetry: %w", err)
 	}
+	defer otelCleanup()
 
 	esClient, err := initElasticsearch(logger, config.Elasticsearch)
 	if err != nil {
@@ -132,7 +134,7 @@ func runServer(config *Config) error {
 	if err != nil {
 		return fmt.Errorf("failed to create new user repository: %w", err)
 	}
-	userService := user.NewService(logger, userRepository, user.ServiceWithStatsDReporter(statsdReporter))
+	userService := user.NewService(logger, userRepository)
 
 	assetRepository, err := postgres.NewAssetRepository(pgClient, userRepository, 0, config.Service.Identity.ProviderDefaultName)
 	if err != nil {
@@ -168,7 +170,6 @@ func runServer(config *Config) error {
 		logger,
 		pgClient,
 		nrApp,
-		statsdReporter,
 		namespaceService,
 		assetService,
 		starService,

core/asset/asset.go

@@ -3,11 +3,12 @@ package asset
 //go:generate mockery --name=Repository -r --case underscore --with-expecter --structname AssetRepository --filename asset_repository.go --output=./mocks
 import (
 	"context"
-	"github.com/raystack/compass/core/namespace"
+	"fmt"
 	"time"
 
-	"github.com/raystack/compass/core/user"
 	"github.com/r3labs/diff/v2"
+	"github.com/raystack/compass/core/namespace"
+	"github.com/raystack/compass/core/user"
 )
 
 type Repository interface {
@@ -22,6 +23,10 @@ type Repository interface {
 	Upsert(ctx context.Context, ns *namespace.Namespace, ast *Asset) (string, error)
 	DeleteByID(ctx context.Context, id string) error
 	DeleteByURN(ctx context.Context, urn string) error
+	SoftDeleteByID(ctx context.Context, id string) (string, error)
+	SoftDeleteByURN(ctx context.Context, urn string) (string, error)
+	GetCountByIsDeleted(ctx context.Context, isDeleted bool) (int, error)
+	HardDeleteByURNs(ctx context.Context, urns []string) error
 	AddProbe(ctx context.Context, ns *namespace.Namespace, assetURN string, probe *Probe) error
 	GetProbes(ctx context.Context, assetURN string) ([]Probe, error)
 	GetProbesWithFilter(ctx context.Context, flt ProbesFilter) (map[string][]Probe, error)
@@ -39,14 +44,18 @@ type Asset struct {
 	URL         string                 `json:"url" diff:"url"`
 	Labels      map[string]string      `json:"labels" diff:"labels"`
 	Owners      []user.User            `json:"owners,omitempty" diff:"owners"`
+	IsDeleted   bool                   `json:"is_deleted" diff:"is_deleted"`
 	CreatedAt   time.Time              `json:"created_at" diff:"-"`
 	UpdatedAt   time.Time              `json:"updated_at" diff:"-"`
+	RefreshedAt *time.Time             `json:"refreshed_at,omitempty" diff:"-"`
 	Version     string                 `json:"version" diff:"-"`
 	UpdatedBy   user.User              `json:"updated_by" diff:"-"`
 	Changelog   diff.Changelog         `json:"changelog,omitempty" diff:"-"`
 	Probes      []Probe                `json:"probes,omitempty"`
 }
 
+var ErrAssetAlreadyDeleted = fmt.Errorf("asset already deleted")
+
 // Diff returns nil changelog with nil error if equal
 // returns wrapped r3labs/diff Changelog struct with nil error if not equal
 func (a *Asset) Diff(otherAsset *Asset) (diff.Changelog, error) {

core/asset/discovery.go

@@ -11,19 +11,49 @@ type DiscoveryRepository interface {
 	Upsert(ctx context.Context, ns *namespace.Namespace, ast *Asset) error
 	DeleteByID(ctx context.Context, ns *namespace.Namespace, assetID string) error
 	DeleteByURN(ctx context.Context, ns *namespace.Namespace, assetURN string) error
+	SoftDeleteByURN(ctx context.Context, ns *namespace.Namespace, assetURN string) error
 	Search(ctx context.Context, cfg SearchConfig) (results []SearchResult, err error)
 	Suggest(ctx context.Context, cfg SearchConfig) (suggestions []string, err error)
+	GroupAssets(ctx context.Context, cfg GroupConfig) ([]GroupResult, error)
+}
+
+// GroupConfig represents configuration for grouping assets
+type GroupConfig struct {
+	GroupBy       []string
+	Filters       SearchFilter
+	IncludeFields []string
+	Size          int
+	Namespace     *namespace.Namespace
+}
+
+// GroupResult represents a single group of assets
+type GroupResult struct {
+	Fields []GroupField
+	Assets []Asset
+}
+
+// GroupField represents a key-value pair identifying a group
+type GroupField struct {
+	Key   string
+	Value string
 }
 
 // SearchFilter is a filter intended to be used as a search
 // criteria for operations involving asset search
 type SearchFilter = map[string][]string
 
+// SearchFlags contains flags that modify search behavior
+type SearchFlags struct {
+	DisableFuzzy    bool
+	EnableHighlight bool
+	IsColumnSearch  bool
+}
+
 // SearchConfig represents a search query along
 // with any corresponding filter(s)
 type SearchConfig struct {
 	// Text to search for
-	Text string `validate:"required"`
+	Text string
 
 	// Filters specifies document level values to look for.
 	// Multiple values can be specified for a single key
@@ -32,12 +62,21 @@ type SearchConfig struct {
 	// Number of relevant results to return
 	MaxResults int
 
+	// Offset is the offset for search results, used for pagination
+	Offset int
+
 	// RankBy is a param to rank based on a specific parameter
 	RankBy string
 
 	// Queries is a param to search a resource based on asset's fields
 	Queries map[string]string
 
+	// IncludeFields specifies which fields to include in search results
+	IncludeFields []string
+
+	// Flags contains optional search flags
+	Flags SearchFlags
+
 	// Namespace under which assets are partitioned. *Required*
 	Namespace *namespace.Namespace `validate:"required"`
 }

core/asset/discovery_test.go

@@ -75,12 +75,12 @@ func TestSearchConfig_Validate(t *testing.T) {
 			wantErr: true,
 		},
 		{
-			name: "fail validation if text is empty",
+			name: "should not fail validation if text is empty but namespace is present",
 			fields: fields{
 				Text:      "",
 				Namespace: &namespace.Namespace{},
 			},
-			wantErr: true,
+			wantErr: false,
 		},
 		{
 			name: "should not fail validation if all required fields are non empty",

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

@@ -23,15 +23,18 @@ const (
 )
 
 type LineageQuery struct {
-	Level     int
-	Direction LineageDirection
+	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
 type LineageRepository interface {
 	GetGraph(ctx context.Context, urn string, query LineageQuery) (LineageGraph, error)
 	Upsert(ctx context.Context, ns *namespace.Namespace, urn string, upstreams, downstreams []string) error
 	DeleteByURN(ctx context.Context, urn string) error
+	DeleteByURNs(ctx context.Context, urns []string) error
 }
 
 type LineageGraph []LineageEdge