Merge pull request #8914 from dokku/feat/vector-cron-sink
Add vector-cron-sink for scheduled cron task output
This commit is contained in:
Jose Diaz-Gonzalez
@@ -2175,7 +2175,10 @@ func installHelmCharts(ctx context.Context, clientset KubernetesClient, shouldIn
|
||||
}
|
||||
|
||||
if chart.ReleaseName == "vector" && chart.Namespace == "vector" {
|
||||
values = updateVectorValues(values)
|
||||
values, err = updateVectorValues(values)
|
||||
if err != nil {
|
||||
return fmt.Errorf("Error updating vector values: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
chartProperties, err := common.PropertyMapGet("scheduler-k3s", "--global", "chart-overrides."+chart.ReleaseName)
|
||||
@@ -2256,26 +2259,129 @@ func installHelmCharts(ctx context.Context, clientset KubernetesClient, shouldIn
|
||||
return nil
|
||||
}
|
||||
|
||||
func updateVectorValues(values map[string]interface{}) map[string]interface{} {
|
||||
value := common.PropertyGet("logs", "--global", "vector-sink")
|
||||
if value == "" {
|
||||
return values
|
||||
const (
|
||||
// kubernetesLogsTransform is the base transform enriching every container log event
|
||||
kubernetesLogsTransform = "kubernetes_container_logs"
|
||||
|
||||
// kubernetesRouterTransform splits container logs into cron and non-cron branches
|
||||
kubernetesRouterTransform = "kubernetes_router"
|
||||
|
||||
// kubernetesCronRemapTransform flattens cron metadata onto the event
|
||||
kubernetesCronRemapTransform = "kubernetes_cron_remap"
|
||||
|
||||
// kubernetesDefaultSink is the console sink shipped in the base values file
|
||||
kubernetesDefaultSink = "default_global_sink"
|
||||
|
||||
// kubernetesGlobalSink receives non-cron logs when a vector-sink is configured
|
||||
kubernetesGlobalSink = "kubernetes_global_sink"
|
||||
|
||||
// kubernetesCronSink receives cron task logs when a vector-cron-sink is configured
|
||||
kubernetesCronSink = "kubernetes_cron_sink"
|
||||
)
|
||||
|
||||
// kubernetesCronRouteTransforms returns the route and remap pair that splits
|
||||
// container logs into cron and non-cron branches.
|
||||
//
|
||||
// Cron pods are labelled app.kubernetes.io/name=cron. The raw cron id exceeds
|
||||
// the Kubernetes label length cap, so it lives in an annotation and is
|
||||
// flattened onto the event here - vector drops any event whose sink template
|
||||
// references a missing field.
|
||||
func kubernetesCronRouteTransforms() map[string]interface{} {
|
||||
return map[string]interface{}{
|
||||
kubernetesRouterTransform: map[string]interface{}{
|
||||
"type": "route",
|
||||
"inputs": []string{kubernetesLogsTransform},
|
||||
"reroute_unmatched": true,
|
||||
"route": map[string]interface{}{
|
||||
logs.CronRouteName: map[string]interface{}{
|
||||
"type": "vrl",
|
||||
"source": `.kubernetes.pod_labels."app.kubernetes.io/name" == "cron"`,
|
||||
},
|
||||
},
|
||||
},
|
||||
kubernetesCronRemapTransform: map[string]interface{}{
|
||||
"type": "remap",
|
||||
"inputs": []string{kubernetesRouterTransform + "." + logs.CronRouteName},
|
||||
"source": ".dokku_app = to_string(.kubernetes.pod_labels.\"app.kubernetes.io/part-of\") ?? \"\"\n" +
|
||||
".dokku_cron_id = to_string(.kubernetes.pod_annotations.\"dokku.com/cron-id\") ?? \"\"",
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// updateVectorValues layers the configured log sinks onto the vector chart
|
||||
// values. Sinks and transforms from the base values file are preserved unless
|
||||
// they are explicitly superseded, so unrelated components such as the
|
||||
// prometheus exporter keep working.
|
||||
func updateVectorValues(values map[string]interface{}) (map[string]interface{}, error) {
|
||||
sinkValue := common.PropertyGet("logs", "--global", "vector-sink")
|
||||
cronSinkValue := common.PropertyGet("logs", "--global", "vector-cron-sink")
|
||||
if sinkValue == "" && cronSinkValue == "" {
|
||||
return values, nil
|
||||
}
|
||||
|
||||
sink, err := logs.SinkValueToConfig("--global", value)
|
||||
if err != nil {
|
||||
return nil
|
||||
customConfig, ok := values["customConfig"].(map[string]interface{})
|
||||
if !ok {
|
||||
return values, errors.New("Missing or invalid customConfig in vector chart values")
|
||||
}
|
||||
|
||||
sink["inputs"] = []string{"kubernetes_container_logs"}
|
||||
|
||||
sinkMap := map[string]interface{}{
|
||||
"kubernetes_global_sink": sink,
|
||||
sinks, ok := customConfig["sinks"].(map[string]interface{})
|
||||
if !ok {
|
||||
sinks = map[string]interface{}{}
|
||||
}
|
||||
|
||||
values["customConfig"].(map[string]interface{})["sinks"] = sinkMap
|
||||
// a configured sink supersedes the console sink from the base values;
|
||||
// without one, the console sink remains the destination for non-cron logs
|
||||
nonCronSinkID := kubernetesDefaultSink
|
||||
if sinkValue != "" {
|
||||
nonCronSinkID = kubernetesGlobalSink
|
||||
delete(sinks, kubernetesDefaultSink)
|
||||
|
||||
return values
|
||||
sink, err := logs.SinkValueToConfig(logs.SinkValueToConfigInput{
|
||||
SinkValue: sinkValue,
|
||||
Inputs: []string{kubernetesLogsTransform},
|
||||
})
|
||||
if err != nil {
|
||||
return values, fmt.Errorf("Error parsing vector-sink: %w", err)
|
||||
}
|
||||
|
||||
// stored as a plain map so that later lookups, and the yaml encoder,
|
||||
// see the same shape as the sinks parsed from the base values file
|
||||
sinks[kubernetesGlobalSink] = map[string]interface{}(sink)
|
||||
}
|
||||
|
||||
if cronSinkValue != "" {
|
||||
transforms, ok := customConfig["transforms"].(map[string]interface{})
|
||||
if !ok {
|
||||
transforms = map[string]interface{}{}
|
||||
}
|
||||
for id, transform := range kubernetesCronRouteTransforms() {
|
||||
transforms[id] = transform
|
||||
}
|
||||
customConfig["transforms"] = transforms
|
||||
|
||||
cronSink, err := logs.SinkValueToConfig(logs.SinkValueToConfigInput{
|
||||
SinkValue: cronSinkValue,
|
||||
Inputs: []string{kubernetesCronRemapTransform},
|
||||
})
|
||||
if err != nil {
|
||||
return values, fmt.Errorf("Error parsing vector-cron-sink: %w", err)
|
||||
}
|
||||
sinks[kubernetesCronSink] = map[string]interface{}(cronSink)
|
||||
|
||||
// cron logs are routed away from the non-cron sink so that each event
|
||||
// lands in exactly one destination
|
||||
nonCronSink, ok := sinks[nonCronSinkID].(map[string]interface{})
|
||||
if !ok {
|
||||
return values, fmt.Errorf("Missing or invalid %s sink in vector chart values", nonCronSinkID)
|
||||
}
|
||||
nonCronSink["inputs"] = []string{kubernetesRouterTransform + "._unmatched"}
|
||||
sinks[nonCronSinkID] = nonCronSink
|
||||
}
|
||||
|
||||
customConfig["sinks"] = sinks
|
||||
values["customConfig"] = customConfig
|
||||
|
||||
return values, nil
|
||||
}
|
||||
|
||||
func installHelperCommands(ctx context.Context) error {
|
||||
|
||||
208
plugins/scheduler-k3s/vector_values_test.go
Normal file
208
plugins/scheduler-k3s/vector_values_test.go
Normal file
@@ -0,0 +1,208 @@
|
||||
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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user