start adding statistics for broker

broker/src/hidra2_broker/database/database_test.go

@@ -12,7 +12,10 @@ func TestMockDataBase(t *testing.T) {
 	db.On("Close").Return()
 	db.On("Copy").Return(nil)
 	db.On("GetNextRecord", "").Return([]byte(""), nil)
-
 	db.Connect("")
 	db.GetNextRecord("")
+	db.Close()
+	db.Copy()
+	var err DBError
+	err.Error()
 }

broker/src/hidra2_broker/database/mock_database.go

@@ -19,7 +19,7 @@ func (db *MockedDatabase) Close() {
 	db.Called()
 }
 
-func (db *MockedDatabase) Copy() Agent{
+func (db *MockedDatabase) Copy() Agent {
 	db.Called()
 	return db
 }

broker/src/hidra2_broker/server/get_next.go

@@ -15,8 +15,8 @@ func extractRequestParameters(r *http.Request) (string, bool) {
 
 func routeGetNext(w http.ResponseWriter, r *http.Request) {
 	r.Header.Set("Content-type", "application/json")
-   // w.Write([]byte("Hello"))
-   // return
+	// w.Write([]byte("Hello"))
+	// return
 	db_name, ok := extractRequestParameters(r)
 	if !ok {
 		w.WriteHeader(http.StatusBadRequest)
@@ -31,6 +31,7 @@ func routeGetNext(w http.ResponseWriter, r *http.Request) {
 func getNextRecord(db_name string) (answer []byte, code int) {
 	db_new := db.Copy()
 	defer db_new.Close()
+	statistics.IncreaseCounter()
 	answer, err := db_new.GetNextRecord(db_name)
 	if err != nil {
 		err_db, ok := err.(*database.DBError)

broker/src/hidra2_broker/server/get_next_test.go

@@ -3,6 +3,7 @@ package server
 import (
 	"errors"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/suite"
 	"hidra2_broker/database"
 	"hidra2_broker/utils"
 	"net/http"
@@ -30,45 +31,57 @@ func TestGetNextWithoutDatabaseName(t *testing.T) {
 	assert.Equal(t, http.StatusNotFound, w.Code, "no database name")
 }
 
-func ExpectCopyClose(mock_db *database.MockedDatabase){
+func ExpectCopyClose(mock_db *database.MockedDatabase) {
 	mock_db.On("Copy").Return(mock_db)
 	mock_db.On("Close").Return()
 }
 
-func TestGetNextWithWrongDatabaseName(t *testing.T) {
-	mock_db := new(database.MockedDatabase)
-	db = mock_db
-	defer func() { db = nil }()
-	ExpectCopyClose(mock_db)
-	mock_db.On("GetNextRecord", "foo").Return([]byte(""),
+type GetNextTestSuite struct {
+	suite.Suite
+	mock_db *database.MockedDatabase
+}
+
+func (suite *GetNextTestSuite) SetupTest() {
+	statistics.Reset()
+	suite.mock_db = new(database.MockedDatabase)
+	db = suite.mock_db
+	ExpectCopyClose(suite.mock_db)
+}
+
+func (suite *GetNextTestSuite) TearDownTest() {
+	assertExpectations(suite.T(), suite.mock_db)
+	db = nil
+}
+
+func TestGetNextTestSuite(t *testing.T) {
+	suite.Run(t, new(GetNextTestSuite))
+}
+
+func (suite *GetNextTestSuite) TestGetNextWithWrongDatabaseName() {
+	suite.mock_db.On("GetNextRecord", "foo").Return([]byte(""),
 		&database.DBError{utils.StatusWrongInput, ""})
 
 	w := doRequest("/database/foo/next")
-	assert.Equal(t, http.StatusBadRequest, w.Code, "wrong database name")
-	assertExpectations(t, mock_db)
+	suite.Equal(http.StatusBadRequest, w.Code, "wrong database name")
 }
 
-func TestGetNextWithInternalDBError(t *testing.T) {
-	mock_db := new(database.MockedDatabase)
-	db = mock_db
-	defer func() { db = nil }()
-	ExpectCopyClose(mock_db)
-	mock_db.On("GetNextRecord", "foo").Return([]byte(""), errors.New(""))
-
+func (suite *GetNextTestSuite) TestGetNextWithInternalDBError() {
+	suite.mock_db.On("GetNextRecord", "foo").Return([]byte(""), errors.New(""))
 	w := doRequest("/database/foo/next")
-	assert.Equal(t, http.StatusInternalServerError, w.Code, "internal error")
-	assertExpectations(t, mock_db)
+	suite.Equal(http.StatusInternalServerError, w.Code, "internal error")
 }
 
-func TestGetNextWithGoodDatabaseName(t *testing.T) {
-	mock_db := new(database.MockedDatabase)
-	db = mock_db
-	defer func() { db = nil }()
-	ExpectCopyClose(mock_db)
-	mock_db.On("GetNextRecord", "dbname").Return([]byte("Hello"), nil)
+func (suite *GetNextTestSuite) TestGetNextWithGoodDatabaseName() {
+	suite.mock_db.On("GetNextRecord", "dbname").Return([]byte("Hello"), nil)
 
 	w := doRequest("/database/dbname/next")
-	assert.Equal(t, http.StatusOK, w.Code, "GetNext OK")
-	assert.Equal(t, "Hello", string(w.Body.Bytes()), "GetNext sends data")
-	assertExpectations(t, mock_db)
+	suite.Equal(http.StatusOK, w.Code, "GetNext OK")
+	suite.Equal("Hello", string(w.Body.Bytes()), "GetNext sends data")
+}
+
+func (suite *GetNextTestSuite) TestGetNextAddsCounter() {
+	suite.mock_db.On("GetNextRecord", "dbname").Return([]byte("Hello"), nil)
+
+	doRequest("/database/dbname/next")
+	suite.Equal(1, statistics.GetCounter(), "GetNext increases counter")
 }

broker/src/hidra2_broker/server/server.go

@@ -12,6 +12,7 @@ type serverSettings struct {
 }
 
 var settings serverSettings
+var statistics serverStatistics
 
 func InitDB(dbAgent database.Agent) error {
 	db = dbAgent

broker/src/hidra2_broker/server/server_nottested.go

@@ -9,7 +9,14 @@ import (
 	"strconv"
 )
 
+func StartStatistics() {
+	statistics.Writer = new(StatisticInfluxDbWriter)
+	statistics.Reset()
+	go statistics.Monitor()
+}
+
 func Start() {
+	StartStatistics()
 	mux := utils.NewRouter(listRoutes)
 	log.Fatal(http.ListenAndServe("localhost:"+strconv.Itoa(settings.Port), http.HandlerFunc(mux.ServeHTTP)))
 }

broker/src/hidra2_broker/server/statistics.go

+package server
+
+import (
+	"fmt"
+	"log"
+	"time"
+)
+
+type statisticsWriter interface {
+	Write(*serverStatistics) error
+}
+
+type serverStatistics struct {
+	counter int
+	Writer  statisticsWriter
+}
+
+func (st *serverStatistics) IncreaseCounter() {
+	st.counter++
+}
+
+func (st *serverStatistics) GetCounter() int {
+	return st.counter
+}
+
+func (st *serverStatistics) Reset() {
+	st.counter = 0
+}
+
+func (st *serverStatistics) WriteStatistic() (err error) {
+	defer func() {
+		if p := recover(); p != nil {
+			err = fmt.Errorf("WriteStatistic error: %v", p)
+		}
+	}()
+	return st.Writer.Write(st)
+}
+
+func (st *serverStatistics) Monitor() {
+	for {
+		time.Sleep(1000 * time.Millisecond)
+		if err := st.WriteStatistic(); err != nil {
+			log.Println(err.Error())
+		}
+		st.Reset()
+	}
+}

broker/src/hidra2_broker/server/statistics_test.go

+package server
+
+import (
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"testing"
+)
+
+type mockWriter struct {
+	mock.Mock
+}
+
+func (writer *mockWriter) Write(statistics *serverStatistics) error {
+	args := writer.Called(statistics)
+	return args.Error(0)
+}
+
+func assertMockWriterExpectations(t *testing.T, mock_writer *mockWriter) {
+	mock_writer.AssertExpectations(t)
+	mock_writer.ExpectedCalls = nil
+}
+
+func TestMonitorOK(t *testing.T) {
+	mock_writer := new(mockWriter)
+	statistics.Writer = mock_writer
+	statistics.Reset()
+	statistics.IncreaseCounter()
+
+	mock_writer.On("Write", &statistics).Return(nil)
+
+	err := statistics.WriteStatistic()
+	assert.Nil(t, err, "Statistics written")
+
+	assertMockWriterExpectations(t, mock_writer)
+}
+
+func TestMonitorCatchesError(t *testing.T) {
+	statistics.Writer = nil
+	err := statistics.WriteStatistic()
+	assert.NotNil(t, err, "Error with nil pointer")
+}

broker/src/hidra2_broker/server/statistics_writer.go

+//+build !test
+
+package server
+
+import (
+	"github.com/influxdata/influxdb/client/v2"
+	"log"
+	"time"
+)
+
+type StatisticLogWriter struct {
+}
+
+func (writer *StatisticLogWriter) Write(statistics *serverStatistics) error {
+	log.Println(statistics.GetCounter())
+	return nil
+}
+
+type StatisticInfluxDbWriter struct {
+}
+
+func (writer *StatisticInfluxDbWriter) Write(statistics *serverStatistics) error {
+	c, err := client.NewHTTPClient(client.HTTPConfig{
+		Addr: "http://localhost:8086",
+	})
+	if err != nil {
+		return err
+	}
+	defer c.Close()
+
+	bp, _ := client.NewBatchPoints(client.BatchPointsConfig{
+		Database:  "db",
+		Precision: "s",
+	})
+
+	//	tags := map[string]string{"rate": "rate-total"}
+	fields := map[string]interface{}{
+		"rate": statistics.GetCounter(),
+	}
+	pt, err := client.NewPoint("RequestsRate", nil, fields, time.Now())
+	if err != nil {
+		return err
+	}
+	bp.AddPoint(pt)
+
+	return c.Write(bp)
+}