Add Kafka topic pattern matching and metadata-based dynamic index naming

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

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情况，适当的调整

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.

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 {

stash/config/topics.go

@@ -0,0 +1,68 @@
+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
+	}
+
+	// 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
+	deadline := 10 * time.Second
+	if err := conn.SetDeadline(time.Now().Add(deadline)); 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
+}

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)
+}

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 {

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)
+		})
+	}
+}

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}}"