This repository was archived by the owner on Aug 23, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 107
Clear cache api #555
Merged
Merged
Clear cache api #555
Changes from all commits
Commits
Show all changes
24 commits
Select commit
Hold shift + click to select a range
096fc81
Add Delete method to accounting
replay 87735cd
unit test for metric deletion in accounting
replay 618ce5b
Add Delete method to chunk cache
replay 985c598
add tests to test the cache delete methods
replay 4f43d6a
add api endpoints to use cache delete
replay 19b0650
add tests for api endpoints and cluster interaction
replay 62d27f8
fix deleting keys from LRU
replay aaaf014
fix unused variables in tests
replay ae3a918
remove total cache size counter and rely on stat
replay cf06a16
add variable names to interfaces
replay e3a4980
remove type CCDelMetricResult
replay 9458c60
do not need to make slice
replay 510a235
test if cache still works after resetting it
replay 13b8874
minor changes according to PR comments
replay 979bea4
improve tests
replay 3a21035
optimize lru delete
Dieterbe f54d2ba
sensible param names for mockCache
Dieterbe 7322dd1
small CCache.Add optimization
Dieterbe 29fa432
comment
Dieterbe b9b25bc
bit more post-Reset testing
Dieterbe 5f49ca0
small comment tweaks
Dieterbe db096fd
remove one pointer per cached metric
replay dfa58d3
remove unnecessary method thisNode()
replay b168558
fix bug in deleting by name and fix tests
replay File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,107 @@ | ||
| package api | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "sync" | ||
|
|
||
| "github.com/grafana/metrictank/api/middleware" | ||
| "github.com/grafana/metrictank/api/models" | ||
| "github.com/grafana/metrictank/api/response" | ||
| "github.com/grafana/metrictank/cluster" | ||
| "github.com/raintank/worldping-api/pkg/log" | ||
| ) | ||
|
|
||
| func (s *Server) ccacheDelete(ctx *middleware.Context, req models.CCacheDelete) { | ||
| res := models.CCacheDeleteResp{} | ||
| code := 200 | ||
|
|
||
| if req.Propagate { | ||
| res.Peers = s.ccacheDeletePropagate(ctx.Req.Context(), &req) | ||
| for _, peer := range res.Peers { | ||
| if peer.Errors > 0 { | ||
| code = 500 | ||
| } | ||
| } | ||
| } | ||
|
|
||
| fullFlush := false | ||
| for _, pattern := range req.Patterns { | ||
| if pattern == "**" { | ||
| fullFlush = true | ||
| } | ||
| } | ||
|
|
||
| if fullFlush { | ||
| delSeries, delArchives := s.Cache.Reset() | ||
| res.DeletedSeries += delSeries | ||
| res.DeletedArchives += delArchives | ||
| } else { | ||
| for _, pattern := range req.Patterns { | ||
| nodes, err := s.MetricIndex.Find(req.OrgId, pattern, 0) | ||
| if err != nil { | ||
| if res.Errors == 0 { | ||
| res.FirstError = err.Error() | ||
| } | ||
| res.Errors += 1 | ||
| code = 500 | ||
| } else { | ||
| for _, node := range nodes { | ||
| for _, def := range node.Defs { | ||
| delSeries, delArchives := s.Cache.DelMetric(def.NameWithTags()) | ||
| res.DeletedSeries += delSeries | ||
| res.DeletedArchives += delArchives | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| response.Write(ctx, response.NewJson(code, res, "")) | ||
| } | ||
|
|
||
| func (s *Server) ccacheDeletePropagate(ctx context.Context, req *models.CCacheDelete) map[string]models.CCacheDeleteResp { | ||
| // we never want to propagate more than once to avoid loops | ||
| req.Propagate = false | ||
|
|
||
| peers := cluster.Manager.MemberList() | ||
| peerResults := make(map[string]models.CCacheDeleteResp) | ||
| var mu sync.Mutex | ||
| var wg sync.WaitGroup | ||
| for _, peer := range peers { | ||
| if peer.IsLocal() { | ||
| continue | ||
| } | ||
| wg.Add(1) | ||
| go func(peer cluster.Node) { | ||
| mu.Lock() | ||
| defer mu.Unlock() | ||
| peerResults[peer.GetName()] = s.ccacheDeleteRemote(ctx, req, peer) | ||
| wg.Done() | ||
| }(peer) | ||
| } | ||
| wg.Wait() | ||
|
|
||
| return peerResults | ||
| } | ||
|
|
||
| func (s *Server) ccacheDeleteRemote(ctx context.Context, req *models.CCacheDelete, peer cluster.Node) models.CCacheDeleteResp { | ||
| var res models.CCacheDeleteResp | ||
|
|
||
| log.Debug("HTTP metricDelete calling %s/ccache/delete", peer.GetName()) | ||
| buf, err := peer.Post(ctx, "ccacheDeleteRemote", "/ccache/delete", *req) | ||
| if err != nil { | ||
| log.Error(4, "HTTP ccacheDelete error querying %s/ccache/delete: %q", peer.GetName(), err) | ||
| res.FirstError = err.Error() | ||
| res.Errors++ | ||
| return res | ||
| } | ||
|
|
||
| err = json.Unmarshal(buf, &res) | ||
| if err != nil { | ||
| log.Error(4, "HTTP ccacheDelete error unmarshaling body from %s/ccache/delete: %q", peer.GetName(), err) | ||
| res.FirstError = err.Error() | ||
| res.Errors++ | ||
| } | ||
|
|
||
| return res | ||
| } |
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,203 @@ | ||
| package api | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "encoding/json" | ||
| "net/http" | ||
| "net/http/httptest" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/grafana/metrictank/api/models" | ||
| "github.com/grafana/metrictank/api/response" | ||
| "github.com/grafana/metrictank/cluster" | ||
| "github.com/grafana/metrictank/conf" | ||
| "github.com/grafana/metrictank/idx/memory" | ||
| "github.com/grafana/metrictank/mdata" | ||
| "github.com/grafana/metrictank/mdata/cache" | ||
| "gopkg.in/raintank/schema.v1" | ||
| ) | ||
|
|
||
| func newSrv(delSeries, delArchives int, key string) (*Server, *cache.MockCache) { | ||
| srv, _ := NewServer() | ||
| srv.RegisterRoutes() | ||
|
|
||
| mdata.SetSingleAgg(conf.Avg, conf.Min, conf.Max) | ||
| mdata.SetSingleSchema(conf.NewRetentionMT(10, 100, 600, 10, true)) | ||
|
|
||
| store := mdata.NewDevnullStore() | ||
| srv.BindBackendStore(store) | ||
|
|
||
| mockCache := cache.NewMockCache() | ||
| mockCache.DelMetricSeries = delSeries | ||
| mockCache.DelMetricArchives = delArchives | ||
| metrics := mdata.NewAggMetrics(store, mockCache, false, 0, 0, 0) | ||
| srv.BindMemoryStore(metrics) | ||
| srv.BindCache(mockCache) | ||
|
|
||
| metricIndex := memory.New() | ||
| metricIndex.AddOrUpdate( | ||
| &schema.MetricData{ | ||
| Id: "123", | ||
| OrgId: 1, | ||
| Name: key, | ||
| Metric: key, | ||
| Interval: 10, | ||
| Value: 1, | ||
| }, | ||
| 0, | ||
| ) | ||
| srv.BindMetricIndex(metricIndex) | ||
| return srv, mockCache | ||
| } | ||
|
|
||
| func TestMetricDelete(t *testing.T) { | ||
| cluster.Init("default", "test", time.Now(), "http", 6060) | ||
|
|
||
| delSeries := 1 | ||
| delArchives := 3 | ||
| testKey := "test.key" | ||
|
|
||
| srv, cache := newSrv(delSeries, delArchives, testKey) | ||
| req, _ := json.Marshal(models.CCacheDelete{ | ||
| Patterns: []string{"test.*"}, | ||
| OrgId: 1, | ||
| Propagate: false, | ||
| }) | ||
|
|
||
| ts := httptest.NewServer(srv.Macaron) | ||
| defer ts.Close() | ||
|
|
||
| res, err := http.Post(ts.URL+"/ccache/delete", "application/json", bytes.NewReader(req)) | ||
| if err != nil { | ||
| t.Fatalf("There was an error in the request: %s", err) | ||
| } | ||
|
|
||
| respParsed := models.CCacheDeleteResp{} | ||
| buf := new(bytes.Buffer) | ||
| buf.ReadFrom(res.Body) | ||
| json.Unmarshal(buf.Bytes(), &respParsed) | ||
|
|
||
| if len(cache.DelMetricKeys) != 1 || cache.DelMetricKeys[0] != testKey { | ||
| t.Fatalf("Expected that key %s has been deleted, but it has not", testKey) | ||
| } | ||
|
|
||
| if respParsed.DeletedSeries != delSeries || respParsed.DeletedArchives != delArchives { | ||
| t.Fatalf("Expected %d series and %d archives to get deleted, but got %d and %d", delSeries, delArchives, respParsed.DeletedSeries, respParsed.DeletedArchives) | ||
| } | ||
| } | ||
|
|
||
| func TestMetricDeleteWithErrorInPropagation(t *testing.T) { | ||
| manager := cluster.InitMock() | ||
|
|
||
| // define how many series/archives are getting deleted by peer 0 | ||
| resp := models.CCacheDeleteResp{ | ||
| DeletedSeries: 1, | ||
| DeletedArchives: 1, | ||
| Errors: 1, | ||
| } | ||
|
|
||
| respEncoded := response.NewJson(500, resp, "") | ||
| buf, _ := respEncoded.Body() | ||
| manager.Peers = append(manager.Peers, cluster.NewMockNode(false, "0", buf)) | ||
|
|
||
| // define how many series/archives are going to get deleted by this server | ||
| delSeries := 1 | ||
| delArchives := 3 | ||
| testKey := "test.key" | ||
|
|
||
| srv, _ := newSrv(delSeries, delArchives, testKey) | ||
| req, err := json.Marshal(models.CCacheDelete{ | ||
| Patterns: []string{"test.*"}, | ||
| OrgId: 1, | ||
| Propagate: true, | ||
| }) | ||
| if err != nil { | ||
| t.Fatalf("Unexpected error when marshaling json: %s", err) | ||
| } | ||
|
|
||
| ts := httptest.NewServer(srv.Macaron) | ||
| defer ts.Close() | ||
|
|
||
| res, err := http.Post(ts.URL+"/ccache/delete", "application/json", bytes.NewReader(req)) | ||
| if err != nil { | ||
| t.Fatalf("There was an error in the request: %s", err) | ||
| } | ||
|
|
||
| expectedCode := 500 | ||
| if res.StatusCode != expectedCode { | ||
| buf2 := new(bytes.Buffer) | ||
| buf2.ReadFrom(res.Body) | ||
| respParsed := models.CCacheDeleteResp{} | ||
| json.Unmarshal(buf2.Bytes(), &respParsed) | ||
| t.Fatalf("Expected status code %d, but got %d:\n%+v", expectedCode, res.StatusCode, respParsed) | ||
| } | ||
| } | ||
|
|
||
| func TestMetricDeletePropagation(t *testing.T) { | ||
| manager := cluster.InitMock() | ||
|
|
||
| expectedDeletedSeries, expectedDeletedArchives := 0, 0 | ||
| for _, peer := range []string{"Peer1", "Peer2", "Peer3"} { | ||
| // define how many series/archives are getting deleted by this peer | ||
| resp := models.CCacheDeleteResp{ | ||
| DeletedSeries: 2, | ||
| DeletedArchives: 5, | ||
| } | ||
| expectedDeletedSeries += resp.DeletedSeries | ||
| expectedDeletedArchives += resp.DeletedArchives | ||
| respEncoded := response.NewJson(200, resp, "") | ||
| buf, _ := respEncoded.Body() | ||
| manager.Peers = append(manager.Peers, cluster.NewMockNode(false, peer, buf)) | ||
| } | ||
|
|
||
| // define how many series/archives are going to get deleted by this server | ||
| delSeries := 1 | ||
| delArchives := 3 | ||
| testKey := "test.key" | ||
|
|
||
| // add up how many series/archives are expected to be deleted | ||
| expectedDeletedSeries += delSeries | ||
| expectedDeletedArchives += delArchives | ||
|
|
||
| srv, cache := newSrv(delSeries, delArchives, testKey) | ||
| req, err := json.Marshal(models.CCacheDelete{ | ||
| Patterns: []string{"test.*"}, | ||
| OrgId: 1, | ||
| Propagate: true, | ||
| }) | ||
| if err != nil { | ||
| t.Fatalf("Unexpected error when marshaling json: %s", err) | ||
| } | ||
|
|
||
| ts := httptest.NewServer(srv.Macaron) | ||
| defer ts.Close() | ||
|
|
||
| res, err := http.Post(ts.URL+"/ccache/delete", "application/json", bytes.NewReader(req)) | ||
| if err != nil { | ||
| t.Fatalf("There was an error in the request: %s", err) | ||
| } | ||
|
|
||
| buf2 := new(bytes.Buffer) | ||
| buf2.ReadFrom(res.Body) | ||
| respParsed := models.CCacheDeleteResp{} | ||
| json.Unmarshal(buf2.Bytes(), &respParsed) | ||
|
|
||
| if len(cache.DelMetricKeys) != 1 || cache.DelMetricKeys[0] != testKey { | ||
| t.Fatalf("Expected that key %s has been deleted, but it has not", testKey) | ||
| } | ||
|
|
||
| deletedArchives := respParsed.DeletedArchives | ||
| deletedSeries := respParsed.DeletedSeries | ||
| for _, peer := range respParsed.Peers { | ||
| deletedArchives += peer.DeletedArchives | ||
| deletedSeries += peer.DeletedSeries | ||
| } | ||
|
|
||
| if deletedSeries != expectedDeletedSeries || deletedArchives != expectedDeletedArchives { | ||
| t.Fatalf( | ||
| "Expected %d series and %d archives to get deleted, but got %d and %d", | ||
| expectedDeletedSeries, expectedDeletedArchives, respParsed.DeletedSeries, respParsed.DeletedArchives, | ||
| ) | ||
| } | ||
| } | ||
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
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
i'm confused about why we tell the cache to return these fake, made up numbers and then check them, while we also check the real numbers above. can this be simplified ? for example have the mockcache return that deleted series == deleted metric keys, and deleted archives is maybe pinned to 3x the number of deleted keys? does this idea make sense? if not, add a comment somewhere that explains this please
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is only testing the request handler, not the cache. It's testing one request with propagation disabled (
TestMetricDelete) and one with propagation enabled (TestMetricDeletePropagation). I'm not sure how creating additional rules likearchives = 3 * serieswould simplify things, that would rather just make it more complicated because in order to understand the test a reader would then first need to be aware of this rule.Maybe I should rename the tests to
TestMetricDeleteRequestHandlerWithoutPropagation()andTestMetricDeleteRequestHandlerWithPropagation?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
it's confusing because the mock cache claims to have deleted a number series and archives that has nothing to do with the delete request we actually issued on it. the numbers don't seem to make sense. we assert that it only received 1 metric key delete , then how could any cache have deleted 3 series if there was only 1 key? i know it's a mock and not a real cache, but the numbers should make more sense I think.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
that would clear things up if the DelMetric method took patterns.
but according to the argument names or the interface function and its implementations
(and the docs for CCache.DelMetric) it only takes 1 metric key, not a pattern.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
i found a bug while looking at that just now: b168558