-
Notifications
You must be signed in to change notification settings - Fork 70
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor(kafka_topic): add KafkaTopic repository
- Loading branch information
Showing
9 changed files
with
262 additions
and
567 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
253 changes: 253 additions & 0 deletions
253
internal/sdkprovider/kafkatopicrepository/repository.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,253 @@ | ||
| package kafkatopicrepository | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strings" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/aiven/aiven-go-client" | ||
| "github.com/avast/retry-go" | ||
| ) | ||
|
|
||
| var ( | ||
| initOnce sync.Once | ||
| singleRep *repository | ||
| ) | ||
|
|
||
| const ( | ||
| // defaultV2ListBatchSize the max size of batch to call V2List | ||
| defaultV2ListBatchSize = 100 | ||
|
|
||
| // defaultV2ListRetryDelay V2List caches results, so we retry it by this delay | ||
| defaultV2ListRetryDelay = 5 * time.Second | ||
|
|
||
| // defaultWorkerCallInterval how often worker should run | ||
| defaultWorkerCallInterval = 5 * time.Second | ||
| ) | ||
|
|
||
| // GetTopic returns KafkaTopic | ||
| // Gathers topics to a batch and fetches them by worker with a rate limit (in background) | ||
| func GetTopic(client topicsClient, project, service, topic string) (*aiven.KafkaTopic, error) { | ||
| // Inits global singleton | ||
| initOnce.Do(func() { | ||
| singleRep = newRepository(client) | ||
| }) | ||
|
|
||
| c := make(chan *response, 1) | ||
| req := &request{ | ||
| project: project, | ||
| service: service, | ||
| topic: topic, | ||
| rsp: c, | ||
| } | ||
|
|
||
| // Adds request to the queue | ||
| singleRep.Lock() | ||
| singleRep.queue = append(singleRep.queue, req) | ||
| singleRep.Unlock() | ||
|
|
||
| // Waits response from channel | ||
| rsp := <-c | ||
| close(c) | ||
| return rsp.topic, rsp.err | ||
| } | ||
|
|
||
| // topicsClient interface for unit tests | ||
| type topicsClient interface { | ||
| // List returns existing topics | ||
| List(project, service string) ([]*aiven.KafkaListTopic, error) | ||
| // V2List returns requested topics configuration | ||
| V2List(project, service string, topicNames []string) ([]*aiven.KafkaTopic, error) | ||
| } | ||
|
|
||
| func newRepository(client topicsClient) *repository { | ||
| r := &repository{ | ||
| client: client, | ||
| v2ListBatchSize: defaultV2ListBatchSize, | ||
| v2ListRetryDelay: defaultV2ListRetryDelay, | ||
| workerCallInterval: defaultWorkerCallInterval, | ||
| } | ||
| go r.worker() | ||
| return r | ||
| } | ||
|
|
||
| type repository struct { | ||
| sync.Mutex | ||
| client topicsClient | ||
| queue []*request | ||
| v2ListBatchSize int | ||
| v2ListRetryDelay time.Duration | ||
| workerCallInterval time.Duration | ||
| } | ||
|
|
||
| // worker runs in background and processes the queue | ||
| func (rep *repository) worker() { | ||
| ticker := time.NewTicker(rep.workerCallInterval) | ||
| for { | ||
| <-ticker.C | ||
| b := rep.withdrawal() | ||
| if b != nil { | ||
| rep.fetch(b) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // withdrawal returns the queue and cleans it | ||
| func (rep *repository) withdrawal() map[string]*request { | ||
| rep.Lock() | ||
| defer rep.Unlock() | ||
|
|
||
| if len(rep.queue) == 0 { | ||
| return nil | ||
| } | ||
|
|
||
| q := make(map[string]*request, len(rep.queue)) | ||
| for _, r := range rep.queue { | ||
| q[r.key()] = r | ||
| } | ||
|
|
||
| // todo: use empty() on go 1.21 | ||
| rep.queue = make([]*request, 0) | ||
| return q | ||
| } | ||
|
|
||
| // fetch fetches requested topics | ||
| // 1. groups topics by service | ||
| // 2. removes "not found" topics | ||
| // 3. requests "found" topics (in chunks) | ||
| func (rep *repository) fetch(queue map[string]*request) { | ||
| // Groups topics by service | ||
| rawByService := make(map[string][]*request, 0) | ||
| for i := range queue { | ||
| r := queue[i] | ||
| key := newKey(r.project, r.service) | ||
| rawByService[key] = append(rawByService[key], r) | ||
| } | ||
|
|
||
| // Removes "not found" topics from the list | ||
| // If we call V2List with at least one "not found" topic, | ||
| // it will return 404 for all topics | ||
| foundByService := make([]map[string]*request, 0, len(rawByService)) | ||
| for _, reqs := range rawByService { | ||
| project := reqs[0].project | ||
| service := reqs[0].service | ||
|
|
||
| // Requests all topics for the service | ||
| list, err := rep.client.List(project, service) | ||
| if err != nil { | ||
| for _, r := range reqs { | ||
| r.send(nil, err) | ||
| } | ||
| } | ||
|
|
||
| // Marks existing topics | ||
| exists := make(map[string]bool, len(list)) | ||
| for _, r := range list { | ||
| exists[newKey(project, service, r.TopicName)] = true | ||
| } | ||
|
|
||
| // Send "not found" and gathers "found" | ||
| found := make(map[string]*request, len(reqs)) | ||
| for i := range reqs { | ||
| r := reqs[i] | ||
| if exists[r.key()] { | ||
| found[r.topic] = r | ||
| } else { | ||
| r.send(nil, aiven.Error{Status: 404, Message: "Topic not found"}) | ||
| } | ||
| } | ||
|
|
||
| // If any exists | ||
| if len(found) != 0 { | ||
| foundByService = append(foundByService, found) | ||
| } | ||
| } | ||
|
|
||
| // Fetches "found" topics configuration | ||
| for _, reqs := range foundByService { | ||
| topicNames := make([]string, 0, len(reqs)) | ||
| for _, r := range reqs { | ||
| topicNames = append(topicNames, r.topic) | ||
| } | ||
|
|
||
| // foundByService are grouped by service | ||
| // We can share this values | ||
| project := reqs[topicNames[0]].project | ||
| service := reqs[topicNames[0]].service | ||
|
|
||
| // Slices topic names by rep.v2ListBatchSize | ||
| // because V2List has a limit | ||
| for _, chunk := range chunksBy(topicNames, rep.v2ListBatchSize) { | ||
| // V2List() and Get() do not get info immediately | ||
| // Some retries should be applied | ||
| var list []*aiven.KafkaTopic | ||
| err := retry.Do(func() error { | ||
| rspList, err := rep.client.V2List(project, service, chunk) | ||
|
|
||
| // 404 means that there is "not found" on the list | ||
| // But previous step should have cleaned it, so now this is a fail | ||
| if aiven.IsNotFound(err) { | ||
| return retry.Unrecoverable(fmt.Errorf("kafka topic list has changed")) | ||
| } | ||
|
|
||
| // Something else happened | ||
| // We have retries in the client, so this is bad | ||
| if err != nil { | ||
| return retry.Unrecoverable(err) | ||
| } | ||
|
|
||
| // This is an old cache, we need to retry until succeed | ||
| if len(rspList) != len(chunk) { | ||
| return fmt.Errorf("got %d topics, expected %d. Retrying", len(rspList), len(chunk)) | ||
| } | ||
|
|
||
| list = rspList | ||
| return nil | ||
| }, retry.Delay(rep.v2ListRetryDelay)) | ||
|
|
||
| if err != nil { | ||
| // Send errors | ||
| for _, r := range reqs { | ||
| r.send(nil, err) | ||
| } | ||
| } else { | ||
| // Sends topics | ||
| for _, t := range list { | ||
| reqs[t.TopicName].send(t, nil) | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func newKey(parts ...string) string { | ||
| return strings.Join(parts, "/") | ||
| } | ||
|
|
||
| type response struct { | ||
| topic *aiven.KafkaTopic | ||
| err error | ||
| } | ||
|
|
||
| type request struct { | ||
| project string | ||
| service string | ||
| topic string | ||
| rsp chan *response | ||
| } | ||
|
|
||
| func (r *request) key() string { | ||
| return newKey(r.project, r.service, r.topic) | ||
| } | ||
|
|
||
| func (r *request) send(topic *aiven.KafkaTopic, err error) { | ||
| r.rsp <- &response{topic: topic, err: err} | ||
| } | ||
|
|
||
| func chunksBy[T any](items []T, size int) (chunks [][]T) { | ||
| for size < len(items) { | ||
| items, chunks = items[size:], append(chunks, items[0:size:size]) | ||
| } | ||
| return append(chunks, items) | ||
| } |
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.