chore(kbagent): keep probe and task events within the Event message limit (#10511)

Author: weicao

pkg/kbagent/service/event_util.go

@@ -0,0 +1,58 @@
+/*
+Copyright (C) 2022-2026 ApeCloud Co., Ltd
+
+This file is part of KubeBlocks project
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package service
+
+import (
+	"encoding/json"
+)
+
+// maxEventMessageLength is the Kubernetes limit for the Event message field;
+// creating an Event with a longer message is rejected by the apiserver.
+const maxEventMessageLength = 1024
+
+// marshalEventWithSizeLimit marshals an event and, when the encoded form
+// exceeds the Event message limit, shrinks the event's free-text fields —
+// message first (keeping its head, where the root error lives), then output —
+// so that the payload stays valid JSON that event consumers can unmarshal.
+// The message and output pointers must reference fields of the event being
+// marshaled.
+func marshalEventWithSizeLimit(event any, message *string, output *[]byte) ([]byte, error) {
+	const marker = "...(truncated)"
+	for {
+		msg, err := json.Marshal(event)
+		if err != nil {
+			return nil, err
+		}
+		if len(msg) <= maxEventMessageLength {
+			return msg, nil
+		}
+		overflow := len(msg) - maxEventMessageLength
+		switch {
+		case len(*message) > len(marker):
+			cut := min(overflow+len(marker), len(*message))
+			*message = (*message)[:len(*message)-cut] + marker
+		case len(*output) > 0:
+			*output = (*output)[:len(*output)-min(overflow, len(*output))]
+		default:
+			// nothing left to shrink; send as-is and let the apiserver decide
+			return msg, nil
+		}
+	}
+}

pkg/kbagent/service/event_util_test.go

@@ -0,0 +1,80 @@
+/*
+Copyright (C) 2022-2026 ApeCloud Co., Ltd
+
+This file is part of KubeBlocks project
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package service
+
+import (
+	"encoding/json"
+	"strings"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/apecloud/kubeblocks/pkg/kbagent/proto"
+)
+
+var _ = Describe("marshal task event with size limit", func() {
+	It("keeps small task events unchanged", func() {
+		event := proto.TaskEvent{
+			Instance: "comp",
+			Task:     "newReplica",
+			Replica:  "pod-1",
+			Message:  "ok",
+		}
+		msg, err := marshalEventWithSizeLimit(&event, &event.Message, &event.Output)
+		Expect(err).Should(BeNil())
+		plain, err := json.Marshal(&event)
+		Expect(err).Should(BeNil())
+		Expect(msg).Should(Equal(plain))
+	})
+
+	It("truncates a long task failure message but keeps its head and valid JSON", func() {
+		event := proto.TaskEvent{
+			Instance: "comp",
+			Task:     "newReplica",
+			Replica:  "pod-1",
+			Code:     -1,
+			Message:  "replication setup failed: connection refused\n" + strings.Repeat("verbose replication log line\n", 200),
+		}
+		msg, err := marshalEventWithSizeLimit(&event, &event.Message, &event.Output)
+		Expect(err).Should(BeNil())
+		Expect(len(msg)).Should(BeNumerically("<=", maxEventMessageLength))
+
+		var decoded proto.TaskEvent
+		Expect(json.Unmarshal(msg, &decoded)).Should(Succeed())
+		Expect(decoded.Code).Should(Equal(int32(-1)))
+		Expect(decoded.Message).Should(HavePrefix("replication setup failed"))
+		Expect(decoded.Message).Should(HaveSuffix("...(truncated)"))
+	})
+
+	It("shrinks oversized task output while keeping valid JSON", func() {
+		event := proto.TaskEvent{
+			Instance: "comp",
+			Task:     "newReplica",
+			Output:   []byte(strings.Repeat("y", 8192)),
+		}
+		msg, err := marshalEventWithSizeLimit(&event, &event.Message, &event.Output)
+		Expect(err).Should(BeNil())
+		Expect(len(msg)).Should(BeNumerically("<=", maxEventMessageLength))
+
+		var decoded proto.TaskEvent
+		Expect(json.Unmarshal(msg, &decoded)).Should(Succeed())
+		Expect(decoded.Task).Should(Equal("newReplica"))
+	})
+})

pkg/kbagent/service/probe.go

@@ -21,7 +21,6 @@ package service
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"net"
 	"os"
@@ -405,7 +404,7 @@ func (r *probeRunner) launchReportLoop(probe *proto.Probe) {
 				return true
 			}
 
-			msg, err := json.Marshal(event)
+			msg, err := marshalEventWithSizeLimit(&event, &event.Message, &event.Output)
 			if err != nil {
 				log(err, "failed to marshal the probe event", retry, periodically)
 				return true

pkg/kbagent/service/probe_test.go

@@ -24,6 +24,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"strings"
 	"time"
 
 	"github.com/fsnotify/fsnotify"
@@ -285,4 +286,54 @@ var _ = Describe("probe", func() {
 
 		// TODO: more test cases
 	})
+
+	Context("marshal event with size limit", func() {
+		It("keeps small events unchanged", func() {
+			event := proto.ProbeEvent{
+				Probe:   "roleProbe",
+				Code:    0,
+				Output:  []byte("leader"),
+				Message: "ok",
+			}
+			msg, err := marshalEventWithSizeLimit(&event, &event.Message, &event.Output)
+			Expect(err).Should(BeNil())
+			plain, err := json.Marshal(event)
+			Expect(err).Should(BeNil())
+			Expect(msg).Should(Equal(plain))
+		})
+
+		It("truncates a long message but keeps its head and valid JSON", func() {
+			longStderr := "DB manager initialize failed: fatal error config file: open /tools/config/dbctl/components/: no such file or directory\n" +
+				strings.Repeat("usage: dbctl [command] [flags] help text filler\n", 200)
+			event := proto.ProbeEvent{
+				Probe:   "roleProbe",
+				Code:    255,
+				Message: longStderr,
+			}
+			msg, err := marshalEventWithSizeLimit(&event, &event.Message, &event.Output)
+			Expect(err).Should(BeNil())
+			Expect(len(msg)).Should(BeNumerically("<=", maxEventMessageLength))
+
+			var decoded proto.ProbeEvent
+			Expect(json.Unmarshal(msg, &decoded)).Should(Succeed())
+			Expect(decoded.Code).Should(Equal(int32(255)))
+			Expect(decoded.Message).Should(HavePrefix("DB manager initialize failed"))
+			Expect(decoded.Message).Should(HaveSuffix("...(truncated)"))
+		})
+
+		It("shrinks the output when the message alone cannot absorb the overflow", func() {
+			event := proto.ProbeEvent{
+				Probe:  "roleProbe",
+				Code:   255,
+				Output: []byte(strings.Repeat("x", 8192)),
+			}
+			msg, err := marshalEventWithSizeLimit(&event, &event.Message, &event.Output)
+			Expect(err).Should(BeNil())
+			Expect(len(msg)).Should(BeNumerically("<=", maxEventMessageLength))
+
+			var decoded proto.ProbeEvent
+			Expect(json.Unmarshal(msg, &decoded)).Should(Succeed())
+			Expect(decoded.Probe).Should(Equal("roleProbe"))
+		})
+	})
 })

pkg/kbagent/service/task.go

@@ -21,7 +21,6 @@ package service
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"reflect"
 	"slices"
@@ -154,7 +153,7 @@ func (s *taskService) wait(ch chan error) error {
 }
 
 func (s *taskService) notify(task proto.Task, event proto.TaskEvent, sync bool) error {
-	msg, err := json.Marshal(&event)
+	msg, err := marshalEventWithSizeLimit(&event, &event.Message, &event.Output)
 	if err == nil {
 		return util.SendEventWithMessage(&s.logger, "task", string(msg), sync)
 	} else {