From dceec5dd5d29baf6fc9dcef93e0196e7dbf3ea4f Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 6 Nov 2025 07:09:58 +0000 Subject: [PATCH 1/5] Initial plan From fbf0a98d9d0833943479e5b935c759cbbb702f24 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 6 Nov 2025 07:18:53 +0000 Subject: [PATCH 2/5] Add topics pattern matching and metadata-based index naming support Co-authored-by: kevwan <1918356+kevwan@users.noreply.github.com> --- stash/config/config.go | 23 ++++++----- stash/config/topics.go | 59 +++++++++++++++++++++++++++ stash/config/topics_test.go | 40 ++++++++++++++++++ stash/es/index.go | 30 ++++++++++++-- stash/es/index_test.go | 81 +++++++++++++++++++++++++++++++++++++ stash/handler/handler.go | 25 ++++++++++++ stash/stash.go | 26 ++++++++---- 7 files changed, 261 insertions(+), 23 deletions(-) create mode 100644 stash/config/topics.go create mode 100644 stash/config/topics_test.go diff --git a/stash/config/config.go b/stash/config/config.go index e03ce46..8a16509 100644 --- a/stash/config/config.go +++ b/stash/config/config.go @@ -35,17 +35,18 @@ type ( KafkaConf struct { service.ServiceConf - Brokers []string - Group string - Topics []string - Offset string `json:",options=first|last,default=last"` - Conns int `json:",default=1"` - Consumers int `json:",default=8"` - Processors int `json:",default=8"` - MinBytes int `json:",default=10240"` // 10K - MaxBytes int `json:",default=10485760"` // 10M - Username string `json:",optional"` - Password string `json:",optional"` + Brokers []string + Group string + Topics []string + TopicsPattern string `json:",optional"` + Offset string `json:",options=first|last,default=last"` + Conns int `json:",default=1"` + Consumers int `json:",default=8"` + Processors int `json:",default=8"` + MinBytes int `json:",default=10240"` // 10K + MaxBytes int `json:",default=10485760"` // 10M + Username string `json:",optional"` + Password string `json:",optional"` } Cluster struct { diff --git a/stash/config/topics.go b/stash/config/topics.go new file mode 100644 index 0000000..791de79 --- /dev/null +++ b/stash/config/topics.go @@ -0,0 +1,59 @@ +package config + +import ( + "context" + "regexp" + "time" + + "github.com/segmentio/kafka-go" + "github.com/zeromicro/go-zero/core/logx" +) + +// FetchMatchingTopics fetches topics from Kafka that match the given pattern. +// If pattern is empty, it returns the Topics list from the config. +func FetchMatchingTopics(c KafkaConf) ([]string, error) { + if c.TopicsPattern == "" { + return c.Topics, nil + } + + // Compile the regex pattern + pattern, err := regexp.Compile(c.TopicsPattern) + if err != nil { + return nil, err + } + + // Connect to Kafka to fetch metadata + conn, err := kafka.DialContext(context.Background(), "tcp", c.Brokers[0]) + if err != nil { + return nil, err + } + defer conn.Close() + + // Set deadline for the operation + if err := conn.SetDeadline(time.Now().Add(10 * time.Second)); err != nil { + return nil, err + } + + // Fetch partition metadata (includes all topics) + partitions, err := conn.ReadPartitions() + if err != nil { + return nil, err + } + + // Extract unique topics that match the pattern + topicSet := make(map[string]struct{}) + for _, partition := range partitions { + if pattern.MatchString(partition.Topic) { + topicSet[partition.Topic] = struct{}{} + } + } + + // Convert to slice + var topics []string + for topic := range topicSet { + topics = append(topics, topic) + } + + logx.Infof("Matched %d topics with pattern '%s': %v", len(topics), c.TopicsPattern, topics) + return topics, nil +} diff --git a/stash/config/topics_test.go b/stash/config/topics_test.go new file mode 100644 index 0000000..cef59be --- /dev/null +++ b/stash/config/topics_test.go @@ -0,0 +1,40 @@ +package config + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/zeromicro/go-zero/core/service" +) + +func TestFetchMatchingTopics_WithExplicitTopics(t *testing.T) { + conf := KafkaConf{ + Topics: []string{"topic1", "topic2", "topic3"}, + } + + topics, err := FetchMatchingTopics(conf) + assert.NoError(t, err) + assert.Equal(t, []string{"topic1", "topic2", "topic3"}, topics) +} + +func TestFetchMatchingTopics_EmptyPattern(t *testing.T) { + conf := KafkaConf{ + Topics: []string{"topic1", "topic2"}, + TopicsPattern: "", + } + + topics, err := FetchMatchingTopics(conf) + assert.NoError(t, err) + assert.Equal(t, []string{"topic1", "topic2"}, topics) +} + +func TestFetchMatchingTopics_InvalidPattern(t *testing.T) { + conf := KafkaConf{ + ServiceConf: service.ServiceConf{}, + Brokers: []string{"localhost:9092"}, + TopicsPattern: "[invalid-regex", + } + + _, err := FetchMatchingTopics(conf) + assert.Error(t, err) +} diff --git a/stash/es/index.go b/stash/es/index.go index a9f096e..a65c67e 100644 --- a/stash/es/index.go +++ b/stash/es/index.go @@ -122,14 +122,36 @@ func buildIndexFormatter(indexFormat string, loc *time.Location) func(map[string continue } - if val, ok := m[attr]; ok { - vals = append(vals, val) + val := getNestedValue(m, attr) + vals = append(vals, val) + } + return fmt.Sprintf(format, vals...) + } +} + +// getNestedValue retrieves a value from a nested map using dot notation. +// For example, "@metadata.kafka.topic" will traverse m["@metadata"]["kafka"]["topic"] +func getNestedValue(m map[string]interface{}, path string) string { + parts := strings.Split(path, ".") + var current interface{} = m + + for _, part := range parts { + if currentMap, ok := current.(map[string]interface{}); ok { + if val, exists := currentMap[part]; exists { + current = val } else { - vals = append(vals, "") + return "" } + } else { + return "" } - return fmt.Sprintf(format, vals...) } + + // Convert the final value to string + if str, ok := current.(string); ok { + return str + } + return fmt.Sprintf("%v", current) } func formatTime(format string, t time.Time) string { diff --git a/stash/es/index_test.go b/stash/es/index_test.go index 8843381..4a94944 100644 --- a/stash/es/index_test.go +++ b/stash/es/index_test.go @@ -61,6 +61,32 @@ func TestBuildIndexFormatter(t *testing.T) { }, expect: "foo-2020/09/13", }, + { + name: "nested metadata kafka topic", + val: "{.@metadata.kafka.topic}-{{yyyy/MM/dd}}", + attrs: map[string]interface{}{ + "@metadata": map[string]interface{}{ + "kafka": map[string]interface{}{ + "topic": "my-topic", + }, + }, + timestampKey: testTime, + }, + expect: "my-topic-2020/09/13", + }, + { + name: "nested metadata with missing path", + val: "{.@metadata.kafka.missing}-{{yyyy/MM/dd}}", + attrs: map[string]interface{}{ + "@metadata": map[string]interface{}{ + "kafka": map[string]interface{}{ + "topic": "my-topic", + }, + }, + timestampKey: testTime, + }, + expect: "-2020/09/13", + }, } for _, test := range tests { @@ -70,3 +96,58 @@ func TestBuildIndexFormatter(t *testing.T) { }) } } + +func TestGetNestedValue(t *testing.T) { + tests := []struct { + name string + m map[string]interface{} + path string + expect string + }{ + { + name: "simple field", + m: map[string]interface{}{ + "field": "value", + }, + path: "field", + expect: "value", + }, + { + name: "nested field", + m: map[string]interface{}{ + "@metadata": map[string]interface{}{ + "kafka": map[string]interface{}{ + "topic": "my-topic", + }, + }, + }, + path: "@metadata.kafka.topic", + expect: "my-topic", + }, + { + name: "missing field", + m: map[string]interface{}{ + "field": "value", + }, + path: "missing", + expect: "", + }, + { + name: "missing nested field", + m: map[string]interface{}{ + "@metadata": map[string]interface{}{ + "kafka": map[string]interface{}{}, + }, + }, + path: "@metadata.kafka.topic", + expect: "", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + result := getNestedValue(test.m, test.path) + assert.Equal(t, test.expect, result) + }) + } +} diff --git a/stash/handler/handler.go b/stash/handler/handler.go index 590603b..b7b55ba 100644 --- a/stash/handler/handler.go +++ b/stash/handler/handler.go @@ -12,6 +12,7 @@ type MessageHandler struct { writer *es.Writer indexer *es.Index filters []filter.FilterFunc + topic string } func NewHandler(writer *es.Writer, indexer *es.Index) *MessageHandler { @@ -21,6 +22,14 @@ func NewHandler(writer *es.Writer, indexer *es.Index) *MessageHandler { } } +func NewHandlerWithTopic(writer *es.Writer, indexer *es.Index, topic string) *MessageHandler { + return &MessageHandler{ + writer: writer, + indexer: indexer, + topic: topic, + } +} + func (mh *MessageHandler) AddFilters(filters ...filter.FilterFunc) { for _, f := range filters { mh.filters = append(mh.filters, f) @@ -33,6 +42,21 @@ func (mh *MessageHandler) Consume(_ context.Context, _, val string) error { return err } + // Inject Kafka metadata if topic is set + if mh.topic != "" { + if _, exists := m["@metadata"]; !exists { + m["@metadata"] = make(map[string]interface{}) + } + if metadata, ok := m["@metadata"].(map[string]interface{}); ok { + if _, exists := metadata["kafka"]; !exists { + metadata["kafka"] = make(map[string]interface{}) + } + if kafkaMeta, ok := metadata["kafka"].(map[string]interface{}); ok { + kafkaMeta["topic"] = mh.topic + } + } + } + index := mh.indexer.GetIndex(m) for _, proc := range mh.filters { if m = proc(m); m == nil { @@ -47,3 +71,4 @@ func (mh *MessageHandler) Consume(_ context.Context, _, val string) error { return mh.writer.Write(index, string(bs)) } + diff --git a/stash/stash.go b/stash/stash.go index 49aeee5..ed05579 100644 --- a/stash/stash.go +++ b/stash/stash.go @@ -18,10 +18,15 @@ import ( var configFile = flag.String("f", "etc/config.yaml", "Specify the config file") -func toKqConf(c config.KafkaConf) []kq.KqConf { - var ret []kq.KqConf +func toKqConf(c config.KafkaConf) ([]kq.KqConf, error) { + // Fetch topics based on pattern or explicit list + topics, err := config.FetchMatchingTopics(c) + if err != nil { + return nil, err + } - for _, topic := range c.Topics { + var ret []kq.KqConf + for _, topic := range topics { ret = append(ret, kq.KqConf{ ServiceConf: c.ServiceConf, Brokers: c.Brokers, @@ -38,7 +43,7 @@ func toKqConf(c config.KafkaConf) []kq.KqConf { }) } - return ret + return ret, nil } func main() { @@ -71,10 +76,15 @@ func main() { loc = time.Local } indexer := es.NewIndex(client, processor.Output.ElasticSearch.Index, loc) - handle := handler.NewHandler(writer, indexer) - handle.AddFilters(filters...) - handle.AddFilters(filter.AddUriFieldFilter("url", "uri")) - for _, k := range toKqConf(processor.Input.Kafka) { + + kqConfs, err := toKqConf(processor.Input.Kafka) + logx.Must(err) + + for _, k := range kqConfs { + // Create a handler with topic information for metadata injection + handle := handler.NewHandlerWithTopic(writer, indexer, k.Topic) + handle.AddFilters(filters...) + handle.AddFilters(filter.AddUriFieldFilter("url", "uri")) group.Add(kq.MustNewQueue(k, handle)) } } From 673a5fdcc5210df6058a638c5fdf19b19b33f13a Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 6 Nov 2025 07:20:58 +0000 Subject: [PATCH 3/5] Update documentation for topics pattern and metadata-based indexing Co-authored-by: kevwan <1918356+kevwan@users.noreply.github.com> --- readme-cn.md | 26 ++++++++++++++++++++++++++ readme.md | 25 +++++++++++++++++++++++++ stash/etc/config.yaml | 9 +++++++++ 3 files changed, 60 insertions(+) diff --git a/readme-cn.md b/readme-cn.md index 8e6934f..a7f01c0 100644 --- a/readme-cn.md +++ b/readme-cn.md @@ -114,6 +114,21 @@ Offset: first #### Offset 可选last和first，默认为last，表示从头从kafka开始读取数据 +#### Topics 和 TopicsPattern +* **Topics**: 明确指定要消费的Kafka主题列表（例如：`["topic1", "topic2"]`） +* **TopicsPattern**: 使用正则表达式模式匹配多个主题（例如：`"^logs-.*"` 匹配所有以"logs-"开头的主题） +* 可以使用 `Topics` 或 `TopicsPattern`，但如果同时指定两者，`TopicsPattern` 优先 +* 使用模式匹配的示例： +```yaml +Input: + Kafka: + Brokers: + - "localhost:9092" + TopicsPattern: "^app-.*-logs$" + Group: mygroup +``` + + ### Filters @@ -164,6 +179,17 @@ Offset: first #### Index 索引名称，indexname-{{yyyy.MM.dd}}表示年.月.日，也可以用{{yyyy-MM-dd}}，格式自己定义 + 支持字段占位符，如 `{.field}` 来使用消息中的值 + 支持嵌套字段的点表示法，例如 `{.@metadata.kafka.topic}` 来使用Kafka主题名 + 使用基于主题的动态索引示例： +```yaml +Output: + ElasticSearch: + Hosts: + - "http://localhost:9200" + Index: "{.@metadata.kafka.topic}-{{yyyy-MM-dd}}" +``` +这将创建按日期分隔的索引，如 `my-topic-2024-01-15`、`my-topic-2024-01-16` 等。 #### MaxChunkBytes 每次往ES提交的bulk大小，默认是5M，可依据ES的io情况，适当的调整 diff --git a/readme.md b/readme.md index 7e80d32..f9d220c 100644 --- a/readme.md +++ b/readme.md @@ -117,6 +117,20 @@ Offset: first #### Offset * Optional last and first, the default is last, which means read data from kafka from the beginning +#### Topics and TopicsPattern +* **Topics**: Explicit list of Kafka topics to consume from (e.g., `["topic1", "topic2"]`) +* **TopicsPattern**: Regular expression pattern to match multiple topics (e.g., `"^logs-.*"` matches all topics starting with "logs-") +* You can use either `Topics` or `TopicsPattern`, but `TopicsPattern` takes precedence if both are specified +* Example with pattern matching: +```yaml +Input: + Kafka: + Brokers: + - "localhost:9092" + TopicsPattern: "^app-.*-logs$" + Group: mygroup +``` + ### Filters @@ -167,6 +181,17 @@ Offset: first #### Index * Index name, indexname-{{yyyy.MM.dd}} for year. Month. Day, or {{yyyy-MM-dd}}, in your own format +* Supports field placeholders like `{.field}` to use values from the message +* Supports nested fields with dot notation, e.g., `{.@metadata.kafka.topic}` to use the Kafka topic name +* Example with dynamic topic-based index: +```yaml +Output: + ElasticSearch: + Hosts: + - "http://localhost:9200" + Index: "{.@metadata.kafka.topic}-{{yyyy-MM-dd}}" +``` +This creates daily indices like `my-topic-2024-01-15`, `my-topic-2024-01-16`, etc. #### MaxChunkBytes * The size of the bulk submitted to ES each time, default is 5M, can be adjusted according to the ES io situation. diff --git a/stash/etc/config.yaml b/stash/etc/config.yaml index be23a2f..edb6f39 100644 --- a/stash/etc/config.yaml +++ b/stash/etc/config.yaml @@ -5,8 +5,11 @@ Clusters: Brokers: - "172.16.186.16:19092" - "172.16.186.17:19092" + # Use Topics for explicit topic list Topics: - k8slog + # Or use TopicsPattern for regex matching (takes precedence over Topics) + # TopicsPattern: "^app-.*-logs$" Group: pro Consumers: 16 Filters: @@ -43,4 +46,10 @@ Clusters: Hosts: - http://172.16.141.4:9200 - http://172.16.141.5:9200 + # Index can use field placeholders and metadata + # {.event} uses the 'event' field from the message + # {.@metadata.kafka.topic} uses the Kafka topic name + # {{yyyy-MM-dd}} uses date formatting Index: "{.event}-{{yyyy-MM-dd}}" + # Example with topic-based index: + # Index: "{.@metadata.kafka.topic}-{{yyyy-MM-dd}}" From b380abc8e06c87bce30b4ad0f0d3368aa64d86c8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 6 Nov 2025 07:22:42 +0000 Subject: [PATCH 4/5] Improve code quality based on review feedback Co-authored-by: kevwan <1918356+kevwan@users.noreply.github.com> --- stash/config/topics.go | 19 ++++++++++++++----- stash/handler/handler.go | 31 +++++++++++++++++++++---------- 2 files changed, 35 insertions(+), 15 deletions(-) diff --git a/stash/config/topics.go b/stash/config/topics.go index 791de79..02258de 100644 --- a/stash/config/topics.go +++ b/stash/config/topics.go @@ -22,15 +22,24 @@ func FetchMatchingTopics(c KafkaConf) ([]string, error) { return nil, err } - // Connect to Kafka to fetch metadata - conn, err := kafka.DialContext(context.Background(), "tcp", c.Brokers[0]) - if err != nil { - return nil, err + // Try connecting to each broker until one succeeds + var conn *kafka.Conn + var lastErr error + for _, broker := range c.Brokers { + conn, err = kafka.DialContext(context.Background(), "tcp", broker) + if err == nil { + break + } + lastErr = err + } + if conn == nil { + return nil, lastErr } defer conn.Close() // Set deadline for the operation - if err := conn.SetDeadline(time.Now().Add(10 * time.Second)); err != nil { + deadline := 10 * time.Second + if err := conn.SetDeadline(time.Now().Add(deadline)); err != nil { return nil, err } diff --git a/stash/handler/handler.go b/stash/handler/handler.go index b7b55ba..943714c 100644 --- a/stash/handler/handler.go +++ b/stash/handler/handler.go @@ -36,6 +36,25 @@ func (mh *MessageHandler) AddFilters(filters ...filter.FilterFunc) { } } +// ensureMetadataStructure ensures the @metadata.kafka structure exists in the message. +func ensureMetadataStructure(m map[string]interface{}) map[string]interface{} { + if _, exists := m["@metadata"]; !exists { + m["@metadata"] = make(map[string]interface{}) + } + metadata, ok := m["@metadata"].(map[string]interface{}) + if !ok { + return nil + } + if _, exists := metadata["kafka"]; !exists { + metadata["kafka"] = make(map[string]interface{}) + } + kafkaMeta, ok := metadata["kafka"].(map[string]interface{}) + if !ok { + return nil + } + return kafkaMeta +} + func (mh *MessageHandler) Consume(_ context.Context, _, val string) error { var m map[string]interface{} if err := jsoniter.Unmarshal([]byte(val), &m); err != nil { @@ -44,16 +63,8 @@ func (mh *MessageHandler) Consume(_ context.Context, _, val string) error { // Inject Kafka metadata if topic is set if mh.topic != "" { - if _, exists := m["@metadata"]; !exists { - m["@metadata"] = make(map[string]interface{}) - } - if metadata, ok := m["@metadata"].(map[string]interface{}); ok { - if _, exists := metadata["kafka"]; !exists { - metadata["kafka"] = make(map[string]interface{}) - } - if kafkaMeta, ok := metadata["kafka"].(map[string]interface{}); ok { - kafkaMeta["topic"] = mh.topic - } + if kafkaMeta := ensureMetadataStructure(m); kafkaMeta != nil { + kafkaMeta["topic"] = mh.topic } } From f095b9ab4ef72e74c27d8e0be2fc0c3e19464e40 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 6 Nov 2025 07:23:53 +0000 Subject: [PATCH 5/5] Complete implementation with security verification Co-authored-by: kevwan <1918356+kevwan@users.noreply.github.com> --- go.mod | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/go.mod b/go.mod index ea183bb..6c5fd80 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ require ( github.com/json-iterator/go v1.1.12 github.com/olivere/elastic/v7 v7.0.32 github.com/rogpeppe/go-internal v1.14.1 + github.com/segmentio/kafka-go v0.4.47 github.com/stretchr/testify v1.11.1 github.com/vjeantet/jodaTime v1.0.0 github.com/zeromicro/go-queue v1.2.2 @@ -42,7 +43,6 @@ require ( github.com/prometheus/client_model v0.6.1 // indirect github.com/prometheus/common v0.62.0 // indirect github.com/prometheus/procfs v0.15.1 // indirect - github.com/segmentio/kafka-go v0.4.47 // indirect github.com/spaolacci/murmur3 v1.1.0 // indirect go.opentelemetry.io/otel v1.24.0 // indirect go.opentelemetry.io/otel/exporters/jaeger v1.17.0 // indirect