Fix retry_file statistic, now belongs to client

Author: Felixoid

client.go

@@ -50,7 +50,7 @@ func (c Client) saveSliceToRetry(metrics []string, backend string) error {
 	f, err := os.OpenFile(retFile, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
 	if err != nil {
 		c.Lc.lg.Println(err)
-		c.Mon.Increase(&c.Mon.stat[backend].dropped, len(metrics))
+		c.Mon.Increase(&c.Mon.clientStat[backend].dropped, len(metrics))
 		return err
 	}
 	defer f.Close()
@@ -64,7 +64,7 @@ func (c Client) saveSliceToRetry(metrics []string, backend string) error {
 		}
 	}
 	if dropped > 0 {
-		c.Mon.Increase(&c.Mon.stat[backend].dropped, dropped)
+		c.Mon.Increase(&c.Mon.clientStat[backend].dropped, dropped)
 	}
 	return c.removeOldDataFromRetryFile(backend)
 }
@@ -103,10 +103,10 @@ func (c Client) saveChannelToRetry(ch chan string, size int, backend string) {
 		}
 	}
 	if dropped > 0 {
-		c.Mon.Increase(&c.Mon.stat[backend].dropped, dropped)
+		c.Mon.Increase(&c.Mon.clientStat[backend].dropped, dropped)
 	}
 	if saved > 0 {
-		c.Mon.Increase(&c.Mon.stat[backend].saved, saved)
+		c.Mon.Increase(&c.Mon.clientStat[backend].saved, saved)
 	}
 	c.removeOldDataFromRetryFile(backend)
 }
@@ -137,7 +137,7 @@ func (c *Client) tryToSendToGraphite(metric string, conn net.Conn) error {
 		return err
 	}
 	backend := conn.RemoteAddr().String()
-	c.Mon.Increase(&c.Mon.stat[backend].sent, 1)
+	c.Mon.Increase(&c.Mon.clientStat[backend].sent, 1)
 	return nil
 }
 
@@ -190,7 +190,7 @@ func (c Client) runBackend(backend string) {
 						connectionFailed = true
 						break
 					} else {
-						c.Mon.Increase(&c.Mon.got.retry, 1)
+						c.Mon.Increase(&c.Mon.clientStat[backend].fromRetry, 1)
 					}
 
 				} else {
@@ -277,7 +277,7 @@ func (c Client) Run() {
 				select {
 				case c.mainChannels[backend] <- metric:
 				default:
-					c.Mon.Increase(&c.Mon.stat[backend].dropped, 1)
+					c.Mon.Increase(&c.Mon.clientStat[backend].dropped, 1)
 				}
 			}
 		}
@@ -289,7 +289,7 @@ func (c Client) Run() {
 				select {
 				case c.monChannels[backend] <- metric:
 				default:
-					c.Mon.Increase(&c.Mon.stat[backend].dropped, 1)
+					c.Mon.Increase(&c.Mon.clientStat[backend].dropped, 1)
 				}
 			}
 		}

config.go

@@ -258,7 +258,7 @@ func (conf *Config) GenerateLocalConfig() (*LocalConfig, error) {
 	}
 
 	// There are 3 metrics per backend
-	MonitorMetrics := 4 + len(carbonAddrsTCP)*3
+	MonitorMetrics := 3 + len(carbonAddrsTCP)*4
 
 	return &LocalConfig{
 		hostname:       hostname,

grafsy_internal_test.go

@@ -12,33 +12,34 @@ import (
 var cleanMonitoring = &Monitoring{
 	Conf: conf,
 	Lc:   lc,
-	got: source{
-		retry: 0,
-		dir:   0,
-		net:   0,
+	serverStat: serverStat{
+		dir:     0,
+		invalid: 0,
+		net:     0,
 	},
-	stat: map[string]*stat{
-		"127.0.0.1:2003": &stat{
+	clientStat: map[string]*clientStat{
+		"127.0.0.1:2003": &clientStat{
 			saved:   0,
 			sent:    0,
 			dropped: 0,
 		},
-		"127.0.0.1:2004": &stat{
+		"127.0.0.1:2004": &clientStat{
 			saved:   0,
 			sent:    0,
 			dropped: 0,
 		},
 	},
-	invalid: 0,
 }
 
 // These variables defined to prevent reading the config multiple times
 // and avoid code duplication
 var conf, lc, configError = getConfigs()
 
 // There are 3 metrics per backend
-var MonitorMetrics = 4 + len(conf.CarbonAddrs)*3
 var mon, monError = generateMonitoringObject()
+var serverStatMetrics = reflect.ValueOf(mon.serverStat).NumField()
+var clientStatMetrics = reflect.TypeOf(mon.clientStat).Elem().Elem().NumField()
+var MonitorMetrics = serverStatMetrics + len(conf.CarbonAddrs)*clientStatMetrics
 var cli = Client{
 	Conf: conf,
 	Lc:   lc,
@@ -56,29 +57,29 @@ var testMetrics = []string{
 }
 
 func generateMonitoringObject() (*Monitoring, error) {
-	s := source{
-		net:   1,
-		dir:   2,
-		retry: 3,
-	}
 
 	return &Monitoring{
 		Conf: conf,
 		Lc:   lc,
-		got:  s,
-		stat: map[string]*stat{
-			"127.0.0.1:2003": &stat{
-				dropped: 1,
-				saved:   3,
-				sent:    4,
+		serverStat: serverStat{
+			net:     1,
+			invalid: 4,
+			dir:     2,
+		},
+		clientStat: map[string]*clientStat{
+			"127.0.0.1:2003": &clientStat{
+				1,
+				3,
+				2,
+				4,
 			},
-			"127.0.0.1:2004": &stat{
-				dropped: 1,
-				saved:   3,
-				sent:    4,
+			"127.0.0.1:2004": &clientStat{
+				1,
+				3,
+				2,
+				4,
 			},
 		},
-		invalid: 2,
 	}, nil
 }
 
@@ -123,6 +124,7 @@ func TestMonitoring_generateOwnMonitoring(t *testing.T) {
 
 	mon.generateOwnMonitoring()
 	if len(mon.Lc.monitoringChannel) != MonitorMetrics {
+		t.Logf("Fix the amount of metrics for server stats to %v and client stats to %v\n", serverStatMetrics, clientStatMetrics)
 		t.Errorf("Mismatch amount of the monitor metrics: expected=%v, gotten=%v", MonitorMetrics, len(mon.Lc.monitoringChannel))
 	}
 }
@@ -209,7 +211,7 @@ func TestClient_tryToSendToGraphite(t *testing.T) {
 	}
 
 	// Create monitoring structure for statistic
-	cli.Mon.stat[conn.RemoteAddr().String()] = &stat{0, 0, 0}
+	cli.Mon.clientStat[conn.RemoteAddr().String()] = &clientStat{0, 0, 0, 0}
 
 	for _, metric := range testMetrics {
 		cli.tryToSendToGraphite(metric, conn)

monitoring.go

@@ -18,37 +18,37 @@ type Monitoring struct {
 	Lc *LocalConfig
 
 	// Structure with amount of metrics from client.
-	got source
+	serverStat serverStat
 
 	// Statistic per carbon receiver
-	stat map[string]*stat
+	clientStat map[string]*clientStat
+}
+
+// The source of metric daemon got.
+type serverStat struct {
+	// Amount of metrics from directory.
+	dir int
 
 	// Amount of invalid metrics.
 	invalid int
-}
 
-// The source of metric daemon got.
-type source struct {
 	// Amount of metrics from network.
 	net int
+}
 
-	// Amount of metrics from directory.
-	dir int
+// The statistic of metrics per backend
+type clientStat struct {
+	// Amount of dropped metrics.
+	dropped int
 
 	// Amount of metrics from retry file.
-	retry int
-}
+	fromRetry int
 
-// The statistic of metrics per backend
-type stat struct {
 	// Amount of saved metrics.
 	saved int
 
 	// Amount of sent metrics.
 	sent int
-
-	// Amount of dropped metrics.
-	dropped int
 }
 
 var statLock sync.Mutex
@@ -61,18 +61,18 @@ func (m *Monitoring) generateOwnMonitoring() {
 	statLock.Lock()
 
 	monitorSlice := []string{
-		fmt.Sprintf("%s.got.net %v %v", path, m.got.net, now),
-		fmt.Sprintf("%s.got.dir %v %v", path, m.got.dir, now),
-		fmt.Sprintf("%s.got.retry %v %v", path, m.got.retry, now),
-		fmt.Sprintf("%s.invalid %v %v", path, m.invalid, now),
+		fmt.Sprintf("%s.got.net %v %v", path, m.serverStat.net, now),
+		fmt.Sprintf("%s.got.dir %v %v", path, m.serverStat.dir, now),
+		fmt.Sprintf("%s.invalid %v %v", path, m.serverStat.invalid, now),
 	}
 
 	for _, carbonAddrTCP := range m.Lc.carbonAddrsTCP {
 		backend := carbonAddrTCP.String()
 		backendString := strings.Replace(backend, ".", "_", -1)
-		monitorSlice = append(monitorSlice, fmt.Sprintf("%s.%s.saved %v %v", path, backendString, m.stat[backend].saved, now))
-		monitorSlice = append(monitorSlice, fmt.Sprintf("%s.%s.sent %v %v", path, backendString, m.stat[backend].sent, now))
-		monitorSlice = append(monitorSlice, fmt.Sprintf("%s.%s.dropped %v %v", path, backendString, m.stat[backend].dropped, now))
+		monitorSlice = append(monitorSlice, fmt.Sprintf("%s.%s.dropped %v %v", path, backendString, m.clientStat[backend].dropped, now))
+		monitorSlice = append(monitorSlice, fmt.Sprintf("%s.%s.from_retry %v %v", path, backendString, m.clientStat[backend].fromRetry, now))
+		monitorSlice = append(monitorSlice, fmt.Sprintf("%s.%s.saved %v %v", path, backendString, m.clientStat[backend].saved, now))
+		monitorSlice = append(monitorSlice, fmt.Sprintf("%s.%s.sent %v %v", path, backendString, m.clientStat[backend].sent, now))
 	}
 
 	statLock.Unlock()
@@ -84,7 +84,7 @@ func (m *Monitoring) generateOwnMonitoring() {
 			m.Lc.lg.Printf("Too many metrics in the MON queue! This is very bad")
 			for _, carbonAddrTCP := range m.Lc.carbonAddrsTCP {
 				backend := carbonAddrTCP.String()
-				m.Increase(&m.stat[backend].dropped, 1)
+				m.Increase(&m.clientStat[backend].dropped, 1)
 			}
 		}
 	}
@@ -94,12 +94,12 @@ func (m *Monitoring) generateOwnMonitoring() {
 func (m *Monitoring) clean() {
 	for _, carbonAddrTCP := range m.Lc.carbonAddrsTCP {
 		backend := carbonAddrTCP.String()
-		m.stat[backend].saved = 0
-		m.stat[backend].sent = 0
-		m.stat[backend].dropped = 0
+		m.clientStat[backend].dropped = 0
+		m.clientStat[backend].fromRetry = 0
+		m.clientStat[backend].saved = 0
+		m.clientStat[backend].sent = 0
 	}
-	m.invalid = 0
-	m.got = source{0, 0, 0}
+	m.serverStat = serverStat{0, 0, 0}
 }
 
 // Increase metric value in the thread safe way
@@ -113,10 +113,11 @@ func (m *Monitoring) Increase(metric *int, value int) {
 // Should be run in separate goroutine.
 func (m *Monitoring) Run() {
 	statLock.Lock()
-	m.stat = make(map[string]*stat)
+	m.clientStat = make(map[string]*clientStat)
 	for _, carbonAddrTCP := range m.Lc.carbonAddrsTCP {
 		backend := carbonAddrTCP.String()
-		m.stat[backend] = &stat{
+		m.clientStat[backend] = &clientStat{
+			0,
 			0,
 			0,
 			0,
@@ -128,8 +129,8 @@ func (m *Monitoring) Run() {
 		statLock.Lock()
 		for _, carbonAddrTCP := range m.Lc.carbonAddrsTCP {
 			backend := carbonAddrTCP.String()
-			if m.stat[backend].dropped != 0 {
-				m.Lc.lg.Printf("Too many metrics in the main buffer of %s server. Had to drop incommings: %d", backend, m.stat[backend].dropped)
+			if m.clientStat[backend].dropped != 0 {
+				m.Lc.lg.Printf("Too many metrics in the main buffer of %s server. Had to drop incommings: %d", backend, m.clientStat[backend].dropped)
 			}
 		}
 		m.clean()

server.go

@@ -94,7 +94,7 @@ func (s Server) aggrMetricsWithPrefix() {
 		if dropped > 0 {
 			for _, carbonAddrTCP := range s.Lc.carbonAddrsTCP {
 				backend := carbonAddrTCP.String()
-				s.Mon.Increase(&s.Mon.stat[backend].dropped, dropped)
+				s.Mon.Increase(&s.Mon.clientStat[backend].dropped, dropped)
 			}
 		}
 	}
@@ -135,15 +135,15 @@ func (s Server) cleanAndUseIncomingData(metrics []string) {
 			}
 		} else {
 			if metric != "" {
-				s.Mon.Increase(&s.Mon.invalid, 1)
+				s.Mon.Increase(&s.Mon.serverStat.invalid, 1)
 				s.Lc.lg.Printf("Removing bad metric '%s' from the list", metric)
 			}
 		}
 	}
 	if dropped > 0 {
 		for _, carbonAddrTCP := range s.Lc.carbonAddrsTCP {
 			backend := carbonAddrTCP.String()
-			s.Mon.Increase(&s.Mon.stat[backend].dropped, dropped)
+			s.Mon.Increase(&s.Mon.clientStat[backend].dropped, dropped)
 		}
 	}
 }
@@ -153,7 +153,7 @@ func (s Server) handleRequest(conn net.Conn) {
 	defer conn.Close()
 	conBuf := bufio.NewReader(conn)
 	for {
-		s.Mon.Increase(&s.Mon.got.net, 1)
+		s.Mon.Increase(&s.Mon.serverStat.net, 1)
 		metric, err := conBuf.ReadString('\n')
 		// Even if error occurred we still put "metric" into analysis, cause it can be a valid metric, but without \n
 		s.cleanAndUseIncomingData([]string{strings.Replace(strings.Replace(metric, "\r", "", -1), "\n", "", -1)})
@@ -173,7 +173,7 @@ func (s Server) handleDirMetrics() {
 		}
 		for _, f := range files {
 			resultsList, _ := readMetricsFromFile(s.Conf.MetricDir + "/" + f.Name())
-			s.Mon.Increase(&s.Mon.got.dir, len(resultsList))
+			s.Mon.Increase(&s.Mon.serverStat.dir, len(resultsList))
 			s.cleanAndUseIncomingData(resultsList)
 		}