Make the channels way easier

backend/backend.go

@@ -26,7 +26,6 @@ import (
 // Channels are use for unscheduled backend
 type Channels struct {
 	Request *chan Point
-	Done    *chan bool
 }
 
 // Backend Interface

backend/prometheus.go

@@ -4,7 +4,6 @@ package backend
 import (
 	"fmt"
 	"log"
-	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 )
@@ -20,8 +19,7 @@ func (backend *Config) Collect(ch chan<- prometheus.Metric) {
 	log.Println("prometheus: requesting metrics")
 
 	request := make(chan Point, 100)
-	done := make(chan bool)
-	channels := Channels{Request: &request, Done: &done}
+	channels := Channels{Request: &request}
 
 	select {
 	case *queries <- channels:
@@ -33,43 +31,13 @@ func (backend *Config) Collect(ch chan<- prometheus.Metric) {
 
 	// points received
 	points := 0
-	// handle timeout between point reception
-	rectimer := time.NewTimer(100 * time.Millisecond)
-	// check that the collection threads have finished
-	recdone := false
-	for {
-		select {
-		case point := <-*channels.Request:
-			// reset timer
-			if !rectimer.Stop() {
-				select {
-				case <-rectimer.C:
-				default:
-				}
-			}
-			rectimer.Reset(100 * time.Millisecond)
-			// increase points
-			points++
-			// send point to prometheus
-			backend.PrometheusSend(ch, point)
-		case <-*channels.Done:
-			recdone = true
-			// reset timer
-			if !rectimer.Stop() {
-				select {
-				case <-rectimer.C:
-				default:
-				}
-			}
-			rectimer.Reset(100 * time.Millisecond)
-		case <-rectimer.C:
-			// only exit when done and timeout
-			if recdone {
-				log.Printf("prometheus: sent %d points", points)
-				return
-			}
-		}
+	for point := range *channels.Request {
+		// increase points
+		points++
+		// send point to prometheus
+		backend.PrometheusSend(ch, point)
 	}
+	log.Printf("prometheus: sent %d points", points)
 }
 
 //PrometheusSend sends a point to prometheus

backend/thinprometheusclient.go

@@ -7,7 +7,6 @@ import (
 	"log"
 	"sort"
 	"strings"
-	"time"
 
 	"github.com/cblomart/vsphere-graphite/utils"
 	"github.com/valyala/fasthttp"
@@ -53,8 +52,7 @@ func requestHandler(ctx *fasthttp.RequestCtx) {
 	}
 	// prepare the channels for the request
 	request := make(chan Point, 100)
-	done := make(chan bool)
-	channels := Channels{Request: &request, Done: &done}
+	channels := Channels{Request: &request}
 	// create a buffer to organise metrics per type
 	buffer := map[string][]string{}
 	log.Println("thinprom: sending query request")
@@ -66,49 +64,20 @@ func requestHandler(ctx *fasthttp.RequestCtx) {
 		ctx.Error("Query buffer full", fasthttp.StatusConflict)
 		return
 	}
-	// start a timeout
-	timeout := time.NewTimer(100 * time.Millisecond)
+
 	// collected points
 	points := 0
-	// receive done
-	recdone := false
+
 	log.Println("thinprom: waiting for query results")
-L:
-	for {
-		select {
-		case point := <-*channels.Request:
-			// reset timer
-			if !timeout.Stop() {
-				select {
-				case <-timeout.C:
-				default:
-				}
-			}
-			timeout.Reset(100 * time.Millisecond)
-			// increased received points
-			points++
-			// add point to the buffer
-			addToThinPrometheusBuffer(buffer, &point)
-		case <-*channels.Done:
-			// finish consuming metrics and break loop
-			log.Println("thinprom: signaled the end of the collection")
-			recdone = true
-			// reset timer
-			if !timeout.Stop() {
-				select {
-				case <-timeout.C:
-				default:
-				}
-			}
-			timeout.Reset(100 * time.Millisecond)
-		case <-timeout.C:
-			// stop timer
-			if recdone {
-				log.Printf("thinprom: sent %d points", points)
-				break L
-			}
-		}
+	for point := range *channels.Request {
+		// increased received points
+		points++
+		// add point to the buffer
+		addToThinPrometheusBuffer(buffer, &point)
 	}
+	log.Println("thinprom: signaled the end of the collection")
+	log.Printf("thinprom: sent %d points", points)
+
 	ctx.SetContentType("text/plain; charset=utf8")
 	var outbuff bytes.Buffer
 	for key, vals := range buffer {

vsphere-graphite-example.json

@@ -1,6 +1,8 @@
 {
   "Domain": ".contoso.com",
   "ReplacePoint": false,
+  "CPUProfiling": false,
+  "MEMProfiling": false,
   "Properties": [ "all" ],
   "Interval": 60,
   "FlushSize": 1000,

vsphere-graphite.go

@@ -329,8 +329,7 @@ func (service *Service) Manage() (string, error) {
 					go queryVCenter(*vcenter, conf, request.Request, &wg)
 				}
 				wg.Wait()
-				//time.Sleep(5 * time.Second)
-				*request.Done <- true
+				close(*request.Request)
 				cleanup <- true
 			}()
 		case <-ticker.C: