When a configuration error occurs, the change should be rejected and an error message should be provided, rather than generating an empty configuration.

[ahfuzhang]

I use CRD ClusterVectorPipeline to use jsonline sinks. Yaml is like this:

apiVersion: observability.kaasops.io/v1alpha1
kind: ClusterVectorPipeline
metadata:
  name: vlogs
  namespace: {{ .Release.Namespace }}
spec:
  sinks:
    vlogs:
      # see: https://vector.dev/docs/reference/configuration/sinks/http/
      # jsonline mode
      type: http
      inputs:
        - container-logs-transform
      uri: {{ .Values.vector.endpoint }}
      batch:
        timeout_secs: 5
        max_bytes: 20971520
      buffer:
        type: memory
        max_events: 15000
        when_full: drop_newest
      method: POST
      encoding:
        codec: json
      request:
        retry_attempts: 3
        compression: {{ .Values.vector.compression | default "none" }}
      framing:
        method: newline_delimited

But it not works.
I check the secret:

KUBECONFIG=~/my-test-k8s.yaml kubectl get secret vector-agent -n logging -o json | jq -r '.data."agent.json"' | base64 -d

The config content is:

{
  "api": {
    "address": "0.0.0.0:8686"
  },
  "data_dir": "/vector-data-dir",
  "sinks": {
    "internalMetricsSink": {
      "inputs": [
        "internalMetricsSource"
      ],
      "type": "prometheus_exporter"
    }
  },
  "sources": {
    "internalMetricsSource": {
      "type": "internal_metrics"
    }
  },
  "transforms": {}
}

Maybe, I wrote wrong yaml, but why I didn't got any feedback.

I check the code from github.com/kaasops/vector-operator/internal/pipeline/pipeline.go:
see:

vector-operator/internal/pipeline/pipeline.go

Line 80 in f205135

vp.IsValid() &&

	if filter.Scope == AllPipelines || filter.Scope == NamespacedPipeline {
 
		if filter.Scope == NamespacedPipeline && filter.Namespace == "" {
			return nil, fmt.Errorf("namespace not specified")
		}
		vps, err := GetVectorPipelines(ctx, client)
		if err != nil {
			return nil, err
		}
		if len(vps) != 0 {
			for _, vp := range vps {
				if !vp.IsDeleted() &&
					vp.IsValid() &&  // when vp.IsValid() is false, this line should return error
					vp.GetRole() == filter.Role &&
					(filter.Scope == AllPipelines || vp.Namespace == filter.Namespace) &&
					MatchLabels(matchLabels, vp.Labels) {
					validPipelines = append(validPipelines, vp.DeepCopy())
				}
			}
		}
	}
       // when len(validPipelines)!=len(vps), should return error detail, never use default config

And this line:
see

vector-operator/api/v1alpha1/vectorpipeline_types.go

Line 34 in f205135

Sinks *runtime.RawExtension `json:"sinks,omitempty"`

// VectorPipelineSpec defines the desired state of VectorPipeline
type VectorPipelineSpec struct {
	// +kubebuilder:pruning:PreserveUnknownFields
	Sources *runtime.RawExtension `json:"sources,omitempty"`
	// +kubebuilder:pruning:PreserveUnknownFields
	Transforms *runtime.RawExtension `json:"transforms,omitempty"`
	// +kubebuilder:pruning:PreserveUnknownFields
	Sinks *runtime.RawExtension `json:"sinks,omitempty"`
}

Why not use a struct to replace *runtime.RawExtension ?
There needs to be a mechanism to detect configuration errors early and provide feedback.

Thanks.