-
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.
* Fix delete updates * Update pkg/storage/stores/shipper/compactor/deletion/delete_requests_client.go Co-authored-by: Michel Hollands <42814411+MichelHollands@users.noreply.github.com> * Add compactor address to the storage config * review comments * Review feedback Co-authored-by: Michel Hollands <42814411+MichelHollands@users.noreply.github.com>
- Loading branch information
1 parent
3b3fcf6
commit 9b2786b
Showing
8 changed files
with
284 additions
and
29 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
174 changes: 174 additions & 0 deletions
174
pkg/storage/stores/shipper/compactor/deletion/delete_requests_client.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,174 @@ | ||
| package deletion | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| "net/http" | ||
| "net/url" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/go-kit/log/level" | ||
|
|
||
| "github.com/grafana/loki/pkg/util/log" | ||
| ) | ||
|
|
||
| const ( | ||
| orgHeaderKey = "X-Scope-OrgID" | ||
| getDeletePath = "/loki/api/v1/delete" | ||
| ) | ||
|
|
||
| type DeleteRequestsClient interface { | ||
| GetAllDeleteRequestsForUser(ctx context.Context, userID string) ([]DeleteRequest, error) | ||
| Stop() | ||
| } | ||
|
|
||
| type deleteRequestsClient struct { | ||
| url string | ||
| httpClient httpClient | ||
| mu sync.RWMutex | ||
|
|
||
| cache map[string][]DeleteRequest | ||
| cacheDuration time.Duration | ||
|
|
||
| stopChan chan struct{} | ||
| } | ||
|
|
||
| type httpClient interface { | ||
| Do(*http.Request) (*http.Response, error) | ||
| } | ||
|
|
||
| type DeleteRequestsStoreOption func(c *deleteRequestsClient) | ||
|
|
||
| func WithRequestClientCacheDuration(d time.Duration) DeleteRequestsStoreOption { | ||
| return func(c *deleteRequestsClient) { | ||
| c.cacheDuration = d | ||
| } | ||
| } | ||
|
|
||
| func NewDeleteRequestsClient(addr string, c httpClient, opts ...DeleteRequestsStoreOption) (DeleteRequestsClient, error) { | ||
| u, err := url.Parse(addr) | ||
| if err != nil { | ||
| level.Error(log.Logger).Log("msg", "error parsing url", "err", err) | ||
| return nil, err | ||
| } | ||
| u.Path = getDeletePath | ||
|
|
||
| client := &deleteRequestsClient{ | ||
| url: u.String(), | ||
| httpClient: c, | ||
| cacheDuration: 5 * time.Minute, | ||
| cache: make(map[string][]DeleteRequest), | ||
| stopChan: make(chan struct{}), | ||
| } | ||
|
|
||
| for _, o := range opts { | ||
| o(client) | ||
| } | ||
|
|
||
| go client.updateLoop() | ||
| return client, nil | ||
| } | ||
|
|
||
| func (c *deleteRequestsClient) GetAllDeleteRequestsForUser(ctx context.Context, userID string) ([]DeleteRequest, error) { | ||
| if cachedRequests, ok := c.getCachedRequests(userID); ok { | ||
| return cachedRequests, nil | ||
| } | ||
|
|
||
| requests, err := c.getRequestsFromServer(ctx, userID) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| c.mu.Lock() | ||
| defer c.mu.Unlock() | ||
| c.cache[userID] = requests | ||
|
|
||
| return requests, nil | ||
| } | ||
|
|
||
| func (c *deleteRequestsClient) getCachedRequests(userID string) ([]DeleteRequest, bool) { | ||
| c.mu.RLock() | ||
| defer c.mu.RUnlock() | ||
|
|
||
| res, ok := c.cache[userID] | ||
| return res, ok | ||
| } | ||
|
|
||
| func (c *deleteRequestsClient) Stop() { | ||
| close(c.stopChan) | ||
| } | ||
|
|
||
| func (c *deleteRequestsClient) updateLoop() { | ||
| t := time.NewTicker(c.cacheDuration) | ||
| for { | ||
| select { | ||
| case <-t.C: | ||
| c.updateCache() | ||
| case <-c.stopChan: | ||
| return | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (c *deleteRequestsClient) updateCache() { | ||
| userIDs := c.currentUserIDs() | ||
|
|
||
| newCache := make(map[string][]DeleteRequest) | ||
| for _, userID := range userIDs { | ||
| deleteReq, err := c.getRequestsFromServer(context.Background(), userID) | ||
| if err != nil { | ||
| level.Error(log.Logger).Log("msg", "error getting delete requests from the store", "err", err) | ||
| continue | ||
| } | ||
| newCache[userID] = deleteReq | ||
| } | ||
|
|
||
| c.mu.Lock() | ||
| defer c.mu.Unlock() | ||
| c.cache = newCache | ||
| } | ||
|
|
||
| func (c *deleteRequestsClient) currentUserIDs() []string { | ||
| c.mu.RLock() | ||
| defer c.mu.RUnlock() | ||
|
|
||
| userIDs := make([]string, 0, len(c.cache)) | ||
| for userID := range c.cache { | ||
| userIDs = append(userIDs, userID) | ||
| } | ||
|
|
||
| return userIDs | ||
| } | ||
|
|
||
| func (c *deleteRequestsClient) getRequestsFromServer(ctx context.Context, userID string) ([]DeleteRequest, error) { | ||
| req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.url, nil) | ||
| if err != nil { | ||
| level.Error(log.Logger).Log("msg", "error getting delete requests from the store", "err", err) | ||
| return nil, err | ||
| } | ||
|
|
||
| req.Header.Set(orgHeaderKey, userID) | ||
|
|
||
| resp, err := c.httpClient.Do(req) | ||
| if err != nil { | ||
| level.Error(log.Logger).Log("msg", "error getting delete requests from the store", "err", err) | ||
| return nil, err | ||
| } | ||
| defer resp.Body.Close() | ||
|
|
||
| if resp.StatusCode/100 != 2 { | ||
| err := fmt.Errorf("unexpected status code: %d", resp.StatusCode) | ||
| level.Error(log.Logger).Log("msg", "error getting delete requests from the store", "err", err) | ||
| return nil, err | ||
| } | ||
|
|
||
| var deleteRequests []DeleteRequest | ||
| if err := json.NewDecoder(resp.Body).Decode(&deleteRequests); err != nil { | ||
| level.Error(log.Logger).Log("msg", "error marshalling response", "err", err) | ||
| return nil, err | ||
| } | ||
|
|
||
| return deleteRequests, nil | ||
| } |
Oops, something went wrong.