feat: allow pinning the pipeline role with spec.role

api/v1alpha1/vectorpipeline_types.go

@@ -40,6 +40,10 @@ type PipelineSecretBackend struct {
 
 // VectorPipelineSpec defines the desired state of VectorPipeline
 type VectorPipelineSpec struct {
+	// Role pins the pipeline to a Vector role. When empty, the operator infers the role
+	// from the source types.
+	// +kubebuilder:validation:Enum=agent;aggregator
+	Role *VectorPipelineRole `json:"role,omitempty"`
 	// +kubebuilder:pruning:PreserveUnknownFields
 	Sources *runtime.RawExtension `json:"sources,omitempty"`
 	// +kubebuilder:pruning:PreserveUnknownFields

config/crd/bases/observability.kaasops.io_clustervectorpipelines.yaml

@@ -52,6 +52,14 @@ spec:
           spec:
             description: VectorPipelineSpec defines the desired state of VectorPipeline
             properties:
+              role:
+                description: |-
+                  Role pins the pipeline to a Vector role. When empty, the operator infers the role
+                  from the source types.
+                enum:
+                - agent
+                - aggregator
+                type: string
               secret:
                 additionalProperties:
                   description: PipelineSecretBackend declares a named secret backend

config/crd/bases/observability.kaasops.io_vectorpipelines.yaml

@@ -53,6 +53,14 @@ spec:
           spec:
             description: VectorPipelineSpec defines the desired state of VectorPipeline
             properties:
+              role:
+                description: |-
+                  Role pins the pipeline to a Vector role. When empty, the operator infers the role
+                  from the source types.
+                enum:
+                - agent
+                - aggregator
+                type: string
               secret:
                 additionalProperties:
                   description: PipelineSecretBackend declares a named secret backend

docs/design.md

@@ -31,7 +31,7 @@ Specification access to [this](https://github.com/kaasops/vector-operator/blob/m
 The `VectorPipeline` is a namespace-scoped CRD.
 The `VectorPipeline` CRD defines Sources, Transforms and Sinks rules for Vector.
 All `VectorPipelines`, with validated configuration file, added to Vector configuration file.
-The pipeline role is determined automatically based on source types:
+The pipeline role is determined automatically based on source types, or pinned with `spec.role`:
 
 **Agent role** (routed to Vector DaemonSet):
 - Only [kubernetes_logs](https://vector.dev/docs/reference/configuration/sources/kubernetes_logs/) source type is allowed
@@ -44,6 +44,9 @@ The pipeline role is determined automatically based on source types:
 
 ## Restrictions
 - All sources in a pipeline must belong to the same role. Mixing agent and aggregator source types is not allowed.
+- An aggregator pipeline in this scope cannot use a source that reads the node (`kubernetes_logs`,
+  `file`, `journald`, `docker_logs`, `host_metrics`). The aggregator is shared and mounts the host
+  log paths, so such a source would collect data from other namespaces.
 
 ## Specification
 Specification access to [this](https://github.com/kaasops/vector-operator/blob/main/docs/specification.md#vectorpipelinespec-clustervectorpipelinespec) page
@@ -52,7 +55,7 @@ Specification access to [this](https://github.com/kaasops/vector-operator/blob/m
 The `ClusterVectorPipeline` is a cluster-scoped CRD.
 The `ClusterVectorPipeline` CRD defines Sources, Transforms and Sinks rules for Vector.
 All `ClusterVectorPipelines`, with validated configuration file, added to Vector configuration file.
-The pipeline role is determined automatically based on source types:
+The pipeline role is determined automatically based on source types, or pinned with `spec.role`:
 
 **Agent role** (routed to Vector DaemonSet):
 - Supports all agent source types: `kubernetes_logs`, `file`, `journald`, `host_metrics`, `docker_logs`, etc.
@@ -64,6 +67,8 @@ The pipeline role is determined automatically based on source types:
 
 ## Restrictions
 - All sources in a pipeline must belong to the same role. Mixing agent and aggregator source types is not allowed.
+- `spec.role` skips source type classification, so it also covers source types the operator does not
+  recognise and types that belong to both roles. Vector still validates the resulting config.
 
 ## Specification
 Specification access to [this](https://github.com/kaasops/vector-operator/blob/main/docs/specification.md#vectorpipelinespec-clustervectorpipelinespec) page

docs/specification.md

@@ -146,6 +146,10 @@
 
 # VectorPipelineSpec (ClusterVectorPipelineSpec)
 <table>
+    <tr>
+      <td>role</td>
+      <td>Pins the pipeline to a Vector role, <code>agent</code> or <code>aggregator</code>. Unset by default, in which case the role is inferred from the source types</td>
+    </tr>
     <tr>
       <td>sources</td>
       <td>List of Sources</td>

helm/charts/vector-operator/crds/observability.kaasops.io_clustervectorpipelines.yaml

@@ -52,6 +52,14 @@ spec:
           spec:
             description: VectorPipelineSpec defines the desired state of VectorPipeline
             properties:
+              role:
+                description: |-
+                  Role pins the pipeline to a Vector role. When empty, the operator infers the role
+                  from the source types.
+                enum:
+                - agent
+                - aggregator
+                type: string
               secret:
                 additionalProperties:
                   description: PipelineSecretBackend declares a named secret backend

helm/charts/vector-operator/crds/observability.kaasops.io_vectorpipelines.yaml

@@ -53,6 +53,14 @@ spec:
           spec:
             description: VectorPipelineSpec defines the desired state of VectorPipeline
             properties:
+              role:
+                description: |-
+                  Role pins the pipeline to a Vector role. When empty, the operator infers the role
+                  from the source types.
+                enum:
+                - agent
+                - aggregator
+                type: string
               secret:
                 additionalProperties:
                   description: PipelineSecretBackend declares a named secret backend

internal/config/config.go

@@ -36,6 +36,7 @@ import (
 var (
 	ErrNotAllowedSourceType   = errors.New("type kubernetes_logs only allowed")
 	ErrClusterScopeNotAllowed = errors.New("logs from external namespace not allowed")
+	ErrHostSourceNotAllowed   = errors.New("host source types not allowed on a namespaced pipeline")
 )
 
 type VectorConfigParams struct {
@@ -119,10 +120,15 @@ func (c *VectorConfig) MarshalJSON() ([]byte, error) {
 	return jsonByte, nil
 }
 
-func (c *PipelineConfig) VectorRole() (*vectorv1alpha1.VectorPipelineRole, error) {
+// VectorRole infers the role from the source types unless the pipeline pins one. Pinning skips
+// classification, so it also covers types the operator does not know and types in both maps.
+func (c *PipelineConfig) VectorRole(pinned *vectorv1alpha1.VectorPipelineRole) (*vectorv1alpha1.VectorPipelineRole, error) {
 	if len(c.Sources) == 0 {
 		return nil, fmt.Errorf("sources list is empty")
 	}
+	if pinned != nil {
+		return pinned, nil
+	}
 	agentCount := 0
 	aggregatorCount := 0
 	for _, s := range c.Sources {
@@ -147,6 +153,16 @@ func (c *PipelineConfig) VectorRole() (*vectorv1alpha1.VectorPipelineRole, error
 	return nil, fmt.Errorf("unknown vector role")
 }
 
+// ValidateAggregatorSources rejects host sources. Callers apply it to namespaced pipelines only.
+func (c *PipelineConfig) ValidateAggregatorSources() error {
+	for name, s := range c.Sources {
+		if isHostSource(s.Type) {
+			return fmt.Errorf("source %s has type %s: %w", name, s.Type, ErrHostSourceNotAllowed)
+		}
+	}
+	return nil
+}
+
 type SPGroup struct {
 	PipelineName string
 	Namespace    string

internal/config/config_test.go

@@ -0,0 +1,143 @@
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	vectorv1alpha1 "github.com/kaasops/vector-operator/api/v1alpha1"
+)
+
+func role(r vectorv1alpha1.VectorPipelineRole) *vectorv1alpha1.VectorPipelineRole {
+	return &r
+}
+
+func sources(types ...string) map[string]*Source {
+	m := make(map[string]*Source, len(types))
+	for i, t := range types {
+		m[string(rune('a'+i))] = &Source{Type: t}
+	}
+	return m
+}
+
+func TestVectorRole(t *testing.T) {
+	tests := []struct {
+		name    string
+		sources map[string]*Source
+		pinned  *vectorv1alpha1.VectorPipelineRole
+		want    vectorv1alpha1.VectorPipelineRole
+		wantErr string
+	}{
+		{
+			name:    "infers agent",
+			sources: sources(KubernetesLogsType, JournaldType),
+			want:    vectorv1alpha1.VectorPipelineRoleAgent,
+		},
+		{
+			name:    "infers aggregator",
+			sources: sources(KafkaType, SyslogType),
+			want:    vectorv1alpha1.VectorPipelineRoleAggregator,
+		},
+		{
+			// Every agent type also counts as an aggregator (the fallthrough in VectorRole),
+			// so any mix of the two lands on aggregator.
+			name:    "mixed types infer aggregator",
+			sources: sources(KubernetesLogsType, KafkaType),
+			want:    vectorv1alpha1.VectorPipelineRoleAggregator,
+		},
+		{
+			name:    "rejects unclassified type",
+			sources: sources("brand_new_source"),
+			wantErr: "unsupported source type: brand_new_source",
+		},
+		{
+			name:    "pin wins over inference",
+			sources: sources(PrometheusRemoteWriteType),
+			pinned:  role(vectorv1alpha1.VectorPipelineRoleAggregator),
+			want:    vectorv1alpha1.VectorPipelineRoleAggregator,
+		},
+		{
+			name:    "pin accepts an unclassified type",
+			sources: sources("brand_new_source"),
+			pinned:  role(vectorv1alpha1.VectorPipelineRoleAggregator),
+			want:    vectorv1alpha1.VectorPipelineRoleAggregator,
+		},
+		{
+			name:    "pin wins over a mixed inference",
+			sources: sources(KubernetesLogsType, KafkaType),
+			pinned:  role(vectorv1alpha1.VectorPipelineRoleAgent),
+			want:    vectorv1alpha1.VectorPipelineRoleAgent,
+		},
+		{
+			name:    "pin still needs a source",
+			pinned:  role(vectorv1alpha1.VectorPipelineRoleAggregator),
+			wantErr: "sources list is empty",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := &PipelineConfig{Sources: tt.sources}
+			got, err := c.VectorRole(tt.pinned)
+			if tt.wantErr != "" {
+				require.Error(t, err)
+				assert.EqualError(t, err, tt.wantErr)
+				assert.Nil(t, got)
+				return
+			}
+			require.NoError(t, err)
+			require.NotNil(t, got)
+			assert.Equal(t, tt.want, *got)
+		})
+	}
+}
+
+func TestValidateAggregatorSources(t *testing.T) {
+	tests := []struct {
+		name    string
+		sources map[string]*Source
+		wantErr bool
+	}{
+		{name: "network source", sources: sources(KafkaType, SyslogType)},
+		{
+			// The motivating case from #218: an agent-classified listener that a namespaced
+			// pipeline must still be able to pin to an aggregator.
+			name:    "prometheus listeners",
+			sources: sources(PrometheusRemoteWriteType, PrometheusPushgatewayType, PrometheusScrapeType),
+		},
+		{name: "kubernetes events", sources: sources(kubernetesEventsType)},
+		{name: "unclassified source", sources: sources("brand_new_source")},
+		{name: "kubernetes logs", sources: sources(KubernetesLogsType), wantErr: true},
+		{name: "file", sources: sources(FileType), wantErr: true},
+		{name: "journald", sources: sources(JournaldType), wantErr: true},
+		{name: "docker logs", sources: sources(DockerLogsType), wantErr: true},
+		{name: "host metrics", sources: sources(HostMetricsType), wantErr: true},
+		{name: "host source mixed with a network source", sources: sources(KafkaType, KubernetesLogsType), wantErr: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := (&PipelineConfig{Sources: tt.sources}).ValidateAggregatorSources()
+			if tt.wantErr {
+				require.ErrorIs(t, err, ErrHostSourceNotAllowed)
+				return
+			}
+			require.NoError(t, err)
+		})
+	}
+}
+
+// A source type in both maps resolves to agent, so an aggregator can only ever receive it
+// through an explicit pin.
+func TestVectorRolePinReachesAggregatorForDualRoleSource(t *testing.T) {
+	c := &PipelineConfig{Sources: sources(OpenTelemetryType)}
+
+	inferred, err := c.VectorRole(nil)
+	require.NoError(t, err)
+	assert.Equal(t, vectorv1alpha1.VectorPipelineRoleAgent, *inferred)
+
+	pinned, err := c.VectorRole(role(vectorv1alpha1.VectorPipelineRoleAggregator))
+	require.NoError(t, err)
+	assert.Equal(t, vectorv1alpha1.VectorPipelineRoleAggregator, *pinned)
+}

internal/config/vector_source_types.go

@@ -92,6 +92,22 @@ var agentTypes = map[string]struct{}{
 	PrometheusScrapeType:      {},
 }
 
+// Host sources read the node or the container runtime instead of receiving data over the network.
+// An aggregator is shared and mounts the host log paths, so one of these in a namespaced pipeline
+// would collect data from other namespaces.
+var hostSourceTypes = map[string]struct{}{
+	DockerLogsType:     {},
+	FileType:           {},
+	HostMetricsType:    {},
+	JournaldType:       {},
+	KubernetesLogsType: {},
+}
+
+func isHostSource(name string) bool {
+	_, ok := hostSourceTypes[name]
+	return ok
+}
+
 func isAggregator(name string) bool {
 	_, ok := aggregatorTypes[name]
 	return ok

internal/controller/pipeline_controller.go

@@ -319,7 +319,7 @@ func (r *PipelineReconciler) reconcile(ctx context.Context, req ctrl.Request) (c
 		return ctrl.Result{}, nil
 	}
 
-	pipelineVectorRole, err := p.VectorRole()
+	pipelineVectorRole, err := resolvePipelineRole(p, pipelineCR)
 	if err != nil {
 		if err := pipeline.SetFailedStatus(ctx, r.Client, pipelineCR, err.Error(), basePipeline); err != nil {
 			log.Error(err, "Failed to set pipeline status")
@@ -560,6 +560,21 @@ func (r *PipelineReconciler) reconcile(ctx context.Context, req ctrl.Request) (c
 	return ctrl.Result{}, nil
 }
 
+// resolvePipelineRole pins or infers the pipeline role. Validating here rather than in the config
+// builder fails only the offending pipeline, not the aggregator's whole config.
+func resolvePipelineRole(cfg *config.PipelineConfig, p pipeline.Pipeline) (*v1alpha1.VectorPipelineRole, error) {
+	role, err := cfg.VectorRole(p.GetSpec().Role)
+	if err != nil {
+		return nil, err
+	}
+	if *role == v1alpha1.VectorPipelineRoleAggregator && p.GetNamespace() != "" {
+		if err := cfg.ValidateAggregatorSources(); err != nil {
+			return nil, err
+		}
+	}
+	return role, nil
+}
+
 func (r *PipelineReconciler) getPipeline(ctx context.Context, req ctrl.Request) (pipeline pipeline.Pipeline, err error) {
 	if req.Namespace != "" {
 		vp := &v1alpha1.VectorPipeline{}