Backport 7804 into 2.7.1 (#7896)

use grpc for communicating with compactor for query time filtering of
data requested for deletion (#7804)

Manual backport of #7804 and #7814

**What this PR does / why we need it**:
Add grpc support to compactor for getting delete requests and gen number
for query time filtering.
Since these requests are internal to Loki, it would be good to use grpc
instead of HTTP same as all the internal requests we do in Loki.

I have added a new config for accepting the grpc address of the
compactor. I tried having just the existing config and detecting if it
is a grpc server, but it was hard to do it reliably, considering the
different deployment modes we support. I think it is safe to keep it the
same and eventually deprecate the existing config.

**Checklist**
- [x] Documentation added
- [x] Tests updated
- [x] `CHANGELOG.md` updated

(cherry picked from commit 1410808)

Co-authored-by: Sandeep Sukhani <[email protected]>

Author: trevorwhitney

CHANGELOG.md

@@ -45,10 +45,16 @@ Check the history of the branch FIXME.
 
 #### Loki
 
+##### Enhancements
+
+* [7804](https://github.com/grafana/loki/pull/7804) **sandeepsukhani**: Use grpc for communicating with compactor for query time filtering of data requested for deletion.
+
 ##### Fixes
+
 * [7453](https://github.com/grafana/loki/pull/7453) **periklis**: Add single compactor http client for delete and gennumber clients
 
 ##### Changes
+
 * [7877](https://github.com/grafana/loki/pull/7877)A **trevorwhitney**: Due to a known bug with experimental new delete mode feature, the default delete mode has been changed to `filter-only`.
 
 ## 2.7.0

clients/pkg/promtail/targets/cloudflare/fields.go

@@ -31,7 +31,7 @@ var (
 	allFields = append(extendedFields, []string{
 		"BotScore", "BotScoreSrc", "ClientRequestBytes", "ClientSrcPort", "ClientXRequestedWith", "CacheTieredFill", "EdgeResponseCompressionRatio", "EdgeServerIP", "FirewallMatchesSources",
 		"FirewallMatchesActions", "FirewallMatchesRuleIDs", "OriginResponseBytes", "OriginResponseTime", "ClientDeviceType", "WAFFlags", "WAFMatchedVar", "EdgeColoID",
-        "RequestHeaders", "ResponseHeaders",
+		"RequestHeaders", "ResponseHeaders",
 	}...)
 )
 

clients/pkg/promtail/targets/lokipush/pushtargetmanager_test.go

@@ -68,7 +68,7 @@ func Test_validateJobName(t *testing.T) {
 					},
 				},
 			},
-			wantErr: false,
+			wantErr:     false,
 			expectedJob: "job_1_2_3_4_job",
 		},
 	}

docs/sources/configuration/_index.md

@@ -2911,6 +2911,10 @@ This way, one doesn't have to replicate configuration in multiple places.
 # CLI flag: -common.compactor-address
 [compactor_address: <string> | default = ""]
 
+# Address and port number where the compactor grpc requests are being served.
+# CLI flag: -common.compactor-grpc-address
+[compactor_grpc_address: <string> | default = ""]
+
 ## analytics
 
 The `analytics` block configures the reporting of Loki analytics to grafana.com.

pkg/logqlmodel/logqlmodel.go

@@ -1,9 +1,10 @@
 package logqlmodel
 
 import (
-	"github.com/grafana/loki/pkg/querier/queryrange/queryrangebase/definitions"
 	"github.com/prometheus/prometheus/promql/parser"
 
+	"github.com/grafana/loki/pkg/querier/queryrange/queryrangebase/definitions"
+
 	"github.com/grafana/loki/pkg/logproto"
 	"github.com/grafana/loki/pkg/logqlmodel/stats"
 )

pkg/logqlmodel/metadata/context_test.go

@@ -5,8 +5,9 @@ import (
 	"errors"
 	"testing"
 
-	"github.com/grafana/loki/pkg/querier/queryrange/queryrangebase/definitions"
 	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/loki/pkg/querier/queryrange/queryrangebase/definitions"
 )
 
 func TestHeaders(t *testing.T) {

pkg/loki/common/common.go

@@ -43,6 +43,9 @@ type Config struct {
 
 	// CompactorAddress is the http address of the compactor in the form http://host:port
 	CompactorAddress string `yaml:"compactor_address"`
+
+	// CompactorAddress is the grpc address of the compactor in the form host:port
+	CompactorGRPCAddress string `yaml:"compactor_grpc_address"`
 }
 
 func (c *Config) RegisterFlags(f *flag.FlagSet) {
@@ -57,6 +60,7 @@ func (c *Config) RegisterFlags(f *flag.FlagSet) {
 	throwaway.Var((*flagext.StringSlice)(&c.InstanceInterfaceNames), "common.instance-interface-names", "List of network interfaces to read address from.")
 
 	f.StringVar(&c.CompactorAddress, "common.compactor-address", "", "the http address of the compactor in the form http://host:port")
+	f.StringVar(&c.CompactorGRPCAddress, "common.compactor-grpc-address", "", "the grpc address of the compactor in the form host:port")
 }
 
 type Storage struct {

pkg/loki/loki.go

@@ -45,6 +45,7 @@ import (
 	"github.com/grafana/loki/pkg/storage"
 	"github.com/grafana/loki/pkg/storage/config"
 	"github.com/grafana/loki/pkg/storage/stores/indexshipper/compactor"
+	compactor_client "github.com/grafana/loki/pkg/storage/stores/indexshipper/compactor/client"
 	"github.com/grafana/loki/pkg/storage/stores/indexshipper/compactor/deletion"
 	"github.com/grafana/loki/pkg/storage/stores/series/index"
 	"github.com/grafana/loki/pkg/storage/stores/shipper/indexgateway"
@@ -69,30 +70,31 @@ type Config struct {
 	UseBufferedLogger bool `yaml:"use_buffered_logger"`
 	UseSyncLogger     bool `yaml:"use_sync_logger"`
 
-	Common           common.Config            `yaml:"common,omitempty"`
-	Server           server.Config            `yaml:"server,omitempty"`
-	InternalServer   internalserver.Config    `yaml:"internal_server,omitempty"`
-	Distributor      distributor.Config       `yaml:"distributor,omitempty"`
-	Querier          querier.Config           `yaml:"querier,omitempty"`
-	CompactorClient  compactor.ClientConfig   `yaml:"delete_client,omitempty"`
-	IngesterClient   client.Config            `yaml:"ingester_client,omitempty"`
-	Ingester         ingester.Config          `yaml:"ingester,omitempty"`
-	StorageConfig    storage.Config           `yaml:"storage_config,omitempty"`
-	IndexGateway     indexgateway.Config      `yaml:"index_gateway"`
-	ChunkStoreConfig config.ChunkStoreConfig  `yaml:"chunk_store_config,omitempty"`
-	SchemaConfig     config.SchemaConfig      `yaml:"schema_config,omitempty"`
-	LimitsConfig     validation.Limits        `yaml:"limits_config,omitempty"`
-	TableManager     index.TableManagerConfig `yaml:"table_manager,omitempty"`
-	Worker           worker.Config            `yaml:"frontend_worker,omitempty"`
-	Frontend         lokifrontend.Config      `yaml:"frontend,omitempty"`
-	Ruler            ruler.Config             `yaml:"ruler,omitempty"`
-	QueryRange       queryrange.Config        `yaml:"query_range,omitempty"`
-	RuntimeConfig    runtimeconfig.Config     `yaml:"runtime_config,omitempty"`
-	MemberlistKV     memberlist.KVConfig      `yaml:"memberlist"`
-	Tracing          tracing.Config           `yaml:"tracing"`
-	CompactorConfig  compactor.Config         `yaml:"compactor,omitempty"`
-	QueryScheduler   scheduler.Config         `yaml:"query_scheduler"`
-	UsageReport      usagestats.Config        `yaml:"analytics"`
+	Common              common.Config               `yaml:"common,omitempty"`
+	Server              server.Config               `yaml:"server,omitempty"`
+	InternalServer      internalserver.Config       `yaml:"internal_server,omitempty"`
+	Distributor         distributor.Config          `yaml:"distributor,omitempty"`
+	Querier             querier.Config              `yaml:"querier,omitempty"`
+	CompactorHTTPClient compactor_client.HTTPConfig `yaml:"compactor_client,omitempty"`
+	CompactorGRPCClient compactor_client.GRPCConfig `yaml:"compactor_grpc_client,omitempty"`
+	IngesterClient      client.Config               `yaml:"ingester_client,omitempty"`
+	Ingester            ingester.Config             `yaml:"ingester,omitempty"`
+	StorageConfig       storage.Config              `yaml:"storage_config,omitempty"`
+	IndexGateway        indexgateway.Config         `yaml:"index_gateway"`
+	ChunkStoreConfig    config.ChunkStoreConfig     `yaml:"chunk_store_config,omitempty"`
+	SchemaConfig        config.SchemaConfig         `yaml:"schema_config,omitempty"`
+	LimitsConfig        validation.Limits           `yaml:"limits_config,omitempty"`
+	TableManager        index.TableManagerConfig    `yaml:"table_manager,omitempty"`
+	Worker              worker.Config               `yaml:"frontend_worker,omitempty"`
+	Frontend            lokifrontend.Config         `yaml:"frontend,omitempty"`
+	Ruler               ruler.Config                `yaml:"ruler,omitempty"`
+	QueryRange          queryrange.Config           `yaml:"query_range,omitempty"`
+	RuntimeConfig       runtimeconfig.Config        `yaml:"runtime_config,omitempty"`
+	MemberlistKV        memberlist.KVConfig         `yaml:"memberlist"`
+	Tracing             tracing.Config              `yaml:"tracing"`
+	CompactorConfig     compactor.Config            `yaml:"compactor,omitempty"`
+	QueryScheduler      scheduler.Config            `yaml:"query_scheduler"`
+	UsageReport         usagestats.Config           `yaml:"analytics"`
 }
 
 // RegisterFlags registers flag.
@@ -115,7 +117,8 @@ func (c *Config) RegisterFlags(f *flag.FlagSet) {
 	c.Common.RegisterFlags(f)
 	c.Distributor.RegisterFlags(f)
 	c.Querier.RegisterFlags(f)
-	c.CompactorClient.RegisterFlags(f)
+	c.CompactorHTTPClient.RegisterFlags(f)
+	c.CompactorGRPCClient.RegisterFlags(f)
 	c.IngesterClient.RegisterFlags(f)
 	c.Ingester.RegisterFlags(f)
 	c.StorageConfig.RegisterFlags(f)

pkg/loki/modules.go

@@ -12,8 +12,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/grafana/loki/pkg/querier/queryrange/queryrangebase"
-
 	"github.com/NYTimes/gziphandler"
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
@@ -43,6 +41,7 @@ import (
 	"github.com/grafana/loki/pkg/lokifrontend/frontend/v2/frontendv2pb"
 	"github.com/grafana/loki/pkg/querier"
 	"github.com/grafana/loki/pkg/querier/queryrange"
+	"github.com/grafana/loki/pkg/querier/queryrange/queryrangebase"
 	"github.com/grafana/loki/pkg/ruler"
 	base_ruler "github.com/grafana/loki/pkg/ruler/base"
 	"github.com/grafana/loki/pkg/runtime"
@@ -54,6 +53,8 @@ import (
 	"github.com/grafana/loki/pkg/storage/config"
 	"github.com/grafana/loki/pkg/storage/stores/indexshipper"
 	"github.com/grafana/loki/pkg/storage/stores/indexshipper/compactor"
+	compactor_client "github.com/grafana/loki/pkg/storage/stores/indexshipper/compactor/client"
+	"github.com/grafana/loki/pkg/storage/stores/indexshipper/compactor/client/grpc"
 	"github.com/grafana/loki/pkg/storage/stores/indexshipper/compactor/deletion"
 	"github.com/grafana/loki/pkg/storage/stores/indexshipper/compactor/generationnumber"
 	"github.com/grafana/loki/pkg/storage/stores/series/index"
@@ -676,41 +677,53 @@ func (t *Loki) initQueryFrontendTripperware() (_ services.Service, err error) {
 func (t *Loki) initCacheGenerationLoader() (_ services.Service, err error) {
 	var client generationnumber.CacheGenClient
 	if t.supportIndexDeleteRequest() {
-		compactorAddress, err := t.compactorAddress()
-		if err != nil {
-			return nil, err
-		}
-
-		httpClient, err := compactor.NewCompactorHTTPClient(t.Cfg.CompactorClient)
+		compactorAddress, isGRPCAddress, err := t.compactorAddress()
 		if err != nil {
 			return nil, err
 		}
 
-		client, err = generationnumber.NewGenNumberClient(compactorAddress, httpClient)
-		if err != nil {
-			return nil, err
+		reg := prometheus.WrapRegistererWith(prometheus.Labels{"for": "cache_gen", "client_type": t.Cfg.Target.String()}, prometheus.DefaultRegisterer)
+		if isGRPCAddress {
+			client, err = compactor_client.NewGRPCClient(compactorAddress, t.Cfg.CompactorGRPCClient, reg)
+			if err != nil {
+				return nil, err
+			}
+		} else {
+			client, err = compactor_client.NewHTTPClient(compactorAddress, t.Cfg.CompactorHTTPClient)
+			if err != nil {
+				return nil, err
+			}
 		}
 	}
 
 	t.cacheGenerationLoader = generationnumber.NewGenNumberLoader(client, prometheus.DefaultRegisterer)
-	return services.NewIdleService(nil, nil), nil
+	return services.NewIdleService(nil, func(failureCase error) error {
+		t.cacheGenerationLoader.Stop()
+		return nil
+	}), nil
 }
 
 func (t *Loki) supportIndexDeleteRequest() bool {
 	return config.UsingObjectStorageIndex(t.Cfg.SchemaConfig.Configs)
 }
 
-func (t *Loki) compactorAddress() (string, error) {
+// compactorAddress returns the configured address of the compactor.
+// It prefers grpc address over http. If the address is grpc then the bool would be true otherwise false
+func (t *Loki) compactorAddress() (string, bool, error) {
 	if t.Cfg.isModuleEnabled(All) || t.Cfg.isModuleEnabled(Read) {
 		// In single binary or read modes, this module depends on Server
-		return fmt.Sprintf("http://127.0.0.1:%d", t.Cfg.Server.HTTPListenPort), nil
+		return fmt.Sprintf("%s:%d", t.Cfg.Server.GRPCListenAddress, t.Cfg.Server.GRPCListenPort), true, nil
 	}
 
-	if t.Cfg.Common.CompactorAddress == "" {
-		return "", errors.New("query filtering for deletes requires 'compactor_address' to be configured")
+	if t.Cfg.Common.CompactorAddress == "" && t.Cfg.Common.CompactorGRPCAddress == "" {
+		return "", false, errors.New("query filtering for deletes requires 'compactor_grpc_address' or 'compactor_address' to be configured")
 	}
 
-	return t.Cfg.Common.CompactorAddress, nil
+	if t.Cfg.Common.CompactorGRPCAddress != "" {
+		return t.Cfg.Common.CompactorGRPCAddress, true, nil
+	}
+
+	return t.Cfg.Common.CompactorAddress, false, nil
 }
 
 func (t *Loki) initQueryFrontend() (_ services.Service, err error) {
@@ -1007,6 +1020,7 @@ func (t *Loki) initCompactor() (services.Service, error) {
 		t.Server.HTTP.Path("/loki/api/v1/delete").Methods("GET").Handler(t.addCompactorMiddleware(t.compactor.DeleteRequestsHandler.GetAllDeleteRequestsHandler))
 		t.Server.HTTP.Path("/loki/api/v1/delete").Methods("DELETE").Handler(t.addCompactorMiddleware(t.compactor.DeleteRequestsHandler.CancelDeleteRequestHandler))
 		t.Server.HTTP.Path("/loki/api/v1/cache/generation_numbers").Methods("GET").Handler(t.addCompactorMiddleware(t.compactor.DeleteRequestsHandler.GetCacheGenerationNumberHandler))
+		grpc.RegisterCompactorServer(t.Server.GRPC, t.compactor.DeleteRequestsGRPCHandler)
 	}
 
 	return t.compactor, nil
@@ -1123,17 +1137,26 @@ func (t *Loki) deleteRequestsClient(clientType string, limits *validation.Overri
 		return deletion.NewNoOpDeleteRequestsStore(), nil
 	}
 
-	compactorAddress, err := t.compactorAddress()
+	compactorAddress, isGRPCAddress, err := t.compactorAddress()
 	if err != nil {
 		return nil, err
 	}
 
-	httpClient, err := compactor.NewCompactorHTTPClient(t.Cfg.CompactorClient)
-	if err != nil {
-		return nil, err
+	reg := prometheus.WrapRegistererWith(prometheus.Labels{"for": "delete_requests", "client_type": clientType}, prometheus.DefaultRegisterer)
+	var compactorClient deletion.CompactorClient
+	if isGRPCAddress {
+		compactorClient, err = compactor_client.NewGRPCClient(compactorAddress, t.Cfg.CompactorGRPCClient, reg)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		compactorClient, err = compactor_client.NewHTTPClient(compactorAddress, t.Cfg.CompactorHTTPClient)
+		if err != nil {
+			return nil, err
+		}
 	}
 
-	client, err := deletion.NewDeleteRequestsClient(compactorAddress, httpClient, t.deleteClientMetrics, clientType)
+	client, err := deletion.NewDeleteRequestsClient(compactorClient, t.deleteClientMetrics, clientType)
 	if err != nil {
 		return nil, err
 	}

pkg/querier/queryrange/queryrangebase/middleware_test.go

@@ -31,3 +31,5 @@ type fakeGenNumberLoader struct {
 func (l *fakeGenNumberLoader) GetResultsCacheGenNumber(tenantIDs []string) string {
 	return l.genNumber
 }
+
+func (l *fakeGenNumberLoader) Stop() {}

pkg/querier/queryrange/queryrangebase/results_cache.go

@@ -68,6 +68,7 @@ func NewResultsCacheMetrics(registerer prometheus.Registerer) *ResultsCacheMetri
 
 type CacheGenNumberLoader interface {
 	GetResultsCacheGenNumber(tenantIDs []string) string
+	Stop()
 }
 
 // ResultsCacheConfig is the config for the results cache.

pkg/querier/queryrange/queryrangebase/results_cache_test.go

@@ -1054,3 +1054,5 @@ func newMockCacheGenNumberLoader() CacheGenNumberLoader {
 func (mockCacheGenNumberLoader) GetResultsCacheGenNumber(tenantIDs []string) string {
 	return ""
 }
+
+func (l mockCacheGenNumberLoader) Stop() {}