send search logs to Elasticsearch via fluentd

* adds fluent.conf for debugging
* send search logs to fluentd
* send search logs from fluentd to elasticsearch

_fluentd/etc/fluent.conf

@@ -0,0 +1,13 @@
+<source>
+  @type forward
+  port 24224
+</source>
+
+<match blog.search>
+  @type elasticsearch
+  host elasticsearch
+  port 9200
+  logstash_format true
+  logstash_prefix search
+  type_name searchlogs
+</match>

controller/article.go

@@ -11,18 +11,19 @@ import (
 
 	"golang.org/x/net/context"
 
-	"github.com/gin-gonic/contrib/sessions"
 	"github.com/efkbook/blog-sample/model"
-	csrf "github.com/utrack/gin-csrf"
-
+	"github.com/fluent/fluent-logger-golang/fluent"
+	"github.com/gin-gonic/contrib/sessions"
 	"github.com/gin-gonic/gin"
+	csrf "github.com/utrack/gin-csrf"
 	elastic "gopkg.in/olivere/elastic.v5"
 )
 
 // Article is controller for requests to articles.
 type Article struct {
-	DB *sql.DB
-	ES *elastic.Client
+	DB     *sql.DB
+	ES     *elastic.Client
+	Fluent *fluent.Fluent
 }
 
 // Root indicates / path as top page.
@@ -173,9 +174,25 @@ func (t *Article) Delete(c *gin.Context) {
 	c.Redirect(301, "/")
 }
 
+type SearchLog struct {
+	Query  string `msg:"query"`
+	UserID int64  `msg:"user_id"`
+}
+
 // TODO search document by elasticsearch
 func (t *Article) Search(c *gin.Context) {
 	queryString := c.Query("q")
+
+	sess := sessions.Default(c)
+	searchLog := SearchLog{Query: queryString}
+	if uid, ok := sess.Get("uid").(int64); ok {
+		searchLog.UserID = uid
+	}
+	if err := t.Fluent.Post("blog.search", searchLog); err != nil {
+		// NOTE: if posting search log to fluentd failed, not panic.
+		log.Printf("post to fluentd failed.: %s", err)
+	}
+
 	query := elastic.NewQueryStringQuery(queryString).DefaultField("body")
 	result, err := t.ES.Search().Index("article").Query(query).Sort("created", false).Do(context.Background())
 	if err != nil {

docker-compose.yml

@@ -35,6 +35,10 @@ services:
     build:
       context: .
       dockerfile: Dockerfile-fluentd
+    volumes:
+      - ./_fluentd/etc:/fluentd/etc
+    environment:
+      - FLUENTD_CONF=fluent.conf
     ports:
       - 24224:24224
 

main.go

@@ -14,6 +14,7 @@ import (
 
 	csrf "github.com/utrack/gin-csrf"
 
+	"github.com/fluent/fluent-logger-golang/fluent"
 	"github.com/gin-gonic/contrib/sessions"
 	"github.com/gin-gonic/gin"
 	_ "github.com/mattn/go-sqlite3"
@@ -26,6 +27,7 @@ type Server struct {
 	db     *sql.DB
 	Engine *gin.Engine
 	es     *elastic.Client
+	fluent *fluent.Fluent
 }
 
 // Close makes the database connection to close.
@@ -55,6 +57,12 @@ func (s *Server) Init(dbconf, env string) {
 	}
 	s.es = client
 
+	logger, err := fluent.New(fluent.Config{})
+	if err != nil {
+		log.Fatalf("initialize fluentd client failed: %s", err)
+	}
+	s.fluent = logger
+
 	// NOTE: define helper func to use from templates here.
 	t := template.Must(template.New("").Funcs(template.FuncMap{
 		"LoggedIn":    controller.LoggedIn,
@@ -91,7 +99,7 @@ func (s *Server) Run(addr ...string) {
 
 // Route setting router for this blog.
 func (s *Server) Route() {
-	article := &controller.Article{DB: s.db, ES: s.es}
+	article := &controller.Article{DB: s.db, ES: s.es, Fluent: s.fluent}
 	user := &controller.User{DB: s.db}
 
 	auth := s.Engine.Group("/")