diff --git a/api/v1alpha1/vectorpipeline_types.go b/api/v1alpha1/vectorpipeline_types.go index 3c62794..0e2aa7e 100644 --- a/api/v1alpha1/vectorpipeline_types.go +++ b/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 diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 22e1a94..63e065e 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -798,6 +798,11 @@ func (in *VectorPipelineList) DeepCopyObject() runtime.Object { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *VectorPipelineSpec) DeepCopyInto(out *VectorPipelineSpec) { *out = *in + if in.Role != nil { + in, out := &in.Role, &out.Role + *out = new(VectorPipelineRole) + **out = **in + } if in.Sources != nil { in, out := &in.Sources, &out.Sources *out = new(runtime.RawExtension) diff --git a/config/crd/bases/observability.kaasops.io_clustervectorpipelines.yaml b/config/crd/bases/observability.kaasops.io_clustervectorpipelines.yaml index ef086a4..7a01f05 100644 --- a/config/crd/bases/observability.kaasops.io_clustervectorpipelines.yaml +++ b/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 diff --git a/config/crd/bases/observability.kaasops.io_vectorpipelines.yaml b/config/crd/bases/observability.kaasops.io_vectorpipelines.yaml index f87a0eb..28bb273 100644 --- a/config/crd/bases/observability.kaasops.io_vectorpipelines.yaml +++ b/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 diff --git a/docs/design.md b/docs/design.md index 99faae9..4e10502 100644 --- a/docs/design.md +++ b/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 diff --git a/docs/specification.md b/docs/specification.md index 96512ca..2f11884 100644 --- a/docs/specification.md +++ b/docs/specification.md @@ -146,6 +146,10 @@ # VectorPipelineSpec (ClusterVectorPipelineSpec) + + + + diff --git a/helm/charts/vector-operator/crds/observability.kaasops.io_clustervectorpipelines.yaml b/helm/charts/vector-operator/crds/observability.kaasops.io_clustervectorpipelines.yaml index ef086a4..7a01f05 100644 --- a/helm/charts/vector-operator/crds/observability.kaasops.io_clustervectorpipelines.yaml +++ b/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 diff --git a/helm/charts/vector-operator/crds/observability.kaasops.io_vectorpipelines.yaml b/helm/charts/vector-operator/crds/observability.kaasops.io_vectorpipelines.yaml index f87a0eb..28bb273 100644 --- a/helm/charts/vector-operator/crds/observability.kaasops.io_vectorpipelines.yaml +++ b/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 diff --git a/internal/config/config.go b/internal/config/config.go index 27a4944..1e5bb4c 100644 --- a/internal/config/config.go +++ b/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 diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..082fd26 --- /dev/null +++ b/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) +} diff --git a/internal/config/vector_source_types.go b/internal/config/vector_source_types.go index aa0c21d..c2049e4 100644 --- a/internal/config/vector_source_types.go +++ b/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 diff --git a/internal/controller/pipeline_controller.go b/internal/controller/pipeline_controller.go index 1c02a47..71a7386 100644 --- a/internal/controller/pipeline_controller.go +++ b/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{} diff --git a/internal/controller/pipeline_controller_test.go b/internal/controller/pipeline_controller_test.go index 28a624e..c9d60f3 100644 --- a/internal/controller/pipeline_controller_test.go +++ b/internal/controller/pipeline_controller_test.go @@ -24,6 +24,7 @@ import ( . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/reconcile" @@ -86,4 +87,133 @@ var _ = Describe("VectorPipeline Controller", func() { // Example: If you expect a certain status condition after reconciliation, verify it here. }) }) + + Context("When the pipeline pins an unknown spec.role", func() { + It("should be rejected by the apiserver", func() { + bogus := v1alpha1.VectorPipelineRole("custom-role") + pipeline := &v1alpha1.ClusterVectorPipeline{ + ObjectMeta: metav1.ObjectMeta{Name: "bogus-role-pipeline"}, + Spec: v1alpha1.VectorPipelineSpec{Role: &bogus}, + } + + err := k8sClient.Create(context.Background(), pipeline) + Expect(errors.IsInvalid(err)).To(BeTrue(), "expected an Invalid error, got %v", err) + }) + }) + + Context("When resolving the pipeline role", func() { + ctx := context.Background() + aggregatorName := "unselected-aggregator" + // prometheus_remote_write is an agent source type, so inference alone never sends a + // pipeline carrying it to an aggregator. + prwSources := &runtime.RawExtension{Raw: []byte(`{"prw_in":{"type":"prometheus_remote_write","address":"0.0.0.0:9098"}}`)} + sinks := &runtime.RawExtension{Raw: []byte(`{"out":{"type":"blackhole","inputs":["prw_in"]}}`)} + aggregatorRole := v1alpha1.VectorPipelineRoleAggregator + + reconciler := func() *PipelineReconciler { + return &PipelineReconciler{ + Client: k8sClient, + Scheme: k8sClient.Scheme(), + Clientset: clientset, + ConfigCheckTimeout: configCheckTimeout, + VectorAgentEventCh: make(chan event.GenericEvent, 1), + VectorAggregatorsEventCh: make(chan event.GenericEvent, 1), + ClusterVectorAggregatorsEventCh: make(chan event.GenericEvent, 1), + } + } + + // The aggregator selects labels no pipeline here carries, so the reconcile skips + // configcheck (no vector pod to run in envtest) and still resolves the role. + BeforeEach(func() { + aggregator := &v1alpha1.ClusterVectorAggregator{ + ObjectMeta: metav1.ObjectMeta{Name: aggregatorName}, + Spec: v1alpha1.ClusterVectorAggregatorSpec{ + VectorAggregatorCommon: v1alpha1.VectorAggregatorCommon{ + Selector: &v1alpha1.VectorSelectorSpec{MatchLabels: map[string]string{"pipeline": "other"}}, + }, + }, + } + Expect(k8sClient.Create(ctx, aggregator)).To(Succeed()) + }) + + AfterEach(func() { + aggregator := &v1alpha1.ClusterVectorAggregator{ObjectMeta: metav1.ObjectMeta{Name: aggregatorName}} + Expect(k8sClient.Delete(ctx, aggregator)).To(Succeed()) + }) + + It("should use the pinned role instead of the inferred one", func() { + name := types.NamespacedName{Name: "pinned-cluster-pipeline"} + pipeline := &v1alpha1.ClusterVectorPipeline{ + ObjectMeta: metav1.ObjectMeta{Name: name.Name}, + Spec: v1alpha1.VectorPipelineSpec{Role: &aggregatorRole, Sources: prwSources, Sinks: sinks}, + } + Expect(k8sClient.Create(ctx, pipeline)).To(Succeed()) + DeferCleanup(func() { Expect(k8sClient.Delete(ctx, pipeline)).To(Succeed()) }) + + _, err := reconciler().Reconcile(ctx, reconcile.Request{NamespacedName: name}) + Expect(err).NotTo(HaveOccurred()) + + reconciled := &v1alpha1.ClusterVectorPipeline{} + Expect(k8sClient.Get(ctx, name, reconciled)).To(Succeed()) + Expect(reconciled.GetRole()).To(Equal(v1alpha1.VectorPipelineRoleAggregator)) + Expect(reconciled.IsValid()).To(BeTrue()) + }) + + It("should infer the role when spec.role is empty", func() { + name := types.NamespacedName{Name: "unpinned-cluster-pipeline"} + pipeline := &v1alpha1.ClusterVectorPipeline{ + ObjectMeta: metav1.ObjectMeta{Name: name.Name}, + Spec: v1alpha1.VectorPipelineSpec{Sources: prwSources, Sinks: sinks}, + } + Expect(k8sClient.Create(ctx, pipeline)).To(Succeed()) + DeferCleanup(func() { Expect(k8sClient.Delete(ctx, pipeline)).To(Succeed()) }) + + _, err := reconciler().Reconcile(ctx, reconcile.Request{NamespacedName: name}) + Expect(err).NotTo(HaveOccurred()) + + reconciled := &v1alpha1.ClusterVectorPipeline{} + Expect(k8sClient.Get(ctx, name, reconciled)).To(Succeed()) + Expect(reconciled.GetRole()).To(Equal(v1alpha1.VectorPipelineRoleAgent)) + }) + + It("should let a namespaced pipeline pin a network listener to the aggregator", func() { + name := types.NamespacedName{Name: "pinned-namespaced-pipeline", Namespace: "default"} + pipeline := &v1alpha1.VectorPipeline{ + ObjectMeta: metav1.ObjectMeta{Name: name.Name, Namespace: name.Namespace}, + Spec: v1alpha1.VectorPipelineSpec{Role: &aggregatorRole, Sources: prwSources, Sinks: sinks}, + } + Expect(k8sClient.Create(ctx, pipeline)).To(Succeed()) + DeferCleanup(func() { Expect(k8sClient.Delete(ctx, pipeline)).To(Succeed()) }) + + _, err := reconciler().Reconcile(ctx, reconcile.Request{NamespacedName: name}) + Expect(err).NotTo(HaveOccurred()) + + reconciled := &v1alpha1.VectorPipeline{} + Expect(k8sClient.Get(ctx, name, reconciled)).To(Succeed()) + Expect(reconciled.GetRole()).To(Equal(v1alpha1.VectorPipelineRoleAggregator)) + Expect(reconciled.IsValid()).To(BeTrue()) + }) + + It("should reject a host source on a namespaced aggregator pipeline", func() { + name := types.NamespacedName{Name: "host-source-pipeline", Namespace: "default"} + pipeline := &v1alpha1.VectorPipeline{ + ObjectMeta: metav1.ObjectMeta{Name: name.Name, Namespace: name.Namespace}, + Spec: v1alpha1.VectorPipelineSpec{ + Role: &aggregatorRole, + Sources: &runtime.RawExtension{Raw: []byte(`{"logs":{"type":"kubernetes_logs"}}`)}, + Sinks: &runtime.RawExtension{Raw: []byte(`{"out":{"type":"blackhole","inputs":["logs"]}}`)}, + }, + } + Expect(k8sClient.Create(ctx, pipeline)).To(Succeed()) + DeferCleanup(func() { Expect(k8sClient.Delete(ctx, pipeline)).To(Succeed()) }) + + _, err := reconciler().Reconcile(ctx, reconcile.Request{NamespacedName: name}) + Expect(err).NotTo(HaveOccurred()) + + reconciled := &v1alpha1.VectorPipeline{} + Expect(k8sClient.Get(ctx, name, reconciled)).To(Succeed()) + Expect(reconciled.IsValid()).To(BeFalse()) + Expect(*reconciled.Status.Reason).To(ContainSubstring("host source types not allowed")) + }) + }) }) diff --git a/internal/pipeline/hash_test.go b/internal/pipeline/hash_test.go index a0c5a64..0d60e3f 100644 --- a/internal/pipeline/hash_test.go +++ b/internal/pipeline/hash_test.go @@ -106,6 +106,27 @@ func TestGetPipelineHashStableWithoutAnnotation(t *testing.T) { assert.Equal(t, int64(hash.Get(legacy)), *h) } +// Pinning spec.role must change the hash so the pipeline is revalidated and rerouted. +func TestGetPipelineHashTracksRole(t *testing.T) { + spec := func(role *v1alpha1.VectorPipelineRole) *v1alpha1.VectorPipeline { + return &v1alpha1.VectorPipeline{ + ObjectMeta: metav1.ObjectMeta{Name: "p", Namespace: "ns"}, + Spec: v1alpha1.VectorPipelineSpec{ + Role: role, + Sources: &runtime.RawExtension{Raw: []byte(`{"logs":{"type":"kubernetes_logs"}}`)}, + }, + } + } + aggregator := v1alpha1.VectorPipelineRoleAggregator + + h1, err := GetPipelineHash(spec(nil)) + require.NoError(t, err) + h2, err := GetPipelineHash(spec(&aggregator)) + require.NoError(t, err) + + assert.NotEqual(t, *h1, *h2) +} + // Toggling the annotation must read as "changed" (IsPipelineChanged == false) so the // pipeline reconcile propagates to an agent rebuild. func TestIsPipelineChangedDetectsConfigOptimizationToggle(t *testing.T) {
rolePins the pipeline to a Vector role, agent or aggregator. Unset by default, in which case the role is inferred from the source types
sources List of Sources