[OWL-2005] Add logging for too late metric received by OWL

.gitignore

@@ -4,3 +4,7 @@ package_cache_tmp
 bin
 open-falcon
 open-falcon*.tar.gz
+
+# Untrack the source code of the dependencies
+vendor/!vendor.json
+vendor/*

.travis.yml

@@ -33,8 +33,8 @@ before_install:
   - cd $TRAVIS_BUILD_DIR
 
 script:
-  - make misspell-check
-  - make fmt-check
+  - make setup-govendor
+  - make check-all
   - make all
   - make pack
   - make GO_TEST_COVERAGE_FILE=coverage.txt go-test

Makefile

@@ -52,35 +52,60 @@ GO_TEST_PROPS_SEP :=
 GO_TEST_PROPS_FILE :=
 GO_TEST_COVERAGE_FILE :=
 
-all: install $(CMD) $(TARGET)
+all: setup-govendor govendor-check $(CMD) $(TARGET)
+
+check-all: misspell-check fmt-check govendor-check
 
 misspell: build_gofile_listfile .get_misspell
 	@echo "Inline fix mis-spelled files.";
 	$(XARGS_CMD) misspell -w <$(LISTFILE_OF_GO_FILES);
 
 misspell-check: build_gofile_listfile .get_misspell
 	check_cmd="$(XARGS_CMD) misspell -error <$(LISTFILE_OF_GO_FILES)"; \
-	echo $$check_cmd; \
+	echo Check misspelling of GoLang ...; \
+	echo -e "\t$$check_cmd"; \
 	check_output=$$(eval "$$check_cmd"); \
 	test -z "$$check_output" || { \
 		echo -e "misspell capture error:\n $$check_output\n"; \
 		echo "[HELP]" Use \"make misspell\" to fix files inline."(Don't forget to commit changed files)"; \
 		exit 1; \
 	}
+	echo -e "[PASS]\n"
 
 fmt: build_gofile_listfile
 	@echo "Inline fix mis-formatted files.";
 	$(XARGS_CMD) $(GOFMT) -l -w <$(LISTFILE_OF_GO_FILES);
 
 fmt-check: build_gofile_listfile
 	check_cmd="$(XARGS_CMD) $(GOFMT) -d <$(LISTFILE_OF_GO_FILES)"; \
-	echo $$check_cmd; \
+	echo Check formatter of GoLang ...; \
+	echo -e "\t$$check_cmd"; \
 	check_output=$$(eval "$$check_cmd"); \
 	test -z "$$check_output" || { \
 		echo -e "gofmt capture error:\n $$check_output\n"; \
 		echo "[HELP]" Use \"make fmt\" to fix files inline."(Don't forget to commit changed files)"; \
 		exit 1; \
-	}
+	};
+	echo -e "[PASS]\n"
+
+# Asserts that there is no external libaray
+govendor-check:
+	@echo -n "Check +external of govendor(\"govendor list +external\") ... "
+	@external_libs=`govendor list +external`; \
+	if test -n "$$external_libs"; then \
+		echo -e "There are external library. You should use govendor add \"<lib path>\" .\n"; \
+		echo -e $$external_libs; \
+		exit 1; \
+	fi
+	@echo "[PASS]"
+	@echo -n "Check +unused of govendor(\"govendor list +unused\") ... "
+	@unused_libs=`govendor list +unused`; \
+	if test -n "$$unused_libs"; then \
+		echo -e "There are unused library. You should use govendor remove \"<lib path>\" .\n"; \
+		echo -e $$unused_libs; \
+		exit 1; \
+	fi
+	@echo "[PASS]"
 
 build_gofile_listfile:
 	echo Generate "$(LISTFILE_OF_GO_FILES)" file for GoLang files.
@@ -110,12 +135,10 @@ $(CMD):
 $(TARGET): $(TARGET_SOURCE)
 	go build -ldflags "-X main.GitCommit=`git rev-parse --short HEAD` -X main.Version=$(VERSION)" -o $@
 
-checkvendor:
+setup-govendor:
 	@hash govendor > /dev/null 2>&1; if [ $$? -ne 0 ]; then \
 		go get -u github.com/kardianos/govendor; \
 	fi
-
-install: checkvendor
 	govendor sync
 
 checkbin: bin/ config/ open-falcon cfg.json
@@ -147,6 +170,6 @@ clean:
 	@rm -rf open-falcon-v$(VERSION).tar.g
 
 .PHONY: install clean all aggregator graph hbs judge nodata query sender task transfer fe f2e-api coverage
-.PHONY: fmt misspell fmt-check misspell-check .get_misspell build_gofile_listfile go-test
+.PHONY: fmt misspell fmt-check misspell-check check-all .get_misspell build_gofile_listfile go-test
 
 .SILENT: build_gofile_listfile misspell-check fmt-check go-test .get_misspell

common/model/event.go

@@ -2,38 +2,38 @@ package model
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/Cepave/open-falcon-backend/common/utils"
 )
 
 // 机器监控和实例监控都会产生Event，共用这么一个struct
 type Event struct {
-	Id          string            `json:"id"`
-	Strategy    *Strategy         `json:"strategy"`
-	Expression  *Expression       `json:"expression"`
-	Status      string            `json:"status"` // OK or PROBLEM
-	Endpoint    string            `json:"endpoint"`
-	LeftValue   float64           `json:"leftValue"`
-	CurrentStep int               `json:"currentStep"`
-	EventTime   int64             `json:"eventTime"`
-	PushedTags  map[string]string `json:"pushedTags"`
+	Id              string            `json:"id"`
+	Strategy        *Strategy         `json:"strategy"`
+	Expression      *Expression       `json:"expression"`
+	Status          string            `json:"status"` // OK or PROBLEM
+	Endpoint        string            `json:"endpoint"`
+	LeftValue       float64           `json:"leftValue"`
+	CurrentStep     int               `json:"currentStep"`
+	PushedTags      map[string]string `json:"pushedTags"`
+	EventTime       int64             `json:"eventTime"`
+	SourceTimestamp int64             `json:"sourceTimestamp"`
 }
 
 func (this *Event) FormattedTime() string {
 	return utils.UnixTsFormat(this.EventTime)
 }
 
 func (this *Event) String() string {
+	eventTime := time.Unix(this.EventTime, 0)
+	sourceTimestamp := time.Unix(this.SourceTimestamp, 0)
+
 	return fmt.Sprintf(
-		"<Endpoint:%s, Status:%s, Strategy:%v, Expression:%v, LeftValue:%s, CurrentStep:%d, PushedTags:%v, TS:%s>",
-		this.Endpoint,
-		this.Status,
-		this.Strategy,
-		this.Expression,
-		utils.ReadableFloat(this.LeftValue),
-		this.CurrentStep,
-		this.PushedTags,
-		this.FormattedTime(),
+		"<Endpoint:%s, Status:%s, Strategy:%v, Expression:%v, LeftValue:%s, CurrentStep:%d, PushedTags:%v, Event Time[%s], Source Time[%s]>",
+		this.Endpoint, this.Status, this.Strategy, this.Expression,
+		utils.ReadableFloat(this.LeftValue), this.CurrentStep, this.PushedTags,
+		eventTime, sourceTimestamp,
 	)
 }
 

common/model/judge.go

@@ -2,27 +2,31 @@ package model
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/Cepave/open-falcon-backend/common/utils"
 )
 
 type JudgeItem struct {
-	Endpoint  string            `json:"endpoint"`
-	Metric    string            `json:"metric"`
-	Value     float64           `json:"value"`
-	Timestamp int64             `json:"timestamp"`
-	JudgeType string            `json:"judgeType"`
-	Tags      map[string]string `json:"tags"`
+	Endpoint        string            `json:"endpoint"`
+	Metric          string            `json:"metric"`
+	Value           float64           `json:"value"`
+	JudgeType       string            `json:"judgeType"`
+	Tags            map[string]string `json:"tags"`
+	Timestamp       int64             `json:"timestamp"`
+	SourceTimestamp int64             `json:"source_timestamp"`
 }
 
 func (this *JudgeItem) String() string {
-	return fmt.Sprintf("<Endpoint:%s, Metric:%s, Value:%f, Timestamp:%d, JudgeType:%s Tags:%v>",
-		this.Endpoint,
-		this.Metric,
-		this.Value,
-		this.Timestamp,
-		this.JudgeType,
-		this.Tags)
+	alignTime := time.Unix(this.Timestamp, 0)
+	sourceTime := time.Unix(this.SourceTimestamp, 0)
+
+	return fmt.Sprintf(
+		"<Endpoint[%s], Metric[%s], Value[%f], Timestamp(align)[%s], Source Timestamp[%s], JudgeType[%s] Tags[%v]>",
+		this.Endpoint, this.Metric, this.Value,
+		alignTime, sourceTime,
+		this.JudgeType, this.Tags,
+	)
 }
 
 func (this *JudgeItem) PrimaryKey() string {

common/model/metric.go

@@ -46,13 +46,14 @@ func (t *JsonMetaData) String() string {
 }
 
 type MetaData struct {
-	Metric      string            `json:"metric"`
-	Endpoint    string            `json:"endpoint"`
-	Timestamp   int64             `json:"timestamp"`
-	Step        int64             `json:"step"`
-	Value       float64           `json:"value"`
-	CounterType string            `json:"counterType"`
-	Tags        map[string]string `json:"tags"`
+	Metric       string            `json:"metric"`
+	Endpoint     string            `json:"endpoint"`
+	Timestamp    int64             `json:"timestamp"`
+	Step         int64             `json:"step"`
+	Value        float64           `json:"value"`
+	CounterType  string            `json:"counterType"`
+	Tags         map[string]string `json:"tags"`
+	SourceMetric *MetricValue      `json:"-"`
 }
 
 func (t *MetaData) String() string {

common/net/listener.go

@@ -0,0 +1,108 @@
+// Provides various utilities to construct listener
+package net
+
+import (
+	"net"
+	"sync"
+)
+
+// Initializes TCP listener
+//
+// error object would be viable if anything get failed
+func InitTcpListener(address string) (net.Listener, error) {
+	tcpAddr, err := net.ResolveTCPAddr("tcp", address)
+	if err != nil {
+		return nil, err
+	}
+
+	listener, err := net.ListenTCP("tcp", tcpAddr)
+	if err != nil {
+		return nil, err
+	}
+
+	logger.Infof("Init TCP Address[ %s ]", address)
+
+	return listener, nil
+}
+
+// [MUST] Initializes TCP listener
+//
+// Panic would raise if anything get failed
+func MustInitTcpListener(address string) net.Listener {
+	listener, err := InitTcpListener(address)
+
+	if err != nil {
+		logger.Fatalf("Cannot initialize listener: %v", err)
+		panic(err.Error())
+	}
+
+	return listener
+}
+
+// Constructs the controller for usage on "net.Listener"
+func NewListenerController(listener net.Listener) *ListenerController {
+	return &ListenerController{
+		Listener: listener,
+		working:  false,
+		lock:     &sync.Mutex{},
+	}
+}
+
+type ListenerController struct {
+	net.Listener
+
+	working bool
+	lock    *sync.Mutex
+}
+
+// This method would keep accepting message of socket
+//
+// This method would use go routine to call your conn handler.
+func (c *ListenerController) AcceptLoop(connHandler func(conn net.Conn)) {
+	c.lock.Lock()
+	if c.working {
+		return
+	}
+	c.working = true
+	c.lock.Unlock()
+
+	for {
+		conn, err := c.Accept()
+		if !c.working {
+			break
+		}
+
+		if err != nil {
+			logger.Errorf("Accept message has error: %v", err)
+			continue
+		}
+
+		go func(lambdaConn net.Conn) {
+			defer func() {
+				p := recover()
+				if p != nil {
+					logger.Panicf("Connection handler has error: %v", p)
+				}
+			}()
+
+			connHandler(lambdaConn)
+		}(conn)
+	}
+}
+
+// Close this controller if any looping is running
+func (c *ListenerController) Close() {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	if !c.working {
+		return
+	}
+
+	c.working = false
+
+	err := c.Listener.Close()
+	if err != nil {
+		logger.Errorf("Close listener[%v] has error: %v", c.Listener.Addr(), err)
+	}
+}

common/net/listener_test.go

@@ -0,0 +1,59 @@
+package net
+
+import (
+	"fmt"
+	"net"
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Initialize TCP listener", func() {
+	Context("Normal initialization", func() {
+		It("Listening on 0.0.0.0:19595", func() {
+			var err error
+
+			listener, err := InitTcpListener("0.0.0.0:19595")
+			defer listener.Close()
+
+			Expect(err).To(Succeed())
+
+			By("[Failed] Listening on 0.0.0.0:19595 again")
+			_, err = InitTcpListener("0.0.0.0:19595")
+			GinkgoT().Logf("Error content: %v", err)
+			Expect(err).To(HaveOccurred())
+		})
+	})
+})
+
+var _ = Describe("Controller for TCP listener", func() {
+	Context("Accept in Loop", func() {
+		listener := MustInitTcpListener("0.0.0.0:19596")
+		testedCtrl := NewListenerController(listener)
+
+		accepted := false
+
+		BeforeEach(func() {
+			go testedCtrl.AcceptLoop(func(conn net.Conn) {
+				accepted = true
+			})
+		})
+		AfterEach(func() {
+			testedCtrl.Close()
+		})
+
+		It("Ensure the accepting is running(sending message)", func() {
+			conn, err := net.Dial("tcp", "127.0.0.1:19596")
+			defer conn.Close()
+
+			Expect(err).To(Succeed())
+			fmt.Fprintf(conn, "Hello World!!")
+
+			Eventually(
+				func() bool { return accepted },
+				2*time.Second, 200*time.Millisecond,
+			).Should(BeTrue())
+		})
+	})
+})