support yyyy-MM-dd in index format

@@ -8,6 +8,8 @@ require (
 	github.com/json-iterator/go v1.1.10
 	github.com/mailru/easyjson v0.7.3 // indirect
 	github.com/olivere/elastic v6.2.34+incompatible
+	github.com/stretchr/testify v1.5.1
 	github.com/tal-tech/go-queue v1.0.1
 	github.com/tal-tech/go-zero v1.0.13
+	github.com/vjeantet/jodaTime v1.0.0
 )

@@ -215,6 +215,7 @@ github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3
 github.com/spf13/pflag v1.0.1/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
@@ -236,6 +237,8 @@ github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1
 github.com/tmc/grpc-websocket-proxy v0.0.0-20171017195756-830351dc03c6/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/urfave/cli v1.20.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA=
 github.com/urfave/cli v1.22.4/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
+github.com/vjeantet/jodaTime v1.0.0 h1:Fq2K9UCsbTFtKbHpe/L7C57XnSgbZ5z+gyGpn7cTE3s=
+github.com/vjeantet/jodaTime v1.0.0/go.mod h1:gA+i8InPfZxL1ToHaDpzi6QT/npjl3uPlcV4cxDNerI=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c h1:u40Z8hqBAAQyv+vATcGgV0YCnDjqSL7/q/JyPhhJSPk=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
 github.com/xdg/stringprep v1.0.0 h1:d9X0esnoa3dFsV0FG35rAT0RIhYFlPq7MiP+DW89La0=

@@ -58,7 +58,9 @@ Processors:
       Hosts:
         - "172.16.141.4:9200"
         - "172.16.141.5:9200"
-      Index: {.event}-2006.01.02
+      # {.event}是json输入的event属性值
+      # {{yyyy-MM-dd}}表示日期，比如2020-09-09
+      Index: {.event}-{{yyyy-MM-dd}}
 ```
 
 ### 微信交流群

@@ -58,5 +58,7 @@ Processors:
       Hosts:
         - "172.16.141.4:9200"
         - "172.16.141.5:9200"
-      Index: {.event}-2006.01.02
+      # {.event} is the value of the json attribute from input
+      # {{yyyy-MM-dd}} means date, like 2020-09-09
+      Index: {.event}-{{yyyy-MM-dd}}
 ```

@@ -12,16 +12,21 @@ import (
 	"github.com/tal-tech/go-zero/core/lang"
 	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/core/syncx"
+	"github.com/vjeantet/jodaTime"
 )
 
 const (
 	timestampFormat = "2006-01-02T15:04:05.000Z"
 	timestampKey    = "@timestamp"
+	leftBrace       = '{'
+	rightBrace      = '}'
+	dot             = '.'
 )
 
 const (
 	stateNormal = iota
 	stateWrap
+	stateVar
 	stateDot
 )
 
@@ -39,29 +44,9 @@ type (
 )
 
 func NewIndex(client *elastic.Client, indexFormat string, loc *time.Location) *Index {
-	var formatter func(map[string]interface{}) string
-	format, attrs := getFormat(indexFormat)
-	if len(attrs) > 0 {
-		formatter = func(m map[string]interface{}) string {
-			var vals []interface{}
-			for _, attr := range attrs {
-				if val, ok := m[attr]; ok {
-					vals = append(vals, val)
-				} else {
-					vals = append(vals, "")
-				}
-			}
-			return getTime(m).In(loc).Format(fmt.Sprintf(format, vals...))
-		}
-	} else {
-		formatter = func(m map[string]interface{}) string {
-			return getTime(m).In(loc).Format(format)
-		}
-	}
-
 	return &Index{
 		client:      client,
-		indexFormat: formatter,
+		indexFormat: buildIndexFormatter(indexFormat, loc),
 		indices:     make(map[string]lang.PlaceholderType),
 		sharedCalls: syncx.NewSharedCalls(),
 	}
@@ -121,6 +106,36 @@ func (idx *Index) ensureIndex(index string) error {
 	return err
 }
 
+func buildIndexFormatter(indexFormat string, loc *time.Location) func(map[string]interface{}) string {
+	format, attrs, timePos := getFormat(indexFormat)
+	if len(attrs) == 0 {
+		return func(m map[string]interface{}) string {
+			return format
+		}
+	}
+
+	return func(m map[string]interface{}) string {
+		var vals []interface{}
+		for i, attr := range attrs {
+			if i == timePos {
+				vals = append(vals, formatTime(attr, getTime(m).In(loc)))
+				continue
+			}
+
+			if val, ok := m[attr]; ok {
+				vals = append(vals, val)
+			} else {
+				vals = append(vals, "")
+			}
+		}
+		return fmt.Sprintf(format, vals...)
+	}
+}
+
+func formatTime(format string, t time.Time) string {
+	return jodaTime.Format(format, t)
+}
+
 func getTime(m map[string]interface{}) time.Time {
 	if ti, ok := m[timestampKey]; ok {
 		if ts, ok := ti.(string); ok {
@@ -133,32 +148,59 @@ func getTime(m map[string]interface{}) time.Time {
 	return time.Now()
 }
 
-func getFormat(indexFormat string) (format string, attrs []string) {
+func getFormat(indexFormat string) (format string, attrs []string, timePos int) {
 	var state = stateNormal
 	var builder strings.Builder
 	var keyBuf strings.Builder
+	timePos = -1
+	writeHolder := func() {
+		if keyBuf.Len() > 0 {
+			attrs = append(attrs, keyBuf.String())
+			keyBuf.Reset()
+			builder.WriteString("%s")
+		}
+	}
+
 	for _, ch := range indexFormat {
-		switch ch {
-		case '{':
-			state = stateWrap
-		case '.':
-			if state == stateWrap {
-				state = stateDot
-			} else {
+		switch state {
+		case stateNormal:
+			switch ch {
+			case leftBrace:
+				state = stateWrap
+			default:
 				builder.WriteRune(ch)
 			}
-		case '}':
-			state = stateNormal
-			if keyBuf.Len() > 0 {
-				attrs = append(attrs, keyBuf.String())
-				builder.WriteString("%s")
+		case stateWrap:
+			switch ch {
+			case leftBrace:
+				state = stateVar
+			case dot:
+				state = stateDot
+				keyBuf.Reset()
+			case rightBrace:
+				state = stateNormal
+				timePos = len(attrs)
+				writeHolder()
+			default:
+				keyBuf.WriteRune(ch)
 			}
-		default:
-			if state == stateDot {
+		case stateVar:
+			switch ch {
+			case rightBrace:
+				state = stateWrap
+			default:
 				keyBuf.WriteRune(ch)
-			} else {
-				builder.WriteRune(ch)
 			}
+		case stateDot:
+			switch ch {
+			case rightBrace:
+				state = stateNormal
+				writeHolder()
+			default:
+				keyBuf.WriteRune(ch)
+			}
+		default:
+			builder.WriteRune(ch)
 		}
 	}
 

+package es
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const testTime = "2020-09-13T08:22:29.294Z"
+
+func TestBuildIndexFormatter(t *testing.T) {
+	tests := []struct {
+		name   string
+		val    string
+		attrs  map[string]interface{}
+		expect string
+	}{
+		{
+			name:   "plain text only",
+			val:    "yyyy/MM/dd",
+			expect: "yyyy/MM/dd",
+		},
+		{
+			name:   "time only",
+			val:    "{{yyyy/MM/dd}}",
+			expect: time.Now().Format("2006/01/02"),
+		},
+		{
+			name: "attr without time",
+			val:  "{.event}",
+			attrs: map[string]interface{}{
+				"event": "foo",
+			},
+			expect: "foo",
+		},
+		{
+			name: "attr with time",
+			val:  "{.event}-{{yyyy/MM/dd}}",
+			attrs: map[string]interface{}{
+				"event":      "foo",
+				timestampKey: testTime,
+			},
+			expect: "foo-2020/09/13",
+		},
+		{
+			name: "attr with time, with missing",
+			val:  "{.event}-{.foo}-{{yyyy/MM/dd}}",
+			attrs: map[string]interface{}{
+				"event":      "foo",
+				timestampKey: testTime,
+			},
+			expect: "foo--2020/09/13",
+		},
+		{
+			name: "attr with time, leading alphas",
+			val:  "{the.event}-{{yyyy/MM/dd}}",
+			attrs: map[string]interface{}{
+				"event":      "foo",
+				timestampKey: testTime,
+			},
+			expect: "foo-2020/09/13",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			formatter := buildIndexFormatter(test.val, time.Local)
+			assert.Equal(t, test.expect, formatter(test.attrs))
+		})
+	}
+}

@@ -39,4 +39,4 @@ Processors:
       Hosts:
         - "172.16.141.4:9200"
         - "172.16.141.5:9200"
-      Index: "{.event}-{{yy-mm-dd}}"
+      Index: "{.event}-{{yyyy-MM-dd}}"