Index v2

config.toml.sample

@@ -2,7 +2,7 @@ listener = "0.0.0.0:8080"
 #tlslistener = "0.0.0.0:8443"
 
 #data = "/data"
-#elasticsearch_url = "http://127.0.0.1:9200"
+#elasticsearch_url = "http://127.0.0.1:9200/server"
 
 #socks = "socks4://127.0.0.1:9050"
 

server/indexer.go

@@ -3,51 +3,63 @@ package server
 import (
 	"context"
 	"time"
-
+	"net/url"
 	"github.com/pborman/uuid"
 	"gopkg.in/olivere/elastic.v5"
+	"errors"
 )
 
-func (p *Server) indexer() {
+func (p *Server) indexer() (err error) {
 	log.Info("Indexer started...")
-	defer log.Info("Indexer stopped...")
 
-	es, err := elastic.NewClient(elastic.SetURL(p.ElasticsearchURL), elastic.SetSniff(false))
+	u, err := url.Parse(p.ElasticsearchURL)
 	if err != nil {
-		panic(err)
+	 	return errors.New("Error parsing url: " + p.ElasticsearchURL)
+
 	}
 
-	bulk := es.Bulk()
-
-	count := 0
-	for {
-		select {
-		case doc := <-p.index:
-			docId := uuid.NewUUID()
-			bulk = bulk.Add(elastic.NewBulkIndexRequest().
-				Index("server").
-				Type("pairs").
-				Id(docId.String()).
-				Doc(doc),
-			)
+	if u.Path == "" || u.Path[1:] == "" {
+         	return errors.New("Index is not set in elasticsearch_url: " + p.ElasticsearchURL)
+        }
+
+	es, err := elastic.NewClient(elastic.SetURL(u.Host), elastic.SetSniff(false))
+	if err != nil {
+		return err
+	}
+	go func() {
+		defer log.Info("Indexer stopped...")
+		bulk := es.Bulk()
+		count := 0
+		for {
+			select {
+			case doc := <-p.index:
+				docId := uuid.NewUUID()
+				bulk = bulk.Add(elastic.NewBulkIndexRequest().
+					Index(u.Path[1:]).
+					Type("pairs").
+					Id(docId.String()).
+					Doc(doc),
+				)
 
-			log.Debugf("Indexed message with id %s", docId.String())
+				log.Debugf("Indexed message with id %s", docId.String())
 
-			// pretty.Print(doc)
-			if bulk.NumberOfActions() < 10 {
-				continue
+				// pretty.Print(doc)
+				if bulk.NumberOfActions() < 10 {
+					continue
+				}
+			case <-time.After(time.Second * 10):
+			}
+
+			if bulk.NumberOfActions() == 0 {
+			} else if response, err := bulk.Do(context.Background()); err != nil {
+				log.Errorf("Error indexing: %s", err.Error())
+			} else {
+				indexed := response.Indexed()
+				count += len(indexed)
+
+				log.Infof("Bulk indexing: %d total %d.\n", len(indexed), count)
 			}
-		case <-time.After(time.Second * 10):
-		}
-
-		if bulk.NumberOfActions() == 0 {
-		} else if response, err := bulk.Do(context.Background()); err != nil {
-			log.Errorf("Error indexing: %s", err.Error())
-		} else {
-			indexed := response.Indexed()
-			count += len(indexed)
-
-			log.Infof("Bulk indexing: %d total %d.\n", len(indexed), count)
 		}
-	}
+	}()
+return nil
 }

server/proxy.go

@@ -96,7 +96,11 @@ func (c *Server) Run() {
 	defer log.Info("Ares stopped....")
 
 	if c.ElasticsearchURL != "" {
-		go c.indexer()
+		err := c.indexer()
+                if err != nil {
+			log.Error(err)
+		return 
+		}
 	}
 
 	var router = mux.NewRouter()