Uses the new parsing structure for RBAC parsing

apis/v1beta1/config.go

@@ -27,9 +27,11 @@ import (
 	"github.com/go-logr/logr"
 	"gopkg.in/yaml.v3"
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 
 	"github.com/open-telemetry/opentelemetry-operator/internal/components"
 	"github.com/open-telemetry/opentelemetry-operator/internal/components/exporters"
+	"github.com/open-telemetry/opentelemetry-operator/internal/components/processors"
 	"github.com/open-telemetry/opentelemetry-operator/internal/components/receivers"
 )
 
@@ -139,9 +141,43 @@ type Config struct {
 	Service    Service    `json:"service" yaml:"service"`
 }
 
+// getRbacRulesForComponentKinds gets the RBAC Rules for the given ComponentKind(s).
+func (c *Config) getRbacRulesForComponentKinds(logger logr.Logger, componentKinds ...ComponentKind) ([]rbacv1.PolicyRule, error) {
+	var rules []rbacv1.PolicyRule
+	enabledComponents := c.GetEnabledComponents()
+	for _, componentKind := range componentKinds {
+		var retriever components.ParserRetriever
+		var cfg AnyConfig
+		switch componentKind {
+		case KindReceiver:
+			retriever = receivers.ReceiverFor
+			cfg = c.Receivers
+		case KindExporter:
+			retriever = exporters.ParserFor
+			cfg = c.Exporters
+		case KindProcessor:
+			retriever = processors.ProcessorFor
+			if c.Processors == nil {
+				cfg = AnyConfig{}
+			} else {
+				cfg = *c.Processors
+			}
+		}
+		for componentName := range enabledComponents[componentKind] {
+			// TODO: Clean up the naming here and make it simpler to use a retriever.
+			parser := retriever(componentName)
+			if parsedRules, err := parser.GetRBACRules(logger, cfg.Object[componentName]); err != nil {
+				return nil, err
+			} else {
+				rules = append(rules, parsedRules...)
+			}
+		}
+	}
+	return rules, nil
+}
+
 // getPortsForComponentKinds gets the ports for the given ComponentKind(s).
 func (c *Config) getPortsForComponentKinds(logger logr.Logger, componentKinds ...ComponentKind) ([]corev1.ServicePort, error) {
-
 	var ports []corev1.ServicePort
 	enabledComponents := c.GetEnabledComponents()
 	for _, componentKind := range componentKinds {
@@ -187,6 +223,10 @@ func (c *Config) GetAllPorts(logger logr.Logger) ([]corev1.ServicePort, error) {
 	return c.getPortsForComponentKinds(logger, KindReceiver, KindExporter)
 }
 
+func (c *Config) GetAllRbacRules(logger logr.Logger) ([]rbacv1.PolicyRule, error) {
+	return c.getRbacRulesForComponentKinds(logger, KindReceiver, KindExporter, KindProcessor)
+}
+
 // Yaml encodes the current object and returns it as a string.
 func (c *Config) Yaml() (string, error) {
 	var buf bytes.Buffer

internal/components/component.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 
 	"github.com/open-telemetry/opentelemetry-operator/internal/naming"
@@ -34,37 +35,44 @@ var (
 	PortNotFoundErr       = errors.New("port should not be empty")
 )
 
-// PortParser is a function that returns a list of servicePorts given a config of type T.
-type PortParser[T any] func(logger logr.Logger, name string, o *Option, config T) ([]corev1.ServicePort, error)
-
 type PortRetriever interface {
 	GetPortNum() (int32, error)
 	GetPortNumOrDefault(logr.Logger, int32) int32
 }
 
-type Option struct {
+// PortParser is a function that returns a list of servicePorts given a config of type T.
+type PortParser[T any] func(logger logr.Logger, name string, defaultPort *corev1.ServicePort, config T) ([]corev1.ServicePort, error)
+
+// RBACRuleGenerator is a function that generates a list of RBAC Rules given a configuration of type T
+// It's expected that type T is the configuration used by a parser.
+type RBACRuleGenerator[T any] func(logger logr.Logger, config T) ([]rbacv1.PolicyRule, error)
+type PortBuilderOption[T any] func(*Option[T])
+
+type Option[T any] struct {
 	protocol    corev1.Protocol
 	appProtocol *string
 	targetPort  intstr.IntOrString
 	nodePort    int32
 	name        string
 	port        int32
+	portParser  PortParser[T]
+	rbacGen     RBACRuleGenerator[T]
 }
 
-func NewOption(name string, port int32) *Option {
-	return &Option{
+func NewOption[T any](name string, port int32) *Option[T] {
+	return &Option[T]{
 		name: name,
 		port: port,
 	}
 }
 
-func (o *Option) Apply(opts ...PortBuilderOption) {
+func (o *Option[T]) Apply(opts ...PortBuilderOption[T]) {
 	for _, opt := range opts {
 		opt(o)
 	}
 }
 
-func (o *Option) GetServicePort() *corev1.ServicePort {
+func (o *Option[T]) GetServicePort() *corev1.ServicePort {
 	return &corev1.ServicePort{
 		Name:        naming.PortName(o.name, o.port),
 		Port:        o.port,
@@ -75,22 +83,32 @@ func (o *Option) GetServicePort() *corev1.ServicePort {
 	}
 }
 
-type PortBuilderOption func(*Option)
+func WithRBACRuleGenerator[T any](r RBACRuleGenerator[T]) PortBuilderOption[T] {
+	return func(opt *Option[T]) {
+		opt.rbacGen = r
+	}
+}
 
-func WithTargetPort(targetPort int32) PortBuilderOption {
-	return func(opt *Option) {
+func WithPortParser[T any](p PortParser[T]) PortBuilderOption[T] {
+	return func(opt *Option[T]) {
+		opt.portParser = p
+	}
+}
+
+func WithTargetPort[T any](targetPort int32) PortBuilderOption[T] {
+	return func(opt *Option[T]) {
 		opt.targetPort = intstr.FromInt32(targetPort)
 	}
 }
 
-func WithAppProtocol(proto *string) PortBuilderOption {
-	return func(opt *Option) {
+func WithAppProtocol[T any](proto *string) PortBuilderOption[T] {
+	return func(opt *Option[T]) {
 		opt.appProtocol = proto
 	}
 }
 
-func WithProtocol(proto corev1.Protocol) PortBuilderOption {
-	return func(opt *Option) {
+func WithProtocol[T any](proto corev1.Protocol) PortBuilderOption[T] {
+	return func(opt *Option[T]) {
 		opt.protocol = proto
 	}
 }
@@ -137,6 +155,9 @@ type Parser interface {
 	// of the form "name" or "type/name"
 	Ports(logger logr.Logger, name string, config interface{}) ([]corev1.ServicePort, error)
 
+	// GetRBACRules returns the rbac rules for this component
+	GetRBACRules(logger logr.Logger, config interface{}) ([]rbacv1.PolicyRule, error)
+
 	// ParserType returns the type of this parser
 	ParserType() string
 

internal/components/exporters/helpers.go

@@ -38,7 +38,7 @@ func ParserFor(name string) components.Parser {
 		return parser
 	}
 	// We want the default for exporters to fail silently.
-	return components.NewGenericParser[any](components.ComponentType(name), components.UnsetPort, nil)
+	return components.NewGenericParser[any](components.ComponentType(name), components.UnsetPort)
 }
 
 var (

internal/components/generic_parser.go

@@ -20,6 +20,7 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/mitchellh/mapstructure"
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 )
 
 var (
@@ -30,8 +31,20 @@ var (
 // functionality to idempotent functions.
 type GenericParser[T any] struct {
 	name       string
+	option     *Option[T]
 	portParser PortParser[T]
-	option     *Option
+	rbacGen    RBACRuleGenerator[T]
+}
+
+func (g *GenericParser[T]) GetRBACRules(logger logr.Logger, config interface{}) ([]rbacv1.PolicyRule, error) {
+	if g.rbacGen == nil {
+		return nil, nil
+	}
+	var parsed T
+	if err := mapstructure.Decode(config, &parsed); err != nil {
+		return nil, err
+	}
+	return g.rbacGen(logger, parsed)
 }
 
 func (g *GenericParser[T]) Ports(logger logr.Logger, name string, config interface{}) ([]corev1.ServicePort, error) {
@@ -42,7 +55,7 @@ func (g *GenericParser[T]) Ports(logger logr.Logger, name string, config interfa
 	if err := mapstructure.Decode(config, &parsed); err != nil {
 		return nil, err
 	}
-	return g.portParser(logger, name, g.option, parsed)
+	return g.portParser(logger, name, g.option.GetServicePort(), parsed)
 }
 
 func (g *GenericParser[T]) ParserType() string {
@@ -53,8 +66,8 @@ func (g *GenericParser[T]) ParserName() string {
 	return fmt.Sprintf("__%s", g.name)
 }
 
-func NewGenericParser[T any](name string, port int32, parser PortParser[T], opts ...PortBuilderOption) *GenericParser[T] {
-	o := NewOption(name, port)
+func NewGenericParser[T any](name string, port int32, opts ...PortBuilderOption[T]) *GenericParser[T] {
+	o := NewOption[T](name, port)
 	o.Apply(opts...)
-	return &GenericParser[T]{name: name, portParser: parser, option: o}
+	return &GenericParser[T]{name: name, portParser: o.portParser, rbacGen: o.rbacGen, option: o}
 }

internal/components/generic_parser_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 
 	"github.com/open-telemetry/opentelemetry-operator/internal/components"
 )
@@ -41,7 +42,7 @@ func TestGenericParser_GetPorts(t *testing.T) {
 	tests := []testCase[*components.SingleEndpointConfig]{
 		{
 			name: "valid config with endpoint",
-			g:    components.NewGenericParser[*components.SingleEndpointConfig]("test", 0, components.ParseSingleEndpoint),
+			g:    components.NewGenericParser[*components.SingleEndpointConfig]("test", 0, components.WithPortParser(components.ParseSingleEndpoint)),
 			args: args{
 				logger: logr.Discard(),
 				config: map[string]interface{}{
@@ -58,7 +59,7 @@ func TestGenericParser_GetPorts(t *testing.T) {
 		},
 		{
 			name: "valid config with listen_address",
-			g:    components.NewGenericParser[*components.SingleEndpointConfig]("test", 0, components.ParseSingleEndpoint),
+			g:    components.NewGenericParser[*components.SingleEndpointConfig]("test", 0, components.WithPortParser(components.ParseSingleEndpoint)),
 			args: args{
 				logger: logr.Discard(),
 				config: map[string]interface{}{
@@ -75,7 +76,7 @@ func TestGenericParser_GetPorts(t *testing.T) {
 		},
 		{
 			name: "valid config with listen_address with option",
-			g:    components.NewGenericParser[*components.SingleEndpointConfig]("test", 0, components.ParseSingleEndpoint, components.WithProtocol(corev1.ProtocolUDP)),
+			g:    components.NewGenericParser[*components.SingleEndpointConfig]("test", 0, components.WithPortParser(components.ParseSingleEndpoint), components.WithProtocol[*components.SingleEndpointConfig](corev1.ProtocolUDP)),
 			args: args{
 				logger: logr.Discard(),
 				config: map[string]interface{}{
@@ -93,7 +94,7 @@ func TestGenericParser_GetPorts(t *testing.T) {
 		},
 		{
 			name: "invalid config with no endpoint or listen_address",
-			g:    components.NewGenericParser[*components.SingleEndpointConfig]("test", 0, components.ParseSingleEndpoint),
+			g:    components.NewGenericParser[*components.SingleEndpointConfig]("test", 0, components.WithPortParser(components.ParseSingleEndpoint)),
 			args: args{
 				logger: logr.Discard(),
 				config: map[string]interface{}{},
@@ -113,3 +114,89 @@ func TestGenericParser_GetPorts(t *testing.T) {
 		})
 	}
 }
+
+func TestGenericParser_GetRBACRules(t *testing.T) {
+	type args struct {
+		logger logr.Logger
+		config interface{}
+	}
+	type testCase[T any] struct {
+		name    string
+		g       *components.GenericParser[T]
+		args    args
+		want    []rbacv1.PolicyRule
+		wantErr assert.ErrorAssertionFunc
+	}
+
+	rbacGenFunc := func(logger logr.Logger, config components.SingleEndpointConfig) ([]rbacv1.PolicyRule, error) {
+		if config.Endpoint == "" && config.ListenAddress == "" {
+			return nil, fmt.Errorf("either endpoint or listen_address must be specified")
+		}
+		return []rbacv1.PolicyRule{
+			{
+				APIGroups: []string{""},
+				Resources: []string{"pods"},
+				Verbs:     []string{"get", "list"},
+			},
+		}, nil
+	}
+
+	tests := []testCase[components.SingleEndpointConfig]{
+		{
+			name: "valid config with endpoint",
+			g:    components.NewGenericParser[components.SingleEndpointConfig]("test", components.UnsetPort, components.WithRBACRuleGenerator(rbacGenFunc)),
+			args: args{
+				logger: logr.Discard(),
+				config: map[string]interface{}{
+					"endpoint": "http://localhost:8080",
+				},
+			},
+			want: []rbacv1.PolicyRule{
+				{
+					APIGroups: []string{""},
+					Resources: []string{"pods"},
+					Verbs:     []string{"get", "list"},
+				},
+			},
+			wantErr: assert.NoError,
+		},
+		{
+			name: "valid config with listen_address",
+			g:    components.NewGenericParser[components.SingleEndpointConfig]("test", components.UnsetPort, components.WithRBACRuleGenerator(rbacGenFunc)),
+			args: args{
+				logger: logr.Discard(),
+				config: map[string]interface{}{
+					"listen_address": "0.0.0.0:9090",
+				},
+			},
+			want: []rbacv1.PolicyRule{
+				{
+					APIGroups: []string{""},
+					Resources: []string{"pods"},
+					Verbs:     []string{"get", "list"},
+				},
+			},
+			wantErr: assert.NoError,
+		},
+		{
+			name: "invalid config with no endpoint or listen_address",
+			g:    components.NewGenericParser[components.SingleEndpointConfig]("test", components.UnsetPort, components.WithRBACRuleGenerator(rbacGenFunc)),
+			args: args{
+				logger: logr.Discard(),
+				config: map[string]interface{}{},
+			},
+			want:    nil,
+			wantErr: assert.Error,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := tt.g.GetRBACRules(tt.args.logger, tt.args.config)
+			if !tt.wantErr(t, err, fmt.Sprintf("GetRBACRules(%v, %v)", tt.args.logger, tt.args.config)) {
+				return
+			}
+			assert.Equalf(t, tt.want, got, "GetRBACRules(%v, %v)", tt.args.logger, tt.args.config)
+		})
+	}
+}