add {.event} in index format

@@ -16,7 +16,7 @@ type (
 
 	ElasticSearchConf struct {
 		Hosts         []string
-		DailyIndexPrefix string
+		Index         string
 		DocType       string `json:",default=doc"`
 		TimeZone      string `json:",optional"`
 		MaxChunkBytes int    `json:",default=1048576"`

@@ -2,40 +2,79 @@ package es
 
 import (
 	"context"
+	"fmt"
+	"strings"
 	"sync"
 	"time"
 
 	"github.com/olivere/elastic"
 	"github.com/tal-tech/go-zero/core/fx"
+	"github.com/tal-tech/go-zero/core/lang"
 	"github.com/tal-tech/go-zero/core/logx"
 	"github.com/tal-tech/go-zero/core/syncx"
 )
 
-const sharedCallsKey = "ensureIndex"
+const (
+	sharedCallsKey  = "ensureIndex"
+	timestampFormat = "2006-01-02T15:04:05.000Z"
+	timestampKey    = "@timestamp"
+)
+
+const (
+	stateNormal = iota
+	stateWrap
+	stateDot
+)
 
 type (
-	IndexFormat func(time.Time) string
+	IndexFormat func(m map[string]interface{}) string
 	IndexFunc   func() string
 
 	Index struct {
 		client      *elastic.Client
 		indexFormat IndexFormat
-		index       string
+		indices     map[string]lang.PlaceholderType
 		lock        sync.RWMutex
 		sharedCalls syncx.SharedCalls
 	}
 )
 
-func NewIndex(client *elastic.Client, indexFormat IndexFormat) *Index {
+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)
+				}
+			}
+			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: indexFormat,
+		indexFormat: formatter,
+		indices:     make(map[string]lang.PlaceholderType),
 		sharedCalls: syncx.NewSharedCalls(),
 	}
 }
 
-func (idx *Index) GetIndex(t time.Time) string {
-	index := idx.indexFormat(t)
+func (idx *Index) GetIndex(m map[string]interface{}) string {
+	index := idx.indexFormat(m)
+	idx.lock.RLock()
+	if _, ok := idx.indices[index]; ok {
+		idx.lock.RUnlock()
+		return index
+	}
+
+	idx.lock.RUnlock()
 	if err := idx.ensureIndex(index); err != nil {
 		logx.Error(err)
 	}
@@ -43,17 +82,14 @@ func (idx *Index) GetIndex(t time.Time) string {
 }
 
 func (idx *Index) ensureIndex(index string) error {
-	idx.lock.RLock()
-	if index == idx.index {
-		idx.lock.RUnlock()
-		return nil
-	}
-	idx.lock.RUnlock()
-
 	_, err := idx.sharedCalls.Do(sharedCallsKey, func() (i interface{}, err error) {
 		idx.lock.Lock()
 		defer idx.lock.Unlock()
 
+		if _, ok := idx.indices[index]; ok {
+			return nil, nil
+		}
+
 		existsService := elastic.NewIndicesExistsService(idx.client)
 		existsService.Index([]string{index})
 		exist, err := existsService.Do(context.Background())
@@ -61,7 +97,6 @@ func (idx *Index) ensureIndex(index string) error {
 			return nil, err
 		}
 		if exist {
-			idx.index = index
 			return nil, nil
 		}
 
@@ -74,8 +109,53 @@ func (idx *Index) ensureIndex(index string) error {
 			return nil, err
 		}
 
-		idx.index = index
+		idx.indices[index] = lang.Placeholder
 		return nil, nil
 	})
 	return err
 }
+
+func getTime(m map[string]interface{}) time.Time {
+	if ti, ok := m[timestampKey]; ok {
+		if ts, ok := ti.(string); ok {
+			if t, err := time.Parse(timestampFormat, ts); err == nil {
+				return t
+			}
+		}
+	}
+
+	return time.Now()
+}
+
+func getFormat(indexFormat string) (format string, attrs []string) {
+	var state = stateNormal
+	var builder strings.Builder
+	var keyBuf strings.Builder
+	for _, ch := range indexFormat {
+		switch ch {
+		case '{':
+			state = stateWrap
+		case '.':
+			if state == stateWrap {
+				state = stateDot
+			} else {
+				builder.WriteRune(ch)
+			}
+		case '}':
+			state = stateNormal
+			if keyBuf.Len() > 0 {
+				attrs = append(attrs, keyBuf.String())
+				builder.WriteString("%s")
+			}
+		default:
+			if state == stateDot {
+				keyBuf.WriteRune(ch)
+			} else {
+				builder.WriteRune(ch)
+			}
+		}
+	}
+
+	format = builder.String()
+	return
+}

@@ -2,8 +2,8 @@ package es
 
 import (
 	"context"
-	"time"
 
+	jsoniter "github.com/json-iterator/go"
 	"github.com/olivere/elastic"
 	"github.com/tal-tech/go-stash/stash/config"
 	"github.com/tal-tech/go-zero/core/executors"
@@ -18,8 +18,8 @@ type (
 		inserter *executors.ChunkExecutor
 	}
 
-	valueWithTime struct {
-		t   time.Time
+	valueWithIndex struct {
+		index string
 		val   string
 	}
 )
@@ -43,9 +43,16 @@ func NewWriter(c config.ElasticSearchConf, indexer *Index) (*Writer, error) {
 	return &writer, nil
 }
 
-func (w *Writer) Write(t time.Time, val string) error {
-	return w.inserter.Add(valueWithTime{
-		t:   t,
+func (w *Writer) Write(m map[string]interface{}) error {
+	bs, err := jsoniter.Marshal(m)
+	if err != nil {
+		return err
+	}
+
+	index := w.indexer.GetIndex(m)
+	val := string(bs)
+	return w.inserter.Add(valueWithIndex{
+		index: index,
 		val:   val,
 	}, len(val))
 }
@@ -53,8 +60,8 @@ func (w *Writer) Write(t time.Time, val string) error {
 func (w *Writer) execute(vals []interface{}) {
 	var bulk = w.client.Bulk()
 	for _, val := range vals {
-		pair := val.(valueWithTime)
-		req := elastic.NewBulkIndexRequest().Index(w.indexer.GetIndex(pair.t)).Type(w.docType).Doc(pair.val)
+		pair := val.(valueWithIndex)
+		req := elastic.NewBulkIndexRequest().Index(pair.index).Type(w.docType).Doc(pair.val)
 		bulk.Add(req)
 	}
 	_, err := bulk.Do(context.Background())

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

 package handler
 
 import (
-	"time"
-
 	jsoniter "github.com/json-iterator/go"
 	"github.com/tal-tech/go-stash/stash/es"
 	"github.com/tal-tech/go-stash/stash/filter"
 )
 
-const (
-	timestampFormat = "2006-01-02T15:04:05.000Z"
-	timestampKey    = "@timestamp"
-)
-
 type MessageHandler struct {
 	writer  *es.Writer
 	filters []filter.FilterFunc
@@ -42,22 +35,5 @@ func (mh *MessageHandler) Consume(_, val string) error {
 		}
 	}
 
-	bs, err := jsoniter.Marshal(m)
-	if err != nil {
-		return err
-	}
-
-	return mh.writer.Write(mh.getTime(m), string(bs))
-}
-
-func (mh *MessageHandler) getTime(m map[string]interface{}) time.Time {
-	if ti, ok := m[timestampKey]; ok {
-		if ts, ok := ti.(string); ok {
-			if t, err := time.Parse(timestampFormat, ts); err == nil {
-				return t
-			}
-		}
-	}
-
-	return time.Now()
+	return mh.writer.Write(m)
 }

@@ -37,7 +37,6 @@ func main() {
 		)
 		logx.Must(err)
 
-		indexFormat := processor.Output.ElasticSearch.DailyIndexPrefix + dateFormat
 		var loc *time.Location
 		if len(processor.Output.ElasticSearch.TimeZone) > 0 {
 			loc, err = time.LoadLocation(processor.Output.ElasticSearch.TimeZone)
@@ -45,10 +44,7 @@ func main() {
 		} else {
 			loc = time.Local
 		}
-		indexer := es.NewIndex(client, func(t time.Time) string {
-			return t.In(loc).Format(indexFormat)
-		})
-
+		indexer := es.NewIndex(client, processor.Output.ElasticSearch.Index, loc)
 		filters := filter.CreateFilters(processor)
 		writer, err := es.NewWriter(processor.Output.ElasticSearch, indexer)
 		logx.Must(err)