Refine migrate

modules/api/test/api_test.go

@@ -20,6 +20,7 @@ import (
 	"os"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/masato25/resty"
 	log "github.com/sirupsen/logrus"
@@ -555,31 +556,60 @@ func TestGraph(t *testing.T) {
 
 	rc := resty.New()
 	rc.SetHeader("Apitoken", api_token)
+	var rr *[]map[string]interface{} = &[]map[string]interface{}{}
 
 	Convey("Get endpoint list: GET /graph/endpoint", t, func() {
-		r := []map[string]interface{}{}
 		resp, _ := rc.R().SetQueryParam("q", ".+").
-			SetResult(&r).
+			SetResult(rr).
 			Get(fmt.Sprintf("%s/graph/endpoint", api_v1))
 		So(resp.StatusCode(), ShouldEqual, 200)
-		So(len(r), ShouldBeGreaterThanOrEqualTo, 0)
+		So(len(*rr), ShouldBeGreaterThanOrEqualTo, 0)
 
-		if len(r) == 0 {
-			return
-		}
+	})
 
-		eid := r[0]["id"]
-		r = []map[string]interface{}{}
-		Convey("Get counter list: GET /graph/endpoint_counter", func() {
-			resp, _ := rc.R().
-				SetQueryParam("eid", fmt.Sprintf("%v", eid)).
-				SetQueryParam("metricQuery", ".+").
-				SetQueryParam("limit", "1").
-				SetResult(&r).
-				Get(fmt.Sprintf("%s/graph/endpoint_counter", api_v1))
-			So(resp.StatusCode(), ShouldEqual, 200)
-			So(r, ShouldNotBeEmpty)
-		})
+	if len(*rr) == 0 {
+		return
+	}
+
+	eid := (*rr)[0]["id"]
+	endpoint := (*rr)[0]["endpoint"]
+	Convey("Get counter list: GET /graph/endpoint_counter", t, func() {
+		resp, _ := rc.R().
+			SetQueryParam("eid", fmt.Sprintf("%v", eid)).
+			SetQueryParam("metricQuery", ".+").
+			SetQueryParam("limit", "1").
+			SetResult(rr).
+			Get(fmt.Sprintf("%s/graph/endpoint_counter", api_v1))
+		So(resp.StatusCode(), ShouldEqual, 200)
+		So(*rr, ShouldNotBeEmpty)
+	})
+
+	if len(*rr) == 0 {
+		return
+	}
+
+	counter := (*rr)[0]["counter"]
+	step := (*rr)[0]["step"]
+
+	now := time.Now()
+	start_ts := now.Add(time.Duration(-1) * time.Hour).Unix()
+	end_ts := now.Unix()
+
+	Convey("Query counter history: POST /graph/history", t, func() {
+		resp, _ := rc.R().
+			SetBody(map[string]interface{}{
+				"step":       step,
+				"consol_fun": "AVERAGE",
+				"start_time": start_ts,
+				"end_time":   end_ts,
+				"hostnames":  []string{endpoint.(string)},
+				"counters":   []string{counter.(string)},
+			}).
+			SetResult(rr).
+			Post(fmt.Sprintf("%s/graph/history", api_v1))
+		log.Info(resp)
+		So(resp.StatusCode(), ShouldEqual, 200)
+		So(*rr, ShouldNotBeEmpty)
 	})
 }
 

modules/graph/api/graph.go

@@ -48,7 +48,7 @@ func (this *Graph) GetRrd(key string, rrdfile *g.File) (err error) {
 
 	items := store.GraphItems.PopAll(key)
 	if len(items) > 0 {
-		rrdtool.FlushFile(rrdfile.Filename, md5, items)
+		rrdtool.CommitFile(rrdfile.Filename, md5, items)
 	}
 
 	rrdfile.Body, err = rrdtool.ReadFile(rrdfile.Filename, md5)
@@ -178,7 +178,11 @@ func (this *Graph) Query(param cmodel.GraphQueryParam, resp *cmodel.GraphQueryRe
 		// read data from rrd file
 		// 从RRD中获取数据不包含起始时间点
 		// 例: start_ts=1484651400,step=60,则第一个数据时间为1484651460)
-		datas, _ = rrdtool.Fetch(filename, md5, param.ConsolFun, start_ts-int64(step), end_ts, step)
+		var err error
+		datas, err = rrdtool.Fetch(filename, md5, param.ConsolFun, start_ts-int64(step), end_ts, step)
+		if err != nil {
+			log.Debugf("rrdtool fetch %s error: %v", filename, err)
+		}
 		datas_size = len(datas)
 	}
 

modules/graph/api/rpc.go

@@ -16,7 +16,7 @@ package api
 
 import (
 	"container/list"
-	"log"
+	log "github.com/sirupsen/logrus"
 	"net"
 	"net/rpc"
 	"sync"
@@ -52,7 +52,7 @@ func init() {
 
 func Start() {
 	if !g.Config().Rpc.Enabled {
-		log.Println("rpc.Start warning, not enabled")
+		log.Info("rpc.Start warning, not enabled")
 		return
 	}
 	addr := g.Config().Rpc.Listen
@@ -65,7 +65,7 @@ func Start() {
 	if err != nil {
 		log.Fatalf("rpc.Start error, listen %s failed, %s", addr, err)
 	} else {
-		log.Println("rpc.Start ok, listening on", addr)
+		log.Info("rpc.Start ok, listening on", addr)
 	}
 
 	rpc.Register(new(Graph))
@@ -99,7 +99,7 @@ func Start() {
 
 	select {
 	case <-Close_chan:
-		log.Println("rpc, recv sigout and exiting...")
+		log.Info("rpc, recv sigout and exiting...")
 		listener.Close()
 		Close_done_chan <- 1
 

modules/graph/cron/clean.go

@@ -29,7 +29,7 @@
 package cron
 
 import (
-	"log"
+	log "github.com/sirupsen/logrus"
 	"strings"
 	"time"
 
@@ -85,7 +85,7 @@ func DeleteInvalidItems() int {
 
 	pfc.Gauge("GraphItemsCacheCnt", int64(currentCnt))
 	pfc.Gauge("GraphItemsCacheInvalidCnt", int64(deleteCnt))
-	log.Printf("GraphItemsCache: Count=>%d, DeleteInvalid=>%d", currentCnt, deleteCnt)
+	log.Infof("GraphItemsCache: Count=>%d, DeleteInvalid=>%d", currentCnt, deleteCnt)
 
 	return deleteCnt
 }
@@ -107,7 +107,7 @@ func DeleteInvalidHistory() int {
 
 	pfc.Gauge("HistoryCacheCnt", int64(currentCnt))
 	pfc.Gauge("HistoryCacheInvalidCnt", int64(deleteCnt))
-	log.Printf("HistoryCache: Count=>%d, DeleteInvalid=>%d", currentCnt, deleteCnt)
+	log.Infof("HistoryCache: Count=>%d, DeleteInvalid=>%d", currentCnt, deleteCnt)
 
 	return deleteCnt
 }

modules/graph/g/cfg.go

@@ -16,7 +16,7 @@ package g
 
 import (
 	"encoding/json"
-	"log"
+	log "github.com/sirupsen/logrus"
 	"strconv"
 	"sync/atomic"
 	"unsafe"
@@ -77,24 +77,24 @@ func Config() *GlobalConfig {
 
 func ParseConfig(cfg string) {
 	if cfg == "" {
-		log.Fatalln("config file not specified: use -c $filename")
+		log.Fatal("config file not specified: use -c $filename")
 	}
 
 	if !file.IsExist(cfg) {
-		log.Fatalln("config file specified not found:", cfg)
+		log.Fatal("config file specified not found:", cfg)
 	}
 
 	ConfigFile = cfg
 
 	configContent, err := file.ToTrimString(cfg)
 	if err != nil {
-		log.Fatalln("read config file", cfg, "error:", err.Error())
+		log.Fatal("read config file", cfg, "error:", err.Error())
 	}
 
 	var c GlobalConfig
 	err = json.Unmarshal([]byte(configContent), &c)
 	if err != nil {
-		log.Fatalln("parse config file", cfg, "error:", err.Error())
+		log.Fatal("parse config file", cfg, "error:", err.Error())
 	}
 
 	if c.Migrate.Enabled && len(c.Migrate.Cluster) == 0 {
@@ -112,5 +112,5 @@ func ParseConfig(cfg string) {
 	// set config
 	atomic.StorePointer(&ptr, unsafe.Pointer(&c))
 
-	log.Println("g.ParseConfig ok, file", cfg)
+	log.Info("g.ParseConfig ok, file", cfg)
 }

modules/graph/g/db.go

@@ -17,7 +17,7 @@ package g
 import (
 	"database/sql"
 	_ "github.com/go-sql-driver/mysql"
-	"log"
+	log "github.com/sirupsen/logrus"
 	"sync"
 )
 
@@ -33,11 +33,11 @@ func InitDB() {
 	var err error
 	DB, err = makeDbConn()
 	if DB == nil || err != nil {
-		log.Fatalln("g.InitDB, get db conn fail", err)
+		log.Fatal("g.InitDB, get db conn fail", err)
 	}
 
 	dbConnMap = make(map[string]*sql.DB)
-	log.Println("g.InitDB ok")
+	log.Info("g.InitDB ok")
 }
 
 func GetDbConn(connName string) (c *sql.DB, e error) {

modules/graph/g/g.go

@@ -15,7 +15,6 @@
 package g
 
 import (
-	"log"
 	"runtime"
 )
 
@@ -55,7 +54,6 @@ const (
 	DEFAULT_STEP    = 60      //s
 	MIN_STEP        = 30      //s
 	CLEAN_CACHE     = 86400   //s the step that clean GraphItems/historyCache Cache
-	CACHE_DELAY     = 1800    //s
 	CACHE_TIME      = 1800000 //ms
 	FLUSH_DISK_STEP = 1000    //ms
 	FLUSH_MIN_COUNT = 6       //  flush counter to disk when its number of monitoring data greater than FLUSH_MIN_COUNT
@@ -71,5 +69,4 @@ const (
 
 func init() {
 	runtime.GOMAXPROCS(runtime.NumCPU())
-	log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile)
 }

modules/graph/http/common.go

@@ -16,10 +16,13 @@ package http
 
 import (
 	"github.com/open-falcon/falcon-plus/modules/graph/g"
+	"github.com/open-falcon/falcon-plus/modules/graph/rrdtool"
 	"github.com/open-falcon/falcon-plus/modules/graph/store"
 	"github.com/toolkits/file"
 
+	"fmt"
 	"github.com/gin-gonic/gin"
+	log "github.com/sirupsen/logrus"
 	"strconv"
 	"time"
 )
@@ -73,4 +76,18 @@ func configCommonRoutes() {
 		}
 		JSONR(c, 200, rt)
 	})
+
+	router.GET("/api/v2/counter/migrate", func(c *gin.Context) {
+		counter := rrdtool.GetCounterV2()
+		log.Debug("migrating counter v2:", fmt.Sprintf("%+v", counter))
+		c.JSON(200, counter)
+	})
+
+	//compatible with open-falcon v0.1
+	router.GET("/counter/migrate", func(c *gin.Context) {
+		cnt := rrdtool.GetCounter()
+		log.Debug("migrating counter:", cnt)
+		c.JSON(200, gin.H{"msg": "ok", "counter": cnt})
+	})
+
 }

modules/graph/http/http.go

@@ -16,7 +16,6 @@ package http
 
 import (
 	"encoding/json"
-	"fmt"
 	log "github.com/sirupsen/logrus"
 	"net"
 	"net/http"
@@ -25,7 +24,6 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/open-falcon/falcon-plus/modules/graph/g"
-	"github.com/open-falcon/falcon-plus/modules/graph/rrdtool"
 )
 
 type Dto struct {
@@ -102,19 +100,6 @@ func Start() {
 	configProcRoutes()
 	configIndexRoutes()
 
-	router.GET("/api/v2/counter/migrate", func(c *gin.Context) {
-		counter := rrdtool.GetCounterV2()
-		log.Debug("migrating counter v2:", fmt.Sprintf("%+v", counter))
-		c.JSON(200, counter)
-	})
-
-	//compatible with open-falcon v0.1
-	router.GET("/counter/migrate", func(c *gin.Context) {
-		cnt := rrdtool.GetCounter()
-		log.Debug("migrating counter:", cnt)
-		c.JSON(200, gin.H{"msg": "ok", "counter": cnt})
-	})
-
 	addr := g.Config().Http.Listen
 	if addr == "" {
 		return

modules/graph/index/cache.go

@@ -17,7 +17,7 @@ package index
 import (
 	"database/sql"
 	"fmt"
-	"log"
+	log "github.com/sirupsen/logrus"
 	"strconv"
 	"strings"
 	"sync"
@@ -97,7 +97,7 @@ func GetEndpointFromCache(endpoint string) (int64, bool) {
 	var id int64 = -1
 	err := g.DB.QueryRow("SELECT id FROM endpoint WHERE endpoint = ?", endpoint).Scan(&id)
 	if err != nil && err != sql.ErrNoRows {
-		log.Println("query endpoint id fail,", err)
+		log.Error("query endpoint id fail,", err)
 		return -1, false
 	}
 
@@ -132,7 +132,7 @@ func GetCounterFromCache(endpointId int64, counter string) (dsType string, step
 	err = g.DB.QueryRow("SELECT type, step FROM endpoint_counter WHERE endpoint_id = ? and counter = ?",
 		endpointId, counter).Scan(&dsType, &step)
 	if err != nil && err != sql.ErrNoRows {
-		log.Println("query type and step fail", err)
+		log.Error("query type and step fail", err)
 		return
 	}