Курнат Андрей
108 lines
3.0 KiB
Go
108 lines
3.0 KiB
Go
package kafka
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net"
|
|
"time"
|
|
|
|
"github.com/segmentio/kafka-go/protocol/describeacls"
|
|
)
|
|
|
|
// DescribeACLsRequest represents a request sent to a kafka broker to describe
|
|
// existing ACLs.
|
|
type DescribeACLsRequest struct {
|
|
// Address of the kafka broker to send the request to.
|
|
Addr net.Addr
|
|
|
|
// Filter to filter ACLs on.
|
|
Filter ACLFilter
|
|
}
|
|
|
|
type ACLFilter struct {
|
|
ResourceTypeFilter ResourceType
|
|
ResourceNameFilter string
|
|
// ResourcePatternTypeFilter was added in v1 and is not available prior to that.
|
|
ResourcePatternTypeFilter PatternType
|
|
PrincipalFilter string
|
|
HostFilter string
|
|
Operation ACLOperationType
|
|
PermissionType ACLPermissionType
|
|
}
|
|
|
|
// DescribeACLsResponse represents a response from a kafka broker to an ACL
|
|
// describe request.
|
|
type DescribeACLsResponse struct {
|
|
// The amount of time that the broker throttled the request.
|
|
Throttle time.Duration
|
|
|
|
// Error that occurred while attempting to describe
|
|
// the ACLs.
|
|
Error error
|
|
|
|
// ACL resources returned from the describe request.
|
|
Resources []ACLResource
|
|
}
|
|
|
|
type ACLResource struct {
|
|
ResourceType ResourceType
|
|
ResourceName string
|
|
PatternType PatternType
|
|
ACLs []ACLDescription
|
|
}
|
|
|
|
type ACLDescription struct {
|
|
Principal string
|
|
Host string
|
|
Operation ACLOperationType
|
|
PermissionType ACLPermissionType
|
|
}
|
|
|
|
func (c *Client) DescribeACLs(ctx context.Context, req *DescribeACLsRequest) (*DescribeACLsResponse, error) {
|
|
m, err := c.roundTrip(ctx, req.Addr, &describeacls.Request{
|
|
Filter: describeacls.ACLFilter{
|
|
ResourceTypeFilter: int8(req.Filter.ResourceTypeFilter),
|
|
ResourceNameFilter: req.Filter.ResourceNameFilter,
|
|
ResourcePatternTypeFilter: int8(req.Filter.ResourcePatternTypeFilter),
|
|
PrincipalFilter: req.Filter.PrincipalFilter,
|
|
HostFilter: req.Filter.HostFilter,
|
|
Operation: int8(req.Filter.Operation),
|
|
PermissionType: int8(req.Filter.PermissionType),
|
|
},
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("kafka.(*Client).DescribeACLs: %w", err)
|
|
}
|
|
|
|
res := m.(*describeacls.Response)
|
|
resources := make([]ACLResource, len(res.Resources))
|
|
|
|
for resourceIdx, respResource := range res.Resources {
|
|
descriptions := make([]ACLDescription, len(respResource.ACLs))
|
|
|
|
for descriptionIdx, respDescription := range respResource.ACLs {
|
|
descriptions[descriptionIdx] = ACLDescription{
|
|
Principal: respDescription.Principal,
|
|
Host: respDescription.Host,
|
|
Operation: ACLOperationType(respDescription.Operation),
|
|
PermissionType: ACLPermissionType(respDescription.PermissionType),
|
|
}
|
|
}
|
|
|
|
resources[resourceIdx] = ACLResource{
|
|
ResourceType: ResourceType(respResource.ResourceType),
|
|
ResourceName: respResource.ResourceName,
|
|
PatternType: PatternType(respResource.PatternType),
|
|
ACLs: descriptions,
|
|
}
|
|
}
|
|
|
|
ret := &DescribeACLsResponse{
|
|
Throttle: makeDuration(res.ThrottleTimeMs),
|
|
Error: makeError(res.ErrorCode, res.ErrorMessage),
|
|
Resources: resources,
|
|
}
|
|
|
|
return ret, nil
|
|
}
|