Merge pull request #5 from matttproud/feature/scraping-infrastructure

Full End-to-End Scraping Infrastructure

main.go

@@ -19,6 +19,7 @@ import (
 	"github.com/matttproud/prometheus/storage/metric/leveldb"
 	"log"
 	"os"
+	"time"
 )
 
 func main() {
@@ -32,21 +33,22 @@ func main() {
 		m.Close()
 	}()
 
+	results := make(chan retrieval.Result, 4096)
+
 	t := &retrieval.Target{
 		Address:  "http://localhost:8080/metrics.json",
+		Deadline: time.Second * 5,
+		Interval: time.Second * 3,
 	}
 
-	for i := 0; i < 100000; i++ {
-		c, err := t.Scrape()
-		if err != nil {
-			fmt.Println(err)
-			continue
-		}
+	manager := retrieval.NewTargetManager(results, 1)
+	manager.Add(t)
 
-		for _, s := range c {
+	for {
+		result := <-results
+		fmt.Printf("result -> %s\n", result)
+		for _, s := range result.Samples {
 			m.AppendSample(&s)
 		}
-
-		fmt.Printf("Finished %d\n", i)
 	}
 }

retrieval/target.go

@@ -10,10 +10,17 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
-
 package retrieval
 
 import (
+	"encoding/json"
+	"fmt"
+	"github.com/matttproud/prometheus/model"
+	"io/ioutil"
+	"log"
+	"math"
+	"net/http"
+	"strconv"
 	"time"
 )
 
@@ -25,24 +32,86 @@ const (
 	UNREACHABLE
 )
 
+const (
+	MAXIMUM_BACKOFF = time.Minute * 30
+)
+
 type Target struct {
 	scheduledFor     time.Time
 	unreachableCount int
 	state            TargetState
 
 	Address  string
+	Deadline time.Duration
+
+	// XXX: Move this to a field with the target manager initialization instead of here.
 	Interval time.Duration
 }
 
-// KEPT FOR LEGACY COMPATIBILITY; PENDING REFACTOR
-
-func (t *Target) Scrape() (samples []model.Sample, err error) {
-	defer func() {
-		if err != nil {
-			t.state = ALIVE
+type Result struct {
+	Err     error
+	Samples []model.Sample
+	Target  Target
 }
+
+func (t *Target) reschedule(s TargetState) {
+	currentState := t.state
+
+	switch currentState {
+	case UNKNOWN, UNREACHABLE:
+		switch s {
+		case ALIVE:
+			t.unreachableCount = 0
+		case UNREACHABLE:
+			backoff := MAXIMUM_BACKOFF
+			exponential := time.Duration(math.Pow(2, float64(t.unreachableCount))) * time.Second
+			if backoff > exponential {
+				backoff = exponential
+			}
+
+			t.scheduledFor = time.Now().Add(backoff)
+			t.unreachableCount++
+
+			log.Printf("%s unavailable %s times deferred for %s.", t, t.unreachableCount, backoff)
+		default:
+		}
+	case ALIVE:
+		switch s {
+		case UNREACHABLE:
+			t.unreachableCount++
+		}
+	default:
+	}
+
+	if s != currentState {
+		log.Printf("%s transitioning from %s to %s.", t, currentState, s)
+	}
+
+	t.state = s
+}
+
+func (t *Target) Scrape(results chan Result) (err error) {
+	result := Result{}
+
+	defer func() {
+		futureState := t.state
+
+		switch err {
+		case nil:
+			futureState = ALIVE
+		default:
+			futureState = UNREACHABLE
+		}
+
+		t.reschedule(futureState)
+
+		result.Err = err
+		results <- result
 	}()
 
+	done := make(chan bool)
+
+	go func() {
 		ti := time.Now()
 		resp, err := http.Get(t.Address)
 		if err != nil {
@@ -90,7 +159,7 @@ func (t *Target) Scrape() (samples []model.Sample, err error) {
 					m[model.LabelName(baseK)] = model.LabelValue(baseV)
 				}
 
-			samples = append(samples, s)
+				result.Samples = append(result.Samples, s)
 			case "histogram":
 				values, ok := asMap["value"].(map[string]interface{})
 				if !ok {
@@ -122,10 +191,20 @@ func (t *Target) Scrape() (samples []model.Sample, err error) {
 						m[model.LabelName(baseK)] = model.LabelValue(baseV)
 					}
 
-				samples = append(samples, s)
+					result.Samples = append(result.Samples, s)
 				}
 			}
 		}
 
+		done <- true
+	}()
+
+	select {
+	case <-done:
+		break
+	case <-time.After(t.Deadline):
+		err = fmt.Errorf("Target %s exceeded %s deadline.", t, t.Deadline)
+	}
+
 	return
 }

retrieval/targetmanager.go

@@ -0,0 +1,66 @@
+// Copyright 2013 Prometheus Team
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package retrieval
+
+import (
+	"container/heap"
+	"log"
+	"time"
+)
+
+type TargetManager interface {
+	acquire()
+	release()
+	Add(t *Target)
+	Remove(t *Target)
+}
+
+type targetManager struct {
+	requestAllowance chan bool
+	pools            map[time.Duration]TargetPool
+	results          chan Result
+}
+
+func NewTargetManager(results chan Result, requestAllowance int) TargetManager {
+	return targetManager{
+		requestAllowance: make(chan bool, requestAllowance),
+		results:          results,
+		pools:            make(map[time.Duration]TargetPool),
+	}
+}
+
+func (m targetManager) acquire() {
+	m.requestAllowance <- true
+}
+
+func (m targetManager) release() {
+	<-m.requestAllowance
+}
+
+func (m targetManager) Add(t *Target) {
+	targetPool, ok := m.pools[t.Interval]
+
+	if !ok {
+		targetPool.manager = m
+		log.Printf("Pool %s does not exist; creating and starting...", t.Interval)
+		go targetPool.Run(m.results, t.Interval)
+	}
+
+	heap.Push(&targetPool, t)
+	m.pools[t.Interval] = targetPool
+}
+
+func (m targetManager) Remove(t *Target) {
+	panic("not implemented")
+}

retrieval/targetpool.go

@@ -1,38 +1,92 @@
 package retrieval
 
 import (
-	"encoding/json"
-	"github.com/matttproud/prometheus/model"
-	"io/ioutil"
-	"net/http"
-	"strconv"
+	"container/heap"
+	"log"
 	"time"
 )
 
-type TargetPool []*Target
+type TargetPool struct {
+	done    chan bool
+	targets []*Target
+	manager TargetManager
+}
+
+func NewTargetPool(m TargetManager) (p TargetPool) {
+	p.manager = m
+
+	return
+}
 
 func (p TargetPool) Len() int {
-	return len(p)
+	return len(p.targets)
 }
 
 func (p TargetPool) Less(i, j int) bool {
-	return p[i].scheduledFor.Before(p[j].scheduledFor)
+	return p.targets[i].scheduledFor.Before(p.targets[j].scheduledFor)
 }
 
 func (p *TargetPool) Pop() interface{} {
-	oldPool := *p
+	oldPool := p.targets
 	futureLength := p.Len() - 1
 	element := oldPool[futureLength]
 	futurePool := oldPool[0:futureLength]
-	*p = futurePool
+	p.targets = futurePool
 
 	return element
 }
 
 func (p *TargetPool) Push(element interface{}) {
-	*p = append(*p, element.(*Target))
+	p.targets = append(p.targets, element.(*Target))
 }
 
 func (p TargetPool) Swap(i, j int) {
-	p[i], p[j] = p[j], p[i]
+	p.targets[i], p.targets[j] = p.targets[j], p.targets[i]
+}
+
+func (p *TargetPool) Run(results chan Result, interval time.Duration) {
+	ticker := time.Tick(interval)
+
+	for {
+		select {
+		case <-ticker:
+			p.runIteration(results)
+		case <-p.done:
+			log.Printf("TargetPool exiting...")
+			break
+		}
+	}
+}
+
+func (p TargetPool) Stop() {
+	p.done <- true
+}
+
+func (p *TargetPool) runSingle(results chan Result, t *Target) {
+	p.manager.acquire()
+	defer p.manager.release()
+
+	t.Scrape(results)
+}
+
+func (p *TargetPool) runIteration(results chan Result) {
+	for i := 0; i < p.Len(); i++ {
+		target := heap.Pop(p).(*Target)
+		if target == nil {
+			break
+		}
+
+		now := time.Now()
+
+		if target.scheduledFor.After(now) {
+			heap.Push(p, target)
+
+			break
+		}
+
+		go func() {
+			p.runSingle(results, target)
+			heap.Push(p, target)
+		}()
+	}
 }