-
Notifications
You must be signed in to change notification settings - Fork 32
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Luiz Pegoraro
committed
Sep 19, 2023
1 parent
6759d35
commit 1e5ece6
Showing
6 changed files
with
198 additions
and
75 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| package consumer | ||
|
|
||
| import ( | ||
| "github.com/go-redis/redis/v8" | ||
| "github.com/orb-community/orb/sinker/redis/producer" | ||
| "go.uber.org/zap" | ||
| ) | ||
|
|
||
| type SinkerKeyExpirationListener interface { | ||
| // Listen to the sinker key expiration | ||
| SubscribeToKeyExpiration() error | ||
| ReceiveMessage(message interface{}) error | ||
| } | ||
|
|
||
| type sinkerKeyExpirationListener struct { | ||
| logger *zap.Logger | ||
| cacheRedisClient redis.Client | ||
| idleProducer producer.SinkIdleProducer | ||
| } |
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,70 @@ | ||
| package producer | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "github.com/go-redis/redis/v8" | ||
| "go.uber.org/zap" | ||
| "time" | ||
| ) | ||
|
|
||
| type SinkerKey struct { | ||
| OwnerID string | ||
| SinkID string | ||
| Size string | ||
| LastActivity time.Time | ||
| } | ||
|
|
||
| func (s *SinkerKey) Encode() map[string]interface{} { | ||
| return map[string]interface{}{ | ||
| "owner_id": s.OwnerID, | ||
| "sink_id": s.SinkID, | ||
| "size": s.Size, | ||
| "last_activity": s.LastActivity.Format(time.RFC3339), | ||
| } | ||
| } | ||
|
|
||
| const DefaultExpiration = 5 * time.Minute | ||
|
|
||
| type SinkerKeyService interface { | ||
| // AddNewSinkerKey Add New Sinker Key with default Expiration of 5 minutes | ||
| AddNewSinkerKey(ctx context.Context, key SinkerKey) error | ||
| // RenewSinkerKey Increment Expiration of Sinker Key | ||
| RenewSinkerKey(ctx context.Context, key SinkerKey) error | ||
| } | ||
|
|
||
| type sinkerKeyService struct { | ||
| logger *zap.Logger | ||
| cacheRepository redis.Client | ||
| } | ||
|
|
||
| func NewSinkerKeyService(logger *zap.Logger, cacheRepository redis.Client) SinkerKeyService { | ||
| return &sinkerKeyService{logger: logger, cacheRepository: cacheRepository} | ||
| } | ||
|
|
||
| // RenewSinkerKey Increment Expiration of Sinker Key | ||
| func (s *sinkerKeyService) RenewSinkerKey(ctx context.Context, key SinkerKey) error { | ||
| // If key does not exist, create new entry | ||
| cmd := s.cacheRepository.Expire(ctx, "orb.sinker", DefaultExpiration) | ||
| if cmd.Err() != nil { | ||
| s.logger.Error("error sending event to sinker event store", zap.Error(cmd.Err())) | ||
| return cmd.Err() | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (s *sinkerKeyService) AddNewSinkerKey(ctx context.Context, sink SinkerKey) error { | ||
| // Create sinker key in redis Hashset with default expiration of 5 minutes | ||
| key := fmt.Sprintf("orb.sinker.key-%s:%s", sink.OwnerID, sink.SinkID) | ||
| cmd := s.cacheRepository.HSet(ctx, key, sink.Encode()) | ||
| if cmd.Err() != nil { | ||
| s.logger.Error("error sending event to sinker event store", zap.Error(cmd.Err())) | ||
| return cmd.Err() | ||
| } | ||
| err := s.RenewSinkerKey(ctx, sink) | ||
| if err != nil { | ||
| s.logger.Error("error setting expiration to sinker event store", zap.Error(cmd.Err())) | ||
| return cmd.Err() | ||
| } | ||
| return nil | ||
| } |
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,58 @@ | ||
| package producer | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/go-redis/redis/v8" | ||
| "go.uber.org/zap" | ||
| "time" | ||
| ) | ||
|
|
||
| type SinkActivityProducer interface { | ||
| // PublishSinkActivity to be used to publish the sink activity to the sinker, mainly used by Otel Bridge Service | ||
| PublishSinkActivity(ctx context.Context, event SinkActivityEvent) error | ||
| } | ||
|
|
||
| type SinkActivityEvent struct { | ||
| OwnerID string | ||
| SinkID string | ||
| State string | ||
| Size string | ||
| Timestamp time.Time | ||
| } | ||
|
|
||
| func (s *SinkActivityEvent) Encode() map[string]interface{} { | ||
| return map[string]interface{}{ | ||
| "owner_id": s.OwnerID, | ||
| "sink_id": s.SinkID, | ||
| "state": s.State, | ||
| "size": s.Size, | ||
| "timestamp": s.Timestamp.Format(time.RFC3339), | ||
| } | ||
| } | ||
|
|
||
| var _ SinkActivityProducer = (*sinkActivityProducer)(nil) | ||
|
|
||
| type sinkActivityProducer struct { | ||
| logger *zap.Logger | ||
| redisStreamClient redis.Client | ||
| } | ||
|
|
||
| func NewSinkActivityProducer(logger *zap.Logger, redisStreamClient redis.Client) SinkActivityProducer { | ||
| return &sinkActivityProducer{logger: logger, redisStreamClient: redisStreamClient} | ||
| } | ||
|
|
||
| // PublishSinkActivity BridgeService will notify stream of sink activity | ||
| func (sp *sinkActivityProducer) PublishSinkActivity(ctx context.Context, event SinkActivityEvent) error { | ||
| const maxLen = 1000 | ||
| record := &redis.XAddArgs{ | ||
| Stream: "orb.sink_activity", | ||
| Values: event.Encode(), | ||
| MaxLen: maxLen, | ||
| Approx: true, | ||
| } | ||
| err := sp.redisStreamClient.XAdd(ctx, record).Err() | ||
| if err != nil { | ||
| sp.logger.Error("error sending event to sinker event store", zap.Error(err)) | ||
| } | ||
| return err | ||
| } |
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,37 @@ | ||
| package producer | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/go-redis/redis/v8" | ||
| "time" | ||
| ) | ||
|
|
||
| type SinkIdleEvent struct { | ||
| OwnerID string | ||
| SinkID string | ||
| State string | ||
| Size string | ||
| Timestamp time.Time | ||
| } | ||
|
|
||
| func (s *SinkIdleEvent) Encode() map[string]interface{} { | ||
| return map[string]interface{}{ | ||
| "owner_id": s.OwnerID, | ||
| "sink_id": s.SinkID, | ||
| "state": s.State, | ||
| "size": s.Size, | ||
| "timestamp": s.Timestamp.Format(time.RFC3339), | ||
| } | ||
| } | ||
|
|
||
| type SinkIdleProducer interface { | ||
| // PublishSinkIdle to be used to publish the sink activity to the sinker, mainly used by Otel Bridge Service | ||
| PublishSinkIdle(ctx context.Context, event SinkIdleEvent) error | ||
| } | ||
|
|
||
| var _ SinkIdleProducer = (*sinkIdleProducer)(nil) | ||
|
|
||
| type sinkIdleProducer struct { | ||
| logger *zap.Logger | ||
| redisStreamClient redis.Client | ||
| } |