add TagPrefix Option

main.go

@@ -24,12 +24,12 @@ import (
 var (
 	kinesisStreamName     string
 	checkpointTablePrefix string
-	defaultTag            string
 	tagKey                string
 
-	fluentSocket string
-	fluentHost   = "localhost"
-	fluentPort   = 24224
+	fluentTagPrefix string
+	fluentSocket    string
+	fluentHost      = "localhost"
+	fluentPort      = 24224
 
 	dogStatsdAddr string
 	dogStatsdTags []string
@@ -47,12 +47,9 @@ func init() {
 		log.Fatalln("env CHECKPOINT_TABLE_PREFIX is required")
 	}
 
-	defaultTag = os.Getenv("DEFAULT_TAG")
-	if len(defaultTag) == 0 {
-		defaultTag = "default"
-	}
 	tagKey = os.Getenv("TAG_KEY")
 
+	fluentTagPrefix = os.Getenv("FLUENT_TAG_PREFIX")
 	fluentSocket = os.Getenv("FLUENT_SOCKET")
 	if len(os.Getenv("FLUENT_HOST")) != 0 {
 		fluentHost = os.Getenv("FLUENT_HOST")
@@ -78,17 +75,20 @@ func initFluentLogger(retry int) *fluent.Fluent {
 	if len(fluentSocket) > 0 {
 		cfg = fluent.Config{
 			MarshalAsJSON:    true,
+			TagPrefix:        fluentTagPrefix,
 			FluentNetwork:    "unix",
 			FluentSocketPath: fluentSocket,
 		}
 	} else {
 		cfg = fluent.Config{
 			MarshalAsJSON: true,
+			TagPrefix:     fluentTagPrefix,
 			FluentHost:    fluentHost,
 			FluentPort:    fluentPort,
 		}
 	}
 
+
 	_l, err := fluent.New(cfg)
 	if err != nil {
 		if retry == 0 {
@@ -203,10 +203,11 @@ func main() {
 			if tagKey != "" {
 				if val, ok := message[tagKey].(string); ok {
 					tag = val
+				} else {
+					tag = "unknown"
 				}
-			}
-			if tag == "" {
-				tag = defaultTag
+			} else {
+				tag = "default"
 			}
 			if err := l.Post(tag, message); err != nil {
 				log.Fatalf("Failed to post: %v", err)