bitly · Jun 16, 2015 · Jun 16, 2015 · Jun 16, 2015 · Jun 17, 2015 · Jun 17, 2015
Showing with 160 additions and 29 deletions.

+3 −0 .whitesource

+15 −0 conf/statsdaemon.conf

+21 −0 scripts/jenkins.go.ws.sh

+26 −0 scripts/udpstats.pl

+95 −29 statsdaemon.go
diff --git a/.whitesource b/.whitesource
@@ -0,0 +1,3 @@
+{
+  "settingsInheritedFrom": "akiunlocks/whitesource-config@main"
+}
diff --git a/conf/statsdaemon.conf b/conf/statsdaemon.conf
@@ -0,0 +1,15 @@
+# statsite
+#
+
+description     "statsdaemon StatsD service"
+
+start on virtual-filesystems
+stop on runlevel [06]
+
+respawn
+respawn limit 5 30
+limit nofile 65550 65550
+
+script
+  exec /opt/statsdaemon/statsdaemon -prefix="prefix.stats." -graphite="carbon.hostedgraphite.com:2003,127.0.0.1:2003" >>/mnt/log/statsdaemon.log 2>&1
+end script
diff --git a/scripts/jenkins.go.ws.sh b/scripts/jenkins.go.ws.sh
@@ -0,0 +1,21 @@
+#!/bin/bash
+
+#Using s3://eyeview-deploy/tools/go1.4.1.linux-amd64.tar.gz  aka https://storage.googleapis.com/golang/go1.4.1.linux-amd64.tar.gz
+
+#export GOROOT=/opt/go_1.8
+export GOROOT=/opt/go
+export GOPATH=`pwd`
+export PATH=$GOROOT/bin:$PATH
+echo "gopath:"
+echo $GOPATH
+
+#HACK: It doesn't look like there's a way to change the git local directory for checkout so moving files manually for now
+echo "cleaning src/github.com dir"
+rm -rf src/github.com || /bin/true
+mkdir -p src/github.com/eyeview/statsdaemon || /bin/true
+find . -maxdepth 1|egrep -v '.git|src|bin|pkg|deploy.|^.$|workspace'|xargs mv -f -t src/github.com/eyeview/statsdaemon/
+
+cd src/github.com/eyeview/statsdaemon/
+${GOROOT}/bin/go build
+#${GOROOT}/bin/go test ./...
+
diff --git a/scripts/udpstats.pl b/scripts/udpstats.pl
@@ -0,0 +1,26 @@
+#!/usr/bin/perl
+
+my ($rec, $err, $rec_n, $err_n) = (0,0,0,0);
+
+while (true) {
+
+	my @stats = `netstat -s|grep Udp: -A4`;
+#Udp:
+#    851851898 packets received
+#    4389457 packets to unknown port received.
+#    3449421204 packet receive errors
+#    5634268 packets sent 
+
+	for my $l (@stats) {
+		$rec_n = $1 if ($l =~ /(\d+) packets received/);
+		$err_n = $1 if ($l =~ /(\d+) packet receive errors/);
+	}
+	my $p_n = ($rec_n - $rec);
+	my $r = 1.0* ($err_n - $err) / $p_n;
+	print "Error rate $r\t packets $p_n\n" unless ($rec == 0 && $err == 0);
+
+	$rec = $rec_n;
+	$err = $err_n;
+
+	sleep(3);
+}
diff --git a/statsdaemon.go b/statsdaemon.go
@@ -116,6 +116,7 @@ var (
 	timers          = make(map[string]Uint64Slice)
 	countInactivity = make(map[string]int64)
 	sets            = make(map[string][]string)
+	updates         = make(map[string]int64)
 )
 
 func monitor() {
@@ -148,6 +149,8 @@ func packetHandler(s *Packet) {
 		counters[*receiveCounter] += 1
 	}
 
+	updates[s.Bucket] +=1
+
 	switch s.Modifier {
 	case "ms":
 		_, ok := timers[s.Bucket]
@@ -206,32 +209,35 @@ func submit(deadline time.Time) error {
 		return nil
 	}
 
-	client, err := net.Dial("tcp", *graphiteAddress)
+	num += processUpdates(&buffer, now)
+	num += processCounters(&buffer, now)
+	num += processGauges(&buffer, now)
+	num += processTimers(&buffer, now, percentThreshold)
+	num += processSets(&buffer, now)
+	if num == 0 {
+		return nil
+	}
+
+	addresses := strings.Split(*graphiteAddress, ",") //TODO - do this once in the beginning
+	for _, address := range addresses {
+		go send(address, deadline, buffer, num)
+	}
+
+	return nil
+}
+
+func send(address string, deadline time.Time, buffer bytes.Buffer, num int64) error {
+	client, err := net.Dial("tcp", address)
 	if err != nil {
-		if *debug {
-			log.Printf("WARNING: resetting counters when in debug mode")
-			processCounters(&buffer, now)
-			processGauges(&buffer, now)
-			processTimers(&buffer, now, percentThreshold)
-			processSets(&buffer, now)
-		}
 		errmsg := fmt.Sprintf("dialing %s failed - %s", *graphiteAddress, err)
 		return errors.New(errmsg)
 	}
-	defer client.Close()
 
 	err = client.SetDeadline(deadline)
 	if err != nil {
 		return err
 	}
-
-	num += processCounters(&buffer, now)
-	num += processGauges(&buffer, now)
-	num += processTimers(&buffer, now, percentThreshold)
-	num += processSets(&buffer, now)
-	if num == 0 {
-		return nil
-	}
+	defer client.Close()
 
 	if *debug {
 		for _, line := range bytes.Split(buffer.Bytes(), []byte("\n")) {
@@ -248,23 +254,49 @@ func submit(deadline time.Time) error {
 		return errors.New(errmsg)
 	}
 
-	log.Printf("sent %d stats to %s", num, *graphiteAddress)
+	log.Printf("sent %d stats to %s", num, address)
 
 	return nil
+
+}
+
+func processUpdates(buffer *bytes.Buffer, now int64) int64 {
+	if int64(len(updates)) == 0 {
+		log.Printf("No Updates found")
+		return 0
+	}
+	var maxBucket string
+	var maxValue int64
+	for bucket, value := range updates {
+		if maxBucket == "" || maxValue < value {
+			maxBucket = bucket
+			maxValue = value
+		}
+		delete(updates, bucket)
+	}
+	maxBucket = strings.Replace(maxBucket, "counters","updates",-1)
+	maxBucket = strings.Replace(maxBucket, "timers","updates",-1)
+	maxBucket = strings.Replace(maxBucket, "gauges","updates",-1)
+	log.Printf("Max updates is %s", maxBucket)
+	fmt.Fprintf(buffer, "%s %d %d\n", maxBucket, maxValue, now)
+	return 1
 }
 
 func processCounters(buffer *bytes.Buffer, now int64) int64 {
 	var num int64
 	// continue sending zeros for counters for a short period of time even if we have no new data
 	for bucket, value := range counters {
-		fmt.Fprintf(buffer, "%s %d %d\n", bucket, value, now)
+		//fmt.Fprintf(buffer, "%s.count %d %d\n", bucket, value, now)
+		rate := float64(value) / float64(*flushInterval)
+		fmt.Fprintf(buffer, "%s.rate %f %d\n", bucket, rate, now)
 		delete(counters, bucket)
 		countInactivity[bucket] = 0
 		num++
 	}
 	for bucket, purgeCount := range countInactivity {
 		if purgeCount > 0 {
-			fmt.Fprintf(buffer, "%s %d %d\n", bucket, 0, now)
+			//fmt.Fprintf(buffer, "%s %d %d\n", bucket, 0, now)
+			fmt.Fprintf(buffer, "%s.rate %d %d\n", bucket, 0, now)
 			num++
 		}
 		countInactivity[bucket] += 1
@@ -499,17 +531,26 @@ func parseLine(line []byte) *Packet {
 	}
 
 	var (
-		err   error
-		value interface{}
+		err      error
+		value    interface{}
+		stattype string
 	)
 
 	switch typeCode {
 	case "c":
 		value, err = strconv.ParseInt(string(val), 10, 64)
 		if err != nil {
-			log.Printf("ERROR: failed to ParseInt %s - %s", string(val), err)
-			return nil
+			//try to round a float
+			value, err = strconv.ParseFloat(string(val), 64)
+			if (err != nil) {
+				log.Printf("ERROR: failed to Parse (type c): %s - %s", string(val), err)
+				return nil
+			} else {
+				value = round(value.(float64))
+			}
+
 		}
+		stattype = "counters."
 	case "g":
 		var rel, neg bool
 		var s string
@@ -531,32 +572,57 @@ func parseLine(line []byte) *Packet {
 
 		value, err = strconv.ParseUint(s, 10, 64)
 		if err != nil {
-			log.Printf("ERROR: failed to ParseUint %s - %s", string(val), err)
-			return nil
+			//try to round a float
+			value, err = strconv.ParseFloat(string(val), 64)
+			if (err != nil) {
+				log.Printf("ERROR: failed to Parse (type g): %s - %s", string(val), err)
+				return nil
+			} else {
+				value = rounduint64(value.(float64))
+			}
 		}
 
 		value = GaugeData{rel, neg, value.(uint64)}
+		stattype = "gauges."
 	case "s":
 		value = string(val)
+		stattype = "timers."
 	case "ms":
 		value, err = strconv.ParseUint(string(val), 10, 64)
 		if err != nil {
-			log.Printf("ERROR: failed to ParseUint %s - %s", string(val), err)
-			return nil
+			//try to round a float
+			value, err = strconv.ParseFloat(string(val), 64)
+			if (err != nil) {
+				log.Printf("ERROR: failed to Parse (type ms): %s - %s", string(val), err)
+				return nil
+			} else {
+				value = rounduint64(value.(float64))
+			}
 		}
+		stattype = "timers."
 	default:
 		log.Printf("ERROR: unrecognized type code %q", typeCode)
 		return nil
 	}
 
 	return &Packet{
-		Bucket:   sanitizeBucket(*prefix + string(name) + *postfix),
+		Bucket:   sanitizeBucket(*prefix + stattype + string(name) + *postfix),
 		Value:    value,
 		Modifier: typeCode,
 		Sampling: sampling,
 	}
 }
 
+func round(val float64) int64 {
+    if val < 0 { return int64(val-0.5) }
+    return int64(val+0.5)
+}
+
+func rounduint64(val float64) uint64 {
+    if val < 0 { return uint64(val-0.5) }
+    return uint64(val+0.5)
+}
+
 func logParseFail(line []byte) {
 	if *debug {
 		log.Printf("ERROR: failed to parse line: %q\n", string(line))