dokku/plugins/scheduler-k3s/vector_values_test.go

package scheduler_k3s

import (
	"testing"

	"github.com/dokku/dokku/plugins/common"
	"gopkg.in/yaml.v3"
)

// baseVectorValues mirrors the shipped templates/helm-config/vector.yaml
// closely enough to exercise the sink and transform merging
const baseVectorValues = `
customConfig:
  sources:
    kubernetes_logs:
      type: kubernetes_logs
    internal_metrics:
      type: internal_metrics
  transforms:
    kubernetes_container_logs:
      type: remap
      inputs:
        - kubernetes_logs
      source: |
        .container = .kubernetes.container_name
  sinks:
    default_global_sink:
      type: console
      inputs:
        - kubernetes_container_logs
      encoding:
        codec: json
    prom_exporter:
      type: prometheus_exporter
      inputs:
        - internal_metrics
      address: 0.0.0.0:9090
`

func setupVectorValuesTest(t *testing.T, sink string, cronSink string) map[string]interface{} {
	t.Helper()
	t.Setenv("PLUGIN_PATH", "/var/lib/dokku/plugins")
	t.Setenv("PLUGIN_ENABLED_PATH", "/var/lib/dokku/plugins/enabled")
	t.Setenv("DOKKU_LIB_ROOT", t.TempDir())
	t.Setenv("DOKKU_SYSTEM_USER", "root")
	t.Setenv("DOKKU_SYSTEM_GROUP", "root")

	if err := common.PropertySetup("logs"); err != nil {
		t.Fatalf("PropertySetup: %v", err)
	}
	if sink != "" {
		if err := common.PropertyWrite("logs", "--global", "vector-sink", sink); err != nil {
			t.Fatalf("PropertyWrite vector-sink: %v", err)
		}
	}
	if cronSink != "" {
		if err := common.PropertyWrite("logs", "--global", "vector-cron-sink", cronSink); err != nil {
			t.Fatalf("PropertyWrite vector-cron-sink: %v", err)
		}
	}

	values := map[string]interface{}{}
	if err := yaml.Unmarshal([]byte(baseVectorValues), &values); err != nil {
		t.Fatalf("Unmarshal base values: %v", err)
	}

	return values
}

func vectorSinks(t *testing.T, values map[string]interface{}) map[string]interface{} {
	t.Helper()
	customConfig, ok := values["customConfig"].(map[string]interface{})
	if !ok {
		t.Fatal("customConfig is not a map")
	}
	sinks, ok := customConfig["sinks"].(map[string]interface{})
	if !ok {
		t.Fatal("sinks is not a map")
	}
	return sinks
}

func vectorTransforms(t *testing.T, values map[string]interface{}) map[string]interface{} {
	t.Helper()
	customConfig, ok := values["customConfig"].(map[string]interface{})
	if !ok {
		t.Fatal("customConfig is not a map")
	}
	transforms, ok := customConfig["transforms"].(map[string]interface{})
	if !ok {
		t.Fatal("transforms is not a map")
	}
	return transforms
}

func sinkInput(t *testing.T, sinks map[string]interface{}, sinkID string) string {
	t.Helper()
	sink, ok := sinks[sinkID].(map[string]interface{})
	if !ok {
		t.Fatalf("sink %q is missing or not a map", sinkID)
	}

	switch inputs := sink["inputs"].(type) {
	case []string:
		return inputs[0]
	case []interface{}:
		return inputs[0].(string)
	default:
		t.Fatalf("sink %q has unexpected inputs type %T", sinkID, sink["inputs"])
		return ""
	}
}

func TestUpdateVectorValuesNoSinks(t *testing.T) {
	values := setupVectorValuesTest(t, "", "")

	updated, err := updateVectorValues(values)
	if err != nil {
		t.Fatalf("updateVectorValues() error = %v", err)
	}

	sinks := vectorSinks(t, updated)
	if _, ok := sinks[kubernetesDefaultSink]; !ok {
		t.Error("default_global_sink should remain when no sink is configured")
	}
	if _, ok := sinks[kubernetesGlobalSink]; ok {
		t.Error("kubernetes_global_sink should not exist when no sink is configured")
	}
}

// TestUpdateVectorValuesPreservesPromExporter guards against the sinks map
// being replaced wholesale, which previously dropped the prometheus exporter
// that the chart still exposes on port 9090
func TestUpdateVectorValuesPreservesPromExporter(t *testing.T) {
	values := setupVectorValuesTest(t, "console://?encoding[codec]=json", "")

	updated, err := updateVectorValues(values)
	if err != nil {
		t.Fatalf("updateVectorValues() error = %v", err)
	}

	sinks := vectorSinks(t, updated)
	if _, ok := sinks["prom_exporter"]; !ok {
		t.Error("prom_exporter should survive configuring a vector-sink")
	}
	if _, ok := sinks[kubernetesDefaultSink]; ok {
		t.Error("default_global_sink should be superseded by the configured sink")
	}
	if got := sinkInput(t, sinks, kubernetesGlobalSink); got != kubernetesLogsTransform {
		t.Errorf("global sink input = %v, want %v", got, kubernetesLogsTransform)
	}
}

func TestUpdateVectorValuesCronSinkOnly(t *testing.T) {
	values := setupVectorValuesTest(t, "", "console://?encoding[codec]=text")

	updated, err := updateVectorValues(values)
	if err != nil {
		t.Fatalf("updateVectorValues() error = %v", err)
	}

	transforms := vectorTransforms(t, updated)
	if _, ok := transforms[kubernetesRouterTransform]; !ok {
		t.Error("router transform should be added")
	}
	if _, ok := transforms[kubernetesLogsTransform]; !ok {
		t.Error("base container logs transform should be preserved")
	}

	sinks := vectorSinks(t, updated)
	if got := sinkInput(t, sinks, kubernetesCronSink); got != kubernetesCronRemapTransform {
		t.Errorf("cron sink input = %v, want %v", got, kubernetesCronRemapTransform)
	}

	// without a configured plain sink, the console default takes the unmatched branch
	if got := sinkInput(t, sinks, kubernetesDefaultSink); got != kubernetesRouterTransform+"._unmatched" {
		t.Errorf("default sink input = %v, want %v._unmatched", got, kubernetesRouterTransform)
	}
}

func TestUpdateVectorValuesBothSinks(t *testing.T) {
	values := setupVectorValuesTest(t, "console://?encoding[codec]=json", "console://?encoding[codec]=text")

	updated, err := updateVectorValues(values)
	if err != nil {
		t.Fatalf("updateVectorValues() error = %v", err)
	}

	sinks := vectorSinks(t, updated)
	if got := sinkInput(t, sinks, kubernetesGlobalSink); got != kubernetesRouterTransform+"._unmatched" {
		t.Errorf("global sink input = %v, want %v._unmatched", got, kubernetesRouterTransform)
	}
	if got := sinkInput(t, sinks, kubernetesCronSink); got != kubernetesCronRemapTransform {
		t.Errorf("cron sink input = %v, want %v", got, kubernetesCronRemapTransform)
	}
	if _, ok := sinks["prom_exporter"]; !ok {
		t.Error("prom_exporter should survive configuring both sinks")
	}
}

func TestUpdateVectorValuesInvalidCustomConfig(t *testing.T) {
	values := setupVectorValuesTest(t, "console://", "")
	delete(values, "customConfig")

	if _, err := updateVectorValues(values); err == nil {
		t.Fatal("updateVectorValues() expected an error for missing customConfig")
	}
}