-
Notifications
You must be signed in to change notification settings - Fork 3.5k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
use grpc for communicating with compactor for query time filtering of…
… data requested for deletion (#7804) **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
- Loading branch information
1 parent
d3615f7
commit 1410808
Showing
22 changed files
with
2,321 additions
and
326 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
100 changes: 100 additions & 0 deletions
100
pkg/storage/stores/indexshipper/compactor/client/grpc.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| package client | ||
|
|
||
| import ( | ||
| "context" | ||
| "flag" | ||
| "github.com/grafana/loki/pkg/storage/stores/indexshipper/compactor/deletion" | ||
| "github.com/prometheus/common/model" | ||
| "github.com/weaveworks/common/user" | ||
|
|
||
| "github.com/grafana/dskit/grpcclient" | ||
| deletion_grpc "github.com/grafana/loki/pkg/storage/stores/indexshipper/compactor/client/grpc" | ||
| "github.com/prometheus/client_golang/prometheus" | ||
| "github.com/prometheus/client_golang/prometheus/promauto" | ||
| "github.com/weaveworks/common/instrument" | ||
| "google.golang.org/grpc" | ||
| ) | ||
|
|
||
| type GRPCConfig struct { | ||
| GRPCClientConfig grpcclient.Config `yaml:",inline"` | ||
| } | ||
|
|
||
| // RegisterFlags registers flags. | ||
| func (cfg *GRPCConfig) RegisterFlags(f *flag.FlagSet) { | ||
| cfg.GRPCClientConfig.RegisterFlagsWithPrefix("", f) | ||
| } | ||
|
|
||
| type compactorGRPCClient struct { | ||
| cfg GRPCConfig | ||
|
|
||
| GRPCClientRequestDuration *prometheus.HistogramVec | ||
| conn *grpc.ClientConn | ||
| grpcClient deletion_grpc.CompactorClient | ||
| } | ||
|
|
||
| // NewGRPCClient supports only methods which are used for internal communication of Loki like | ||
| // loading delete requests and cache gen numbers for query time filtering. | ||
| func NewGRPCClient(addr string, cfg GRPCConfig, r prometheus.Registerer) (deletion.CompactorClient, error) { | ||
| client := &compactorGRPCClient{ | ||
| cfg: cfg, | ||
| GRPCClientRequestDuration: promauto.With(r).NewHistogramVec(prometheus.HistogramOpts{ | ||
| Namespace: "loki_compactor", | ||
| Name: "grpc_request_duration_seconds", | ||
| Help: "Time (in seconds) spent serving requests when using compactor GRPC client", | ||
| Buckets: instrument.DefBuckets, | ||
| }, []string{"operation", "status_code"}), | ||
| } | ||
|
|
||
| dialOpts, err := cfg.GRPCClientConfig.DialOption(grpcclient.Instrument(client.GRPCClientRequestDuration)) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| client.conn, err = grpc.Dial(addr, dialOpts...) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| client.grpcClient = deletion_grpc.NewCompactorClient(client.conn) | ||
| return client, nil | ||
| } | ||
|
|
||
| func (s *compactorGRPCClient) Stop() { | ||
| s.conn.Close() | ||
| } | ||
|
|
||
| func (s *compactorGRPCClient) GetAllDeleteRequestsForUser(ctx context.Context, userID string) ([]deletion.DeleteRequest, error) { | ||
| ctx = user.InjectOrgID(ctx, userID) | ||
| grpcResp, err := s.grpcClient.GetDeleteRequests(ctx, &deletion_grpc.GetDeleteRequestsRequest{}) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| deleteRequests := make([]deletion.DeleteRequest, len(grpcResp.DeleteRequests)) | ||
| for i, dr := range grpcResp.DeleteRequests { | ||
| deleteRequests[i] = deletion.DeleteRequest{ | ||
| RequestID: dr.RequestID, | ||
| StartTime: model.Time(dr.StartTime), | ||
| EndTime: model.Time(dr.EndTime), | ||
| Query: dr.Query, | ||
| Status: deletion.DeleteRequestStatus(dr.Status), | ||
| CreatedAt: model.Time(dr.CreatedAt), | ||
| } | ||
| } | ||
|
|
||
| return deleteRequests, nil | ||
| } | ||
|
|
||
| func (s *compactorGRPCClient) GetCacheGenerationNumber(ctx context.Context, userID string) (string, error) { | ||
| ctx = user.InjectOrgID(ctx, userID) | ||
| grpcResp, err := s.grpcClient.GetCacheGenNumbers(ctx, &deletion_grpc.GetCacheGenNumbersRequest{}) | ||
| if err != nil { | ||
| return "", err | ||
| } | ||
|
|
||
| return grpcResp.ResultsCacheGen, nil | ||
| } | ||
|
|
||
| func (s *compactorGRPCClient) Name() string { | ||
| return "grpc_client" | ||
| } |
Oops, something went wrong.