Skip to content

Commit 9a82177

Browse files
odouglas-esodouglas-es
committed
TMP
1 parent 17cd591 commit 9a82177

File tree

12 files changed

+372
-68
lines changed

12 files changed

+372
-68
lines changed

‎api/frontend.proto‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -137,13 +137,22 @@ message DeleteTicketsRequest {
137137
}
138138

139139
message GetIndexedTicketCountRequest{
140-
141140
}
142141

143142
message GetIndexedTicketCountResponse{
144143
int32 count =1;
145144
}
146145

146+
message GetExpiredTicketsRequest{
147+
// The limit on how many ticket Ids to return
148+
int32 limit = 1;
149+
}
150+
151+
message GetExpiredTicketsResponse{
152+
// Expired TicketIds of generated Tickets to be deleted.
153+
repeated string ticket_ids = 1;
154+
}
155+
147156
// The FrontendService implements APIs to manage and query status of a Tickets.
148157
service FrontendService {
149158
// CreateTicket assigns an unique TicketId to the input Ticket and record it in state storage.
@@ -255,4 +264,11 @@ service FrontendService {
255264
get: "/v1/frontendservice/tickets/count"
256265
};
257266
}
267+
268+
// GetExpiredTickets returns the ticket Ids corresponding to tickets outside the valid window
269+
rpc GetExpiredTickets(GetExpiredTicketsRequest) returns (GetExpiredTicketsResponse) {
270+
option (google.api.http) = {
271+
get: "/v1/frontendservice/tickets/expired"
272+
};
273+
}
258274
}

‎api/frontend.swagger.json‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -384,6 +384,46 @@
384384
]
385385
}
386386
},
387+
"/v1/frontendservice/tickets/expired": {
388+
"get": {
389+
"summary": "GetExpiredTickets returns the ticket Ids corresponding to tickets outside the valid window",
390+
"operationId": "FrontendService_GetExpiredTickets",
391+
"responses": {
392+
"200": {
393+
"description": "A successful response.",
394+
"schema": {
395+
"$ref": "#/definitions/openmatchGetExpiredTicketsResponse"
396+
}
397+
},
398+
"404": {
399+
"description": "Returned when the resource does not exist.",
400+
"schema": {
401+
"type": "string",
402+
"format": "string"
403+
}
404+
},
405+
"default": {
406+
"description": "An unexpected error response.",
407+
"schema": {
408+
"$ref": "#/definitions/rpcStatus"
409+
}
410+
}
411+
},
412+
"parameters": [
413+
{
414+
"name": "limit",
415+
"description": "The limit on how many ticket Ids to return",
416+
"in": "query",
417+
"required": false,
418+
"type": "integer",
419+
"format": "int32"
420+
}
421+
],
422+
"tags": [
423+
"FrontendService"
424+
]
425+
}
426+
},
387427
"/v1/frontendservice/tickets/{ticket_id}": {
388428
"get": {
389429
"summary": "GetTicket get the Ticket associated with the specified TicketId.",
@@ -631,6 +671,18 @@
631671
}
632672
}
633673
},
674+
"openmatchGetExpiredTicketsResponse": {
675+
"type": "object",
676+
"properties": {
677+
"ticket_ids": {
678+
"type": "array",
679+
"items": {
680+
"type": "string"
681+
},
682+
"description": "Expired TicketIds of generated Tickets to be deleted."
683+
}
684+
}
685+
},
634686
"openmatchGetIndexedTicketCountResponse": {
635687
"type": "object",
636688
"properties": {

‎internal/app/frontend/frontend_service.go‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -486,3 +486,17 @@ func (s *frontendService) GetIndexedTicketCount(ctx context.Context, _ *pb.GetIn
486486

487487
return resp, nil
488488
}
489+
490+
// GetExpiredTickets returns the ticket Ids corresponding to tickets outside the valid window
491+
func (s *frontendService) GetExpiredTickets(ctx context.Context, req *pb.GetExpiredTicketsRequest) (*pb.GetExpiredTicketsResponse, error) {
492+
ticketIds, err := s.store.GetExpiredTicketIDs(ctx, int(req.GetLimit()))
493+
if err != nil {
494+
return nil, err
495+
}
496+
497+
resp := &pb.GetExpiredTicketsResponse{
498+
TicketIds: ticketIds,
499+
}
500+
501+
return resp, nil
502+
}

‎internal/app/query/cache.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -148,7 +148,7 @@ func updateTicketCache(store statestore.Service, value interface{}) error {
148148
t := time.Now()
149149
previousCount := len(tickets)
150150
// get all indexed tickets within the valid time window
151-
currentAll, err := store.GetIndexedIDSetWithTTL(context.Background())
151+
currentAll, err := store.GetIndexedIDSetWithTTL(context.Background(), 0)
152152
if err != nil {
153153
return err
154154
}

‎internal/app/synchronizer/synchronizer_service.go‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -303,12 +303,6 @@ Registration:
303303
if err != nil {
304304
logger.Errorf("Failed to clean up backfills, %s", err.Error())
305305
}
306-
307-
// TODO: this should probably be enabled much later like after a day of running without it for the transition for Prod
308-
err = s.store.CleanupTickets(ctx)
309-
if err != nil {
310-
logger.Errorf("Failed to clean up tickets, %s", err.Error())
311-
}
312306
}
313307

314308
///////////////////////////////////////

‎internal/statestore/instrumented.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -232,15 +232,15 @@ func (is *instrumentedService) DeleteTicketCompletely(ctx context.Context, id st
232232
}
233233

234234
// GetExpiredTicketIDs gets all ticket IDs which are expired
235-
func (is *instrumentedService) GetExpiredTicketIDs(ctx context.Context) ([]string, error) {
235+
func (is *instrumentedService) GetExpiredTicketIDs(ctx context.Context, limit int) ([]string, error) {
236236
ctx, span := trace.StartSpan(ctx, "statestore/instrumented.GetExpiredTicketIDs")
237237
defer span.End()
238-
return is.s.GetExpiredTicketIDs(ctx)
238+
return is.s.GetExpiredTicketIDs(ctx, 0)
239239
}
240240

241241
// GetIndexedIDSetWithTTL returns the ids of all tickets currently indexed but within a given TTL.
242-
func (is *instrumentedService) GetIndexedIDSetWithTTL(ctx context.Context) (map[string]struct{}, error) {
242+
func (is *instrumentedService) GetIndexedIDSetWithTTL(ctx context.Context, limit int) (map[string]struct{}, error) {
243243
ctx, span := trace.StartSpan(ctx, "statestore/instrumented.GetIndexedIDSetWithTTL")
244244
defer span.End()
245-
return is.s.GetIndexedIDSetWithTTL(ctx)
245+
return is.s.GetIndexedIDSetWithTTL(ctx, 0)
246246
}

‎internal/statestore/public.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -135,10 +135,10 @@ type Service interface {
135135
DeleteTicketCompletely(ctx context.Context, id string) error
136136

137137
// GetExpiredTicketIDs gets all ticket IDs which are expired
138-
GetExpiredTicketIDs(ctx context.Context) ([]string, error)
138+
GetExpiredTicketIDs(ctx context.Context, limit int) ([]string, error)
139139

140140
// GetIndexedIDSetWithTTL returns the ids of all tickets currently indexed but within a given TTL.
141-
GetIndexedIDSetWithTTL(ctx context.Context) (map[string]struct{}, error)
141+
GetIndexedIDSetWithTTL(ctx context.Context, limit int) (map[string]struct{}, error)
142142
}
143143

144144
// New creates a Service based on the configuration.

‎internal/statestore/ticket.go‎

Lines changed: 21 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,11 @@ import (
1818
"context"
1919

2020
"fmt"
21-
"github.com/sirupsen/logrus"
2221
"sync"
2322
"time"
2423

24+
"github.com/sirupsen/logrus"
25+
2526
"github.com/cenkalti/backoff"
2627
"github.com/gomodule/redigo/redis"
2728
"github.com/pkg/errors"
@@ -359,7 +360,7 @@ func (rb *redisBackend) GetIndexedIDSet(ctx context.Context) (map[string]struct{
359360
}
360361

361362
// GetIndexedIDSetWithTTL returns the ids of all tickets currently indexed but within a given TTL.
362-
func (rb *redisBackend) GetIndexedIDSetWithTTL(ctx context.Context) (map[string]struct{}, error) {
363+
func (rb *redisBackend) GetIndexedIDSetWithTTL(ctx context.Context, limit int) (map[string]struct{}, error) {
363364
redisConn, err := rb.redisPool.GetContext(ctx)
364365
if err != nil {
365366
return nil, status.Errorf(codes.Unavailable, "GetIndexedIDSetWithTTL, failed to connect to redis: %v", err)
@@ -378,10 +379,16 @@ func (rb *redisBackend) GetIndexedIDSetWithTTL(ctx context.Context) (map[string]
378379
}
379380

380381
curTimeUnix := curTime.UnixNano()
382+
args := redis.Args{
383+
allTicketsWithTTL, curTimeUnix, "+inf",
384+
}
385+
if limit > 0 {
386+
args = args.Add("LIMIT", 0, limit)
387+
}
381388
// fetch only tickets with a score or ttl ahead of or equal to current time
382-
idsIndexed, err := redis.Strings(redisConn.Do("ZRANGEBYSCORE", allTicketsWithTTL, curTimeUnix, "+inf"))
389+
idsIndexed, err := redis.Strings(redisConn.Do("ZRANGEBYSCORE", args...))
383390
if err != nil {
384-
return nil, status.Errorf(codes.Internal, "error getting all indexed ticket ids %v", err)
391+
return nil, status.Errorf(codes.Internal, "error getting indexed ticket ids %v", err)
385392
}
386393

387394
r := make(map[string]struct{}, len(idsIndexed))
@@ -696,7 +703,7 @@ func (rb *redisBackend) newConstantBackoffStrategy() backoff.BackOff {
696703
}
697704

698705
// GetExpiredTicketIDs gets all ticket IDs which are expired
699-
func (rb *redisBackend) GetExpiredTicketIDs(ctx context.Context) ([]string, error) {
706+
func (rb *redisBackend) GetExpiredTicketIDs(ctx context.Context, limit int) ([]string, error) {
700707
redisConn, err := rb.redisPool.GetContext(ctx)
701708
if err != nil {
702709
return nil, status.Errorf(codes.Unavailable, "GetExpiredBackfillIDs, failed to connect to redis: %v", err)
@@ -708,8 +715,15 @@ func (rb *redisBackend) GetExpiredTicketIDs(ctx context.Context) ([]string, erro
708715
endTimeInt := curTime.Add(-ticketTTL).UnixNano() // anything before the now - ttl
709716
startTimeInt := 0 // unix epoc start time
710717

718+
args := redis.Args{
719+
allTicketsWithTTL, startTimeInt, endTimeInt,
720+
}
721+
if limit > 0 {
722+
args = append(args, "LIMIT", 0, limit)
723+
}
724+
711725
// Filter out ticket IDs that are fetched but not assigned within TTL time (ms).
712-
expiredTicketIds, err := redis.Strings(redisConn.Do("ZRANGEBYSCORE", allTicketsWithTTL, startTimeInt, endTimeInt))
726+
expiredTicketIds, err := redis.Strings(redisConn.Do("ZRANGEBYSCORE", args...))
713727
if err != nil {
714728
return nil, status.Errorf(codes.Internal, "error getting expired tickets %v", err)
715729
}
@@ -778,7 +792,7 @@ func (rb *redisBackend) cleanupTicketsWorker(ctx context.Context, ticketIDsCh <-
778792
}
779793

780794
func (rb *redisBackend) CleanupTickets(ctx context.Context) error {
781-
expiredTicketIDs, err := rb.GetExpiredTicketIDs(ctx)
795+
expiredTicketIDs, err := rb.GetExpiredTicketIDs(ctx, 0)
782796
if err != nil {
783797
return err
784798
}

0 commit comments

Comments
 (0)