-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
acl describe and create + other cleanup and refactoring
- Loading branch information
Showing
38 changed files
with
2,426 additions
and
92 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,173 @@ | ||
| package logical_broker | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
|
|
||
| "github.com/go-logr/logr" | ||
| "github.com/puzpuzpuz/xsync" | ||
| "github.com/rmb938/krouter/pkg/kafka/logical_broker/internal_topics_pb" | ||
| "github.com/rmb938/krouter/pkg/kafka/logical_broker/models" | ||
| "github.com/twmb/franz-go/pkg/kgo" | ||
| "google.golang.org/protobuf/proto" | ||
| ) | ||
|
|
||
| type Authorizer struct { | ||
| log logr.Logger | ||
|
|
||
| kafkaClient *kgo.Client | ||
|
|
||
| aclsSyncOnce sync.Once | ||
| syncedChan chan struct{} | ||
|
|
||
| acls *xsync.MapOf[*models.ACL] | ||
| } | ||
|
|
||
| func NewAuthorizer(log logr.Logger, kafkaClient *kgo.Client) (*Authorizer, error) { | ||
| authorizer := &Authorizer{ | ||
| log: log.WithName("authorizer"), | ||
| kafkaClient: kafkaClient, | ||
| syncedChan: make(chan struct{}), | ||
| acls: xsync.NewMapOf[*models.ACL](), | ||
| } | ||
|
|
||
| return authorizer, nil | ||
| } | ||
|
|
||
| func (c *Authorizer) CreateTestACLs() error { | ||
| acls := []struct { | ||
| Key *internal_topics_pb.ACLMessageKey | ||
| Value *internal_topics_pb.ACLMessageValue | ||
| }{ | ||
| { | ||
| Key: &internal_topics_pb.ACLMessageKey{ | ||
| Operation: internal_topics_pb.ACLMessageKey_OPERATION_WRITE, | ||
| ResourceType: internal_topics_pb.ACLMessageKey_RESOURCE_TYPE_TOPIC, | ||
| PatternType: internal_topics_pb.ACLMessageKey_PATTERN_TYPE_LITERAL, | ||
| ResourceName: "topic-1", | ||
| Principal: "User:user1", | ||
| }, | ||
| Value: &internal_topics_pb.ACLMessageValue{ | ||
| Permission: internal_topics_pb.ACLMessageValue_PERMISSION_ALLOW, | ||
| }, | ||
| }, | ||
| { | ||
| Key: &internal_topics_pb.ACLMessageKey{ | ||
| Operation: internal_topics_pb.ACLMessageKey_OPERATION_READ, | ||
| ResourceType: internal_topics_pb.ACLMessageKey_RESOURCE_TYPE_TOPIC, | ||
| PatternType: internal_topics_pb.ACLMessageKey_PATTERN_TYPE_LITERAL, | ||
| ResourceName: "topic-1", | ||
| Principal: "User:user2", | ||
| }, | ||
| Value: &internal_topics_pb.ACLMessageValue{ | ||
| Permission: internal_topics_pb.ACLMessageValue_PERMISSION_DENY, | ||
| }, | ||
| }, | ||
| { | ||
| Key: &internal_topics_pb.ACLMessageKey{ | ||
| Operation: internal_topics_pb.ACLMessageKey_OPERATION_READ, | ||
| ResourceType: internal_topics_pb.ACLMessageKey_RESOURCE_TYPE_GROUP, | ||
| PatternType: internal_topics_pb.ACLMessageKey_PATTERN_TYPE_PREFIXED, | ||
| ResourceName: "my-group1", | ||
| Principal: "User:user2", | ||
| }, | ||
| Value: &internal_topics_pb.ACLMessageValue{ | ||
| Permission: internal_topics_pb.ACLMessageValue_PERMISSION_DENY, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| for _, acl := range acls { | ||
| aclMessageKeyBytes, err := proto.Marshal(acl.Key) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| aclMessageValueBytes, err := proto.Marshal(acl.Value) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| record := kgo.KeySliceRecord(aclMessageKeyBytes, aclMessageValueBytes) | ||
| record.Topic = InternalTopicAcls | ||
| produceResp := c.kafkaClient.ProduceSync(context.TODO(), record) | ||
| if produceResp.FirstErr() != nil { | ||
| return produceResp.FirstErr() | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (c *Authorizer) WaitSynced() { | ||
| <-c.syncedChan | ||
| } | ||
|
|
||
| func (c *Authorizer) GetAcls(resourceType models.ACLResourceType, resourceName *string, patternType models.ACLPatternType, principal *string, operation models.ACLOperation, permission models.ACLPermission) []*models.ACL { | ||
| var acls []*models.ACL | ||
|
|
||
| c.acls.Range(func(_ string, acl *models.ACL) bool { | ||
| if resourceType != models.ACLResourceTypeAny && resourceType != acl.ResourceType { | ||
| return true | ||
| } | ||
|
|
||
| if resourceName != nil && *resourceName != acl.ResourceName { | ||
| return true | ||
| } | ||
|
|
||
| if patternType != models.ACLPatternTypeAny && patternType != acl.PatternType { | ||
| return true | ||
| } | ||
|
|
||
| if principal != nil && *principal != acl.Principal { | ||
| return true | ||
| } | ||
|
|
||
| if operation != models.ACLOperationAny && operation != acl.Operation { | ||
| return true | ||
| } | ||
|
|
||
| if permission != models.ACLPermissionAny && permission != acl.Permission { | ||
| return true | ||
| } | ||
|
|
||
| acls = append(acls, acl) | ||
| return true | ||
| }) | ||
|
|
||
| return acls | ||
| } | ||
|
|
||
| func (c *Authorizer) CreateAcl(operation models.ACLOperation, resourceType models.ACLResourceType, patternType models.ACLPatternType, resourceName string, principal string, permission models.ACLPermission) error { | ||
|
|
||
| aclKey := &internal_topics_pb.ACLMessageKey{ | ||
| Operation: internal_topics_pb.ACLMessageKey_Operation(operation), | ||
| ResourceType: internal_topics_pb.ACLMessageKey_ResourceType(resourceType), | ||
| PatternType: internal_topics_pb.ACLMessageKey_PatternType(patternType), | ||
| ResourceName: resourceName, | ||
| Principal: principal, | ||
| } | ||
|
|
||
| aclValue := &internal_topics_pb.ACLMessageValue{ | ||
| Permission: internal_topics_pb.ACLMessageValue_Permission(permission), | ||
| } | ||
|
|
||
| aclMessageKeyBytes, err := proto.Marshal(aclKey) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| aclMessageValueBytes, err := proto.Marshal(aclValue) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| record := kgo.KeySliceRecord(aclMessageKeyBytes, aclMessageValueBytes) | ||
| record.Topic = InternalTopicAcls | ||
| produceResp := c.kafkaClient.ProduceSync(context.TODO(), record) | ||
| if produceResp.FirstErr() != nil { | ||
| return produceResp.FirstErr() | ||
| } | ||
|
|
||
| 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,128 @@ | ||
| package logical_broker | ||
|
|
||
| import ( | ||
| "context" | ||
| "os" | ||
| "strconv" | ||
| "sync" | ||
|
|
||
| "github.com/mitchellh/hashstructure/v2" | ||
| "github.com/rmb938/krouter/pkg/kafka/logical_broker/internal_topics_pb" | ||
| "github.com/rmb938/krouter/pkg/kafka/logical_broker/models" | ||
| "github.com/twmb/franz-go/pkg/kgo" | ||
| "google.golang.org/protobuf/proto" | ||
| ) | ||
|
|
||
| const ( | ||
| InternalTopicAcls = "__krouter_acls" | ||
| ) | ||
|
|
||
| func (c *Authorizer) ConsumeAcls() { | ||
| kafkaClient := c.kafkaClient | ||
|
|
||
| highWaterMarks := make(map[int32]int64) | ||
| lowWaterMarks := make(map[int32]int64) | ||
|
|
||
| for { | ||
| fetches := kafkaClient.PollFetches(context.TODO()) | ||
| if fetches.IsClientClosed() { | ||
| c.log.Info("Topic ACL Kafka Client Closed") | ||
| os.Exit(1) | ||
| return | ||
| } | ||
|
|
||
| fetchErrors := fetches.Errors() | ||
| for _, fetchError := range fetches.Errors() { | ||
| c.log.Error(fetchError.Err, "Error polling fetches for acl consumer", "partition", fetchError.Partition) | ||
| } | ||
| if len(fetchErrors) > 0 { | ||
| os.Exit(1) | ||
| } | ||
|
|
||
| fetches.EachTopic(func(topic kgo.FetchTopic) { | ||
| topic.EachPartition(func(partition kgo.FetchPartition) { | ||
| if _, ok := lowWaterMarks[partition.Partition]; !ok { | ||
| lowWaterMarks[partition.Partition] = 0 | ||
| } | ||
| highWaterMarks[partition.Partition] = partition.HighWatermark | ||
| }) | ||
| }) | ||
|
|
||
| for iter := fetches.RecordIter(); !iter.Done(); { | ||
| record := iter.Next() | ||
|
|
||
| lowWaterMarks[record.Partition] = record.Offset + 1 | ||
|
|
||
| key := record.Key | ||
| value := record.Value | ||
|
|
||
| if string(key) != InternalControlKey { | ||
| aclKey := &internal_topics_pb.ACLMessageKey{} | ||
| aclValue := &internal_topics_pb.ACLMessageValue{} | ||
| err := proto.Unmarshal(key, aclKey) | ||
| if err != nil { | ||
| // We don't exit and return here because it'll crash all instances | ||
| // instead we just ignore the message | ||
| c.log.Error(err, "error parsing acl key", "key", key) | ||
| aclKey = nil | ||
| } | ||
|
|
||
| // value may be nil so don't parse if it is | ||
| if value != nil { | ||
| err = proto.Unmarshal(value, aclValue) | ||
| if err != nil { | ||
| // We don't exit and return here because it'll crash all instances | ||
| // instead we just ignore the message | ||
| c.log.Error(err, "error parsing acl value", "value", value) | ||
| aclValue = nil | ||
| } | ||
| } | ||
|
|
||
| // if the key and value was parsed correctly do stuff | ||
| if aclKey != nil && aclValue != nil { | ||
| aclModel := &models.ACL{ | ||
| Operation: models.ACLOperation(aclKey.Operation), | ||
| ResourceType: models.ACLResourceType(aclKey.ResourceType), | ||
| PatternType: models.ACLPatternType(aclKey.PatternType), | ||
| ResourceName: aclKey.ResourceName, | ||
| Principal: aclKey.Principal, | ||
| } | ||
|
|
||
| hashInt, err := hashstructure.Hash(aclModel, hashstructure.FormatV2, nil) | ||
| if err != nil { | ||
| c.log.Error(err, "hashing acl for map key") | ||
| } | ||
| hashKey := strconv.FormatUint(hashInt, 10) | ||
|
|
||
| aclModel.Permission = models.ACLPermission(aclValue.Permission) | ||
|
|
||
| if value == nil { | ||
| // if no value delete it | ||
| c.acls.Delete(hashKey) | ||
| } else { | ||
| c.acls.Store(hashKey, aclModel) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| c.shouldBeSynced(&c.aclsSyncOnce, highWaterMarks, lowWaterMarks) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (c *Authorizer) shouldBeSynced(once *sync.Once, highWaterMarks, lowWaterMarks map[int32]int64) { | ||
| synced := true | ||
| for partitionIndex, highMark := range highWaterMarks { | ||
| if lowMark, ok := lowWaterMarks[partitionIndex]; ok { | ||
| if highMark != lowMark { | ||
| synced = false | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if synced { | ||
| once.Do(func() { | ||
| c.syncedChan <- struct{}{} | ||
| }) | ||
| } | ||
| } |
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.